skade-flight/src/sqlinfo.rs
//! What this server tells a generic driver about itself.
//!
//! `CommandGetSqlInfo` mirrors ODBC's `SQLGetInfo`; `CommandGetXdbcTypeInfo`
//! mirrors `SQLGetTypeInfo`. The Flight SQL ODBC driver and Power BI's
//! connector answer their own capability questions from these two tables and
//! only fold the SQL they say we support. So they are generated here from the
//! typed builders — a finite, checkable declaration — and they declare what
//! DataFusion 55 actually plans, no more.
//!
//! Bitmask infos use `1 << ordinal` per the proto; ordinal infos carry the
//! enum value.
use std::collections::BTreeMap;
use std::sync::LazyLock;
use arrow_flight::sql::metadata::{SqlInfoData, SqlInfoDataBuilder, XdbcTypeInfo, XdbcTypeInfoData, XdbcTypeInfoDataBuilder};
use arrow_flight::sql::{
Nullable, Searchable, SqlInfo, SqlNullOrdering, SqlOuterJoinsSupportLevel, SqlSupportedCaseSensitivity,
SqlSupportedGroupBy, SqlSupportedResultSetConcurrency, SqlSupportedResultSetType, SqlSupportedSubqueries,
SqlSupportedTransaction, SqlSupportedUnions, SqlSupportsConvert, SupportedSqlGrammar, XdbcDataType, XdbcDatetimeSubcode,
};
pub const SERVER_NAME: &str = "skade-flight (DataFusion over skade Iceberg)";
/// Arrow columnar format version the IPC writer emits.
pub const ARROW_FORMAT_VERSION: &str = "1.5";
fn bits<T: Copy + Into<i32>>(items: &[T]) -> i32 {
items.iter().fold(0i32, |m, v| m | (1i32 << (*v).into()))
}
/// Scalar functions, by ODBC family, under their ODBC NAMES. The Flight SQL
/// ODBC driver turns these lists into `SQLGetInfo` bitmasks by looking each
/// name up in its own tables (`scalar_function_reporter.cc`: UCASE, LOCATE,
/// DAYOFMONTH, … — never `upper`), and Power Query folds only the functions
/// the bitmask says exist, emitting them as `{fn UCASE(x)}` escapes. odbc.rs
/// translates every name listed here to DataFusion. Names the driver would
/// not recognise are omitted; TIMESTAMPDIFF is listed because odbc.rs
/// implements it for every SQL_TSI unit.
pub const NUMERIC_FUNCTIONS: &[&str] = &[
"ABS", "ACOS", "ASIN", "ATAN", "ATAN2", "CEILING", "COS", "COT", "DEGREES", "EXP", "FLOOR", "LOG", "LOG10",
"MOD", "PI", "POWER", "RADIANS", "RAND", "ROUND", "SIGN", "SIN", "SQRT", "TAN", "TRUNCATE",
];
pub const STRING_FUNCTIONS: &[&str] = &[
"ASCII", "CHAR", "CHAR_LENGTH", "CHARACTER_LENGTH", "CONCAT", "INSERT", "LCASE", "LEFT", "LENGTH", "LOCATE",
"LTRIM", "OCTET_LENGTH", "REPEAT", "REPLACE", "RIGHT", "RTRIM", "SPACE", "SUBSTRING", "UCASE",
];
pub const DATETIME_FUNCTIONS: &[&str] = &[
"CURDATE", "CURTIME", "CURRENT_DATE", "CURRENT_TIME", "CURRENT_TIMESTAMP", "DAYNAME", "DAYOFMONTH",
"DAYOFWEEK", "DAYOFYEAR", "EXTRACT", "HOUR", "MINUTE", "MONTH", "MONTHNAME", "NOW", "QUARTER", "SECOND",
"TIMESTAMPADD", "TIMESTAMPDIFF", "WEEK", "YEAR",
];
/// No "CONVERT" here on purpose: with it the driver reports
/// SQL_FN_CVT_CONVERT and Power Query prefers `{fn convert(x, SQL_T)}`;
/// without it, plain `CAST`. Both work (odbc.rs), CAST is the cleaner wire.
pub const SYSTEM_FUNCTIONS: &[&str] = &["DATABASE", "IFNULL", "USER"];
/// The `GetSqlInfo` table. Built once.
pub static SQL_INFO: LazyLock<SqlInfoData> = LazyLock::new(|| {
let mut b = SqlInfoDataBuilder::new();
// ── server ──────────────────────────────────────────────────────────
b.append(SqlInfo::FlightSqlServerName, SERVER_NAME);
b.append(SqlInfo::FlightSqlServerVersion, env!("CARGO_PKG_VERSION"));
b.append(SqlInfo::FlightSqlServerArrowVersion, ARROW_FORMAT_VERSION);
b.append(SqlInfo::FlightSqlServerReadOnly, true);
b.append(SqlInfo::FlightSqlServerSql, true);
b.append(SqlInfo::FlightSqlServerSubstrait, false);
b.append(SqlInfo::FlightSqlServerTransaction, SqlSupportedTransaction::None as i32);
b.append(SqlInfo::FlightSqlServerCancel, false);
b.append(SqlInfo::FlightSqlServerBulkIngestion, false);
b.append(SqlInfo::FlightSqlServerIngestTransactionsSupported, false);
b.append(SqlInfo::FlightSqlServerStatementTimeout, 0i32);
b.append(SqlInfo::FlightSqlServerTransactionTimeout, 0i32);
// ── DDL: none, read-only ────────────────────────────────────────────
b.append(SqlInfo::SqlDdlCatalog, false);
b.append(SqlInfo::SqlDdlSchema, false);
b.append(SqlInfo::SqlDdlTable, false);
// ── identifiers: DataFusion lowercases unquoted, preserves quoted ───
b.append(SqlInfo::SqlIdentifierCase, SqlSupportedCaseSensitivity::SqlCaseSensitivityLowercase as i32);
b.append(SqlInfo::SqlIdentifierQuoteChar, "\"");
b.append(SqlInfo::SqlQuotedIdentifierCase, SqlSupportedCaseSensitivity::SqlCaseSensitivityUnknown as i32);
b.append(SqlInfo::SqlAllTablesAreSelectable, true);
b.append(SqlInfo::SqlNullOrdering, SqlNullOrdering::SqlNullsSortedHigh as i32);
b.append(SqlInfo::SqlNumericFunctions, NUMERIC_FUNCTIONS);
b.append(SqlInfo::SqlStringFunctions, STRING_FUNCTIONS);
b.append(SqlInfo::SqlSystemFunctions, SYSTEM_FUNCTIONS);
b.append(SqlInfo::SqlDatetimeFunctions, DATETIME_FUNCTIONS);
b.append(SqlInfo::SqlSearchStringEscape, "\\");
b.append(SqlInfo::SqlExtraNameCharacters, "");
b.append(SqlInfo::SqlSupportsColumnAliasing, true);
b.append(SqlInfo::SqlNullPlusNullIsNull, true);
b.append(SqlInfo::SqlSupportsConvert, convert_matrix());
b.append(SqlInfo::SqlSupportsTableCorrelationNames, true);
b.append(SqlInfo::SqlSupportsDifferentTableCorrelationNames, true);
b.append(SqlInfo::SqlSupportsExpressionsInOrderBy, true);
b.append(SqlInfo::SqlSupportsOrderByUnrelated, true);
// ⚠ ONE bit each for 522/525/526. The ODBC driver tests the LOWEST bit
// first (`if (v & 1<<ENTRY) … else if INTERMEDIATE … else if FULL`), so a
// mask with several bits reports the LEAST capable level. GROUP BY
// UNRELATED (columns not in the SELECT) is the capable answer and the
// one DataFusion supports; it implies BEYOND_SELECT.
b.append(SqlInfo::SqlSupportedGroupBy, bits(&[SqlSupportedGroupBy::SqlGroupByUnrelated]));
b.append(SqlInfo::SqlSupportsLikeEscapeClause, true);
b.append(SqlInfo::SqlSupportsNonNullableColumns, true);
// EXTENDED only (→ SQL_OAC_LEVEL2); see the bit-order note above.
b.append(SqlInfo::SqlSupportedGrammar, bits(&[SupportedSqlGrammar::SqlExtendedGrammar]));
// `SupportedAnsi92SqlGrammarLevel` is not re-exported by arrow-flight 59; the
// ordinals are ENTRY = 0, INTERMEDIATE = 1, FULL = 2 (FlightSql.proto).
// FULL only (→ SQL_SC_SQL92_FULL): Power Query assumes derived-table
// support only at FULL, and DataFusion has it. See the bit-order note.
b.append(SqlInfo::SqlAnsi92SupportedLevel, bits(&[2i32]));
b.append(SqlInfo::SqlSupportsIntegrityEnhancementFacility, false);
b.append(SqlInfo::SqlOuterJoinsSupportLevel, SqlOuterJoinsSupportLevel::SqlFullOuterJoins as i32);
b.append(SqlInfo::SqlSchemaTerm, "schema");
b.append(SqlInfo::SqlProcedureTerm, "procedure");
b.append(SqlInfo::SqlCatalogTerm, "catalog");
b.append(SqlInfo::SqlCatalogAtStart, true);
b.append(SqlInfo::SqlSchemasSupportedActions, 0i32);
b.append(SqlInfo::SqlCatalogsSupportedActions, 0i32);
b.append(SqlInfo::SqlSupportedPositionedCommands, 0i32);
b.append(SqlInfo::SqlSelectForUpdateSupported, false);
b.append(SqlInfo::SqlStoredProceduresSupported, false);
b.append(
SqlInfo::SqlSupportedSubqueries,
bits(&[
SqlSupportedSubqueries::SqlSubqueriesInComparisons,
SqlSupportedSubqueries::SqlSubqueriesInExists,
SqlSupportedSubqueries::SqlSubqueriesInIns,
SqlSupportedSubqueries::SqlSubqueriesInQuantifieds,
]),
);
b.append(SqlInfo::SqlCorrelatedSubqueriesSupported, true);
b.append(
SqlInfo::SqlSupportedUnions,
bits(&[SqlSupportedUnions::SqlUnion, SqlSupportedUnions::SqlUnionAll]),
);
// 0 = no limit / unknown, for every SQL_MAX_*.
for id in [
SqlInfo::SqlMaxBinaryLiteralLength,
SqlInfo::SqlMaxCharLiteralLength,
SqlInfo::SqlMaxColumnNameLength,
SqlInfo::SqlMaxColumnsInGroupBy,
SqlInfo::SqlMaxColumnsInIndex,
SqlInfo::SqlMaxColumnsInOrderBy,
SqlInfo::SqlMaxColumnsInSelect,
SqlInfo::SqlMaxColumnsInTable,
SqlInfo::SqlMaxConnections,
SqlInfo::SqlMaxCursorNameLength,
SqlInfo::SqlMaxIndexLength,
SqlInfo::SqlDbSchemaNameLength,
SqlInfo::SqlMaxProcedureNameLength,
SqlInfo::SqlMaxCatalogNameLength,
SqlInfo::SqlMaxRowSize,
SqlInfo::SqlMaxStatementLength,
SqlInfo::SqlMaxStatements,
SqlInfo::SqlMaxTableNameLength,
SqlInfo::SqlMaxTablesInSelect,
SqlInfo::SqlMaxUsernameLength,
] {
b.append(id, 0i64);
}
b.append(SqlInfo::SqlMaxRowSizeIncludesBlobs, true);
// ── transactions: none ──────────────────────────────────────────────
b.append(SqlInfo::SqlDefaultTransactionIsolation, 0i64);
b.append(SqlInfo::SqlTransactionsSupported, false);
b.append(SqlInfo::SqlSupportedTransactionsIsolationLevels, 0i32);
b.append(SqlInfo::SqlDataDefinitionCausesTransactionCommit, false);
b.append(SqlInfo::SqlDataDefinitionsInTransactionsIgnored, false);
// ── result sets: forward-only, read-only ────────────────────────────
b.append(
SqlInfo::SqlSupportedResultSetTypes,
bits(&[SqlSupportedResultSetType::SqlResultSetTypeForwardOnly]),
);
b.append(SqlInfo::SqlSupportedConcurrenciesForResultSetUnspecified, 0i32);
b.append(
SqlInfo::SqlSupportedConcurrenciesForResultSetForwardOnly,
bits(&[SqlSupportedResultSetConcurrency::SqlResultSetConcurrencyReadOnly]),
);
b.append(SqlInfo::SqlSupportedConcurrenciesForResultSetScrollSensitive, 0i32);
b.append(SqlInfo::SqlSupportedConcurrenciesForResultSetScrollInsensitive, 0i32);
b.append(SqlInfo::SqlBatchUpdatesSupported, false);
b.append(SqlInfo::SqlSavepointsSupported, false);
b.append(SqlInfo::SqlNamedParametersSupported, false);
b.append(SqlInfo::SqlLocatorsUpdateCopy, false);
b.append(SqlInfo::SqlStoredFunctionsUsingCallSyntaxSupported, false);
b.build().expect("static SqlInfoData builds")
});
/// `CAST` matrix: DataFusion casts freely between numerics and strings, and
/// between strings and temporals.
fn convert_matrix() -> BTreeMap<i32, Vec<i32>> {
use SqlSupportsConvert as C;
let numeric = [
C::SqlConvertBigint, C::SqlConvertInteger, C::SqlConvertSmallint, C::SqlConvertTinyint,
C::SqlConvertDecimal, C::SqlConvertNumeric, C::SqlConvertFloat, C::SqlConvertReal,
C::SqlConvertBit,
];
let text = [C::SqlConvertChar, C::SqlConvertVarchar, C::SqlConvertLongvarchar];
let temporal = [C::SqlConvertDate, C::SqlConvertTime, C::SqlConvertTimestamp];
let ids = |xs: &[C]| xs.iter().map(|c| *c as i32).collect::<Vec<_>>();
let mut m = BTreeMap::new();
for src in numeric {
let mut to = ids(&numeric);
to.extend(ids(&text));
m.insert(src as i32, to);
}
for src in text {
let mut to = ids(&numeric);
to.extend(ids(&text));
to.extend(ids(&temporal));
m.insert(src as i32, to);
}
for src in temporal {
let mut to = ids(&text);
to.extend(ids(&temporal));
if src == C::SqlConvertTimestamp {
to.push(C::SqlConvertBigint as i32);
}
m.insert(src as i32, to);
}
m
}
/// The `GetXdbcTypeInfo` table: the Arrow types an Iceberg table can carry,
/// as ODBC sees them. Built once.
pub static XDBC_TYPE_INFO: LazyLock<XdbcTypeInfoData> = LazyLock::new(|| {
let mut b = XdbcTypeInfoDataBuilder::new();
let base = |name: &str, dt: XdbcDataType, size: i32| XdbcTypeInfo {
type_name: name.to_string(),
data_type: dt,
column_size: Some(size),
literal_prefix: None,
literal_suffix: None,
// Stays None: XdbcTypeInfoDataBuilder writes a null for Some(vec![]) too,
// so an empty list cannot be expressed here. create_params_never_null()
// turns the nulls into empty lists after the batch is built.
create_params: None,
nullable: Nullable::NullabilityNullable,
case_sensitive: false,
searchable: Searchable::Full,
unsigned_attribute: Some(false),
fixed_prec_scale: false,
auto_increment: Some(false),
local_type_name: Some(name.to_string()),
minimum_scale: None,
maximum_scale: None,
sql_data_type: dt,
datetime_subcode: None,
num_prec_radix: Some(10),
interval_precision: None,
};
b.append(XdbcTypeInfo { fixed_prec_scale: true, ..base("BOOLEAN", XdbcDataType::XdbcBit, 1) });
b.append(XdbcTypeInfo { num_prec_radix: Some(2), ..base("TINYINT", XdbcDataType::XdbcTinyint, 8) });
b.append(XdbcTypeInfo { num_prec_radix: Some(2), ..base("SMALLINT", XdbcDataType::XdbcSmallint, 16) });
b.append(XdbcTypeInfo { num_prec_radix: Some(2), ..base("INTEGER", XdbcDataType::XdbcInteger, 32) });
b.append(XdbcTypeInfo { num_prec_radix: Some(2), ..base("BIGINT", XdbcDataType::XdbcBigint, 64) });
b.append(XdbcTypeInfo { num_prec_radix: Some(2), ..base("FLOAT", XdbcDataType::XdbcReal, 24) });
b.append(XdbcTypeInfo { num_prec_radix: Some(2), ..base("DOUBLE", XdbcDataType::XdbcDouble, 53) });
b.append(XdbcTypeInfo {
create_params: Some(vec!["precision".into(), "scale".into()]),
fixed_prec_scale: true,
minimum_scale: Some(0),
maximum_scale: Some(38),
..base("DECIMAL", XdbcDataType::XdbcDecimal, 38)
});
b.append(XdbcTypeInfo {
literal_prefix: Some("'".into()),
literal_suffix: Some("'".into()),
case_sensitive: true,
unsigned_attribute: None,
auto_increment: None,
num_prec_radix: None,
..base("VARCHAR", XdbcDataType::XdbcVarchar, i32::MAX)
});
b.append(XdbcTypeInfo {
literal_prefix: Some("X'".into()),
literal_suffix: Some("'".into()),
unsigned_attribute: None,
auto_increment: None,
num_prec_radix: None,
searchable: Searchable::None,
..base("VARBINARY", XdbcDataType::XdbcVarbinary, i32::MAX)
});
b.append(XdbcTypeInfo {
literal_prefix: Some("DATE '".into()),
literal_suffix: Some("'".into()),
unsigned_attribute: None,
auto_increment: None,
num_prec_radix: None,
// ODBC SQL_CODE_DATE = 1; the proto names ordinal 1 after the interval code.
datetime_subcode: Some(XdbcDatetimeSubcode::XdbcSubcodeYear),
..base("DATE", XdbcDataType::XdbcDate, 10)
});
b.append(XdbcTypeInfo {
literal_prefix: Some("TIME '".into()),
literal_suffix: Some("'".into()),
unsigned_attribute: None,
auto_increment: None,
num_prec_radix: None,
datetime_subcode: Some(XdbcDatetimeSubcode::XdbcSubcodeTime),
..base("TIME", XdbcDataType::XdbcTime, 15)
});
b.append(XdbcTypeInfo {
literal_prefix: Some("TIMESTAMP '".into()),
literal_suffix: Some("'".into()),
unsigned_attribute: None,
auto_increment: None,
num_prec_radix: None,
datetime_subcode: Some(XdbcDatetimeSubcode::XdbcSubcodeTimestamp),
..base("TIMESTAMP", XdbcDataType::XdbcTimestamp, 29)
});
b.build().expect("static XdbcTypeInfoData builds")
});
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn tables_build_and_are_nonempty() {
let batch = SQL_INFO.record_batch(None).unwrap();
assert!(batch.num_rows() > 60, "{}", batch.num_rows());
let types = XDBC_TYPE_INFO.record_batch(None).unwrap();
assert_eq!(types.num_rows(), 13);
}
#[test]
fn create_params_is_an_empty_list_never_a_null() {
use arrow_array::Array;
use arrow_array::cast::AsArray;
// The raw builder cannot say "empty list", so the unrepaired batch is null-heavy.
let raw = XDBC_TYPE_INFO.record_batch(None).unwrap();
let (i, _) = raw.schema().column_with_name("create_params").unwrap();
assert!(raw.column(i).null_count() > 0, "the bug this guards is gone from arrow-flight");
// What we actually serve: no nulls, and only DECIMAL carries parameters.
let batch = &*XDBC_TYPE_INFO_BATCH;
let params = batch.column(i).as_list::<i32>();
assert_eq!(params.null_count(), 0, "Power BI calls Enumerable.Select on this column");
let named: Vec<_> = (0..batch.num_rows()).filter(|&r| !params.value(r).is_empty()).collect();
assert_eq!(named.len(), 1, "only DECIMAL declares precision and scale");
}
#[test]
fn bitmask_is_one_shl_ordinal() {
assert_eq!(bits(&[SqlSupportedGroupBy::SqlGroupByUnrelated, SqlSupportedGroupBy::SqlGroupByBeyondSelect]), 3);
assert_eq!(bits(&[SqlSupportedResultSetType::SqlResultSetTypeForwardOnly]), 2);
}
}
/// The full `GetSqlInfo` batch (no id filter), built once: the common call
/// shape from a driver is "everything", and it costs nothing per call.
pub static SQL_INFO_BATCH: LazyLock<arrow_array::RecordBatch> =
LazyLock::new(|| SQL_INFO.record_batch(None).expect("static sql_info batch"));
/// Rewrite `create_params` so a type with no creation parameters carries an
/// empty list instead of a null.
///
/// Power BI's ADBC layer reads the column straight into
/// `Enumerable.Select(source, ..)`, which throws `ArgumentNullException` on a
/// null — `AdbcEngineConnection.GetTypeInfo` dies and the navigator cannot open
/// a table. Only DECIMAL declares parameters here, so 12 of the 13 rows were
/// null and any table whose first column was not a decimal hit it.
///
/// arrow-flight's `XdbcTypeInfoDataBuilder` cannot emit an empty list: it calls
/// `append_null()` for `None` *and* for `Some(vec![])`. So the batch is repaired
/// afterwards. A null list slot already has equal start and end offsets, which
/// is exactly an empty list, so dropping the validity buffer is the whole fix —
/// no offsets move and no values are touched.
pub fn create_params_never_null(batch: arrow_array::RecordBatch) -> arrow_array::RecordBatch {
use arrow_array::Array;
use arrow_array::cast::AsArray;
let Some((idx, _)) = batch.schema().column_with_name("create_params") else {
return batch;
};
let list = batch.column(idx).as_list::<i32>();
if list.null_count() == 0 {
return batch;
}
let (field, offsets, values, _nulls) = list.clone().into_parts();
let mut columns = batch.columns().to_vec();
columns[idx] = std::sync::Arc::new(arrow_array::ListArray::new(field, offsets, values, None));
arrow_array::RecordBatch::try_new(batch.schema(), columns).expect("create_params keeps its type")
}
/// The full `GetXdbcTypeInfo` batch (no type filter), built once.
pub static XDBC_TYPE_INFO_BATCH: LazyLock<arrow_array::RecordBatch> = LazyLock::new(|| {
create_params_never_null(XDBC_TYPE_INFO.record_batch(None).expect("static xdbc batch"))
});
/// A metadata answer pre-encoded as Flight frames: built once, every reply
/// is refcounted `Bytes` clones of the schema + data messages. Zero encode,
/// zero copy per call.
pub struct StaticFrames {
frames: Vec<arrow_flight::FlightData>,
}
impl StaticFrames {
pub fn encode(batch: &arrow_array::RecordBatch) -> Self {
use futures::StreamExt;
let stream = arrow_flight::encode::FlightDataEncoderBuilder::new()
.with_schema(batch.schema())
.build(futures::stream::iter([Ok(batch.clone())]));
// Encoding a batch we hold is infallible in practice; a failure here
// would be a bug in the static tables and should surface at startup.
let frames = futures::executor::block_on(stream.collect::<Vec<_>>())
.into_iter()
.collect::<Result<Vec<_>, _>>()
.expect("static metadata encodes");
Self { frames }
}
/// The frames as a ready stream.
pub fn stream(
&self,
) -> std::pin::Pin<Box<dyn futures::Stream<Item = Result<arrow_flight::FlightData, tonic::Status>> + Send + 'static>>
{
Box::pin(futures::stream::iter(self.frames.clone().into_iter().map(Ok)))
}
}
pub static SQL_INFO_FRAMES: LazyLock<StaticFrames> = LazyLock::new(|| StaticFrames::encode(&SQL_INFO_BATCH));
pub static XDBC_TYPE_INFO_FRAMES: LazyLock<StaticFrames> = LazyLock::new(|| StaticFrames::encode(&XDBC_TYPE_INFO_BATCH));
/// `GetTableTypes` and `GetCatalogs` never change for this server.
pub static TABLE_TYPES_FRAMES: LazyLock<StaticFrames> = LazyLock::new(|| {
let mut b = arrow_flight::sql::CommandGetTableTypes::default().into_builder();
b.append("TABLE");
StaticFrames::encode(&b.build().expect("table types batch"))
});
pub static CATALOGS_FRAMES: LazyLock<StaticFrames> = LazyLock::new(|| {
let mut b = arrow_flight::sql::CommandGetCatalogs::default().into_builder();
b.append(crate::engine::CATALOG);
StaticFrames::encode(&b.build().expect("catalogs batch"))
});