fix: map pushed-down filter columns by position instead of by name - #25259
Conversation
github-actions
Bot
added
core
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25259 +/- ##
==========================================
- Coverage 81.95% 81.91% -0.04%
==========================================
Files 1133 1134 +1
Lines 423828 425684 +1856
Branches 423828 425684 +1856
==========================================
+ Hits 347344 348701 +1357
- Misses 55890 56303 +413
- Partials 20594 20680 +86 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Physical filter pushdown resolved a pushed filter's columns in the child schema by name. When the child contains duplicate column names, such as a.id and b.id from a nested join, a join dynamic filter on the second column was rewritten onto the first and silently dropped matching rows. Make FilterRemapper positional only: an identity mapping for nodes that preserve their input schema, or an explicit parent-output to child-input mapping. HashJoin builds the mapping from its column indices and output projection (semi joins map output keys to the paired key), FilterExec maps through its embedded projection, ProjectionExec substitutes the expression at each output position, and AggregateExec maps grouping outputs to the input column each grouping expression reads. `from_child_with_allowed_indices` is kept as a deprecated wrapper that resolves the allowed indices positionally.
haohuaijin
changed the title
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
github-actions
Bot
added
the
documentation
|
Hi @adriangb @nuno-faria could you take a look when you have time? Thanks! |
|
@jayzhan211 wonder if you'd be interested in reviewing this change since you are also working on dynamic filter pushdown improvements? |
jayzhan211
left a comment
jayzhan211
left a comment
There was a problem hiding this comment.
Thanks @haohuaijin ,
Two related suggestions on FilterRemapper and the deprecated from_child_with_allowed_indices.
1. Model the two resolution modes as an enum instead of Option<HashMap>
column_mapping: None currently doubles as "identity", and remap_column then keys a second rule (the name check) on is_none(). Two behaviours are encoded in one absent value and their validation rules live in different places. Naming the modes lets each variant own its rule, with no cross-field conditional:
/// How a parent output position resolves to a child input position.
enum ColumnMapping {
/// Output position `i` reads child position `i`, and the child field at
/// that position must carry the same name. Used by schema-preserving
/// nodes such as sort, repartition and coalesce.
Identity,
/// Explicit output -> input positions supplied by a node that projects,
/// reorders or pairs columns (joins, aggregates, projected filters).
/// Names may differ, so the caller is trusted.
Explicit(HashMap<usize, usize>),
}
pub(crate) struct FilterRemapper {
child_schema: SchemaRef,
mapping: ColumnMapping,
}
impl FilterRemapper {
pub(crate) fn new(child_schema: SchemaRef) -> Self {
Self { child_schema, mapping: ColumnMapping::Identity }
}
pub(crate) fn with_column_mapping(
child_schema: SchemaRef,
column_mapping: HashMap<usize, usize>,
) -> Self {
Self { child_schema, mapping: ColumnMapping::Explicit(column_mapping) }
}
fn remap_column(&self, col: &Column) -> Option<Column> {
let index = match &self.mapping {
ColumnMapping::Identity => {
let field = self.child_schema.fields().get(col.index())?;
(field.name() == col.name()).then_some(col.index())?
}
ColumnMapping::Explicit(mapping) => *mapping.get(&col.index())?,
};
let field = self.child_schema.fields().get(index)?;
Some(Column::new(field.name(), index))
}
}try_remap, from_child, from_child_with_column_mapping and the FilterExec / ProjectionExec call sites are unchanged. Identity keeps the name check as a safety net, so a downstream node that calls from_child while reordering columns gets "not pushed" rather than a wrong column, which is what main did.
2. Keep the legacy semantics inside the deprecated function
from_child_with_allowed_indices now builds an identity map and goes through the trusted Explicit path, so it drops the name check entirely. The documented use of this public API was join-style nodes whose parent positions differ from the child's, relying on the name lookup. Such a downstream caller now gets parent index n_left + i resolved to child index n_left + i: out of range is dropped, in range it silently becomes a different column. That was correct on main, so it is a wrong-results regression hidden behind a deprecation warning, which is the failure class this PR is fixing.
Since the parent filters are available in the function, it can reproduce the old resolution exactly through the new positional API, and the legacy code disappears with the function when the deprecation window ends:
use datafusion_physical_expr::utils::collect_columns;
#[deprecated(
since = "56.0.0",
note = "use `from_child` for matching schemas or `from_child_with_column_mapping` when positions differ"
)]
pub fn from_child_with_allowed_indices(
parent_filters: &[Arc<dyn PhysicalExpr>],
allowed_indices: HashSet<usize>,
child: &Arc<dyn crate::ExecutionPlan>,
) -> Result<Self> {
if parent_filters.is_empty() {
return Ok(Self::empty());
}
// Preserve the historical behaviour for callers that have not migrated:
// an allowed parent column resolves to the first child field with the
// same name. New code must pass positions explicitly.
let child_schema = child.schema();
let column_mapping: HashMap<usize, usize> = parent_filters
.iter()
.flat_map(collect_columns)
.filter(|col| allowed_indices.contains(&col.index()))
.filter_map(|col| {
child_schema
.index_of(col.name())
.ok()
.map(|child_index| (col.index(), child_index))
})
.collect();
Self::from_child_with_column_mapping(parent_filters, column_mapping, child)
}With this, the paragraph in 56.0.0.md saying the deprecated function "no longer preserves the previous name-based mapping" can go; the note becomes "deprecated, migrate to from_child_with_column_mapping because name resolution is ambiguous with duplicate field names". test_from_child_with_allowed_indices_resolves_by_position would need to change to assert the legacy behaviour (or be replaced by a test that an unresolvable name is not pushed).
If you would rather not carry any name-based path at all, removing the function outright is the other safe option, since a compile error cannot produce wrong results. Deprecating it with changed semantics is the one combination I would avoid.
|
It seems this PR might solve the issue I just open #25296 |
|
Thanks @jayzhan211 @adriangb |
4 participants
Which issue does this PR close?
enable_join_dynamic_filter_pushdownsilently drops rows: the pushed filter is remapped by column NAME and lands on the wrong same-named column #25244.FilterExecwith a projection can discard matching rows with duplicate column names #25262.ProjectionExeccan map duplicate output aliases to the wrong expression #25263.AggregateExeccan confuse same-named grouping columns #25264.Rationale for this change
Enabling
datafusion.optimizer.enable_join_dynamic_filter_pushdowncan silently drop matching rows when the probe side of a join contains several columns with the same name, for examplea.idandb.idfrom a nested join. Physical filter pushdown resolved a pushed filter's columns in the child schema by name, so a predicate on the secondidcolumn was rewritten to the firstidcolumn, which holds a different value.The same name-based mapping also affects TopK dynamic filters under the default configuration (#25296). For a nested join of orders and payments that both expose
amount,ORDER BY p.amount LIMIT 1can push the payment threshold ontoorders.amount. Later matching orders are incorrectly pruned, returning(100, 30)instead of(300, 10).Every operator that forwards parent filters had the same weakness, in slightly different forms:
HashJoinExeccomputed which output columns belong to each side by position, but then resolved the child column by name.FilterExec,SortExec,RepartitionExec,CoalesceBatchesExec,UnionExecand similar nodes used the generic name lookup even though their output positions equal their input positions.FilterExecwith an embedded projection resolved unconsumed parent filters back into input coordinates by name.ProjectionExeclooked up output aliases by name to find the expression to substitute, so two outputs aliasedidboth mapped to the first.AggregateExecrestricted pushdown to grouping positions but resolved the input column by name.The join-filter regression queries return one row with dynamic filtering disabled and zero rows with it enabled across these shapes. The
FilterExec,ProjectionExecandAggregateExeccases were found while auditing the remaining name-based paths for #25244; they share the root cause, so this PR fixes them together. The TopK case in #25296 is fixed by the same positional mapping.What changes are included in this PR?
Built-in operators now remap filter columns by position. The deprecated public API retains its historical name-based resolution for compatibility.
FilterRemapperuses aColumnMappingenum:Identitypreserves positions and checks that names match;Explicituses a caller-supplied parent-output to child-input mapping and allows names to differ.ChildFilterDescription::from_childnow maps by position. NewChildFilterDescription::from_child_with_column_mappingtakes an explicitHashMap<usize, usize>.HashJoinExecbuilds the explicit mapping from itscolumn_indicesand output projection. For semi joins, output join keys on the emitted side are mapped to the paired key on the other side, which also supports differently named keys.FilterExecmaps through its embedded projection in both pushdown phases and when folding unconsumed parent filters back into its predicate.ProjectionExecsubstitutes the expression at each output position instead of looking the alias up in the output schema.AggregateExecmaps each grouping output position to the input column that grouping expression reads; non-column grouping expressions are not forwarded, as before.ChildFilterDescription::from_child_with_allowed_indicesis kept as a deprecated wrapper that preserves name-based resolution to the first matching child field. It translates allowed parent column references into an explicit positional mapping; new callers should supply positions directly to avoid ambiguous duplicate names.What is the testing strategy for this PR?
dynamic_filter_pushdown_config.sltgains four regression queries over two small Parquet tables, run once with join dynamic filtering disabled and once enabled, asserting identical rows: nested joins withRepartitionExecbetween them (the query fromenable_join_dynamic_filter_pushdownsilently drops rows: the pushed filter is remapped by column NAME and lands on the wrong same-named column #25244), aFilterExecwith an embedded projection, aProjectionExecwith duplicate aliases, and anAggregateExecgrouping on same-named columns. All four return zero rows onmainwith dynamic filtering enabled.dynamic_filter_pushdown_config.sltalso covers Filter pushdown picks the wrong column when a join output has duplicated column names, causing wrong results with TopK dynamic filters #25296 using nested joins of orders, payments and customers, with one row per batch and one row per orders Parquet row group. The query returns(300, 10)with TopK dynamic filtering disabled, enabled, and enabled together with Parquet filter pushdown. Onmainat85d4cbb0a9, both enabled cases reproduce the incorrect(100, 30)result; all three pass on this branch.datafusion/core/tests/physical_optimizer/filter_pushdown.rsgains focused tests that callgather_filters_for_pushdowndirectly onRepartitionExec,HashJoinExec(duplicate child columns with and without projection, both semi join directions with differently named keys, and a semi join whose key is not a plain column),FilterExecwith a projection in both phases,ProjectionExecwith duplicate aliases, andAggregateExecwith reordered same-named grouping columns. Further tests cover the deprecatedfrom_child_with_allowed_indiceswrapper preserving the first name match, accepting allowed parent indices outside the child schema, and rejecting unresolvable names, the identity mapping rejecting a column whose name differs from the child field at that position, andFilterExecrejecting an unconsumed parent filter outside its projection.Passed locally on the updated implementation:
cargo fmt --all./ci/scripts/doc_prettier_check.sh --write --allow-dirtycargo test --profile ci -p datafusion --test core_integration physical_optimizer::filter_pushdown— 70 tests passed.cargo test --profile ci --test sqllogictests -- dynamic_filter_pushdown_config.slt— passed.mainin both TopK-enabled configurations with the expected wrong-result mismatch.Are there any user-facing changes?
Queries that push join or TopK dynamic filters through operators with duplicate column names now return the correct rows, including the default-configuration TopK query in #25296.
API change in
datafusion-physical-plan:ChildFilterDescription::from_child_with_allowed_indicesis deprecated in favour offrom_childand the newfrom_child_with_column_mapping; the deprecated function preserves its previous name-based behavior. Callers should migrate to explicit positions because duplicate names make name resolution ambiguous.ChildFilterDescription::from_childalso resolves by position, which only affects callers that used it on a node whose output positions differ from its child's.