Skip to content

Commit e9ccd27

Browse files
fix: pandas bulk path falls back to SQL on ANY failure, not just privilege errors (#34)
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 <noreply@anthropic.com>
1 parent 55d8300 commit e9ccd27

3 files changed

Lines changed: 95 additions & 7 deletions

File tree

pyproject.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[project]
22
name = "model-ledger"
3-
version = "0.7.10"
3+
version = "0.7.11"
44
description = "Developer-first model inventory and governance framework for SR 11-7, EU AI Act, and NIST AI RMF compliance"
55
readme = "README.md"
66
requires-python = ">=3.10"

src/model_ledger/backends/snowflake.py

Lines changed: 38 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,18 @@
66

77
from __future__ import annotations
88

9+
import contextlib
910
import json
11+
import logging
1012
import threading
1113
from collections.abc import Callable
1214
from datetime import datetime, timezone
1315
from typing import Any
1416

1517
from model_ledger.core.ledger_models import ModelRef, Snapshot, Tag
1618

19+
logger = logging.getLogger(__name__)
20+
1721
BATCH_SIZE = 50
1822

1923
# Snowflake error code raised when a session's auth token has idle-expired
@@ -317,9 +321,23 @@ def _flush_models_pandas(self) -> bool:
317321
)
318322
self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
319323
except Exception as e:
320-
if _is_privilege_error(e):
321-
return False
322-
raise
324+
# The pandas path is an optimization, not a correctness
325+
# requirement: privilege denials (no CREATE TEMPORARY TABLE) and
326+
# transport limitations (sessions whose backing protocol cannot
327+
# run PUT file transfers — write_pandas dies inside
328+
# file_transfer_agent, e.g. KeyError('command')) both land here.
329+
# Any failure falls back to the DDL-free SQL path, which is
330+
# idempotent and carries all columns. Never leave the buffer
331+
# poisoned: a raised exception here would re-fire on every
332+
# subsequent flush-before-read and 500 the whole API.
333+
logger.warning(
334+
"pandas bulk path failed (%s: %s); falling back to SQL flush",
335+
type(e).__name__,
336+
e,
337+
)
338+
with contextlib.suppress(Exception):
339+
self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
340+
return False
323341
return True
324342

325343
def _flush_models_sql(self) -> None:
@@ -415,9 +433,23 @@ def _flush_snapshots_pandas(self) -> bool:
415433
)
416434
self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
417435
except Exception as e:
418-
if _is_privilege_error(e):
419-
return False
420-
raise
436+
# The pandas path is an optimization, not a correctness
437+
# requirement: privilege denials (no CREATE TEMPORARY TABLE) and
438+
# transport limitations (sessions whose backing protocol cannot
439+
# run PUT file transfers — write_pandas dies inside
440+
# file_transfer_agent, e.g. KeyError('command')) both land here.
441+
# Any failure falls back to the DDL-free SQL path, which is
442+
# idempotent and carries all columns. Never leave the buffer
443+
# poisoned: a raised exception here would re-fire on every
444+
# subsequent flush-before-read and 500 the whole API.
445+
logger.warning(
446+
"pandas bulk path failed (%s: %s); falling back to SQL flush",
447+
type(e).__name__,
448+
e,
449+
)
450+
with contextlib.suppress(Exception):
451+
self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
452+
return False
421453
return True
422454

423455
def _flush_snapshots_sql(self) -> None:

tests/test_backends/test_snowflake_ledger.py

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1066,3 +1066,59 @@ def test_requires_connection_or_factory(self):
10661066

10671067
with pytest.raises(ValueError, match="connection"):
10681068
SnowflakeLedgerBackend(schema="TEST_SCHEMA")
1069+
1070+
1071+
def test_pandas_path_failure_falls_back_to_sql(monkeypatch):
1072+
"""A pandas bulk-path failure that is NOT a privilege error (e.g. the
1073+
backing session's protocol cannot run PUT file transfers — write_pandas
1074+
raises KeyError('command') from file_transfer_agent) must fall back to
1075+
the SQL path instead of raising. A raised exception leaves the buffer
1076+
poisoned: every subsequent flush-before-read re-fires it and the whole
1077+
API 500s (observed in production 2026-07-10).
1078+
"""
1079+
import sys
1080+
import types
1081+
1082+
from model_ledger.backends.snowflake import SnowflakeLedgerBackend
1083+
1084+
statements: list[str] = []
1085+
1086+
class FakeConn:
1087+
_session_parameters = {"QUERY_TAG": "test"}
1088+
1089+
class PandasCapableSession:
1090+
"""Looks pandas-capable (has _connection w/ _session_parameters)."""
1091+
1092+
_connection = FakeConn()
1093+
1094+
def sql(self, query: str, params: Any = None) -> MockCollectResult:
1095+
statements.append(query)
1096+
return MockCollectResult([[0]])
1097+
1098+
def exploding_write_pandas(*args, **kwargs):
1099+
raise KeyError("command")
1100+
1101+
fake_tools = types.ModuleType("snowflake.connector.pandas_tools")
1102+
fake_tools.write_pandas = exploding_write_pandas
1103+
monkeypatch.setitem(sys.modules, "snowflake.connector.pandas_tools", fake_tools)
1104+
1105+
backend = SnowflakeLedgerBackend(schema="TEST_SCHEMA", connection=PandasCapableSession())
1106+
backend.save_model(
1107+
ModelRef(
1108+
name="fraud_scorer",
1109+
owner="risk-team",
1110+
model_type="scoring_model",
1111+
tier="unclassified",
1112+
purpose="",
1113+
)
1114+
)
1115+
backend.flush() # must NOT raise
1116+
1117+
merges = [s for s in statements if "MERGE INTO" in s.upper() and ".MODELS" in s.upper()]
1118+
assert merges, "SQL fallback MERGE never ran after pandas-path failure"
1119+
assert not backend._model_buffer, "buffer not cleared — would re-fire on every flush"
1120+
1121+
# Second flush must be a no-op, not a re-explosion.
1122+
before = len(statements)
1123+
backend.flush()
1124+
assert len(statements) == before

0 commit comments

Comments
 (0)