Made the Arrow-Flight transport optional behind a flight feature.

With `flight` off there is no `tonic` / `arrow-flight` and the in-memory transport is the
default, so distributed plans still run. The integration suite runs over the in-memory transport
in both build configurations: `start_localhost_context` builds an `InMemoryWorkerTransport`
cluster instead of a gRPC one, and the gRPC harness moved to `start_localhost_flight_context`. A
`unit-test-flight-transport` job sets `DATAFUSION_DISTRIBUTED_TEST_TRANSPORT=flight` to keep full
Flight coverage.

For the no-flight suite to run, not just build, the `tpch` / `tpcds` / `clickbench` /
`stateful_data_cleanup` dataset tests move into the benchmarks crate: as a dev-dependency of the
library they re-enabled `flight` on every test build through feature unification. With that gone,
`cargo test --no-default-features --features integration --lib --tests` is genuinely Flight-free,
and a `unit-test-no-flight` job runs it. The only tests still gated on `flight` are the ones that
need a real wire: the `URLEmitter` routing tests, which assert per-URL worker identity, and the
network-boundary connection metrics (`bytes_transferred`, latency). The rest, including the
worker-hook and metrics tests, register `InMemoryWorkerTransport` directly and run either way.

Author: mdashti

.github/workflows/ci.yml

@@ -32,6 +32,8 @@ jobs:
           components: clippy
       - run: cargo clippy --all-targets --all-features -- -D warnings
 
+  # The generic suite runs over the in-memory transport (what `start_localhost_context` builds by
+  # default), so it covers the distributed paths without a gRPC server.
   unit-test:
     runs-on: ubuntu-latest
     steps:
@@ -41,12 +43,50 @@ jobs:
       - uses: ./.github/actions/setup
       - run: cargo test --features integration
 
+  # The same suite over the Arrow-Flight gRPC transport. `start_localhost_context` switches to it
+  # when `DATAFUSION_DISTRIBUTED_TEST_TRANSPORT=flight`, so both transports get full coverage.
+  unit-test-flight-transport:
+    runs-on: ubuntu-latest
+    env:
+      DATAFUSION_DISTRIBUTED_TEST_TRANSPORT: flight
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          lfs: true
+      - uses: ./.github/actions/setup
+      - run: cargo test --features integration
+
+  # Builds the lib and the suite with `flight` off: no `tonic` / `arrow-flight`, in-memory transport
+  # as the default. The dataset suites moved to the benchmarks crate, so the suite no longer drags
+  # the benchmarks dev-dependency in (which re-enabled `flight` through feature unification).
+  # Examples are left out: they demo the Flight cluster, so `--all-features` clippy covers them.
+  no-flight:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/setup
+        with:
+          components: clippy
+      - run: cargo build --no-default-features --features integration
+      - run: cargo clippy --no-default-features --features integration --lib --tests -- -D warnings
+
+  # The integration suite over the in-memory transport with `flight` off, so the no-flight runtime
+  # path is run, not just built. Flight-specific tests gate themselves out.
+  unit-test-no-flight:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          lfs: true
+      - uses: ./.github/actions/setup
+      - run: cargo test --no-default-features --features integration --lib --tests
+
   tpch-test:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/setup
-      - run: cargo test --features tpch --test 'tpch_*'
+      - run: cargo test -p datafusion-distributed-benchmarks --features tpch --test 'tpch_*'
 
   tpcds-correctness-test:
     runs-on: ubuntu-latest
@@ -59,9 +99,9 @@ jobs:
       - uses: ./.github/actions/setup
       - uses: actions/cache@v4
         with:
-          path: testdata/tpcds/main.zip
+          path: benchmarks/testdata/tpcds/main.zip
           key: "main.zip"
-      - run: cargo test --features tpcds --test tpcds_correctness_test shard${{ matrix.shard }}
+      - run: cargo test -p datafusion-distributed-benchmarks --features tpcds --test tpcds_correctness_test shard${{ matrix.shard }}
 
   tpcds-plans-test:
     runs-on: ubuntu-latest
@@ -70,9 +110,9 @@ jobs:
       - uses: ./.github/actions/setup
       - uses: actions/cache@v4
         with:
-          path: testdata/tpcds/main.zip
+          path: benchmarks/testdata/tpcds/main.zip
           key: "main.zip"
-      - run: cargo test --features tpcds --test tpcds_plans_test
+      - run: cargo test -p datafusion-distributed-benchmarks --features tpcds --test tpcds_plans_test
 
   clickbench-test:
     runs-on: ubuntu-latest
@@ -81,9 +121,9 @@ jobs:
       - uses: ./.github/actions/setup
       - uses: actions/cache@v4
         with:
-          path: testdata/clickbench/
+          path: benchmarks/testdata/clickbench/
           key: "data"
-      - run: cargo test --features clickbench --test 'clickbench_*'
+      - run: cargo test -p datafusion-distributed-benchmarks --features clickbench --test 'clickbench_*'
 
   format-check:
     runs-on: ubuntu-latest

Cargo.lock

@@ -23,13 +23,13 @@ datafusion = { workspace = true, features = [
   "datetime_expressions",
 ] }
 datafusion-proto = { workspace = true }
-arrow-flight = "58"
+arrow-flight = { version = "58", optional = true }
 arrow-select = "58"
 arrow-ipc = { version = "58", features = ["zstd"] }
 async-trait = "0.1.89"
 tokio = { version = "1.48", features = ["full"] }
