crates/graphar-flight/src/convert.rs
//! FalkorDB GRAPH.QUERY response (Redis RESP) → Arrow RecordBatch.
//!
//! Response shape (non-compact mode):
//! Array[
//! Array[col_name, ...] ← headers (index 0)
//! Array[Array[val, ...], ...] ← data rows (index 1)
//! Array[stat_string, ...] ← stats (index 2, ignored)
//! ]
use std::sync::Arc;
use arrow_array::{
ArrayRef, RecordBatch,
builder::{BooleanBuilder, Float64Builder, Int64Builder, StringBuilder},
};
use arrow_schema::{DataType, Field, SchemaRef};
use redis::Value;
use crate::error::{FlightSqlError, Result};
/// Convert a raw `GRAPH.QUERY` Redis response to a `RecordBatch` using a
/// **pre-registered** schema to guide type coercion.
///
/// Schema fields must be listed in the same order as the `RETURN` clause.
pub fn response_to_batch(response: Value, schema: &SchemaRef) -> Result<RecordBatch> {
let mut outer = match response {
Value::Array(v) => v,
other => {
return Err(FlightSqlError::MalformedResponse(format!(
"expected Array at top level, got {other:?}"
)));
}
};
if outer.len() < 2 {
return Ok(RecordBatch::new_empty(schema.clone()));
}
let rows_val = outer.remove(1);
let rows = match rows_val {
Value::Array(r) => r,
Value::Nil => vec![],
other => {
return Err(FlightSqlError::MalformedResponse(format!(
"expected Array for data rows, got {other:?}"
)));
}
};
let ncols = schema.fields().len();
let nrows = rows.len();
let mut builders: Vec<ColumnBuilder> = schema
.fields()
.iter()
.map(|f| ColumnBuilder::new(f.data_type(), nrows))
.collect();
for (row_idx, row_val) in rows.into_iter().enumerate() {
let cells = match row_val {
Value::Array(c) => c,
other => {
return Err(FlightSqlError::MalformedResponse(format!(
"row {row_idx}: expected Array, got {other:?}"
)));
}
};
if cells.len() != ncols {
return Err(FlightSqlError::ColumnCountMismatch {
schema: ncols,
row: cells.len(),
});
}
for (col_idx, cell) in cells.iter().enumerate() {
builders[col_idx].append(cell, row_idx)?;
}
}
let columns: Vec<ArrayRef> = builders.into_iter().map(|b| b.finish()).collect();
Ok(RecordBatch::try_new(schema.clone(), columns)?)
}
/// Convert a raw `GRAPH.QUERY` response to a `RecordBatch` **without a
/// pre-registered schema**.
///
/// Column names are taken from the FalkorDB response header row. The Arrow
/// type for each column is inferred by inspecting the first non-null Redis
/// value in that column:
///
/// | Redis value | Arrow type |
/// |---|---|
/// | `Int` | `Int64` |
/// | `Double` | `Float64` |
/// | `Boolean` | `Boolean` |
/// | `BulkString` / `SimpleString` / other | `Utf8` |
/// | all-null column | `Utf8` (fallback) |
///
/// This means integer IDs like `_gar_id`, `src`, and `dst` are returned as
/// `Int64` by default — no registration needed.
pub fn response_to_batch_auto(response: Value) -> Result<RecordBatch> {
let outer = match response {
Value::Array(v) => v,
other => {
return Err(FlightSqlError::MalformedResponse(format!(
"expected Array at top level, got {other:?}"
)));
}
};
if outer.len() < 2 {
return Ok(RecordBatch::new_empty(Arc::new(
arrow_schema::Schema::empty(),
)));
}
let col_names = extract_string_vec(&outer[0])?;
let rows: Vec<&Value> = match &outer[1] {
Value::Array(r) => r.iter().collect(),
Value::Nil => vec![],
other => {
return Err(FlightSqlError::MalformedResponse(format!(
"expected Array for data rows, got {other:?}"
)));
}
};
let ncols = col_names.len();
let nrows = rows.len();
if ncols == 0 {
return Ok(RecordBatch::new_empty(Arc::new(
arrow_schema::Schema::empty(),
)));
}
// Pass 1 — sniff the Arrow type for each column in a single row-major scan
// that short-circuits once every column's type is known (M4).
let col_types = sniff_column_types(&rows, ncols);
let schema = Arc::new(arrow_schema::Schema::new(
col_names
.iter()
.zip(&col_types)
.map(|(n, dt)| Field::new(n.as_str(), dt.clone(), true))
.collect::<Vec<_>>(),
));
// Pass 2 — fill typed builders.
let mut builders: Vec<ColumnBuilder> = col_types
.iter()
.map(|dt| ColumnBuilder::new(dt, nrows))
.collect();
for (row_idx, row_val) in rows.iter().enumerate() {
let cells = match row_val {
Value::Array(c) => c,
other => {
return Err(FlightSqlError::MalformedResponse(format!(
"row {row_idx}: expected Array, got {other:?}"
)));
}
};
// Validate the row width before filling builders. Without this a ragged
// row (fewer cells than columns) silently under-fills the trailing
// builders, and the mismatch only surfaces later as an opaque
// `RecordBatch::try_new` length error. Mirror the schema-driven
// `response_to_batch`, which already returns `ColumnCountMismatch`.
if cells.len() != ncols {
return Err(FlightSqlError::ColumnCountMismatch {
schema: ncols,
row: cells.len(),
});
}
for (col_idx, cell) in cells.iter().enumerate() {
builders[col_idx].append(cell, row_idx)?;
}
}
let columns: Vec<ArrayRef> = builders.into_iter().map(|b| b.finish()).collect();
Ok(RecordBatch::try_new(schema, columns)?)
}
/// Map a single Redis cell to the Arrow type it fixes for its column, or `None`
/// for a `Nil` cell (which leaves the column undetermined).
///
/// This is the *one* place the per-cell type rule lives; both the original
/// per-column sniff and the single-pass sniff agree because they call it.
#[inline]
fn cell_type(cell: &Value) -> Option<DataType> {
match cell {
Value::Nil => None, // null: keep the column undetermined
Value::Int(_) => Some(DataType::Int64),
Value::Double(_) => Some(DataType::Float64),
Value::Boolean(_) => Some(DataType::Boolean),
_ => Some(DataType::Utf8),
}
}
/// Read an integer-ish column of a batch this module produced into an owned
/// `Vec<Option<i64>>`, coercing to i64 whatever width/sign the sniff emitted.
///
/// **This is the inverse of [`sniff_column_types`], and it lives here — next to
/// the sniff whose behaviour it compensates for — because every consumer that
/// reads an integer column back out of a `response_to_batch_auto` batch needs
/// exactly this coercion, and a copy that drifts from the sniff is a live bug.**
///
/// FalkorDB's `id(a)` arrives as a Redis `Int`, which [`cell_type`] maps to
/// `Int64`. But the sniff only fixes a column's type once it observes a non-null
/// cell: on a **zero-row** result (e.g. an empty graph, or a `MATCH` that matched
/// nothing) every column falls back to `Utf8`. A hard `Int64Array` downcast then
/// fails, and korp's graph view reported the misleading
/// `query must RETURN an Int64 src` error even though the query was correct.
///
/// So we accept the whole integer family AND `Utf8` (parsing each string to i64),
/// coercing to i64 — an empty result yields an empty (but present) column, not an
/// error. Returns `None` only if the column is genuinely absent or a type we
/// can't read as an integer (e.g. Float/Boolean), so a real shape mistake still
/// surfaces a clear error.
///
/// The `Utf8` arm is not only the zero-row case: `cell_type` maps every non
/// Int/Double/Boolean cell to `Utf8`, so a `BulkString` id from a driver in
/// string mode reads back correctly too.
pub fn i64_column(batch: &RecordBatch, name: &str) -> Option<Vec<Option<i64>>> {
use arrow_array::cast::AsArray;
use arrow_array::types::{
Int8Type, Int16Type, Int32Type, Int64Type, UInt8Type, UInt16Type, UInt32Type, UInt64Type,
};
use arrow_array::{Array, StringArray};
let col = batch.column_by_name(name)?;
let n = col.len();
let out: Vec<Option<i64>> = match col.data_type() {
DataType::Int64 => {
let a = col.as_primitive::<Int64Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i)))
.collect()
}
DataType::Int32 => {
let a = col.as_primitive::<Int32Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i) as i64))
.collect()
}
DataType::Int16 => {
let a = col.as_primitive::<Int16Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i) as i64))
.collect()
}
DataType::Int8 => {
let a = col.as_primitive::<Int8Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i) as i64))
.collect()
}
DataType::UInt64 => {
let a = col.as_primitive::<UInt64Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i) as i64))
.collect()
}
DataType::UInt32 => {
let a = col.as_primitive::<UInt32Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i) as i64))
.collect()
}
DataType::UInt16 => {
let a = col.as_primitive::<UInt16Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i) as i64))
.collect()
}
DataType::UInt8 => {
let a = col.as_primitive::<UInt8Type>();
(0..n)
.map(|i| (!a.is_null(i)).then(|| a.value(i) as i64))
.collect()
}
// The zero-row / all-null fallback the sniffer leaves as Utf8 — parse the
// strings (and yield an empty vec when there are no rows).
DataType::Utf8 => {
let a = col.as_any().downcast_ref::<StringArray>()?;
(0..n)
.map(|i| {
if a.is_null(i) {
None
} else {
a.value(i).trim().parse::<i64>().ok()
}
})
.collect()
}
_ => return None,
};
Some(out)
}
/// Read a **float-ish** column of a batch this module produced into an owned
/// `Vec<Option<f64>>`, coercing to f64 whatever the sniff emitted.
///
/// **The f64 sibling of [`i64_column`], and it lives here for the same reason: next to
/// [`sniff_column_types`], whose behaviour it compensates for.**
///
/// The compensation is not hypothetical. FalkorDB stores a node's `lon`/`lat` as
/// doubles, but over RESP2 they arrive as **bulk strings**, which [`cell_type`] maps to
/// `Utf8` — so a reader that downcast to `Float64Array` and nothing else got `None` for
/// every coordinate on a live FalkorDB while passing every fixture test that built the
/// batch with a `Float64Array`. MEASURED on oden 2026-09-02: a 77 916-junction Stockholm
/// slice arrived at korp's canvas with `geo_nodes: 0` and drew as a hairball, twice,
/// against a server that had answered perfectly.
///
/// So: `Float64`/`Float32`, the whole integer family (a whole-degree coordinate comes
/// back as an `Int`), and `Utf8` (parsed). `None` only when the column is absent or is a
/// type no number can be read out of, so a real shape mistake still surfaces.
pub fn f64_column(batch: &RecordBatch, name: &str) -> Option<Vec<Option<f64>>> {
use arrow_array::cast::AsArray;
use arrow_array::types::{
Float32Type, Float64Type, Int8Type, Int16Type, Int32Type, Int64Type, UInt8Type, UInt16Type,
UInt32Type, UInt64Type,
};
use arrow_array::{Array, StringArray};
let col = batch.column_by_name(name)?;
let n = col.len();
macro_rules! prim {
($t:ty) => {{
let a = col.as_primitive::<$t>();
(0..n).map(|i| (!a.is_null(i)).then(|| f64::from(a.value(i)))).collect()
}};
}
macro_rules! prim_as {
($t:ty) => {{
let a = col.as_primitive::<$t>();
(0..n).map(|i| (!a.is_null(i)).then(|| a.value(i) as f64)).collect()
}};
}
let out: Vec<Option<f64>> = match col.data_type() {
DataType::Float64 => prim!(Float64Type),
DataType::Float32 => prim!(Float32Type),
DataType::Int32 => prim!(Int32Type),
DataType::Int16 => prim!(Int16Type),
DataType::Int8 => prim!(Int8Type),
DataType::UInt32 => prim!(UInt32Type),
DataType::UInt16 => prim!(UInt16Type),
DataType::UInt8 => prim!(UInt8Type),
DataType::Int64 => prim_as!(Int64Type),
DataType::UInt64 => prim_as!(UInt64Type),
// The RESP2 bulk-string case AND the zero-row / all-null fallback — both land
// here as Utf8. This arm is the whole reason the function exists.
DataType::Utf8 => {
let a = col.as_any().downcast_ref::<StringArray>()?;
(0..n)
.map(|i| {
if a.is_null(i) { None } else { a.value(i).trim().parse::<f64>().ok() }
})
.collect()
}
_ => return None,
};
Some(out)
}
/// Infer the Arrow type of every column in a **single row-major pass** that
/// stops as soon as all columns are determined (M4 — single-pass type sniff).
///
/// ## Equivalence to the previous per-column scan
///
/// The previous code called `sniff_column_type(rows, ci)` once per column,
/// each call walking `rows` top-to-bottom and returning the type of the
/// **first non-null cell** in that column (falling back to `Utf8` for an
/// all-null/empty column). Crucially the rule is *first-non-null wins* — there
/// is **no type promotion**: a column's type never changes once a non-null cell
/// is seen, regardless of what later rows contain (e.g. an `Int` first row
/// followed by a `Double` row still yields `Int64`; the builder then coerces
/// the later cells, see `redis_to_i64`). Because the type is a pure function of
/// the *first non-null cell per column* and is independent across columns, the
/// order in which we visit cells does not affect the result, so we can fuse the
/// `ncols` independent passes into one row-major pass:
///
/// * `types[ci]` is written exactly once — on the first non-null cell of column
/// `ci` — and never overwritten, matching "first non-null wins".
/// * columns whose every cell is `Nil` (or that have no cell in any row) are
/// never written and keep the `Utf8` default, matching the all-null fallback.
/// * once `remaining == 0` every column is fixed and no later cell could change
/// any answer, so we stop early — saving the wasted tail scan M4 targets.
///
/// Rows shorter than `ncols` (a malformed/ragged response) are handled exactly
/// as before: `cells.get(ci)` simply yields `None` for the missing trailing
/// columns, leaving them undetermined just like the per-column scan's
/// `cells.get(col_idx)` miss did.
fn sniff_column_types(rows: &[&Value], ncols: usize) -> Vec<DataType> {
// Default for all-null / empty columns is Utf8 (the original fallback).
let mut types = vec![DataType::Utf8; ncols];
// `determined[ci]` flips true the moment column `ci`'s type is fixed.
let mut determined = vec![false; ncols];
let mut remaining = ncols;
for row in rows {
if remaining == 0 {
break; // every column resolved — the rest of the scan is wasted work
}
let Value::Array(cells) = row else { continue };
for ci in 0..ncols {
if determined[ci] {
continue; // already fixed by an earlier row — skip
}
if let Some(cell) = cells.get(ci)
&& let Some(dt) = cell_type(cell)
{
types[ci] = dt;
determined[ci] = true;
remaining -= 1;
}
}
}
types
}
// ── Per-column builder ────────────────────────────────────────────────────────
enum ColumnBuilder {
Int64(Int64Builder),
Float64(Float64Builder),
Bool(BooleanBuilder),
Utf8(StringBuilder),
}
impl ColumnBuilder {
fn new(dt: &DataType, capacity: usize) -> Self {
match dt {
DataType::Int8
| DataType::Int16
| DataType::Int32
| DataType::Int64
| DataType::UInt8
| DataType::UInt16
| DataType::UInt32
| DataType::UInt64 => Self::Int64(Int64Builder::with_capacity(capacity)),
DataType::Float16 | DataType::Float32 | DataType::Float64 => {
Self::Float64(Float64Builder::with_capacity(capacity))
}
DataType::Boolean => Self::Bool(BooleanBuilder::with_capacity(capacity)),
_ => Self::Utf8(StringBuilder::with_capacity(capacity, capacity * 16)),
}
}
fn append(&mut self, val: &Value, row: usize) -> Result<()> {
match self {
Self::Int64(b) => b.append_option(redis_to_i64(val, row)?),
Self::Float64(b) => b.append_option(redis_to_f64(val, row)?),
Self::Bool(b) => b.append_option(redis_to_bool(val, row)?),
// Append directly into the builder; for the common valid-UTF-8
// BulkString this borrows (Cow::Borrowed) with no owned-String copy.
Self::Utf8(b) => append_redis_str(b, val),
}
Ok(())
}
fn finish(self) -> ArrayRef {
match self {
Self::Int64(mut b) => Arc::new(b.finish()),
Self::Float64(mut b) => Arc::new(b.finish()),
Self::Bool(mut b) => Arc::new(b.finish()),
Self::Utf8(mut b) => Arc::new(b.finish()),
}
}
}
// ── Redis Value → scalar coercions ───────────────────────────────────────────
fn redis_to_i64(v: &Value, row: usize) -> Result<Option<i64>> {
match v {
Value::Nil => Ok(None),
Value::Int(i) => Ok(Some(*i)),
Value::Double(d) => Ok(Some(*d as i64)),
Value::BulkString(b) => {
let s = std::str::from_utf8(b).unwrap_or("");
s.parse::<i64>()
.map(Some)
.map_err(|_| malformed(row, "expected i64", s))
}
Value::SimpleString(s) => s
.parse::<i64>()
.map(Some)
.map_err(|_| malformed(row, "expected i64", s)),
other => Err(malformed(
row,
"expected Int for Int64",
&format!("{other:?}"),
)),
}
}
fn redis_to_f64(v: &Value, row: usize) -> Result<Option<f64>> {
match v {
Value::Nil => Ok(None),
Value::Double(d) => Ok(Some(*d)),
Value::Int(i) => Ok(Some(*i as f64)),
Value::BulkString(b) => {
let s = std::str::from_utf8(b).unwrap_or("");
s.parse::<f64>()
.map(Some)
.map_err(|_| malformed(row, "expected f64", s))
}
Value::SimpleString(s) => s
.parse::<f64>()
.map(Some)
.map_err(|_| malformed(row, "expected f64", s)),
other => Err(malformed(
row,
"expected Double for Float64",
&format!("{other:?}"),
)),
}
}
fn redis_to_bool(v: &Value, row: usize) -> Result<Option<bool>> {
match v {
Value::Nil => Ok(None),
Value::Boolean(b) => Ok(Some(*b)),
Value::Int(i) => Ok(Some(*i != 0)),
Value::BulkString(b) => match b.as_slice() {
b"true" | b"1" => Ok(Some(true)),
b"false" | b"0" => Ok(Some(false)),
other => Err(malformed(
row,
"expected bool",
std::str::from_utf8(other).unwrap_or("?"),
)),
},
other => Err(malformed(row, "expected Boolean", &format!("{other:?}"))),
}
}
/// Append a redis value to a `StringBuilder` without an intermediate owned
/// `String` for the common cases (`from_utf8_lossy` is `Cow::Borrowed` when the
/// bytes are valid UTF-8, and the builder copies into its own buffer anyway).
fn append_redis_str(b: &mut StringBuilder, v: &Value) {
match v {
Value::Nil => b.append_null(),
Value::BulkString(bytes) => b.append_value(String::from_utf8_lossy(bytes)),
Value::SimpleString(s) => b.append_value(s),
Value::Int(i) => b.append_value(i.to_string()),
Value::Double(d) => b.append_value(d.to_string()),
Value::Boolean(x) => b.append_value(if *x { "true" } else { "false" }),
other => b.append_value(format!("{other:?}")),
}
}
fn extract_string_vec(v: &Value) -> Result<Vec<String>> {
match v {
Value::Array(items) => items
.iter()
.map(|item| match item {
Value::BulkString(b) => Ok(String::from_utf8_lossy(b).into_owned()),
Value::SimpleString(s) => Ok(s.clone()),
other => Ok(format!("{other:?}")),
})
.collect(),
other => Err(FlightSqlError::MalformedResponse(format!(
"expected Array for column names, got {other:?}"
))),
}
}
fn malformed(row: usize, expected: &str, got: &str) -> FlightSqlError {
FlightSqlError::MalformedResponse(format!("row {row}: {expected}, got '{got}'"))
}
#[cfg(test)]
mod tests {
use super::*;
fn header(names: &[&str]) -> Value {
Value::Array(
names
.iter()
.map(|n| Value::BulkString(n.as_bytes().to_vec()))
.collect(),
)
}
fn row(cells: Vec<Value>) -> Value {
Value::Array(cells)
}
/// A well-formed 2-column response round-trips through the schema-less path.
#[test]
fn auto_well_formed_rows_build_batch() {
let resp = Value::Array(vec![
header(&["id", "name"]),
Value::Array(vec![
row(vec![Value::Int(1), Value::BulkString(b"a".to_vec())]),
row(vec![Value::Int(2), Value::BulkString(b"b".to_vec())]),
]),
]);
let batch = response_to_batch_auto(resp).unwrap();
assert_eq!(batch.num_rows(), 2);
assert_eq!(batch.num_columns(), 2);
}
/// A ragged row (fewer cells than columns) must be rejected with a precise
/// `ColumnCountMismatch`, not silently under-fill the trailing builder and
/// surface later as an opaque `RecordBatch::try_new` length error.
#[test]
fn auto_short_row_is_column_count_mismatch() {
let resp = Value::Array(vec![
header(&["id", "name"]),
Value::Array(vec![
row(vec![Value::Int(1), Value::BulkString(b"a".to_vec())]),
row(vec![Value::Int(2)]), // ragged: only 1 cell for 2 columns
]),
]);
let err = response_to_batch_auto(resp).unwrap_err();
match err {
FlightSqlError::ColumnCountMismatch { schema, row } => {
assert_eq!(schema, 2);
assert_eq!(row, 1);
}
other => panic!("expected ColumnCountMismatch, got {other:?}"),
}
}
/// An over-wide row (more cells than columns) is likewise a mismatch, where
/// previously the extra cell was silently dropped.
#[test]
fn auto_long_row_is_column_count_mismatch() {
let resp = Value::Array(vec![
header(&["id"]),
Value::Array(vec![row(vec![Value::Int(1), Value::Int(99)])]),
]);
let err = response_to_batch_auto(resp).unwrap_err();
assert!(matches!(
err,
FlightSqlError::ColumnCountMismatch { schema: 1, row: 2 }
));
}
/// A nested Cypher value (a `map`/`node`/`list` — anything Redis returns as a
/// non-scalar `Value`) has no scalar Arrow type, so [`cell_type`] falls back
/// to `Utf8` and the cell is rendered via its `Debug` form. This pins that
/// documented fallback: `RETURN {name:'alice', age:30}` (a map) collapses to
/// a `Utf8` column whose value is the map's debug string, rather than
/// erroring or silently dropping the column.
#[test]
fn auto_nested_cypher_value_falls_back_to_utf8() {
// A Cypher map value, as FalkorDB returns it over RESP.
let cypher_map = Value::Map(vec![
(
Value::BulkString(b"name".to_vec()),
Value::BulkString(b"alice".to_vec()),
),
(Value::BulkString(b"age".to_vec()), Value::Int(30)),
]);
// A Cypher list value (nested array).
let cypher_list = Value::Array(vec![Value::Int(1), Value::Int(2), Value::Int(3)]);
let resp = Value::Array(vec![
header(&["id", "props", "tags"]),
Value::Array(vec![row(vec![
Value::Int(7),
cypher_map.clone(),
cypher_list.clone(),
])]),
]);
let batch = response_to_batch_auto(resp).unwrap();
assert_eq!(batch.num_rows(), 1);
assert_eq!(batch.num_columns(), 3);
// The nested columns sniff to Utf8 (the fallback), the scalar id stays Int64.
assert_eq!(batch.schema().field(0).data_type(), &DataType::Int64);
assert_eq!(batch.schema().field(1).data_type(), &DataType::Utf8);
assert_eq!(batch.schema().field(2).data_type(), &DataType::Utf8);
// The cell is the value's Debug rendering (what `append_redis_str` emits
// for a non-scalar), so no data is lost — it is stringified, not dropped.
let props = batch
.column(1)
.as_any()
.downcast_ref::<arrow_array::StringArray>()
.unwrap();
assert_eq!(props.value(0), format!("{cypher_map:?}"));
let tags = batch
.column(2)
.as_any()
.downcast_ref::<arrow_array::StringArray>()
.unwrap();
assert_eq!(tags.value(0), format!("{cypher_list:?}"));
}
/// ZERO-LOSS pin for the decode path: the full `RecordBatch` (schema AND
/// every cell, nulls included) for a varied fixture — ints with Nils,
/// doubles, booleans, bulk + simple strings, an all-null column (Utf8
/// fallback) and a nested map (Debug-stringified). Built by hand with Arrow
/// arrays, so any byte the decoder shifts fails the assert.
#[test]
fn auto_decode_varied_fixture_pins_whole_batch() {
use arrow_array::{BooleanArray, Float64Array, Int64Array, StringArray};
let map = Value::Map(vec![(
Value::BulkString(b"k".to_vec()),
Value::Int(1),
)]);
let resp = Value::Array(vec![
header(&["id", "score", "ok", "name", "ghost", "props"]),
Value::Array(vec![
row(vec![
Value::Int(7),
Value::Double(0.5),
Value::Boolean(true),
Value::BulkString(b"alpha".to_vec()),
Value::Nil,
map.clone(),
]),
row(vec![
Value::Nil,
Value::Nil,
Value::Nil,
Value::SimpleString("beta".into()),
Value::Nil,
Value::Nil,
]),
row(vec![
Value::Int(-3),
Value::Double(f64::MAX),
Value::Boolean(false),
Value::Nil,
Value::Nil,
map.clone(),
]),
]),
Value::Array(vec![]), // stats, ignored
]);
let batch = response_to_batch_auto(resp).unwrap();
let want = RecordBatch::try_new(
Arc::new(arrow_schema::Schema::new(vec![
Field::new("id", DataType::Int64, true),
Field::new("score", DataType::Float64, true),
Field::new("ok", DataType::Boolean, true),
Field::new("name", DataType::Utf8, true),
Field::new("ghost", DataType::Utf8, true), // all-null fallback
Field::new("props", DataType::Utf8, true), // nested → Debug string
])),
vec![
Arc::new(Int64Array::from(vec![Some(7), None, Some(-3)])) as ArrayRef,
Arc::new(Float64Array::from(vec![Some(0.5), None, Some(f64::MAX)])),
Arc::new(BooleanArray::from(vec![Some(true), None, Some(false)])),
Arc::new(StringArray::from(vec![Some("alpha"), Some("beta"), None])),
Arc::new(StringArray::from(vec![None::<&str>, None, None])),
Arc::new(StringArray::from(vec![
Some(format!("{map:?}")),
None,
Some(format!("{map:?}")),
])),
],
)
.unwrap();
assert_eq!(batch, want, "decode must reproduce the batch exactly");
}
/// ZERO-LOSS pin for the schema-driven path: registered types drive the
/// coercions (a Double cell into Int64, a numeric BulkString into Int64 and
/// Float64, Int 0/1 into Boolean), exactly as today.
#[test]
fn schema_decode_coercions_pin_whole_batch() {
use arrow_array::{BooleanArray, Float64Array, Int64Array, StringArray};
let schema: SchemaRef = Arc::new(arrow_schema::Schema::new(vec![
Field::new("i", DataType::Int64, true),
Field::new("f", DataType::Float64, true),
Field::new("b", DataType::Boolean, true),
Field::new("s", DataType::Utf8, true),
]));
let resp = Value::Array(vec![
header(&["i", "f", "b", "s"]),
Value::Array(vec![
row(vec![
Value::Double(2.9), // → 2 (as-cast, the documented coercion)
Value::Int(4), // → 4.0
Value::Int(1), // → true
Value::Int(9), // → "9"
]),
row(vec![
Value::BulkString(b"11".to_vec()), // parsed
Value::BulkString(b"2.25".to_vec()), // parsed
Value::BulkString(b"false".to_vec()),
Value::Double(1.5), // → "1.5"
]),
row(vec![Value::Nil, Value::Nil, Value::Nil, Value::Nil]),
]),
]);
let batch = response_to_batch(resp, &schema).unwrap();
let want = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from(vec![Some(2), Some(11), None])) as ArrayRef,
Arc::new(Float64Array::from(vec![Some(4.0), Some(2.25), None])),
Arc::new(BooleanArray::from(vec![Some(true), Some(false), None])),
Arc::new(StringArray::from(vec![Some("9"), Some("1.5"), None])),
],
)
.unwrap();
assert_eq!(batch, want, "schema-driven decode must coerce exactly as pinned");
}
/// ZERO-LOSS pin on FIRST-ERROR order (row-major): a bad cell in row 0
/// must win over a ragged row 1, and a ragged row 0 must win over a bad
/// cell in row 1 — whatever order the cells are visited in internally.
#[test]
fn decode_first_error_is_row_major() {
let schema: SchemaRef = Arc::new(arrow_schema::Schema::new(vec![
Field::new("i", DataType::Int64, true),
Field::new("j", DataType::Int64, true),
]));
// Bad cell (row 0, col 1) beats ragged row 1.
let resp = Value::Array(vec![
header(&["i", "j"]),
Value::Array(vec![
row(vec![Value::Int(1), Value::BulkString(b"not-a-number".to_vec())]),
row(vec![Value::Int(2)]),
]),
]);
let err = response_to_batch(resp, &schema).unwrap_err();
assert!(
matches!(&err, FlightSqlError::MalformedResponse(m) if m.starts_with("row 0:")),
"row-0 cell error must win over the row-1 raggedness, got {err:?}"
);
// Auto path: the column sniffs Int64 from row 0, the row-1 string cell
// is the coercion error, and it must beat the ragged row 2.
let resp_auto = Value::Array(vec![
header(&["i", "j"]),
Value::Array(vec![
row(vec![Value::Int(1), Value::Int(1)]),
row(vec![Value::Int(2), Value::BulkString(b"nope".to_vec())]),
row(vec![Value::Int(3)]),
]),
]);
let auto_err = response_to_batch_auto(resp_auto).unwrap_err();
assert!(
matches!(&auto_err, FlightSqlError::MalformedResponse(m) if m.starts_with("row 1:")),
"auto path: the row-1 cell error must win over the row-2 raggedness, got {auto_err:?}"
);
// Ragged row 0 beats the bad cell in row 1.
let resp = Value::Array(vec![
header(&["i", "j"]),
Value::Array(vec![
row(vec![Value::Int(1)]),
row(vec![Value::Int(2), Value::BulkString(b"nope".to_vec())]),
]),
]);
let err = response_to_batch(resp.clone(), &schema).unwrap_err();
assert!(
matches!(err, FlightSqlError::ColumnCountMismatch { schema: 2, row: 1 }),
"the row-0 width mismatch must win"
);
let auto_err = response_to_batch_auto(resp).unwrap_err();
assert!(
matches!(auto_err, FlightSqlError::ColumnCountMismatch { schema: 2, row: 1 }),
"auto path likewise"
);
}
/// The same Utf8 fallback drives cell-type sniffing directly: a non-scalar
/// `Value` fixes a column to `Utf8`, while `Nil` leaves it undetermined.
#[test]
fn cell_type_maps_nested_values_to_utf8() {
assert_eq!(cell_type(&Value::Int(1)), Some(DataType::Int64));
assert_eq!(cell_type(&Value::Double(1.0)), Some(DataType::Float64));
assert_eq!(cell_type(&Value::Boolean(true)), Some(DataType::Boolean));
assert_eq!(cell_type(&Value::Nil), None);
// Nested Cypher shapes → Utf8.
assert_eq!(cell_type(&Value::Map(vec![])), Some(DataType::Utf8));
assert_eq!(cell_type(&Value::Array(vec![])), Some(DataType::Utf8));
assert_eq!(
cell_type(&Value::BulkString(b"x".to_vec())),
Some(DataType::Utf8)
);
}
}