Skip to content

Fix snowflake loader stream state store #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fordN wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix snowflake loader stream state store #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3a3d1b2
snowflake loader: fix stream state store config, allow setting its own
fordN Dec 11, 2025
7f9b201
logging: suppress acero alignment warnings
fordN Dec 11, 2025
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
7 changes: 7 additions & 0 deletions src/amp/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,12 @@
"""Amp - Flight SQL client with comprehensive data loading capabilities."""

import os

# Suppress PyArrow acero alignment warnings before any imports
# These warnings are informational and cannot be controlled via logging
# Valid values: ignore, warn, reallocate, error
os.environ.setdefault('ACERO_ALIGNMENT_HANDLING', 'ignore')

from amp.client import Client, QueryBuilder
from amp.registry import RegistryClient

Expand Down
11 changes: 4 additions & 7 deletions src/amp/loaders/implementations/snowflake_loader.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -831,13 +831,10 @@ def connect(self) -> None:
self._create_stage()

# Replace in-memory state store with Snowflake-backed store if configured
state_config = getattr(self.config, 'state', None)
if state_config:
storage = getattr(state_config, 'storage', None)
enabled = getattr(state_config, 'enabled', True)
if storage == 'snowflake' and enabled:
self.logger.info('Using Snowflake-backed persistent state store')
self.state_store = SnowflakeStreamStateStore(self.connection, self.cursor, self.logger)
# Base class already parsed state config and stored it in self.state_storage and self.state_enabled
if self.state_storage == 'snowflake' and self.state_enabled:
self.logger.info('Using Snowflake-backed persistent state store')
self.state_store = SnowflakeStreamStateStore(self.connection, self.cursor, self.logger)
# Otherwise, state store is initialized in base class with in-memory storage (default)

self._is_connected = True
Expand Down
Loading