From 5d142955fa18c66dfb2cc6dcc995d581e3d5e24b Mon Sep 17 00:00:00 2001 From: Vignesh Narayanaswamy Date: Fri, 10 Jul 2026 09:37:27 -0700 Subject: [PATCH] fix: pandas bulk path falls back to SQL on ANY failure, not just privilege errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Observed in production: on a session whose backing protocol cannot run PUT file transfers, write_pandas dies inside file_transfer_agent with KeyError('command') — not a privilege error — so the fallback never engaged, the exception propagated, and because the buffer was never cleared, every subsequent flush-before-read re-fired it (all reads 500 on that process). The pandas path is an optimization, not a correctness requirement: any failure now logs a warning, best-effort drops the staging table, and falls back to the DDL-free SQL path (idempotent, carries all columns as of v0.7.10). Regression test simulates the poisoned-buffer scenario and asserts fallback + buffer clear + quiet second flush. Bumps version to 0.7.11. Co-Authored-By: Claude Fable 5 --- pyproject.toml | 2 +- src/model_ledger/backends/snowflake.py | 44 ++++++++++++--- tests/test_backends/test_snowflake_ledger.py | 56 ++++++++++++++++++++ 3 files changed, 95 insertions(+), 7 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e89f5a4..6e04056 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "model-ledger" -version = "0.7.10" +version = "0.7.11" description = "Developer-first model inventory and governance framework for SR 11-7, EU AI Act, and NIST AI RMF compliance" readme = "README.md" requires-python = ">=3.10" diff --git a/src/model_ledger/backends/snowflake.py b/src/model_ledger/backends/snowflake.py index 1d9d479..972c10c 100644 --- a/src/model_ledger/backends/snowflake.py +++ b/src/model_ledger/backends/snowflake.py @@ -6,7 +6,9 @@ from __future__ import annotations +import contextlib import json +import logging import threading from collections.abc import Callable from datetime import datetime, timezone @@ -14,6 +16,8 @@ from model_ledger.core.ledger_models import ModelRef, Snapshot, Tag +logger = logging.getLogger(__name__) + BATCH_SIZE = 50 # Snowflake error code raised when a session's auth token has idle-expired @@ -317,9 +321,23 @@ def _flush_models_pandas(self) -> bool: ) self._exec_no_result(f"DROP TABLE IF EXISTS {staging}") except Exception as e: - if _is_privilege_error(e): - return False - raise + # The pandas path is an optimization, not a correctness + # requirement: privilege denials (no CREATE TEMPORARY TABLE) and + # transport limitations (sessions whose backing protocol cannot + # run PUT file transfers — write_pandas dies inside + # file_transfer_agent, e.g. KeyError('command')) both land here. + # Any failure falls back to the DDL-free SQL path, which is + # idempotent and carries all columns. Never leave the buffer + # poisoned: a raised exception here would re-fire on every + # subsequent flush-before-read and 500 the whole API. + logger.warning( + "pandas bulk path failed (%s: %s); falling back to SQL flush", + type(e).__name__, + e, + ) + with contextlib.suppress(Exception): + self._exec_no_result(f"DROP TABLE IF EXISTS {staging}") + return False return True def _flush_models_sql(self) -> None: @@ -415,9 +433,23 @@ def _flush_snapshots_pandas(self) -> bool: ) self._exec_no_result(f"DROP TABLE IF EXISTS {staging}") except Exception as e: - if _is_privilege_error(e): - return False - raise + # The pandas path is an optimization, not a correctness + # requirement: privilege denials (no CREATE TEMPORARY TABLE) and + # transport limitations (sessions whose backing protocol cannot + # run PUT file transfers — write_pandas dies inside + # file_transfer_agent, e.g. KeyError('command')) both land here. + # Any failure falls back to the DDL-free SQL path, which is + # idempotent and carries all columns. Never leave the buffer + # poisoned: a raised exception here would re-fire on every + # subsequent flush-before-read and 500 the whole API. + logger.warning( + "pandas bulk path failed (%s: %s); falling back to SQL flush", + type(e).__name__, + e, + ) + with contextlib.suppress(Exception): + self._exec_no_result(f"DROP TABLE IF EXISTS {staging}") + return False 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 d72346c..8130cba 100644 --- a/tests/test_backends/test_snowflake_ledger.py +++ b/tests/test_backends/test_snowflake_ledger.py @@ -1066,3 +1066,59 @@ def test_requires_connection_or_factory(self): with pytest.raises(ValueError, match="connection"): SnowflakeLedgerBackend(schema="TEST_SCHEMA") + + +def test_pandas_path_failure_falls_back_to_sql(monkeypatch): + """A pandas bulk-path failure that is NOT a privilege error (e.g. the + backing session's protocol cannot run PUT file transfers — write_pandas + raises KeyError('command') from file_transfer_agent) must fall back to + the SQL path instead of raising. A raised exception leaves the buffer + poisoned: every subsequent flush-before-read re-fires it and the whole + API 500s (observed in production 2026-07-10). + """ + import sys + import types + + from model_ledger.backends.snowflake import SnowflakeLedgerBackend + + statements: list[str] = [] + + class FakeConn: + _session_parameters = {"QUERY_TAG": "test"} + + class PandasCapableSession: + """Looks pandas-capable (has _connection w/ _session_parameters).""" + + _connection = FakeConn() + + def sql(self, query: str, params: Any = None) -> MockCollectResult: + statements.append(query) + return MockCollectResult([[0]]) + + def exploding_write_pandas(*args, **kwargs): + raise KeyError("command") + + fake_tools = types.ModuleType("snowflake.connector.pandas_tools") + fake_tools.write_pandas = exploding_write_pandas + monkeypatch.setitem(sys.modules, "snowflake.connector.pandas_tools", fake_tools) + + backend = SnowflakeLedgerBackend(schema="TEST_SCHEMA", connection=PandasCapableSession()) + backend.save_model( + ModelRef( + name="fraud_scorer", + owner="risk-team", + model_type="scoring_model", + tier="unclassified", + purpose="", + ) + ) + backend.flush() # must NOT raise + + merges = [s for s in statements if "MERGE INTO" in s.upper() and ".MODELS" in s.upper()] + assert merges, "SQL fallback MERGE never ran after pandas-path failure" + assert not backend._model_buffer, "buffer not cleared — would re-fire on every flush" + + # Second flush must be a no-op, not a re-explosion. + before = len(statements) + backend.flush() + assert len(statements) == before