test: add datafusion-tests workspace crate for cross-crate integration tests

Introduce a new workspace member `datafusion-tests` at datafusion/tests
dedicated to integration tests that need to depend on both `datafusion`
and `datafusion-proto`. Placing such tests in either of those crates'
own `tests/` directory would close a dev-dependency cycle caught by the
workspace's circular-dependency check (see `dev/depcheck`).

The first two tests exercise end-to-end SQL round-trips proving the
DynamicFilterPhysicalExpr wiring added in preceding commits actually
drives scan-level pruning:

- `hash_join_dynamic_filter_prunes_via_sql`: INNER JOIN with a WHERE
  on the build side; after round-trip the probe-side `ParquetSource`
  emits fewer rows than the full table.
- `topk_dynamic_filter_prunes_files_via_sql`: ORDER BY ... LIMIT 1
  over two single-row parquet files; after round-trip the second file
  is pruned by row-group statistics because TopK's dynamic filter
  tightens after reading the first, and the scan sees exactly one row.

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

Author: adriangb

Cargo.lock

@@ -55,6 +55,7 @@ members = [
     "datafusion/sql",
     "datafusion/sqllogictest",
     "datafusion/substrait",
+    "datafusion/tests",
     "datafusion/wasmtest",
     "datafusion-cli",
     "datafusion-examples",

Cargo.toml

@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "datafusion-tests"
+description = "Integration tests that span multiple DataFusion crates (e.g. datafusion + datafusion-proto)"
+publish = false
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+repository = { workspace = true }
+license = { workspace = true }
+authors = { workspace = true }
+rust-version = { workspace = true }
+
+[lints]
+workspace = true
+
+[lib]
+# Empty lib to give Cargo a target; tests/ is where the actual code lives.
+path = "src/lib.rs"
+
+[dev-dependencies]
+arrow = { workspace = true }
+datafusion = { workspace = true, default-features = true }
+datafusion-common = { workspace = true }
+datafusion-physical-plan = { workspace = true }
+datafusion-proto = { workspace = true }
+parquet = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread"] }

datafusion/tests/Cargo.toml

@@ -0,0 +1,23 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Cross-crate integration tests for DataFusion.
+//!
+//! This crate intentionally has no library code — it exists only to host
+//! integration tests under `tests/` that need to depend on both `datafusion`
+//! and `datafusion-proto` simultaneously, which can't live in either of those
+//! crates without creating a dev-dependency cycle.

datafusion/tests/src/lib.rs

@@ -0,0 +1,250 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! End-to-end tests asserting that `datafusion-proto` round-trip preserves
+//! the identity + shared mutable state of `DynamicFilterPhysicalExpr`
+//! instances attached to operators that produce them (`HashJoinExec` and
+//! `SortExec`'s TopK). Running the deserialized plan drives updates on the
+//! shared filter, and the pushed predicate on the scan side prunes rows
+//! accordingly.
+//!
+//! These tests live in a standalone crate (`datafusion-tests`) because they
+//! need dev-dependencies on both `datafusion` and `datafusion-proto`; putting
+//! them in either of those crates' own `tests/` directory would close a
+//! dev-dependency cycle caught by the workspace's circular-dependency check.
+
+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
+/// `ParquetSource`. After proto round-trip, the `HashJoinExec`'s dynamic
+/// filter Arc and the pushed predicate still share mutable state, so
+/// build-side `update()` during execution is visible to the scan and the
+/// scan actually prunes rows.
+#[tokio::test]
+async fn hash_join_dynamic_filter_prunes_via_sql() -> datafusion_common::Result<()> {
+    let config = SessionConfig::new()
+        .set_bool("datafusion.execution.parquet.pushdown_filters", true);
+    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("build", parquet_path, ParquetReadOptions::default())
+        .await?;
+    ctx.register_parquet("probe", parquet_path, ParquetReadOptions::default())
+        .await?;
+
+    // Self-join with a selective WHERE on the build side. The build side
+    // reduces to a single row; the dynamic filter derived from it is pushed
+    // into the probe-side scan.
+    let sql = "SELECT p.n_name FROM probe p \
+               INNER JOIN build b ON p.n_nationkey = b.n_nationkey \
+               WHERE b.n_nationkey = 5";
+    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 hj = find_hash_join(&result_plan)
+        .expect("deserialized plan should contain a HashJoinExec");
+    assert!(
+        hj.dynamic_filter().is_some(),
+        "deserialized HashJoinExec should carry a dynamic filter"
+    );
+
+    let _batches = collect(Arc::clone(&result_plan), ctx.task_ctx()).await?;
+
+    // The probe side is HashJoinExec.right; walk down to its DataSourceExec.
+    let probe_scan = find_data_source(hj.children()[1])
+        .expect("probe-side should terminate in a DataSourceExec");
+    let rows = probe_scan
+        .metrics()
+        .and_then(|m| m.output_rows())
+        .expect("DataSourceExec should record output_rows");
+
+    // Full probe table is 20 rows; the dynamic filter must prune below that.
+    assert!(
+        rows < 20,
+        "probe-side scan emitted {rows} rows; dynamic filter did not prune \
+         (full table is 20 rows)"
+    );
+
+    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);
+    }
+    for child in plan.children() {
+        if let Some(hj) = find_hash_join(child) {
+            return Some(hj);
+        }
+    }
+    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);
+    }
+    for child in plan.children() {
+        if let Some(dse) = find_data_source(child) {
+            return Some(dse);
+        }
+    }
+    None
+}