Skip schema DDL on write-init when tables already exist

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:
@@ -322,6 +353,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 +405,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)
 

tests/test_backends/test_snowflake_ledger.py

@@ -623,3 +623,170 @@ 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 == []
+
+
+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