fix(cadastral): single-statement match UPDATE — OFFSET chunk skipped rows

The chunked matcher paged the candidate set with OFFSET over the very predicate
it mutates (building_cadastral_number IS NULL). Matched rows drop out, so OFFSET
(reset to 0 OR advanced) skips un-processed rows: once >= batch_size unmatched
listings (beyond threshold) accumulate at the low-id front, the chunk matches 0
and the loop breaks early, leaving higher-id matchable listings unfilled.
Counterexample (batch=3, matchable ids 1,4,7,10): matches 1 and 4 then breaks,
missing 7 and 10. Unit tests used single-chunk fixtures so never caught it.

The full UPDATE is ~5s for ~41k listings, so replace the loop with ONE set-based
LATERAL KNN UPDATE over all candidates — correct and fast. batch_size kept in the
signature for schedule-param compat (now unused).

tradein-mvp/backend/app/tasks/cadastral_geo_match.py

@@ -122,41 +122,36 @@ def refresh_cad_buildings_local(db: Session) -> int:
 # LATERAL UPDATE only over that chunk — bounds lock/WAL footprint on the ~43k table and lets
 # the run loop checkpoint a heartbeat between chunks. only_missing=true makes each chunk drain
 # monotonically (filled rows drop out of the candidate set), so the loop terminates.
