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/05/26 10:19:59 UTC
(datafusion) branch gh-readonly-queue/main/pr-22451-bdf8a6d0ad0706ccf29d85caf2436039e3bf77d3 created (now e8a641b314)
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-22451-bdf8a6d0ad0706ccf29d85caf2436039e3bf77d3
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at e8a641b314 refactor: add `try_to_proto` to `HashTableLookupExpr` (#22451)
This branch includes the following new commits:
new e8a641b314 refactor: add `try_to_proto` to `HashTableLookupExpr` (#22451)
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: add `try_to_proto` to `HashTableLookupExpr` (#22451)
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-22451-bdf8a6d0ad0706ccf29d85caf2436039e3bf77d3
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit e8a641b3146e6770e6966efbfa7906868e749617
Author: Anurag Tryambak Raut <12...@users.noreply.github.com>
AuthorDate: Tue May 26 15:49:37 2026 +0530
refactor: add `try_to_proto` to `HashTableLookupExpr` (#22451)
## Which issue does this PR close?
Part of #22435
## What changes are included?
Adds `try_to_proto` to `HashTableLookupExpr` so it participates in the
expression-local serialization pattern introduced in #21929.
`HashTableLookupExpr` holds a runtime `Arc<Map>` that cannot be
serialized, so `try_to_proto` replaces it with `lit(true)`. This is
safe because the filter is a performance optimisation only — `lit(true)`
passes all rows and the join produces correct results either way.
The centralized arm in `to_proto.rs` remains as a fallback for now.
Cleanup can follow in a separate PR once this lands.
## Are these changes tested?
Yes — covered by the existing `roundtrip_hash_table_lookup_expr_to_lit`
test in `datafusion/proto/tests/cases/roundtrip_physical_plan.rs`.
## Are there any user-facing changes?
No.
---------
Co-authored-by: Anurag Tryambak Raut <an...@Anurags-MacBook-Air.local>
---
Cargo.lock | 2 ++
datafusion/physical-plan/Cargo.toml | 7 +++++
.../src/joins/hash_join/partitioned_hash_eval.rs | 35 ++++++++++++++++++++--
datafusion/proto/Cargo.toml | 2 +-
datafusion/proto/src/physical_plan/to_proto.rs | 27 +----------------
5 files changed, 43 insertions(+), 30 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index a421d6d992..7bd9afc040 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2448,6 +2448,8 @@ dependencies = [
"datafusion-functions-window-common",
"datafusion-physical-expr",
"datafusion-physical-expr-common",
+ "datafusion-proto-common",
+ "datafusion-proto-models",
"futures",
"half",
"hashbrown 0.17.1",
diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml
index 5a05173eb3..c64d3cad69 100644
--- a/datafusion/physical-plan/Cargo.toml
+++ b/datafusion/physical-plan/Cargo.toml
@@ -42,6 +42,11 @@ force_hash_collisions = []
test_utils = ["arrow/test_utils"]
tokio_coop = []
tokio_coop_fallback = []
+proto = [
+ "dep:datafusion-proto-models",
+ "dep:datafusion-proto-common",
+ "datafusion-physical-expr-common/proto",
+]
[lib]
name = "datafusion_physical_plan"
@@ -65,6 +70,8 @@ datafusion-functions-aggregate-common = { workspace = true }
datafusion-functions-window-common = { workspace = true }
datafusion-physical-expr = { workspace = true, default-features = true }
datafusion-physical-expr-common = { workspace = true }
+datafusion-proto-common = { workspace = true, optional = true }
+datafusion-proto-models = { workspace = true, optional = true }
futures = { workspace = true }
half = { workspace = true }
hashbrown = { workspace = true }
diff --git a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs
index 0daac0bb86..46b087ad70 100644
--- a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs
@@ -215,7 +215,6 @@ pub struct HashTableLookupExpr {
/// Description for display
description: String,
}
-
impl HashTableLookupExpr {
/// Create a new HashTableLookupExpr
///
@@ -241,7 +240,6 @@ impl HashTableLookupExpr {
}
}
}
-
impl std::fmt::Debug for HashTableLookupExpr {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let cols = self
@@ -337,7 +335,38 @@ impl PhysicalExpr for HashTableLookupExpr {
}
}
}
-
+ #[cfg(feature = "proto")]
+ fn try_to_proto(
+ &self,
+ _ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
+ ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+ use datafusion_proto_models::protobuf;
+ use datafusion_proto_models::protobuf::physical_expr_node::ExprType;
+
+ // HashTableLookupExpr holds a runtime Arc<Map> (the build-side hash
+ // table) that cannot be serialized, so it is replaced with lit(true).
+ //
+ // Dynamic filtering is a performance optimisation only — replacing the
+ // lookup with lit(true) preserves correctness by allowing all rows
+ // through.
+ //
+ // If a plan is serialized before execution, HashTableLookupExpr is not
+ // yet present in the dynamic filter expression.
+ //
+ // If a plan is serialized after execution, any runtime-created
+ // HashTableLookupExpr is replaced during serialization. Re-executing
+ // the plan requires reset_state(), after which HashJoinExec rebuilds
+ // fresh dynamic filters at runtime.
+ let value = datafusion_proto_common::ScalarValue {
+ value: Some(datafusion_proto_common::scalar_value::Value::BoolValue(
+ true,
+ )),
+ };
+ Ok(Some(protobuf::PhysicalExprNode {
+ expr_id: None,
+ expr_type: Some(ExprType::Literal(value)),
+ }))
+ }
fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.description)
}
diff --git a/datafusion/proto/Cargo.toml b/datafusion/proto/Cargo.toml
index 53e7cd78dc..cfff8a9494 100644
--- a/datafusion/proto/Cargo.toml
+++ b/datafusion/proto/Cargo.toml
@@ -65,7 +65,7 @@ datafusion-expr = { workspace = true }
datafusion-functions-table = { workspace = true }
datafusion-physical-expr = { workspace = true, features = ["proto"] }
datafusion-physical-expr-common = { workspace = true, features = ["proto"] }
-datafusion-physical-plan = { workspace = true }
+datafusion-physical-plan = { workspace = true, features = ["proto"] }
datafusion-proto-common = { workspace = true }
datafusion-proto-models = { workspace = true }
object_store = { workspace = true }
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs
index 5dd643c84b..c359f651c0 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -39,7 +39,7 @@ use datafusion_physical_plan::expressions::{
CaseExpr, CastExpr, DynamicFilterPhysicalExpr, IsNotNullExpr, IsNullExpr, Literal,
NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
};
-use datafusion_physical_plan::joins::{HashExpr, HashTableLookupExpr};
+use datafusion_physical_plan::joins::HashExpr;
use datafusion_physical_plan::udaf::AggregateFunctionExpr;
use datafusion_physical_plan::windows::{PlainAggregateWindowExpr, WindowUDFExpr};
use datafusion_physical_plan::{Partitioning, PhysicalExpr, WindowExpr};
@@ -302,31 +302,6 @@ pub fn serialize_physical_expr_with_converter(
return Ok(node);
}
- // HashTableLookupExpr is used for dynamic filter pushdown in hash joins.
- // It contains an Arc<dyn JoinHashMapType> (the build-side hash table) which
- // cannot be serialized - the hash table is a runtime structure built during
- // execution on the build side.
- //
- // We replace it with lit(true) which is safe because:
- // 1. The filter is a performance optimization, not a correctness requirement
- // 2. lit(true) passes all rows, so no valid rows are incorrectly filtered out
- // 3. The join itself will still produce correct results, just without the
- // benefit of early filtering on the probe side
- //
- // In distributed execution, the remote worker won't have access to the hash
- // table anyway, so the best we can do is skip this optimization.
- if expr.downcast_ref::<HashTableLookupExpr>().is_some() {
- let value = datafusion_proto_common::ScalarValue {
- value: Some(datafusion_proto_common::scalar_value::Value::BoolValue(
- true,
- )),
- };
- return Ok(protobuf::PhysicalExprNode {
- expr_id,
- expr_type: Some(protobuf::physical_expr_node::ExprType::Literal(value)),
- });
- }
-
if let Some(expr) = expr.downcast_ref::<UnKnownColumn>() {
Ok(protobuf::PhysicalExprNode {
expr_id,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org