Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions target_snowflake/connector.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,21 @@ def inspector(self) -> sqlalchemy.Inspector:
self._inspector = sqlalchemy.inspect(self._engine)
return self._inspector

def invalidate_table_cache(self, full_table_name: str) -> None:
"""Discard cached reflection state for a table after DDL.

Dropping the `table_cache` entry alone is not enough. The Inspector
memoises reflection for its lifetime, and snowflake-sqlalchemy caches
columns per *schema*, so a table created after the first reflection of
that schema stays invisible and reflecting it raises NoSuchTableError.
The Inspector is dropped so the next reflection queries Snowflake.

Args:
full_table_name: the table whose cached state is now stale.
"""
self.table_cache.pop(full_table_name, None)
self._inspector = None

def get_table_columns(
self,
full_table_name: str | FullyQualifiedName,
Expand Down
4 changes: 2 additions & 2 deletions target_snowflake/sinks.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ def setup(self) -> None:
self.conform_name(self.schema_name, object_type="schema"),
)

self.connector.table_cache.pop(self.full_table_name, None)
self.connector.invalidate_table_cache(self.full_table_name)

try:
self.connector.prepare_table(
Expand All @@ -105,7 +105,7 @@ def setup(self) -> None:
)
raise

self.connector.table_cache.pop(self.full_table_name, None)
self.connector.invalidate_table_cache(self.full_table_name)

if self.config.get("load_method", "upsert") == "overwrite":
self.logger.info("load_method=overwrite: truncating %s", self.full_table_name)
Expand Down
34 changes: 34 additions & 0 deletions tests/unit/test_connector.py
Original file line number Diff line number Diff line change
Expand Up @@ -193,3 +193,37 @@ def test_format_identifier(connector: SnowflakeConnector, config: dict, identifi
connector.config.update(config)
formatted = connector.format_identifier(identifier)
assert formatted == expected_formatted


def test_invalidate_table_cache_drops_inspector(connector: SnowflakeConnector):
"""Invalidating a table must drop the cached Inspector, not just table_cache.

snowflake-sqlalchemy memoises columns per schema on the Inspector, so
reusing it after a CREATE TABLE reflects the schema as it was before the
table existed and raises NoSuchTableError.
"""
connector.table_cache["DB.SCHEMA.USERS"] = {"id": mock.Mock()}
connector.table_cache["DB.SCHEMA.POSTS"] = {"id": mock.Mock()}
connector._inspector = mock.Mock(spec=sa.Inspector)

connector.invalidate_table_cache("DB.SCHEMA.USERS")

assert "DB.SCHEMA.USERS" not in connector.table_cache
assert "DB.SCHEMA.POSTS" in connector.table_cache
assert connector._inspector is None


def test_inspector_rebuilt_after_invalidation(connector: SnowflakeConnector):
"""A fresh Inspector is built on next access, so reflection re-queries."""
connector._cached_engine = mock.Mock()

with mock.patch("sqlalchemy.inspect") as inspect_mock:
inspect_mock.side_effect = [mock.sentinel.stale, mock.sentinel.fresh]

assert connector.inspector is mock.sentinel.stale
assert connector.inspector is mock.sentinel.stale # cached

connector.invalidate_table_cache("DB.SCHEMA.USERS")

assert connector.inspector is mock.sentinel.fresh
assert inspect_mock.call_count == 2
41 changes: 41 additions & 0 deletions tests/unit/test_sinks.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,11 @@
from unittest import mock

import pytest
import sqlalchemy as sa
from singer_sdk.helpers._batch import JSONLinesEncoding

from target_snowflake.arrow_batch import ArrowEncoding
from target_snowflake.connector import SnowflakeConnector
from target_snowflake.sinks import SnowflakeSink


Expand Down Expand Up @@ -148,3 +150,42 @@ def fake_convert(manifest, output_dir, *, clean_up_source_files): # noqa: ARG00
SnowflakeSink._process_arrow_batch_files(fake_sink, ["file:///tmp/a.arrow"]) # noqa: SLF001

assert not os.path.exists(captured_output_dir["path"])


def test_setup_invalidates_cached_inspector():
"""setup() must drop the connector's cached Inspector after creating the table.

snowflake-sqlalchemy memoises columns per schema on the Inspector, so a table
created here stays invisible to later reflection unless the Inspector is
dropped. That surfaced as NoSuchTableError in activate_version for every
stream after the first.
"""
connector = SnowflakeConnector(
config={
"user": "test_user",
"password": "test_password",
"account": "test_account",
"database": "TEST_DATABASE",
},
)
connector._inspector = mock.Mock(spec=sa.Inspector)
connector.table_cache["DB.SCHEMA.TABLE"] = {"id": mock.Mock()}
connector.prepare_schema = mock.MagicMock()
connector.prepare_table = mock.MagicMock()

fake_sink = SimpleNamespace(
schema_name="SCHEMA",
full_table_name="DB.SCHEMA.TABLE",
schema={},
key_properties=[],
config={},
logger=mock.MagicMock(),
connector=connector,
conform_name=lambda name, object_type=None: name, # noqa: ARG005
conform_schema=lambda schema: schema,
)

SnowflakeSink.setup(fake_sink)

assert connector._inspector is None
assert "DB.SCHEMA.TABLE" not in connector.table_cache