diff --git a/README.md b/README.md index 1c4af53..477f615 100644 --- a/README.md +++ b/README.md @@ -125,9 +125,15 @@ iceberg_rest( table_location_layout=None, # Custom table location pattern register_new_tables=False, # Register tables found in storage hard_delete_column="_dlt_deleted_at", # Column for hard deletes + internal_table_prefix="_dlt", # Physical prefix for dlt metadata tables ) ``` +`internal_table_prefix` is opt-in and does not change dlt's logical schema. +Leave the default to create the standard `_dlt_loads`, `_dlt_version`, and +`_dlt_pipeline_state` tables. Set it to `"dlt"` for catalogs such as AWS S3 +Tables that reject identifiers beginning with an underscore. + ## Catalog Examples diff --git a/src/dlt_iceberg/destination_client.py b/src/dlt_iceberg/destination_client.py index db529e4..8e112ac 100644 --- a/src/dlt_iceberg/destination_client.py +++ b/src/dlt_iceberg/destination_client.py @@ -166,6 +166,12 @@ class IcebergRestConfiguration(DestinationClientConfiguration): # Set to None to disable hard delete hard_delete_column: Optional[str] = "_dlt_deleted_at" + # Physical prefix for dlt's internal metadata tables. The default preserves + # dlt's canonical names (_dlt_loads, _dlt_version, _dlt_pipeline_state). + # Catalogs that reject leading underscores, such as AWS S3 Tables, can set + # this to "dlt" without changing the logical names used by dlt. + internal_table_prefix: str = "_dlt" + class IcebergRestLoadJob(RunnableLoadJob): @@ -238,6 +244,27 @@ def __init__( # SQL client instance (created lazily) self._sql_client = None + def _physical_table_name(self, table_name: str) -> str: + """Map canonical dlt metadata names to their configured physical names.""" + logical_prefix = "_dlt_" + if not table_name.startswith(logical_prefix): + return table_name + + prefix = self.config.internal_table_prefix.rstrip("_") + if not prefix: + raise ValueError("internal_table_prefix must contain at least one character") + return f"{prefix}_{table_name.removeprefix(logical_prefix)}" + + def _table_identifier(self, table_name: str) -> str: + return f"{self.config.namespace}.{self._physical_table_name(table_name)}" + + def _is_internal_physical_table(self, table_name: str) -> bool: + return table_name in { + self._physical_table_name(self.schema.loads_table_name), + self._physical_table_name(self.schema.version_table_name), + self._physical_table_name(self.schema.state_table_name), + } + # ---- WithSqlClient interface ---- @property @@ -273,7 +300,7 @@ def get_open_table_location(self, table_format: TTableFormat, table_name: str) - # Try to get location from catalog try: catalog = self._get_catalog() - identifier = f"{self.config.namespace}.{table_name}" + identifier = self._table_identifier(table_name) iceberg_table = catalog.load_table(identifier) return iceberg_table.location() except NoSuchTableError: @@ -285,7 +312,10 @@ def get_open_table_location(self, table_format: TTableFormat, table_name: str) - warehouse = self.config.warehouse or "" if warehouse and not warehouse.endswith("/"): warehouse += "/" - return f"{warehouse}{self.config.namespace}/{table_name}" + return ( + f"{warehouse}{self.config.namespace}/" + f"{self._physical_table_name(table_name)}" + ) def load_open_table(self, table_format: TTableFormat, table_name: str, **kwargs: Any) -> Any: """Load and return a PyIceberg Table object.""" @@ -295,7 +325,7 @@ def load_open_table(self, table_format: TTableFormat, table_name: str, **kwargs: from dlt.common.destination.exceptions import DestinationUndefinedEntity catalog = self._get_catalog() - identifier = f"{self.config.namespace}.{table_name}" + identifier = self._table_identifier(table_name) try: return catalog.load_table(identifier) @@ -313,7 +343,7 @@ def _get_newest_schema(self, schema_name: str) -> Optional[StorageSchemaInfo]: """Get newest schema version by schema name using predicate pushdown.""" try: catalog = self._get_catalog() - identifier = f"{self.config.namespace}.{self.schema.version_table_name}" + identifier = self._table_identifier(self.schema.version_table_name) iceberg_table = catalog.load_table(identifier) # Use row_filter for predicate pushdown - only scan matching rows @@ -346,7 +376,7 @@ def _get_schema_by_hash(self, version_hash: str) -> Optional[StorageSchemaInfo]: """Get schema by exact version hash using predicate pushdown.""" try: catalog = self._get_catalog() - identifier = f"{self.config.namespace}.{self.schema.version_table_name}" + identifier = self._table_identifier(self.schema.version_table_name) iceberg_table = catalog.load_table(identifier) # Use row_filter for predicate pushdown @@ -394,7 +424,7 @@ def get_stored_state(self, pipeline_name: str) -> Optional[StateInfo]: """Loads pipeline state from the _dlt_pipeline_state table using predicate pushdown.""" try: catalog = self._get_catalog() - identifier = f"{self.config.namespace}.{self.schema.state_table_name}" + identifier = self._table_identifier(self.schema.state_table_name) iceberg_table = catalog.load_table(identifier) # Use row_filter for predicate pushdown @@ -460,7 +490,7 @@ def _derive_schema_from_iceberg_tables(self, schema_name: str) -> Optional[Stora derived_tables = {} for table_id in tables: table_name = table_id[1] - if table_name.startswith('_dlt_'): + if self._is_internal_physical_table(table_name): continue # Skip dlt metadata tables try: @@ -612,7 +642,7 @@ def _write_schema_to_storage(self) -> None: catalog = self._get_catalog() version_table_name = self.schema.version_table_name - identifier = f"{self.config.namespace}.{version_table_name}" + identifier = self._table_identifier(version_table_name) # Schema data to write # Use naive datetime (no timezone) to match Iceberg TimestampType @@ -753,7 +783,7 @@ def _get_table_location(self, table_name: str) -> Optional[str]: location = self.config.table_location_layout.format( namespace=self.config.namespace, dataset_name=self.config.namespace, # In dlt, dataset_name maps to namespace - table_name=table_name, + table_name=self._physical_table_name(table_name), ) # If layout is relative (doesn't start with protocol), prepend warehouse @@ -837,7 +867,7 @@ def _register_tables_from_storage(self, catalog, namespace: str) -> None: latest_metadata = os.path.join(metadata_path, metadata_files[0]) try: - identifier = f"{namespace}.{table_name}" + identifier = self._table_identifier(table_name) catalog.register_table( identifier=identifier, metadata_location=f"file://{latest_metadata}", @@ -872,7 +902,7 @@ def initialize_storage(self, truncate_tables: Optional[Iterable[str]] = None) -> # Handle truncation if requested if truncate_tables: for table_name in truncate_tables: - identifier = f"{namespace}.{table_name}" + identifier = self._table_identifier(table_name) try: catalog.drop_table(identifier) logger.info(f"Truncated table {identifier}") @@ -910,7 +940,7 @@ def drop_tables(self, *table_names: str, delete_schema: bool = True) -> None: """ catalog = self._get_catalog() for name in table_names: - identifier = f"{self.config.namespace}.{name}" + identifier = self._table_identifier(name) try: if hasattr(catalog, "purge_table"): catalog.purge_table(identifier) @@ -921,9 +951,7 @@ def drop_tables(self, *table_names: str, delete_schema: bool = True) -> None: pass if delete_schema: - version_identifier = ( - f"{self.config.namespace}.{self.schema.version_table_name}" - ) + version_identifier = self._table_identifier(self.schema.version_table_name) try: version_table = catalog.load_table(version_identifier) version_table.delete(EqualTo("schema_name", self.schema.name)) @@ -1000,7 +1028,7 @@ def complete_load(self, load_id: str) -> None: # Process each table for table_name, file_data in pending_files.items(): - identifier = f"{namespace}.{table_name}" + identifier = self._table_identifier(table_name) try: self._commit_table_files( @@ -1026,7 +1054,7 @@ def complete_load(self, load_id: str) -> None: def _store_completed_load(self, catalog, load_id: str) -> None: """Persist a load completion row in the internal _dlt_loads table.""" loads_table_name = self.schema.loads_table_name - identifier = f"{self.config.namespace}.{loads_table_name}" + identifier = self._table_identifier(loads_table_name) inserted_at = pendulum.now("UTC").naive() load_row_schema = pa.schema([ diff --git a/tests/test_class_based_atomic.py b/tests/test_class_based_atomic.py index 9de587c..df87831 100644 --- a/tests/test_class_based_atomic.py +++ b/tests/test_class_based_atomic.py @@ -348,5 +348,58 @@ def batch_2(): shutil.rmtree(temp_dir, ignore_errors=True) +def test_internal_table_prefix_remaps_only_physical_metadata_tables(): + """Keep dlt's logical names while supporting restrictive catalogs.""" + temp_dir = tempfile.mkdtemp() + warehouse_path = f"{temp_dir}/warehouse" + catalog_path = f"{temp_dir}/catalog.db" + + try: + from dlt_iceberg import iceberg_rest + from pyiceberg.catalog import load_catalog + + @dlt.resource(name="events", write_disposition="append") + def events(): + yield {"event_id": 1, "value": 10} + + pipeline = dlt.pipeline( + pipeline_name="test_internal_table_prefix", + destination=iceberg_rest( + catalog_uri=f"sqlite:///{catalog_path}", + warehouse=f"file://{warehouse_path}", + namespace="test_ns", + internal_table_prefix="dlt", + ), + dataset_name="test_dataset", + ) + + first_load = pipeline.run(events()) + second_load = pipeline.run(events()) + + assert not first_load.has_failed_jobs + assert not second_load.has_failed_jobs + + catalog = load_catalog( + "dlt_catalog", + type="sql", + uri=f"sqlite:///{catalog_path}", + warehouse=f"file://{warehouse_path}", + ) + table_names = {name for _, name in catalog.list_tables("test_ns")} + + assert "events" in table_names + assert {"dlt_loads", "dlt_version"} <= table_names + assert not any(name.startswith("_dlt_") for name in table_names) + assert len(catalog.load_table("test_ns.dlt_loads").scan().to_arrow()) == 2 + + with pipeline.destination_client() as client: + assert ( + client._physical_table_name(client.schema.state_table_name) + == "dlt_pipeline_state" + ) + finally: + shutil.rmtree(temp_dir, ignore_errors=True) + + if __name__ == "__main__": pytest.main([__file__, "-v", "-s"])