Harden distributed deduplication and adaptive execution

.github/workflows/unit-tests.yml

@@ -2,18 +2,33 @@ name: Unit tests
 
 on:
   pull_request:
-    types: [opened, synchronize, reopened, ready_for_review]
+    branches: [main]
+  push:
+    branches: [main]
 
 permissions:
   contents: read
 
 concurrency:
-  group: unit-tests-${{ github.event.pull_request.number }}
+  group: unit-tests-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
 
 jobs:
+  workflow-lint:
+    name: Validate GitHub workflows
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v6
+
+      - name: Run actionlint
+        uses: raven-actions/actionlint@v2
+
   validate:
     name: Ruff and lightweight tests
+    needs: workflow-lint
     runs-on: ubuntu-latest
     timeout-minutes: 30
 
@@ -25,6 +40,14 @@ jobs:
       - name: Check out repository
         uses: actions/checkout@v6
 
+      - name: Cache Conda packages
+        uses: actions/cache@v4
+        with:
+          path: ~/conda_pkgs_dir
+          key: conda-${{ runner.os }}-${{ hashFiles('environment.dev.yaml') }}
+          restore-keys: |
+            conda-${{ runner.os }}-
+
       - name: Create development environment
         uses: conda-incubator/setup-miniconda@v4
         with:
@@ -33,6 +56,15 @@ jobs:
           miniforge-version: latest
           conda-solver: libmamba
           auto-activate-base: false
+          use-only-tar-bz2: true
+
+      - name: Cache pre-commit environments
+        uses: actions/cache@v4
+        with:
+          path: ~/.cache/pre-commit
+          key: pre-commit-${{ runner.os }}-${{ hashFiles('.pre-commit-config.yaml') }}
+          restore-keys: |
+            pre-commit-${{ runner.os }}-
 
       - name: Show environment
         run: |

.gitignore

@@ -149,8 +149,5 @@ process*
 
 # other
 config-new.yaml
-config.all.yaml
-config.test.elais.yaml
 flags_translation_reports/
-test_data/make_collection
-.codex
+.codex

.pre-commit-config.yaml

@@ -1,9 +1,25 @@
 repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v6.0.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-json
+      - id: check-toml
+      - id: check-merge-conflict
+      - id: check-added-large-files
+      - id: check-case-conflict
+      - id: detect-private-key
+      - id: debug-statements
+      - id: no-commit-to-branch
+        args: [--branch, main]
+
   - repo: local
     hooks:
       - id: ruff-essential
-        name: Ruff essential checks
-        entry: ruff check --force-exclude --select=E9,F,B006,B008,B012,B018,B023,B904
+        name: Ruff essential checks and import sorting
+        entry: ruff check --fix --force-exclude --select=E9,F,I,B006,B008,B012,B018,B023,B904
         language: system
         types_or: [python, pyi]
         exclude: ^notebooks/

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2021 LIneA IT 
+Copyright (c) 2021 LIneA IT
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -32,13 +32,6 @@ unit tests:
 pytest
 ```
 
-Integration tests execute the complete pipeline, may manage a separate runtime
-environment, and are excluded from the default suite. Run them explicitly with:
-
-```bash
-pytest -m slow
-```
-
 ### Pre-commit checks
 
 With `pipe_crd_dev` activated, install the repository hook once:
@@ -58,32 +51,26 @@ Run the same checks manually across the repository with:
 pre-commit run --all-files
 ```
 
-Pull requests run the same Ruff and lightweight unit-test checks automatically
-through GitHub Actions. New commits cancel any older validation run still in
-progress for that pull request.
+Pull requests targeting `main` and commits merged into `main` run the same Ruff
+and lightweight unit-test checks automatically through GitHub Actions. The
+workflow also validates its own GitHub Actions configuration. New commits cancel
+older validation runs still in progress for the same pull request or branch.
 
 ### Test data
 
-This repository currently contains a basic dataset, for testing purposes only. The ideal is to connect the pipelines to systems with access to a larger datasets.
+The repository contains three small sample catalogs used by
+`config.template.yaml` for local smoke runs. Unit tests create their own
+temporary fixtures and do not depend on these files.
 
 ### Install
 
 The only requirement is to have `micromamba` available in `PATH`:
 
 ```bash
 git clone https://github.com/linea-it/pzserver_combine_redshift_dedup && cd pzserver_combine_redshift_dedup
-./setup.sh
-source env.sh
-```
-
-To install all pipelines at once:
-
-```bash
 ./install.sh
 ```
 
-The `setup.sh` will suggest a directory where the pipelines and datasets are installed, type 'yes' to confirm or 'no' to configure the desired path in each case with the respective environment variables and then run again `setup.sh`.
-
 The installation script creates the `pipe_crd` environment with `micromamba`.
 
 By default the scripts use `MAMBA_ROOT_PREFIX="$HOME/.micromamba"`. On a Slurm cluster, point this variable to a persistent location visible to the jobs if needed:
@@ -95,20 +82,9 @@ export MAMBA_ROOT_PREFIX=/path/to/shared/or/persistent/micromamba
 
 ## Run a pipeline
 
