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/09/16 13:50:40 UTC
(datafusion) branch gh-readonly-queue/main/pr-25178-2cb6b54085ad36f412717b6a79b683d78ae2f48e created (now b0b547136f)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25178-2cb6b54085ad36f412717b6a79b683d78ae2f48e
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at b0b547136f refactor: exhaustively destructure dynamic filter and scalar subquery proto hooks (#25178)
This branch includes the following new commits:
new b0b547136f refactor: exhaustively destructure dynamic filter and scalar subquery proto hooks (#25178)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: refactor: exhaustively destructure dynamic filter and scalar subquery proto hooks (#25178)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25178-2cb6b54085ad36f412717b6a79b683d78ae2f48e
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b0b547136fbe03caca5c2374da9cd88989bd9f77
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Wed Sep 16 13:50:18 2026 +0000
refactor: exhaustively destructure dynamic filter and scalar subquery proto hooks (#25178)
## Which issue does this PR close?
Closes #24618.
## Rationale for this change
This change makes the compiler catch fields that serialization does not
handle. When a field is added to a dynamic filter, scalar subquery
expression, or its protobuf payload, the corresponding hook must account
for it. This helps prevent new fields from being silently dropped when
expressions are encoded and decoded.
## What changes are included in this PR?
All four encoding and decoding hooks for `DynamicFilterPhysicalExpr` and
`ScalarSubqueryExpr` explicitly list every field without a rest pattern.
Dynamic filter encoding also destructures the snapshot of its shared
inner state, and decoding handles the expression ID carried by the
enclosing message.
Comments explain why caches, watch channels, cached type and nullability
checks, and subquery results are omitted, including how they are rebuilt
or supplied.
The PR also replaces one use of unstable `std::assert_matches` in an
existing execution test with `assert!(…is_ok())`, allowing the required
checks to run on the pinned stable compiler.
## What is the testing strategy for this PR?
Existing tests cover filter identity, shared updates, remapped children,
subquery metadata, nested result scopes, and execution after
serialization. The physical expression suite passed 1,739 tests with 2
ignored, and all 13 relevant protobuf integration tests passed.
The full extended workspace run passed 11,591 Rust tests, with 8
ignored, and completed all 516 SQLLogicTest files. Formatting, Clippy
with all targets and features and warnings denied, and the complete
`./dev/rust_lint.sh` suite also passed.
## Are there any user-facing changes?
No.
---
datafusion/execution/src/async_stream.rs | 3 +-
.../src/expressions/dynamic_filters/mod.rs | 56 +++++++++++++++-------
datafusion/physical-expr/src/scalar_subquery.rs | 38 +++++++++------
3 files changed, 64 insertions(+), 33 deletions(-)
diff --git a/datafusion/execution/src/async_stream.rs b/datafusion/execution/src/async_stream.rs
index 0462c53a0f..a5e1c49ae9 100644
--- a/datafusion/execution/src/async_stream.rs
+++ b/datafusion/execution/src/async_stream.rs
@@ -302,7 +302,6 @@ mod test {
use crate::{async_stream, async_try_stream};
use futures::stream::FusedStream;
use futures::{Stream, StreamExt, pin_mut};
- use std::assert_matches;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -464,7 +463,7 @@ mod test {
pin_mut!(s);
for i in 0..3 {
- assert_matches!(tx.send(i).await, Ok(_));
+ assert!(tx.send(i).await.is_ok());
assert_eq!(Some(i), s.next().await);
}
diff --git a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs
index 15544d09e2..4d384e952e 100644
--- a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs
+++ b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs
@@ -606,13 +606,22 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
use datafusion_proto_models::protobuf;
use datafusion_proto_models::protobuf::physical_expr_node::ExprType;
- let children = self
- .children
+ let Self {
+ children,
+ remapped_children,
+ current_cache: _, // Runtime cache, repopulated by current().
+ inner,
+ state_watch: _, // Runtime channel, recreated from inner state by from_parts().
+ data_type: _, // Cached test invariant, recomputed from the expression.
+ nullable: _, // Cached test invariant, recomputed from the expression.
+ } = self;
+
+ let children = children
.iter()
.map(|c| ctx.encode_child(c))
.collect::<Result<Vec<_>>>()?;
- let remapped_children = match &self.remapped_children {
+ let remapped_children = match remapped_children {
Some(remapped) => remapped
.iter()
.map(|c| ctx.encode_child(c))
@@ -620,18 +629,23 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
None => vec![],
};
- let inner = self.inner.read().clone();
- let inner_expr = Box::new(ctx.encode_child(&inner.expr)?);
+ let Inner {
+ expression_id,
+ generation,
+ expr,
+ is_complete,
+ } = inner.read().clone();
+ let inner_expr = Box::new(ctx.encode_child(&expr)?);
Ok(Some(protobuf::PhysicalExprNode {
- expr_id: Some(inner.expression_id),
+ expr_id: Some(expression_id),
expr_type: Some(ExprType::DynamicFilter(Box::new(
protobuf::PhysicalDynamicFilterNode {
children,
remapped_children,
- generation: inner.generation,
+ generation,
inner_expr: Some(inner_expr),
- is_complete: inner.is_complete,
+ is_complete,
},
))),
}))
@@ -649,28 +663,36 @@ impl DynamicFilterPhysicalExpr {
proto: &datafusion_proto_models::protobuf::PhysicalExprNode,
ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
) -> Result<Arc<dyn PhysicalExpr>> {
+ use datafusion_proto_models::protobuf;
use datafusion_proto_models::protobuf::physical_expr_node::ExprType;
- let ExprType::DynamicFilter(df) = proto.expr_type.as_ref().ok_or_else(|| {
+ let protobuf::PhysicalExprNode { expr_id, expr_type } = proto;
+ let ExprType::DynamicFilter(df) = expr_type.as_ref().ok_or_else(|| {
internal_datafusion_err!("Missing expr_type in PhysicalExprNode")
})?
else {
return Err(internal_datafusion_err!("Expected DynamicFilter expr_type"));
};
+ let protobuf::PhysicalDynamicFilterNode {
+ children,
+ remapped_children,
+ generation,
+ inner_expr,
+ is_complete,
+ } = df.as_ref();
// Decode original children
- let children = df
- .children
+ let children = children
.iter()
.map(|c| ctx.decode(c))
.collect::<Result<Vec<_>>>()?;
// Decode remapped children (empty vec means None)
- let remapped_children = if df.remapped_children.is_empty() {
+ let remapped_children = if remapped_children.is_empty() {
None
} else {
Some(
- df.remapped_children
+ remapped_children
.iter()
.map(|c| ctx.decode(c))
.collect::<Result<Vec<_>>>()?,
@@ -678,13 +700,13 @@ impl DynamicFilterPhysicalExpr {
};
// Decode the inner expression
- let inner_expr_proto = df.inner_expr.as_ref().ok_or_else(|| {
+ let inner_expr_proto = inner_expr.as_ref().ok_or_else(|| {
internal_datafusion_err!("Missing inner_expr in PhysicalDynamicFilterNode")
})?;
let inner_expr = ctx.decode(inner_expr_proto)?;
// Restore the expression_id from the outer PhysicalExprNode
- let expression_id = proto.expr_id.ok_or_else(|| {
+ let expression_id = expr_id.ok_or_else(|| {
internal_datafusion_err!(
"Missing expr_id in PhysicalExprNode for DynamicFilter"
)
@@ -692,9 +714,9 @@ impl DynamicFilterPhysicalExpr {
let inner = Inner {
expression_id,
- generation: df.generation,
+ generation: *generation,
expr: inner_expr,
- is_complete: df.is_complete,
+ is_complete: *is_complete,
};
Ok(Arc::new(Self::from_parts(
diff --git a/datafusion/physical-expr/src/scalar_subquery.rs b/datafusion/physical-expr/src/scalar_subquery.rs
index 389baf3505..927df8c2ee 100644
--- a/datafusion/physical-expr/src/scalar_subquery.rs
+++ b/datafusion/physical-expr/src/scalar_subquery.rs
@@ -171,18 +171,25 @@ impl PhysicalExpr for ScalarSubqueryExpr {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_common::utils::usize_to_wire;
use datafusion_proto_models::protobuf;
+
+ let Self {
+ field,
+ index,
+ results: _, // Runtime state, supplied by ScalarSubqueryExec on decode.
+ } = self;
+
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::ScalarSubquery(
protobuf::PhysicalScalarSubqueryExprNode {
- data_type: Some(self.field.data_type().try_into()?),
- nullable: self.field.is_nullable(),
+ data_type: Some(field.data_type().try_into()?),
+ nullable: field.is_nullable(),
index: usize_to_wire(
- self.index.as_usize(),
+ index.as_usize(),
"ScalarSubqueryExpr",
"index",
)?,
- metadata: self.field.metadata().clone(),
+ metadata: field.metadata().clone(),
},
)),
}))
@@ -214,22 +221,25 @@ impl ScalarSubqueryExpr {
protobuf::physical_expr_node::ExprType::ScalarSubquery,
"ScalarSubqueryExpr",
);
- let data_type = require_proto_field(
- sq.data_type.as_ref(),
- "ScalarSubqueryExpr",
- "data_type",
- )?
- .try_into()?;
- let metadata = if sq.metadata.is_empty() {
+ let protobuf::PhysicalScalarSubqueryExprNode {
+ data_type,
+ nullable,
+ index,
+ metadata,
+ } = sq;
+ let data_type =
+ require_proto_field(data_type.as_ref(), "ScalarSubqueryExpr", "data_type")?
+ .try_into()?;
+ let metadata = if metadata.is_empty() {
None
} else {
- Some(FieldMetadata::from(sq.metadata.clone()))
+ Some(FieldMetadata::from(metadata.clone()))
};
Ok(Arc::new(ScalarSubqueryExpr::new_with_metadata(
data_type,
- sq.nullable,
+ *nullable,
metadata,
- SubqueryIndex::new(sq.index as usize),
+ SubqueryIndex::new(*index as usize),
results.clone(),
)))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org