chore: make streamed_exec default

python/deltalake/table.py

@@ -976,7 +976,7 @@ def merge(
         error_on_type_mismatch: bool = True,
         writer_properties: Optional[WriterProperties] = None,
         large_dtypes: Optional[bool] = None,
-        streamed_exec: bool = False,
+        streamed_exec: bool = True,
         custom_metadata: Optional[Dict[str, str]] = None,
         post_commithook_properties: Optional[PostCommitHookProperties] = None,
         commit_properties: Optional[CommitProperties] = None,
@@ -994,7 +994,8 @@ def merge(
             error_on_type_mismatch: specify if merge will return error if data types are mismatching :default = True
             writer_properties: Pass writer properties to the Rust parquet writer
             large_dtypes: Deprecated, will be removed in 1.0
-            streamed_exec: Will execute MERGE using a LazyMemoryExec plan
+            streamed_exec: Will execute MERGE using a LazyMemoryExec plan, this improves memory pressure for large source tables. Enabling streamed_exec
+                implicitly disables source table stats to derive an early_pruning_predicate
             arrow_schema_conversion_mode: Large converts all types of data schema into Large Arrow types, passthrough keeps string/binary/list types untouched
             custom_metadata: Deprecated and will be removed in future versions. Use commit_properties instead.
             post_commithook_properties: properties for the post commit hook. If None, default values are used.

python/tests/test_generated_columns.py

@@ -220,7 +220,9 @@ def test_merge_with_g_during_schema_evolution(
     expected_data = pa.Table.from_pydict(
         {"id": [1, 2], "gc": [5, 5]}, schema=pa.schema([id_col, gc])
     )
-    assert table_with_gc.to_pyarrow_table() == expected_data
+    assert (
+        table_with_gc.to_pyarrow_table().sort_by([("id", "ascending")]) == expected_data
+    )
 
 
 def test_raise_when_gc_passed_merge_statement_during_schema_evolution(

python/tests/test_merge.py

@@ -1116,6 +1116,7 @@ def test_merge_timestamps_partitioned_2344(tmp_path: pathlib.Path, timezone, pre
         predicate="s.datetime = t.datetime",
         source_alias="s",
         target_alias="t",
+        streamed_exec=False,  # only with streamed execution off can we use stats to create a pruning predicate
     ).when_matched_update_all().when_not_matched_insert_all().execute()
 
     result = dt.to_pyarrow_table()
@@ -1441,13 +1442,15 @@ def test_merge_on_decimal_3033(tmp_path):
         predicate="target.timestamp = source.timestamp",
         source_alias="source",
         target_alias="target",
+        streamed_exec=False,
     ).when_matched_update_all().when_not_matched_insert_all().execute()
 
     dt.merge(
         source=table,
         predicate="target.timestamp = source.timestamp AND target.altitude = source.altitude",
         source_alias="source",
         target_alias="target",
+        streamed_exec=False,  # only with streamed execution off can we use stats to create a pruning predicate
     ).when_matched_update_all().when_not_matched_insert_all().execute()
 
     string_predicate = dt.history(1)[0]["operationParameters"]["predicate"]