Pyiceberg bumped, improved error messages, batching for iceberg

Author: anuunchin

dlt/common/libs/deltalake.py

@@ -118,6 +118,7 @@ def merge_delta_table(
     table: DeltaTable,
     data: Union[pa.Table, pa.RecordBatchReader],
     schema: TTableSchema,
+    load_table_name: str,
 ) -> None:
     """Merges in-memory Arrow data into on-disk Delta table."""
 
@@ -149,7 +150,10 @@ def merge_delta_table(
 
         qry.execute()
     else:
-        raise ValueError(f'Merge strategy "{strategy}" not supported.')
+        raise ValueError(
+            f'Merge strategy "{strategy}" is not supported for Delta tables. '
+            f'Table: "{load_table_name}".'
+        )
 
 
 def get_delta_tables(

dlt/common/libs/pyiceberg.py

@@ -68,6 +68,7 @@ def merge_iceberg_table(
     table: IcebergTable,
     data: pa.Table,
     schema: TTableSchema,
+    load_table_name: str,
 ) -> None:
     """Merges in-memory Arrow data into on-disk Iceberg table."""
     strategy = schema["x-merge-strategy"]  # type: ignore[typeddict-item]
@@ -81,15 +82,22 @@ def merge_iceberg_table(
         else:
             join_cols = get_columns_names_with_prop(schema, "primary_key")
 
-        table.upsert(
-            df=ensure_iceberg_compatible_arrow_data(data),
-            join_cols=join_cols,
-            when_matched_update_all=True,
-            when_not_matched_insert_all=True,
-            case_sensitive=True,
-        )
+        for rb in data.to_batches(max_chunksize=1_000):
+            batch_tbl = pa.Table.from_batches([rb])
+            batch_tbl = ensure_iceberg_compatible_arrow_data(batch_tbl)
+
+            table.upsert(
+                df=batch_tbl,
+                join_cols=join_cols,
+                when_matched_update_all=True,
+                when_not_matched_insert_all=True,
+                case_sensitive=True,
+            )
     else:
-        raise ValueError(f'Merge strategy "{strategy}" not supported.')
+        raise ValueError(
+            f'Merge strategy "{strategy}" is not supported for Iceberg tables. '
+            f'Table: "{load_table_name}".'
+        )
 
 
 def get_sql_catalog(

dlt/destinations/impl/filesystem/filesystem.py

@@ -190,6 +190,7 @@ def run(self) -> None:
                     table=delta_table,
                     data=arrow_rbr,
                     schema=self._load_table,
+                    load_table_name=self.load_table_name,
                 )
             else:
                 location = self._job_client.get_open_table_location("delta", self.load_table_name)
@@ -239,6 +240,7 @@ def run(self) -> None:
                 table=table,
                 data=self.arrow_dataset.to_table(),
                 schema=self._load_table,
+                load_table_name=self.load_table_name,
             )
         else:
             write_iceberg_table(

docs/website/docs/dlt-ecosystem/destinations/iceberg.md

@@ -119,13 +119,9 @@ The [S3-compatible](./filesystem.md#using-s3-compatible-storage) interface for G
 ## Iceberg Azure scheme
 The `az` [scheme](./filesystem.md#supported-schemes) is not supported when using the `iceberg` table format. Please use the `abfss` scheme. This is because `pyiceberg`, which dlt used under the hood, currently does not support `az`.
 
-## Table format `merge` support (**experimental**)
+## Table format `merge` support
 The [`upsert`](../../general-usage/merge-loading.md#upsert-strategy) merge strategy is supported for `iceberg`. This strategy requires that the input data contains no duplicate rows based on the key columns, and that the target table also does not contain duplicates on those keys.
 
-:::caution
-The `upsert` merge strategy for the filesystem destination with Iceberg table format is **experimental**.
-:::
-
 ```py
 @dlt.resource(
     write_disposition={"disposition": "merge", "strategy": "upsert"},