DM-56033: Remove APDB preload from Daytime pipeline

bps/LSSTCam/bps_Daytime.yaml

@@ -1,4 +1,4 @@
-pipelineYaml: $AP_PIPE_DIR/pipelines/LSSTCam/ApPipe.yaml
+pipelineYaml: $AP_PIPE_DIR/pipelines/LSSTCam/ApPipeDaytime.yaml
 
 project: ApPipe
 campaign: AP-daytime

bps/clustering/clustering_Daytime.yaml

@@ -21,9 +21,6 @@ cluster:
   preload:
     pipetasks: mpSkyEphemerisQuery,getRegionTimeFromVisit
     dimensions: group,detector
-  preloadApdb:
-    pipetasks: loadDiaCatalogs,analyzeLoadDiaCatalogsMetrics
-    dimensions: group,detector
   singleFrame:
     pipetasks: isr,calibrateImage,analyzePreliminarySummaryStats
     dimensions: visit,detector
@@ -52,6 +49,6 @@ ordering:
   associationOrder:
     ordering_type: sort
     findDependencyMethod: sink
-    labels: preloadApdb,association
+    labels: association
     dimensions: visit
     blocking: False

doc/lsst.ap.pipe/pipeline-overview.rst

@@ -33,6 +33,11 @@ to verify the output.
 :doc:`ap_pipe <index>` is entirely written in Python. Key contents include:
 
 - :file:`ApPipe.yaml`: a `~lsst.pipe.base.Pipeline` configuration for running the entire AP Pipeline.
+- :file:`LSSTCam/ApPipeDaytime.yaml`: the variant used for daytime (non-real-time) LSSTCam processing.
+  It drops ``loadDiaCatalogs`` and has `~lsst.ap.association.DiaPipelineTask` read the DIAObject and
+  DIASource history from the APDB during association, so that the duplicate DIASource check sees rows
+  written by any earlier pass over the same image.
+  Prompt Processing must not use it; the preload is what keeps the APDB out of its latency-critical path.
 
 By default the pipeline is limited to running on data taken in filter bands whose names match those used by the Rubin Observatory LSST Camera (that is `ugrizy`).
 In order to run on bands outside of these filters, one must add the associated columns to the `~lsst.dax.apdb.Apdb` schema and add the band names to the config of `~lsst.ap.association.DiaPipelineTask`.

pipelines/LSSTCam/ApPipeDaytime.yaml

@@ -0,0 +1,50 @@
+description: >-
+  AP pipeline for daytime (non-real-time) processing of LSSTCam data.
+
+  Association reads the DiaObject, DiaSource, and DiaForcedSource history
+  directly from the APDB rather than consuming preloaded catalogs, so the
+  duplicate DiaSource check in associateApdb can see rows written by an
+  earlier pass over the same image. The preloaded catalogs are built before
+  the image is processed and so cannot contain those rows; reusing them let
+  a reprocessing run write duplicate diaSources to the APDB (DM-55633).
+  loadDiaCatalogs is therefore dropped from this pipeline entirely, which
+  also stops --skip-existing-in from reviving a stale preload.
+
+  Prompt Processing must not use this pipeline. Its preload exists to keep
+  the APDB out of the latency-critical path, and it guards its own retries
+  with Apdb.containsVisitDetector.
+instrument: lsst.obs.lsst.LsstCam
+imports:
+  - location: $AP_PIPE_DIR/pipelines/LSSTCam/ApPipe.yaml
+    exclude:
+      - loadDiaCatalogs
+      - analyzeLoadDiaCatalogsMetrics
+    # EDIT drops the excluded labels from the inherited subsets. The
+    # default, DROP, would delete preload and promptQaMetrics outright
+    # and force apPipe to be restated here.
+    labeledSubsetModifyMode: EDIT
+
+tasks:
+  associateApdb:
+    class: lsst.ap.association.DiaPipelineTask
+    config:
+      # Load all of the APDB catalogs in association, instead of relying on
+      # preloaded catalogs from loadDiaCatalogs
+      doReloadAllApdbCatalogs: True
+  analyzeDiaSourceAssociationMetrics:
+    class: lsst.analysis.tools.tasks.TaskMetadataAnalysisTask
+    config:
+      # Publish the APDB read timings that analyzeLoadDiaCatalogsMetrics used
+      # to provide. The metric names match the ones that task emitted, so
+      # they reach the same Sasquatch topics as in Prompt Processing; only
+      # the dataset type carrying them changes, since the timings now come
+      # from a visit-dimensioned quantum.
+      # Setting `metrics` replaces the dict, so the inherited entries are
+      # repeated here.
+      atools.associationMetadataMetrics.metrics:
+        numTotalSolarSystemObjects: ct
+        numAssociatedSsObjects: ct
+        writeToApdbDuration: s
+        loadDiaObjectsDuration: s
+        loadDiaSourcesDuration: s
+        loadDiaForcedSourcesDuration: s

pipelines/_ingredients/ApPipe.yaml

@@ -312,8 +312,14 @@ subsets:
       Requires prompt subset to be run first.
 contracts:
   - detectAndMeasureDiaSource.doSkySources == filterDiaSource.doRemoveSkySources
-  # Both loadDiaCatalogs and associateApdb connect to the APDB, so make sure they use the same configuration
-  - loadDiaCatalogs.apdb_config_url == associateApdb.apdb_config_url
+  # Both loadDiaCatalogs and associateApdb connect to the APDB, so make sure
+  # they use the same configuration. associateApdb must not also reload the
+  # full history, which would leave the preload unused; ApPipeDaytime.yaml
+  # drops loadDiaCatalogs instead.
+  - contract: loadDiaCatalogs.apdb_config_url == associateApdb.apdb_config_url
+              and not associateApdb.doReloadAllApdbCatalogs
+    msg: "loadDiaCatalogs and associateApdb must share an APDB config, and
+          associateApdb.doReloadAllApdbCatalogs must be False when loadDiaCatalogs runs"
   # to reduce latency, we need two calls to the sattle service when active
   - calibrateImage.run_sattle == detectAndMeasureDiaSource.run_sattle
   # Inputs and outputs must match. For consistency, contracts are written in execution order:

scripts/LSSTCam/submit_ap_daytime.sh

@@ -39,7 +39,7 @@ BLOCKS_SQL="($(printf "'%s'," $BLOCKS | sed 's/,$//'))"
 
 # Pipeline and butler config must mirror bps_Daytime.yaml — we replicate them
 # here because we build the quantum graph ourselves before calling BPS.
-PIPELINE_YAML="${AP_PIPE_DIR}/pipelines/LSSTCam/ApPipe.yaml"
+PIPELINE_YAML="${AP_PIPE_DIR}/pipelines/LSSTCam/ApPipeDaytime.yaml"
 APDB_CONFIG="s3://embargo@rubin-summit-users/apdb_config/cassandra/pp_apdb_lsstcam.yaml"
 BUTLER_CONFIG="embargo"
 INPUT_COLLECTIONS="LSSTCam/defaults,LSSTCam/templates,LSSTCam/runs/prompt-${DAY_OBS}"

tests/test_pipelines.py

@@ -40,6 +40,7 @@ def setUp(self):
         # Each pipeline file should have a subset that represents it in
         # higher-level pipelines.
         self.synonyms = {"ApPipe.yaml": "apPipe",
+                         "ApPipeDaytime.yaml": "apPipe",
                          "ApPipeWithIsrTaskLSST.yaml": "apPipe",
                          "ApPipeWithPreconvolution.yaml": "apPipe",
                          "ApPipeWithFakes.yaml": "apPipe",