-tonic = { version = "0.14.1", features = ["transport"] }
-tower = "0.5.2"
+tonic = { version = "0.14.1", features = ["transport"], optional = true }
+tower = { version = "0.5.2", optional = true }
 http = "1.3.1"
 itertools = "0.14.0"
 futures = "0.3.31"
@@ -49,7 +49,7 @@ crossbeam-queue = "0.3"
 sysinfo = { version = "0.30", optional = true }
 sketches-ddsketch = { version = "0.3", features = ["use_serde"] }
 bincode = "1"
-tonic-prost = "0.14.2"
+tonic-prost = { version = "0.14.2", optional = true }
 
 # integration_tests deps
 insta = { version = "1.46.0", features = ["filters"], optional = true }
@@ -58,19 +58,22 @@ arrow = { version = "58", optional = true, features = ["test_utils"] }
 hyper-util = { version = "0.1.16", optional = true }
 
 [features]
+default = ["flight"]
+# Arrow Flight gRPC transport. When off, the crate builds with no `tonic` / `arrow-flight`
+# and the in-memory transport becomes the default: distributed plans run inside the current
+# process. Multi-process execution then needs an embedder-registered transport.
+flight = ["dep:arrow-flight", "dep:tonic", "dep:tonic-prost", "dep:tower"]
 avro = ["datafusion/avro"]
+# Independent from `flight` so the integration suite runs against both the gRPC transport
+# (`--features integration`) and the in-memory one (`--no-default-features --features
+# integration`). The Flight-specific tests gate themselves on `flight`.
 integration = ["insta", "parquet", "arrow", "hyper-util"]
 
 system-metrics = ["sysinfo"]
 
-tpch = ["integration"]
-tpcds = ["integration"]
-clickbench = ["integration"]
-slow-tests = []
 sysinfo = ["dep:sysinfo"]
 
 [dev-dependencies]
-datafusion-distributed-benchmarks = { path = "benchmarks" }
 structopt = "0.3"
 insta = { version = "1.46.0", features = ["filters"] }
 parquet = "58"

Cargo.toml

@@ -33,9 +33,20 @@ aws-sdk-ec2 = "1"
 openssl = { version = "0.10", features = ["vendored"] } # Keep this. Necessary for the remote benchmarks worker.
 mimalloc = "0.1"
 
+[features]
+# Gates for the dataset test suites under `tests/`. They live here instead of the library because
+# they depend on it: as a dev-dependency of the library they re-enable `flight` on every test build
+# through feature unification, which makes the no-flight config untestable.
+tpch = []
+tpcds = []
+clickbench = []
+slow-tests = []
+
 [dev-dependencies]
 criterion = "0.5"
 sysinfo = "0.30"
+pretty_assertions = "1.4"
+test-case = "3.3.1"
 
 [build-dependencies]
 built = { version = "0.8", features = ["git2", "chrono"] }

benchmarks/Cargo.toml

@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "clickbench", test))]
+#[cfg(all(feature = "clickbench", test))]
 mod tests {
     use datafusion::arrow::array::RecordBatch;
     use datafusion::common::plan_err;

tests/clickbench_correctness_test.rs …rks/tests/clickbench_correctness_test.rstests/clickbench_correctness_test.rs renamed to benchmarks/tests/clickbench_correctness_test.rs tests/clickbench_correctness_test.rs renamed to benchmarks/tests/clickbench_correctness_test.rs

@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "clickbench", test))]
+#[cfg(all(feature = "clickbench", test))]
 mod tests {
     use datafusion::error::Result;
     use datafusion_distributed::test_utils::in_memory_channel_resolver::start_in_memory_context;

tests/clickbench_plans_test.rs benchmarks/tests/clickbench_plans_test.rstests/clickbench_plans_test.rs renamed to benchmarks/tests/clickbench_plans_test.rs

@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpch", test))]
+#[cfg(all(feature = "tpch", test))]
 mod tests {
     use datafusion::common::instant::Instant;
     use datafusion::error::Result;

tests/stateful_data_cleanup.rs benchmarks/tests/stateful_data_cleanup.rstests/stateful_data_cleanup.rs renamed to benchmarks/tests/stateful_data_cleanup.rs

@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpcds", test))]
+#[cfg(all(feature = "tpcds", test))]
 mod tests {
     use datafusion::arrow::array::RecordBatch;
     use datafusion::common::plan_err;

tests/tpcds_correctness_test.rs …nchmarks/tests/tpcds_correctness_test.rstests/tpcds_correctness_test.rs renamed to benchmarks/tests/tpcds_correctness_test.rs tests/tpcds_correctness_test.rs renamed to benchmarks/tests/tpcds_correctness_test.rs

@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpcds", test))]
+#[cfg(all(feature = "tpcds", test))]
 mod tests {
     use datafusion::error::Result;
     use datafusion_distributed::test_utils::in_memory_channel_resolver::start_in_memory_context;

tests/tpcds_plans_test.rs benchmarks/tests/tpcds_plans_test.rstests/tpcds_plans_test.rs renamed to benchmarks/tests/tpcds_plans_test.rs

@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpch", test))]
+#[cfg(all(feature = "tpch", test))]
 mod tests {
     use datafusion::physical_plan::execute_stream;
     use datafusion::prelude::SessionContext;