skade-flight/tests/tpch_flight_sql.rs
//! **TPC-H, all 22, through the Flight SQL door — and it is a GATE.**
//!
//! The run this replaces was a report: failures were recorded, not panicked on,
//! and the only assertion was that the tables had loaded. It measured, on SF=1,
//! `q1 A/F count_order = 1478493`, `q6 revenue = 123141078.2283`, 21/22 executed
//! (Q15's `CREATE VIEW` refused), every row count matching the published
//! cardinalities — and would have gone green with all of that wrong.
//!
//! Here every number is asserted. The queries are `bench/tpch_shared.rs`'s (ONE
//! text, shared by inclusion), the data is `tpchgen`'s, and the expected values are
//! the TPC-H Tools `dbgen` answer sets that `tpchgen` bundles for SF=1
//! (`tpchgen::q_and_a::answers_sf1`, "derived from the TPC-H Tools distribution").
//! The static table below is checked AGAINST those bundled answers at test time, so
//! it cannot drift from its source.
//!
//! Two doors:
//! * [`tpch_22_execute_over_flight_sql`] — always on, SF from `TPCH_SF` (default
//! 0.01): every query must plan, stream and return rows; no value gate (only SF=1
//! has published answers).
//! * [`tpch_sf1_answers_are_canonical_over_flight_sql`] — `#[ignore]`, generates
//! SF=1 (8.66 M rows, ~1 GB under `$TMPDIR`) and gates cardinality, Q1, Q6.
//!
//! The tempdir honours `TMPDIR`; `/tmp` on oden is RAM (LAW 6), so run the SF=1
//! gate with `TMPDIR=~/scratch/cargo/skade-b5/tpch-tmp`.
#[path = "../../bench/tpch_shared.rs"]
mod tpch;
use std::sync::Arc;
use std::time::{Duration, Instant};
use arrow_array::RecordBatch;
use futures::TryStreamExt;
use iceberg::Catalog;
use skade_flight::arrow_flight::sql::client::FlightSqlServiceClient;
use skade_flight::{Auth, Engine};
use tonic::transport::Channel;
/// TPC-H "first N rows" per query (spec §2.4, the queries that return only the
/// top of their result). Everything else returns its whole result.
fn first_n(n: i32) -> Option<usize> {
match n {
2 => Some(100),
3 => Some(10),
10 => Some(20),
18 => Some(100),
21 => Some(100),
_ => None,
}
}
/// Canonical row count of every SF=1 answer set — the TPC-H Tools `dbgen`
/// answers as bundled by `tpchgen::q_and_a::answers_sf1` (data lines of each
/// `Q<n>_ANSWER`, header excluded). Q2/Q3/Q10/Q21 are the "first N" cut; Q18's
/// full result (57) is under its N (100).
const SF1_ROWS: [(i32, usize); 22] = [
(1, 4),
(2, 100),
(3, 10),
(4, 5),
(5, 5),
(6, 1),
(7, 4),
(8, 2),
(9, 175),
(10, 20),
(11, 1048),
(12, 2),
(13, 42),
(14, 1),
(15, 1),
(16, 18314),
(17, 1),
(18, 57),
(19, 1),
(20, 186),
(21, 100),
(22, 7),
];
/// Q1, the `A`/`F` row's `count_order` — the published SF=1 answer.
const Q1_AF_COUNT_ORDER: i64 = 1_478_493;
/// Q6 `revenue` — the published SF=1 answer (the TBL carries two decimals,
/// `123141078.23`; the exact sum over the generated data is this).
const Q6_REVENUE: f64 = 123_141_078.2283;
/// Data lines of a bundled TBL answer (header dropped).
fn bundled_rows(n: i32) -> usize {
tpchgen::q_and_a::answers_sf1::answer(n)
.expect("tpchgen bundles answers 1..=22 for SF=1")
.lines()
.filter(|l| !l.trim().is_empty())
.count()
- 1
}
struct Live {
_dir: tempfile::TempDir,
client: FlightSqlServiceClient<Channel>,
_server: tokio::task::JoinHandle<()>,
rows_loaded: u64,
load: Duration,
}
/// Generate + ingest the 8 tables into a skade warehouse's DEFAULT namespace
/// (bare names resolve over the door), then serve it over Flight SQL on a
/// loopback port.
async fn bring_up(sf: f64) -> Live {
let dir = tempfile::tempdir().unwrap();
let wh = skade::Warehouse::open(dir.path()).await.unwrap();
let catalog: Arc<dyn Catalog> = wh.catalog();
let parts = if sf >= 0.5 { 8 } else { 1 };
let t0 = Instant::now();
let (ctx, rows_loaded) = tpch::build_ctx_in_ns(catalog, skade::DEFAULT_NAMESPACE, sf, parts)
.await
.expect("TPC-H ingest into the warehouse");
let load = t0.elapsed();
drop(ctx);
let engine = Arc::new(Engine::with_options(wh, Duration::from_millis(50), 8).await.unwrap());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let auth = Arc::new(Auth::open());
let server = tokio::spawn(async move {
skade_flight::serve_on(engine, auth, listener, Default::default(), futures::future::pending())
.await
.unwrap();
});
let channel = Channel::from_shared(format!("http://{addr}")).unwrap().connect().await.unwrap();
Live { _dir: dir, client: FlightSqlServiceClient::new(channel), _server: server, rows_loaded, load }
}
async fn run(client: &mut FlightSqlServiceClient<Channel>, sql: &str) -> Result<Vec<RecordBatch>, String> {
let info = client.execute(sql.to_string(), None).await.map_err(|e| format!("plan: {e}"))?;
let ticket = info.endpoint[0].ticket.clone().ok_or("no ticket on endpoint 0")?;
client
.do_get(ticket)
.await
.map_err(|e| format!("do_get: {e}"))?
.try_collect()
.await
.map_err(|e| format!("stream: {e}"))
}
fn rows(b: &[RecordBatch]) -> usize {
b.iter().map(|x| x.num_rows()).sum()
}
/// The value of column `col` on the first row whose first `key.len()` columns
/// equal `key`, rendered through arrow-cast to text (works for Decimal/Float/Int).
fn cell(batches: &[RecordBatch], key: &[&str], col: &str) -> String {
for b in batches {
let text: Vec<arrow_array::StringArray> = (0..b.num_columns())
.map(|i| {
let a = arrow_cast::cast(b.column(i), &arrow_schema::DataType::Utf8).unwrap();
a.as_any().downcast_ref::<arrow_array::StringArray>().unwrap().clone()
})
.collect();
let ci = b.schema().index_of(col).unwrap_or_else(|_| panic!("no column {col}"));
for r in 0..b.num_rows() {
if key.iter().enumerate().all(|(i, k)| text[i].value(r).trim() == *k) {
return text[ci].value(r).to_string();
}
}
}
panic!("no row keyed {key:?}")
}
/// Run all 22 (`query_sql`, the engine-accepted texts) and return
/// `(n, rows, wall)` per query; panics on the first query that does not execute
/// — a door that cannot run a query is not measured, it is broken.
async fn run_22(client: &mut FlightSqlServiceClient<Channel>) -> Vec<(i32, Vec<RecordBatch>, Duration)> {
let mut out = Vec::with_capacity(22);
for n in 1..=22 {
let t = Instant::now();
let b = run(client, &tpch::query_sql(n)).await.unwrap_or_else(|e| panic!("Q{n} over Flight SQL: {e}"));
let wall = t.elapsed();
eprintln!(" Q{n:<2} {:>6} rows {:>7.1} ms", rows(&b), wall.as_secs_f64() * 1e3);
out.push((n, b, wall));
}
out
}
/// **Every one of the 22 plans, streams and returns rows over the door.** Default
/// SF 0.01 (~87 k rows) so it runs in the ordinary suite; `TPCH_SF` scales it.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn tpch_22_execute_over_flight_sql() {
let sf: f64 = std::env::var("TPCH_SF").ok().and_then(|v| v.parse().ok()).unwrap_or(0.01);
let mut live = bring_up(sf).await;
assert!(live.rows_loaded > 0, "no TPC-H rows generated at SF={sf}");
eprintln!("loaded {} rows in {:.1}s (SF={sf})", live.rows_loaded, live.load.as_secs_f64());
let results = run_22(&mut live.client).await;
for (n, b, _) in &results {
// Q16 at SF 0.01 legitimately has rows; every query has at least one row
// at every SF the spec defines (Q6, Q14, Q17, Q19 are single-row aggregates).
assert!(rows(b) > 0, "Q{n} returned no rows at SF={sf}");
}
}
/// **SF=1: the published answers, gated.** Cardinality of all 22 against the
/// bundled `dbgen` answer sets, Q1's A/F `count_order`, Q6's `revenue`, and the
/// static table above against its source.
///
/// RED-proven (LAW 2): with `Q1_AF_COUNT_ORDER` set to 1_478_494 this panics on
/// the Q1 line; with `SF1_ROWS[8]` set to `(9, 176)` it panics on the
/// table-vs-source line before any query runs.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "requires: SF=1 TPC-H generation (8.66 M rows, ~1 GB under $TMPDIR, ~30 s ingest) — run with --ignored and TMPDIR on md1, not /tmp"]
async fn tpch_sf1_answers_are_canonical_over_flight_sql() {
// The table is a render of the bundled answers; prove it before spending a
// minute generating data.
for (n, expect) in SF1_ROWS {
assert_eq!(bundled_rows(n), expect, "SF1_ROWS drifted from tpchgen's bundled dbgen answer for Q{n}");
}
let mut live = bring_up(1.0).await;
eprintln!("loaded {} rows in {:.1}s (SF=1)", live.rows_loaded, live.load.as_secs_f64());
assert_eq!(live.rows_loaded, 8_661_245, "SF=1 is 8 661 245 rows across the 8 tables");
let results = run_22(&mut live.client).await;
let mut bad = Vec::new();
for (n, b, _) in &results {
let expect = SF1_ROWS.iter().find(|(q, _)| q == n).unwrap().1;
let got = match first_n(*n) {
Some(k) => rows(b).min(k),
None => rows(b),
};
if got != expect {
bad.push(format!("Q{n}: {got} rows, canonical {expect}"));
}
}
assert!(bad.is_empty(), "cardinality off the published SF=1 answers:\n{}", bad.join("\n"));
let q1 = &results[0].1;
let count_order: i64 = cell(q1, &["A", "F"], "count_order").parse().unwrap();
assert_eq!(count_order, Q1_AF_COUNT_ORDER, "Q1 A/F count_order");
let q6 = &results[5].1;
let revenue: f64 = cell(q6, &[], "revenue").parse().unwrap();
assert!(
(revenue - Q6_REVENUE).abs() < 0.00005,
"Q6 revenue {revenue} vs published {Q6_REVENUE} (4 decimals)"
);
let total: Duration = results.iter().map(|r| r.2).sum();
let mut walls: Vec<Duration> = results.iter().map(|r| r.2).collect();
walls.sort();
eprintln!(
"TPC-H SF=1 over Flight SQL: 22/22 canonical; total {:.1}s, median {:.0} ms, slowest Q{} {:.1}s",
total.as_secs_f64(),
walls[11].as_secs_f64() * 1e3,
results.iter().max_by_key(|r| r.2).unwrap().0,
walls[21].as_secs_f64()
);
}
/// **Q15 as the SPEC writes it — `CREATE VIEW revenue0 …; SELECT …; DROP VIEW` —
/// either RUNS or is REFUSED BY NAME. Never an internal error.**
///
/// The engine-accepted `query_sql(15)` is the CTE form and is gated above; a
/// generic client (Power BI, the ODBC driver) sends the view form, one statement
/// at a time. MEASURED 2026-09-10 over the door, before any fix:
///
/// ```text
/// code: 'Internal error', message: "datafusion: External error:
/// Unexpected => register_table does not support tables with data."
/// ```
///
/// That is DataFusion trying to write the view into the Iceberg schema provider
/// and the provider refusing — surfaced as a 500 that names neither "view" nor
/// what the client could do instead. This test lands RED on that (THE LAW: the
/// failing test lands before the fix) and stays the gate for both acceptable
/// outcomes: a session-scoped temp view that makes the three statements run and
/// return the one Q15 row, or a refusal that says "view" and points at the CTE
/// rewrite. `Internal error` is the one thing it never accepts.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn q15_spec_create_view_runs_or_is_refused_by_name_never_an_internal_error() {
let mut live = bring_up(0.01).await;
let spec = tpch::query_sql_spec(15);
let stmts: Vec<String> = spec.split(';').map(|s| s.trim().to_string()).filter(|s| !s.is_empty()).collect();
assert_eq!(stmts.len(), 3, "create view / select / drop view: {stmts:?}");
assert!(stmts[0].to_lowercase().starts_with("create view"), "{}", stmts[0]);
match run(&mut live.client, &stmts[0]).await {
Err(e) => {
eprintln!("Q15 spec form refused: {e}");
assert!(
!e.contains("Internal error"),
"a client-facing refusal, never a 500 — the door must say what it cannot do: {e}"
);
let l = e.to_lowercase();
assert!(
l.contains("view") && (l.contains("with ") || l.contains("cte")),
"the refusal must name the view and the CTE rewrite the client can send instead: {e}"
);
}
Ok(_) => {
let b = run(&mut live.client, &stmts[1]).await.expect("SELECT over the session view");
assert_eq!(rows(&b), 1, "Q15 is one row (the top supplier)");
run(&mut live.client, &stmts[2]).await.expect("DROP VIEW of the session view");
eprintln!("Q15 spec form ran through a session view: 1 row");
}
}
}