diff --git a/.github/workflows/tpcds.yml b/.github/workflows/tpcds.yml index fa2e282045..9b803b1eb2 100644 --- a/.github/workflows/tpcds.yml +++ b/.github/workflows/tpcds.yml @@ -61,14 +61,14 @@ jobs: # sitting for hours with no useful signal. timeout-minutes: 45 strategy: - # One red leg must not cancel the other; the two task-packing modes are - # independent signals. + # One red leg must not cancel the others; the legs are independent signals. fail-fast: false matrix: include: - label: "1 partition per task" task_args: "" slug: "mpt1" + partitions: "16" # 4 rather than a larger cap because the scheduler clamps the slice to # the executor's free vcores (`budget.vcores.min(cap)`), and the # executor below runs `--concurrent-tasks 4`. A higher cap would be @@ -76,6 +76,18 @@ jobs: - label: "4 partitions per task" task_args: "-c ballista.scheduler.max_partitions_per_task=4" slug: "mpt4" + partitions: "16" + # Low-partition legs: target_partitions=1 exercises the degenerate + # single-task path; =2 exercises low-partition task packing + # (the #2186 bug class). + - label: "target_partitions=1" + task_args: "" + slug: "tp1" + partitions: "1" + - label: "target_partitions=2, 4 partitions per task" + task_args: "-c ballista.scheduler.max_partitions_per_task=4" + slug: "tp2-mpt4" + partitions: "2" steps: - name: Install dependencies run: | @@ -180,7 +192,7 @@ jobs: # This matrix leg runs the suite under the default (static) planner at # one task-packing setting. The tpcds binary internally loops all # non-skipped queries and exits non-zero on any failure, so a single - # invocation covers the whole suite. The other leg runs in parallel. + # invocation covers the whole suite. The other legs run in parallel. # # Coverage for the adaptive planner (AQE on) is being added # separately; it currently fails on pre-existing bugs. @@ -188,7 +200,7 @@ jobs: ./target/tpch-ci/tpcds \ --host 127.0.0.1 --port 50050 \ --path "$DATA_DIR" \ - --partitions 16 \ + --partitions ${{ matrix.partitions }} \ --verify \ -c datafusion.optimizer.prefer_hash_join=false \ ${{ matrix.task_args }} diff --git a/.github/workflows/tpch.yml b/.github/workflows/tpch.yml index 04a68d44ec..7d6b6119fa 100644 --- a/.github/workflows/tpch.yml +++ b/.github/workflows/tpch.yml @@ -15,7 +15,7 @@ # specific language governing permissions and limitations # under the License. -name: TPC-H SF10 +name: TPC-H permissions: contents: read @@ -52,23 +52,61 @@ on: jobs: tpch-sf10: - name: TPC-H SF10 (${{ matrix.label }}) + name: TPC-H (${{ matrix.label }}) runs-on: ubuntu-latest container: image: amd64/rust + # Cap the job well under the 6-hour default so a hung query fails fast + # and frees the runner rather than sitting for hours with no useful + # signal. + timeout-minutes: 60 strategy: fail-fast: false matrix: include: - - label: "AQE off" + - label: "SF10, AQE off" planner_args: "" slug: "aqe-off" - - label: "AQE on" + scale_factor: "10" + partitions: "16" + empty_rotation: false + - label: "SF10, AQE on" planner_args: "-c ballista.planner.adaptive.enabled=true" slug: "aqe-on" - - label: "AQE on, multi-partition tasks" + scale_factor: "10" + partitions: "16" + empty_rotation: false + - label: "SF10, AQE on, multi-partition tasks" planner_args: "-c ballista.planner.adaptive.enabled=true -c ballista.scheduler.max_partitions_per_task=0" slug: "aqe-on-mpt" + scale_factor: "10" + partitions: "16" + empty_rotation: false + # Low-partition legs run at SF1 to keep CI runtime bounded; low + # partition counts are what triggered the union-restriction and + # single-task-path bug class (#2186). + - label: "SF1, AQE on, 1 partition" + planner_args: "-c ballista.planner.adaptive.enabled=true" + slug: "aqe-on-sf1-p1" + scale_factor: "1" + partitions: "1" + empty_rotation: false + - label: "SF1, AQE on, 2 partitions" + planner_args: "-c ballista.planner.adaptive.enabled=true" + slug: "aqe-on-sf1-p2" + scale_factor: "1" + partitions: "2" + empty_rotation: false + # Empty-stage leg: a rotating set of tables is registered zero-row, + # forcing empty intermediate stages through PropagateEmptyExecRule + # (the #2185/#2194 bug class). Rotation is keyed on day-of-week in + # the run step below. + - label: "SF1, AQE on, empty tables" + planner_args: "-c ballista.planner.adaptive.enabled=true" + slug: "aqe-on-sf1-empty" + scale_factor: "1" + partitions: "16" + empty_rotation: true steps: - name: Install dependencies run: | @@ -88,7 +126,7 @@ jobs: with: # Share the build cache across all matrix legs — the compiled # binaries are identical; only the per-suite CLI args differ. - shared-key: tpch-sf10 + shared-key: tpch - name: Build Ballista binaries run: | @@ -102,16 +140,19 @@ jobs: with: tool: tpchgen-cli@2.0.2 - - name: Generate TPC-H SF10 data + - name: Generate TPC-H data run: | mkdir -p "$RUNNER_TEMP/tpch-data" tpchgen-cli \ - --scale-factor 10 \ + --scale-factor ${{ matrix.scale_factor }} \ --parts 16 \ --format=parquet \ --output-dir "$RUNNER_TEMP/tpch-data" - name: Run TPC-H queries against Ballista cluster + # Primary timeout, on the step rather than the job, so that a hang + # still runs the log-upload step below. + timeout-minutes: 45 env: DATA_DIR: ${{ runner.temp }}/tpch-data WORK_DIR: ${{ runner.temp }}/work @@ -168,26 +209,47 @@ jobs: done nc -z 127.0.0.1 50051 || { echo "executor did not start"; exit 1; } + # Empty-tables rotation: keyed on day-of-week so different + # query/empty-table combinations get coverage over time while any + # single run stays cheap. Deterministic within a day. + EMPTY_ARGS="" + if [ "${{ matrix.empty_rotation }}" = "true" ]; then + case $(( $(date +%u) % 3 )) in + 0) EMPTY_TABLES="lineitem" ;; + 1) EMPTY_TABLES="orders" ;; + 2) EMPTY_TABLES="supplier,nation" ;; + esac + echo "Empty tables for this run: $EMPTY_TABLES" + EMPTY_ARGS="--empty-tables $EMPTY_TABLES" + fi + # This matrix leg runs one planner configuration against the whole - # SF10 dataset. The other legs run in parallel jobs. + # generated dataset. The other legs run in parallel jobs. # q16 omitted: still unsupported (matches benchmarks/run.sh). for q in 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 17 18 19 20 21 22; do echo "::group::[${{ matrix.label }}] Query $q" + # shellcheck disable=SC2086 # EMPTY_ARGS is intentionally unquoted: + # it expands to zero words or a flag plus value, never a single + # word containing spaces. ./target/tpch-ci/tpch benchmark ballista \ --host 127.0.0.1 --port 50050 \ --query "$q" \ --path "$DATA_DIR" \ --format parquet \ - --partitions 16 \ + --partitions ${{ matrix.partitions }} \ --iterations 1 \ --verify \ + $EMPTY_ARGS \ -c datafusion.optimizer.prefer_hash_join=false \ ${{ matrix.planner_args }} echo "::endgroup::" done - name: Upload cluster logs on failure - if: failure() + # `!success()` rather than `failure()` so a timed-out or cancelled + # run still surfaces its scheduler/executor logs — the hang case is + # exactly when they are most worth having. + if: ${{ !success() }} uses: actions/upload-artifact@v7 with: name: tpch-sf10-cluster-logs-${{ matrix.slug }} diff --git a/benchmarks/README.md b/benchmarks/README.md index 93716965b4..5eb180271d 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -400,4 +400,12 @@ scheduler + executor on every push/PR touching `ballista/**` or `EmptyExec invalid partition` assertion (issue #2047); re-enable an AQE-on run once that is fixed. +### Forcing empty intermediate stages + +`--empty-tables lineitem,orders` registers the named tables as zero-row +Parquet tables (schema preserved) in both the Ballista session and the +`--verify` oracle. Every query touching an emptied table then produces empty +intermediate stages, exercising the scheduler's empty-stage plan rewrites. +Requires `--format parquet` (TPC-H) and is incompatible with `--expected`. + [1]: http://www.tpc.org/tpch/ diff --git a/benchmarks/src/bin/tpcds.rs b/benchmarks/src/bin/tpcds.rs index a93b746977..5b37c940ae 100644 --- a/benchmarks/src/bin/tpcds.rs +++ b/benchmarks/src/bin/tpcds.rs @@ -19,13 +19,14 @@ use ballista::extension::SessionConfigExt; use ballista::prelude::SessionContextExt; use ballista_benchmarks::{ answer_statement_index, compare_results, execute_query_capturing_answer, - register_parquet_tables, + parse_empty_tables, prepare_empty_tables, register_parquet_tables, }; use ballista_core::object_store::{ session_config_with_s3_support, session_state_with_s3_support, }; use datafusion::error::{DataFusionError, Result}; use datafusion::prelude::{SessionConfig, SessionContext}; +use std::collections::HashMap; use std::fs; use std::time::Instant; use structopt::StructOpt; @@ -132,6 +133,12 @@ struct Opt { /// Verify each Ballista result against single-process DataFusion. #[structopt(long = "verify")] verify: bool, + + /// Comma-separated table names to register as zero-row tables (schema + /// preserved) in both the Ballista and oracle contexts, forcing empty + /// intermediate stages through the distributed planner. + #[structopt(long = "empty-tables")] + empty_tables: Option, } /// Split a query file into statements, dropping full-line `--` comments and @@ -222,6 +229,7 @@ async fn run_one_query( opt: &Opt, address: &str, oracle_ctx: Option<&SessionContext>, + overrides: &HashMap, query: usize, ) -> Result<()> { let sqls = get_query_sql(query) @@ -247,7 +255,7 @@ async fn run_one_query( let ctx = SessionContext::remote_with_state(address, state) .await .map_err(|e| DataFusionError::Execution(format!("connect: {e}")))?; - register_parquet_tables(&ctx, TABLES, opt.path.as_str(), opt.debug) + register_parquet_tables(&ctx, TABLES, opt.path.as_str(), overrides, opt.debug) .await .map_err(|e| DataFusionError::Execution(format!("register-tables: {e}")))?; @@ -309,6 +317,18 @@ async fn main() -> Result<()> { let opt = Opt::from_args(); let address = format!("df://{}:{}", opt.host, opt.port); + // Zero-row mirrors are built once and shared by every query's Ballista + // session and by the oracle, so both sides see identical empty tables. + let empty_overrides = match &opt.empty_tables { + Some(spec) => { + let names = parse_empty_tables(spec, TABLES)?; + let dir = std::env::temp_dir() + .join(format!("tpcds-empty-tables-{}", std::process::id())); + prepare_empty_tables(&opt.path, &names, &dir).await? + } + None => std::collections::HashMap::new(), + }; + // Oracle context (single-process DataFusion), built once when verifying. // Not per-query, so a failure here is genuinely fatal to the whole run. let oracle_ctx = if opt.verify { @@ -316,7 +336,14 @@ async fn main() -> Result<()> { .with_target_partitions(opt.partitions) .with_batch_size(opt.batch_size); let ctx = SessionContext::new_with_config(cfg); - register_parquet_tables(&ctx, TABLES, opt.path.as_str(), opt.debug).await?; + register_parquet_tables( + &ctx, + TABLES, + opt.path.as_str(), + &empty_overrides, + opt.debug, + ) + .await?; Some(ctx) } else { None @@ -325,7 +352,10 @@ async fn main() -> Result<()> { let mut failures: Vec<(usize, String)> = vec![]; for query in selected_queries(opt.query, SKIP) { - if let Err(e) = run_one_query(&opt, &address, oracle_ctx.as_ref(), query).await { + if let Err(e) = + run_one_query(&opt, &address, oracle_ctx.as_ref(), &empty_overrides, query) + .await + { eprintln!("Query {query} FAILED: {e}"); failures.push((query, e.to_string())); } diff --git a/benchmarks/src/bin/tpch.rs b/benchmarks/src/bin/tpch.rs index 2f20804d7b..29fbf886a7 100644 --- a/benchmarks/src/bin/tpch.rs +++ b/benchmarks/src/bin/tpch.rs @@ -21,6 +21,7 @@ use ballista::extension::SessionConfigExt; use ballista::prelude::SessionContextExt; use ballista_benchmarks::{ answer_statement_index, compare_results, execute_query_capturing_answer, find_path, + parse_empty_tables, prepare_empty_tables, }; use ballista_core::object_store::{ session_config_with_s3_support, session_state_with_s3_support, @@ -59,6 +60,7 @@ use rand::prelude::*; use serde::{Deserialize, Serialize}; use std::ops::Div; use std::{ + collections::HashMap, fs::{self, File}, io::Write, iter::Iterator, @@ -142,6 +144,13 @@ struct BallistaBenchmarkOpt { /// same data. #[structopt(long = "verify")] verify: bool, + + /// Comma-separated table names to register as zero-row tables (schema + /// preserved) in both the Ballista and oracle contexts, forcing empty + /// intermediate stages through the distributed planner. Requires + /// `--format parquet`; incompatible with `--expected`. + #[structopt(long = "empty-tables")] + empty_tables: Option, } #[derive(Debug, StructOpt, Clone)] @@ -366,6 +375,29 @@ fn select_queries(query: Option, skip: &[usize]) -> Result> { Ok(selected) } +/// Parse and validate `--empty-tables`. The mirrors are Parquet files and the +/// schema is inferred from Parquet data, so the flag requires +/// `--format parquet`; canned expected results assume full data, so it +/// conflicts with `--expected`. Returns an empty list when the flag is unset. +fn validate_empty_tables(opt: &BallistaBenchmarkOpt) -> Result> { + let Some(spec) = &opt.empty_tables else { + return Ok(vec![]); + }; + if opt.file_format != "parquet" { + return Err(DataFusionError::Plan( + "--empty-tables requires --format parquet".to_string(), + )); + } + if opt.expected_results.is_some() { + return Err(DataFusionError::Plan( + "--empty-tables cannot be combined with --expected \ + (canned answers assume full data)" + .to_string(), + )); + } + parse_empty_tables(spec, TABLES) +} + #[allow(clippy::await_holding_lock)] async fn benchmark_datafusion(opt: DataFusionBenchmarkOpt) -> Result> { println!("Running benchmarks with the following options: {opt:?}"); @@ -406,6 +438,7 @@ async fn benchmark_datafusion(opt: DataFusionBenchmarkOpt) -> Result Result<()> { // Determine which queries to run let query_numbers = select_queries(opt.query, &opt.skip)?; + // Zero-row mirrors are built once and shared by every query's Ballista + // session and by the oracle, so both sides see identical empty tables. + let empty_names = validate_empty_tables(&opt)?; + let empty_overrides = if empty_names.is_empty() { + HashMap::new() + } else { + let dir = std::env::temp_dir() + .join(format!("tpch-empty-tables-{}", std::process::id())); + prepare_empty_tables(&opt.path, &empty_names, &dir).await? + }; + let oracle_ctx = if opt.verify { let cfg = SessionConfig::new() .with_target_partitions(opt.partitions) @@ -559,6 +603,7 @@ async fn benchmark_ballista(opt: BallistaBenchmarkOpt) -> Result<()> { opt.path.as_str(), opt.file_format.as_str(), opt.partitions, + &empty_overrides, ) .await?; Some(ctx) @@ -581,6 +626,7 @@ async fn benchmark_ballista(opt: BallistaBenchmarkOpt) -> Result<()> { &address, &opt, oracle_ctx.as_ref(), + &empty_overrides, query, &sqls, &mut query_run, @@ -624,6 +670,7 @@ async fn run_ballista_query( address: &str, opt: &BallistaBenchmarkOpt, oracle_ctx: Option<&SessionContext>, + overrides: &HashMap, query: usize, sqls: &[String], query_run: &mut QueryRun, @@ -650,7 +697,14 @@ async fn run_ballista_query( let state = session_state_with_s3_support(config)?; let ctx = SessionContext::remote_with_state(address, state).await?; - register_tables(opt.path.as_str(), opt.file_format.as_str(), &ctx, opt.debug).await?; + register_tables( + opt.path.as_str(), + opt.file_format.as_str(), + &ctx, + overrides, + opt.debug, + ) + .await?; let answer_idx = answer_statement_index(sqls); let mut batches = vec![]; @@ -879,7 +933,7 @@ async fn loadtest_ballista(opt: BallistaLoadtestOpt) -> Result<()> { let sql_path = opt.sql_path.to_str().unwrap().to_string(); for ctx in &clients { - register_tables(path, file_format, ctx, opt.debug).await?; + register_tables(path, file_format, ctx, &HashMap::new(), opt.debug).await?; } let request_per_thread = request_amount.div(concurrency); @@ -964,8 +1018,15 @@ async fn register_datafusion_tables( path: &str, file_format: &str, partitions: usize, + overrides: &HashMap, ) -> Result<()> { for table in TABLES { + if let Some(p) = overrides.get(*table) { + ctx.register_parquet(*table, p, ParquetReadOptions::default()) + .await + .map_err(|e| DataFusionError::Plan(format!("{e:?}")))?; + continue; + } let table_provider = { let mut session_state = ctx.state(); get_table(&mut session_state, path, table, file_format, partitions).await? @@ -979,9 +1040,19 @@ async fn register_tables( path: &str, file_format: &str, ctx: &SessionContext, + overrides: &HashMap, debug: bool, ) -> Result<()> { for &table in TABLES { + if let Some(p) = overrides.get(table) { + if debug { + println!("Registering table '{table}' as EMPTY from {p}"); + } + ctx.register_parquet(table, p, ParquetReadOptions::default()) + .await + .map_err(|e| DataFusionError::Plan(format!("{e:?}")))?; + continue; + } match file_format { // dbgen creates .tbl ('|' delimited) files without header "tbl" => { @@ -1731,6 +1802,57 @@ mod tests { assert!(select_queries(None, &(1..=22).collect::>()).is_err()); } + #[test] + fn empty_tables_requires_parquet_format() { + let opt = BallistaBenchmarkOpt::from_iter([ + "tpch", + "-p", + "/data", + "--format", + "csv", + "--empty-tables", + "lineitem", + ]); + let err = validate_empty_tables(&opt).unwrap_err().to_string(); + assert!(err.contains("requires --format parquet"), "{err}"); + } + + #[test] + fn empty_tables_conflicts_with_expected_results() { + let opt = BallistaBenchmarkOpt::from_iter([ + "tpch", + "-p", + "/data", + "--format", + "parquet", + "--empty-tables", + "lineitem", + "--expected", + "/answers", + ]); + let err = validate_empty_tables(&opt).unwrap_err().to_string(); + assert!(err.contains("--expected"), "{err}"); + } + + #[test] + fn empty_tables_parses_valid_names() { + let opt = BallistaBenchmarkOpt::from_iter([ + "tpch", + "-p", + "/data", + "--format", + "parquet", + "--empty-tables", + "supplier,nation", + ]); + assert_eq!( + validate_empty_tables(&opt).unwrap(), + vec!["supplier".to_string(), "nation".to_string()] + ); + let none = BallistaBenchmarkOpt::from_iter(["tpch", "-p", "/data"]); + assert!(validate_empty_tables(&none).unwrap().is_empty()); + } + fn run_with(queries: Vec) -> BenchmarkRun { BenchmarkRun { benchmark_version: "test".to_string(), diff --git a/benchmarks/src/lib.rs b/benchmarks/src/lib.rs index 2f05647dee..b5a60dbd03 100644 --- a/benchmarks/src/lib.rs +++ b/benchmarks/src/lib.rs @@ -26,7 +26,9 @@ use datafusion::arrow::record_batch::RecordBatch; use datafusion::arrow::util::display::array_value_to_string; use datafusion::error::{DataFusionError, Result}; use datafusion::prelude::{ParquetReadOptions, SessionContext}; +use std::collections::HashMap; use std::path::Path; +use std::sync::Arc; /// Maximum relative-or-absolute difference tolerated between floating-point /// cells. Distributed (Ballista) and single-process (DataFusion) execution @@ -241,20 +243,95 @@ pub async fn execute_query_capturing_answer( Ok(answer) } +/// Parse a `--empty-tables` argument (comma-separated table names) against a +/// benchmark's table list. Returns the validated names, deduplicated in first- +/// seen order. Unknown names are an error rather than being ignored, since a +/// typo would otherwise silently produce a full-data run. +pub fn parse_empty_tables(spec: &str, valid: &[&str]) -> Result> { + let mut out: Vec = vec![]; + for name in spec.split(',') { + let name = name.trim(); + if name.is_empty() { + continue; + } + if !valid.contains(&name) { + return Err(DataFusionError::Plan(format!( + "--empty-tables: unknown table '{name}'" + ))); + } + if !out.iter().any(|n| n == name) { + out.push(name.to_string()); + } + } + if out.is_empty() { + return Err(DataFusionError::Plan( + "--empty-tables: no table names given".to_string(), + )); + } + Ok(out) +} + +/// Write a zero-row Parquet mirror for each named table, inferring the schema +/// from the real Parquet data under `data_path`. Returns +/// `table name -> mirror path` overrides for table registration, so the +/// Ballista and oracle contexts can register the same provably-empty tables +/// and `--verify` remains a valid comparison. +pub async fn prepare_empty_tables( + data_path: &str, + tables: &[String], + out_dir: &Path, +) -> Result> { + use datafusion::parquet::arrow::ArrowWriter; + + std::fs::create_dir_all(out_dir)?; + let schema_ctx = SessionContext::new(); + let mut overrides = HashMap::new(); + for table in tables { + let real_path = find_path(data_path, table, "parquet")?; + let df = schema_ctx + .read_parquet(&real_path, ParquetReadOptions::default()) + .await?; + let schema = Arc::new(df.schema().as_arrow().clone()); + let dest = out_dir.join(format!("{table}.parquet")); + let file = std::fs::File::create(&dest)?; + let writer = ArrowWriter::try_new(file, schema, None) + .map_err(|e| DataFusionError::Execution(format!("{e}")))?; + writer + .close() + .map_err(|e| DataFusionError::Execution(format!("{e}")))?; + overrides.insert(table.clone(), dest.to_string_lossy().into_owned()); + } + Ok(overrides) +} + /// Register each named table as a Parquet source under `path`, inferring the /// schema from the files. Works for both a single `.parquet` file and a -/// partitioned `
/` directory (see `find_path`). +/// partitioned `
/` directory (see `find_path`). Tables present in +/// `overrides` are registered from the override path instead (used by +/// `--empty-tables` to substitute zero-row mirrors). pub async fn register_parquet_tables( ctx: &SessionContext, tables: &[&str], path: &str, + overrides: &HashMap, debug: bool, ) -> Result<()> { for &table in tables { - let table_path = find_path(path, table, "parquet")?; - if debug { - println!("Registering table '{table}' from Parquet at {table_path}"); - } + let table_path = match overrides.get(table) { + Some(p) => { + if debug { + println!("Registering table '{table}' as EMPTY from {p}"); + } + p.clone() + } + None => { + let p = find_path(path, table, "parquet")?; + if debug { + println!("Registering table '{table}' from Parquet at {p}"); + } + p + } + }; ctx.register_parquet(table, &table_path, ParquetReadOptions::default()) .await .map_err(|e| DataFusionError::Plan(format!("{e:?}")))?; @@ -365,4 +442,60 @@ mod tests { .unwrap(); assert!(compare_results(&[a], &[b]).is_ok()); } + + #[test] + fn parse_empty_tables_validates_and_dedupes() { + let valid = &["lineitem", "orders", "nation"]; + assert_eq!( + parse_empty_tables("lineitem, orders,lineitem", valid).unwrap(), + vec!["lineitem".to_string(), "orders".to_string()] + ); + let err = parse_empty_tables("linetem", valid) + .unwrap_err() + .to_string(); + assert!(err.contains("unknown table 'linetem'"), "{err}"); + assert!(parse_empty_tables("", valid).is_err()); + assert!(parse_empty_tables(" , ", valid).is_err()); + } + + #[tokio::test] + async fn prepare_empty_tables_writes_zero_row_mirror() { + use datafusion::parquet::arrow::ArrowWriter; + use datafusion::prelude::SessionContext; + + let tmp = tempfile::tempdir().unwrap(); + let data_dir = tmp.path().join("data"); + std::fs::create_dir_all(&data_dir).unwrap(); + + // One-row "t1" table as the real data. + let batch = i64_batch(vec![42]); + let file = std::fs::File::create(data_dir.join("t1.parquet")).unwrap(); + let mut writer = ArrowWriter::try_new(file, batch.schema(), None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + let out_dir = tmp.path().join("empty"); + let overrides = prepare_empty_tables( + data_dir.to_str().unwrap(), + &["t1".to_string()], + &out_dir, + ) + .await + .unwrap(); + + // The mirror must be readable, zero-row, and schema-identical. + let ctx = SessionContext::new(); + let df = ctx + .read_parquet(overrides["t1"].clone(), ParquetReadOptions::default()) + .await + .unwrap(); + let schema = df.schema().as_arrow().clone(); + let batches = df.collect().await.unwrap(); + let rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(rows, 0); + assert_eq!( + comparable_schema(&schema), + comparable_schema(&batch.schema()) + ); + } }