When not matched by source triggers full scan

[aldder]

Environment

Delta-rs version: 0.25.5

Hello,
I have a scenario where i need to delete some data through merge operations on a partitioned table.
I explicitly set partition values in the merge predicate and delete predicate, and i expected that all files related to other partitions would have been skipped.

import os
import shutil
import polars as pl
from deltalake import DeltaTable


if os.path.exists("delta"):
    shutil.rmtree("delta")


df = pl.DataFrame({
    'foo': ['a', 'a', 'a', 'b', 'b'],  # partition col
    'bar': [1, 2, 3, 1, 2],
})

df.write_delta(
    'delta',
    delta_write_options={
        'partition_by': ['foo'],
    }
)

df2 = pl.DataFrame({
    'foo': ['a', 'a'],
    'bar': [1, 2],
})

out = df2.write_delta(
    'delta',
    mode='merge',
    delta_merge_options={
        'predicate': "tgt.foo = src.foo AND tgt.bar = src.bar AND tgt.foo = 'a'",
        'target_alias': 'tgt',
        'source_alias': 'src',
    }
).when_not_matched_by_source_delete(
    predicate="tgt.foo = 'a'"
).execute()

print(out)

output:

{
'num_source_rows': 2, 
'num_target_rows_inserted': 0, 
'num_target_rows_updated': 0, 
'num_target_rows_deleted': 1, 
'num_target_rows_copied': 2,
'num_output_rows': 2, 
'num_target_files_scanned': 2, 
'num_target_files_skipped_during_scan': 0,
'num_target_files_added': 1, 
'num_target_files_removed': 1, 
'execution_time_ms': 46, 
'scan_time_ms': 0, 
'rewrite_time_ms': 27
}

[ion-elgreco]

@aldder when not matched by source triggers a full scan

https://github.com/delta-io/delta-rs/blob/226c3976f76be29c3eb0cab6b4ecf1d1328cc8a3/crates/core/src/operations/merge/mod.rs#L842C1-L844C13