skade-flight/tests/datatype_matrix.rs
//! THE DATATYPE MATRIX — every Arrow type skade's Iceberg bridge accepts,
//! written through skade, read back over Flight SQL by the stock client, and
//! exercised per cell: wire type, values, predicate, ORDER BY, CAST to
//! VARCHAR, aggregates, and the ODBC type row a driver would look up.
//!
//! Every cell is one `functional_status` row (`skade-flight/datatype/<col>`)
//! so `cargo test --features testmatrix` turns this file into matrix rows;
//! set `SKADE_FLIGHT_MATRIX_OUT=<path.jsonl>` to have them written. Without
//! the feature the same cells run and the same assertions fire — the file is
//! NEVER feature-gated as a whole (green-by-vacuum is the failure mode the
//! silenced-tests guard exists for).
//!
//! Known bridge facts the matrix pins rather than hides:
//! - Iceberg has no 16-bit int: `Int16` widens to `Int32` on the way back.
//! - Strings/binary may come back as any member of their family
//! (`Utf8`/`LargeUtf8`/`Utf8View`); the cell records which.
//! - `Timestamp(µs, Some(tz))` comes back with whatever tz label the reader
//! attaches (`UTC` / `+00:00`); the cell records which.
use std::sync::Arc;
use std::time::Duration;
use arrow_array::builder::{BinaryBuilder, Decimal128Builder};
use arrow_array::{
Array, ArrayRef, BooleanArray, Date32Array, Float32Array, Float64Array, Int16Array, Int32Array,
Int64Array, RecordBatch, StringArray, Time64MicrosecondArray, TimestampMicrosecondArray,
TimestampNanosecondArray,
};
use arrow_schema::{DataType, Field, Schema, TimeUnit};
use futures::TryStreamExt;
use skade_flight::arrow_flight::sql::CommandGetXdbcTypeInfo;
use skade_flight::arrow_flight::sql::client::FlightSqlServiceClient;
use skade_flight::{Auth, Engine};
use tonic::transport::Channel;
// ── the table ───────────────────────────────────────────────────────────────
struct Col {
name: &'static str,
arrow: DataType,
/// The type we expect on the wire (family-loose for strings/binary/tz).
wire: DataType,
/// ODBC type name the driver looks up in GetXdbcTypeInfo.
xdbc: &'static str,
/// A predicate that matches exactly ROW 1.
predicate: &'static str,
values: ArrayRef,
}
const TS1: i64 = 1_704_198_896_000_000; // 2024-01-02T12:34:56Z in µs
const TS3: i64 = 1_704_285_296_000_000; // 2024-01-03T12:34:56Z
fn columns() -> Vec<Col> {
let dec = {
let mut b = Decimal128Builder::new().with_precision_and_scale(18, 3).unwrap();
b.append_value(1_500);
b.append_null();
b.append_value(3_250);
b.finish()
};
let bin = {
let mut b = BinaryBuilder::new();
b.append_value([1u8, 2]);
b.append_null();
b.append_value([3u8]);
b.finish()
};
vec![
Col {
name: "b",
arrow: DataType::Boolean,
wire: DataType::Boolean,
xdbc: "BOOLEAN",
predicate: "b = true",
values: Arc::new(BooleanArray::from(vec![Some(true), None, Some(false)])),
},
Col {
name: "i16",
arrow: DataType::Int16,
wire: DataType::Int32,
xdbc: "INTEGER",
predicate: "i16 = 1",
values: Arc::new(Int16Array::from(vec![Some(1), None, Some(3)])),
},
Col {
name: "i32",
arrow: DataType::Int32,
wire: DataType::Int32,
xdbc: "INTEGER",
predicate: "i32 = 1",
values: Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])),
},
Col {
name: "i64",
arrow: DataType::Int64,
wire: DataType::Int64,
xdbc: "BIGINT",
predicate: "i64 = 1",
values: Arc::new(Int64Array::from(vec![Some(1), None, Some(3)])),
},
Col {
name: "f32",
arrow: DataType::Float32,
wire: DataType::Float32,
xdbc: "FLOAT",
predicate: "f32 < 2.0",
values: Arc::new(Float32Array::from(vec![Some(1.5), None, Some(3.25)])),
},
Col {
name: "f64",
arrow: DataType::Float64,
wire: DataType::Float64,
xdbc: "DOUBLE",
predicate: "f64 = 1.5",
values: Arc::new(Float64Array::from(vec![Some(1.5), None, Some(3.25)])),
},
Col {
name: "dec",
arrow: DataType::Decimal128(18, 3),
wire: DataType::Decimal128(18, 3),
xdbc: "DECIMAL",
predicate: "dec = 1.500",
values: Arc::new(dec),
},
Col {
name: "d",
arrow: DataType::Date32,
wire: DataType::Date32,
xdbc: "DATE",
predicate: "d = DATE '2024-01-02'",
values: Arc::new(Date32Array::from(vec![Some(19724), None, Some(19725)])),
},
Col {
name: "t",
arrow: DataType::Time64(TimeUnit::Microsecond),
wire: DataType::Time64(TimeUnit::Microsecond),
xdbc: "TIME",
predicate: "t = TIME '12:34:56'",
values: Arc::new(Time64MicrosecondArray::from(vec![
Some(45_296_000_000),
None,
Some(46_296_000_000),
])),
},
Col {
name: "ts",
arrow: DataType::Timestamp(TimeUnit::Microsecond, None),
wire: DataType::Timestamp(TimeUnit::Microsecond, None),
xdbc: "TIMESTAMP",
predicate: "ts = TIMESTAMP '2024-01-02 12:34:56'",
values: Arc::new(TimestampMicrosecondArray::from(vec![Some(TS1), None, Some(TS3)])),
},
Col {
name: "tstz",
arrow: DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
wire: DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
xdbc: "TIMESTAMP",
predicate: "date_part('day', tstz) = 2",
values: Arc::new(
TimestampMicrosecondArray::from(vec![Some(TS1), None, Some(TS3)]).with_timezone("UTC"),
),
},
Col {
name: "tsns",
arrow: DataType::Timestamp(TimeUnit::Nanosecond, None),
wire: DataType::Timestamp(TimeUnit::Nanosecond, None),
xdbc: "TIMESTAMP",
predicate: "tsns = TIMESTAMP '2024-01-02 12:34:56.000000123'",
values: Arc::new(TimestampNanosecondArray::from(vec![
Some(TS1 * 1000 + 123),
None,
Some(TS3 * 1000 + 456),
])),
},
Col {
name: "s",
arrow: DataType::Utf8,
wire: DataType::Utf8,
xdbc: "VARCHAR",
predicate: "s LIKE 'a%'",
values: Arc::new(StringArray::from(vec![Some("ada"), None, Some("cyd")])),
},
Col {
name: "bin",
arrow: DataType::Binary,
wire: DataType::Binary,
xdbc: "VARBINARY",
predicate: "bin = X'0102'",
values: Arc::new(bin),
},
]
}
fn table(cols: &[Col]) -> (Schema, RecordBatch) {
let mut fields = vec![Field::new("id", DataType::Int64, false)];
let mut arrays: Vec<ArrayRef> = vec![Arc::new(Int64Array::from(vec![1, 2, 3]))];
for c in cols {
fields.push(Field::new(c.name, c.arrow.clone(), true));
arrays.push(c.values.clone());
}
let schema = Schema::new(fields);
(schema.clone(), RecordBatch::try_new(Arc::new(schema), arrays).unwrap())
}
/// Same family? Strings, binaries and tz-labelled timestamps are compared
/// loosely; everything else exactly.
fn same_family(expected: &DataType, actual: &DataType) -> bool {
use DataType::*;
match (expected, actual) {
(Utf8, Utf8 | LargeUtf8 | Utf8View) => true,
(Binary, Binary | LargeBinary | BinaryView) => true,
(Timestamp(u1, Some(_)), Timestamp(u2, Some(_))) => u1 == u2,
(e, a) => e == a,
}
}
/// Cells that are RED today for a reason outside this crate, named here so the
/// matrix stays honest AND stable: an unexpected red fails the test, and so
/// does an unexpected GREEN (the gap closed — delete the row). LAW 2: red
/// before green, and a caller must be able to tell working from broken.
///
/// The `t/predicate` row that sat here (DataFusion 55 plans `TIME '..'` as
/// Time64(ns) and could not coerce it to an Iceberg TIME column, Time64(µs))
/// closed on 2026-09-10: the engine lowers an exactly-representable TIME
/// literal to microseconds before type coercion (`time_literal.rs`).
const KNOWN_RED: &[(&str, &str, &str)] = &[];
// ── the harness ─────────────────────────────────────────────────────────────
struct Cell {
col: &'static str,
check: &'static str,
ok: bool,
detail: String,
}
struct Matrix {
cells: Vec<Cell>,
}
impl Matrix {
fn record(&mut self, col: &'static str, check: &'static str, ok: bool, detail: impl Into<String>) {
let detail = detail.into();
skade_flight::functional_status(&format!("skade-flight/datatype/{col}"), check, ok, &detail);
self.cells.push(Cell { col, check, ok, detail });
}
fn render(&self) -> String {
let checks = ["wire_type", "values", "predicate", "order_by", "cast_varchar", "aggregate", "xdbc_row", "field_metadata"];
let mut s = String::new();
s.push_str(&format!("{:<6}", "col"));
for c in checks {
s.push_str(&format!(" {:<12}", c));
}
s.push('\n');
let mut cols: Vec<&str> = self.cells.iter().map(|c| c.col).collect();
cols.dedup();
for col in cols {
s.push_str(&format!("{:<6}", col));
for check in checks {
let mark = match self.cells.iter().find(|c| c.col == col && c.check == check) {
Some(c) if c.ok => "✅",
Some(_) => "❌",
None => "·",
};
s.push_str(&format!(" {:<12}", mark));
}
s.push('\n');
}
for c in self.cells.iter().filter(|c| !c.ok) {
s.push_str(&format!(" ❌ {}/{}: {}\n", c.col, c.check, c.detail));
}
s
}
}
async fn fetch(c: &mut FlightSqlServiceClient<Channel>, sql: &str) -> Result<Vec<RecordBatch>, String> {
let info = c.execute(sql.to_string(), None).await.map_err(|e| e.to_string())?;
let ticket = info.endpoint[0].ticket.clone().ok_or("no ticket")?;
c.do_get(ticket).await.map_err(|e| e.to_string())?.try_collect().await.map_err(|e| e.to_string())
}
fn one_column(batches: &[RecordBatch]) -> Result<ArrayRef, String> {
let arrays: Vec<&dyn Array> = batches.iter().map(|b| b.column(0).as_ref()).collect();
if arrays.is_empty() {
return Err("no batches".into());
}
arrow_select::concat::concat(&arrays).map_err(|e| e.to_string())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn datatype_matrix_over_flight_sql() {
let cols = columns();
let dir = tempfile::tempdir().unwrap();
let wh = skade::Warehouse::open(dir.path().join("wh")).await.unwrap();
let (schema, batch) = table(&cols);
wh.create_table("types", &schema).await.unwrap().append(&[batch]).await.unwrap();
let engine = Arc::new(Engine::with_options(wh, Duration::from_secs(60), 2).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(skade_flight::serve_on(engine, auth, listener, Default::default(), futures::future::pending()));
let channel = Channel::from_shared(format!("http://{addr}")).unwrap().connect().await.unwrap();
let mut c = FlightSqlServiceClient::new(channel);
// The ODBC type table, once.
let info = c.get_xdbc_type_info(CommandGetXdbcTypeInfo::default()).await.unwrap();
let xdbc: Vec<RecordBatch> =
c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect().await.unwrap();
let xdbc_names: Vec<String> = xdbc
.iter()
.flat_map(|b| {
let a = b.column_by_name("type_name").unwrap();
let a = a.as_any().downcast_ref::<StringArray>().unwrap();
(0..a.len()).map(|i| a.value(i).to_string()).collect::<Vec<_>>()
})
.collect();
// The table schema as GetTables(include_schema) hands it to a driver, with
// the per-field metadata the ODBC driver dereferences without a null check.
let info = c
.get_tables(skade_flight::arrow_flight::sql::CommandGetTables { include_schema: true, ..Default::default() })
.await
.unwrap();
let tables: Vec<RecordBatch> =
c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect().await.unwrap();
let names = tables[0].column_by_name("table_name").unwrap();
let names = names.as_any().downcast_ref::<StringArray>().unwrap();
let row = (0..names.len()).find(|i| names.value(*i) == "types").expect("types listed");
let ipc = tables[0].column_by_name("table_schema").unwrap();
let ipc = ipc.as_any().downcast_ref::<arrow_array::BinaryArray>().unwrap().value(row);
let listed = arrow_ipc::convert::try_schema_from_ipc_buffer(ipc).expect("table_schema decodes");
let mut m = Matrix { cells: Vec::new() };
for col in &cols {
let n = col.name;
// field_metadata: TYPE_NAME present, equal to the xdbc name, searchable.
let r = listed.field_with_name(n).map_err(|e| e.to_string()).and_then(|f| {
let md = f.metadata();
let tn = md.get("ARROW:FLIGHT:SQL:TYPE_NAME").ok_or("no TYPE_NAME")?;
let s = md.get("ARROW:FLIGHT:SQL:IS_SEARCHABLE").ok_or("no IS_SEARCHABLE")?;
if tn != col.xdbc {
return Err(format!("TYPE_NAME {tn}, want {}", col.xdbc));
}
if s != "1" {
return Err(format!("IS_SEARCHABLE {s}"));
}
Ok(format!("{tn} p={} s={}", md.get("ARROW:FLIGHT:SQL:PRECISION").cloned().unwrap_or_default(), md.get("ARROW:FLIGHT:SQL:SCALE").cloned().unwrap_or_default()))
});
let (ok, detail) = match r { Ok(d) => (true, d), Err(e) => (false, e) };
m.record(n, "field_metadata", ok, detail);
// wire_type + values
match fetch(&mut c, &format!("SELECT {n} FROM types ORDER BY id")).await.and_then(|b| one_column(&b)) {
Ok(actual) => {
let dt = actual.data_type().clone();
m.record(n, "wire_type", same_family(&col.wire, &dt), format!("{:?} -> {dt:?}", col.arrow));
// Compare in the expected wire type so a family-loose type still
// has its VALUES checked exactly.
let cmp = match arrow_cast::cast(&actual, &col.wire) {
Ok(a) => {
let expected = arrow_cast::cast(&col.values, &col.wire).unwrap();
if a.as_ref() == expected.as_ref() {
Ok(())
} else {
Err(format!("got {a:?}, want {expected:?}"))
}
}
Err(e) => Err(e.to_string()),
};
m.record(n, "values", cmp.is_ok(), cmp.err().unwrap_or_default());
}
Err(e) => {
m.record(n, "wire_type", false, e.clone());
m.record(n, "values", false, e);
}
}
// predicate: matches exactly row 1
let r = fetch(&mut c, &format!("SELECT id FROM types WHERE {}", col.predicate)).await.and_then(|b| {
let ids = one_column(&b)?;
let ids = ids.as_any().downcast_ref::<Int64Array>().ok_or("id not Int64")?;
if ids.len() == 1 && ids.value(0) == 1 { Ok(()) } else { Err(format!("matched ids {ids:?}")) }
});
m.record(n, "predicate", r.is_ok(), format!("{} {}", col.predicate, r.err().unwrap_or_default()));
// order_by: DESC NULLS LAST puts row 3 first, null last
let r = fetch(&mut c, &format!("SELECT id FROM types ORDER BY {n} DESC NULLS LAST")).await.and_then(|b| {
let ids = one_column(&b)?;
let ids = ids.as_any().downcast_ref::<Int64Array>().ok_or("id not Int64")?;
let v: Vec<i64> = ids.values().to_vec();
// Booleans: false < true so DESC gives row 1 (true) first.
let want = if n == "b" { vec![1, 3, 2] } else { vec![3, 1, 2] };
if v == want { Ok(()) } else { Err(format!("order {v:?}, want {want:?}")) }
});
m.record(n, "order_by", r.is_ok(), r.err().unwrap_or_default());
// cast_varchar
let r = fetch(&mut c, &format!("SELECT CAST({n} AS VARCHAR) FROM types ORDER BY id")).await.and_then(|b| {
let a = one_column(&b)?;
match a.data_type() {
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => {
if a.null_count() == 1 { Ok(format!("{:?}", a)) } else { Err(format!("nulls {}", a.null_count())) }
}
other => Err(format!("cast gave {other:?}")),
}
});
let (ok, detail) = match r { Ok(d) => (true, d), Err(e) => (false, e) };
m.record(n, "cast_varchar", ok, detail);
// aggregate: count skips the null; min/max exist
let r = fetch(&mut c, &format!("SELECT count({n}), min({n}), max({n}) FROM types")).await.and_then(|b| {
let cnt = b[0].column(0).as_any().downcast_ref::<Int64Array>().ok_or("count not Int64")?.value(0);
if cnt == 2 && !b[0].column(1).is_null(0) && !b[0].column(2).is_null(0) {
Ok(())
} else {
Err(format!("count {cnt}, min null {}, max null {}", b[0].column(1).is_null(0), b[0].column(2).is_null(0)))
}
});
m.record(n, "aggregate", r.is_ok(), r.err().unwrap_or_default());
// xdbc_row: the driver can look the type up
let has = xdbc_names.iter().any(|x| x == col.xdbc);
m.record(n, "xdbc_row", has, format!("{} in GetXdbcTypeInfo", col.xdbc));
}
let rendered = m.render();
eprintln!("\n{rendered}");
#[cfg(feature = "testmatrix")]
{
let rows = skade_flight::drain_functional_rows();
eprintln!("testmatrix: {} functional rows drained", rows.len());
if let Ok(p) = std::env::var("SKADE_FLIGHT_MATRIX_OUT") {
use nornir_testmatrix::TestSink;
nornir_testmatrix::JsonFileSink::new(&p).append(&rows).unwrap();
eprintln!("testmatrix: written to {p}");
}
}
let known = |c: &Cell| KNOWN_RED.iter().find(|(col, check, _)| *col == c.col && *check == c.check);
let unexpected_red: Vec<&Cell> = m.cells.iter().filter(|c| !c.ok && known(c).is_none()).collect();
let unexpected_green: Vec<&Cell> = m.cells.iter().filter(|c| c.ok && known(c).is_some()).collect();
for c in m.cells.iter().filter(|c| !c.ok) {
if let Some((_, _, why)) = known(c) {
eprintln!(" known red {}/{}: {why}", c.col, c.check);
}
}
assert!(unexpected_red.is_empty(), "{} unexpected red cell(s):\n{rendered}", unexpected_red.len());
assert!(
unexpected_green.is_empty(),
"{} KNOWN_RED cell(s) turned green — the gap closed, delete the row(s): {:?}",
unexpected_green.len(),
unexpected_green.iter().map(|c| (c.col, c.check)).collect::<Vec<_>>()
);
}