(datafusion) branch main updated: test(proto): add missing physical plan round-trip coverage (#24172)

[gi...@apache.org]

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 0f68e23bde test(proto): add missing physical plan round-trip coverage (#24172)
0f68e23bde is described below

commit 0f68e23bde46c19d2bc944132cc37bf9055c6f13
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Mon Aug 10 07:40:13 2026 -0400

    test(proto): add missing physical plan round-trip coverage (#24172)
    
    ## Which issue does this PR close?
    
    - Part of #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 #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 #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 #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 #24167
    
    These tests pass on unmodified `main` — verified, not assumed. They
    describe serialization behaviour that already exists, and none of them
    depends on #24167's changes.
    
    That gives a useful property: merged first, they pin the current wire
    behaviour independently, which makes #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 <no...@anthropic.com>
---
 .../proto/tests/cases/roundtrip_physical_plan.rs   | 156 +++++++++++++++++++++
 1 file changed, 156 insertions(+)

diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 19249cf97b..3f03e58656 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -88,6 +88,7 @@ use datafusion::physical_plan::scalar_subquery::{
     ScalarSubqueryExec, ScalarSubqueryLink,
 };
 use datafusion::physical_plan::sorts::sort::SortExec;
+use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
 use datafusion::physical_plan::union::{InterleaveExec, UnionExec};
 use datafusion::physical_plan::unnest::{ListUnnest, UnnestExec};
 use datafusion::physical_plan::windows::{
@@ -1199,6 +1200,161 @@ fn roundtrip_sort_preserve_partitioning() -> Result<()> {
     ))
 }
 
+/// `SortExec::fetch` turns a sort into a top-k sort. Losing it during serde
+/// would silently widen the result set, so exercise the `Some(..)` state
+/// explicitly (`roundtrip_sort` only covers `None`).
+///
+/// `SortExec` currently derives `Debug`, so `roundtrip_test`'s
+/// `format!("{plan:?}")` comparison does observe `fetch`. The assertions below
+/// go through the accessor instead so that this coverage does not silently
+/// disappear if `SortExec` ever grows a hand-written `Debug` impl.
+#[test]
+fn roundtrip_sort_with_fetch() -> Result<()> {
+    let field_a = Field::new("a", DataType::Boolean, false);
+    let field_b = Field::new("b", DataType::Int64, false);
+    let schema = Arc::new(Schema::new(vec![field_a, field_b]));
+    let sort_exprs: LexOrdering = [
+        PhysicalSortExpr {
+            expr: col("a", &schema)?,
+            options: SortOptions {
+                descending: true,
+                nulls_first: false,
+            },
+        },
+        PhysicalSortExpr {
+            expr: col("b", &schema)?,
+            options: SortOptions {
+                descending: false,
+                nulls_first: true,
+            },
+        },
+    ]
+    .into();
+
+    let ctx = SessionContext::new();
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DefaultPhysicalProtoConverter {};
+
+    let roundtripped = roundtrip_test_and_return(
+        Arc::new(
+            SortExec::new(
+                sort_exprs.clone(),
+                Arc::new(EmptyExec::new(Arc::clone(&schema))),
+            )
+            .with_fetch(Some(7)),
+        ),
+        &ctx,
+        &codec,
+        &proto_converter,
+    )?;
+    let roundtripped = roundtripped
+        .downcast_ref::<SortExec>()
+        .expect("should decode back into a SortExec");
+    assert_eq!(roundtripped.fetch(), Some(7));
+    assert_eq!(roundtripped.expr(), &sort_exprs);
+
+    // `fetch` combined with `preserve_partitioning`, since both share the same
+    // proto node.
+    let roundtripped = roundtrip_test_and_return(
+        Arc::new(
+            SortExec::new(sort_exprs.clone(), Arc::new(EmptyExec::new(schema)))
+                .with_fetch(Some(3))
+                .with_preserve_partitioning(true),
+        ),
+        &ctx,
+        &codec,
+        &proto_converter,
+    )?;
+    let roundtripped = roundtripped
+        .downcast_ref::<SortExec>()
+        .expect("should decode back into a SortExec");
+    assert_eq!(roundtripped.fetch(), Some(3));
+    assert!(roundtripped.preserve_partitioning());
+    Ok(())
+}
+
+/// Round trip a [`SortPreservingMergeExec`], which had no dedicated round trip
+/// test at all.
+///
+/// Covers everything that is actually on the wire for this plan: the input, the
+/// sort expressions and `fetch` in both its `None` and `Some(..)` states.
+///
+/// Note that `SortPreservingMergeExec::enable_round_robin_repartition` is
+/// deliberately *not* asserted on here: it has no field in
+/// `SortPreservingMergeExecNode`, so it is not serialized and decoding always
+/// restores the `true` default from `SortPreservingMergeExec::new`. Asserting
+/// round trip equality on it would give a false sense of coverage.
+///
+/// `SortPreservingMergeExec` derives `Debug`, so `roundtrip_test`'s
+/// `format!("{plan:?}")` comparison does observe `expr` and `fetch`. The
+/// assertions below use the accessors so the coverage survives a future
+/// hand-written `Debug` impl.
+#[test]
+fn roundtrip_sort_preserving_merge() -> Result<()> {
+    let field_a = Field::new("a", DataType::Boolean, false);
+    let field_b = Field::new("b", DataType::Int64, false);
+    let schema = Arc::new(Schema::new(vec![field_a, field_b]));
+    let sort_exprs: LexOrdering = [
+        PhysicalSortExpr {
+            expr: col("a", &schema)?,
+            options: SortOptions {
+                descending: true,
+                nulls_first: false,
+            },
+        },
+        PhysicalSortExpr {
+            expr: col("b", &schema)?,
+            options: SortOptions {
+                descending: false,
+                nulls_first: true,
+            },
+        },
+    ]
+    .into();
+
+    let ctx = SessionContext::new();
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DefaultPhysicalProtoConverter {};
+
+    // No fetch: `fetch` is encoded as -1 and must decode back to `None`.
+    let roundtripped = roundtrip_test_and_return(
+        Arc::new(SortPreservingMergeExec::new(
+            sort_exprs.clone(),
+            Arc::new(EmptyExec::new(Arc::clone(&schema))),
+        )),
+        &ctx,
+        &codec,
+        &proto_converter,
+    )?;
+    let roundtripped = roundtripped
+        .downcast_ref::<SortPreservingMergeExec>()
+        .expect("should decode back into a SortPreservingMergeExec");
+    assert_eq!(roundtripped.fetch(), None);
+    assert_eq!(roundtripped.expr(), &sort_exprs);
+    assert_eq!(roundtripped.input().schema(), schema);
+
+    // With a fetch: dropping it would turn a bounded merge into an unbounded
+    // one and change the query result.
+    let roundtripped = roundtrip_test_and_return(
+        Arc::new(
+            SortPreservingMergeExec::new(
+                sort_exprs.clone(),
+                Arc::new(EmptyExec::new(Arc::clone(&schema))),
+            )
+            .with_fetch(Some(11)),
+        ),
+        &ctx,
+        &codec,
+        &proto_converter,
+    )?;
+    let roundtripped = roundtripped
+        .downcast_ref::<SortPreservingMergeExec>()
+        .expect("should decode back into a SortPreservingMergeExec");
+    assert_eq!(roundtripped.fetch(), Some(11));
+    assert_eq!(roundtripped.expr(), &sort_exprs);
+    Ok(())
+}
+
 #[test]
 fn roundtrip_coalesce_batches_with_fetch() -> Result<()> {
     let field_a = Field::new("a", DataType::Boolean, false);


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org