Merge pull request #194 from lsst-sqre/tickets/DM-52588

DM-52588: Add initial implementation of timeseries join on `visit` and `detector`

Author: JeremyMcCormick

changelog.d/20250924_133732_jeremym_DM_52588.md

@@ -0,0 +1,6 @@
+### New features
+
+- Added support for joining on ``visit`` and ``detector`` columns when
+  constructing time series using ``join_style`` parameter, which can be
+  either ``ccdVisit`` (the original scheme) or ``visit_detector`` for
+  the new one.

src/datalinker/handlers/external.py

@@ -160,6 +160,9 @@ async def timeseries(
             pattern=ADQL_FOREIGN_COLUMN_REGEX,
         ),
     ] = None,
+    join_style: Annotated[
+        Literal["ccdVisit", "visit_detector"], Query(title="Join style")
+    ] = "ccdVisit",
     tap_metadata: Annotated[TAPMetadata, Depends(tap_metadata_dependency)],
     logger: Annotated[BoundLogger, Depends(logger_dependency)],
 ) -> str:
@@ -169,9 +172,13 @@ async def timeseries(
     # In those cases we have to join with another table on ccdVisitId.
     if join_time_column:
         join_table, time_column = join_time_column.rsplit(".", 1)
+        if join_style == "visit_detector":
+            join_clause = "(s.visit = t.visitId AND s.detector = t.detector)"
+        else:
+            join_clause = "s.ccdVisitId = t.ccdVisitId"
         adql = (
             f"SELECT t.{time_column},{columns} FROM {table} AS s"
-            f" JOIN {join_table} AS t ON s.ccdVisitId = t.ccdVisitId"
+            f" JOIN {join_table} AS t ON {join_clause}"
         )
     else:
         adql = f"SELECT {columns} FROM {table} AS s"

tests/data/columns-principal.yaml

@@ -306,6 +306,29 @@ tables:
     tap:principal: []
   dp02_test_PREOPS863_00.Visit:
     tap:principal: []
+  dp1.CcdVisit:
+    tap:principal:
+    - band
+    - dec
+    - detector
+    - expTime
+    - magLim
+    - psfSigma
+    - ra
+    - seeing
+    - zenithDistance
+    - visitId
+    - expMidptMJD
+  dp1.ForcedSource:
+    tap:principal:
+    - band
+    - detector
+    - objectId
+    - psfDiffFlux
+    - psfDiffFluxErr
+    - psfFlux
+    - psfFluxErr
+    - visit
   ivoa.ObsCore:
     tap:principal:
     - dataproduct_type

tests/handlers/external_test.py

@@ -189,6 +189,67 @@ async def test_timeseries_detail(client: AsyncClient) -> None:
     }
 
 
+@pytest.mark.asyncio
+async def test_timeseries_join_style(client: AsyncClient) -> None:
+    r = await client.get(
+        "/api/datalink/timeseries",
+        params={
+            "id": 18446744073709551617,
+            "table": "dp1.ForcedSource",
+            "id_column": "diaObjectId",
+            "detail": "principal",
+            "join_time_column": "dp1.CcdVisit.expMidptMJD",
+            "join_style": "ccdVisit",
+        },
+    )
+    assert r.status_code == 307
+    url = urlparse(r.headers["Location"])
+    assert url.path == "/api/tap/sync"
+    query = parse_qs(url.query)
+    assert query == {
+        "LANG": ["ADQL"],
+        "REQUEST": ["doQuery"],
+        "QUERY": [
+            (
+                "SELECT t.expMidptMJD,s.band,s.detector,s.objectId,"
+                "s.psfDiffFlux,s.psfDiffFluxErr,s.psfFlux,s.psfFluxErr,s.visit"
+                " FROM dp1.ForcedSource"
+                " AS s JOIN dp1.CcdVisit AS t ON s.ccdVisitId = t.ccdVisitId"
+                " WHERE s.diaObjectId = 18446744073709551617"
+            )
+        ],
+    }
+
+    r = await client.get(
+        "/api/datalink/timeseries",
+        params={
+            "id": 18446744073709551617,
+            "table": "dp1.ForcedSource",
+            "id_column": "diaObjectId",
+            "detail": "principal",
+            "join_time_column": "dp1.CcdVisit.expMidptMJD",
+            "join_style": "visit_detector",
+        },
+    )
+    assert r.status_code == 307
+    url = urlparse(r.headers["Location"])
+    assert url.path == "/api/tap/sync"
+    query = parse_qs(url.query)
+    assert query == {
+        "LANG": ["ADQL"],
+        "REQUEST": ["doQuery"],
+        "QUERY": [
+            (
+                "SELECT t.expMidptMJD,s.band,s.detector,s.objectId,"
+                "s.psfDiffFlux,s.psfDiffFluxErr,s.psfFlux,s.psfFluxErr,s.visit"
+                " FROM dp1.ForcedSource AS s JOIN dp1.CcdVisit AS t"
+                " ON (s.visit = t.visitId AND s.detector = t.detector)"
+                " WHERE s.diaObjectId = 18446744073709551617"
+            )
+        ],
+    }
+
+
 @pytest.mark.asyncio
 async def test_links(
     client: AsyncClient, mock_butler: MockButler, mock_discovery: Discovery