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).

Add an end-to-end SQL test in
core/tests/physical_optimizer/filter_pushdown.rs that writes two
single-row parquet files (`a.parquet` with key=1, `b.parquet` with
key=2), runs `ORDER BY n_nationkey ASC LIMIT 1` with
`target_partitions=1`, round-trips via DeduplicatingProtoConverter,
executes, and asserts the scan emitted exactly 1 row — b.parquet was
pruned by row-group statistics after TopK saw a's row and tightened
the shared dynamic filter.

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

Author: adriangb

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -3383,18 +3383,23 @@ fn test_filter_pushdown_through_sort_with_projection() {
 mod proto_roundtrip {
     use std::sync::Arc;
 
+    use arrow::array::{Int64Array, StringArray};
+    use arrow::datatypes::{DataType, Field, Schema};
+    use arrow::record_batch::RecordBatch;
     use datafusion::datasource::source::DataSourceExec;
     use datafusion::physical_plan::collect;
     use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext};
     use datafusion_physical_plan::ExecutionPlan;
     use datafusion_physical_plan::joins::HashJoinExec;
+    use datafusion_physical_plan::sorts::sort::SortExec;
     use datafusion_proto::bytes::{
         physical_plan_from_bytes_with_proto_converter,
         physical_plan_to_bytes_with_proto_converter,
     };
     use datafusion_proto::physical_plan::{
         DeduplicatingProtoConverter, DefaultPhysicalExtensionCodec,
     };
+    use parquet::arrow::ArrowWriter;
 
     /// End-to-end: a SQL hash join with a selective WHERE on the build side
     /// produces a dynamic filter that's pushed into the probe-side
@@ -3465,6 +3470,104 @@ mod proto_roundtrip {
         Ok(())
     }
 
+    /// End-to-end: an `ORDER BY ... LIMIT 1` over two single-row parquet
+    /// files (`a.parquet` with key=1, `b.parquet` with key=2).
+    ///
+    ///  1. `SortExec` with `TopK(fetch=1)` creates a dynamic filter pushed
+    ///     into the underlying `ParquetSource`.
+    ///  2. After proto round-trip, the `SortExec`'s filter and the scan's
+    ///     pushed predicate share the same inner state (cached by
+    ///     `expression_id`).
+    ///  3. With `target_partitions=1`, both files are read sequentially. After
+    ///     `a.parquet` is read, TopK's single best row has key=1 and the
+    ///     filter tightens.
+    ///  4. When the scan opens `b.parquet` (min=max=2), row-group statistics
+    ///     prune it — the scan never yields b's row.
+    ///
+    /// Observable proof: `DataSourceExec::metrics().output_rows() == 1`.
+    /// Without the round-trip wiring, TopK would update a disconnected
+    /// filter and the scan would emit both rows.
+    #[tokio::test]
+    async fn topk_dynamic_filter_prunes_files_via_sql() -> datafusion_common::Result<()> {
+        let tmp = tempfile::TempDir::new()?;
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("n_nationkey", DataType::Int64, false),
+            Field::new("n_name", DataType::Utf8, false),
+        ]));
+        let write_file =
+            |name: &str, key: i64, value: &str| -> datafusion_common::Result<()> {
+                let path = tmp.path().join(name);
+                let batch = RecordBatch::try_new(
+                    Arc::clone(&schema),
+                    vec![
+                        Arc::new(Int64Array::from(vec![key])),
+                        Arc::new(StringArray::from(vec![value])),
+                    ],
+                )?;
+                let file = std::fs::File::create(&path)?;
+                let mut writer = ArrowWriter::try_new(file, batch.schema(), None)?;
+                writer.write(&batch)?;
+                writer.close()?;
+                Ok(())
+            };
+        write_file("a.parquet", 1, "foo")?;
+        write_file("b.parquet", 2, "bar")?;
+
+        let config = SessionConfig::new()
+            .set_bool("datafusion.execution.parquet.pushdown_filters", true)
+            .with_target_partitions(1);
+        let ctx = SessionContext::new_with_config(config);
+        ctx.register_parquet(
+            "t",
+            tmp.path().to_str().unwrap(),
+            ParquetReadOptions::default(),
+        )
+        .await?;
+
+        let sql = "SELECT n_name FROM t ORDER BY n_nationkey ASC LIMIT 1";
+        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");
+        assert!(
+            sort.dynamic_filter().is_some(),
+            "deserialized SortExec should carry a TopK dynamic filter"
+        );
+
+        let batches = collect(Arc::clone(&result_plan), ctx.task_ctx()).await?;
+        let total_out: usize = batches.iter().map(|b| b.num_rows()).sum();
+        assert_eq!(total_out, 1, "ORDER BY LIMIT 1 should emit one row");
+
+        let scan = find_data_source(&result_plan)
+            .expect("deserialized plan should contain a DataSourceExec");
+        let rows = scan
+            .metrics()
+            .and_then(|m| m.output_rows())
+            .expect("DataSourceExec should record output_rows");
+        assert_eq!(
+            rows, 1,
+            "scan emitted {rows} rows; expected exactly 1 (only a.parquet), \
+             because the TopK dynamic filter should prune b.parquet once TopK \
+             has seen a's row"
+        );
+
+        Ok(())
+    }
+
     fn find_hash_join(plan: &Arc<dyn ExecutionPlan>) -> Option<&HashJoinExec> {
         if let Some(hj) = plan.downcast_ref::<HashJoinExec>() {
             return Some(hj);
@@ -3477,6 +3580,18 @@ mod proto_roundtrip {
         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);

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,
                 },
             ))),
         })