proto: round-trip SortExec's TopK dynamic filter

Add optional PhysicalExprNode dynamic_filter field to SortExecNode.
Emit from to_proto via SortExec::dynamic_filter(); on deserialize parse
it and install via SortExec::with_dynamic_filter after the usual
with_fetch / with_preserve_partitioning chain. The with_fetch step may
auto-create a TopK filter when fetch is set; with_dynamic_filter then
replaces it with the one from the sender so the id matches the
pushed-down scan's copy (shared via the id cache).

The accompanying SQL test uses an ORDER BY ... LIMIT query over a
parquet file, round-trips via DeduplicatingProtoConverter, executes,
and verifies:
  - SortExec and the scan's pushed predicate share the same expression_id
  - generation == 1 before execution (filter not yet updated)
  - generation > 1 after execution (TopK's update took effect) AND
    the scan-side wrapper observes the same generation — proving that
    updates propagate across the round-trip via shared `inner`.

Scan row-count pruning isn't asserted because the single row-group
parquet file decodes all rows before the filter is updated (a TopK
scan-pruning optimization concern, orthogonal to proto correctness).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

datafusion/proto/proto/datafusion.proto

@@ -1352,6 +1352,11 @@ message SortExecNode {
   // Maximum number of highest/lowest rows to fetch; negative means no limit
   int64 fetch = 3;
   bool preserve_partitioning = 4;
+  // TopK dynamic filter built by the sort and observed by pushed-down scans.
+  // Only present when `fetch` is set. Carried here so receiver's SortExec
+  // shares mutable state with the scan sites via the DeduplicatingProtoConverter's
+  // expr_id cache.
+  optional PhysicalExprNode dynamic_filter = 5;
 }
 
 message SortPreservingMergeExecNode {

datafusion/proto/src/generated/pbjson.rs

@@ -1680,10 +1680,28 @@ impl protobuf::PhysicalPlanNode {
             return internal_err!("SortExec requires an ordering");
         };
         let fetch = (sort.fetch >= 0).then_some(sort.fetch as _);
-        let new_sort = SortExec::new(ordering, input)
+        let input_schema = input.schema();
+        let mut new_sort = SortExec::new(ordering, input)
             .with_fetch(fetch)
             .with_preserve_partitioning(sort.preserve_partitioning);
 
+        if let Some(proto_filter) = &sort.dynamic_filter {
+            let filter_expr = proto_converter.proto_to_physical_expr(
+                proto_filter,
+                ctx,
+                &input_schema,
+                codec,
+            )?;
+            let df = Arc::downcast::<DynamicFilterPhysicalExpr>(filter_expr).map_err(
+                |_| {
+                    proto_error(
+                        "SortExec.dynamic_filter was not a DynamicFilterPhysicalExpr",
+                    )
+                },
+            )?;
+            new_sort = new_sort.with_dynamic_filter(df);
+        }
+
         Ok(Arc::new(new_sort))
     }
 
@@ -3125,6 +3143,13 @@ impl protobuf::PhysicalPlanNode {
                 })
             })
             .collect::<Result<Vec<_>>>()?;
+        let dynamic_filter = exec
+            .dynamic_filter()
+            .map(|df| {
+                let df_expr: Arc<dyn PhysicalExpr> = df as _;
+                proto_converter.physical_expr_to_proto(&df_expr, codec)
+            })
+            .transpose()?;
         Ok(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(PhysicalPlanType::Sort(Box::new(
                 protobuf::SortExecNode {
@@ -3135,6 +3160,7 @@ impl protobuf::PhysicalPlanNode {
                         _ => -1,
                     },
                     preserve_partitioning: exec.preserve_partitioning(),
+                    dynamic_filter,
                 },
             ))),
         })

datafusion/proto/src/generated/prost.rs

@@ -3178,6 +3178,18 @@ fn find_hash_join(plan: &Arc<dyn ExecutionPlan>) -> Option<&HashJoinExec> {
     None
 }
 
