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 92f4e8f3ee add ExecutionPlan::dynamic_expressions_produced() method
(#24068)
92f4e8f3ee is described below
commit 92f4e8f3eeb7b8426399063b4226a591d0c062b3
Author: Jayant Shrivastava <[email protected]>
AuthorDate: Fri Aug 7 11:41:40 2026 -0400
add ExecutionPlan::dynamic_expressions_produced() method (#24068)
## Which issue does this PR close?
- Closes: https://github.com/apache/datafusion/issues/23814
- Closes:
https://github.com/datafusion-contrib/datafusion-distributed/issues/584
## Rationale for this change
External users who wish to propagate dynamic filter updates across
network boundaries need to know in which direction updates need to flow.
Thus `ExecutionPlan::apply_expressions` is not enough. The proposal in
https://github.com/apache/datafusion/issues/23814 is to provide a
separate API for dynamic filter producers.
## What changes are included in this PR?
This change adds a new method
`ExecutionPlan::dynamic_expressions_produced(&self) -> Vec<Arc<dyn
PhysicalExpr>>` which should return dynamic filters produced by the
`ExecutionPlan`.
## Are these changes tested?
Yes. Adds a new invariant that all expressions returned by
`ExecutionPlan::dynamic_expressions_produced` have an expression id.
## Are there any user-facing changes?
There's a new API, `ExecutionPlan::dynamic_expressions_produced(&self)
-> Vec<Arc<dyn PhysicalExpr>>`
---
.../tests/physical_optimizer/filter_pushdown.rs | 81 +++++++++++++++++++-
datafusion/ffi/src/execution_plan.rs | 73 +++++++++++++++++-
datafusion/ffi/src/physical_expr/mod.rs | 15 ++++
datafusion/ffi/src/tests/mod.rs | 14 +++-
datafusion/ffi/tests/ffi_execution_plan.rs | 18 +++++
datafusion/physical-plan/src/aggregates/mod.rs | 39 +++++++---
datafusion/physical-plan/src/execution_plan.rs | 86 +++++++++++++++++++++-
.../physical-plan/src/joins/hash_join/exec.rs | 36 +++++----
datafusion/physical-plan/src/sorts/sort.rs | 57 +++++++++-----
.../proto/tests/cases/roundtrip_physical_plan.rs | 18 +++--
10 files changed, 380 insertions(+), 57 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
index e6f51266c4..f2b1f66c28 100644
--- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
+++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
@@ -34,7 +34,11 @@ use datafusion::{
scalar::ScalarValue,
};
use datafusion_catalog::memory::DataSourceExec;
-use datafusion_common::config::ConfigOptions;
+use datafusion_common::{
+ JoinType,
+ config::ConfigOptions,
+ tree_node::{TreeNode, TreeNodeRecursion},
+};
use datafusion_datasource::{
PartitionedFile, file_groups::FileGroup,
file_scan_config::FileScanConfigBuilder,
};
@@ -60,6 +64,7 @@ use datafusion_physical_plan::{
coalesce_partitions::CoalescePartitionsExec,
collect,
filter::{FilterExec, FilterExecBuilder},
+ joins::{HashJoinExec, PartitionMode},
projection::ProjectionExec,
repartition::RepartitionExec,
sorts::sort::SortExec,
@@ -2914,12 +2919,16 @@ async fn
test_hashjoin_hash_table_pushdown_collect_left() {
);
}
-// Not portable to sqllogictest: asserts on
`HashJoinExec::dynamic_filter_for_test().is_used()`
-// which is a debug-only API. The observable behavior (probe-side scan
+// Not portable to sqllogictest: asserts on the dynamic filter's Arc ownership.
+// The observable behavior (probe-side scan
// receiving the dynamic filter when the data source supports it) is
// already covered by the simpler CollectLeft port in
push_down_filter_parquet.slt;
// the with_support(false) branch has no SQL analog (parquet always supports
// pushdown).
+#[expect(
+ deprecated,
+ reason = "the borrowed getter avoids adding a producer Arc that is_used
would count"
+)]
#[tokio::test]
async fn test_hashjoin_dynamic_filter_pushdown_is_used() {
use datafusion_common::JoinType;
@@ -3101,6 +3110,72 @@ async fn test_filter_with_projection_pushdown() {
assert_batches_eq!(expected, &result);
}
+#[test]
+fn test_discover_dynamic_expression_producers() {
+ fn producer_count(plan: &Arc<dyn ExecutionPlan>) -> usize {
+ let mut count = 0;
+ plan.apply(|node| {
+ count += node.dynamic_expressions_produced().len();
+ Ok(TreeNodeRecursion::Continue)
+ })
+ .expect("plan traversal should succeed");
+ count
+ }
+
+ let build_schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Utf8, false),
+ Field::new("b", DataType::Int32, false),
+ ]));
+ let build_scan = TestScanBuilder::new(Arc::clone(&build_schema))
+ .with_support(true)
+ .with_batches(vec![
+ record_batch!(("a", Utf8, ["foo", "bar"]), ("b", Int32, [1,
2])).unwrap(),
+ ])
+ .build();
+
+ let probe_schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Utf8, false),
+ Field::new("c", DataType::Float64, false),
+ ]));
+ let probe_scan = TestScanBuilder::new(Arc::clone(&probe_schema))
+ .with_support(true)
+ .with_batches(vec![
+ record_batch!(
+ ("a", Utf8, ["foo", "bar", "baz", "qux"]),
+ ("c", Float64, [1.0, 2.0, 3.0, 4.0])
+ )
+ .unwrap(),
+ ])
+ .build();
+
+ let plan = Arc::new(
+ HashJoinExec::try_new(
+ build_scan,
+ probe_scan,
+ vec![(
+ col("a", &build_schema).unwrap(),
+ col("a", &probe_schema).unwrap(),
+ )],
+ None,
+ &JoinType::Inner,
+ None,
+ PartitionMode::CollectLeft,
+ datafusion_common::NullEquality::NullEqualsNothing,
+ false,
+ )
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+ assert_eq!(producer_count(&plan), 0);
+
+ let mut config = ConfigOptions::default();
+ config.optimizer.enable_dynamic_filter_pushdown = true;
+ config.execution.parquet.pushdown_filters = true;
+ let optimized_plan = FilterPushdown::new_post_optimization()
+ .optimize(plan, &config)
+ .unwrap();
+ assert_eq!(producer_count(&optimized_plan), 1);
+}
+
// ==== Filter pushdown through SortExec tests ====
/// FilterExec above a plain SortExec (no fetch) should be pushed below it.
diff --git a/datafusion/ffi/src/execution_plan.rs
b/datafusion/ffi/src/execution_plan.rs
index 087a351b69..5191b2fc03 100644
--- a/datafusion/ffi/src/execution_plan.rs
+++ b/datafusion/ffi/src/execution_plan.rs
@@ -33,6 +33,7 @@ use tokio::runtime::Handle;
use crate::config::FFI_ConfigOptions;
use crate::execution::FFI_TaskContext;
+use crate::physical_expr::FFI_PhysicalExpr;
use crate::physical_expr::metrics::FFI_MetricsSet;
use crate::plan_properties::FFI_PlanProperties;
use crate::record_batch_stream::FFI_RecordBatchStream;
@@ -50,6 +51,10 @@ pub struct FFI_ExecutionPlan {
/// Return a vector of children plans
pub children: unsafe extern "C" fn(plan: &Self) -> SVec<FFI_ExecutionPlan>,
+ /// Return the dynamic expressions produced by this plan node.
+ pub dynamic_expressions_produced:
+ unsafe extern "C" fn(plan: &Self) -> SVec<FFI_PhysicalExpr>,
+
pub with_new_children:
unsafe extern "C" fn(plan: &Self, children: SVec<Self>) ->
FFI_Result<Self>,
@@ -138,6 +143,16 @@ unsafe extern "C" fn children_fn_wrapper(
.collect()
}
+unsafe extern "C" fn dynamic_expressions_produced_fn_wrapper(
+ plan: &FFI_ExecutionPlan,
+) -> SVec<FFI_PhysicalExpr> {
+ plan.inner()
+ .dynamic_expressions_produced()
+ .into_iter()
+ .map(FFI_PhysicalExpr::from)
+ .collect()
+}
+
unsafe extern "C" fn with_new_children_fn_wrapper(
plan: &FFI_ExecutionPlan,
children: SVec<FFI_ExecutionPlan>,
@@ -306,6 +321,7 @@ impl FFI_ExecutionPlan {
Self {
properties: properties_fn_wrapper,
children: children_fn_wrapper,
+ dynamic_expressions_produced:
dynamic_expressions_produced_fn_wrapper,
with_new_children: with_new_children_fn_wrapper,
name: name_fn_wrapper,
execute: execute_fn_wrapper,
@@ -442,6 +458,15 @@ impl ExecutionPlan for ForeignExecutionPlan {
}
}
+ fn dynamic_expressions_produced(
+ &self,
+ ) -> Vec<Arc<dyn datafusion_physical_plan::PhysicalExpr>> {
+ unsafe { (self.plan.dynamic_expressions_produced)(&self.plan) }
+ .iter()
+ .map(<Arc<dyn datafusion_physical_plan::PhysicalExpr>>::from)
+ .collect()
+ }
+
fn repartitioned(
&self,
target_partitions: usize,
@@ -474,8 +499,9 @@ impl ExecutionPlan for ForeignExecutionPlan {
#[cfg(any(test, feature = "integration-tests"))]
pub mod tests {
- use datafusion_physical_plan::Partitioning;
+ use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr,
lit};
use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
+ use datafusion_physical_plan::{Partitioning, PhysicalExpr};
use super::*;
@@ -483,6 +509,7 @@ pub mod tests {
pub struct EmptyExec {
props: Arc<PlanProperties>,
children: Vec<Arc<dyn ExecutionPlan>>,
+ dynamic_expressions: Vec<Arc<dyn PhysicalExpr>>,
metrics: Option<MetricsSet>,
statistics: Option<Statistics>,
}
@@ -497,6 +524,7 @@ pub mod tests {
Boundedness::Bounded,
)),
children: Vec::default(),
+ dynamic_expressions: Vec::default(),
metrics: None,
statistics: None,
}
@@ -511,6 +539,14 @@ pub mod tests {
self.statistics = Some(statistics);
self
}
+
+ pub fn with_dynamic_expressions(
+ mut self,
+ dynamic_expressions: Vec<Arc<dyn PhysicalExpr>>,
+ ) -> Self {
+ self.dynamic_expressions = dynamic_expressions;
+ self
+ }
}
impl DisplayAs for EmptyExec {
@@ -543,6 +579,7 @@ pub mod tests {
Ok(Arc::new(EmptyExec {
props: Arc::clone(&self.props),
children,
+ dynamic_expressions: self.dynamic_expressions.clone(),
metrics: self.metrics.clone(),
statistics: self.statistics.clone(),
}))
@@ -556,6 +593,10 @@ pub mod tests {
unimplemented!()
}
+ fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ self.dynamic_expressions.iter().map(Arc::clone).collect()
+ }
+
fn metrics(&self) -> Option<MetricsSet> {
self.metrics.clone()
}
@@ -571,6 +612,10 @@ pub mod tests {
}
}
+ pub(crate) fn create_dynamic_filter() -> Arc<DynamicFilterPhysicalExpr> {
+ Arc::new(DynamicFilterPhysicalExpr::new(vec![], lit(true)))
+ }
+
#[test]
fn test_round_trip_ffi_execution_plan() -> Result<()> {
let schema = Arc::new(arrow::datatypes::Schema::new(vec![
@@ -600,6 +645,32 @@ pub mod tests {
Ok(())
}
+ #[test]
+ fn test_ffi_execution_plan_dynamic_expressions_produced() -> Result<()> {
+ let schema = Arc::new(arrow::datatypes::Schema::empty());
+ let dynamic_filter = create_dynamic_filter();
+ let expected_id = dynamic_filter
+ .expression_id()
+ .expect("dynamic filters always have an expression ID");
+ let expression: Arc<dyn PhysicalExpr> = Arc::clone(&dynamic_filter) as
_;
+ let original_plan =
+
Arc::new(EmptyExec::new(schema).with_dynamic_expressions(vec![expression]));
+
+ let mut ffi_plan = FFI_ExecutionPlan::new(original_plan, None);
+ ffi_plan.library_marker_id = crate::mock_foreign_marker_id;
+ let foreign_plan: Arc<dyn ExecutionPlan> = (&ffi_plan).try_into()?;
+ foreign_plan.check_invariants(
+ datafusion_physical_plan::execution_plan::InvariantLevel::Always,
+ )?;
+
+ let produced = foreign_plan.dynamic_expressions_produced();
+ assert_eq!(produced.len(), 1);
+ assert_eq!(produced[0].expression_id(), Some(expected_id));
+ drop(foreign_plan);
+ assert_eq!(produced[0].expression_id(), Some(expected_id));
+ Ok(())
+ }
+
#[test]
fn test_ffi_execution_plan_children() -> Result<()> {
let schema = Arc::new(arrow::datatypes::Schema::new(vec![
diff --git a/datafusion/ffi/src/physical_expr/mod.rs
b/datafusion/ffi/src/physical_expr/mod.rs
index 9a3ee27393..debe0bd7bf 100644
--- a/datafusion/ffi/src/physical_expr/mod.rs
+++ b/datafusion/ffi/src/physical_expr/mod.rs
@@ -120,6 +120,8 @@ pub struct FFI_PhysicalExpr {
pub is_volatile_node: unsafe extern "C" fn(&Self) -> bool,
+ pub expression_id: unsafe extern "C" fn(&Self) -> FFI_Option<u64>,
+
// Display trait
pub display: unsafe extern "C" fn(&Self) -> SString,
@@ -387,6 +389,13 @@ unsafe extern "C" fn is_volatile_node_fn_wrapper(expr:
&FFI_PhysicalExpr) -> boo
let expr = expr.inner();
expr.is_volatile_node()
}
+
+unsafe extern "C" fn expression_id_fn_wrapper(
+ expr: &FFI_PhysicalExpr,
+) -> FFI_Option<u64> {
+ expr.inner().expression_id().into()
+}
+
unsafe extern "C" fn display_fn_wrapper(expr: &FFI_PhysicalExpr) -> SString {
let expr = expr.inner();
format!("{expr}").into()
@@ -434,6 +443,7 @@ unsafe extern "C" fn clone_fn_wrapper(expr:
&FFI_PhysicalExpr) -> FFI_PhysicalEx
snapshot: snapshot_fn_wrapper,
snapshot_generation: snapshot_generation_fn_wrapper,
is_volatile_node: is_volatile_node_fn_wrapper,
+ expression_id: expression_id_fn_wrapper,
display: display_fn_wrapper,
hash: hash_fn_wrapper,
clone: clone_fn_wrapper,
@@ -477,6 +487,7 @@ impl From<Arc<dyn PhysicalExpr>> for FFI_PhysicalExpr {
snapshot: snapshot_fn_wrapper,
snapshot_generation: snapshot_generation_fn_wrapper,
is_volatile_node: is_volatile_node_fn_wrapper,
+ expression_id: expression_id_fn_wrapper,
display: display_fn_wrapper,
hash: hash_fn_wrapper,
clone: clone_fn_wrapper,
@@ -713,6 +724,10 @@ impl PhysicalExpr for ForeignPhysicalExpr {
fn is_volatile_node(&self) -> bool {
unsafe { (self.expr.is_volatile_node)(&self.expr) }
}
+
+ fn expression_id(&self) -> Option<u64> {
+ unsafe { (self.expr.expression_id)(&self.expr).into() }
+ }
}
impl Eq for ForeignPhysicalExpr {}
diff --git a/datafusion/ffi/src/tests/mod.rs b/datafusion/ffi/src/tests/mod.rs
index 357953674a..ad7e069546 100644
--- a/datafusion/ffi/src/tests/mod.rs
+++ b/datafusion/ffi/src/tests/mod.rs
@@ -28,6 +28,7 @@ use datafusion_common::stats::Precision;
use datafusion_common::{ColumnStatistics, Statistics};
use datafusion_common::{Result, ScalarValue};
use datafusion_expr::{Expr, TableType};
+use datafusion_physical_expr::PhysicalExpr;
use datafusion_physical_plan::ExecutionPlan;
use sync_provider::create_sync_table_provider;
use udf_udaf_udwf::{
@@ -40,7 +41,7 @@ use crate::catalog_provider::FFI_CatalogProvider;
use crate::catalog_provider_list::FFI_CatalogProviderList;
use crate::config::extension_options::FFI_ExtensionOptions;
use crate::execution_plan::FFI_ExecutionPlan;
-use crate::execution_plan::tests::EmptyExec;
+use crate::execution_plan::tests::{EmptyExec, create_dynamic_filter};
use crate::physical_optimizer::FFI_PhysicalOptimizerRule;
use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
use crate::proto::physical_extension_codec::FFI_PhysicalExtensionCodec;
@@ -112,6 +113,8 @@ pub struct ForeignLibraryModule {
pub create_empty_exec: extern "C" fn() -> FFI_ExecutionPlan,
+ pub create_exec_with_dynamic_expressions: extern "C" fn() ->
FFI_ExecutionPlan,
+
pub create_exec_with_statistics: extern "C" fn() -> FFI_ExecutionPlan,
pub create_table_with_statistics:
@@ -177,6 +180,14 @@ pub(crate) extern "C" fn create_empty_exec() ->
FFI_ExecutionPlan {
FFI_ExecutionPlan::new(plan, None)
}
+pub(crate) extern "C" fn create_exec_with_dynamic_expressions() ->
FFI_ExecutionPlan {
+ let schema = Arc::new(Schema::empty());
+ let expression: Arc<dyn PhysicalExpr> = create_dynamic_filter();
+ let plan =
+
Arc::new(EmptyExec::new(schema).with_dynamic_expressions(vec![expression]));
+ FFI_ExecutionPlan::new(plan, None)
+}
+
/// Returns canonical statistics used by both the producer and consumer sides
of
/// the integration tests so round-trips can be asserted without hard-coding
/// the values in two places.
@@ -275,6 +286,7 @@ pub extern "C" fn datafusion_ffi_get_module() ->
ForeignLibraryModule {
create_rank_udwf: create_ffi_rank_func,
create_extension_options: config::create_extension_options,
create_empty_exec,
+ create_exec_with_dynamic_expressions,
create_exec_with_statistics,
create_table_with_statistics,
create_physical_optimizer_rule:
diff --git a/datafusion/ffi/tests/ffi_execution_plan.rs
b/datafusion/ffi/tests/ffi_execution_plan.rs
index 7d04e828bd..a26ba9a1a6 100644
--- a/datafusion/ffi/tests/ffi_execution_plan.rs
+++ b/datafusion/ffi/tests/ffi_execution_plan.rs
@@ -26,6 +26,7 @@ mod tests {
use datafusion_ffi::execution_plan::{ExecutionPlanPrivateData,
tests::EmptyExec};
use datafusion_ffi::tests::utils::get_module;
use datafusion_physical_plan::ExecutionPlan;
+ use datafusion_physical_plan::execution_plan::InvariantLevel;
use std::sync::Arc;
#[test]
@@ -63,6 +64,23 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_ffi_execution_plan_dynamic_expressions_cross_library()
+ -> Result<(), DataFusionError> {
+ let module = get_module()?;
+ let plan = (module.create_exec_with_dynamic_expressions)();
+ let plan: Arc<dyn ExecutionPlan> = (&plan).try_into()?;
+ assert!(plan.is::<ForeignExecutionPlan>());
+ plan.check_invariants(InvariantLevel::Always)?;
+
+ let produced = plan.dynamic_expressions_produced();
+ assert_eq!(produced.len(), 1);
+ assert!(produced[0].expression_id().is_some());
+ drop(plan);
+ assert!(produced[0].expression_id().is_some());
+ Ok(())
+ }
+
#[test]
fn test_ffi_execution_plan_new_sets_runtimes_on_children()
-> Result<(), DataFusionError> {
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs
b/datafusion/physical-plan/src/aggregates/mod.rs
index ad5ee3db17..6af67d0482 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -1087,6 +1087,10 @@ impl AggregateExec {
}
/// Returns the dynamic filter expression for this aggregate, if set.
+ #[deprecated(
+ since = "55.0.0",
+ note = "Use ExecutionPlan::dynamic_expressions_produced instead"
+ )]
pub fn dynamic_filter_expr(&self) ->
Option<&Arc<DynamicFilterPhysicalExpr>> {
self.dynamic_filter.as_ref().map(|df| &df.filter)
}
@@ -1945,6 +1949,16 @@ impl ExecutionPlan for AggregateExec {
vec![&self.input]
}
+ fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ self.dynamic_filter
+ .iter()
+ .map(|dynamic_filter| {
+ Arc::<DynamicFilterPhysicalExpr>::clone(&dynamic_filter.filter)
+ as Arc<dyn PhysicalExpr>
+ })
+ .collect()
+ }
+
fn with_new_children(
self: Arc<Self>,
children: Vec<Arc<dyn ExecutionPlan>>,
@@ -2175,14 +2189,12 @@ impl ExecutionPlan for AggregateExec {
limit: options.limit() as u64,
descending: options.descending(),
});
- let dynamic_filter = match self.dynamic_filter_expr() {
- Some(filter) => {
- let expr: Arc<dyn PhysicalExpr> =
- Arc::clone(filter) as Arc<dyn PhysicalExpr>;
- Some(ctx.encode_expr(&expr)?)
- }
- None => None,
- };
+ let dynamic_filter = self
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
+ .map(|expr| ctx.encode_expr(&expr))
+ .transpose()?;
Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(
@@ -7509,11 +7521,14 @@ mod tests {
lit(false),
));
let agg = agg.with_dynamic_filter_expr(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());
// The aggregate's filter should now resolve to the new inner
expression.
- let swapped = agg
- .dynamic_filter_expr()
- .expect("should still have dynamic filter")
+ let swapped = produced[0]
+ .downcast_ref::<DynamicFilterPhysicalExpr>()
+ .expect("produced expression should be a
DynamicFilterPhysicalExpr")
.current()?;
assert_eq!(format!("{swapped}"), format!("{}", lit(false)));
@@ -7557,7 +7572,7 @@ mod tests {
child,
Arc::clone(&schema),
)?;
- assert!(agg.dynamic_filter_expr().is_none());
+ assert!(agg.dynamic_expressions_produced().is_empty());
let df = Arc::new(DynamicFilterPhysicalExpr::new(
vec![col("a", &schema)?],
diff --git a/datafusion/physical-plan/src/execution_plan.rs
b/datafusion/physical-plan/src/execution_plan.rs
index 11a8d69a37..bc4ece5adf 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -39,6 +39,7 @@ pub use datafusion_physical_expr::{
};
use std::any::Any;
+use std::collections::HashSet;
use std::fmt::Debug;
use std::sync::{Arc, LazyLock};
@@ -165,6 +166,22 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send +
Sync {
check_default_invariants(self, check)
}
+ /// Returns the dynamic expressions produced by this plan node.
+ ///
+ /// A dynamic expression is produced when this node updates or completes
its
+ /// runtime state during execution. Expressions that this node only
consumes
+ /// must not be returned. This method is shallow and does not include
dynamic
+ /// expressions produced by child plans.
+ ///
+ /// Each returned expression must have a [`PhysicalExpr::expression_id`]
+ /// since all dynamic expressions such as [`DynamicFilterPhysicalExpr`]
+ /// have an expression id.
+ ///
+ /// [`DynamicFilterPhysicalExpr`]:
datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr
+ fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ Vec::new()
+ }
+
/// Specifies simple per-child input distribution requirements.
///
/// Deprecated: override [`Self::input_distribution_requirements`] instead.
@@ -1317,6 +1334,27 @@ macro_rules! check_len {
};
}
+/// All dynamic expressions must have an expression id.
+fn check_dynamic_expression_invariants<P: ExecutionPlan + ?Sized>(
+ plan: &P,
+) -> Result<()> {
+ let mut produced_ids = HashSet::new();
+ for expr in plan.dynamic_expressions_produced() {
+ let Some(expression_id) = expr.expression_id() else {
+ return internal_err!(
+ "{}::dynamic_expressions_produced returned an expression
without an expression ID",
+ plan.name()
+ );
+ };
+ assert_or_internal_err!(
+ produced_ids.insert(expression_id),
+ "{}::dynamic_expressions_produced returned duplicate expression ID
{expression_id}",
+ plan.name()
+ );
+ }
+ Ok(())
+}
+
/// Checks a set of invariants that apply to all ExecutionPlan implementations.
/// Returns an error if the given node does not conform.
pub fn check_default_invariants<P: ExecutionPlan + ?Sized>(
@@ -1330,6 +1368,7 @@ pub fn check_default_invariants<P: ExecutionPlan +
?Sized>(
check_len!(plan, benefits_from_input_partitioning, children_len);
plan.input_distribution_requirements()
.check_invariants(plan, check)?;
+ check_dynamic_expression_invariants(plan)?;
Ok(())
}
@@ -1732,13 +1771,26 @@ mod tests {
use arrow::array::{DictionaryArray, Int32Array, NullArray, RunArray};
use arrow::datatypes::{DataType, Field, Schema};
+ use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr,
lit};
#[derive(Debug)]
- pub struct EmptyExec;
+ pub struct EmptyExec {
+ dynamic_expressions: Vec<Arc<dyn PhysicalExpr>>,
+ }
impl EmptyExec {
pub fn new(_schema: SchemaRef) -> Self {
- Self
+ Self {
+ dynamic_expressions: vec![],
+ }
+ }
+
+ fn with_dynamic_expressions(
+ mut self,
+ dynamic_expressions: Vec<Arc<dyn PhysicalExpr>>,
+ ) -> Self {
+ self.dynamic_expressions = dynamic_expressions;
+ self
}
}
@@ -1772,6 +1824,10 @@ mod tests {
unimplemented!()
}
+ fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ self.dynamic_expressions.iter().map(Arc::clone).collect()
+ }
+
fn execute(
&self,
_partition: usize,
@@ -1789,6 +1845,32 @@ mod tests {
}
}
+ #[test]
+ fn test_dynamic_expression_invariants() -> Result<()> {
+ let schema = Arc::new(Schema::empty());
+ let dynamic: Arc<dyn PhysicalExpr> =
+ Arc::new(DynamicFilterPhysicalExpr::new(vec![], lit(true)));
+ let valid = EmptyExec::new(Arc::clone(&schema))
+ .with_dynamic_expressions(vec![Arc::clone(&dynamic)]);
+ check_default_invariants(&valid, InvariantLevel::Always)?;
+
+ let missing_id =
+
EmptyExec::new(Arc::clone(&schema)).with_dynamic_expressions(vec![lit(true)]);
+ let error = check_default_invariants(&missing_id,
InvariantLevel::Always)
+ .unwrap_err()
+ .strip_backtrace();
+ assert!(error.contains("without an expression ID"), "{error}");
+
+ let duplicate = EmptyExec::new(schema)
+ .with_dynamic_expressions(vec![Arc::clone(&dynamic), dynamic]);
+ let error = check_default_invariants(&duplicate,
InvariantLevel::Always)
+ .unwrap_err()
+ .strip_backtrace();
+ assert!(error.contains("duplicate expression ID"), "{error}");
+
+ Ok(())
+ }
+
#[derive(Debug)]
pub struct RenamedEmptyExec;
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs
b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 9a1c0b0f63..7cc2644668 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -964,8 +964,11 @@ impl HashJoinExec {
self.null_equality
}
- /// Get the dynamic filter expression for testing purposes.
- /// Returns the dynamic filter expression for this hash join, if set.
+ /// Returns the dynamic filter expression produced by this hash join, if
set.
+ #[deprecated(
+ since = "55.0.0",
+ note = "Use ExecutionPlan::dynamic_expressions_produced instead"
+ )]
pub fn dynamic_filter_expr(&self) ->
Option<&Arc<DynamicFilterPhysicalExpr>> {
self.dynamic_filter.as_ref().map(|df| &df.filter)
}
@@ -1330,6 +1333,16 @@ impl ExecutionPlan for HashJoinExec {
vec![&self.left, &self.right]
}
+ fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ self.dynamic_filter
+ .iter()
+ .map(|dynamic_filter| {
+ Arc::<DynamicFilterPhysicalExpr>::clone(&dynamic_filter.filter)
+ as Arc<dyn PhysicalExpr>
+ })
+ .collect()
+ }
+
/// Creates a new HashJoinExec with different children while preserving
configuration.
///
/// This method is called during query optimization when the optimizer
creates new
@@ -1808,12 +1821,10 @@ impl ExecutionPlan for HashJoinExec {
.transpose()?;
let dynamic_filter = self
- .dynamic_filter_expr()
- .map(|df| {
- let df_expr: Arc<dyn PhysicalExpr> =
- Arc::clone(df) as Arc<dyn PhysicalExpr>;
- ctx.encode_expr(&df_expr)
- })
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
+ .map(|expr| ctx.encode_expr(&expr))
.transpose()?;
Ok(Some(protobuf::PhysicalPlanNode {
@@ -6866,7 +6877,7 @@ mod tests {
NullEquality::NullEqualsNothing,
false,
)?;
- assert!(join.dynamic_filter_expr().is_none());
+ assert!(join.dynamic_expressions_produced().is_empty());
let df = Arc::new(DynamicFilterPhysicalExpr::new(
vec![Arc::new(Column::new("b1", 1)) as _],
@@ -6874,11 +6885,10 @@ mod tests {
));
let join = join.with_dynamic_filter_expr(Arc::clone(&df))?;
- let restored = join
- .dynamic_filter_expr()
- .expect("should have dynamic filter");
+ let produced = join.dynamic_expressions_produced();
+ assert_eq!(produced.len(), 1);
assert_eq!(
- restored
+ produced[0]
.expression_id()
.expect("DynamicFilterPhysicalExpr always has an
expression_id"),
df.expression_id()
diff --git a/datafusion/physical-plan/src/sorts/sort.rs
b/datafusion/physical-plan/src/sorts/sort.rs
index 4b30aede7d..b1b6a84fd9 100644
--- a/datafusion/physical-plan/src/sorts/sort.rs
+++ b/datafusion/physical-plan/src/sorts/sort.rs
@@ -1097,6 +1097,10 @@ impl SortExec {
}
/// Returns the dynamic filter expression for this sort (TopK), if set.
+ #[deprecated(
+ since = "55.0.0",
+ note = "Use ExecutionPlan::dynamic_expressions_produced instead"
+ )]
pub fn dynamic_filter_expr(&self) ->
Option<Arc<DynamicFilterPhysicalExpr>> {
self.filter.as_ref().map(|f| f.read().expr())
}
@@ -1274,6 +1278,13 @@ impl ExecutionPlan for SortExec {
vec![&self.input]
}
+ fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ self.filter
+ .iter()
+ .map(|filter| filter.read().expr() as Arc<dyn PhysicalExpr>)
+ .collect()
+ }
+
fn benefits_from_input_partitioning(&self) -> Vec<bool> {
vec![false]
}
@@ -1575,13 +1586,12 @@ impl ExecutionPlan for SortExec {
})
})
.collect::<Result<Vec<_>>>()?;
- let dynamic_filter = match self.dynamic_filter_expr() {
- Some(df) => {
- let df_expr: Arc<dyn PhysicalExpr> = df;
- Some(ctx.encode_expr(&df_expr)?)
- }
- None => None,
- };
+ let dynamic_filter = self
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
+ .map(|expr| ctx.encode_expr(&expr))
+ .transpose()?;
Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(
protobuf::physical_plan_node::PhysicalPlanType::Sort(Box::new(
@@ -3189,9 +3199,9 @@ mod tests {
.with_fetch(Some(10));
// SortExec with fetch creates a dynamic filter automatically.
- let original_id = sort
- .dynamic_filter_expr()
- .expect("should have dynamic filter with fetch")
+ let produced = sort.dynamic_expressions_produced();
+ assert_eq!(produced.len(), 1);
+ let original_id = produced[0]
.expression_id()
.expect("DynamicFilterPhysicalExpr always has an expression_id");
@@ -3204,9 +3214,9 @@ mod tests {
.expression_id()
.expect("DynamicFilterPhysicalExpr always has an expression_id");
let sort = sort.with_dynamic_filter_expr(Arc::clone(&new_df))?;
- let restored_id = sort
- .dynamic_filter_expr()
- .expect("should still have dynamic filter")
+ let produced = sort.dynamic_expressions_produced();
+ assert_eq!(produced.len(), 1);
+ let restored_id = produced[0]
.expression_id()
.expect("DynamicFilterPhysicalExpr always has an expression_id");
assert_eq!(restored_id, new_id);
@@ -3236,6 +3246,19 @@ mod tests {
);
}
+ fn dynamic_filter_produced(
+ plan: &dyn ExecutionPlan,
+ ) -> Arc<DynamicFilterPhysicalExpr> {
+ let expr = plan
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
+ .expect("plan should produce a dynamic filter");
+ (expr as Arc<dyn std::any::Any + Send + Sync>)
+ .downcast::<DynamicFilterPhysicalExpr>()
+ .expect("produced expression should be a
DynamicFilterPhysicalExpr")
+ }
+
#[tokio::test]
async fn test_preserved_topk_filter_waits_for_all_sort_partitions() ->
Result<()> {
let schema = Arc::new(Schema::new(vec![Field::new("a",
DataType::Int32, false)]));
@@ -3259,9 +3282,7 @@ mod tests {
.with_fetch(Some(2))
.with_preserve_partitioning(true);
- let dynamic_filter = sort
- .dynamic_filter_expr()
- .expect("fetch sort should create a dynamic filter");
+ let dynamic_filter = dynamic_filter_produced(&sort);
let sort = Arc::new(sort);
let task_ctx = Arc::new(TaskContext::default());
@@ -3308,9 +3329,7 @@ mod tests {
.with_preserve_partitioning(true)
.with_fetch(Some(2));
- let dynamic_filter = sort
- .dynamic_filter_expr()
- .expect("fetch sort should keep the dynamic filter");
+ let dynamic_filter = dynamic_filter_produced(&sort);
assert_eq!(
dynamic_filter
.expression_id()
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index e7bcffea13..a05f2a6ee3 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -4557,7 +4557,9 @@ fn test_hash_join_with_dynamic_filter_roundtrip() ->
Result<()> {
.downcast_ref::<HashJoinExec>()
.expect("Should be HashJoinExec");
let deserialized_hash_join_df = deserialized_join
- .dynamic_filter_expr()
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
.expect("HashJoinExec should have a dynamic filter after roundtrip");
// Extract the dynamic filter pushed down to the probe side's
ParquetSource.
@@ -4565,7 +4567,7 @@ fn test_hash_join_with_dynamic_filter_roundtrip() ->
Result<()> {
// The HashJoinExec's dynamic filter and the probe side's predicate should
// refer to the same underlying expression.
- let plan_df: Arc<dyn PhysicalExpr> = deserialized_hash_join_df.clone();
+ let plan_df = deserialized_hash_join_df;
assert_dynamic_filters_equal(&plan_df, &deserialized_predicate);
assert_dynamic_filter_update_is_visible(&plan_df,
&deserialized_predicate)?;
@@ -4710,7 +4712,9 @@ fn test_aggregate_with_dynamic_filter_roundtrip() ->
Result<()> {
.downcast_ref::<AggregateExec>()
.expect("Should be AggregateExec");
let deserialized_agg_df = deserialized_agg
- .dynamic_filter_expr()
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
.expect("AggregateExec should have a dynamic filter after roundtrip");
// Extract the dynamic filter pushed down to the child ParquetSource.
@@ -4718,7 +4722,7 @@ fn test_aggregate_with_dynamic_filter_roundtrip() ->
Result<()> {
// The AggregateExec's dynamic filter and the child's predicate should
// refer to the same underlying expression.
- let plan_df: Arc<dyn PhysicalExpr> = deserialized_agg_df.clone();
+ let plan_df = deserialized_agg_df;
assert_dynamic_filters_equal(&plan_df, &deserialized_predicate);
assert_dynamic_filter_update_is_visible(&plan_df,
&deserialized_predicate)?;
@@ -4778,7 +4782,9 @@ fn test_sort_topk_with_dynamic_filter_roundtrip() ->
Result<()> {
.downcast_ref::<SortExec>()
.expect("Should be SortExec");
let deserialized_sort_df = deserialized_sort
- .dynamic_filter_expr()
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
.expect("SortExec should have a dynamic filter after roundtrip");
// Extract the dynamic filter pushed down to the child ParquetSource.
@@ -4786,7 +4792,7 @@ fn test_sort_topk_with_dynamic_filter_roundtrip() ->
Result<()> {
// The SortExec's dynamic filter and the child's predicate should
// refer to the same underlying expression.
- let plan_df: Arc<dyn PhysicalExpr> = deserialized_sort_df;
+ let plan_df = deserialized_sort_df;
assert_dynamic_filters_equal(&plan_df, &deserialized_predicate);
assert_dynamic_filter_update_is_visible(&plan_df,
&deserialized_predicate)?;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
