From c4a6e197bab79cc1c5b8b8caa86372655467a998 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sat, 18 Jul 2026 11:32:49 +1000 Subject: [PATCH 1/4] test(protocol): add non-Ibis stub-backend conformance guard Standing guard that the Backend protocol stays satisfiable by a second, non-Ibis backend before Iceberg (the only other implementer) is retired. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_unit/core/test_protocol.py | 65 +++++++++++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/tests/test_unit/core/test_protocol.py b/tests/test_unit/core/test_protocol.py index 5893dc4..7c5553b 100644 --- a/tests/test_unit/core/test_protocol.py +++ b/tests/test_unit/core/test_protocol.py @@ -89,3 +89,68 @@ def test_ibis_backend_satisfies_protocol_including_in_transaction(): be = IbisBackend(dialect="duckdb", database=":memory:") assert isinstance(be, Backend) # runtime_checkable: presence of protocol methods assert callable(be.in_transaction) + + +class _StubBackend: + """Minimal non-Ibis structural implementer of the Backend protocol. + + Standing guard that the protocol stays satisfiable by a second backend and + does not silently collapse into 'whatever IbisBackend does'. Not shipped, + not registered — test-only. + """ + + name = "stub" + + def connect(self): + return self + + def close(self): + return self + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def list_tables(self, namespace=None): + return [] + + def list_namespaces(self, catalog=None): + return [] + + def list_catalogs(self): + return [] + + def inspect_table(self, name, namespace=None): + return TableInfo(name=name, columns=[]) + + def inspect_namespace(self, name): + return NamespaceInfo(location=Namespace(), tables=[]) + + def inspect_catalog(self, catalog=None): + return CatalogInfo(name=catalog or "stub", namespaces=[]) + + def raw_driver_connection(self): + raise RuntimeError("stub has no driver handle") + + @property + def supports_transactions(self): + return False + + def transaction(self, *, required=True): + raise RuntimeError("stub does not support transactions") + + def in_transaction(self): + return False + + +def test_non_ibis_stub_satisfies_backend_protocol(): + stub = _StubBackend() + assert isinstance(stub, Backend) + # catalog tier — the seam Iceberg originally motivated, kept generic + assert stub.list_catalogs() == [] + assert stub.inspect_catalog().name == "stub" + assert stub.list_namespaces(catalog="anything") == [] + assert stub.in_transaction() is False + assert stub.supports_transactions is False From e423331ac4323fa7ce5b977da4a4cface104f97e Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sat, 18 Jul 2026 11:38:51 +1000 Subject: [PATCH 2/4] chore: retire the Iceberg backend Remove backends/iceberg/, its two settings modules, its tests, the PYICEBERG* enum members, the pyiceberg_rest auth adapter, and the Iceberg mypy carve-out. The backend-agnostic infrastructure (Backend protocol catalog tier, inspection model, registry, settings/adapter framework) is retained unchanged. Surviving tests unwired from Iceberg references. Co-Authored-By: Claude Opus 4.8 (1M context) --- pyproject.toml | 12 +- src/mountainash_data/__init__.py | 7 - .../backends/iceberg/__init__.py | 0 .../backends/iceberg/_types.py | 113 ---- .../backends/iceberg/backend.py | 111 ---- .../backends/iceberg/catalogs/__init__.py | 0 .../backends/iceberg/catalogs/rest.py | 219 ------- .../backends/iceberg/connection.py | 617 ------------------ .../backends/iceberg/inspect.py | 82 --- .../backends/iceberg/operations.py | 434 ------------ src/mountainash_data/core/_warn.py | 6 +- src/mountainash_data/core/constants.py | 3 - src/mountainash_data/core/inspection.py | 2 +- src/mountainash_data/core/registry.py | 7 +- .../core/settings/__init__.py | 2 - .../core/settings/adapters/pyiceberg_rest.py | 15 - .../core/settings/adapters/registry.py | 3 +- .../core/settings/pyiceberg_rest.py | 70 -- .../backends/ibis/test_namespace_hierarchy.py | 2 - .../backends/iceberg/COVERAGE_GAP.md | 17 - tests/test_unit/backends/iceberg/__init__.py | 0 .../backends/iceberg/test_backend.py | 83 --- .../backends/iceberg/test_iceberg_auth.py | 40 -- .../backends/iceberg/test_namespace.py | 23 - .../settings/adapters/test_auth_adapters.py | 6 +- .../settings/backends/test_pyiceberg_rest.py | 37 -- .../core/settings/test_config_shaping.py | 9 - .../test_unit/core/settings/test_registry.py | 4 +- tests/test_unit/core/test_protocol.py | 9 +- 29 files changed, 14 insertions(+), 1919 deletions(-) delete mode 100644 src/mountainash_data/backends/iceberg/__init__.py delete mode 100644 src/mountainash_data/backends/iceberg/_types.py delete mode 100644 src/mountainash_data/backends/iceberg/backend.py delete mode 100644 src/mountainash_data/backends/iceberg/catalogs/__init__.py delete mode 100644 src/mountainash_data/backends/iceberg/catalogs/rest.py delete mode 100644 src/mountainash_data/backends/iceberg/connection.py delete mode 100644 src/mountainash_data/backends/iceberg/inspect.py delete mode 100644 src/mountainash_data/backends/iceberg/operations.py delete mode 100644 src/mountainash_data/core/settings/adapters/pyiceberg_rest.py delete mode 100644 src/mountainash_data/core/settings/pyiceberg_rest.py delete mode 100644 tests/test_unit/backends/iceberg/COVERAGE_GAP.md delete mode 100644 tests/test_unit/backends/iceberg/__init__.py delete mode 100644 tests/test_unit/backends/iceberg/test_backend.py delete mode 100644 tests/test_unit/backends/iceberg/test_iceberg_auth.py delete mode 100644 tests/test_unit/backends/iceberg/test_namespace.py delete mode 100644 tests/test_unit/core/settings/backends/test_pyiceberg_rest.py diff --git a/pyproject.toml b/pyproject.toml index e835438..8a3b056 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,18 +102,8 @@ exclude_lines = ["no cov", "if __name__ == .__main__:", "if TYPE_CHECKING:"] [tool.mypy] # Internal mountainash siblings ship no py.typed marker, and optional drivers -# (trino/google/pyiceberg/mountainash_dataframes) are not installed in the type +# (trino/google/mountainash_dataframes) are not installed in the type # env — treat all unresolved/untyped third-party imports as Any rather than noise. ignore_missing_imports = true disable_error_code = ["import-untyped"] -# Pre-existing type debt in the Iceberg backend (catalog_backend None-narrowing, -# pyiceberg Any-typed surfaces, hook-signature mismatches) — NOT introduced by -# the auth-client migration and in code paths that require the optional pyiceberg -# + mountainash_dataframes stack to exercise. Carved out for a separate Iceberg -# type-hardening pass; the auth-migration code (factories/adapters/registry/ibis) -# is type-checked normally. -[[tool.mypy.overrides]] -module = "mountainash_data.backends.iceberg.*" -ignore_errors = true - diff --git a/src/mountainash_data/__init__.py b/src/mountainash_data/__init__.py index 31d189c..5dfecc5 100644 --- a/src/mountainash_data/__init__.py +++ b/src/mountainash_data/__init__.py @@ -3,7 +3,6 @@ Public API: Backend — protocol (core.protocol) IbisBackend — ibis-style relational backends (backends.ibis.backend) - IcebergBackend — iceberg-style table-format catalogs (backends.iceberg.backend) CatalogInfo, NamespaceInfo, TableInfo, ColumnInfo — inspection model """ @@ -18,11 +17,6 @@ from mountainash_data.core.namespace import Namespace, NamespaceLike from mountainash_data.backends.ibis.backend import IbisBackend -try: - from mountainash_data.backends.iceberg.backend import IcebergBackend -except ImportError: - IcebergBackend = None # type: ignore[assignment,misc] - __all__ = [ "__version__", "Backend", @@ -33,5 +27,4 @@ "Namespace", "NamespaceLike", "IbisBackend", - "IcebergBackend", ] diff --git a/src/mountainash_data/backends/iceberg/__init__.py b/src/mountainash_data/backends/iceberg/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/mountainash_data/backends/iceberg/_types.py b/src/mountainash_data/backends/iceberg/_types.py deleted file mode 100644 index 7b482a5..0000000 --- a/src/mountainash_data/backends/iceberg/_types.py +++ /dev/null @@ -1,113 +0,0 @@ -"""Iceberg to PyArrow type conversion helpers. - -Extracted from the legacy base_pyiceberg_connection.py / -base_pyiceberg_operations.py during the Phase 3 deduplication. These -functions take Iceberg schema fields and return their PyArrow equivalents. - -Pure functions, no classes. No external mountainash_data imports. -""" - -from __future__ import annotations - -import pyarrow as pa -from pyiceberg.types import ( - BinaryType, - BooleanType, - DateType, - DecimalType, - DoubleType, - FixedType, - FloatType, - IntegerType, - ListType, - LongType, - MapType, - StringType, - StructType, - TimestampType, - TimeType, - UUIDType, -) - - -def iceberg_type_to_pyarrow(iceberg_type) -> pa.DataType: - """Convert a single Iceberg field type to its PyArrow equivalent. - - Supports primitive types and the common composite types (list, map, - struct). Unknown types fall back to ``pa.string()``. - - Args: - iceberg_type: An Iceberg type object (e.g. ``BooleanType()``, - ``ListType(...)``, etc.) - - Returns: - The corresponding ``pa.DataType``. - """ - if isinstance(iceberg_type, BooleanType): - return pa.bool_() - elif isinstance(iceberg_type, IntegerType): - return pa.int32() - elif isinstance(iceberg_type, LongType): - return pa.int64() - elif isinstance(iceberg_type, FloatType): - return pa.float32() - elif isinstance(iceberg_type, DoubleType): - return pa.float64() - elif isinstance(iceberg_type, DateType): - return pa.date32() - elif isinstance(iceberg_type, TimeType): - return pa.time64("us") - elif isinstance(iceberg_type, TimestampType): - if iceberg_type.with_timezone: - return pa.timestamp("us", tz="UTC") - else: - return pa.timestamp("us") - elif isinstance(iceberg_type, StringType): - return pa.string() - elif isinstance(iceberg_type, UUIDType): - # UUIDs are usually handled as strings in PyArrow - return pa.string() - elif isinstance(iceberg_type, BinaryType): - return pa.binary() - elif isinstance(iceberg_type, DecimalType): - return pa.decimal128(iceberg_type.precision, iceberg_type.scale) - elif isinstance(iceberg_type, FixedType): - return pa.binary(iceberg_type.length) - elif isinstance(iceberg_type, ListType): - # Recursive call for list element type (one level deep) - inner_type = iceberg_type.element_type - pa_element_type = iceberg_type_to_pyarrow(inner_type) - return pa.list_(pa_element_type) - elif isinstance(iceberg_type, MapType): - # For maps, we default to string keys and values - return pa.map_(pa.string(), pa.string()) - elif isinstance(iceberg_type, StructType): - # For structs, we create a nested field structure - struct_fields = [ - pa.field( - nested_field.name, - pa.string(), - nullable=not nested_field.required, - ) - for nested_field in iceberg_type.fields - ] - return pa.struct(struct_fields) - else: - # Default fallback for unrecognised types - return pa.string() - - -def iceberg_schema_to_pyarrow(iceberg_schema) -> pa.Schema: - """Convert a full Iceberg Schema to a PyArrow Schema. - - Args: - iceberg_schema: A ``pyiceberg.schema.Schema`` instance. - - Returns: - Equivalent ``pa.Schema``. - """ - pa_fields = [] - for field in iceberg_schema.fields: - pa_type = iceberg_type_to_pyarrow(field.field_type) - pa_fields.append(pa.field(field.name, pa_type, nullable=not field.required)) - return pa.schema(pa_fields) diff --git a/src/mountainash_data/backends/iceberg/backend.py b/src/mountainash_data/backends/iceberg/backend.py deleted file mode 100644 index 5f57a69..0000000 --- a/src/mountainash_data/backends/iceberg/backend.py +++ /dev/null @@ -1,111 +0,0 @@ -"""IcebergBackend — implements core.protocol.Backend for iceberg catalogs.""" - -from __future__ import annotations - -import contextlib -import typing as t - -from mountainash_data.backends.iceberg.catalogs.rest import IcebergRestConnection -from mountainash_data.backends.iceberg.connection import IcebergConnectionBase -from mountainash_data.core.errors import TransactionUnsupportedError -from mountainash_data.core._warn import warn_once - - -_CATALOG_REGISTRY: dict[str, type[IcebergConnectionBase]] = { - "rest": IcebergRestConnection, -} - - -class IcebergBackend: - """Iceberg backend — single entry point for iceberg catalog interaction. - - Construction takes a catalog type (e.g. ``'rest'``) and config kwargs. - ``connect()`` returns ``self``. Use as a context manager. - """ - - name = "iceberg" - - def __init__(self, catalog: str, **config: t.Any) -> None: - if catalog not in _CATALOG_REGISTRY: - raise KeyError( - f"Unknown iceberg catalog type {catalog!r}. " - f"Available: {sorted(_CATALOG_REGISTRY)}" - ) - self._catalog_cls = _CATALOG_REGISTRY[catalog] - self._config = config - self._conn: IcebergConnectionBase | None = None - - def connect(self) -> IcebergBackend: - """Open a connection. Returns self for fluent chaining.""" - if self._conn is None: - self._conn = self._catalog_cls(**self._config) - return self - - def close(self) -> IcebergBackend: - """Release the connection. Idempotent. Returns self.""" - if self._conn is not None: - if hasattr(self._conn, "close"): - self._conn.close() - self._conn = None - return self - - def __enter__(self) -> IcebergBackend: - self.connect() - return self - - def __exit__(self, *args: t.Any) -> None: - self.close() - - def _require_connected(self) -> IcebergConnectionBase: - if self._conn is None: - raise RuntimeError( - "IcebergBackend is not connected. Call connect() first." - ) - return self._conn - - def list_tables(self, namespace: t.Any = None) -> list[str]: - return self._require_connected().list_tables(namespace=namespace) - - def list_namespaces(self, catalog: str | None = None) -> list[str]: - return self._require_connected().list_namespaces(catalog=catalog) - - def list_catalogs(self) -> list[str]: - return self._require_connected().list_catalogs() - - def inspect_table(self, name: str, namespace: t.Any = None) -> t.Any: - return self._require_connected().inspect_table(name, namespace=namespace) - - def inspect_namespace(self, name: str) -> t.Any: - return self._require_connected().inspect_namespace(name) - - def inspect_catalog(self, catalog: str | None = None) -> t.Any: - return self._require_connected().inspect_catalog(catalog=catalog) - - def raw_driver_connection(self) -> t.Any: - """Return the underlying pyiceberg Catalog (native handle).""" - return self._require_connected().catalog_backend - - @property - def supports_transactions(self) -> bool: - return False - - @contextlib.contextmanager - def transaction(self, *, required: bool = True): - """Iceberg has no connection-level cross-table transaction; declines. - - required=True raises; required=False warns ONCE and no-ops. (pyiceberg - offers table-scoped transactions — a future capability, not this one.) - """ - if required: - raise TransactionUnsupportedError( - "iceberg has no connection-level transaction; use table-scoped " - "pyiceberg transactions, or call transaction(required=False)." - ) - warn_once("iceberg", "iceberg has no transaction support; transaction() is a no-op.") - yield - - def in_transaction(self) -> bool: - """Iceberg has no connection-level unit of work (supports_transactions - is False); nothing can be active. pyiceberg table-scoped transactions - are a separate future capability.""" - return False diff --git a/src/mountainash_data/backends/iceberg/catalogs/__init__.py b/src/mountainash_data/backends/iceberg/catalogs/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/mountainash_data/backends/iceberg/catalogs/rest.py b/src/mountainash_data/backends/iceberg/catalogs/rest.py deleted file mode 100644 index ceadb73..0000000 --- a/src/mountainash_data/backends/iceberg/catalogs/rest.py +++ /dev/null @@ -1,219 +0,0 @@ -"""REST catalog implementation for the Iceberg backend. - -Merges the legacy pyiceberg_rest_connection.py and pyiceberg_rest_operations.py -into a single concrete connection class for the REST catalog. - -NOTE on the ``_upsert`` override: - The legacy REST files both contain an ``_upsert`` implementation that uses - raw SQL cursor operations (DuckDB-style syntax). This appears to be a - copy-paste artefact from a DuckDB connection — it is incorrect for an Iceberg - REST catalog and would fail at runtime. It is preserved here verbatim to - avoid silently changing behaviour, but is marked as broken. The base - ``upsert()`` (which delegates to ``operations.upsert()``) is the correct - path for REST catalogs. This override will be removed in Phase 6. - -NOTE on ``_list_tables``: - The legacy ``pyiceberg_rest_operations.py`` contained a classmethod version - of ``_list_tables`` that referenced ``cls.catalog_backend`` — which is an - instance property and cannot be accessed on the class. That implementation - was broken. The correct instance-method implementation is provided here, - delegating to ``self.catalog_backend.list_tables()``. -""" - -from __future__ import annotations - -import contextlib -import typing as t -import uuid - -import ibis.expr.types.relations as ir -from pydantic_settings import BaseSettings -from pyiceberg.catalog import Catalog - -from mountainash_settings import SettingsParameters -from mountainash_data.core.constants import CONST_DB_BACKEND -from mountainash_data.core.settings import PyIcebergRestBackendProfile -from mountainash_data.backends.iceberg.connection import IcebergConnectionBase - - -class IcebergRestConnection(IcebergConnectionBase): - """Concrete Iceberg connection for a REST catalog endpoint.""" - - def __init__( - self, - db_auth_settings_parameters: SettingsParameters, - connection_mode: t.Optional[str] = None, - ) -> None: - self._catalog_backend: t.Optional[Catalog] = None - self.supports_upsert = True - super().__init__(db_auth_settings_parameters=db_auth_settings_parameters) - - # ------------------------------------------------------------------ - # Abstract property implementations - # ------------------------------------------------------------------ - - @property - def catalog_backend(self) -> t.Optional[Catalog]: - return self._catalog_backend - - @property - def db_backend_name(self) -> str: - return CONST_DB_BACKEND.PYICEBERG - - @property - def settings_class(self) -> t.Type[BaseSettings]: - return PyIcebergRestBackendProfile - - # ------------------------------------------------------------------ - # List tables (instance-method version; fixes the broken classmethod - # that existed in the legacy pyiceberg_rest_operations.py) - # ------------------------------------------------------------------ - - def _list_tables( - self, - namespace: str | t.Tuple[str, ...] | None = None, - ) -> t.List[str]: - """Return table names within ``namespace`` from the REST catalog.""" - return ( - self.catalog_backend.list_tables(namespace=namespace) - if self.catalog_backend is not None - else [] - ) - - # ------------------------------------------------------------------ - # Legacy _upsert override (preserved verbatim from the REST legacy files; - # flagged as incorrect for REST catalogs — see module docstring). - # ------------------------------------------------------------------ - - def _upsert( - self, - table_name: str, - df: ir.Table | t.Any, - natural_key_columns: list[str] | str, - data_columns: list[str] | str, - database: str | None = None, - schema: str | None = None, - ) -> None: - """BROKEN: legacy SQL-cursor upsert copied from DuckDB implementation. - - This method uses DuckDB cursor semantics (``catalog_backend.con.cursor``) - which are not available on a pyiceberg RestCatalog. It is preserved here - for behavioural parity with the legacy code only. - - Consumers should call ``upsert()`` (the base class method) instead, - which delegates to ``operations.upsert()`` and uses native pyiceberg - merge semantics. - """ - if isinstance(natural_key_columns, str): - natural_key_columns = [natural_key_columns] - - if len(natural_key_columns) == 0: - raise ValueError("Natural Keys must be provided") - - if isinstance(data_columns, str): - data_columns = [data_columns] - - if len(data_columns) == 0: - raise ValueError("Data Columns must be provided") - - if not self.table_exists(table_name=table_name, database=database): - raise ValueError(f"Target Upsert table '{table_name}' does not exist") - - table_suffix: str = str(uuid.uuid4()).replace("-", "") - staging_table_name = f"temp_upsert_{table_suffix}" - - list_all_columns = natural_key_columns + data_columns - sql_all_columns = ", ".join(list_all_columns) if list_all_columns else "" - sql_natural_keys = ", ".join(natural_key_columns) if natural_key_columns else "" - sql_value_fields = ( - ", ".join([f"{col} = excluded.{col}" for col in data_columns]) - if data_columns - else "" - ) - - upsert_sql = ( - f"INSERT INTO {database}.{table_name}({sql_all_columns}) " - f"SELECT {sql_all_columns} FROM {staging_table_name} " - f"ON CONFLICT ({sql_natural_keys}) DO UPDATE SET {sql_value_fields}" - ) - - with contextlib.closing(self.catalog_backend.con.cursor()) as cur: - cur.execute("BEGIN TRANSACTION;") - cur.register(staging_table_name, df) - cur.execute(f"{upsert_sql}") - cur.unregister(staging_table_name) - cur.execute("COMMIT;") - - # ------------------------------------------------------------------ - # Index helpers (legacy, REST-specific; rarely used) - # ------------------------------------------------------------------ - - def unique_index_exists( - self, - table_name: str, - natural_key_columns: list[str], - database: str | None = None, - ) -> bool: - """Check if a unique index exists (legacy DuckDB-style check).""" - if not natural_key_columns: - return - - index_name = self.create_unique_index_name(table_name, natural_key_columns) - - check_index_sql = f""" - SELECT COUNT(*) as index_exists - FROM pg_catalog.pg_indexes - WHERE indexname = '{index_name}' - AND tablename = '{table_name}' - """ - - index_exists = ( - self.run_sql_as_catalog_dataframe(check_index_sql) - .get_column_as_list("index_exists")[0] - > 0 - ) - - return index_exists - - def create_unique_index( - self, - table_name: str, - natural_key_columns: list[str], - database: str | None = None, - ) -> bool: - """Create a unique index (legacy DuckDB-style operation).""" - if isinstance(natural_key_columns, str): - natural_key_columns = [natural_key_columns] - - if len(natural_key_columns) == 0: - raise ValueError("Natural Keys must be provided") - - index_exists = self.unique_index_exists( - table_name=table_name, - natural_key_columns=natural_key_columns, - database=database, - ) - - if not index_exists: - qualified_table = table_name - if database: - qualified_table = f"{database}.{qualified_table}" - - index_name = self.create_unique_index_name(table_name, natural_key_columns) - sql_natural_keys = ", ".join(natural_key_columns) - create_index_sql = ( - f"CREATE UNIQUE INDEX {index_name} " - f"ON {qualified_table} ({sql_natural_keys});" - ) - - with contextlib.closing(self.catalog_backend.con.cursor()) as cur: - cur.execute(create_index_sql) - - def create_unique_index_name( - self, - table_name: str, - natural_key_columns: list[str], - ) -> str: - """Generate a canonical index name from table and key columns.""" - natural_key_columns.sort() - return f"idx_{table_name}_{'_'.join(natural_key_columns)}" diff --git a/src/mountainash_data/backends/iceberg/connection.py b/src/mountainash_data/backends/iceberg/connection.py deleted file mode 100644 index ff7ec24..0000000 --- a/src/mountainash_data/backends/iceberg/connection.py +++ /dev/null @@ -1,617 +0,0 @@ -"""Iceberg connection: catalog/namespace lifecycle and read-side inspection. - -Created in Phase 3 by deduplicating and splitting the legacy -base_pyiceberg_connection.py and base_pyiceberg_operations.py. - -This module provides ``IcebergConnectionBase``, the abstract base for all -Iceberg catalog connections. It handles: -- Lifecycle: open/close/reconnect -- Catalog and namespace accessors -- Table loading with retry (``table()`` / ``get_schema()``) -- Inspection: ``list_namespaces``, ``list_tables``, ``inspect_table``, - ``inspect_namespace``, ``inspect_catalog`` -- Schema caching -- Thin delegation wrappers to ``operations.py`` for all mutations - -There is deliberately no ``to_relation()`` here (or anywhere in this -package): bridging to the unified ``mountainash`` package is pull-side — -``ma.relation(...)`` over a backend-native handle. For Iceberg that handle -is not an ibis Table; reaching mountainash goes via an Arrow scan of the -loaded ``table()``, or via an Ibis engine reading the catalog. -""" - -from __future__ import annotations - -import typing as t -from abc import abstractmethod -from time import sleep - -from mountainash_data.core.connection import BaseDBConnection -from mountainash_data.core.constants import ( - CONST_DB_ABSTRACTION_LAYER, - CONST_DB_PROVIDER_TYPE, -) -from mountainash_data.core.inspection import ( - CatalogInfo, - NamespaceInfo, - TableInfo, -) -from mountainash_data.core.namespace import Namespace, NamespaceLike - -from mountainash_data.core.factories.connection_factory import build_driver_kwargs -from mountainash_dataframes import DataFrameUtils, SupportedDataFrames -from mountainash_dataframes.constants import CONST_DATAFRAME_FRAMEWORK -from mountainash_settings import SettingsParameters - -from pyiceberg.catalog import Catalog -from pyiceberg.catalog.rest import RestCatalog -from pyiceberg.partitioning import PartitionSpec -from pyiceberg.schema import Schema -from pyiceberg.table import Table -from pyiceberg.table.sorting import SortOrder - - -class IcebergConnectionBase(BaseDBConnection): - """Abstract base for Iceberg catalog connections. - - Concrete subclasses (e.g. ``IcebergRestConnection``) implement - ``catalog_backend``, ``db_backend_name``, and ``settings_class``. - """ - - def __init__( - self, - db_auth_settings_parameters: SettingsParameters, - ) -> None: - super().__init__(db_auth_settings_parameters=db_auth_settings_parameters) - self._schema_cache: dict = {} - - # ------------------------------------------------------------------ - # Abstract properties (implemented by concrete subclasses) - # ------------------------------------------------------------------ - - @property - @abstractmethod - def catalog_backend(self) -> t.Optional[Catalog | t.Any]: - """The live pyiceberg Catalog handle, or None if not yet connected.""" - ... - - # ------------------------------------------------------------------ - # BaseDBConnection abstract properties - # ------------------------------------------------------------------ - - @property - def db_abstraction_layer(self) -> CONST_DB_ABSTRACTION_LAYER: - return CONST_DB_ABSTRACTION_LAYER.PYICEBERG - - @property - def provider_type(self) -> CONST_DB_PROVIDER_TYPE: - """Database provider identifier.""" - return CONST_DB_PROVIDER_TYPE.PYICEBERG_REST - - # ------------------------------------------------------------------ - # Lifecycle - # ------------------------------------------------------------------ - - def connect( - self, - connection_string: t.Optional[str] = None, - connection_kwargs: t.Optional[t.Dict[str, t.Any]] = None, - *, - auth_profile: t.Any = None, - **kwargs: t.Any, - ) -> Catalog: - """Ensure a catalog connection is open, returning the backend handle. - - Idempotent. ``auth_profile`` is L2 credential data (a *AuthProfile) - composed onto the catalog config at connect time. Precedence: - profile-derived config < explicit ``connection_kwargs``/``kwargs``. - """ - if self.catalog_backend is None: - self.connect_default( - auth_profile=auth_profile, **(connection_kwargs or {}), **kwargs - ) - return self.catalog_backend - - def connect_default(self, *, auth_profile: t.Any = None, **kwargs: t.Any) -> Catalog: - """Connect using settings credentials plus an optional auth profile. - - Precedence: profile-derived config < explicit ``kwargs``. - """ - if self.catalog_backend is None: - connection_kwargs = self._build_catalog_kwargs(auth_profile, **kwargs) - self._catalog_backend: RestCatalog = RestCatalog(**connection_kwargs) - return self.catalog_backend - - def _build_catalog_kwargs(self, auth_profile: t.Any = None, **kwargs: t.Any) -> dict: - """Build RestCatalog kwargs from settings + auth (no pyiceberg use here). - - Explicit ``kwargs`` override profile-derived config. - """ - settings_class = self.db_auth_settings_parameters.settings_class - if settings_class is None: - raise ValueError("Settings class is required for the database connection") - obj_settings = settings_class.get_settings( - settings_parameters=self.db_auth_settings_parameters - ) - connection_kwargs = build_driver_kwargs(obj_settings, auth_profile) - connection_kwargs.update(kwargs) - return connection_kwargs - - def _connect( - self, - connection_kwargs: t.Optional[t.Dict[str, str]], - ) -> Catalog: - """Low-level connect hook. Raise NotImplementedError by default.""" - raise NotImplementedError - - def close(self) -> None: - """Release the connection. Idempotent.""" - self.disconnect() - - def disconnect(self) -> None: - """Close the connection to the catalog.""" - if self.catalog_backend is not None: - self._catalog_backend = None - - def is_connected(self) -> bool: - """Return True if a live catalog handle exists.""" - return self.catalog_backend is not None - - # ------------------------------------------------------------------ - # Schema cache - # ------------------------------------------------------------------ - - def get_schema( - self, - table_name: str | t.Tuple[str, ...], - refresh: bool = False, - ) -> t.Optional[Schema]: - """Return the Iceberg schema for ``table_name``, with caching. - - Args: - table_name: Table identifier. - refresh: Force a cache bypass when True. - - Returns: - ``Schema`` object, or None if the table cannot be loaded. - """ - if not refresh and table_name in self._schema_cache: - return self._schema_cache[table_name] - - table_ref = self.table(table_name) - if table_ref is None: - return None - - try: - schema = table_ref.schema() - self._schema_cache[table_name] = schema - return schema - except Exception as e: - print(f"Error getting schema for {table_name}: {e}") - return None - - def clear_schema_cache( - self, - table_name: t.Optional[str | t.Tuple[str, ...]] = None, - ) -> None: - """Clear schema cache for a specific table or all tables. - - Args: - table_name: If provided, clear only that entry. If None, - clear the whole cache. - """ - if table_name: - if table_name in self._schema_cache: - del self._schema_cache[table_name] - else: - self._schema_cache.clear() - - # ------------------------------------------------------------------ - # Table loading - # ------------------------------------------------------------------ - - def table( - self, - table_name: str | t.Tuple[str, ...], - max_attempts: int = 3, - retry_delay: float = 0.5, - ) -> t.Optional[Table]: - """Load a table reference with built-in retry logic. - - Args: - table_name: Table identifier. - max_attempts: Maximum number of load attempts. - retry_delay: Seconds to wait between retries. - - Returns: - ``Table`` reference, or None after all attempts fail. - """ - self.connect() - - for attempt in range(max_attempts): - table_ref = self.catalog_backend.load_table(table_name) - - if table_ref is not None: - return table_ref - - if attempt < max_attempts - 1: - print( - f"Table reference for {table_name} returned None " - f"(attempt {attempt + 1}/{max_attempts}), retrying..." - ) - sleep(retry_delay) - - print( - f"Failed to get table reference for {table_name} " - f"after {max_attempts} attempts" - ) - return None - - # Alias used by the protocol and some legacy callers - def load_table( - self, - table_name: str | t.Tuple[str, ...], - ) -> t.Optional[Table]: - """Alias for ``table()`` with default retry settings.""" - return self.table(table_name) - - # ------------------------------------------------------------------ - # Inspection (satisfies core.protocol.Connection) - # ------------------------------------------------------------------ - - def _catalog_name(self) -> str: - return getattr(self.catalog_backend, "name", "iceberg") - - def _check_catalog(self, catalog: str | None) -> None: - """iceberg's catalog IS the connection; a foreign catalog is unaddressable.""" - if catalog is not None and catalog != self._catalog_name(): - raise ValueError( - f"iceberg connection is bound to catalog {self._catalog_name()!r}; " - f"cannot address catalog {catalog!r}." - ) - - def list_catalogs(self) -> list[str]: - """One catalog per iceberg connection.""" - self.connect() - return [self._catalog_name()] - - def list_namespaces(self, catalog: str | None = None) -> list[str]: - """Return all namespaces visible to this catalog connection. - - Args: - catalog: Must match this connection's catalog name, or be None. - - Returns: - List of namespace names (dotted for multi-level). - """ - self.connect() - self._check_catalog(catalog) - if self.catalog_backend is None: - return [] - raw = self.catalog_backend.list_namespaces() - # DEFERRED (DEBT-11): multi-level namespaces are joined to a dotted string; - # this is NOT round-trippable as a bare string. Structured round-trip is - # tracked in DEBT-11 (list[Namespace] vs dotted list[str] fork). - return [".".join(ns) if isinstance(ns, (tuple, list)) else str(ns) for ns in raw] - - def list_tables(self, namespace: NamespaceLike = None) -> list[str]: - """Return the names of tables in ``namespace``. - - Args: - namespace: Namespace to list. Backend-specific subclasses - implement ``_list_tables`` to provide the actual listing. - - Returns: - List of table name strings. - """ - self.connect() - ns = Namespace.coerce(namespace) - self._check_catalog(ns.catalog) - # pyiceberg accepts a namespace tuple directly (behavior-preserving). - return self._list_tables(namespace=(ns.path or None)) - - def _list_tables( - self, - namespace: str | t.Tuple[str, ...] | None = None, - ) -> list[str]: - """Hook for subclasses to implement namespace-scoped table listing.""" - raise NotImplementedError - - def inspect_table( - self, - name: str, - namespace: NamespaceLike = None, - ) -> TableInfo: - """Return shared-model metadata for one table. - - Args: - name: Simple table name. - namespace: Namespace the table belongs to, if known. - - Returns: - ``TableInfo`` populated from the live table schema. - """ - from mountainash_data.backends.iceberg.inspect import table_to_info - - ns = Namespace.coerce(namespace) - self._check_catalog(ns.catalog) - identifier = (*ns.path, name) if ns.path else name - iceberg_table = self.table(identifier) - if iceberg_table is None: - raise ValueError(f"Table not found: {identifier!r}") - return table_to_info(iceberg_table, name=name, location=ns) - - def inspect_namespace(self, name: str) -> NamespaceInfo: - """Return shared-model metadata for one namespace. - - Args: - name: Namespace identifier. - - Returns: - ``NamespaceInfo`` with table names discovered from the catalog. - """ - from mountainash_data.backends.iceberg.inspect import namespace_to_info - - ns = Namespace.coerce(name) - self._check_catalog(ns.catalog) - table_names = self._list_tables(namespace=(ns.path or None)) - return namespace_to_info(ns.path, table_names) - - def inspect_catalog(self, catalog: str | None = None) -> CatalogInfo: - """Return shared-model metadata for the connection's catalog. - - Returns: - ``CatalogInfo`` containing all namespaces and their tables. - """ - from mountainash_data.backends.iceberg.inspect import ( - catalog_to_info, - namespace_to_info, - ) - - self.connect() - self._check_catalog(catalog) - catalog_name = self._catalog_name() - raw_namespaces = self.catalog_backend.list_namespaces() - - namespace_infos = [] - for ns in raw_namespaces: - # DEFERRED (DEBT-11): deep namespaces are flattened to their first - # segment here, mirroring current behavior. Faithful round-trip is DEBT-11. - ns_name = ns[0] if isinstance(ns, (tuple, list)) else str(ns) - try: - table_names = self._list_tables(namespace=(ns_name,)) - except NotImplementedError: - table_names = [] - namespace_infos.append(namespace_to_info((ns_name,), table_names)) - - return catalog_to_info(catalog_name, namespace_infos) - - # ------------------------------------------------------------------ - # Table and view existence checks - # ------------------------------------------------------------------ - - def table_exists( - self, - table_name: str | t.Tuple[str, ...] | None = None, - ) -> bool: - """Return True if the table exists in the catalog.""" - self.connect() - return ( - self.catalog_backend.table_exists(table_name) - if self.catalog_backend is not None - else None - ) - - def view_exists( - self, - view_name: str | t.Tuple[str, ...] | None = None, - ) -> bool: - """Return True if the view exists in the catalog.""" - self.connect() - return ( - self.catalog_backend.view_exists(view_name) - if self.catalog_backend is not None - else None - ) - - def rename_table( - self, - old_name: str, - new_name: str, - ) -> None: - """Rename a table in the catalog.""" - self.connect() - return ( - self._rename_table(old_name=old_name, new_name=new_name) - if self.catalog_backend is not None - else None - ) - - def _rename_table(self, old_name: str, new_name: str) -> None: - """Hook for subclasses to implement table renaming.""" - raise NotImplementedError - - # ------------------------------------------------------------------ - # SQL passthrough (not supported) - # ------------------------------------------------------------------ - - def run_sql( - self, - query: str, - schema: Schema | None = None, - dialect: str | None = None, - ) -> t.Optional[Table]: - """Raise NotImplementedError — Iceberg does not support raw SQL.""" - raise NotImplementedError - - # ------------------------------------------------------------------ - # DataFrame accessors - # ------------------------------------------------------------------ - - def table_as_ibis_dataframe( - self, - table_name: str, - tablename_prefix: t.Optional[str] = None, - ) -> t.Optional[SupportedDataFrames]: - """Return the table as an Ibis-compatible dataframe. - - Args: - table_name: Simple table name. - tablename_prefix: Optional prefix for Ibis table registration. - - Returns: - Ibis-backed dataframe, or None if the table could not be loaded. - """ - result: Table | None = self.table(table_name=table_name) - return DataFrameUtils.to_ibis(result, tablename_prefix=tablename_prefix) - - def table_as_polars_dataframe( - self, - table_name: str, - tablename_prefix: t.Optional[str] = None, - dataframe_framework: t.Optional[str] = CONST_DATAFRAME_FRAMEWORK.POLARS, - ) -> t.Optional[SupportedDataFrames]: - """Return the table as a Polars dataframe. - - Args: - table_name: Simple table name. - tablename_prefix: Unused; kept for API symmetry with ibis accessor. - dataframe_framework: Target framework; defaults to POLARS. - - Returns: - Dataframe in the requested framework, or None if not loaded. - """ - result: Table | None = self.table(table_name=table_name) - if result is None: - return None - - if dataframe_framework is None: - dataframe_framework = CONST_DATAFRAME_FRAMEWORK.POLARS - - return DataFrameUtils.cast_dataframe(result, dataframe_framework=dataframe_framework) - - def table_as_native_dataframe( - self, - object_name: str, - schema: str | None = None, - database: str | None = None, - dataframe_framework: t.Optional[str] = CONST_DATAFRAME_FRAMEWORK.POLARS, - ) -> t.Optional[SupportedDataFrames]: - """Return the table as a native dataframe in the specified framework.""" - return self.table_as_polars_dataframe( - table_name=object_name, - dataframe_framework=dataframe_framework, - ) - - # ------------------------------------------------------------------ - # Mutation delegation wrappers (delegate to operations.py) - # All wrappers use a local import to avoid circular dependencies. - # ------------------------------------------------------------------ - - def create_table( - self, - table_name: str | t.Tuple[str, ...], - schema: Schema, - df: t.Optional[t.Any] = None, - location: str | None = None, - partition_spec: t.Optional[PartitionSpec] = None, - sort_order: t.Optional[SortOrder] = None, - overwrite: t.Optional[bool] = False, - ) -> t.Optional[Table]: - """Create an Iceberg table. Delegates to ``operations.create_table``.""" - from mountainash_data.backends.iceberg import operations - - return operations.create_table( - self, - table_name, - schema, - df=df, - location=location, - partition_spec=partition_spec, - sort_order=sort_order, - overwrite=overwrite, - ) - - def drop_table( - self, - table_name: str | t.Tuple[str, ...], - purge: t.Optional[bool] = False, - ) -> bool: - """Drop an Iceberg table. Delegates to ``operations.drop_table``.""" - from mountainash_data.backends.iceberg import operations - - return operations.drop_table(self, table_name, purge=purge) - - def insert( - self, - table_name: str | t.Tuple[str, ...], - df: t.Any, - prevent_duplicates: t.Optional[bool] = False, - ) -> bool: - """Insert data. Delegates to ``operations.insert``.""" - from mountainash_data.backends.iceberg import operations - - return operations.insert(self, table_name, df, prevent_duplicates) - - def upsert( - self, - table_name: str | t.Tuple[str, ...], - df: t.Any, - natural_key_columns: list[str] | None = None, - when_matched_update_all: bool = True, - when_not_matched_insert_all: bool = True, - case_sensitive: bool = True, - ) -> None: - """Upsert data. Delegates to ``operations.upsert``.""" - from mountainash_data.backends.iceberg import operations - - return operations.upsert( - self, - table_name, - df, - natural_key_columns=natural_key_columns, - when_matched_update_all=when_matched_update_all, - when_not_matched_insert_all=when_not_matched_insert_all, - case_sensitive=case_sensitive, - ) - - def truncate( - self, - table_name: str | t.Tuple[str, ...], - ) -> None: - """Truncate a table. Delegates to ``operations.truncate``.""" - from mountainash_data.backends.iceberg import operations - - return operations.truncate(self, table_name) - - def create_view( - self, - view_name: str | t.Tuple[str, ...], - ) -> None: - """Create a view. Delegates to ``operations.create_view``.""" - from mountainash_data.backends.iceberg import operations - - return operations.create_view(self, view_name) - - def drop_view( - self, - view_name: str | t.Tuple[str, ...], - ) -> bool: - """Drop a view. Delegates to ``operations.drop_view``.""" - from mountainash_data.backends.iceberg import operations - - return operations.drop_view(self, view_name) - - # ------------------------------------------------------------------ - # Retry helper (exposed for subclass use) - # ------------------------------------------------------------------ - - def retry_operation( - self, - operation_func: t.Callable, - max_attempts: int = 3, - ) -> t.Any: - """Retry a PyIceberg operation that might fail due to commit conflicts.""" - from mountainash_data.backends.iceberg import operations - - return operations.retry_operation(operation_func, max_attempts) diff --git a/src/mountainash_data/backends/iceberg/inspect.py b/src/mountainash_data/backends/iceberg/inspect.py deleted file mode 100644 index b6bf45e..0000000 --- a/src/mountainash_data/backends/iceberg/inspect.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Iceberg to core.inspection conversion. - -Helpers that take pyiceberg Table objects and produce TableInfo / -NamespaceInfo / CatalogInfo dataclasses from core.inspection. -""" - -from __future__ import annotations - -import typing as t - -from mountainash_data.core.inspection import ( - CatalogInfo, - ColumnInfo, - NamespaceInfo, - TableInfo, -) -from mountainash_data.core.namespace import Namespace - - -def table_to_info( - iceberg_table, - *, - name: str, - location: Namespace = Namespace(), -) -> TableInfo: - """Convert a pyiceberg Table object into a TableInfo. - - Args: - iceberg_table: A ``pyiceberg.table.Table`` instance. - name: The simple table name (without namespace prefix). - location: The table's namespace/catalog location. - - Returns: - A ``TableInfo`` populated from the table's current schema. - """ - columns = [ - ColumnInfo( - name=field.name, - type_name=str(field.field_type), - nullable=not field.required, - ) - for field in iceberg_table.schema().fields - ] - return TableInfo(name=name, columns=columns, location=location) - - -def namespace_to_info( - namespace_path: t.Sequence[str], - table_names: t.Sequence[str], -) -> NamespaceInfo: - """Build a NamespaceInfo from a namespace path and its table names. - - Args: - namespace_path: The namespace path segments. - table_names: Names of tables within this namespace. - - Returns: - A populated ``NamespaceInfo``. - """ - return NamespaceInfo( - location=Namespace(path=tuple(namespace_path)), - tables=list(table_names), - ) - - -def catalog_to_info( - catalog_name: str, - namespace_infos: t.Sequence[NamespaceInfo], -) -> CatalogInfo: - """Build a CatalogInfo from a sequence of NamespaceInfo objects. - - Args: - catalog_name: The catalog identifier. - namespace_infos: Pre-built NamespaceInfo objects for each namespace. - - Returns: - A populated ``CatalogInfo``. - """ - return CatalogInfo( - name=catalog_name, - namespaces=list(namespace_infos), - ) diff --git a/src/mountainash_data/backends/iceberg/operations.py b/src/mountainash_data/backends/iceberg/operations.py deleted file mode 100644 index 3b1d6bb..0000000 --- a/src/mountainash_data/backends/iceberg/operations.py +++ /dev/null @@ -1,434 +0,0 @@ -"""Iceberg table mutations: create, drop, insert, upsert, truncate, view ops. - -Created in Phase 3 by deduplicating and splitting the legacy -base_pyiceberg_connection.py and base_pyiceberg_operations.py. - -These functions take an active connection (IcebergConnectionBase instance) and -perform mutations. They are NOT class methods — they are module-level functions -that the connection class delegates to when consumers call e.g. -``connection.insert(...)``. - -The ``connection`` parameter in each function is an IcebergConnectionBase -instance (typed as Any to avoid a circular import; the structural contract is -that it exposes ``.catalog_backend``, ``.table()``, ``.table_exists()``, and -``._schema_cache``). -""" - -from __future__ import annotations - -import typing as t -from datetime import date, datetime -from time import sleep - -import pyarrow as pa -from mountainash_dataframes import DataFrameUtils, SupportedDataFrames -from pyiceberg.partitioning import PartitionSpec -from pyiceberg.schema import Schema -from pyiceberg.table import Table -from pyiceberg.table.sorting import SortOrder - -from mountainash_data.backends.iceberg._types import iceberg_type_to_pyarrow - - -# --------------------------------------------------------------------------- -# DataFrame preparation helper -# --------------------------------------------------------------------------- - - -def prepare_dataframe_for_iceberg( - df: SupportedDataFrames, - target_schema: t.Optional[Schema] = None, - target_table: t.Optional[Table] = None, -) -> pa.Table: - """Preprocess a dataframe to match Iceberg table schema requirements. - - Args: - df: Input dataframe (any SupportedDataFrames type). - target_schema: Iceberg schema to cast to. If None and - ``target_table`` is supplied, the schema is read from the table. - target_table: Iceberg table to extract schema from when - ``target_schema`` is None. - - Returns: - PyArrow table cast to the target schema. - """ - if target_schema is None and target_table is None: - return DataFrameUtils.to_pyarrow(df=df) - - if target_schema is None and target_table is not None: - target_schema = target_table.schema() - - df = DataFrameUtils.to_pyarrow(df=df) - - pa_schema = pa.schema([]) - - for field in target_schema.fields: - iceberg_type = field.field_type - iceberg_nullable = not field.required - pa_type = iceberg_type_to_pyarrow(iceberg_type) - pa_schema = pa_schema.append(pa.field(field.name, pa_type, nullable=iceberg_nullable)) - - # Cast the dataframe to the new schema - cast_arrays = [] - for field in pa_schema: - if field.name in df.column_names: - col_index = df.column_names.index(field.name) - - if pa.types.is_timestamp(field.type) and not pa.types.is_timestamp( - df.column(col_index).type - ): - try: - values = df.column(col_index).to_pylist() - converted = [ - datetime.fromisoformat(str(v)) if v is not None else None - for v in values - ] - cast_arrays.append(pa.array(converted, type=field.type)) - except Exception: - cast_arrays.append(df.column(col_index).cast(field.type, safe=True)) - elif pa.types.is_date(field.type) and not pa.types.is_date( - df.column(col_index).type - ): - try: - values = df.column(col_index).to_pylist() - converted = [ - date.fromisoformat(str(v)) if v is not None else None - for v in values - ] - cast_arrays.append(pa.array(converted, type=field.type)) - except Exception: - cast_arrays.append(df.column(col_index).cast(field.type, safe=True)) - elif pa.types.is_decimal(field.type): - try: - values = df.column(col_index).to_pylist() - from decimal import Decimal - - converted = [ - Decimal(str(v)) if v is not None else None for v in values - ] - cast_arrays.append(pa.array(converted, type=field.type)) - except Exception: - cast_arrays.append(df.column(col_index).cast(field.type, safe=True)) - else: - cast_arrays.append(df.column(col_index).cast(field.type, safe=True)) - else: - # Create empty column for missing fields - cast_arrays.append(pa.array([None] * len(df), type=field.type)) - - return pa.Table.from_arrays(cast_arrays, schema=pa_schema) - - -# --------------------------------------------------------------------------- -# Retry helper -# --------------------------------------------------------------------------- - - -def retry_operation(operation_func: t.Callable, max_attempts: int = 3) -> t.Any: - """Retry a PyIceberg operation that might fail due to commit conflicts. - - Args: - operation_func: Zero-argument callable to retry. - max_attempts: Maximum number of attempts. - - Returns: - Return value of ``operation_func`` on success. - - Raises: - RuntimeError: If all attempts fail. - """ - attempts = 0 - last_error = None - - while attempts < max_attempts: - try: - return operation_func() - except Exception as e: - attempts += 1 - print(e) - if attempts >= max_attempts: - break - sleep(0.2) - - raise last_error if last_error else RuntimeError("Operation failed for unknown reason") - - -# --------------------------------------------------------------------------- -# Table mutations -# --------------------------------------------------------------------------- - - -def create_table( - connection: t.Any, - table_name: str | t.Tuple[str, ...], - schema: Schema, - df: t.Optional[t.Any] = None, - location: str | None = None, - partition_spec: t.Optional[PartitionSpec] = None, - sort_order: t.Optional[SortOrder] = None, - overwrite: t.Optional[bool] = False, -) -> t.Optional[Table]: - """Create an Iceberg table. - - Args: - connection: Active ``IcebergConnectionBase`` instance. - table_name: Table identifier (string or tuple of namespace parts). - schema: Iceberg schema for the new table. - df: Optional dataframe to append immediately after creation. - location: Optional storage location override. - partition_spec: Optional partition specification. - sort_order: Optional sort order. - overwrite: If True, drop an existing table first. - - Returns: - The created Iceberg Table object, or None if backend unavailable. - """ - connection.connect() - - params: dict[str, t.Any] = { - "identifier": table_name, - "schema": schema, - } - - if location is not None: - params["location"] = location - if partition_spec is not None: - params["partition_spec"] = partition_spec - if sort_order is not None: - params["sort_order"] = sort_order - - if not overwrite and connection.table_exists(table_name): - print( - f"Cannot create table - table already exists: {table_name}. " - "Set overwrite=True." - ) - return False - - if overwrite: - connection.catalog_backend.drop_table(table_name) - - obj_table = ( - connection.catalog_backend.create_table(**params) - if connection.catalog_backend is not None - else None - ) - - # Refresh the schema cache after recreating the table - connection.get_schema(table_name, refresh=True) - - if df is not None: - pa_df = prepare_dataframe_for_iceberg(df, target_schema=schema) - obj_table.append(pa_df) - - return obj_table - - -def drop_table( - connection: t.Any, - table_name: str | t.Tuple[str, ...], - purge: t.Optional[bool] = False, -) -> bool: - """Drop an Iceberg table. - - Args: - connection: Active ``IcebergConnectionBase`` instance. - table_name: Table identifier. - purge: If True, purge data files as well. - - Returns: - True if the table was dropped, False otherwise. - """ - connection.connect() - - try: - if connection.table_exists(table_name): - if purge: - ( - connection.catalog_backend.purge_table(table_name) - if connection.catalog_backend is not None - else None - ) - else: - ( - connection.catalog_backend.drop_table(table_name) - if connection.catalog_backend is not None - else None - ) - return True - return False - except Exception: - return False - - -def insert( - connection: t.Any, - table_name: str | t.Tuple[str, ...], - df: t.Any, - prevent_duplicates: t.Optional[bool] = False, -) -> bool: - """Insert (append) data into an Iceberg table. - - Args: - connection: Active ``IcebergConnectionBase`` instance. - table_name: Table identifier. - df: Input dataframe (any SupportedDataFrames type). - prevent_duplicates: If True, use upsert semantics that skip - existing rows instead of appending. - - Returns: - True on success, False on failure. - """ - connection.connect() - - try: - schema = connection.get_schema(table_name) - pa_df = prepare_dataframe_for_iceberg(df, target_schema=schema) - table = connection.table(table_name) - - if prevent_duplicates: - print( - f"NOTE: insert operation on {table_name} may have prevented " - "duplicates being added to the table. Records matching the " - "existing keys were dropped. If you wish to perform a full " - "insert set 'prevent_duplicates=False'. If you wish to also " - "update existing rows use upsert()" - ) - ( - table.upsert( - df=pa_df, - when_matched_update_all=False, - when_not_matched_insert_all=True, - ) - if connection.catalog_backend is not None - else None - ) - else: - print( - f"NOTE: insert operation on {table_name} performing a straight " - "append, which may cause duplicates to be added to the table. " - "To avoid duplicates set 'prevent_duplicates=True'. To update " - "rows that match existing natural keys use upsert()" - ) - ( - table.append(df=pa_df) - if connection.catalog_backend is not None - else None - ) - - return True - except Exception: - return False - - -def upsert( - connection: t.Any, - table_name: str | t.Tuple[str, ...], - df: t.Any, - natural_key_columns: list[str] | None = None, - when_matched_update_all: bool = True, - when_not_matched_insert_all: bool = True, - case_sensitive: bool = True, -) -> None: - """Upsert data into an Iceberg table. - - Args: - connection: Active ``IcebergConnectionBase`` instance. - table_name: Table identifier. - df: Input dataframe (any SupportedDataFrames type). - natural_key_columns: Column(s) used to match existing rows. - Maps to ``join_cols`` in pyiceberg's ``Table.upsert()``. - when_matched_update_all: Update matched rows (default True). - when_not_matched_insert_all: Insert unmatched rows (default True). - case_sensitive: Case-sensitive key comparison (default True). - """ - connection.connect() - - schema = connection.get_schema(table_name) - pa_df = prepare_dataframe_for_iceberg(df, target_schema=schema) - - params: dict[str, t.Any] = { - "df": pa_df, - "when_matched_update_all": when_matched_update_all, - "when_not_matched_insert_all": when_not_matched_insert_all, - "case_sensitive": case_sensitive, - } - - if natural_key_columns is not None: - params["join_cols"] = natural_key_columns - - table = connection.table(table_name) - ( - table.upsert(**params) - if connection.catalog_backend is not None - else None - ) - - -def truncate( - connection: t.Any, - table_name: str | t.Tuple[str, ...], -) -> None: - """Truncate an Iceberg table (not yet implemented). - - Args: - connection: Active ``IcebergConnectionBase`` instance. - table_name: Table identifier. - - Raises: - NotImplementedError: Always — pyiceberg does not expose a native - truncate; this requires deleting all files manually. - """ - raise NotImplementedError( - "truncate() is not implemented for the Iceberg backend. " - "pyiceberg does not expose a native truncate operation." - ) - - -# --------------------------------------------------------------------------- -# View operations -# --------------------------------------------------------------------------- - - -def create_view( - connection: t.Any, - view_name: str | t.Tuple[str, ...], -) -> None: - """Create an Iceberg view (not yet implemented). - - Args: - connection: Active ``IcebergConnectionBase`` instance. - view_name: View identifier. - - Raises: - NotImplementedError: Always — view creation is not yet wired up. - """ - raise NotImplementedError( - "create_view() is not implemented for the Iceberg backend." - ) - - -def drop_view( - connection: t.Any, - view_name: str | t.Tuple[str, ...], -) -> bool: - """Drop an Iceberg view. - - Args: - connection: Active ``IcebergConnectionBase`` instance. - view_name: View identifier. - - Returns: - True if the view was dropped, False otherwise. - """ - connection.connect() - - try: - if connection.view_exists(view_name): - ( - connection.catalog_backend.drop_view(view_name) - if connection.catalog_backend is not None - else None - ) - return True - return False - except Exception: - return False diff --git a/src/mountainash_data/core/_warn.py b/src/mountainash_data/core/_warn.py index 882d86a..2fa0152 100644 --- a/src/mountainash_data/core/_warn.py +++ b/src/mountainash_data/core/_warn.py @@ -1,8 +1,8 @@ """Process-wide "warn at most once per key" helper (Gap 3, fable finding 6). -Shared by the ibis transaction machinery and the iceberg backend so a no-op -transaction() on an unsupported backend warns once per dialect, not per call. -Lives in core/ so neither backend imports the other. +Shared by the ibis transaction machinery so a no-op transaction() on an +unsupported backend warns once per dialect, not per call. Lives in core/ so +backends don't need to import each other. """ from __future__ import annotations diff --git a/src/mountainash_data/core/constants.py b/src/mountainash_data/core/constants.py index 90bb391..6fb5b19 100644 --- a/src/mountainash_data/core/constants.py +++ b/src/mountainash_data/core/constants.py @@ -21,7 +21,6 @@ class CONST_DB_PROVIDER_TYPE(Enum): DUCKDB =auto() MOTHERDUCK = auto() TRINO = auto() - PYICEBERG_REST = auto() ORACLE = auto() CLICKHOUSE = auto() DATABRICKS = auto() @@ -85,7 +84,6 @@ class CONST_DB_ABSTRACTION_LAYER(Enum): - FUGUE (str): Fugue database abstraction layer. """ IBIS = auto() - PYICEBERG = auto() class CONST_DB_BACKEND(StrEnum): """ @@ -124,7 +122,6 @@ class CONST_DB_BACKEND(StrEnum): MATERIALIZE = "MATERIALIZE" RISINGWAVE = "RISINGWAVE" DRUID = "DRUID" - PYICEBERG = "PYICEBERG" # POLARS = "POLARS" # PANDAS = "PANDAS" diff --git a/src/mountainash_data/core/inspection.py b/src/mountainash_data/core/inspection.py index b35e171..f9b8f65 100644 --- a/src/mountainash_data/core/inspection.py +++ b/src/mountainash_data/core/inspection.py @@ -1,6 +1,6 @@ """Shared physical-layer metadata model. -Both ibis and iceberg backends populate these dataclasses from their +Backends populate these dataclasses from their native introspection APIs, giving consumers a uniform shape regardless of which backend produced them. """ diff --git a/src/mountainash_data/core/registry.py b/src/mountainash_data/core/registry.py index 7a7ed18..41075b4 100644 --- a/src/mountainash_data/core/registry.py +++ b/src/mountainash_data/core/registry.py @@ -1,6 +1,7 @@ -"""Backend registry — populated in Phase 4 once IbisBackend and -IcebergBackend exist. This module is intentionally a placeholder for -now so that imports from core.registry don't break across phases.""" +"""Backend registry — name -> factory lookup for Backend implementations. + +Backend-agnostic infrastructure: any backend can register itself here so +callers can construct one by name without importing its module.""" from __future__ import annotations diff --git a/src/mountainash_data/core/settings/__init__.py b/src/mountainash_data/core/settings/__init__.py index 6b7f8f7..a53f22e 100644 --- a/src/mountainash_data/core/settings/__init__.py +++ b/src/mountainash_data/core/settings/__init__.py @@ -41,7 +41,6 @@ from .materialize import MaterializeBackendProfile from .risingwave import RisingWaveBackendProfile from .druid import DruidBackendProfile -from .pyiceberg_rest import PyIcebergRestBackendProfile import warnings as _warnings @@ -83,5 +82,4 @@ def __getattr__(name: str): "PySparkBackendProfile", "TrinoBackendProfile", "ExasolBackendProfile", "ImpalaBackendProfile", "MaterializeBackendProfile", "RisingWaveBackendProfile", "DruidBackendProfile", - "PyIcebergRestBackendProfile", ] diff --git a/src/mountainash_data/core/settings/adapters/pyiceberg_rest.py b/src/mountainash_data/core/settings/adapters/pyiceberg_rest.py deleted file mode 100644 index 5a4ebd3..0000000 --- a/src/mountainash_data/core/settings/adapters/pyiceberg_rest.py +++ /dev/null @@ -1,15 +0,0 @@ -"""PyIceberg REST adapters.""" -from __future__ import annotations -import typing as t - - -def headers_compose(profile: t.Any, base: dict[str, t.Any]) -> dict[str, t.Any]: - out = dict(base) - if profile.HEADERS: - for hk, hv in profile.HEADERS.items(): - out[f"header.{hk}"] = hv - return out - - -def token(auth, base): - return {**base, "token": auth.TOKEN.get_secret_value()} diff --git a/src/mountainash_data/core/settings/adapters/registry.py b/src/mountainash_data/core/settings/adapters/registry.py index 0221ca3..7124f56 100644 --- a/src/mountainash_data/core/settings/adapters/registry.py +++ b/src/mountainash_data/core/settings/adapters/registry.py @@ -9,7 +9,7 @@ ) from mountainash_data.core.constants import CONST_DB_PROVIDER_TYPE as P from . import (sql as _sql, trino as _trino, snowflake as _snow, bigquery as _bq, - databricks as _dbx, mssql as _mssql, redshift as _rs, pyiceberg_rest as _ice, + databricks as _dbx, mssql as _mssql, redshift as _rs, motherduck as _md) _AUTH_ADAPTERS: dict[tuple[t.Any, type], t.Callable[[t.Any, dict], dict]] = { @@ -29,7 +29,6 @@ (P.MSSQL, AzureADAuthProfile): _mssql.azure_ad, (P.REDSHIFT, PasswordAuthProfile): _rs.password, (P.REDSHIFT, IAMAuthProfile): _rs.iam, - (P.PYICEBERG_REST, TokenAuthProfile): _ice.token, } for _p in (P.POSTGRESQL, P.MYSQL, P.CLICKHOUSE, P.MATERIALIZE, P.RISINGWAVE, P.DRUID, P.SINGLESTOREDB, P.IMPALA, P.EXASOL): diff --git a/src/mountainash_data/core/settings/pyiceberg_rest.py b/src/mountainash_data/core/settings/pyiceberg_rest.py deleted file mode 100644 index f2d846f..0000000 --- a/src/mountainash_data/core/settings/pyiceberg_rest.py +++ /dev/null @@ -1,70 +0,0 @@ -"""PyIceberg REST catalog backend settings. - -Spec: ``mountainash-central/04.planning/mountainash-data/superpowers/specs/2026-04-15-settings-audit/pyiceberg_rest.md``. -Driver: https://py.iceberg.apache.org/configuration/ -""" - -from __future__ import annotations - -import typing as t - -from pydantic import SecretStr - -from ..constants import CONST_DB_PROVIDER_TYPE -from mountainash_auth_client import TokenAuthProfile -from .adapters import pyiceberg_rest as _ice -from .descriptor import BackendSpec, ParameterSpec -from .profile import BackendProfile -from .registry import register - - -PYICEBERG_REST_SPEC = BackendSpec( - name="pyiceberg_rest", - provider_type=CONST_DB_PROVIDER_TYPE.PYICEBERG_REST, - connection_string_scheme=None, # uri= kwarg, not URL form - supported_auth=(TokenAuthProfile,), - parameters=[ - ParameterSpec(name="CATALOG_NAME", type=str, tier="core", - driver_key="name"), - ParameterSpec(name="CATALOG_URI", type=str, tier="core", - driver_key="uri"), - ParameterSpec(name="WAREHOUSE", type=t.Optional[str], tier="core", - default=None, driver_key="warehouse"), - ParameterSpec(name="VERIFY_SSL", type=bool, tier="advanced", - default=True, driver_key="verify-ssl"), - # S3 family (driver_key emits dotted keys directly) - ParameterSpec(name="S3_REGION", type=t.Optional[str], tier="advanced", - default=None, driver_key="s3.region"), - ParameterSpec(name="S3_ENDPOINT", type=t.Optional[str], - tier="advanced", default=None, driver_key="s3.endpoint"), - ParameterSpec(name="S3_ACCESS_KEY_ID", type=t.Optional[str], - tier="advanced", default=None, - driver_key="s3.access-key-id"), - ParameterSpec(name="S3_SECRET_ACCESS_KEY", - type=t.Optional[SecretStr], tier="advanced", - default=None, secret=True, - driver_key="s3.secret-access-key"), - ParameterSpec(name="S3_SESSION_TOKEN", type=t.Optional[SecretStr], - tier="advanced", default=None, secret=True, - driver_key="s3.session-token"), - # SigV4 - ParameterSpec(name="REST_SIGV4_ENABLED", type=t.Optional[bool], - tier="advanced", default=None, - driver_key="rest.sigv4-enabled"), - ParameterSpec(name="REST_SIGNING_REGION", type=t.Optional[str], - tier="advanced", default=None, - driver_key="rest.signing-region"), - ParameterSpec(name="REST_SIGNING_NAME", type=t.Optional[str], - tier="advanced", default=None, - driver_key="rest.signing-name"), - # HEADERS: no driver_key — adapter expands to header. = v - ParameterSpec(name="HEADERS", type=t.Optional[dict[str, str]], - tier="advanced", default=None), - ], -) - - -@register -class PyIcebergRestBackendProfile(BackendProfile): - __spec__ = PYICEBERG_REST_SPEC - __adapters__ = {CONST_DB_PROVIDER_TYPE.PYICEBERG_REST: _ice.headers_compose} diff --git a/tests/test_unit/backends/ibis/test_namespace_hierarchy.py b/tests/test_unit/backends/ibis/test_namespace_hierarchy.py index ce48445..bb52cc3 100644 --- a/tests/test_unit/backends/ibis/test_namespace_hierarchy.py +++ b/tests/test_unit/backends/ibis/test_namespace_hierarchy.py @@ -8,8 +8,6 @@ not at module import time — so no pytest.importorskip guard is needed here; _RecordingConn stands in for the ibis connection object entirely. Regression: the database= keyword is gone from the public surface. - -Iceberg deep-namespace round-trip is deferred to DEBT-11 (see spec §10 note). """ from __future__ import annotations diff --git a/tests/test_unit/backends/iceberg/COVERAGE_GAP.md b/tests/test_unit/backends/iceberg/COVERAGE_GAP.md deleted file mode 100644 index 6aab2e5..0000000 --- a/tests/test_unit/backends/iceberg/COVERAGE_GAP.md +++ /dev/null @@ -1,17 +0,0 @@ -# Iceberg test coverage gap - -Before Phase 3 (writing-plans audit), no tests existed for any of: - -- `databases/connections/pyiceberg/base_pyiceberg_connection.py` (884 LOC) -- `databases/connections/pyiceberg/pyiceberg_rest_connection.py` (205 LOC) -- `databases/operations/pyiceberg/base_pyiceberg_operations.py` (868 LOC) -- `databases/operations/pyiceberg/pyiceberg_rest_operations.py` (207 LOC) - -The Phase 3 refactor proceeds *without* a regression net for iceberg. -This is acceptable because the user (sole consumer) confirmed iceberg -is in prototype use only. Tests added during this phase target the new -shape (IcebergBackend protocol, inspection model conversion) rather -than reproducing legacy behavior. - -If iceberg moves to production use later, a separate hardening pass -should add tests for the salvaged operations. diff --git a/tests/test_unit/backends/iceberg/__init__.py b/tests/test_unit/backends/iceberg/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/test_unit/backends/iceberg/test_backend.py b/tests/test_unit/backends/iceberg/test_backend.py deleted file mode 100644 index 01822c3..0000000 --- a/tests/test_unit/backends/iceberg/test_backend.py +++ /dev/null @@ -1,83 +0,0 @@ -"""Tests for IcebergBackend factory. - -NOTE: These tests do NOT call ``.connect()`` — that would require a live -iceberg catalog and valid settings parameters. They verify construction, -protocol shape, and config storage only. End-to-end connection tests are -out of scope for this refactor (see COVERAGE_GAP.md). - -pyiceberg is an optional dependency that is not installed in the default -test environment. All tests in this module are skipped when it is absent. -""" - -import warnings -import pytest -from unittest.mock import MagicMock - -pyiceberg = pytest.importorskip("pyiceberg", reason="pyiceberg not installed") - -from mountainash_data.backends.iceberg.backend import IcebergBackend # noqa: E402 -from mountainash_data.core.protocol import Backend # noqa: E402 -from mountainash_data.core.errors import TransactionUnsupportedError # noqa: E402 - - -def test_iceberg_backend_satisfies_protocol(): - backend = IcebergBackend(catalog="rest", uri="http://localhost:8181") - assert isinstance(backend, Backend) - assert backend.name == "iceberg" - - -def test_unknown_catalog_raises(): - with pytest.raises(KeyError, match="Unknown iceberg catalog"): - IcebergBackend(catalog="bogus") - - -def test_iceberg_backend_carries_config(): - backend = IcebergBackend(catalog="rest", uri="http://localhost:8181", token="abc") - assert backend._config == {"uri": "http://localhost:8181", "token": "abc"} - - -def test_raw_driver_connection_returns_catalog(monkeypatch): - be = IcebergBackend(catalog="rest", uri="http://localhost:8181") - fake_conn = MagicMock() - fake_catalog = object() - fake_conn.catalog_backend = fake_catalog - be._conn = fake_conn # simulate connected - assert be.raw_driver_connection() is fake_catalog - - -def test_raw_driver_connection_requires_connected(): - be = IcebergBackend(catalog="rest", uri="http://localhost:8181") - with pytest.raises(RuntimeError, match="not connected"): - be.raw_driver_connection() - - -def test_transaction_required_raises(): - be = IcebergBackend(catalog="rest", uri="http://localhost:8181") - with pytest.raises(TransactionUnsupportedError): - with be.transaction(): - pass - - -def test_transaction_not_required_noops(): - be = IcebergBackend(catalog="rest", uri="http://localhost:8181") - from mountainash_data.core import _warn as _warnmod - _warnmod._WARNED.discard("iceberg") - with warnings.catch_warnings(record=True) as w: - warnings.simplefilter("always") - with be.transaction(required=False): - pass - assert any("iceberg" in str(x.message).lower() for x in w) - - -def test_in_transaction_always_false(): - be = IcebergBackend(catalog="rest", uri="http://localhost:8181") - # No connection-level unit of work — consistent with supports_transactions. - assert be.in_transaction() is False - assert be.supports_transactions is False - - -def test_iceberg_satisfies_protocol_including_in_transaction(): - from mountainash_data.core.protocol import Backend - be = IcebergBackend(catalog="rest", uri="http://localhost:8181") - assert isinstance(be, Backend) - assert callable(be.in_transaction) diff --git a/tests/test_unit/backends/iceberg/test_iceberg_auth.py b/tests/test_unit/backends/iceberg/test_iceberg_auth.py deleted file mode 100644 index 421add4..0000000 --- a/tests/test_unit/backends/iceberg/test_iceberg_auth.py +++ /dev/null @@ -1,40 +0,0 @@ -import pytest - -pytest.importorskip("pyiceberg", reason="pyiceberg not installed") -pytest.importorskip("mountainash_dataframes", reason="mountainash_dataframes not installed") - -from types import SimpleNamespace # noqa: E402 -from unittest.mock import patch # noqa: E402 - -from mountainash_auth_client import TokenAuthProfile # noqa: E402 -from mountainash_data.backends.iceberg.connection import IcebergConnectionBase # noqa: E402 - - -class _ConcreteIceberg(IcebergConnectionBase): - @property - def catalog_backend(self): - return getattr(self, "_catalog_backend", None) - - -# test double: bypass the remaining ABC methods we don't exercise -_ConcreteIceberg.__abstractmethods__ = frozenset() - - -def test_build_catalog_kwargs_threads_auth_and_merges(): - obj_settings = object() - params = SimpleNamespace( - settings_class=SimpleNamespace(get_settings=lambda settings_parameters: obj_settings) - ) - conn = _ConcreteIceberg.__new__(_ConcreteIceberg) - conn.db_auth_settings_parameters = params - - auth = TokenAuthProfile(TOKEN="T") - with patch( - "mountainash_data.backends.iceberg.connection.build_driver_kwargs", - return_value={"uri": "http://x", "token": "T", "name": "c"}, - ) as bk: - out = conn._build_catalog_kwargs(auth, warehouse="w") - - bk.assert_called_once_with(obj_settings, auth) # profile + auth_profile threaded - assert out["warehouse"] == "w" # explicit kwargs win - assert out["uri"] == "http://x" diff --git a/tests/test_unit/backends/iceberg/test_namespace.py b/tests/test_unit/backends/iceberg/test_namespace.py deleted file mode 100644 index 7b24692..0000000 --- a/tests/test_unit/backends/iceberg/test_namespace.py +++ /dev/null @@ -1,23 +0,0 @@ -"""Iceberg namespace interface alignment (DEBT-10; deep-path fidelity = DEBT-11). - -NOTE: There is no live/in-memory iceberg catalog fixture in this repo (see -tests/fixtures/database_fixtures.py and test_backend.py), so this module is -limited to a pure structural check that needs no live catalog. Behavioral / -round-trip coverage against a real catalog is out of scope here and tracked -under DEBT-11. -""" - -import pytest - -pytest.importorskip("pyiceberg") - -from mountainash_data.core.namespace import Namespace, NamespaceLike # noqa: F401 - - -def test_iceberg_connection_satisfies_widened_protocol(): - """The iceberg connection base exposes the widened discovery surface.""" - from mountainash_data.backends.iceberg.connection import IcebergConnectionBase - - for meth in ("list_tables", "list_namespaces", "list_catalogs", - "inspect_table", "inspect_namespace", "inspect_catalog"): - assert hasattr(IcebergConnectionBase, meth) diff --git a/tests/test_unit/core/settings/adapters/test_auth_adapters.py b/tests/test_unit/core/settings/adapters/test_auth_adapters.py index 2df168c..76d895f 100644 --- a/tests/test_unit/core/settings/adapters/test_auth_adapters.py +++ b/tests/test_unit/core/settings/adapters/test_auth_adapters.py @@ -6,7 +6,7 @@ ) from mountainash_data.core.settings.adapters import ( sql as _sql, snowflake as _snow, mssql as _mssql, - redshift as _rs, databricks as _dbx, pyiceberg_rest as _ice, + redshift as _rs, databricks as _dbx, ) @@ -62,10 +62,6 @@ def test_databricks_token(): assert _dbx.token(TokenAuthProfile(TOKEN="tok"), {}) == {"access_token": "tok"} -def test_pyiceberg_token(): - assert _ice.token(TokenAuthProfile(TOKEN="tok"), {"uri": "u"}) == {"uri": "u", "token": "tok"} - - def test_trino_password_builds_basic_auth(): pytest.importorskip("trino") from trino.auth import BasicAuthentication diff --git a/tests/test_unit/core/settings/backends/test_pyiceberg_rest.py b/tests/test_unit/core/settings/backends/test_pyiceberg_rest.py deleted file mode 100644 index a40b4e4..0000000 --- a/tests/test_unit/core/settings/backends/test_pyiceberg_rest.py +++ /dev/null @@ -1,37 +0,0 @@ -# tests/test_unit/core/settings/backends/test_pyiceberg_rest.py -from __future__ import annotations - -import pytest - -from mountainash_data.core.constants import CONST_DB_PROVIDER_TYPE -from mountainash_data.core.settings.pyiceberg_rest import PyIcebergRestBackendProfile - - -@pytest.mark.unit -class TestPyIcebergRestBackendProfile: - def _min(self, **extra): - return PyIcebergRestBackendProfile( - CATALOG_NAME="cat", - CATALOG_URI="https://catalog.example/v1", - **extra, - ) - - def test_warehouse_optional(self): - """Audit regression: WAREHOUSE was over-required.""" - s = self._min() - assert s.WAREHOUSE is None - - def test_emit_plumbs_uri(self): - s = self._min() - kwargs = s.emit(CONST_DB_PROVIDER_TYPE.PYICEBERG_REST) - assert kwargs["uri"] == "https://catalog.example/v1" - assert kwargs["name"] == "cat" - - def test_s3_params_stored(self): - """Audit regression: s3.* family was absent from the spec.""" - s = self._min( - S3_ENDPOINT="https://r2.example.com", - S3_REGION="auto", - ) - assert s.S3_ENDPOINT == "https://r2.example.com" - assert s.S3_REGION == "auto" diff --git a/tests/test_unit/core/settings/test_config_shaping.py b/tests/test_unit/core/settings/test_config_shaping.py index 28a4a64..0894f53 100644 --- a/tests/test_unit/core/settings/test_config_shaping.py +++ b/tests/test_unit/core/settings/test_config_shaping.py @@ -1,7 +1,6 @@ from mountainash_data.core.constants import CONST_DB_PROVIDER_TYPE as P from mountainash_data.core.settings import ( MySQLBackendProfile, MSSQLBackendProfile, SnowflakeBackendProfile, - PyIcebergRestBackendProfile, ) @@ -25,11 +24,3 @@ def test_snowflake_session_parameters_added_only(): assert out["session_parameters"] == {"QUERY_TAG": "etl", "TIMEZONE": "UTC"} assert "query_tag" not in out and "timezone" not in out - -def test_pyiceberg_headers_expand_s3_flat(): - out = PyIcebergRestBackendProfile( - CATALOG_NAME="c", CATALOG_URI="http://x", S3_REGION="us-east-1", - HEADERS={"X-A": "1", "X-B": "2"}, - ).emit(P.PYICEBERG_REST) - assert out["name"] == "c" and out["uri"] == "http://x" and out["s3.region"] == "us-east-1" - assert out["header.X-A"] == "1" and out["header.X-B"] == "2" and "headers" not in out diff --git a/tests/test_unit/core/settings/test_registry.py b/tests/test_unit/core/settings/test_registry.py index a55ee52..410eeed 100644 --- a/tests/test_unit/core/settings/test_registry.py +++ b/tests/test_unit/core/settings/test_registry.py @@ -13,12 +13,12 @@ @pytest.mark.unit class TestDatabasesRegistry: def test_registry_is_populated_after_import(self): - """All 12 backends register themselves at import time.""" + """Every relational backend registers itself at import time.""" import mountainash_data.core.settings # noqa: F401 for name in ["sqlite", "duckdb", "postgresql", "mysql", "mssql", "snowflake", "bigquery", "redshift", "pyspark", - "trino", "motherduck", "pyiceberg_rest"]: + "trino", "motherduck"]: assert name in DATABASES_REGISTRY, f"{name} missing from registry" def test_get_descriptor_returns_correct_type(self): diff --git a/tests/test_unit/core/test_protocol.py b/tests/test_unit/core/test_protocol.py index 7c5553b..3bd3122 100644 --- a/tests/test_unit/core/test_protocol.py +++ b/tests/test_unit/core/test_protocol.py @@ -68,13 +68,6 @@ def test_protocol_declares_raw_driver_connection(): assert hasattr(Backend, "raw_driver_connection") -def test_iceberg_backend_satisfies_widened_protocol(): - import pytest - pytest.importorskip("pyiceberg") - from mountainash_data.backends.iceberg.backend import IcebergBackend - assert hasattr(IcebergBackend, "raw_driver_connection") - - def test_protocol_declares_transaction(): from mountainash_data.core.protocol import Backend assert hasattr(Backend, "transaction") @@ -148,7 +141,7 @@ def in_transaction(self): def test_non_ibis_stub_satisfies_backend_protocol(): stub = _StubBackend() assert isinstance(stub, Backend) - # catalog tier — the seam Iceberg originally motivated, kept generic + # catalog tier — kept generic for any multi-catalog backend assert stub.list_catalogs() == [] assert stub.inspect_catalog().name == "stub" assert stub.list_namespaces(catalog="anything") == [] From d6e354fb2fe32d4cb7f3edd5acc4b15a18b2ac69 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sat, 18 Jul 2026 11:44:35 +1000 Subject: [PATCH 3/4] docs: drop Iceberg backend from README, overview, and CLAUDE.md Package is now Ibis-relational behind a backend-agnostic Backend protocol. Co-Authored-By: Claude Opus 4.8 (1M context) --- CLAUDE.md | 37 ++++++++-------------------------- README.md | 22 +++++++------------- docs/PROJECT_OVERVIEW.md | 43 +++++++++++++++++----------------------- 3 files changed, 33 insertions(+), 69 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 1f93e6f..7aed899 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,7 +4,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## Project Overview -**mountainash-data** provides physical access to backend data services — relational databases via Ibis, and Iceberg table-format catalogs via PyIceberg. It collapses what was previously 13 per-dialect connection classes into a data-driven `DialectSpec` registry, exposes clean `Backend` / `Connection` protocols, and provides factories and a high-level facade (`DatabaseUtils`). +**mountainash-data** provides physical access to backend data services — relational databases via Ibis behind a backend-agnostic `Backend` protocol ready for additional backends. It collapses what was previously 13 per-dialect connection classes into a data-driven `DialectSpec` registry, exposes clean `Backend` / `Connection` protocols, and provides factories and a high-level facade (`DatabaseUtils`). ## Planning, Specs & Principles (live in mountainash-central) @@ -63,11 +63,6 @@ tests, and this `CLAUDE.md` live in this repo's tree. - `BaseIbisOperations` + per-dialect subclasses in `operations.py` - `DialectSpec` registry in `dialects/` -7. **Iceberg backend** (`src/mountainash_data/backends/iceberg/`) - - `IcebergBackend` — catalog-type registry (currently: `"rest"`) - - Connection, operations, and inspection classes - - Requires optional `pyiceberg` dependency - ### Package Structure ``` @@ -83,22 +78,15 @@ src/mountainash_data/ │ ├── settings/ # Per-dialect auth settings (pydantic) │ └── factories/ # ConnectionFactory, OperationsFactory, SettingsFactory └── backends/ - ├── ibis/ - │ ├── backend.py # IbisBackend + IbisConnection (new-style) - │ ├── connection.py # BaseIbisConnection + dialect subclasses - │ ├── operations.py # BaseIbisOperations + dialect subclasses - │ ├── inspect.py # Ibis-specific inspection helpers - │ └── dialects/ # DialectSpec registry (data-driven) - └── iceberg/ - ├── backend.py # IcebergBackend + catalog registry - ├── connection.py # IcebergConnectionBase - ├── operations.py # IcebergOperationsBase - ├── inspect.py # Iceberg inspection helpers - └── catalogs/ # Per-catalog implementations + └── ibis/ + ├── backend.py # IbisBackend + IbisConnection (new-style) + ├── connection.py # BaseIbisConnection + dialect subclasses + ├── operations.py # BaseIbisOperations + dialect subclasses + ├── inspect.py # Ibis-specific inspection helpers + └── dialects/ # DialectSpec registry (data-driven) ``` ### Optional Dependencies (extras) -- **pyiceberg**: Required for `IcebergBackend` - **postgres**: PostgreSQL support (psycopg2-binary, ibis-framework[postgres]) - **mssql**: SQL Server support (pyodbc, ibis-framework[mssql]) - **snowflake**: Snowflake support (snowflake-connector-python, ibis-framework[snowflake]) @@ -196,7 +184,6 @@ tests/ ├── test_unit/ │ ├── core/ # Protocol, inspection tests │ ├── backends/ibis/ # IbisBackend tests -│ ├── backends/iceberg/ # IcebergBackend tests │ ├── factories/ # Factory tests │ ├── databases/ # Legacy-path tests (updated to new paths) │ ├── test_database_utils.py # DatabaseUtils tests @@ -219,7 +206,7 @@ tests/ ## Usage Patterns ```python -from mountainash_data import IbisBackend, IcebergBackend +from mountainash_data import IbisBackend from mountainash_data.core.settings import ( SQLiteAuthSettings, NoAuth, @@ -250,14 +237,6 @@ try: # rel = ma.relation(ibis_table) # compiles against Ibis automatically finally: conn.close() - -# Iceberg backend (requires pyiceberg) -ice = IcebergBackend(catalog="rest", uri="http://localhost:8181") -ice_conn = ice.connect() -try: - namespaces = ice_conn.list_namespaces() -finally: - ice_conn.close() ``` ## Development Environments diff --git a/README.md b/README.md index fb608bb..f15c0b5 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ Mountain Ash - Data Physical access to backend data services — relational databases via Ibis, -and Iceberg table-format catalogs via PyIceberg. +behind a backend-agnostic `Backend` protocol ready for additional backends. @@ -83,18 +83,12 @@ src/mountainash_data/ │ ├── settings/ # Per-dialect auth settings (pydantic) │ └── factories/ # ConnectionFactory, OperationsFactory, SettingsFactory └── backends/ - ├── ibis/ # IbisBackend — 12-dialect registry - │ ├── backend.py # IbisBackend + IbisConnection - │ ├── connection.py # BaseIbisConnection + per-dialect subclasses - │ ├── operations.py # BaseIbisOperations + per-dialect subclasses - │ ├── inspect.py # Ibis-specific inspection helpers - │ └── dialects/ # DialectSpec registry (data-driven) - └── iceberg/ # IcebergBackend — PyIceberg catalogs - ├── backend.py # IcebergBackend + catalog registry - ├── connection.py # IcebergConnectionBase - ├── operations.py # IcebergOperationsBase - ├── inspect.py # Iceberg inspection helpers - └── catalogs/ # Per-catalog implementations (rest, …) + └── ibis/ # IbisBackend — 12-dialect registry + ├── backend.py # IbisBackend + IbisConnection + ├── connection.py # BaseIbisConnection + per-dialect subclasses + ├── operations.py # BaseIbisOperations + per-dialect subclasses + ├── inspect.py # Ibis-specific inspection helpers + └── dialects/ # DialectSpec registry (data-driven) ``` ### Public API @@ -104,7 +98,6 @@ from mountainash_data import ( Backend, # Protocol: what every backend must implement Connection, # Protocol: what every connection must implement IbisBackend, # Ibis-style relational backends (sqlite, duckdb, postgres, …) - IcebergBackend, # Iceberg-style table-format catalogs (pyiceberg required) CatalogInfo, # Physical catalog metadata NamespaceInfo, # Physical namespace/schema metadata TableInfo, # Physical table metadata @@ -118,7 +111,6 @@ from mountainash_data import ( ### Optional Dependencies -- **pyiceberg**: Required for `IcebergBackend`. Not installed by default. - **postgres**: `psycopg2-binary` + `ibis-framework[postgres]` - **mssql**: `pyodbc` + `ibis-framework[mssql]` - **snowflake**: `snowflake-connector-python` + `ibis-framework[snowflake]` diff --git a/docs/PROJECT_OVERVIEW.md b/docs/PROJECT_OVERVIEW.md index 68fd761..7532be3 100644 --- a/docs/PROJECT_OVERVIEW.md +++ b/docs/PROJECT_OVERVIEW.md @@ -5,7 +5,7 @@ Provides unified database connections and dataframe abstractions for multiple ba ## Architecture The package is built on a layered architecture with three main components: -1. **Database Connections Layer** - Abstracts database connections using Ibis framework and PyIceberg +1. **Database Connections Layer** - Abstracts database connections using the Ibis framework 2. **DataFrame Abstraction Layer** - Provides unified dataframe interface across multiple backends 3. **Utilities Layer** - Supporting utilities for data transformation, mapping, and conversion @@ -17,31 +17,25 @@ src/mountainash_data/ ├── databases/ # Database connection layer │ ├── __init__.py │ ├── base_db_connection.py # Abstract base connection -│ ├── ibis/ # Ibis-based connections -│ │ ├── __init__.py -│ │ ├── base_ibis_connection.py -│ │ ├── constants.py -│ │ ├── ibis_connection_factory.py -│ │ └── connections/ # Specific backend implementations -│ │ ├── __init__.py -│ │ ├── bigquery_ibis_connection.py -│ │ ├── duckdb_ibis_connection.py -│ │ ├── motherduck_ibis_connection.py -│ │ ├── mssql_ibis_connection.py -│ │ ├── mysql_ibis_connection.py -│ │ ├── oracle_ibis_connection.py -│ │ ├── postgres_ibis_connection.py -│ │ ├── pyspark_ibis_connection.py -│ │ ├── redshift_ibis_connection.py -│ │ ├── snowflake_ibis_connection.py -│ │ ├── sqlite_ibis_connection.py -│ │ └── trino_ibis_connection.py -│ └── pyiceberg/ # PyIceberg support +│ └── ibis/ # Ibis-based connections │ ├── __init__.py -│ ├── base_pyiceberg_connection.py -│ └── connections/ +│ ├── base_ibis_connection.py +│ ├── constants.py +│ ├── ibis_connection_factory.py +│ └── connections/ # Specific backend implementations │ ├── __init__.py -│ └── pyiceberg_rest_connection.py +│ ├── bigquery_ibis_connection.py +│ ├── duckdb_ibis_connection.py +│ ├── motherduck_ibis_connection.py +│ ├── mssql_ibis_connection.py +│ ├── mysql_ibis_connection.py +│ ├── oracle_ibis_connection.py +│ ├── postgres_ibis_connection.py +│ ├── pyspark_ibis_connection.py +│ ├── redshift_ibis_connection.py +│ ├── snowflake_ibis_connection.py +│ ├── sqlite_ibis_connection.py +│ └── trino_ibis_connection.py ├── dataframes/ # DataFrame abstraction layer │ ├── __init__.py │ ├── base_dataframe.py # Abstract dataframe interface @@ -101,7 +95,6 @@ Core package providing unified data access layer with key classes: Database connection abstraction supporting multiple backends: - **Base Layer**: `BaseDBConnection` abstract interface - **Ibis Layer**: Connection implementations for 12+ database backends -- **PyIceberg Layer**: Data lake connectivity via Apache Iceberg ### dataframes/ DataFrame abstraction and utilities: From 6441ca558a0abaef589d4cfe825010a94747bb11 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sat, 18 Jul 2026 11:49:03 +1000 Subject: [PATCH 4/4] docs: scrub Iceberg from quickstart, dbt-integration, package-overview MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the README/CLAUDE.md scrub — remove the broken IcebergBackend quickstart example and PyIceberg capability claims now that the backend is retired. Historical hiivmind review artifacts intentionally left untouched. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/assets/package_overview.md | 2 -- docs/dbt_integration.md | 4 +--- docs/quickstart.md | 15 --------------- 3 files changed, 1 insertion(+), 20 deletions(-) diff --git a/docs/assets/package_overview.md b/docs/assets/package_overview.md index b9a44dc..c03b6cc 100644 --- a/docs/assets/package_overview.md +++ b/docs/assets/package_overview.md @@ -9,7 +9,6 @@ 1. **Database Connections Layer** (`src/mountainash_data/databases/`) - Base database connection abstraction (`BaseDBConnection`) - Ibis-based connections supporting multiple backends (SQLite, DuckDB, PostgreSQL, SQL Server, etc.) - - PyIceberg support for data lake operations - Connection factory pattern for backend instantiation 2. **DataFrame Abstraction Layer** (`src/mountainash_data/dataframes/`) @@ -43,7 +42,6 @@ src/mountainash_data/ │ │ │ ├── postgres_ibis_connection.py │ │ │ └── [other backends...] │ │ └── ibis_connection_factory.py -│ └── pyiceberg/ # PyIceberg support ├── dataframes/ # DataFrame abstraction layer │ ├── base_dataframe.py # Abstract dataframe interface │ ├── ibis_dataframe.py # Ibis dataframe implementation diff --git a/docs/dbt_integration.md b/docs/dbt_integration.md index f7b2550..4726498 100644 --- a/docs/dbt_integration.md +++ b/docs/dbt_integration.md @@ -526,7 +526,7 @@ | Snapshots | mountainash-data | mountainash-dataframes | Operations + Join for SCD | | Seeds (Enhanced) | mountainash-dataframes | - | Convert from structured Python data | | Materializations | mountainash-data | - | Connection, Operations | - | External Tables | mountainash-data | mountainash-dataframes | PyIceberg, External file ops | + | External Tables | mountainash-data | mountainash-dataframes | External file ops | --- 🎯 Integration Architecture Pattern @@ -598,7 +598,6 @@ Phase 4: Advanced Features (Innovation) 9. ✅ Custom materializations using mountainash-data connections - 10. ✅ PyIceberg integration for data lakes --- This alignment creates a powerful synergy where: @@ -1640,7 +1639,6 @@ - ✅ PostgreSQL, MySQL, Oracle, MS SQL Server - ✅ Snowflake, BigQuery, Redshift - ✅ PySpark, Trino - - ✅ PyIceberg (data lakes) dbt Integration Points: - ✅ Incremental model upserts diff --git a/docs/quickstart.md b/docs/quickstart.md index 6c0bddf..4e7f0b0 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -15,7 +15,6 @@ pip install mountainash-data[snowflake] # Snowflake pip install mountainash-data[bigquery] # BigQuery pip install mountainash-data[pyspark] # Apache Spark pip install mountainash-data[trino] # Trino -pip install mountainash-data[pyiceberg] # Iceberg catalogs ``` SQLite and DuckDB work out of the box — no extra needed. @@ -214,17 +213,3 @@ with IbisBackend(dialect="duckdb") as backend: | `druid` | `druid://` | | | `pyspark` | `pyspark://` | Requires `[pyspark]` extra | ---- - -## Iceberg catalogs (optional) - -Requires `pip install mountainash-data[pyiceberg]`. - -```python -from mountainash_data import IcebergBackend - -with IcebergBackend(catalog="rest", uri="http://localhost:8181") as backend: - namespaces = backend.list_namespaces() - tables = backend.list_tables(namespace="analytics") - info = backend.inspect_table("events", namespace="analytics") -```