Feat: Allow control over streamed_exec in delta merge upsert

dlt/common/libs/deltalake.py

@@ -118,6 +118,7 @@ def merge_delta_table(
     data: Union[pa.Table, pa.RecordBatchReader],
     schema: TTableSchema,
     load_table_name: str,
+    streamed_exec: bool,
 ) -> None:
     """Merges in-memory Arrow data into on-disk Delta table."""
 
@@ -142,6 +143,7 @@ def merge_delta_table(
                 predicate=predicate,
                 source_alias="source",
                 target_alias="target",
+                streamed_exec=streamed_exec,
             )
             .when_matched_update_all()
             .when_not_matched_insert_all()

dlt/common/storages/configuration.py

@@ -202,6 +202,7 @@ class FilesystemConfiguration(BaseConfiguration):
     """Additional arguments as Config in botocore"""
     deltalake_storage_options: Optional[DictStrAny] = None
     deltalake_configuration: Optional[DictStrOptionalStr] = None
+    deltalake_streamed_exec: bool = True
 
     @property
     def protocol(self) -> str:

dlt/destinations/impl/filesystem/filesystem.py

@@ -200,6 +200,7 @@ def run(self) -> None:
                     data=arrow_rbr,
                     schema=self._load_table,
                     load_table_name=self.load_table_name,
+                    streamed_exec=self._job_client.config.deltalake_streamed_exec,
                 )
             else:
                 location = self._job_client.get_open_table_location("delta", self.load_table_name)

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

@@ -110,6 +110,13 @@ def my_upsert_resource():
 - Deleting records from nested tables not supported
   - This means updates to JSON columns that involve element removals are not propagated. For example, if you first load `{"key": 1, "nested": [1, 2]}` and then load `{"key": 1, "nested": [1]}`, then the record for element `2` will not be deleted from the nested table.
 
+By default, dlt runs Delta table upserts in streamed mode to reduce memory pressure. To enable the use of source table statistics to derive an early pruning predicate, set:
+
+```toml
+[destination.filesystem]
+deltalake_streamed_exec = false
+```
+
 ## Delta table format storage options and configuration
 You can pass storage options and configuration by configuring both `destination.filesystem.deltalake_storage_options` and
 `destination.filesystem.deltalake_configuration`:

tests/load/filesystem/test_filesystem_common.py

@@ -59,6 +59,7 @@ def test_filesystem_configuration() -> None:
         "kwargs": None,
         "deltalake_storage_options": None,
         "deltalake_configuration": None,
+        "deltalake_streamed_exec": True,
     }
 
 
@@ -234,6 +235,7 @@ def test_filesystem_configuration_with_additional_arguments() -> None:
             "delta.minWriterVersion": "7",
             "delta.enableChangeDataFeed": "true",
         },
+        deltalake_streamed_exec=False,
     )
     assert dict(config) == {
         "read_only": False,
@@ -247,6 +249,7 @@ def test_filesystem_configuration_with_additional_arguments() -> None:
             "delta.minWriterVersion": "7",
             "delta.enableChangeDataFeed": "true",
         },
+        "deltalake_streamed_exec": False,
     }