+fn find_sort(plan: &Arc<dyn ExecutionPlan>) -> Option<&SortExec> {
+    if let Some(s) = plan.downcast_ref::<SortExec>() {
+        return Some(s);
+    }
+    for child in plan.children() {
+        if let Some(s) = find_sort(child) {
+            return Some(s);
+        }
+    }
+    None
+}
+
 fn find_data_source(plan: &Arc<dyn ExecutionPlan>) -> Option<&DataSourceExec> {
     if let Some(dse) = plan.downcast_ref::<DataSourceExec>() {
         return Some(dse);
@@ -3190,6 +3202,96 @@ fn find_data_source(plan: &Arc<dyn ExecutionPlan>) -> Option<&DataSourceExec> {
     None
 }
 
+/// End-to-end: a SQL `ORDER BY ... LIMIT` produces a TopK dynamic filter on
+/// the `SortExec` that's pushed into the underlying `ParquetSource`. After
+/// proto round-trip we verify:
+///   1. The `SortExec`'s dynamic filter survived (wired through the builder).
+///   2. The `ParquetSource`'s pushed predicate is the same shared expression.
+///   3. Executing the plan drives TopK's `update()`, and that update is
+///      observable via the pushed predicate — proving inner-state sharing
+///      across proto round-trip for the TopK path.
+///
+/// We don't assert scan row-count pruning here: with a single row-group
+/// parquet file the DataSource has already decoded all rows by the time the
+/// TopK updates its filter, so the scan metric isn't a reliable pruning
+/// signal. The filter-update observation is the invariant this PR adds.
+#[tokio::test]
+async fn roundtrip_topk_dynamic_filter_survives_via_sql() -> Result<()> {
+    use datafusion::physical_expr::expressions::DynamicFilterPhysicalExpr;
+    use datafusion::physical_plan::collect;
+    use datafusion::prelude::SessionConfig;
+    use datafusion_datasource::file_scan_config::FileScanConfig;
+
+    let config = SessionConfig::new()
+        .set_bool("datafusion.execution.parquet.pushdown_filters", true)
+        .with_batch_size(4);
+    let ctx = SessionContext::new_with_config(config);
+    let parquet_path = concat!(
+        env!("CARGO_MANIFEST_DIR"),
+        "/../core/tests/data/tpch_nation_small.parquet"
+    );
+    ctx.register_parquet("nation", parquet_path, ParquetReadOptions::default())
+        .await?;
+
+    let sql = "SELECT n_name FROM nation ORDER BY n_nationkey ASC LIMIT 3";
+    let plan = ctx.sql(sql).await?.create_physical_plan().await?;
+
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DeduplicatingProtoConverter {};
+    let bytes = physical_plan_to_bytes_with_proto_converter(
+        Arc::clone(&plan),
+        &codec,
+        &proto_converter,
+    )?;
+    let result_plan = physical_plan_from_bytes_with_proto_converter(
+        bytes.as_ref(),
+        ctx.task_ctx().as_ref(),
+        &codec,
+        &proto_converter,
+    )?;
+
+    let sort =
+        find_sort(&result_plan).expect("deserialized plan should contain a SortExec");
+    let sort_filter = sort
+        .dynamic_filter()
+        .expect("deserialized SortExec should carry a TopK dynamic filter");
+
+    let scan = find_data_source(&result_plan)
+        .expect("deserialized plan should contain a DataSourceExec");
+    let scan_filter = scan
+        .data_source()
+        .downcast_ref::<FileScanConfig>()
+        .expect("DataSourceExec should have a FileScanConfig")
+        .file_source()
+        .filter()
+        .expect("ParquetSource should have a pushed-down dynamic filter");
+    let scan_df = scan_filter
+        .downcast_ref::<DynamicFilterPhysicalExpr>()
+        .expect("pushed filter should be a DynamicFilterPhysicalExpr");
+
+    // Identity is shared: same id on both sites.
+    assert_eq!(sort_filter.expression_id(), scan_df.expression_id());
+
+    // Before execution the filter hasn't been updated (generation == 1).
+    assert_eq!(sort_filter.snapshot_generation(), 1);
+
+    // Execute the deserialized plan.
+    let _batches = collect(Arc::clone(&result_plan), ctx.task_ctx()).await?;
+
+    // TopK updated its filter during execution; the scan-side wrapper sees
+    // the update because they share `inner`.
+    assert!(
+        sort_filter.snapshot_generation() > 1,
+        "SortExec's dynamic filter should have been updated during execution"
+    );
+    assert_eq!(
+        scan_df.snapshot_generation(),
+        sort_filter.snapshot_generation()
+    );
+
+    Ok(())
+}
+
 /// Tests that `lead` window function with offset and default value args
 /// survives a protobuf round-trip. This is a regression test for a bug
 /// where `expressions()` (used during serialization) returns only the