-To execute, simply:
+Copy and edit the example configuration, then execute the pipeline:
 
 ```bash
-# execute combine redshift catalogs
+cp config.template.yaml config.yaml
 ./run.sh config.yaml process001
 ```
-
-
-## Validation notebook
-
-To validate your test results, use the notebook `validation.ipynb`:
-
-```
-conda install -c conda-forge jupyterlab ipykernel
-python -m ipykernel install --user --name=pipe_crd
-jupyter lab
-```

config.py

@@ -19,9 +19,11 @@ class Instance(BaseModel):
         job_extra_directives: list[str] = ["--propagate", "--time=12:00:00"]
 
     class Scale(BaseModel):
-        minimum_jobs: int = 10
+        minimum_jobs: int = 3
         maximum_jobs: int = 22
-        worker_recovery_timeout_seconds: float = 600.0
+        adaptive_interval_seconds: float = 10.0
+        adaptive_scale_down_delay_seconds: float = 180.0
+        worker_recovery_timeout_seconds: float = 360.0
         worker_recovery_check_interval_seconds: float = 10.0
 
         @model_validator(mode="after")
@@ -30,6 +32,12 @@ def validate_limits(self):
                 raise ValueError("minimum_jobs must be at least 1")
             if self.maximum_jobs < self.minimum_jobs:
                 raise ValueError("maximum_jobs must be >= minimum_jobs")
+            if self.adaptive_interval_seconds <= 0:
+                raise ValueError("adaptive_interval_seconds must be positive")
+            if self.adaptive_scale_down_delay_seconds <= 0:
+                raise ValueError(
+                    "adaptive_scale_down_delay_seconds must be positive"
+                )
             if self.worker_recovery_timeout_seconds <= 0:
                 raise ValueError("worker_recovery_timeout_seconds must be positive")
             if self.worker_recovery_check_interval_seconds <= 0:
@@ -93,7 +101,7 @@ class Columns(BaseModel):
 class Param(BaseModel):
     combine_type: str = "concatenate"
     extra_columns: dict[str, Any] = Field(default_factory=dict)
-    # Valid cuts are 1, 2, 3, 4, 5, 6. Invalid values skip the cut with a warning.
+    # Zero disables the cut; valid active cuts are 1, 2, 3, 4, 5, 6.
     z_flag_homogenized_value_to_cut: float = 3.0
     flags_translation_file: str = str(Path(MAINDIR, "flags_translation.yaml"))
     insert_DP1_footprint_flag: bool = False

config.template.yaml

@@ -23,9 +23,11 @@ executor:
 #        - "--propagate"
 #        - "--time=12:00:00"
 #    scale:
-#      minimum_jobs: 10
+#      minimum_jobs: 3
 #      maximum_jobs: 22
-#      worker_recovery_timeout_seconds: 600
+#      adaptive_interval_seconds: 10
+#      adaptive_scale_down_delay_seconds: 180
+#      worker_recovery_timeout_seconds: 360
 #      worker_recovery_check_interval_seconds: 10
 
 inputs:
@@ -85,7 +87,7 @@ param:
   #   ORIGINAL_ID:
   #     source: id
   #     type: str
-  z_flag_homogenized_value_to_cut: 0.0 # Valid cuts are 1, 2, 3, 4, 5, 6. Invalid values (<=0, >6, non-integer) skip the cut with a warning.
+  z_flag_homogenized_value_to_cut: 0.0 # 0 disables the cut; valid active cuts are 1, 2, 3, 4, 5, 6.
   insert_DP1_footprint_flag: false # Adds is_in_DP1_fields (1/0)
   insert_rubin_footprint_flag: true # Adds is_in_rubin_footprint (1/0)
   flags_translation_file: flags_translation.yaml # File with homogenization rules for z_flag and type

flags_translation.yaml

@@ -9,7 +9,11 @@ tiebreaking_priority:
   # - It exists in all catalogs being crossmatched.
   # - It is numeric (except for 'instrument_type_homogenized', which uses a string-to-priority mapping).
   # - Higher values are considered better; lower values are worse.
+  # Generic priority columns are normalized to float64 per catalog and must
+  # contain at least one valid numeric value in every input catalog.
   # The columns will be evaluated in the order listed to resolve duplicates.
+  # In deduplication modes, z_flag_homogenized is still created and preserved
+  # for star classification (value 6), even when omitted from this ranking.
 
 delta_z_threshold: 0.001    # Maximum allowed redshift difference (|z1 - z2|) to resolve hard ties
                              # Applied only when initial tiebreaking via columns results in a tie (tie_result == 2)
@@ -133,7 +137,7 @@ translation_rules:
       6: 6
     instrument_type_translation:
       default: "s"
-      
+
   ASTRODEEP:
     z_flag_translation:
       conditions:
@@ -149,7 +153,7 @@ translation_rules:
         - expr: "zspec_survey == '-'"
           value: "p"
       default: "p"
-      
+
   ASTRODEEP-JWST:
     z_flag_translation:
       conditions: