@@ -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);