From ab5b539ef94cd8d3cc6c8164d95e76764d0f7c18 Mon Sep 17 00:00:00 2001 From: Vignesh Narayanaswamy Date: Mon, 22 Jun 2026 14:21:42 -0700 Subject: [PATCH 1/2] Skip schema DDL on write-init when tables already exist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SnowflakeLedgerBackend ran CREATE SCHEMA / CREATE TABLE / ALTER on every read_only=False init. Snowflake checks the CREATE/ALTER privilege even for `IF NOT EXISTS`, so a write-enabled deployment whose schema is provisioned out-of-band was forced to hold schema-ownership-level DDL grants just to write rows — otherwise startup failed (or silently fell back to an empty in-memory backend in wrappers that catch the error). Gate the DDL behind a SELECT-only INFORMATION_SCHEMA existence probe: when MODELS/SNAPSHOTS/TAGS (with MODELS.METADATA) already exist, skip the DDL so plain INSERT/UPDATE/SELECT grants suffice. Fresh deployments still auto-provision, and any introspection failure falls back to the CREATE path (no regression for permissioned deployments). Co-Authored-By: Claude Opus 4.8 --- src/model_ledger/backends/snowflake.py | 46 ++++++++++ tests/test_backends/test_snowflake_ledger.py | 97 ++++++++++++++++++++ 2 files changed, 143 insertions(+) diff --git a/src/model_ledger/backends/snowflake.py b/src/model_ledger/backends/snowflake.py index fb1079f..a19e544 100644 --- a/src/model_ledger/backends/snowflake.py +++ b/src/model_ledger/backends/snowflake.py @@ -322,6 +322,15 @@ def _flush_snapshots_sql(self) -> None: ) def _ensure_tables(self) -> None: + # Skip DDL when the schema is already provisioned out-of-band. Snowflake + # checks the CREATE/ALTER privilege even for `IF NOT EXISTS`, so issuing + # this DDL forces a write-enabled deployment to hold CREATE SCHEMA (db) + + # CREATE TABLE (schema) + ALTER (table) — effectively schema ownership — + # when all it needs to write rows is INSERT/UPDATE/SELECT. Existence is + # probed via INFORMATION_SCHEMA (SELECT only); on any doubt we fall + # through to the CREATE path so fresh deployments still auto-provision. + if self._schema_objects_exist(): + return _exec_no_result(self._session, f"CREATE SCHEMA IF NOT EXISTS {self._schema}") _exec_no_result( self._session, @@ -365,6 +374,43 @@ def _ensure_tables(self) -> None: PRIMARY KEY (MODEL_HASH, NAME))""", ) + def _schema_objects_exist(self) -> bool: + """True if MODELS, SNAPSHOTS and TAGS already exist (with MODELS.METADATA). + + Lets a write-enabled deployment whose schema is provisioned externally + run with only INSERT/UPDATE/SELECT — no CREATE/ALTER. Probes are + SELECT-only against INFORMATION_SCHEMA. Returns False (so the caller + falls back to the CREATE path) whenever existence can't be positively + confirmed, preserving auto-provisioning for fresh deployments. + """ + info_schema = ( + f"{self._database}.INFORMATION_SCHEMA" if self._database else "INFORMATION_SCHEMA" + ) + schema_lit = _esc(self._schema_name.upper()) + try: + table_rows = _exec( + self._session, + f"""SELECT TABLE_NAME FROM {info_schema}.TABLES + WHERE TABLE_SCHEMA = {schema_lit} + AND TABLE_TYPE = 'BASE TABLE' + AND TABLE_NAME IN ('MODELS', 'SNAPSHOTS', 'TAGS')""", + ) + present = {str(r["TABLE_NAME"]).upper() for r in table_rows} + if not {"MODELS", "SNAPSHOTS", "TAGS"} <= present: + return False + # MODELS must carry the METADATA column (the backward-compat ALTER + # target); without it, fall through so the migration DDL still runs. + column_rows = _exec( + self._session, + f"""SELECT 1 FROM {info_schema}.COLUMNS + WHERE TABLE_SCHEMA = {schema_lit} + AND TABLE_NAME = 'MODELS' + AND COLUMN_NAME = 'METADATA'""", + ) + return len(column_rows) > 0 + except Exception: + return False + def save_model(self, model: ModelRef) -> None: self._model_buffer.append(model) diff --git a/tests/test_backends/test_snowflake_ledger.py b/tests/test_backends/test_snowflake_ledger.py index e76fcde..2c1878f 100644 --- a/tests/test_backends/test_snowflake_ledger.py +++ b/tests/test_backends/test_snowflake_ledger.py @@ -623,3 +623,100 @@ def test_parity_with_in_memory_fallback(self): ] backend, _ = self._backend(rows) assert backend.composite_summary() == expected + + +class _DDLProbeSession: + """Session that records DDL and serves configurable INFORMATION_SCHEMA probes. + + Lets tests assert whether write-mode init issues CREATE/ALTER, and simulate a + role that lacks DDL privileges (`raise_on_ddl`) or can't introspect + (`raise_on_introspect`). + """ + + def __init__( + self, + *, + tables_present, + metadata_present=True, + raise_on_ddl=False, + raise_on_introspect=False, + ): + self._tables_present = {t.upper() for t in tables_present} + self._metadata_present = metadata_present + self._raise_on_ddl = raise_on_ddl + self._raise_on_introspect = raise_on_introspect + self.ddl: list[str] = [] + + def sql(self, query: str, params: Any = None) -> MockCollectResult: + upper = query.upper().strip() + if "INFORMATION_SCHEMA.TABLES" in upper: + if self._raise_on_introspect: + raise RuntimeError("introspection blocked") + return MockCollectResult([{"TABLE_NAME": t} for t in sorted(self._tables_present)]) + if "INFORMATION_SCHEMA.COLUMNS" in upper: + if self._raise_on_introspect: + raise RuntimeError("introspection blocked") + return MockCollectResult([{"1": 1}] if self._metadata_present else []) + if upper.startswith("CREATE") or upper.startswith("ALTER"): + self.ddl.append(upper) + if self._raise_on_ddl: + raise RuntimeError( + "003001 (42501): SQL access control error: Insufficient " + "privileges to operate on database. Your primary role must " + "have CREATE SCHEMA granted on DATABASE." + ) + return MockCollectResult([]) + return MockCollectResult([]) + + +def _new_backend(session, *, read_only=False): + from model_ledger.backends.snowflake import SnowflakeLedgerBackend + + return SnowflakeLedgerBackend(schema="DB.MODEL_LEDGER", connection=session, read_only=read_only) + + +def test_skip_ddl_when_schema_already_provisioned(): + # All tables + METADATA present, and the role would fail any DDL. Init must + # succeed without issuing a single CREATE/ALTER (least-privilege path). + session = _DDLProbeSession( + tables_present=["MODELS", "SNAPSHOTS", "TAGS"], + metadata_present=True, + raise_on_ddl=True, + ) + _new_backend(session) # must not raise + assert session.ddl == [] + + +def test_auto_provisions_when_tables_absent(): + # Fresh deployment: nothing exists → full DDL runs (auto-provision preserved). + session = _DDLProbeSession(tables_present=[]) + _new_backend(session) + joined = " ".join(session.ddl) + assert "CREATE SCHEMA" in joined + assert joined.count("CREATE TABLE") == 3 + + +def test_falls_back_to_ddl_when_introspection_fails(): + # If existence can't be confirmed, fall through to the CREATE path rather + # than silently skipping (no regression for permissioned deployments). + session = _DDLProbeSession(tables_present=[], raise_on_introspect=True) + _new_backend(session) + assert any(d.startswith("CREATE SCHEMA") for d in session.ddl) + + +def test_skip_requires_metadata_column(): + # Tables exist but MODELS lacks METADATA → must NOT skip; the migration DDL + # (incl. the ALTER ADD COLUMN) needs to run. + session = _DDLProbeSession( + tables_present=["MODELS", "SNAPSHOTS", "TAGS"], + metadata_present=False, + ) + _new_backend(session) + assert any(d.startswith("CREATE SCHEMA") for d in session.ddl) + + +def test_read_only_skips_ensure_tables_entirely(): + # Read-only init must neither introspect nor issue DDL. + session = _DDLProbeSession(tables_present=["MODELS", "SNAPSHOTS", "TAGS"], raise_on_ddl=True) + _new_backend(session, read_only=True) + assert session.ddl == [] From 44304d5331a73e23ff361884d1d955ea2265e296 Mon Sep 17 00:00:00 2001 From: Vignesh Narayanaswamy Date: Mon, 22 Jun 2026 14:38:08 -0700 Subject: [PATCH 2/2] Fall back to SQL write path when the role can't create temp tables MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bulk write path (_flush_models_pandas / _flush_snapshots_pandas) creates a TEMPORARY staging table, which needs CREATE TABLE on the schema. A least-privilege writer with only INSERT/UPDATE/SELECT can't, so the first write would still fail even after the init-DDL fix. Catch the privilege error (SQLSTATE 42501) and fall back to the existing DDL-free SQL MERGE/INSERT path, which needs only INSERT/UPDATE/SELECT. Privileged deployments keep the fast bulk path unchanged. With this, the entire write path (init, models, snapshots, tags) requires no CREATE/ALTER — only INSERT/UPDATE/SELECT. Co-Authored-By: Claude Opus 4.8 --- src/model_ledger/backends/snowflake.py | 119 ++++++++++++------- tests/test_backends/test_snowflake_ledger.py | 70 +++++++++++ 2 files changed, 145 insertions(+), 44 deletions(-) diff --git a/src/model_ledger/backends/snowflake.py b/src/model_ledger/backends/snowflake.py index a19e544..76c1d70 100644 --- a/src/model_ledger/backends/snowflake.py +++ b/src/model_ledger/backends/snowflake.py @@ -42,6 +42,21 @@ def _esc(value: str | None) -> str: return "'" + str(value).replace("'", "''") + "'" +def _is_privilege_error(exc: Exception) -> bool: + """True if exc looks like a Snowflake authorization failure (SQLSTATE 42501). + + Used to fall back from the bulk (temp-table) write path to the DDL-free SQL + path when the role lacks CREATE TABLE, rather than failing the write. + """ + msg = str(exc).lower() + return ( + "insufficient privileges" in msg + or "access control error" in msg + or "not authorized" in msg + or "(42501)" in msg + ) + + def _row_to_model_ref(row: dict[str, Any]) -> ModelRef: metadata = row.get("METADATA") or {} if isinstance(metadata, str): @@ -183,27 +198,36 @@ def _flush_models_pandas(self) -> bool: ) staging = f"{self._schema}.MODELS_STAGING" - _exec_no_result( - self._session, f"CREATE OR REPLACE TEMPORARY TABLE {staging} LIKE {self._schema}.MODELS" - ) - wp_kwargs: dict[str, str] = {"schema": self._schema_name} - if self._database: - wp_kwargs["database"] = self._database - write_pandas(conn, df, "MODELS_STAGING", **wp_kwargs) # type: ignore[arg-type] - _exec_no_result( - self._session, - f""" - MERGE INTO {self._schema}.MODELS t USING {staging} s ON t.MODEL_HASH = s.MODEL_HASH - WHEN MATCHED THEN UPDATE SET - NAME=s.NAME, OWNER=s.OWNER, MODEL_TYPE=s.MODEL_TYPE, - MODEL_ORIGIN=s.MODEL_ORIGIN, TIER=s.TIER, PURPOSE=s.PURPOSE, STATUS=s.STATUS, - LAST_SEEN=s.LAST_SEEN, METADATA=PARSE_JSON(s.METADATA) - WHEN NOT MATCHED THEN INSERT - (MODEL_HASH, NAME, OWNER, MODEL_TYPE, MODEL_ORIGIN, TIER, PURPOSE, STATUS, CREATED_AT, LAST_SEEN, METADATA) - VALUES (s.MODEL_HASH, s.NAME, s.OWNER, s.MODEL_TYPE, s.MODEL_ORIGIN, - s.TIER, s.PURPOSE, s.STATUS, s.CREATED_AT, s.LAST_SEEN, PARSE_JSON(s.METADATA))""", - ) - _exec_no_result(self._session, f"DROP TABLE IF EXISTS {staging}") + # The bulk path needs CREATE TABLE on the schema for the staging table. + # A least-privilege writer (INSERT/UPDATE/SELECT only) can't create it; + # fall back to the DDL-free SQL MERGE path rather than failing the write. + try: + _exec_no_result( + self._session, + f"CREATE OR REPLACE TEMPORARY TABLE {staging} LIKE {self._schema}.MODELS", + ) + wp_kwargs: dict[str, str] = {"schema": self._schema_name} + if self._database: + wp_kwargs["database"] = self._database + write_pandas(conn, df, "MODELS_STAGING", **wp_kwargs) # type: ignore[arg-type] + _exec_no_result( + self._session, + f""" + MERGE INTO {self._schema}.MODELS t USING {staging} s ON t.MODEL_HASH = s.MODEL_HASH + WHEN MATCHED THEN UPDATE SET + NAME=s.NAME, OWNER=s.OWNER, MODEL_TYPE=s.MODEL_TYPE, + MODEL_ORIGIN=s.MODEL_ORIGIN, TIER=s.TIER, PURPOSE=s.PURPOSE, STATUS=s.STATUS, + LAST_SEEN=s.LAST_SEEN, METADATA=PARSE_JSON(s.METADATA) + WHEN NOT MATCHED THEN INSERT + (MODEL_HASH, NAME, OWNER, MODEL_TYPE, MODEL_ORIGIN, TIER, PURPOSE, STATUS, CREATED_AT, LAST_SEEN, METADATA) + VALUES (s.MODEL_HASH, s.NAME, s.OWNER, s.MODEL_TYPE, s.MODEL_ORIGIN, + s.TIER, s.PURPOSE, s.STATUS, s.CREATED_AT, s.LAST_SEEN, PARSE_JSON(s.METADATA))""", + ) + _exec_no_result(self._session, f"DROP TABLE IF EXISTS {staging}") + except Exception as e: + if _is_privilege_error(e): + return False + raise return True def _flush_models_sql(self) -> None: @@ -275,29 +299,36 @@ def _flush_snapshots_pandas(self) -> bool: ) staging = f"{self._schema}.SNAPSHOTS_STAGING" - _exec_no_result( - self._session, - f""" - CREATE OR REPLACE TEMPORARY TABLE {staging} ( - SNAPSHOT_HASH VARCHAR, MODEL_HASH VARCHAR, PARENT_HASH VARCHAR, - TIMESTAMP VARCHAR, ACTOR VARCHAR, EVENT_TYPE VARCHAR, - SOURCE VARCHAR, PAYLOAD VARCHAR, TAGS VARCHAR)""", - ) - wp_kwargs: dict[str, str] = {"schema": self._schema_name} - if self._database: - wp_kwargs["database"] = self._database - write_pandas(conn, df, "SNAPSHOTS_STAGING", **wp_kwargs) # type: ignore[arg-type] - _exec_no_result( - self._session, - f""" - INSERT INTO {self._schema}.SNAPSHOTS - (SNAPSHOT_HASH, MODEL_HASH, PARENT_HASH, TIMESTAMP, ACTOR, EVENT_TYPE, SOURCE, PAYLOAD, TAGS) - SELECT SNAPSHOT_HASH, MODEL_HASH, PARENT_HASH, TIMESTAMP::TIMESTAMP_TZ, ACTOR, EVENT_TYPE, SOURCE, - PARSE_JSON(PAYLOAD), PARSE_JSON(TAGS) - FROM {staging} s - WHERE NOT EXISTS (SELECT 1 FROM {self._schema}.SNAPSHOTS t WHERE t.SNAPSHOT_HASH = s.SNAPSHOT_HASH)""", - ) - _exec_no_result(self._session, f"DROP TABLE IF EXISTS {staging}") + # See _flush_models_pandas: fall back to the DDL-free SQL path when the + # role can't create the staging table. + try: + _exec_no_result( + self._session, + f""" + CREATE OR REPLACE TEMPORARY TABLE {staging} ( + SNAPSHOT_HASH VARCHAR, MODEL_HASH VARCHAR, PARENT_HASH VARCHAR, + TIMESTAMP VARCHAR, ACTOR VARCHAR, EVENT_TYPE VARCHAR, + SOURCE VARCHAR, PAYLOAD VARCHAR, TAGS VARCHAR)""", + ) + wp_kwargs: dict[str, str] = {"schema": self._schema_name} + if self._database: + wp_kwargs["database"] = self._database + write_pandas(conn, df, "SNAPSHOTS_STAGING", **wp_kwargs) # type: ignore[arg-type] + _exec_no_result( + self._session, + f""" + INSERT INTO {self._schema}.SNAPSHOTS + (SNAPSHOT_HASH, MODEL_HASH, PARENT_HASH, TIMESTAMP, ACTOR, EVENT_TYPE, SOURCE, PAYLOAD, TAGS) + SELECT SNAPSHOT_HASH, MODEL_HASH, PARENT_HASH, TIMESTAMP::TIMESTAMP_TZ, ACTOR, EVENT_TYPE, SOURCE, + PARSE_JSON(PAYLOAD), PARSE_JSON(TAGS) + FROM {staging} s + WHERE NOT EXISTS (SELECT 1 FROM {self._schema}.SNAPSHOTS t WHERE t.SNAPSHOT_HASH = s.SNAPSHOT_HASH)""", + ) + _exec_no_result(self._session, f"DROP TABLE IF EXISTS {staging}") + except Exception as e: + if _is_privilege_error(e): + return False + raise return True def _flush_snapshots_sql(self) -> None: diff --git a/tests/test_backends/test_snowflake_ledger.py b/tests/test_backends/test_snowflake_ledger.py index 2c1878f..fcd40f2 100644 --- a/tests/test_backends/test_snowflake_ledger.py +++ b/tests/test_backends/test_snowflake_ledger.py @@ -720,3 +720,73 @@ def test_read_only_skips_ensure_tables_entirely(): session = _DDLProbeSession(tables_present=["MODELS", "SNAPSHOTS", "TAGS"], raise_on_ddl=True) _new_backend(session, read_only=True) assert session.ddl == [] + + +def test_is_privilege_error_detection(): + from model_ledger.backends.snowflake import _is_privilege_error + + assert _is_privilege_error(RuntimeError("003001 (42501): SQL access control error")) + assert _is_privilege_error(Exception("Insufficient privileges to operate on schema")) + assert _is_privilege_error(Exception("User not authorized to perform CREATE TABLE")) + # Non-privilege errors must surface, not be swallowed as a fallback trigger. + assert not _is_privilege_error(Exception("Object 'MODELS' already exists")) + assert not _is_privilege_error(Exception("connection reset by peer")) + + +class _PrivDeniedBulkSession: + """Passes the bulk-path gate (`_session_parameters`) but denies CREATE TABLE, + so the write must fall back to the DDL-free SQL MERGE/INSERT path.""" + + def __init__(self): + self._session_parameters = {} + self.created_temp = False + self.merges: list[str] = [] + self.inserts: list[str] = [] + + def sql(self, query: str, params: Any = None) -> MockCollectResult: + upper = query.upper().strip() + if "INFORMATION_SCHEMA.TABLES" in upper: + return MockCollectResult([{"TABLE_NAME": t} for t in ("MODELS", "SNAPSHOTS", "TAGS")]) + if "INFORMATION_SCHEMA.COLUMNS" in upper: + return MockCollectResult([{"1": 1}]) + if "TEMPORARY TABLE" in upper: + self.created_temp = True + raise RuntimeError( + "003001 (42501): SQL access control error: Insufficient " + "privileges to operate on schema 'MODEL_LEDGER'" + ) + if upper.startswith("MERGE INTO"): + self.merges.append(query) + return MockCollectResult([]) + if upper.startswith("INSERT INTO"): + self.inserts.append(query) + return MockCollectResult([]) + return MockCollectResult([]) + + +def test_write_falls_back_to_sql_when_temp_table_denied(): + # Requires the bulk-path deps so the temp-table CREATE is actually reached. + pytest.importorskip("pandas") + pytest.importorskip("snowflake.connector.pandas_tools") + from model_ledger.backends.snowflake import SnowflakeLedgerBackend + + session = _PrivDeniedBulkSession() + # init must skip DDL (tables already present), else it fails before any write. + backend = SnowflakeLedgerBackend(schema="DB.MODEL_LEDGER", connection=session) + backend.save_model(_make_model()) + backend.append_snapshot( + Snapshot( + snapshot_hash="snap-fallback", + model_hash="abc123", + timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc), + actor="t", + event_type="registered", + source="test", + payload={}, + ) + ) + backend.flush() # must not raise + + assert session.created_temp is True # bulk path was attempted + assert len(session.merges) >= 1 # models persisted via SQL MERGE + assert len(session.inserts) >= 1 # snapshots persisted via SQL INSERT