Support custom dlt metadata table prefixes

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.
+
 </details>
 
 ## Catalog Examples

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([

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"])