fix(syntax-receipts): import the first Snowflake batch's runs faithfully

docs/syntax-coverage.md

@@ -168,7 +168,21 @@ that edits or removes a committed one.
 - `outcome`, `errno` and `sqlstate`.
 - `phase`: one of `accepted`, `syntax_error`, `semantic_error`, `runtime_error` or
   `unsupported_feature`.
-- `phase_basis`: why the importer chose the phase.
+- `phase_basis`: why the importer chose the phase. The importer applies these rules in order:
+  1. An uncaught exception raised while a block ran is a `runtime_error`, unless its errno is
+     1003, which is refused.
+  2. Errno 2 `Unsupported feature` is an `unsupported_feature`.
+  3. An errno of 100000 or more is a `runtime_error`, unless its SQLSTATE is in class 42. Then it
+     is refused unless the errno is a reviewed one, which is a `semantic_error`.
+  4. Errno 1003 under the `SQL compilation error` prefix is a `syntax_error`. Any other errno
+     under that prefix is a `semantic_error`, a compilation error after parsing.
+  5. Any other error is refused unless its errno is a reviewed one, which is a `semantic_error`.
+
+  A reviewed errno is one whose message shows that Snowflake parsed the statement and then
+  refused it. Examples are a duplicate secret name (509008), a missing object to undrop (90025),
+  access control (3001), a required property (3172), an account restriction (93171) and an
+  application-only operation (98620). Internal errors (603), statement timeouts (630) and errors
+  that may come from the parser itself stay refused.
 - `raw_record_sha256`, `run_audit_sha256` and `packet_sha256`: the hashes of the private raw
   record, the private run audit and the submitted packet.
 - Optional provenance: `case_hash`, `input_sha256`, `statement_index`, `statements_in_input`,

scripts/syntax_receipts/_helpers/audit.py

@@ -170,6 +170,15 @@ def error_fields(*, raw: Row) -> tuple[int | None, str | None, str]:
     return fields.get("errno"), fields.get("sqlstate"), fields.get("raw_msg") or ""
 
 
+def client_failure(*, raw: Row) -> str | None:
+    """Why a raw record holds no answer from Snowflake: the connector failed before one arrived."""
+    fields: Row = raw.get("connector_fields") or {}
+    if fields.get("case_result") is not False:
+        return None
+    error: str = f"{fields.get('exception_class')}, errno {fields.get('errno')}"
+    return f"client-side connector failure ({error}); Snowflake's answer was not observed"
+
+
 def connector_query_id(*, raw: Row) -> str | None:
     """The query id the connector reported for a raw record."""
     return (raw.get("connector_fields") or {}).get("query_id")

scripts/syntax_receipts/_helpers/ddl_run.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 
 from scripts.syntax_receipts._helpers.audit import (
+    client_failure,
     connector_query_id,
     error_fields,
     packet_rows,
@@ -66,7 +67,24 @@ def import_ddl_run(*, run: CaptureRun, index: dict[str, Path]) -> Imported:
                 )
             )
             continue
-        setup: list[str] = _setup(start=start, raw=raw, position=position) + case.get("setup", [])
+        failure: str | None = client_failure(raw=raw)
+        if failure is not None:
+            found.refusals.append(refusal(run_id=run.path.name, row_id=row_id, reason=failure))
+            continue
+        run_setup: list[str] = _setup(start=start, raw=raw, position=position)
+        fixtures: list[str] | None = _fixtures(
+            packet_setup=case.get("setup", []), run_setup=run_setup
+        )
+        if fixtures is None:
+            found.refusals.append(
+                refusal(
+                    run_id=run.path.name,
+                    row_id=row_id,
+                    reason="run template in the packet setup differs from the setup the run made",
+                )
+            )
+            continue
+        setup: list[str] = run_setup + fixtures
         exact: Row = input_row(sql=case["sql"], setup=setup)
         found.inputs.setdefault(exact["sql_sha256"] + exact["setup_sha256"], exact)
         errno, sqlstate, message = error_fields(raw=raw)
@@ -214,6 +232,15 @@ def _text(value: object) -> bool:
     return isinstance(value, str) and bool(value)
 
 
+def _fixtures(*, packet_setup: list[str], run_setup: list[str]) -> list[str] | None:
+    """The packet setup after the run setup it may repeat; None for another run's template."""
+    if packet_setup[: len(run_setup)] == run_setup:
+        return packet_setup[len(run_setup) :]
+    if packet_setup and packet_setup[0].startswith(f"CREATE DATABASE {RUN_DATABASE}"):
+        return None
+    return packet_setup
+
+
 def _setup(*, start: Row, raw: Row, position: int) -> list[str]:
     database: str = start["run_database"]
     nonce: str = database.rsplit("_", 1)[-1]

scripts/syntax_receipts/_helpers/packet_run.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 
 from scripts.syntax_receipts._helpers.audit import (
+    client_failure,
     connector_query_id,
     error_fields,
     packet_rows,
@@ -43,6 +44,10 @@ def import_packet_run(*, run: CaptureRun, index: dict[str, Path]) -> Imported:
         except RowRefused as refused:
             found.refusals.append(refusal(run_id=run.path.name, row_id=row_id, reason=str(refused)))
             continue
+        failure: str | None = client_failure(raw=raw)
+        if failure is not None:
+            found.refusals.append(refusal(run_id=run.path.name, row_id=row_id, reason=failure))
+            continue
         sql: Row = input_row(sql=case["sql"], setup=[])
         found.inputs.setdefault(sql["sql_sha256"] + sql["setup_sha256"], sql)
         errno, sqlstate, message = error_fields(raw=raw)

scripts/syntax_receipts/_helpers/receipts.py

@@ -48,6 +48,8 @@ def classify(
         return "syntax_error", "errno 1003 SQL compilation syntax error"
     if errno is not None and compiled:
         return "semantic_error", f"errno {errno} compilation error after parsing"
+    if errno in REVIEWED_SEMANTIC_ERRNOS:
+        return "semantic_error", f"errno {errno} SQLSTATE {sqlstate}, reviewed as semantic"
     return None
 
 

scripts/syntax_receipts/constants.py

@@ -13,7 +13,25 @@
 UNSUPPORTED_PREFIX: str = "Unsupported feature"
 RUNTIME_ERRNO_FLOOR: int = 100000
 SYNTAX_OR_ACCESS_SQLSTATE_CLASS: str = "42"
-REVIEWED_SEMANTIC_ERRNOS: frozenset[int] = frozenset({391858, 392002, 506005, 506034, 506035})
+REVIEWED_SEMANTIC_ERRNOS: frozenset[int] = frozenset(
+    {
+        3001,
+        3172,
+        90025,
+        93171,
+        98620,
+        391858,
+        392002,
+        393941,
+        506005,
+        506034,
+        506035,
+        509000,
+        509001,
+        509008,
+        509013,
+    }
+)
 SCRIPTING_SUBMISSION: str = "anonymous block, as-is"
 EXPLAIN_METHOD: str = "EXPLAIN USING TEXT, once"
 CAPTURE_EXECUTE: str = "execute"
@@ -65,6 +83,7 @@
             2142,
             2211,
             3620,
+            90025,
             90080,
             91083,
             91085,

tests/unit/scripts/syntax_receipts/_test_types.py

@@ -63,7 +63,7 @@ class SetupRunTestCase:
     packet: list[dict[str, object]]
     start: dict[str, object]
     expected_receipts: list[tuple[str, str, str]]
-    expected_setup_tail: list[str]
+    expected_setup: list[str]
     expected_refusals: list[str]
 
 

tests/unit/scripts/syntax_receipts/test_import_runs.py

@@ -56,9 +56,37 @@
     packet_case(case_id="c3", sql=_BROKEN),
 ]
 _FIXTURE: list[str] = ['CREATE TABLE "ORDERS" ("ID" INTEGER)']
+_DDL_TEMPLATE: list[str] = [
+    "CREATE DATABASE <run_database> COMMENT = 'sqlquack ddl sweep <run_nonce>'",
+    "CREATE SCHEMA <run_database>.CASES",
+    "USE SCHEMA <run_database>.CASES",
+    "CREATE SCHEMA <run_database>.<case_schema>",
+    "USE SCHEMA <run_database>.<case_schema>",
+]
+# Fixtures after the template that repeat each other and a template line, all kept.
+_FIXTURES_AFTER: list[str] = [*_FIXTURE, *_FIXTURE, "USE SCHEMA <run_database>.CASES"]
+_DDL_START: dict[str, object] = {
+    "run_database": "SQLQUACK_DDL_ABC123",
+    "setup": [
+        "CREATE DATABASE SQLQUACK_DDL_ABC123 COMMENT = 'sqlquack ddl sweep ABC123'",
+        "CREATE SCHEMA SQLQUACK_DDL_ABC123.CASES",
+        "USE SCHEMA SQLQUACK_DDL_ABC123.CASES",
+    ],
+}
 _SYNTAX_ROW: dict[str, object] = error_row(
     case_id="c1", sql=_GOOD, errno=1003, message="SQL compilation error:\nsyntax error"
 )
+# The connector's own fields for a read timeout: no SQLSTATE class, so the parent code read errno 250003 as an
+# execution-time code and imported a runtime_error receipt.
+_TIMED_OUT_ROW: dict[str, object] = replaced(
+    replaced(
+        error_row(case_id="c2", sql=_OTHER, errno=250003, message="Read timed out."),
+        ("connector_fields", "case_result"),
+        False,
+    ),
+    ("connector_fields", "sqlstate"),
+    "n/a",
+)
 
 
 @pytest.mark.parametrize(
@@ -124,6 +152,37 @@
                 "errno 391858 SQLSTATE 42601, reviewed as semantic",
             ),
         ),
+        ClassifyTestCase(
+            description="a reviewed errno without the compilation prefix is semantic",
+            errno=90025,
+            sqlstate="02000",
+            message="Dynamic_table V1 did not exist or was purged.",
+            expected_classification=(
+                "semantic_error",
+                "errno 90025 SQLSTATE 02000, reviewed as semantic",
+            ),
+        ),
+        ClassifyTestCase(
+            description="an internal error is not established",
+            errno=603,
+            sqlstate="XX000",
+            message="SQL execution internal error:\nProcessing aborted due to error 300010:1",
+            expected_classification=None,
+        ),
+        ClassifyTestCase(
+            description="a statement timeout is not established",
+            errno=630,
+            sqlstate="57014",
+            message="Statement reached its statement or warehouse timeout of 30 second(s).",
+            expected_classification=None,
+        ),
+        ClassifyTestCase(
+            description="an unreviewed errno without the compilation prefix is not established",
+            errno=926,
+            sqlstate="42601",
+            message="External table identifier cannot be empty",
+            expected_classification=None,
+        ),
         ClassifyTestCase(
             description="a syntax error without the compilation prefix is not established",
             errno=1003,
@@ -203,14 +262,31 @@ def test_given_non_syntax_errors_when_categorized_then_snowflake_reason_is_named
                 {"id": "c2", "sql_sha256": sha(b"something else"), "outcome": "accepted_executed"},
                 accepted_row(case_id="c9", sql=_GOOD),
                 error_row(case_id="c3", sql=_OTHER, errno=1003, message="no prefix"),
+                replaced(
+                    replaced(
+                        error_row(
+                            case_id="c4", sql=_OTHER, errno=250003, message="Read timed out."
+                        ),
+                        ("connector_fields", "case_result"),
+                        False,
+                    ),
+                    ("connector_fields", "sqlstate"),
+                    "n/a",
+                ),
             ],
             packet=[
                 packet_case(case_id="c1", sql=_GOOD),
                 packet_case(case_id="c2", sql=_OTHER),
                 packet_case(case_id="c3", sql=_OTHER),
+                packet_case(case_id="c4", sql=_OTHER),
             ],
             expected_receipts=[("c1", "syntax_error")],
-            expected_refusals=[("c2", "raw"), ("c3", "error"), ("c9", "row")],
+            expected_refusals=[
+                ("c2", "raw"),
+                ("c3", "error"),
+                ("c4", "client-side"),
+                ("c9", "row"),
+            ],
             expected_sql_sha256=sha(_GOOD.encode()),
             expected_raw_record_sha256=sha(json.dumps(_SYNTAX_ROW).encode()),
         ),
@@ -344,13 +420,28 @@ def test_given_receipt_published_before_a_field_was_added_when_imported_again_th
                 "measure": "missing_object_with_setup",
             },
             expected_receipts=[("c1", "accepted", "compile_only_explain")],
-            expected_setup_tail=[
+            expected_setup=[
+                "CREATE DATABASE <run_database>",
                 "CREATE SCHEMA <run_database>.<case_schema>",
                 "USE SCHEMA <run_database>.<case_schema>",
                 *_FIXTURE,
             ],
             expected_refusals=["c2"],
         ),
+        SetupRunTestCase(
+            description="a packet that repeats the run's template is not counted twice",
+            rows=[{**_SYNTAX_ROW, "case_schema": "C0000"}],
+            packet=[
+                {
+                    **packet_case(case_id="c1", sql=_GOOD),
+                    "setup": [*_DDL_TEMPLATE, *_FIXTURES_AFTER],
+                }
+            ],
+            start=_DDL_START,
+            expected_receipts=[("c1", "syntax_error", "execute_ddl_in_run_database")],
+            expected_setup=[*_DDL_TEMPLATE, *_FIXTURES_AFTER],
+            expected_refusals=[],
+        ),
     ),
     ids=lambda case: case.description,
 )
@@ -373,9 +464,79 @@ def test_given_fixture_setup_per_item_when_imported_then_receipts_name_that_setu
         test_case.expected_receipts
     )
     receipt_input: tuple[str, str] = (receipts[0]["sql_sha256"], receipts[0]["setup_sha256"])
-    assert setups[receipt_input][-3:] == test_case.expected_setup_tail
+    assert setups[receipt_input] == test_case.expected_setup
     assert [r["row_id"] for r in report["refusals"]] == test_case.expected_refusals
-    assert receipts[0]["measure"] == test_case.start["measure"]
+    assert receipts[0].get("measure") == test_case.start.get("measure")
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        BindingTestCase(
+            description="a client timeout in a run database is refused, not a runtime error",
+            rows=[
+                {**_SYNTAX_ROW, "case_schema": "C0000"},
+                {**_TIMED_OUT_ROW, "case_schema": "C0001"},
+            ],
+            packet=[packet_case(case_id="c1", sql=_GOOD), packet_case(case_id="c2", sql=_OTHER)],
+            expected_receipts=[("c1", "syntax_error")],
+            expected_refusals=[("c2", "client-side")],
+            expected_sql_sha256=sha(_GOOD.encode()),
+            expected_raw_record_sha256=sha(
+                json.dumps({**_SYNTAX_ROW, "case_schema": "C0000"}).encode()
+            ),
+        ),
+        BindingTestCase(
+            description="a packet carrying another run template is refused, not doubled",
+            rows=[
+                {**_SYNTAX_ROW, "case_schema": "C0000"},
+                {
+                    **error_row(case_id="c2", sql=_OTHER, errno=1003, message="x"),
+                    "case_schema": "C0001",
+                },
+                # No case schema: the run setup ends with the shared-schema note, not the template.
+                error_row(case_id="c3", sql=_BROKEN, errno=1003, message="x"),
+            ],
+            packet=[
+                {**packet_case(case_id="c1", sql=_GOOD), "setup": _DDL_TEMPLATE},
+                {
+                    **packet_case(case_id="c2", sql=_OTHER),
+                    "setup": [
+                        _DDL_TEMPLATE[0].replace("ddl sweep", "admin sweep"),
+                        *_DDL_TEMPLATE[1:],
+                    ],
+                },
+                {**packet_case(case_id="c3", sql=_BROKEN), "setup": _DDL_TEMPLATE},
+            ],
+            expected_receipts=[("c1", "syntax_error")],
+            expected_refusals=[("c2", "run"), ("c3", "run")],
+            expected_sql_sha256=sha(_GOOD.encode()),
+            expected_raw_record_sha256=sha(
+                json.dumps({**_SYNTAX_ROW, "case_schema": "C0000"}).encode()
+            ),
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_run_database_rows_that_cannot_be_bound_when_imported_then_they_are_refused(
+    test_case: BindingTestCase, tmp_path: Path
+) -> None:
+    write_run(
+        root=tmp_path,
+        rows=test_case.rows,
+        packet=test_case.packet,
+        start=_DDL_START,
+    )
+
+    imported: Imported = import_runs(runs=tmp_path / "runs", input_dirs=[tmp_path / "inputs"])
+
+    assert [(r["case_id"], r["phase"]) for r in imported.receipts] == test_case.expected_receipts
+    assert (
+        sorted((r["row_id"], r["reason"].split()[0]) for r in imported.refusals)
+        == test_case.expected_refusals
+    )
+    assert imported.receipts[0]["sql_sha256"] == test_case.expected_sql_sha256
+    assert imported.receipts[0]["raw_record_sha256"] == test_case.expected_raw_record_sha256
 
 
 @pytest.mark.parametrize(