-_MATCH_CHUNK_SQL = text(
+# Single set-based LATERAL KNN UPDATE over ALL candidates in one statement.
+# NOT chunked by OFFSET: an OFFSET walk over the `building_cadastral_number IS NULL`
+# predicate is unsound because the UPDATE mutates that very predicate mid-walk —
+# matched rows drop out of the candidate set, so OFFSET (reset OR advance) skips
+# un-processed rows. The full UPDATE is ~5s for ~41k listings (GIST KNN, measured),
+# so a single statement is both correct and fast enough; no chunking needed.
+_MATCH_SQL = text(
     """
-    WITH chunk AS (
-        SELECT l.id, l.geom AS l_geom
+    WITH matched AS (
+        SELECT l.id AS id, m.cad_num AS cad_num
         FROM listings l
+        JOIN LATERAL (
+            SELECT cb.cad_num,
+                   ST_DistanceSphere(cb.geom, l.geom) AS dist_m
+            FROM cad_buildings_local cb
+            WHERE ST_DWithin(cb.geom, l.geom, CAST(:deg_gate AS double precision))
+            ORDER BY cb.geom <-> l.geom
+            LIMIT 1
+        ) m ON true
         WHERE l.geom IS NOT NULL
           AND (
               CAST(:only_missing AS boolean) = false
               OR l.building_cadastral_number IS NULL
           )
-        ORDER BY l.id
-        LIMIT CAST(:batch_size AS int)
-        OFFSET CAST(:offset AS int)
-    ),
-    matched AS (
-        SELECT c.id, m.cad_num
-        FROM chunk c
-        JOIN LATERAL (
-            SELECT cb.cad_num,
-                   ST_DistanceSphere(cb.geom, c.l_geom) AS dist_m
-            FROM cad_buildings_local cb
-            WHERE ST_DWithin(cb.geom, c.l_geom, CAST(:deg_gate AS double precision))
-            ORDER BY cb.geom <-> c.l_geom
-            LIMIT 1
-        ) m ON true
-        WHERE m.dist_m <= CAST(:threshold_m AS double precision)
+          AND m.dist_m <= CAST(:threshold_m AS double precision)
     )
     UPDATE listings l
     SET building_cadastral_number = matched.cad_num
     FROM matched
     WHERE l.id = matched.id
-      AND (
-          CAST(:only_missing AS boolean) = false
-          OR l.building_cadastral_number IS NULL
-      )
     """
 )
 
@@ -199,52 +194,31 @@ def match_listings_to_buildings(
     GEO-NEAREST (approximate): each candidate listing is matched to the nearest
     cad_buildings_local building within `threshold_m` metres. The threshold is logged.
 
-    Chunked by listing id to bound lock/WAL. With only_missing=true the candidate set
-    shrinks each chunk (filled rows drop out), so we iterate with a moving OFFSET reset to 0
-    after every successful chunk (the WHERE re-filters). Returns total listings matched.
+    Single set-based UPDATE over ALL candidates (no chunking). An OFFSET walk would be
+    unsound: the UPDATE mutates the `building_cadastral_number IS NULL` predicate it pages
+    over, so matched rows drop out and OFFSET (reset OR advance) skips un-processed rows.
+    The full UPDATE is ~5s for ~41k listings (GIST KNN), so one statement is correct and
+    fast. `batch_size` is accepted for schedule-param compatibility but no longer used.
+    Returns listings matched.
     """
     logger.info(
-        "match_listings_to_buildings: GEO-NEAREST match threshold_m=%d batch_size=%d "
+        "match_listings_to_buildings: GEO-NEAREST match threshold_m=%d "
         "only_missing=%s (approximate: nearest building, not exact cadastral)",
         threshold_m,
-        batch_size,
         only_missing,
     )
 
     deg_gate = _deg_gate_for(float(threshold_m))
-    total_matched = 0
-    offset = 0
-    while True:
-        result = db.execute(
-            _MATCH_CHUNK_SQL,
-            {
-                "threshold_m": float(threshold_m),
-                "deg_gate": deg_gate,
-                "batch_size": batch_size,
-                "offset": offset,
-                "only_missing": only_missing,
-            },
-        )
-        matched = result.rowcount
-        db.commit()
-        total_matched += matched
-
-        if only_missing:
-            # Filled rows drop out of the candidate set → next chunk starts fresh at 0.
-            # Terminate when a chunk matches nothing (no more matchable candidates).
-            if matched == 0:
-                break
-            offset = 0
-        else:
-            # Re-match all: walk the id space once via OFFSET; stop when a full pass
-            # over the candidate window yielded a short chunk (no rows updated).
-            offset += batch_size
-            if matched == 0:
-                break
-
-        logger.info(
-            "match_listings_to_buildings: chunk matched=%d total=%d", matched, total_matched
-        )
+    result = db.execute(
+        _MATCH_SQL,
+        {
+            "threshold_m": float(threshold_m),
+            "deg_gate": deg_gate,
+            "only_missing": only_missing,
+        },
+    )
+    total_matched = result.rowcount
+    db.commit()
 
     logger.info(
         "match_listings_to_buildings: DONE total_matched=%d threshold_m=%d",

tradein-mvp/backend/tests/tasks/test_cadastral_geo_match.py

@@ -37,7 +37,7 @@ _SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql"
 _MIGRATION_124 = _SQL_DIR / "124_cad_buildings_local.sql"
 _MIGRATION_125 = _SQL_DIR / "125_scrape_schedules_seed_cadastral_geo_match.sql"
 
-_MATCH_SQL = str(cgm._MATCH_CHUNK_SQL.text)
+_MATCH_SQL = str(cgm._MATCH_SQL.text)
 _CANDIDATES_SQL = str(cgm._CANDIDATES_SQL.text)
 _REFRESH_SRC = inspect.getsource(cgm.refresh_cad_buildings_local)
 _MATCH_SRC = inspect.getsource(cgm.match_listings_to_buildings)
@@ -72,7 +72,7 @@ def test_match_uses_lateral_knn_nearest_one() -> None:
     flat = re.sub(r"\s+", " ", _MATCH_SQL)
     assert "JOIN LATERAL" in flat
     # GIST-backed KNN order + single nearest candidate.
-    assert "ORDER BY cb.geom <-> c.l_geom" in flat
+    assert "ORDER BY cb.geom <-> l.geom" in flat
     assert "LIMIT 1" in flat
 
 
@@ -81,11 +81,11 @@ def test_match_gates_with_geometry_dwithin_and_distancesphere_recheck() -> None:
     metric recheck — NOT a geography-cast ST_DWithin in the WHERE (that ran 58s+ unfinished)."""
     flat = re.sub(r"\s+", " ", _MATCH_SQL)
     # GIST-indexable degree-space gate (no geography cast on the filtered table).
-    assert "ST_DWithin(cb.geom, c.l_geom, CAST(:deg_gate AS double precision))" in flat
+    assert "ST_DWithin(cb.geom, l.geom, CAST(:deg_gate AS double precision))" in flat
     # The geography cast must NOT appear in the gate (that defeats the index).
     assert "CAST(cb.geom AS geography)" not in flat
     # True metric distance computed only on the single nearest row, gated by threshold_m.
-    assert "ST_DistanceSphere(cb.geom, c.l_geom) AS dist_m" in flat
+    assert "ST_DistanceSphere(cb.geom, l.geom) AS dist_m" in flat
     assert "m.dist_m <= CAST(:threshold_m AS double precision)" in flat
 
 
@@ -104,10 +104,13 @@ def test_match_only_missing_param_is_bound_and_filters() -> None:
     assert "l.geom IS NOT NULL" in flat
 
 
-def test_match_is_chunked() -> None:
+def test_match_is_single_statement_no_offset() -> None:
+    # OFFSET-paging over the mutated `building_cadastral_number IS NULL` predicate skips
+    # un-processed rows — the matcher must be ONE set-based UPDATE, no OFFSET/LIMIT paging.
     flat = re.sub(r"\s+", " ", _MATCH_SQL)
-    assert "LIMIT CAST(:batch_size AS int)" in flat
-    assert "OFFSET CAST(:offset AS int)" in flat
+    assert "OFFSET" not in flat
+    assert "ST_DistanceSphere" in flat
+    assert "UPDATE listings l" in flat
 
 
 def test_deg_gate_encloses_threshold() -> None:
@@ -222,18 +225,13 @@ class _FakeDB:
         pass
 
 
-def test_match_loop_terminates_and_counts(monkeypatch: pytest.MonkeyPatch) -> None:
-    """only_missing loop: chunks match 3, then 2, then 0 → total 5, stops on empty chunk."""
-    db = _FakeDB(
-        [
-            _FakeResult(3),  # chunk 1
-            _FakeResult(2),  # chunk 2
-            _FakeResult(0),  # chunk 3 → terminate
-        ]
-    )
+def test_match_runs_single_update_and_counts(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Single set-based UPDATE: one execute → rowcount is the match count, one commit."""
+    db = _FakeDB([_FakeResult(5)])
     total = cgm.match_listings_to_buildings(db, threshold_m=50, batch_size=10, only_missing=True)
     assert total == 5
-    assert db.commits == 3  # one per chunk
+    assert db.commits == 1
+    assert len(db.executed) == 1  # exactly one UPDATE — no OFFSET paging
 
 
 def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -> None: