[Feature]: Matched file filtering via IN (...) doesn't scale for large file sets

[ethan-tyler]

Is your feature request related to a problem?

PR #4111 adds scan_files_where_matches() to identify files containing matching rows for DML rewrites (UPDATE now, MERGE/DELETE next). The second scan is restricted to matched files via a plan-time IN (...) predicate:

let file_list = valid_files
    .iter()
    .flat_map(|arr| arr.iter().flatten().map(|v| v.to_string()))
    .map(|v| lit(wrap_partition_value_in_dict(ScalarValue::Utf8(Some(v)))))
    .collect_vec();

let plan = LogicalPlanBuilder::scan("source", table_source, None)?
    .filter(col(FILE_ID_COLUMN_DEFAULT).in_list(file_list, false))?
    // ...

This builds one Expr::Literal per matched file. For rewrites touching thousands of files, that means large expression trees, slow planning, and memory overhead. This is a file level restriction implemented as a DF expression over the synthetic file-id column, not a Parquet row predicate.

Describe the solution you'd like

Scan-planning allowlist:

Instead of constructing Expr::InList with one literal per matched file, plumb an allowlist of matched file ids into the scan planning path and filter the candidate files before building Parquet FileGroups / PartitionedFiles. This avoids plan-time bloat while preserving true file pruning (don't read Parquet for excluded files).

Optional safety net:

exec-time membership filtering in DeltaScanExec (e.g., per run via split_by_file_id_runs() from #4112) to guard against mixed batch edge cases. However, this happens after Parquet reads and doesn't provide I/O savings.

Something like:

1. Add file_id_allowlist: Option<Arc<HashSet<String>>> to DeltaScanConfig
2. Add with_file_id_allowlist() builder method
3. Filter in scan planning when assembling file list / file groups (before Parquet read plan)
4. Optionally also guard in DeltaScanExec for mixed-batch correctness

Describe alternatives you've considered

Alternatives considered:

• Semi-join against MemTable - avoids literal list, uses DF join machinery.
• IN (subquery) - depends on DF planner behavior
• Chunked IN (...) - still builds large expressions, just delays the cliff

Priority

Medium - Would be helpful

Additional context

Related:

• refactor: move files scan to separate function #4111
• fix(datafusion): handle coalesced multi-file batches in next-scan #4112

Contribution

• I'm willing to submit a pull request for this feature
• I can help with testing this feature
• I can help with documentation for this feature

[roeap]

Thanks for opening this.

Since we have several folks who are working with very large logs often arising from streaming ingestion, we should make sure we are not building excessively large expressions here.

In fact the previous implementation DeltaTableProvider had a similar with_files method. The main reason this one did not make onto the new provider is that is was build based on Add actions which came with several constraints in our design. Specifically we need to receive the file information during scanning from delta kernel as it does some processing that allows us to support column mapping and DVs (and other features for that matter).

So pushing the allow list into the scan is a good place to start. When it comes to actually enforcing this we should push this down even further i believe or maybe just have a closer look at whats going on ...

I.e. this is fundamentally limiting log replay. For this we have the kernel internal means via predicate we pass to it. this is currently limited b/c we have yet to implement OpaquePredicate (or similar name :)) for the DF expressions, which mean we cannot yet translate all predicates.

then we have the explicit lists of files, where the main challenge is that we keep the unique representation for a file (i.e. the file url) consistent.

As of now, I believe the right place to do so is ScanFileStream which produces scan file metadata we then use to create the parquet scan. WE also do some stats parsing gymnastics in there so that we can pass relevant stats to the parquet scan as well.

We do similar things in the ScanRowOutStream which right now always processes all statistics. This is only required though until we have gotten rid of the current table provider, which exposes stats on the TableProviderLevel so long before we are even planning the scan. Once that is gone, we should be in a good place to consolidate the two streams and push the matched file filter down into whatver comes out :).

This is all of course open for better approaches and just my initial gut reaction on how to proceed with this.

I know @rtyler and @abhiaagarwal might find this interesting as well.

[ethan-tyler]

@roeap - Thanks for the detailed response and discussion! After digging through the next-scan path, here's how I see the options:

Filter in scan-stream:

Filter matched files in the scan replay path (ScanFileStream / replay_files()) before get_data_scan_plan() turns them into PartitionedFile / file groups. Removes the plan-time IN (...) expression entirely while preserving true file pruning and excluded files never become Parquet scan inputs.

something like:

• Plumb file_allowlist: Option<Arc<HashSet<String>>> into the next-scan path
• Filter in ScanFileStream output or replay_files() before file group construction
• Use ScanFileContext.file_url.to_string() as canonical identity (same as FILE_ID_COLUMN_DEFAULT)

Makes sense that ScanFileStream is the right place for now. Once the old provider is gone, the filter can move to wherever the streams consolidate.

We could consider a eager-snapshot pre-filter as a follow up add on:

EagerSnapshot::files() returns cached RecordBatch with a path column. Pre-filtering before scan_metadata_from(...) is technically possible. However, path contains raw strings (relative, not URL-decoded) while the matched file set uses file_url.to_string().

Would need a normalization step to match representations. Probably not worth it when ScanFileStream filtering is straightforward and already eliminates Parquet planning/I/O for excluded files. Thoughts on this?

To your point, building a ScanBuilder::with_file_allowlist(...) in the kernel would be the better long term solution:

Add a file allowlist to kernel, enforce at scan-file emission time. This would be the leanest abstraction, centralizes file identity normalization. Obviously would need API work but could be something to add to the roadmap.