skade-flight/src/columns.rs
//! Column metadata the Flight SQL ODBC driver reads off every schema field.
//!
//! The driver (dremio lineage and the Apache donation alike) does
//! `field->metadata()->Get("ARROW:FLIGHT:SQL:TYPE_NAME")` WITHOUT a null
//! check on the fields of `GetTables(include_schema)`, and Power Query
//! matches that `TYPE_NAME` against the `type_name` column of
//! `GetXdbcTypeInfo` to decide whether a column is searchable. A field with no
//! metadata crashes the driver; a field whose name is not in the type table
//! is UNSEARCHABLE and nothing folds on it (GizmoSQL changelog, measured
//! against Power BI). So every schema this server hands out — table
//! schemas, `FlightInfo.schema`, prepared `dataset_schema` — goes through
//! [`decorate`].
//!
//! Keys are the ones FlightSql.proto names for column metadata.
use std::collections::HashMap;
use std::sync::Arc;
use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
pub const TYPE_NAME: &str = "ARROW:FLIGHT:SQL:TYPE_NAME";
pub const PRECISION: &str = "ARROW:FLIGHT:SQL:PRECISION";
pub const SCALE: &str = "ARROW:FLIGHT:SQL:SCALE";
pub const IS_SEARCHABLE: &str = "ARROW:FLIGHT:SQL:IS_SEARCHABLE";
pub const IS_CASE_SENSITIVE: &str = "ARROW:FLIGHT:SQL:IS_CASE_SENSITIVE";
pub const IS_AUTO_INCREMENT: &str = "ARROW:FLIGHT:SQL:IS_AUTO_INCREMENT";
pub const IS_READ_ONLY: &str = "ARROW:FLIGHT:SQL:IS_READ_ONLY";
pub const CATALOG_NAME: &str = "ARROW:FLIGHT:SQL:CATALOG_NAME";
pub const SCHEMA_NAME: &str = "ARROW:FLIGHT:SQL:SCHEMA_NAME";
pub const TABLE_NAME: &str = "ARROW:FLIGHT:SQL:TABLE_NAME";
/// The VARCHAR length advertised for strings. The driver's own default
/// (`StringColumnLength`) is 1024; Power BI truncates to what is advertised.
pub const VARCHAR_PRECISION: i32 = 32_767;
/// `(type_name, precision, scale, case_sensitive)` — the ODBC view of an
/// Arrow type. `type_name` MUST be a `type_name` row of `sqlinfo::XDBC_TYPE_INFO`.
pub fn xdbc_type(dt: &DataType) -> (&'static str, i32, i32, bool) {
match dt {
DataType::Boolean => ("BOOLEAN", 1, 0, false),
DataType::Int8 | DataType::UInt8 => ("TINYINT", 3, 0, false),
DataType::Int16 | DataType::UInt16 => ("SMALLINT", 5, 0, false),
DataType::Int32 | DataType::UInt32 => ("INTEGER", 10, 0, false),
DataType::Int64 | DataType::UInt64 => ("BIGINT", 19, 0, false),
DataType::Float16 | DataType::Float32 => ("FLOAT", 7, 0, false),
DataType::Float64 => ("DOUBLE", 15, 0, false),
DataType::Decimal128(p, s) | DataType::Decimal256(p, s) => ("DECIMAL", *p as i32, *s as i32, false),
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => ("VARCHAR", VARCHAR_PRECISION, 0, true),
DataType::Binary | DataType::LargeBinary | DataType::BinaryView | DataType::FixedSizeBinary(_) => {
("VARBINARY", VARCHAR_PRECISION, 0, false)
}
DataType::Date32 | DataType::Date64 => ("DATE", 10, 0, false),
DataType::Time32(_) | DataType::Time64(_) => ("TIME", 15, 6, false),
DataType::Timestamp(unit, _) => {
let scale = match unit {
TimeUnit::Second => 0,
TimeUnit::Millisecond => 3,
TimeUnit::Microsecond => 6,
TimeUnit::Nanosecond => 9,
};
("TIMESTAMP", 29, scale, false)
}
// Everything else the driver renders as JSON text.
_ => ("VARCHAR", VARCHAR_PRECISION, 0, true),
}
}
/// Add the Flight SQL column metadata to every field. Existing metadata keys
/// are kept; ours win on collision. `table` names the owning table for the
/// CATALOG/SCHEMA/TABLE keys when known.
pub fn decorate(schema: &Schema, table: Option<(&str, &str, &str)>) -> SchemaRef {
let fields: Vec<Field> = schema
.fields()
.iter()
.map(|f| {
let (name, precision, scale, case_sensitive) = xdbc_type(f.data_type());
let mut md: HashMap<String, String> = f.metadata().clone();
md.insert(TYPE_NAME.into(), name.into());
md.insert(PRECISION.into(), precision.to_string());
md.insert(SCALE.into(), scale.to_string());
md.insert(IS_SEARCHABLE.into(), "1".into());
md.insert(IS_CASE_SENSITIVE.into(), if case_sensitive { "1" } else { "0" }.into());
md.insert(IS_AUTO_INCREMENT.into(), "0".into());
md.insert(IS_READ_ONLY.into(), "1".into());
if let Some((c, s, t)) = table {
md.insert(CATALOG_NAME.into(), c.into());
md.insert(SCHEMA_NAME.into(), s.into());
md.insert(TABLE_NAME.into(), t.into());
}
f.as_ref().clone().with_metadata(md)
})
.collect();
Arc::new(Schema::new_with_metadata(fields, schema.metadata().clone()))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::sqlinfo::XDBC_TYPE_INFO;
use arrow_array::{Array, StringArray};
#[test]
fn every_type_name_is_in_the_xdbc_table() {
let batch = XDBC_TYPE_INFO.record_batch(None).unwrap();
let names = batch.column_by_name("type_name").unwrap();
let names = names.as_any().downcast_ref::<StringArray>().unwrap();
let known: Vec<&str> = (0..names.len()).map(|i| names.value(i)).collect();
for dt in [
DataType::Boolean,
DataType::Int8,
DataType::Int16,
DataType::Int32,
DataType::Int64,
DataType::Float32,
DataType::Float64,
DataType::Decimal128(18, 3),
DataType::Utf8,
DataType::Binary,
DataType::Date32,
DataType::Time64(TimeUnit::Microsecond),
DataType::Timestamp(TimeUnit::Microsecond, None),
DataType::List(Arc::new(Field::new("x", DataType::Int32, true))),
] {
let (n, _, _, _) = xdbc_type(&dt);
assert!(known.contains(&n), "{dt:?} → {n} not in GetXdbcTypeInfo");
}
}
#[test]
fn decorate_sets_every_key() {
let s = Schema::new(vec![Field::new("a", DataType::Int64, false), Field::new("b", DataType::Utf8, true)]);
let d = decorate(&s, Some(("skade", "main", "t")));
for f in d.fields() {
for k in [TYPE_NAME, PRECISION, SCALE, IS_SEARCHABLE, IS_CASE_SENSITIVE, IS_AUTO_INCREMENT, IS_READ_ONLY, CATALOG_NAME, SCHEMA_NAME, TABLE_NAME] {
assert!(f.metadata().contains_key(k), "{} lacks {k}", f.name());
}
}
assert_eq!(d.field(1).metadata()[TYPE_NAME], "VARCHAR");
assert_eq!(d.field(0).metadata()[PRECISION], "19");
}
}