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-24165-92f4e8f3eeb7b8426399063b4226a591d0c062b3 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit fc846dd36852c8628718175d88ff7bb7c003167e Author: Adrian Garcia Badaracco <[email protected]> AuthorDate: Fri Aug 7 13:39:55 2026 -0500 fix(proto): preserve HashJoinExec fetch across serialization (#24165) ## Which issue does this PR close? <!-- No dedicated issue; found while auditing the plans covered by the `try_to_proto`/`try_from_proto` migration EPIC. --> - Related to #23494 (found while auditing that EPIC's plans for unserialized fields). This is a bug fix, not part of the migration checklist. ## Rationale for this change `HashJoinExec.fetch` was silently dropped by protobuf serialization. `protobuf::HashJoinExecNode` had no `fetch` field, so `HashJoinExec`'s `try_to_proto` never wrote it and `try_from_proto` never restored it: a plan with `fetch = Some(n)` round-tripped to `fetch = None`. This is user-visible. The `limit_pushdown` physical optimizer rule pushes a limit into the join via `ExecutionPlan::with_fetch`, then marks the global state satisfied and drops the enclosing `GlobalLimitExec`. So after a proto round-trip the plan carried no limit at all, and a distributed executor (Ballista/Comet-style, anything that ships physical plans over the wire) returned more rows than the query asked for. ## What changes are included in this PR? - `datafusion.proto`: add `optional uint64 fetch = 12` to `HashJoinExecNode`. The field is **presence-tracked on purpose**, and this is the load-bearing detail for wire compatibility. Messages written by versions predating this field carry no `fetch` at all, and a plain proto3 scalar decodes that absence as `0`. With the negative-sentinel convention used by `SortExecNode`'s `int64 fetch`, `0` would mean "fetch 0 rows" and would silently turn every older plan into an empty result. `optional` gives prost an `Option<u64>` where absent decodes to `None`, which is the correct reading of an older message. A comment in the `.proto` records this. - Regenerated `prost.rs` / `pbjson.rs` via `datafusion/proto-models/regen.sh` (no hand edits). - `hash_join/exec.rs`: write `self.fetch` in the `try_to_proto` hook and restore it in `try_from_proto` via the builder's `with_fetch`, matching how the plan is normally constructed. - New regression test `roundtrip_hash_join_fetch`. The deprecated `PhysicalPlanNodeExt` shims (`try_from_hash_join_exec` / `try_into_hash_join_physical_plan`) delegate straight to these two hooks, so they pick the fix up with no separate change. Verified by reading them rather than assumed. ## Are these changes tested? Yes. `roundtrip_hash_join_fetch` in `datafusion/proto/tests/cases/roundtrip_physical_plan.rs` builds a `HashJoinExec`, applies `with_fetch(Some(7))` the way `limit_pushdown` does, round-trips it through `physical_plan_to_bytes_with_proto_converter` / `physical_plan_from_bytes_with_proto_converter`, and asserts `fetch()` is still `Some(7)`. It also covers `fetch = None`. The assertion deliberately inspects `fetch()` rather than the plan's string form. The existing `roundtrip_test` helper compares `format!("{plan:?}")`, and `HashJoinExec`'s `Debug` output does not include `fetch` — which is exactly why this went unnoticed. I confirmed this empirically: with the encode side reverted, the Debug comparison inside the helper still passes and only the `fetch()` assertion fails (`left: None, right: Some(7)`). Ran locally: - `cargo fmt --all` - `cargo test -p datafusion-proto --test proto_integration` — 215 passed, 0 failed - `cargo test -p datafusion-physical-plan` — 1640 + 9 passed, 0 failed - `cargo clippy --all-targets --all-features` on the touched packages. The changed code is clean; the only two errors reported are pre-existing on an unmodified `main` with my newer local clippy (`uninlined_format_args` in `datafusion/proto-common/src/generated/pbjson.rs` and `needless_pass_by_value` in `datafusion/proto/src/bytes/mod.rs`), in files this PR does not touch. ## Are there any user-facing changes? Yes, a bug fix: a limit pushed into a hash join now survives physical-plan serialization, so distributed executors no longer over-return rows. No API changes. The new proto field is backward and forward compatible in both directions — old readers ignore tag 12, and new readers treat its absence as "no limit". --------- Co-authored-by: Claude Opus 5 <[email protected]> --- .../physical-plan/src/joins/hash_join/exec.rs | 43 +++++++++++---- datafusion/proto-models/proto/datafusion.proto | 8 +++ datafusion/proto-models/src/generated/pbjson.rs | 21 +++++++ datafusion/proto-models/src/generated/prost.rs | 9 +++ .../proto/tests/cases/roundtrip_physical_plan.rs | 64 ++++++++++++++++++++++ 5 files changed, 133 insertions(+), 12 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 7cc2644668..f965e87df8 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1852,6 +1852,7 @@ impl ExecutionPlan for HashJoinExec { }, null_aware: self.null_aware, dynamic_filter, + fetch: self.fetch.map(|f| f as u64), }, )), ), @@ -1866,7 +1867,7 @@ impl HashJoinExec { node: &datafusion_proto_models::protobuf::PhysicalPlanNode, ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, ) -> Result<Arc<dyn ExecutionPlan>> { - use datafusion_common::internal_datafusion_err; + use datafusion_common::{internal_datafusion_err, plan_datafusion_err}; use datafusion_proto_models::protobuf; use std::any::Any; @@ -1943,17 +1944,35 @@ impl HashJoinExec { indices => Some(indices.iter().map(|i| *i as usize).collect()), }; - let mut hash_join = HashJoinExec::try_new( - left, - right, - on, - filter, - &join_type, - projection, - partition_mode, - null_equality, - hashjoin.null_aware, - )?; + // Restore the row limit that `limit_pushdown` may have pushed into the + // join. The field is presence-tracked, so a message written before it + // existed decodes to `None` (no limit) rather than to `Some(0)`. + // + // The conversion is checked, not `as usize`: `fetch` is a `u64` on the + // wire but a `usize` in the plan, and on a 32-bit target `as usize` + // truncates. A fetch of `1 << 32` would become `0` -- not merely a + // wrong limit but the worst one, silently turning the query into an + // empty result. Report the out-of-range value instead. Please do not + // "simplify" this back to `as usize`. + let fetch = hashjoin + .fetch + .map(|f| { + usize::try_from(f).map_err(|_| { + plan_datafusion_err!( + "HashJoinExec: fetch value {f} cannot be represented as usize on this target" + ) + }) + }) + .transpose()?; + + let mut hash_join = HashJoinExecBuilder::new(left, right, on, join_type) + .with_filter(filter) + .with_projection(projection) + .with_partition_mode(partition_mode) + .with_null_equality(null_equality) + .with_null_aware(hashjoin.null_aware) + .with_fetch(fetch) + .build()?; if let Some(dynamic_filter_proto) = &hashjoin.dynamic_filter { // The dynamic filter is a `DynamicFilterPhysicalExpr` over the probe diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index cbc41a7c57..5a8bede195 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1324,6 +1324,14 @@ message HashJoinExecNode { bool null_aware = 10; // Optional dynamic filter expression for pushing down to the probe side. PhysicalExprNode dynamic_filter = 11; + // Optional row limit pushed into the join by the `limit_pushdown` rule. + // + // This is presence-tracked (`optional`) on purpose: messages produced by + // versions predating this field carry no `fetch` at all, and a plain proto3 + // scalar would decode that absence as `0`, i.e. "fetch 0 rows", silently + // turning old plans into empty results. With `optional`, absent decodes to + // `None`, which is the correct reading of an older message. + optional uint64 fetch = 12; } enum StreamPartitionMode { diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index 7f9b9eddc5..9c82ceb4a2 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -9026,6 +9026,9 @@ impl serde::Serialize for HashJoinExecNode { if self.dynamic_filter.is_some() { len += 1; } + if self.fetch.is_some() { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.HashJoinExecNode", len)?; if let Some(v) = self.left.as_ref() { struct_ser.serialize_field("left", v)?; @@ -9063,6 +9066,11 @@ impl serde::Serialize for HashJoinExecNode { if let Some(v) = self.dynamic_filter.as_ref() { struct_ser.serialize_field("dynamicFilter", v)?; } + if let Some(v) = self.fetch.as_ref() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("fetch", ToString::to_string(&v).as_str())?; + } struct_ser.end() } } @@ -9088,6 +9096,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode { "nullAware", "dynamic_filter", "dynamicFilter", + "fetch", ]; #[allow(clippy::enum_variant_names)] @@ -9102,6 +9111,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode { Projection, NullAware, DynamicFilter, + Fetch, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error> @@ -9133,6 +9143,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode { "projection" => Ok(GeneratedField::Projection), "nullAware" | "null_aware" => Ok(GeneratedField::NullAware), "dynamicFilter" | "dynamic_filter" => Ok(GeneratedField::DynamicFilter), + "fetch" => Ok(GeneratedField::Fetch), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -9162,6 +9173,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode { let mut projection__ = None; let mut null_aware__ = None; let mut dynamic_filter__ = None; + let mut fetch__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::Left => { @@ -9227,6 +9239,14 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode { } dynamic_filter__ = map_.next_value()?; } + GeneratedField::Fetch => { + if fetch__.is_some() { + return Err(serde::de::Error::duplicate_field("fetch")); + } + fetch__ = + map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) + ; + } } } Ok(HashJoinExecNode { @@ -9240,6 +9260,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode { projection: projection__.unwrap_or_default(), null_aware: null_aware__.unwrap_or_default(), dynamic_filter: dynamic_filter__, + fetch: fetch__, }) } } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index f763348308..df8f4677c9 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -2029,6 +2029,15 @@ pub struct HashJoinExecNode { /// Optional dynamic filter expression for pushing down to the probe side. #[prost(message, optional, tag = "11")] pub dynamic_filter: ::core::option::Option<PhysicalExprNode>, + /// Optional row limit pushed into the join by the `limit_pushdown` rule. + /// + /// This is presence-tracked (`optional`) on purpose: messages produced by + /// versions predating this field carry no `fetch` at all, and a plain proto3 + /// scalar would decode that absence as `0`, i.e. "fetch 0 rows", silently + /// turning old plans into empty results. With `optional`, absent decodes to + /// `None`, which is the correct reading of an older message. + #[prost(uint64, optional, tag = "12")] + pub fetch: ::core::option::Option<u64>, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct SymmetricHashJoinExecNode { diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index a05f2a6ee3..afd4057d04 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -542,6 +542,70 @@ fn roundtrip_hash_join_projection_states() -> Result<()> { Ok(()) } +/// Regression: `HashJoinExecNode` had no `fetch` field, so the row limit that +/// the `limit_pushdown` physical optimizer rule pushes into the join via +/// `ExecutionPlan::with_fetch` was silently dropped by serde. Because that rule +/// also removes the enclosing `GlobalLimitExec` once the join absorbs the limit, +/// a round-tripped plan had no limit left at all and a distributed executor +/// returned more rows than the query asked for. +/// +/// Note this cannot be covered by `roundtrip_test`: that helper compares +/// `format!("{plan:?}")`, and `HashJoinExec`'s `Debug` output does not include +/// `fetch`, so the before/after strings match even when the value is lost. The +/// assertions below therefore inspect `fetch()` directly. +#[test] +fn roundtrip_hash_join_fetch() -> Result<()> { + let field_a = Field::new("col", DataType::Int64, false); + let schema_left = Arc::new(Schema::new(vec![field_a.clone()])); + let schema_right = Arc::new(Schema::new(vec![field_a])); + let on = vec![( + Arc::new(Column::new("col", schema_left.index_of("col")?)) as _, + Arc::new(Column::new("col", schema_right.index_of("col")?)) as _, + )]; + + // `usize::MAX` and `u32::MAX as usize` pin the decode-side `u64 -> usize` + // conversion: it is a checked `usize::try_from`, and a large fetch must + // survive the round trip exactly rather than being truncated or clamped. + // Both are representable on every target (on a 32-bit target `usize::MAX` + // is simply `u32::MAX`), so this stays portable. The truncating case + // itself -- a `u64` fetch above `usize::MAX` -- is only reachable on a + // 32-bit target and so is not exercised by this test on a 64-bit host. + for fetch in [None, Some(7), Some(u32::MAX as usize), Some(usize::MAX)] { + let join = HashJoinExec::try_new( + Arc::new(EmptyExec::new(Arc::clone(&schema_left))), + Arc::new(EmptyExec::new(Arc::clone(&schema_right))), + on.clone(), + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + NullEquality::NullEqualsNothing, + false, + )?; + + let plan: Arc<dyn ExecutionPlan> = match fetch { + // This is how `limit_pushdown` installs the limit. + Some(fetch) => join + .with_fetch(Some(fetch)) + .expect("HashJoinExec supports fetch"), + None => Arc::new(join), + }; + assert_eq!(plan.fetch(), fetch); + + let ctx = SessionContext::new(); + let codec = DefaultPhysicalExtensionCodec {}; + let proto_converter = DefaultPhysicalProtoConverter {}; + let deserialized = + roundtrip_test_and_return(plan, &ctx, &codec, &proto_converter)?; + + let deserialized_join = deserialized + .downcast_ref::<HashJoinExec>() + .expect("should be a HashJoinExec"); + assert_eq!(deserialized_join.fetch(), fetch); + } + Ok(()) +} + /// Same regression coverage for `NestedLoopJoinExec`, which shares the /// `repeated uint32 projection` proto field shape with `HashJoinExec`. #[test] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
