skade-flight/src/time_literal.rs
//! **A `TIME '…'` literal is planned in nanoseconds; every TIME column here is
//! microseconds.** This rule lowers the literal, so the comparison plans.
//!
//! MEASURED (datatype matrix, 2026-09-10, before this rule):
//!
//! ```text
//! t = TIME '12:34:56' → Internal: datafusion: type_coercion
//! caused by Error during planning: Cannot infer common argument type for
//! comparison operation Time64(µs) = Time64(ns)
//! ```
//!
//! DataFusion 55's SQL planner maps the `TIME` type — and therefore the typed
//! string `TIME '12:34:56'`, which plans as `CAST('12:34:56' AS TIME)` — to
//! `Time64(Nanosecond)`; `TIME(6)` is not supported at all. Its type coercion
//! then refuses `Time64(µs) = Time64(ns)` rather than widen either side. The
//! Iceberg spec fixes `time` at microseconds, and this engine reads Iceberg
//! only, so on this door a TIME literal can only ever be meant against a
//! microsecond column. The driver-side workaround the matrix pinned for a
//! week — `CAST(... AS ...)` or `to_char` — put DataFusion's unit on the user.
//!
//! What is lowered, and what is left alone:
//!
//! * `CAST(<anything> AS Time64(ns))` — the literal's shape — becomes
//! `CAST(… AS Time64(µs))`. A string literal survives the cast either way.
//! * a folded `Time64Nanosecond(v)` scalar becomes `Time64Microsecond(v/1000)`
//! **only when `v` is a whole number of microseconds**. A sub-microsecond
//! literal is left as it was, so the coercion error still fires for a value
//! no Iceberg column could hold — lowering it silently would compare against
//! a different instant than the one written.
//!
//! Runs BEFORE `TypeCoercion` (registered first on the session, and first in
//! the view rule set), because coercion is where the refusal lives.
use std::sync::Arc;
use datafusion::arrow::datatypes::{DataType, FieldRef, TimeUnit};
use datafusion::common::config::ConfigOptions;
use datafusion::common::tree_node::{Transformed, TreeNode};
use datafusion::common::{Result, ScalarValue};
use datafusion::logical_expr::{Cast, Expr, LogicalPlan, TryCast};
use datafusion::optimizer::AnalyzerRule;
/// The rule. Unit struct: it carries no state.
#[derive(Debug, Default)]
pub struct TimeLiteralToMicros;
impl AnalyzerRule for TimeLiteralToMicros {
fn analyze(&self, plan: LogicalPlan, _config: &ConfigOptions) -> Result<LogicalPlan> {
plan.transform_up_with_subqueries(|p| p.map_expressions(|e| e.transform_up(lower)))
.map(|t| t.data)
}
fn name(&self) -> &str {
"skade_flight_time_literal_to_micros"
}
}
/// One expression: the `Time64(ns)` shapes a literal takes, lowered to `µs`.
fn lower(e: Expr) -> Result<Transformed<Expr>> {
let is_ns = |f: &FieldRef| matches!(f.data_type(), DataType::Time64(TimeUnit::Nanosecond));
let to_us = |f: &FieldRef| -> FieldRef { Arc::new(f.as_ref().clone().with_data_type(DataType::Time64(TimeUnit::Microsecond))) };
Ok(match e {
Expr::Cast(Cast { expr, field }) if is_ns(&field) => {
Transformed::yes(Expr::Cast(Cast { expr, field: to_us(&field) }))
}
Expr::TryCast(TryCast { expr, field }) if is_ns(&field) => {
Transformed::yes(Expr::TryCast(TryCast { expr, field: to_us(&field) }))
}
Expr::Literal(ScalarValue::Time64Nanosecond(Some(ns)), meta) if ns % 1000 == 0 => {
Transformed::yes(Expr::Literal(ScalarValue::Time64Microsecond(Some(ns / 1000)), meta))
}
Expr::Literal(ScalarValue::Time64Nanosecond(None), meta) => {
Transformed::yes(Expr::Literal(ScalarValue::Time64Microsecond(None), meta))
}
other => Transformed::no(other),
})
}
/// The rule as the session and the view builder register it.
pub fn rule() -> Arc<dyn AnalyzerRule + Send + Sync> {
Arc::new(TimeLiteralToMicros)
}
/// **The analyzer rule set every session here runs: this rule FIRST, then
/// DataFusion's own.** Order is the point — `SessionContext::add_analyzer_rule`
/// appends after `TypeCoercion`, which is exactly too late, because coercion
/// is where `Time64(µs) = Time64(ns)` is refused.
pub fn analyzer_rules() -> Vec<Arc<dyn AnalyzerRule + Send + Sync>> {
let mut rules = datafusion::optimizer::Analyzer::new().rules;
rules.insert(0, rule());
rules
}
/// A `SessionContext` over `config` with [`analyzer_rules`] in place — the ONE
/// constructor the engine and the tests build sessions through.
pub fn session_context(config: datafusion::prelude::SessionConfig) -> datafusion::prelude::SessionContext {
use datafusion::execution::session_state::SessionStateBuilder;
let state = SessionStateBuilder::new().with_config(config).with_default_features().with_analyzer_rules(analyzer_rules()).build();
datafusion::prelude::SessionContext::new_with_state(state)
}
#[cfg(test)]
mod tests {
use super::*;
use datafusion::logical_expr::lit;
/// **The literal's cast is lowered; a whole-µs scalar is lowered; a
/// sub-µs scalar is NOT.** RED direction: make `lower` return
/// `Transformed::no` for the cast and the first assertion fails; drop the
/// `% 1000` guard and the last one does.
#[test]
fn lowers_exactly_representable_time_literals_and_leaves_the_rest() {
let cast = Expr::Cast(Cast::new(Box::new(lit("12:34:56")), DataType::Time64(TimeUnit::Nanosecond)));
match lower(cast).unwrap().data {
Expr::Cast(Cast { field, .. }) => assert_eq!(field.data_type(), &DataType::Time64(TimeUnit::Microsecond)),
other => panic!("{other:?}"),
}
let whole = Expr::Literal(ScalarValue::Time64Nanosecond(Some(45_296_000_000_000)), None);
let t = lower(whole).unwrap();
assert!(t.transformed);
assert!(matches!(t.data, Expr::Literal(ScalarValue::Time64Microsecond(Some(45_296_000_000)), _)));
let sub_us = Expr::Literal(ScalarValue::Time64Nanosecond(Some(45_296_000_000_123)), None);
let t = lower(sub_us).unwrap();
assert!(!t.transformed, "a value no Iceberg TIME column can hold is not silently lowered");
assert!(matches!(t.data, Expr::Literal(ScalarValue::Time64Nanosecond(Some(45_296_000_000_123)), _)));
let unrelated = lit(1i64);
assert!(!lower(unrelated).unwrap().transformed);
}
/// **Through a session: the comparison the matrix could not plan, plans.**
#[tokio::test]
async fn a_time_predicate_against_a_microsecond_column_plans() {
use datafusion::arrow::array::{Int64Array, RecordBatch, Time64MicrosecondArray};
use datafusion::arrow::datatypes::{Field, Schema};
use datafusion::prelude::SessionContext;
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("t", DataType::Time64(TimeUnit::Microsecond), false),
]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(Int64Array::from(vec![1, 2])), Arc::new(Time64MicrosecondArray::from(vec![45_296_000_000, 1]))],
)
.unwrap();
let sql = "SELECT id FROM x WHERE t = TIME '12:34:56'";
// Without the rule: the measured refusal.
let plain = SessionContext::new();
plain.register_batch("x", batch.clone()).unwrap();
let err = plain.sql(sql).await.unwrap().collect().await.unwrap_err().to_string();
assert!(err.contains("Time64(µs) = Time64(ns)"), "the gap this rule closes: {err}");
// With it — through the one constructor, because ORDER matters: appended
// after TypeCoercion (what `add_analyzer_rule` does) it would never run
// before the refusal.
let ctx = session_context(datafusion::prelude::SessionConfig::new());
ctx.register_batch("x", batch).unwrap();
let rows = ctx.sql(sql).await.unwrap().collect().await.unwrap();
let ids: Vec<i64> = rows.iter().flat_map(|b| b.column(0).as_any().downcast_ref::<Int64Array>().unwrap().values().to_vec()).collect();
assert_eq!(ids, vec![1]);
}
}