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 fc846dd368 fix(proto): preserve HashJoinExec fetch across
serialization (#24165)
fc846dd368 is described below
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]
