You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/25 11:32:19 UTC
(datafusion) branch main updated: chore: deprecate `with_dynamic_filter_expr` on Sort/Aggregate/HashJoin (#24610)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 2f592edca6 chore: deprecate `with_dynamic_filter_expr` on Sort/Aggregate/HashJoin (#24610)
2f592edca6 is described below
commit 2f592edca60a5983b59e1ca0d11a872fa1f8eefe
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Tue Aug 25 11:23:53 2026 +0000
chore: deprecate `with_dynamic_filter_expr` on Sort/Aggregate/HashJoin (#24610)
## Which issue does this PR close?
<!-- No dedicated issue; this is follow-up cleanup for the proto hook
migration EPIC. -->
- Part of #23494.
## Rationale for this change
`SortExec::with_dynamic_filter_expr`,
`AggregateExec::with_dynamic_filter_expr` and
`HashJoinExec::with_dynamic_filter_expr` were added in #22011 for one
reason: at the
time `datafusion-proto` drove serialization from outside the crate where
these plans
are defined, so it needed a public setter to put a decoded dynamic
filter back onto a
plan. The matching getters (`dynamic_filter_expr`) were deprecated in
#24068 in favour
of `ExecutionPlan::dynamic_expressions_produced`.
Serialization now lives next to each plan (EPIC #23494), so the only
caller left for
each setter is that same plan's `try_from_proto`, in the same file.
Nothing outside
DataFusion uses them either: a GitHub code search for
`with_dynamic_filter_expr`
returns only the three definitions in this repository, and grepping
shallow clones of
datafusion-distributed, datafusion-ballista, datafusion-comet, sail,
delta-rs,
iceberg-rust, sedona-db and openobserve turns up no callers.
Handing external code a way to swap a plan's dynamic filter is also the
kind of API
that is easy to misuse: the filter is wired to the plan's internal
state, which is why
each setter has to re-validate the filter's children against the plan's
schema.
## What changes are included in this PR?
- Marks all three `with_dynamic_filter_expr` methods `#[deprecated(since
= "56.0.0")]`.
- Moves each body to a private `set_dynamic_filter`, called by
`try_from_proto` and by
the tests. The public method is a thin shim that delegates, so there is
no duplicated
logic and no behavior change.
The deprecation note follows the wording already used for
`AsyncFuncExec::async_exprs`,
deprecated for the same reason in #23825.
## Are these changes tested?
Covered by the existing tests, which now exercise the private method:
`test_with_dynamic_filter`,
`test_with_dynamic_filter_error_unsupported`,
`test_with_dynamic_filter_error_column_mismatch` (aggregate),
`test_with_dynamic_filter` /
`test_with_dynamic_filter_rejects_invalid_columns` (sort
and hash join), plus the dynamic-filter round-trip tests in
`datafusion-proto`.
`cargo test -p datafusion-physical-plan --features proto --lib` and
`cargo test -p datafusion-proto` pass, and `cargo clippy --all-targets
--workspace
--features avro,integration-tests,extended_tests -- -D warnings` is
clean.
## Are there any user-facing changes?
Three public methods are deprecated. Nothing is removed, so no code
stops compiling.
There is no replacement: dynamic filters are created by the plan itself
and restored by
the plan's own `try_from_proto`. If you have a use case for setting one
from outside,
please comment here or open an issue.
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
---
datafusion/physical-plan/src/aggregates/mod.rs | 32 ++++++++++++++++------
.../physical-plan/src/joins/hash_join/exec.rs | 26 +++++++++++++++---
datafusion/physical-plan/src/sorts/sort.rs | 27 ++++++++++++++----
3 files changed, 68 insertions(+), 17 deletions(-)
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index a7f08ea83a..59d382302a 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -1119,7 +1119,23 @@ impl AggregateExec {
/// Replace the dynamic filter expression. This method errors if the aggregate does not
/// support dynamic filtering or if the filter expression is incompatible with this
/// [`AggregateExec`].
+ #[deprecated(
+ since = "56.0.0",
+ note = "unused by DataFusion; `AggregateExec` restores its dynamic filter in `AggregateExec::try_from_proto`, which sets the field directly. There is no replacement; please open an issue if you have a use case for it."
+ )]
pub fn with_dynamic_filter_expr(
+ self,
+ filter: Arc<DynamicFilterPhysicalExpr>,
+ ) -> Result<Self> {
+ self.set_dynamic_filter(filter)
+ }
+
+ /// Replace the dynamic filter expression, validating that it is compatible
+ /// with this [`AggregateExec`].
+ ///
+ /// Only used to restore the filter when decoding a serialized plan: every
+ /// other code path creates the filter in [`AggregateExec::try_new`].
+ fn set_dynamic_filter(
mut self,
filter: Arc<DynamicFilterPhysicalExpr>,
) -> Result<Self> {
@@ -2695,7 +2711,7 @@ impl AggregateExec {
"AggregateExec dynamic_filter did not decode to a DynamicFilterPhysicalExpr"
)
})?;
- aggregate.with_dynamic_filter_expr(dynamic_filter)?
+ aggregate.set_dynamic_filter(dynamic_filter)?
} else {
let mut aggregate = aggregate;
aggregate.dynamic_filter = None;
@@ -7998,7 +8014,7 @@ mod tests {
}
}
- /// Test that [`AggregateExec::with_dynamic_filter_expr`] overrides the existing dynamic filter
+ /// Test that [`AggregateExec::set_dynamic_filter`] overrides the existing dynamic filter
#[test]
fn test_with_dynamic_filter() -> Result<()> {
let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
@@ -8025,7 +8041,7 @@ mod tests {
vec![col("a", &schema)?],
lit(false),
));
- let agg = agg.with_dynamic_filter_expr(Arc::clone(&new_df))?;
+ let agg = agg.set_dynamic_filter(Arc::clone(&new_df))?;
let produced = agg.dynamic_expressions_produced();
assert_eq!(produced.len(), 1);
assert_eq!(produced[0].expression_id(), new_df.expression_id());
@@ -8050,7 +8066,7 @@ mod tests {
};
// Hard to assert this because the filter is identical. No error means
// the filter was accepted. That's a good enough assertion for now.
- let _agg = agg.with_dynamic_filter_expr(remapped_df)?;
+ let _agg = agg.set_dynamic_filter(remapped_df)?;
Ok(())
}
@@ -8086,7 +8102,7 @@ mod tests {
Ok(())
}
- /// Test that [`AggregateExec::with_dynamic_filter_expr`] errors when the aggregate does not support dynamic filtering
+ /// Test that [`AggregateExec::set_dynamic_filter`] errors when the aggregate does not support dynamic filtering
#[test]
fn test_with_dynamic_filter_error_unsupported() -> Result<()> {
let schema = Arc::new(Schema::new(vec![
@@ -8115,11 +8131,11 @@ mod tests {
vec![col("a", &schema)?],
lit(true),
));
- assert!(agg.with_dynamic_filter_expr(df).is_err());
+ assert!(agg.set_dynamic_filter(df).is_err());
Ok(())
}
- /// Test that [`AggregateExec::with_dynamic_filter_expr`] errors when the column is not in the schema
+ /// Test that [`AggregateExec::set_dynamic_filter`] errors when the column is not in the schema
#[test]
fn test_with_dynamic_filter_error_column_mismatch() -> Result<()> {
let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
@@ -8143,7 +8159,7 @@ mod tests {
vec![Arc::new(Column::new("bad", 99)) as _],
lit(true),
));
- assert!(agg.with_dynamic_filter_expr(df).is_err());
+ assert!(agg.set_dynamic_filter(df).is_err());
Ok(())
}
}
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index fb8077264a..df10a8a5fc 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -1008,7 +1008,25 @@ impl HashJoinExec {
///
/// Validates that the filter's children reference valid columns in
/// the probe (right) side's schema.
+ #[deprecated(
+ since = "56.0.0",
+ note = "unused by DataFusion; `HashJoinExec` restores its dynamic filter in `HashJoinExec::try_from_proto`, which sets the field directly. There is no replacement; please open an issue if you have a use case for it."
+ )]
pub fn with_dynamic_filter_expr(
+ self,
+ filter: Arc<DynamicFilterPhysicalExpr>,
+ ) -> Result<Self> {
+ self.set_dynamic_filter(filter)
+ }
+
+ /// Set the dynamic filter on this hash join, resetting any internal state
+ /// that depends on an existing one and validating that the filter's
+ /// children reference valid columns in the probe (right) side's schema.
+ ///
+ /// Only used to restore the filter when decoding a serialized plan: every
+ /// other code path installs the filter in
+ /// [`ExecutionPlan::handle_child_pushdown_result`].
+ fn set_dynamic_filter(
mut self,
filter: Arc<DynamicFilterPhysicalExpr>,
) -> Result<Self> {
@@ -2101,7 +2119,7 @@ impl HashJoinExec {
"HashJoinExec dynamic_filter did not decode to a DynamicFilterPhysicalExpr"
)
})?;
- hash_join = hash_join.with_dynamic_filter_expr(df)?;
+ hash_join = hash_join.set_dynamic_filter(df)?;
}
Ok(Arc::new(hash_join))
@@ -7851,7 +7869,7 @@ mod tests {
vec![Arc::new(Column::new("b1", 1)) as _],
lit(true),
));
- let join = join.with_dynamic_filter_expr(Arc::clone(&df))?;
+ let join = join.set_dynamic_filter(Arc::clone(&df))?;
let produced = join.dynamic_expressions_produced();
assert_eq!(produced.len(), 1);
@@ -7894,7 +7912,7 @@ mod tests {
NullEquality::NullEqualsNothing,
false,
)?
- .with_dynamic_filter_expr(dynamic_filter)?;
+ .set_dynamic_filter(dynamic_filter)?;
let err = join.swap_inputs(PartitionMode::CollectLeft).unwrap_err();
assert_contains!(
@@ -8143,7 +8161,7 @@ mod tests {
vec![Arc::new(Column::new("bad", 99)) as _],
lit(true),
));
- assert!(join.with_dynamic_filter_expr(df).is_err());
+ assert!(join.set_dynamic_filter(df).is_err());
Ok(())
}
}
diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs
index c91df46351..9880c8b5e5 100644
--- a/datafusion/physical-plan/src/sorts/sort.rs
+++ b/datafusion/physical-plan/src/sorts/sort.rs
@@ -1113,7 +1113,24 @@ impl SortExec {
///
/// Validates that the filter's children reference valid columns in
/// the sort's input schema.
+ #[deprecated(
+ since = "56.0.0",
+ note = "unused by DataFusion; `SortExec` restores its dynamic filter in `SortExec::try_from_proto`, which sets the field directly. There is no replacement; please open an issue if you have a use case for it."
+ )]
pub fn with_dynamic_filter_expr(
+ self,
+ filter: Arc<DynamicFilterPhysicalExpr>,
+ ) -> Result<Self> {
+ self.set_dynamic_filter(filter)
+ }
+
+ /// Replace the dynamic filter expression for this sort, resetting any
+ /// internal state which depends on the previous one and validating that the
+ /// filter's children reference valid columns in the sort's input schema.
+ ///
+ /// Only used to restore the filter when decoding a serialized plan: every
+ /// other code path creates the filter in [`SortExec::with_fetch`].
+ fn set_dynamic_filter(
mut self,
filter: Arc<DynamicFilterPhysicalExpr>,
) -> Result<Self> {
@@ -1726,7 +1743,7 @@ impl SortExec {
"SortExec dynamic_filter did not decode to a DynamicFilterPhysicalExpr"
)
})?;
- new_sort.with_dynamic_filter_expr(df)?
+ new_sort.set_dynamic_filter(df)?
} else {
new_sort
};
@@ -3605,7 +3622,7 @@ mod tests {
.expression_id()
.expect("DynamicFilterPhysicalExpr always has an expression_id");
- // with_dynamic_filter replaces it with a new TopKDynamicFilters.
+ // set_dynamic_filter replaces it with a new TopKDynamicFilters.
let new_df = Arc::new(DynamicFilterPhysicalExpr::new(
vec![Arc::new(Column::new("a", 0)) as _],
lit(true),
@@ -3613,7 +3630,7 @@ mod tests {
let new_id = new_df
.expression_id()
.expect("DynamicFilterPhysicalExpr always has an expression_id");
- let sort = sort.with_dynamic_filter_expr(Arc::clone(&new_df))?;
+ let sort = sort.set_dynamic_filter(Arc::clone(&new_df))?;
let produced = sort.dynamic_expressions_produced();
assert_eq!(produced.len(), 1);
let restored_id = produced[0]
@@ -3725,7 +3742,7 @@ mod tests {
[PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0)))].into(),
input,
)
- .with_dynamic_filter_expr(dynamic_filter)?
+ .set_dynamic_filter(dynamic_filter)?
.with_preserve_partitioning(true)
.with_fetch(Some(2));
@@ -3774,7 +3791,7 @@ mod tests {
vec![Arc::new(Column::new("bad", 99)) as _],
lit(true),
));
- assert!(sort.with_dynamic_filter_expr(df).is_err());
+ assert!(sort.set_dynamic_filter(df).is_err());
Ok(())
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org