Uh oh!
There was an error while loading. Please reload this page.
fix(proto): preserve HashJoinExec fetch across serialization - #24165
Merged
Conversation
`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 because the `limit_pushdown` physical optimizer rule
pushes a limit into the join via `ExecutionPlan::with_fetch` and then
drops the enclosing `GlobalLimitExec`. After a proto round-trip the plan
therefore carried no limit at all, and a distributed executor returned
more rows than the query asked for.
Add `optional uint64 fetch = 12` and wire it through both hooks. The
field is presence-tracked on purpose: messages written before it existed
carry no `fetch`, and a plain proto3 scalar would decode that absence as
`0` -- "fetch 0 rows" -- silently producing empty results. `optional`
gives `None` for absent, which is the correct reading of an older
message.
The existing `roundtrip_test` helper cannot catch this class of bug: it
compares `format!("{plan:?}")`, and `HashJoinExec`'s `Debug` output does
not include `fetch`. The new regression test asserts on `fetch()`
directly and covers both `Some(7)` and `None`.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
26 tasks
adriangb
commented
Aug 7, 2026
Comment on lines
1936
to
1946
ContributorAuthor
There was a problem hiding this comment.
Should we use a builder the whole way through instead of a try_new and then a builder later?
Comment on lines
+1967
to
+1968
| hash_join = hash_join | ||
| .builder() |
ContributorAuthor
There was a problem hiding this comment.
Let's avoid going from instance -> builder and back.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@## main #24165 +/- ##
========================================
Coverage 81.05% 81.05% ========================================
Files 1107 1107 Lines 381407 381574 +167 Branches 381407 381574 +167 ========================================
+ Hits 309139 309275 +136 - Misses 54013 54036 +23 - Partials 18255 18263 +8 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
…roto Address review feedback: the decoder called `HashJoinExec::try_new` and then round-tripped through `.builder().with_fetch(..).build()` just to apply `fetch`. `try_new` is itself a thin wrapper over `HashJoinExecBuilder`, so construct through the builder directly and set `fetch` alongside the other options. No behavior change: `try_new` delegates to the same `HashJoinExecBuilder::new(..).with_filter(..).with_projection(..) .with_partition_mode(..).with_null_equality(..).with_null_aware(..) .build()` chain, so validation, `column_indices`, `join_schema` and the computed `PlanProperties` are identical. `with_dynamic_filter_expr` is a method on `HashJoinExec` (not the builder), so it remains a post-build step. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
adriangb
commented
Aug 7, 2026
ContributorAuthor
@kosiew@kumarUjjawal would one of you be able to review this change please? |
| // the join. The field is presence-tracked, so a message written | ||
| // before it existed decodes to `None` (no limit) rather than to | ||
| // `Some(0)`. | ||
| .with_fetch(hashjoin.fetch.map(|f| f as usize)) |
Contributor
There was a problem hiding this comment.
u64 as usize will silently truncates on 32-bit targets. A fetch of 1 << 32 will becomes 0.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
kumarUjjawal
commented
Aug 7, 2026
Contributor
Thanks @adriangb Left few comments |
Review feedback: `u64 as usize` silently truncates on a 32-bit target, where `usize` is 32 bits. A fetch of `1 << 32` decodes to `0` -- not merely a wrong limit but the worst possible one, since "fetch 0 rows" turns the query into an empty result instead of erroring. Use `usize::try_from` and surface an out-of-range value as an error. Truncating and saturating both misrepresent the plan; an explicit decode failure is the honest outcome, and it is only reachable on a 32-bit target with an absurd fetch. `plan_datafusion_err!` rather than `internal_datafusion_err!`: this decode path reserves the internal-error macros for genuinely malformed nodes (an unknown `PartitionMode` discriminant, a dynamic filter that does not downcast), which really do indicate a bug. A well-formed but unrepresentable `fetch` is not a DataFusion bug -- it is a plan that cannot be expressed on this target -- and `plan_err!` is already this file's idiom for invalid plan configuration. The encode side (`self.fetch.map(|f| f as u64)`) needs no change: `usize` is at most 64 bits on every supported target, so widening to `u64` is lossless. `roundtrip_hash_join_fetch` now also covers `u32::MAX as usize` and `usize::MAX`, pinning that a large fetch round-trips exactly. Both are representable on every target, so the test stays portable. The truncation path itself is only reachable on a 32-bit target and is therefore not covered on a 64-bit CI host. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
adriangb added a commit
to pydantic/datafusion
that referenced
this pull request
Aug 7, 2026
This branch is stacked on the `fetch` fix and, after the rebase onto main, carries its own copy of the `HashJoinExec` decode path. Apply the same correction here so the defect does not survive on this branch and reappear once apache#24165 merges. Review feedback on apache#24165: `u64 as usize` silently truncates on a 32-bit target, where `usize` is 32 bits. A fetch of `1 << 32` decodes to `0` -- not merely a wrong limit but the worst possible one, since "fetch 0 rows" turns the query into an empty result instead of erroring. Use `usize::try_from` and surface an out-of-range value as an error. Truncating and saturating both misrepresent the plan; an explicit decode failure is the honest outcome, and it is only reachable on a 32-bit target with an absurd fetch. `plan_datafusion_err!` rather than `internal_datafusion_err!`: this decode path reserves the internal-error macros for genuinely malformed nodes (an unknown `PartitionMode` discriminant, a dynamic filter that does not downcast), which really do indicate a bug. A well-formed but unrepresentable `fetch` is not a DataFusion bug -- it is a plan that cannot be expressed on this target -- and `plan_err!` is already this file's idiom for invalid plan configuration. The encode side (`fetch.map(|f| f as u64)`) needs no change: `usize` is at most 64 bits on every supported target, so widening to `u64` is lossless. `roundtrip_hash_join_fetch` now also covers `u32::MAX as usize` and `usize::MAX`, pinning that a large fetch round-trips exactly. Both are representable on every target, so the test stays portable. The truncation path itself is only reachable on a 32-bit target and is therefore not covered on a 64-bit CI host. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This was referenced Aug 7, 2026
adriangb
commented
Aug 7, 2026
ContributorAuthor
Thanks @kumarUjjawal, replied! |
kumarUjjawal
approved these changes
Aug 7, 2026
kumarUjjawal
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the response. Looks good 👍
Uh oh!
There was an error while loading. Please reload this page.
Ruchirtripathi pushed a commit
to Ruchirtripathi/datafusion
that referenced
this pull request
Aug 10, 2026
) ## Which issue does this PR close? - Part of apache#24171. This addresses items (a), (b) and (c) of that issue. It does **not** attempt (d) the colocated test tier or (e) splitting up `roundtrip_physical_plan.rs`, both of which are larger and independent, so the issue stays open. Split out of apache#24167 at the reviewer's prompting: @andygrove pointed out on that PR that the "covered by existing round-trip tests" claim did not hold — there is no round-trip test for `SortPreservingMergeExec`. He was right. Rather than bundle new tests with a mechanical refactor, they are here on their own so the two can be reviewed and merged independently. ## Rationale for this change `datafusion/proto/tests/cases/roundtrip_physical_plan.rs` is the safety net for physical-plan serialization. A field that is silently dropped on the wire produces a plan that still *runs* — it just returns different rows. Checking turned up two real holes: 1. **`SortPreservingMergeExec` had no round-trip test at all.** The string `SortPreservingMerge` did not appear anywhere in the file. Nothing constructed one, so nothing on the encode or decode path for that plan was exercised. 2. **`SortExec::fetch` is serialized but was never exercised.** `roundtrip_sort` and `roundtrip_sort_preserve_partitioning` both leave `fetch` as `None`, so the `Some(..)` state had no coverage. `fetch` is what makes a `SortExec` a top-k sort; dropping it on the wire silently widens the result set. Both are cases where a serialization gap changes query results rather than causing a visible failure. ## What changes are included in this PR? Two new tests in `roundtrip_physical_plan.rs`. No production code changes. - **`roundtrip_sort_preserving_merge`** — covers everything actually on the wire for `SortPreservingMergeExecNode`: the input, the sort expressions, and `fetch` in both its `None` (encoded as `-1`) and `Some(11)` states. - **`roundtrip_sort_with_fetch`** — `SortExec` with `fetch: Some(7)`, plus `fetch: Some(3)` combined with `preserve_partitioning: true`, since both live in the same proto node. ### Which tests assert on accessors rather than the helper, and why The file's `roundtrip_test` helper compares `format!("{plan:?}")` before and after. That comparison is only as good as the plan's `Debug` impl — it is blind to any field `Debug` does not print, which is how apache#24165 (`HashJoinExec::fetch`) survived. `SortExec` and `SortPreservingMergeExec` both currently *derive* `Debug`, so the helper does in fact observe `expr`, `fetch` and `preserve_partitioning` today. I checked rather than assumed, and the deliberate-break results below confirm it — the helper is what fires first under each break. But that coverage is incidental: it would vanish the day either plan grows a hand-written `Debug`. So both new tests go through `roundtrip_test_and_return`, downcast, and assert on `fetch()`, `expr()`, `preserve_partitioning()` and the input schema directly, with the helper's string comparison still running as a backstop. ### What is deliberately *not* asserted `SortPreservingMergeExec::enable_round_robin_repartition` is **not** serialized — `SortPreservingMergeExecNode` has only `input`, `expr` and `fetch`, so decode always restores the `true` default from `SortPreservingMergeExec::new`. A round-trip equality assertion would pass whether or not that field were on the wire, so asserting on it would advertise coverage that does not exist. The test carries a comment saying so instead. The same applies to `Global/LocalLimitExec::required_ordering`, which is set by the `enforce_sorting` rule and starts as `None` on a decoded plan. If either field *should* be on the wire, that is a separate change with a wire-format bump, not something to paper over with a test that cannot tell the difference. ### Plans I checked and decided needed nothing I went through the rest of the plans touched by apache#24167 looking for state that is on the wire but exercised by no test. These already have adequate coverage and I did not add to them: | Plan | Existing coverage | |---|---| | `GlobalLimitExec` | `roundtrip_global_limit` (skip 0 / limit 25) and `roundtrip_global_skip_no_limit` (skip 10 / limit `None`) — both `skip` and `fetch` states | | `LocalLimitExec` | `roundtrip_local_limit` | | `FilterExec` | `roundtrip_filter_with_fetch` already asserts `default_selectivity`, `batch_size` and `fetch` on the accessors; `roundtrip_filter_projection_states` covers the projection | | `ProjectionExec` | `roundtrip_projection_source`, `roundtrip_empty_projection` | | `RepartitionExec` | `roundtrip_repartition_preserve_order` (round-robin + `preserve_order`), `roundtrip_range_partitioning`, plus hash-partitioning cases | | `UnionExec` / `InterleaveExec` | `roundtrip_union`, `roundtrip_interleave` — nothing on the wire beyond the children | | `CoalesceBatchesExec` | `roundtrip_coalesce_batches_with_fetch` covers `target_batch_size` and `fetch` in both states | | `CoalescePartitionsExec` | `roundtrip_coalesce_partitions_with_fetch`, both `fetch` states | | `CooperativeExec` | `roundtrip_cooperative` — only the input is on the wire | | `BufferExec` | `roundtrip_buffer` asserts `capacity()` on the accessor | | `EmptyExec` / `PlaceholderRowExec` | `roundtrip_empty_with_partitions`, `roundtrip_placeholder_row_with_partitions` | | `ExplainExec` | `roundtrip_explain` asserts schema, stringified plans and `verbose` on the accessors | | `ScalarSubqueryExec` | `roundtrip_scalar_subquery_exec` and the executing variant | Per the issue, this is not a push for 100% field coverage. These two were the cases where the gap was real and the test was cheap; past them it got contrived fast. ## Are these changes tested? This PR *is* tests. To confirm they are not vacuous, I broke the encode side on purpose and checked each one fails: | Deliberate break | Result | |---|---| | `SortPreservingMergeExec` encode: hardcode `fetch: -1` | `roundtrip_sort_preserving_merge` **FAILS** | | `SortPreservingMergeExec` encode: `.take(1)` on the sort expressions | `roundtrip_sort_preserving_merge` **FAILS** | | `SortExec` encode: hardcode `fetch: -1` | `roundtrip_sort_with_fetch` **FAILS** | Under all three breaks the pre-existing `roundtrip_sort` and `roundtrip_sort_preserve_partitioning` kept passing — a direct demonstration that the gap was real. All breaks reverted; the diff here is test-only. Checks run: - `cargo fmt --all` - `cargo clippy --all-targets --workspace --features avro,integration-tests,extended_tests -- -D warnings` — clean - `cargo test -p datafusion-proto --test proto_integration` — 216 passed, 0 failed (214 before this PR) - `cargo test -p datafusion-physical-plan` — 1641 passed, 0 failed ### Why this is based on `main` rather than on apache#24167 These tests pass on unmodified `main` — verified, not assumed. They describe serialization behaviour that already exists, and none of them depends on apache#24167's changes. That gives a useful property: merged first, they pin the current wire behaviour independently, which makes apache#24167's "wire format unchanged" claim something CI verifies rather than something the PR description asserts. ## Are there any user-facing changes? No. Test-only; no production code touched. Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
adriangb pushed a commit
to pydantic/datafusion
that referenced
this pull request
Aug 10, 2026
Adds the second, colocated test tier proposed in apache#24171 (d), next to the serde hooks the plans already own. `datafusion-physical-plan` gains `proto_test_util`, the plan-level sibling of `datafusion_physical_expr::proto_test_util`: stub `ExecutionPlanEncode` / `ExecutionPlanDecode` implementations a test supplies itself, so a plan's `try_to_proto` / `try_from_proto` can be driven without depending on `datafusion-proto` (which sits above this crate). The stubs count their calls and can fail on the Nth one, so the `?` arms are covered too. On top of that, colocated `proto_tests` modules for the fields where being wrong is expensive and `Debug`-comparing round-trip tests are blind: * `SortExec` / `SortPreservingMergeExec` — `fetch` presence semantics (absent -> `None`, never `Some(0)`), `preserve_partitioning`, the `asc`/`descending` inversion, the TopK dynamic filter, and the reject paths. `SortPreservingMergeExec` also pins that `enable_round_robin_repartition` is deliberately not on the wire. * `HashJoinExec` — the three `projection` states proto3 cannot express directly (the `[u32::MAX]` sentinel), `fetch` presence (the field dropped in apache#24165), the checked `u64` -> `usize` conversion, and `PartitionMode`. * `NestedLoopJoinExec` — the same projection sentinel, which is written out a second time in that file and so is tested a second time here. * `joins::proto` — exhaustive by-name round trips for `JoinType`, `JoinSide` and `NullEquality`, plus a guard that the two numberings really do differ, so nobody "simplifies" the matches into a cast. Per apache#24171 (c) the new tests were verified against a deliberately broken encode side rather than assumed to bite. This supplements the central round-trip tests, it does not replace them: those still prove the real `PhysicalExtensionCodec` works, that dispatch reaches the hook, and that bytes survive bytes.
kosiew pushed a commit
to kosiew/datafusion
that referenced
this pull request
Aug 12, 2026
…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 apache#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 <noreply@anthropic.com>
kosiew pushed a commit
to kosiew/datafusion
that referenced
this pull request
Aug 12, 2026
) ## Which issue does this PR close? - Part of apache#24171. This addresses items (a), (b) and (c) of that issue. It does **not** attempt (d) the colocated test tier or (e) splitting up `roundtrip_physical_plan.rs`, both of which are larger and independent, so the issue stays open. Split out of apache#24167 at the reviewer's prompting: @andygrove pointed out on that PR that the "covered by existing round-trip tests" claim did not hold — there is no round-trip test for `SortPreservingMergeExec`. He was right. Rather than bundle new tests with a mechanical refactor, they are here on their own so the two can be reviewed and merged independently. ## Rationale for this change `datafusion/proto/tests/cases/roundtrip_physical_plan.rs` is the safety net for physical-plan serialization. A field that is silently dropped on the wire produces a plan that still *runs* — it just returns different rows. Checking turned up two real holes: 1. **`SortPreservingMergeExec` had no round-trip test at all.** The string `SortPreservingMerge` did not appear anywhere in the file. Nothing constructed one, so nothing on the encode or decode path for that plan was exercised. 2. **`SortExec::fetch` is serialized but was never exercised.** `roundtrip_sort` and `roundtrip_sort_preserve_partitioning` both leave `fetch` as `None`, so the `Some(..)` state had no coverage. `fetch` is what makes a `SortExec` a top-k sort; dropping it on the wire silently widens the result set. Both are cases where a serialization gap changes query results rather than causing a visible failure. ## What changes are included in this PR? Two new tests in `roundtrip_physical_plan.rs`. No production code changes. - **`roundtrip_sort_preserving_merge`** — covers everything actually on the wire for `SortPreservingMergeExecNode`: the input, the sort expressions, and `fetch` in both its `None` (encoded as `-1`) and `Some(11)` states. - **`roundtrip_sort_with_fetch`** — `SortExec` with `fetch: Some(7)`, plus `fetch: Some(3)` combined with `preserve_partitioning: true`, since both live in the same proto node. ### Which tests assert on accessors rather than the helper, and why The file's `roundtrip_test` helper compares `format!("{plan:?}")` before and after. That comparison is only as good as the plan's `Debug` impl — it is blind to any field `Debug` does not print, which is how apache#24165 (`HashJoinExec::fetch`) survived. `SortExec` and `SortPreservingMergeExec` both currently *derive* `Debug`, so the helper does in fact observe `expr`, `fetch` and `preserve_partitioning` today. I checked rather than assumed, and the deliberate-break results below confirm it — the helper is what fires first under each break. But that coverage is incidental: it would vanish the day either plan grows a hand-written `Debug`. So both new tests go through `roundtrip_test_and_return`, downcast, and assert on `fetch()`, `expr()`, `preserve_partitioning()` and the input schema directly, with the helper's string comparison still running as a backstop. ### What is deliberately *not* asserted `SortPreservingMergeExec::enable_round_robin_repartition` is **not** serialized — `SortPreservingMergeExecNode` has only `input`, `expr` and `fetch`, so decode always restores the `true` default from `SortPreservingMergeExec::new`. A round-trip equality assertion would pass whether or not that field were on the wire, so asserting on it would advertise coverage that does not exist. The test carries a comment saying so instead. The same applies to `Global/LocalLimitExec::required_ordering`, which is set by the `enforce_sorting` rule and starts as `None` on a decoded plan. If either field *should* be on the wire, that is a separate change with a wire-format bump, not something to paper over with a test that cannot tell the difference. ### Plans I checked and decided needed nothing I went through the rest of the plans touched by apache#24167 looking for state that is on the wire but exercised by no test. These already have adequate coverage and I did not add to them: | Plan | Existing coverage | |---|---| | `GlobalLimitExec` | `roundtrip_global_limit` (skip 0 / limit 25) and `roundtrip_global_skip_no_limit` (skip 10 / limit `None`) — both `skip` and `fetch` states | | `LocalLimitExec` | `roundtrip_local_limit` | | `FilterExec` | `roundtrip_filter_with_fetch` already asserts `default_selectivity`, `batch_size` and `fetch` on the accessors; `roundtrip_filter_projection_states` covers the projection | | `ProjectionExec` | `roundtrip_projection_source`, `roundtrip_empty_projection` | | `RepartitionExec` | `roundtrip_repartition_preserve_order` (round-robin + `preserve_order`), `roundtrip_range_partitioning`, plus hash-partitioning cases | | `UnionExec` / `InterleaveExec` | `roundtrip_union`, `roundtrip_interleave` — nothing on the wire beyond the children | | `CoalesceBatchesExec` | `roundtrip_coalesce_batches_with_fetch` covers `target_batch_size` and `fetch` in both states | | `CoalescePartitionsExec` | `roundtrip_coalesce_partitions_with_fetch`, both `fetch` states | | `CooperativeExec` | `roundtrip_cooperative` — only the input is on the wire | | `BufferExec` | `roundtrip_buffer` asserts `capacity()` on the accessor | | `EmptyExec` / `PlaceholderRowExec` | `roundtrip_empty_with_partitions`, `roundtrip_placeholder_row_with_partitions` | | `ExplainExec` | `roundtrip_explain` asserts schema, stringified plans and `verbose` on the accessors | | `ScalarSubqueryExec` | `roundtrip_scalar_subquery_exec` and the executing variant | Per the issue, this is not a push for 100% field coverage. These two were the cases where the gap was real and the test was cheap; past them it got contrived fast. ## Are these changes tested? This PR *is* tests. To confirm they are not vacuous, I broke the encode side on purpose and checked each one fails: | Deliberate break | Result | |---|---| | `SortPreservingMergeExec` encode: hardcode `fetch: -1` | `roundtrip_sort_preserving_merge` **FAILS** | | `SortPreservingMergeExec` encode: `.take(1)` on the sort expressions | `roundtrip_sort_preserving_merge` **FAILS** | | `SortExec` encode: hardcode `fetch: -1` | `roundtrip_sort_with_fetch` **FAILS** | Under all three breaks the pre-existing `roundtrip_sort` and `roundtrip_sort_preserve_partitioning` kept passing — a direct demonstration that the gap was real. All breaks reverted; the diff here is test-only. Checks run: - `cargo fmt --all` - `cargo clippy --all-targets --workspace --features avro,integration-tests,extended_tests -- -D warnings` — clean - `cargo test -p datafusion-proto --test proto_integration` — 216 passed, 0 failed (214 before this PR) - `cargo test -p datafusion-physical-plan` — 1641 passed, 0 failed ### Why this is based on `main` rather than on apache#24167 These tests pass on unmodified `main` — verified, not assumed. They describe serialization behaviour that already exists, and none of them depends on apache#24167's changes. That gives a useful property: merged first, they pin the current wire behaviour independently, which makes apache#24167's "wire format unchanged" claim something CI verifies rather than something the PR description asserts. ## Are there any user-facing changes? No. Test-only; no production code touched. Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
fields). This is a bug fix, not part of the migration checklist.
Rationale for this change
HashJoinExec.fetchwas silently dropped by protobuf serialization.protobuf::HashJoinExecNodehad nofetchfield, soHashJoinExec'stry_to_protonever wrote it andtry_from_protonever restored it: a planwith
fetch = Some(n)round-tripped tofetch = None.This is user-visible. The
limit_pushdownphysical optimizer rule pushes alimit into the join via
ExecutionPlan::with_fetch, then marks the globalstate satisfied and drops the enclosing
GlobalLimitExec. So after a protoround-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: addoptional uint64 fetch = 12toHashJoinExecNode.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
fetchat all, and a plain proto3 scalar decodes that absenceas
0. With the negative-sentinel convention used bySortExecNode'sint64 fetch,0would mean "fetch 0 rows" and would silently turn everyolder plan into an empty result.
optionalgives prost anOption<u64>where absent decodes to
None, which is the correct reading of an oldermessage. A comment in the
.protorecords this.Regenerated
prost.rs/pbjson.rsviadatafusion/proto-models/regen.sh(no hand edits).
hash_join/exec.rs: writeself.fetchin thetry_to_protohook andrestore it in
try_from_protovia the builder'swith_fetch, matching howthe plan is normally constructed.
New regression test
roundtrip_hash_join_fetch.The deprecated
PhysicalPlanNodeExtshims (try_from_hash_join_exec/try_into_hash_join_physical_plan) delegate straight to these two hooks, sothey pick the fix up with no separate change. Verified by reading them rather
than assumed.
Are these changes tested?
Yes.
roundtrip_hash_join_fetchindatafusion/proto/tests/cases/roundtrip_physical_plan.rsbuilds aHashJoinExec, applieswith_fetch(Some(7))the waylimit_pushdowndoes,round-trips it through
physical_plan_to_bytes_with_proto_converter/physical_plan_from_bytes_with_proto_converter, and assertsfetch()isstill
Some(7). It also coversfetch = None.The assertion deliberately inspects
fetch()rather than the plan's stringform. The existing
roundtrip_testhelper comparesformat!("{plan:?}"), andHashJoinExec'sDebugoutput does not includefetch— which is exactly whythis 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 --allcargo test -p datafusion-proto --test proto_integration— 215 passed, 0 failedcargo test -p datafusion-physical-plan— 1640 + 9 passed, 0 failedcargo clippy --all-targets --all-featureson the touched packages. Thechanged code is clean; the only two errors reported are pre-existing on an
unmodified
mainwith my newer local clippy (uninlined_format_argsindatafusion/proto-common/src/generated/pbjson.rsandneedless_pass_by_valueindatafusion/proto/src/bytes/mod.rs), in filesthis 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".