adriangb opened a new pull request, #24165:
URL: https://github.com/apache/datafusion/pull/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".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
