From 85e5f9583854c45bc9f4bc139d2fb3fbbd323c75 Mon Sep 17 00:00:00 2001 From: oliver Date: Thu, 8 Oct 2026 16:09:15 +0800 Subject: [PATCH] Fix SQL compare stall: detach progress commits and skip small sheets. Co-authored-by: Cursor --- netx_api/biz_state/compare_service.py | 68 +++++++++++++++++++++++---- netx_api/biz_state/compare_sql.py | 35 ++++++++++++-- 2 files changed, 90 insertions(+), 13 deletions(-) diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index 57bcbe5..a0b6c93 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -1794,7 +1794,7 @@ def _run_sheet( on_load_progress(side, n) # PostgreSQL path: pushdown-safe sheets join in-DB (BGP-scale). - from .compare_sql import run_sql_sheet_compare, sql_compare_skip_reason + from .compare_sql import SqlCompareSkip, run_sql_sheet_compare, sql_compare_skip_reason skip_reason = sql_compare_skip_reason( db, @@ -1834,6 +1834,18 @@ def _run_sheet( diffs=list(result.get("diffs") or []), mapping_stats=dict(result.get("mapping_stats") or {}), ) + except SqlCompareSkip as skip: + skip_reason = skip.reason or "skip" + _log.info( + "python compare sheet=%s metric=%s skip_sql=%s", + sheet_key(sheet), + mid, + skip_reason, + ) + try: + db.rollback() + except Exception: + pass except Exception: _log.exception( "sql compare fallback sheet=%s metric=%s — using Python engine", @@ -1841,6 +1853,11 @@ def _run_sheet( mid, ) skip_reason = "sql_error_fallback" + # Roll back aborted SQL transaction so Python path can use the session + try: + db.rollback() + except Exception: + pass else: _log.info( "python compare sheet=%s metric=%s skip_sql=%s", @@ -2002,9 +2019,14 @@ def _set_run_progress( sheet: dict[str, Any] | None, started_mono: float, extra: dict[str, Any] | None = None, + detach: bool = False, ) -> None: + """Update run progress. + + ``detach=True`` writes via a fresh session so SQL compare can keep an open + transaction (TEMP CTAS) without mid-flight commits on the worker ``db``. + """ elapsed_ms = int((time.monotonic() - started_mono) * 1000) - prev = dict(run.summary_json or {}) progress = { "phase": phase, "sheet_index": sheet_index, @@ -2015,6 +2037,39 @@ def _set_run_progress( } if extra: progress.update(extra) + title = progress["sheet_title"] or progress["sheet_id"] or "" + message = ( + f"{phase} {sheet_index}/{sheet_total}" + + (f" · {title}" if title else "") + + f" · {elapsed_ms // 1000}s" + )[:1024] + + if detach: + from ..db import SessionLocal + + s = SessionLocal() + try: + r = s.get(BizCompareRun, str(run.id)) + if not r: + return + prev = dict(r.summary_json or {}) + prev["progress"] = progress + r.summary_json = prev + if str(r.status or "") != "cancelled": + r.status = "running" + r.message = message + s.commit() + # Mirror into worker instance for later in-memory reads (do not commit db) + prev_w = dict(run.summary_json or {}) + prev_w["progress"] = progress + run.summary_json = prev_w + if str(run.status or "") != "cancelled": + run.message = message + finally: + s.close() + return + + prev = dict(run.summary_json or {}) prev["progress"] = progress run.summary_json = prev # Re-read status from DB — cancel may have been committed by another session @@ -2022,12 +2077,7 @@ def _set_run_progress( db.expire(run, ["status", "message"]) if str(run.status or "") != "cancelled": run.status = "running" - title = progress["sheet_title"] or progress["sheet_id"] or "" - run.message = ( - f"{phase} {sheet_index}/{sheet_total}" - + (f" · {title}" if title else "") - + f" · {elapsed_ms // 1000}s" - )[:1024] + run.message = message db.commit() @@ -2146,6 +2196,7 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: } if note: extra["engine_note"] = str(note)[:128] + # SQL path: detach progress commits so TEMP CTAS stays in one txn _set_run_progress( db, run, @@ -2155,6 +2206,7 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: sheet=_sheet, started_mono=started_mono, extra=extra, + detach=(eng == "sql"), ) _set_run_progress( diff --git a/netx_api/biz_state/compare_sql.py b/netx_api/biz_state/compare_sql.py index 7df648b..52cb85a 100644 --- a/netx_api/biz_state/compare_sql.py +++ b/netx_api/biz_state/compare_sql.py @@ -18,7 +18,20 @@ from .compare_rules import effective_compare_fields, field_rule_map _log = logging.getLogger("netx.biz_state.compare_sql") +# Below this, Python hash-join is faster and avoids TEMP CTAS / progress races. +_SQL_MIN_ROWS = 20_000 +# Cap a single SQL statement so a stuck planner/lock surfaces as fallback. +_SQL_STATEMENT_TIMEOUT_MS = 180_000 + _FIELD_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +class SqlCompareSkip(Exception): + """Soft skip — caller should use the Python engine (not an error).""" + + def __init__(self, reason: str) -> None: + super().__init__(reason) + self.reason = str(reason or "skip") _SQL_FILTER_OPS = frozenset( {"eq", "==", "ne", "!=", "in", "not_in", "nin", "contains", "empty", "not_empty", "nonempty", "ci_eq"} ) @@ -381,6 +394,9 @@ def run_sql_sheet_compare( tb = f"_netx_cmp_b_{tag}" ta = f"_netx_cmp_a_{tag}" + # Keep worker transaction open across CTAS — progress must NOT commit this session. + db.execute(text(f"SET LOCAL statement_timeout = '{int(_SQL_STATEMENT_TIMEOUT_MS)}'")) + _prog("before", 0, phase="sql_count", note="count") # Raw counts (no row_filters) @@ -407,13 +423,20 @@ def run_sql_sheet_compare( ) _prog("after", raw_a, phase="sql_count") + if max(raw_b, raw_a) < int(_SQL_MIN_ROWS): + raise SqlCompareSkip("small_sheet") + base_params = {"bid": bid_b, "mid": mid, **filter_params} - # Build TEMP sides - for tname, batch_id, side in ((tb, bid_b, "before"), (ta, bid_a, "after")): + # Build TEMP sides (one transaction — no mid-flight commits on ``db``) + _prog("before", raw_b, phase="sql_project", note="temp_before") + for tname, batch_id, side, note in ( + (tb, bid_b, "before", "temp_before"), + (ta, bid_a, "after", "temp_after"), + ): db.execute(text(f"DROP TABLE IF EXISTS {tname}")) params = {**base_params, "bid": batch_id} - _prog(side, raw_b if side == "before" else raw_a, phase="sql_project", note="temp") - # PRESERVE ROWS: compare progress commits must not drop temps mid-run + if side == "after": + _prog(side, raw_a, phase="sql_project", note=note) db.execute( text( f""" @@ -434,7 +457,9 @@ def run_sql_sheet_compare( ), params, ) - db.execute(text(f"CREATE INDEX ON {tname} (rk) WHERE dup_rn = 1")) + db.execute( + text(f"CREATE INDEX IF NOT EXISTS {tname}_rk ON {tname} (rk) WHERE dup_rn = 1") + ) before_n = int( db.execute(text(f"SELECT count(*) FROM {tb}")).scalar() or 0