krishiv_plan/
expression.rs

1//! Versioned, engine-owned public expression and scalar type contract.
2
3use serde::{Deserialize, Serialize};
4
5use crate::PlanError;
6use krishiv_common::sql_util::quote_identifier;
7
8/// Current serialized public-expression envelope version.
9pub const EXPRESSION_FORMAT_VERSION: u16 = 1;
10
11/// Engine-owned data types used at public and wire boundaries.
12#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
13#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
14pub enum ExprDataType {
15    Null,
16    Boolean,
17    Int64,
18    UInt64,
19    Float64,
20    Utf8,
21    Binary,
22    Decimal128 {
23        precision: u8,
24        scale: i8,
25    },
26    Date32,
27    Timestamp {
28        unit: TimeUnit,
29        timezone: Option<String>,
30    },
31    Interval {
32        unit: IntervalUnit,
33    },
34    List(Box<ExprDataType>),
35    Map {
36        key: Box<ExprDataType>,
37        value: Box<ExprDataType>,
38    },
39    Struct(Vec<ExprField>),
40    /// Semi-structured JSON-like data type (Spark VARIANT equivalent).
41    ///
42    /// Stores arbitrary JSON without a fixed schema. Query-time access uses
43    /// `variant_get(column, 'path')` and schema is applied at read time.
44    /// Arrow serialization uses `Binary` with a variant encoding prefix.
45    Variant,
46}
47
48#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
49pub struct ExprField {
50    pub name: String,
51    pub data_type: ExprDataType,
52    pub nullable: bool,
53}
54
55impl ExprDataType {
56    pub fn validate(&self) -> Result<(), PlanError> {
57        match self {
58            Self::Decimal128 { precision, scale }
59                if *precision == 0 || *precision > 38 || scale.unsigned_abs() > *precision =>
60            {
61                Err(PlanError::Validation(format!(
62                    "invalid decimal({precision}, {scale}); precision must be 1..=38 and cover scale"
63                )))
64            }
65            Self::Timestamp {
66                timezone: Some(timezone),
67                ..
68            } if timezone.trim().is_empty() => Err(PlanError::Validation(
69                "timestamp timezone must not be empty".into(),
70            )),
71            Self::List(element) => element.validate(),
72            Self::Map { key, value } => {
73                key.validate()?;
74                value.validate()
75            }
76            Self::Struct(fields) => {
77                let mut names = std::collections::HashSet::new();
78                for field in fields {
79                    if field.name.is_empty() || !names.insert(&field.name) {
80                        return Err(PlanError::Validation(
81                            "struct field names must be non-empty and unique".into(),
82                        ));
83                    }
84                    field.data_type.validate()?;
85                }
86                Ok(())
87            }
88            _ => Ok(()),
89        }
90    }
91}
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
94#[serde(rename_all = "snake_case")]
95pub enum TimeUnit {
96    Second,
97    Millisecond,
98    Microsecond,
99    Nanosecond,
100}
101
102#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
103#[serde(rename_all = "snake_case")]
104pub enum IntervalUnit {
105    YearMonth,
106    DayTime,
107    MonthDayNano,
108}
109
110/// Typed scalar literals. Float values retain their exact IEEE-754 bits.
111#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
112#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
113pub enum ScalarValue {
114    Null,
115    Boolean(bool),
116    Int64(i64),
117    UInt64(u64),
118    Float64(u64),
119    Utf8(String),
120    Binary(Vec<u8>),
121    Decimal128 {
122        value: i128,
123        precision: u8,
124        scale: i8,
125    },
126    Date32(i32),
127    Timestamp {
128        value: i64,
129        unit: TimeUnit,
130        timezone: Option<String>,
131    },
132    Interval {
133        value: i128,
134        unit: IntervalUnit,
135    },
136}
137
138impl ScalarValue {
139    pub fn float64(value: f64) -> Self {
140        Self::Float64(value.to_bits())
141    }
142    pub fn as_f64(&self) -> Option<f64> {
143        match self {
144            Self::Float64(bits) => Some(f64::from_bits(*bits)),
145            _ => None,
146        }
147    }
148
149    /// Render a scalar as a SQL literal for typed prepared-statement binding.
150    pub fn to_sql_literal(&self) -> String {
151        scalar_sql(self)
152    }
153}
154
155#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
156#[serde(rename_all = "snake_case")]
157pub enum BinaryOperator {
158    Eq,
159    NotEq,
160    Gt,
161    GtEq,
162    Lt,
163    LtEq,
164    And,
165    Or,
166    Plus,
167    Minus,
168    Multiply,
169    Divide,
170}
171
172#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
173#[serde(rename_all = "snake_case")]
174pub enum AggregateFunction {
175    Count,
176    Sum,
177    Avg,
178    Min,
179    Max,
180}
181
182#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
183#[serde(rename_all = "snake_case")]
184pub enum SortDirection {
185    Ascending,
186    Descending,
187}
188
189#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
190#[serde(rename_all = "snake_case")]
191pub enum NullOrdering {
192    First,
193    Last,
194}
195
196/// `ROWS` or `RANGE` framing for a window function's frame bounds.
197#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
198#[serde(rename_all = "snake_case")]
199pub enum WindowFrameUnits {
200    Rows,
201    Range,
202}
203
204/// One bound (start or end) of a window frame.
205#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
206#[serde(tag = "bound", rename_all = "snake_case")]
207pub enum WindowFrameBound {
208    UnboundedPreceding,
209    Preceding(u64),
210    CurrentRow,
211    Following(u64),
212    UnboundedFollowing,
213}
214
215/// A window function frame specification, e.g. `ROWS BETWEEN 3 PRECEDING AND CURRENT ROW`.
216#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
217pub struct WindowFrame {
218    pub units: WindowFrameUnits,
219    pub start: WindowFrameBound,
220    pub end: WindowFrameBound,
221}
222
223impl WindowFrame {
224    /// `ROWS BETWEEN start AND end`.
225    pub fn rows(start: WindowFrameBound, end: WindowFrameBound) -> Self {
226        Self {
227            units: WindowFrameUnits::Rows,
228            start,
229            end,
230        }
231    }
232    /// `RANGE BETWEEN start AND end`.
233    pub fn range(start: WindowFrameBound, end: WindowFrameBound) -> Self {
234        Self {
235            units: WindowFrameUnits::Range,
236            start,
237            end,
238        }
239    }
240}
241
242fn window_frame_bound_sql(bound: WindowFrameBound) -> String {
243    match bound {
244        WindowFrameBound::UnboundedPreceding => "UNBOUNDED PRECEDING".to_string(),
245        WindowFrameBound::Preceding(n) => format!("{n} PRECEDING"),
246        WindowFrameBound::CurrentRow => "CURRENT ROW".to_string(),
247        WindowFrameBound::Following(n) => format!("{n} FOLLOWING"),
248        WindowFrameBound::UnboundedFollowing => "UNBOUNDED FOLLOWING".to_string(),
249    }
250}
251
252/// Structured public expression AST.
253#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
254#[serde(tag = "node", rename_all = "snake_case")]
255pub enum Expr {
256    Column {
257        path: Vec<String>,
258    },
259    Literal {
260        value: ScalarValue,
261    },
262    Alias {
263        expression: Box<Expr>,
264        name: String,
265    },
266    Binary {
267        left: Box<Expr>,
268        op: BinaryOperator,
269        right: Box<Expr>,
270    },
271    IsNull {
272        expression: Box<Expr>,
273        negated: bool,
274    },
275    Aggregate {
276        function: AggregateFunction,
277        expression: Option<Box<Expr>>,
278        distinct: bool,
279    },
280    Cast {
281        expression: Box<Expr>,
282        data_type: ExprDataType,
283        safe: bool,
284    },
285    Sort {
286        expression: Box<Expr>,
287        direction: SortDirection,
288        nulls: NullOrdering,
289    },
290    Function {
291        name: String,
292        arguments: Vec<Expr>,
293    },
294    Window {
295        expression: Box<Expr>,
296        partition_by: Vec<Expr>,
297        order_by: Vec<Expr>,
298        /// `ROWS`/`RANGE BETWEEN ... AND ...` frame bounds. `#[serde(default)]`
299        /// so expressions serialized before this field existed still decode
300        /// (as "no frame specified", DataFusion's own OVER-clause default).
301        #[serde(default)]
302        frame: Option<WindowFrame>,
303    },
304    RawSql {
305        sql: String,
306    },
307}
308
309impl Expr {
310    pub fn column(name: &str) -> Self {
311        Self::Column {
312            path: name.split('.').map(ToOwned::to_owned).collect(),
313        }
314    }
315    pub fn literal(value: ScalarValue) -> Self {
316        Self::Literal { value }
317    }
318    pub fn raw(sql: impl Into<String>) -> Self {
319        Self::RawSql { sql: sql.into() }
320    }
321    pub fn alias(self, name: impl Into<String>) -> Self {
322        Self::Alias {
323            expression: Box::new(self),
324            name: name.into(),
325        }
326    }
327    pub fn binary(self, op: BinaryOperator, right: Expr) -> Self {
328        Self::Binary {
329            left: Box::new(self),
330            op,
331            right: Box::new(right),
332        }
333    }
334    pub fn cast(self, data_type: ExprDataType) -> Self {
335        Self::Cast {
336            expression: Box::new(self),
337            data_type,
338            safe: false,
339        }
340    }
341    pub fn try_cast(self, data_type: ExprDataType) -> Self {
342        Self::Cast {
343            expression: Box::new(self),
344            data_type,
345            safe: true,
346        }
347    }
348    pub fn function(name: impl Into<String>, arguments: Vec<Expr>) -> Self {
349        Self::Function {
350            name: name.into(),
351            arguments,
352        }
353    }
354    pub fn over(self, partition_by: Vec<Expr>, order_by: Vec<Expr>) -> Self {
355        Self::Window {
356            expression: Box::new(self),
357            partition_by,
358            order_by,
359            frame: None,
360        }
361    }
362    /// Attach a `ROWS`/`RANGE BETWEEN ... AND ...` frame to a window
363    /// expression built by [`Expr::over`]. A no-op on any other variant —
364    /// there is no frame to attach without a preceding `.over(...)` call.
365    pub fn frame(self, frame: WindowFrame) -> Self {
366        match self {
367            Self::Window {
368                expression,
369                partition_by,
370                order_by,
371                frame: _,
372            } => Self::Window {
373                expression,
374                partition_by,
375                order_by,
376                frame: Some(frame),
377            },
378            other => other,
379        }
380    }
381    pub fn normalize_json(&self) -> Result<String, PlanError> {
382        serde_json::to_string(self).map_err(|error| PlanError::Encode(error.to_string()))
383    }
384    pub fn encode_versioned(&self) -> Result<Vec<u8>, PlanError> {
385        self.validate()?;
386        serde_json::to_vec(&ExpressionEnvelope {
387            version: EXPRESSION_FORMAT_VERSION,
388            expression: self.clone(),
389        })
390        .map_err(|error| PlanError::Encode(error.to_string()))
391    }
392    pub fn decode_versioned(bytes: &[u8]) -> Result<Self, PlanError> {
393        let envelope: ExpressionEnvelope =
394            serde_json::from_slice(bytes).map_err(|error| PlanError::Parse(error.to_string()))?;
395        if envelope.version != EXPRESSION_FORMAT_VERSION {
396            return Err(PlanError::Validation(format!(
397                "unsupported expression format version {}; expected {}",
398                envelope.version, EXPRESSION_FORMAT_VERSION
399            )));
400        }
401        envelope.expression.validate()?;
402        Ok(envelope.expression)
403    }
404    pub fn validate(&self) -> Result<(), PlanError> {
405        match self {
406            Self::Column { path } if path.is_empty() || path.iter().any(String::is_empty) => Err(
407                PlanError::Validation("column path must not be empty".into()),
408            ),
409            Self::Column { .. } => Ok(()),
410            Self::Literal { value } => match value {
411                ScalarValue::Decimal128 {
412                    precision, scale, ..
413                } => ExprDataType::Decimal128 {
414                    precision: *precision,
415                    scale: *scale,
416                }
417                .validate(),
418                ScalarValue::Timestamp { unit, timezone, .. } => ExprDataType::Timestamp {
419                    unit: *unit,
420                    timezone: timezone.clone(),
421                }
422                .validate(),
423                _ => Ok(()),
424            },
425            Self::Alias { expression, name } => {
426                if name.is_empty() {
427                    return Err(PlanError::Validation("alias must not be empty".into()));
428                }
429                expression.validate()
430            }
431            Self::Binary { left, right, .. } => {
432                left.validate()?;
433                right.validate()
434            }
435            Self::IsNull { expression, .. } | Self::Sort { expression, .. } => {
436                expression.validate()
437            }
438            Self::Cast {
439                expression,
440                data_type,
441                ..
442            } => {
443                expression.validate()?;
444                data_type.validate()
445            }
446            Self::Aggregate { expression, .. } => {
447                if let Some(expression) = expression {
448                    expression.validate()?;
449                }
450                Ok(())
451            }
452            Self::Function { name, arguments } => {
453                if name.trim().is_empty() {
454                    return Err(PlanError::Validation(
455                        "function name must not be empty".into(),
456                    ));
457                }
458                arguments.iter().try_for_each(Self::validate)
459            }
460            Self::Window {
461                expression,
462                partition_by,
463                order_by,
464                frame,
465            } => {
466                expression.validate()?;
467                partition_by.iter().try_for_each(Self::validate)?;
468                order_by.iter().try_for_each(Self::validate)?;
469                if let Some(frame) = frame
470                    && order_by.is_empty()
471                    && frame.units == WindowFrameUnits::Range
472                    && !matches!(
473                        (frame.start, frame.end),
474                        (
475                            WindowFrameBound::UnboundedPreceding,
476                            WindowFrameBound::UnboundedFollowing
477                        )
478                    )
479                {
480                    return Err(PlanError::Validation(
481                        "RANGE frame with PRECEDING/FOLLOWING/CURRENT ROW bounds requires ORDER BY"
482                            .into(),
483                    ));
484                }
485                Ok(())
486            }
487            Self::RawSql { sql } => {
488                if sql.trim().is_empty() {
489                    Err(PlanError::Validation(
490                        "raw SQL expression must not be empty".into(),
491                    ))
492                } else {
493                    Ok(())
494                }
495            }
496        }
497    }
498    pub fn to_sql(&self) -> String {
499        match self {
500            Self::Column { path } => path
501                .iter()
502                .map(|part| quote_identifier(part))
503                .collect::<Vec<_>>()
504                .join("."),
505            Self::Literal { value } => scalar_sql(value),
506            Self::Alias { expression, name } => {
507                format!("{} AS {}", expression.to_sql(), quote_identifier(name))
508            }
509            Self::Binary { left, op, right } => format!(
510                "({} {} {})",
511                left.to_sql(),
512                operator_sql(*op),
513                right.to_sql()
514            ),
515            Self::IsNull {
516                expression,
517                negated,
518            } => format!(
519                "({} IS {}NULL)",
520                expression.to_sql(),
521                if *negated { "NOT " } else { "" }
522            ),
523            Self::Aggregate {
524                function,
525                expression,
526                distinct,
527            } => {
528                let argument = expression
529                    .as_ref()
530                    .map(|value| value.to_sql())
531                    .unwrap_or_else(|| "*".into());
532                format!(
533                    "{}({}{argument})",
534                    aggregate_sql(*function),
535                    if *distinct { "DISTINCT " } else { "" }
536                )
537            }
538            Self::Cast {
539                expression,
540                data_type,
541                safe,
542            } => format!(
543                "{}({} AS {})",
544                if *safe { "TRY_CAST" } else { "CAST" },
545                expression.to_sql(),
546                type_sql(data_type)
547            ),
548            Self::Sort {
549                expression,
550                direction,
551                nulls,
552            } => format!(
553                "{} {} NULLS {}",
554                expression.to_sql(),
555                if *direction == SortDirection::Ascending {
556                    "ASC"
557                } else {
558                    "DESC"
559                },
560                if *nulls == NullOrdering::First {
561                    "FIRST"
562                } else {
563                    "LAST"
564                }
565            ),
566            Self::Function { name, arguments } => format!(
567                "{}({})",
568                name,
569                arguments
570                    .iter()
571                    .map(Self::to_sql)
572                    .collect::<Vec<_>>()
573                    .join(", ")
574            ),
575            Self::Window {
576                expression,
577                partition_by,
578                order_by,
579                frame,
580            } => {
581                let mut clauses = Vec::new();
582                if !partition_by.is_empty() {
583                    clauses.push(format!(
584                        "PARTITION BY {}",
585                        partition_by
586                            .iter()
587                            .map(Self::to_sql)
588                            .collect::<Vec<_>>()
589                            .join(", ")
590                    ));
591                }
592                if !order_by.is_empty() {
593                    clauses.push(format!(
594                        "ORDER BY {}",
595                        order_by
596                            .iter()
597                            .map(Self::to_sql)
598                            .collect::<Vec<_>>()
599                            .join(", ")
600                    ));
601                }
602                if let Some(frame) = frame {
603                    let units = match frame.units {
604                        WindowFrameUnits::Rows => "ROWS",
605                        WindowFrameUnits::Range => "RANGE",
606                    };
607                    clauses.push(format!(
608                        "{units} BETWEEN {} AND {}",
609                        window_frame_bound_sql(frame.start),
610                        window_frame_bound_sql(frame.end),
611                    ));
612                }
613                format!("{} OVER ({})", expression.to_sql(), clauses.join(" "))
614            }
615            Self::RawSql { sql } => sql.clone(),
616        }
617    }
618}
619
620#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
621struct ExpressionEnvelope {
622    version: u16,
623    expression: Expr,
624}
625
626fn operator_sql(op: BinaryOperator) -> &'static str {
627    match op {
628        BinaryOperator::Eq => "=",
629        BinaryOperator::NotEq => "<>",
630        BinaryOperator::Gt => ">",
631        BinaryOperator::GtEq => ">=",
632        BinaryOperator::Lt => "<",
633        BinaryOperator::LtEq => "<=",
634        BinaryOperator::And => "AND",
635        BinaryOperator::Or => "OR",
636        BinaryOperator::Plus => "+",
637        BinaryOperator::Minus => "-",
638        BinaryOperator::Multiply => "*",
639        BinaryOperator::Divide => "/",
640    }
641}
642fn aggregate_sql(function: AggregateFunction) -> &'static str {
643    match function {
644        AggregateFunction::Count => "COUNT",
645        AggregateFunction::Sum => "SUM",
646        AggregateFunction::Avg => "AVG",
647        AggregateFunction::Min => "MIN",
648        AggregateFunction::Max => "MAX",
649    }
650}
651fn scalar_sql(value: &ScalarValue) -> String {
652    // H-8 (audit): the prior implementation emitted bare integers for
653    // Date32 / Timestamp / Decimal128 and the bare string "NaN" for
654    // Float64 NaN. Every typed bind with one of these scalars either
655    // raised a backend parse error (NaN) or produced a silently wrong
656    // value (Date32 / Timestamp / Decimal128). The new implementation
657    // produces typed literals that DataFusion, Postgres, MySQL, DuckDB,
658    // and SQLite all accept.
659    use chrono::{DateTime, NaiveDate, Utc};
660    match value {
661        ScalarValue::Null => "NULL".into(),
662        ScalarValue::Boolean(value) => value.to_string().to_ascii_uppercase(),
663        ScalarValue::Int64(value) => value.to_string(),
664        ScalarValue::UInt64(value) => value.to_string(),
665        ScalarValue::Float64(bits) => float_to_sql(f64::from_bits(*bits)),
666        ScalarValue::Utf8(value) => format!("'{}'", value.replace('\'', "''")),
667        ScalarValue::Binary(value) => format!(
668            "X'{}'",
669            value
670                .iter()
671                .map(|byte| format!("{byte:02X}"))
672                .collect::<String>()
673        ),
674        ScalarValue::Decimal128 {
675            value,
676            precision,
677            scale,
678        } => {
679            // H-8: a bare integer like `12345` would parse as BIGINT and
680            // be silently cast to DECIMAL on the consuming side, which
681            // does not preserve the value (e.g. 12345 cast to
682            // DECIMAL(10,2) becomes 12345.00, not 123.45). The CAST
683            // form makes the type explicit.
684            format!("CAST({value} AS DECIMAL({precision},{scale}))")
685        }
686        ScalarValue::Date32(value) => {
687            // H-8: a bare integer for a date is parsed as BIGINT and the
688            // expression becomes type-error vs DATE. Render as
689            // DATE 'YYYY-MM-DD' so the literal matches the column type.
690            let Some(epoch) = NaiveDate::from_ymd_opt(1970, 1, 1) else {
691                return "DATE '1970-01-01'".to_owned();
692            };
693            let date = epoch + chrono::Duration::days(*value as i64);
694            format!("DATE '{}'", date.format("%Y-%m-%d"))
695        }
696        ScalarValue::Timestamp {
697            value,
698            unit,
699            timezone: _,
700        } => {
701            // H-8: a bare integer is type-error vs TIMESTAMP. Render as a
702            // typed TIMESTAMP literal. The unit determines precision.
703            let formatted = match unit {
704                TimeUnit::Second => DateTime::<Utc>::from_timestamp(*value, 0)
705                    .map(|d| d.format("%Y-%m-%dT%H:%M:%SZ").to_string())
706                    .unwrap_or_else(|| format!("from_unixtime({value})")),
707                TimeUnit::Millisecond => DateTime::<Utc>::from_timestamp_millis(*value)
708                    .map(|d| d.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string())
709                    .unwrap_or_else(|| format!("from_unixtime_ms({value})")),
710                TimeUnit::Microsecond => DateTime::<Utc>::from_timestamp_micros(*value)
711                    .map(|d| d.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string())
712                    .unwrap_or_else(|| format!("from_unixtime_us({value})")),
713                TimeUnit::Nanosecond => {
714                    let secs = value.div_euclid(1_000_000_000);
715                    let nsec = value.rem_euclid(1_000_000_000) as u32;
716                    DateTime::<Utc>::from_timestamp(secs, nsec)
717                        .map(|d| d.format("%Y-%m-%dT%H:%M:%S%.9fZ").to_string())
718                        .unwrap_or_else(|| format!("CAST({value} AS BIGINT)"))
719                }
720            };
721            format!("TIMESTAMP '{formatted}'")
722        }
723        ScalarValue::Interval { value, unit } => match unit {
724            IntervalUnit::YearMonth => {
725                // value is i128; render as a year-month interval.
726                format!("INTERVAL '{value} months'")
727            }
728            IntervalUnit::DayTime => {
729                // value is i128; pack lower 64 bits as days, upper 64 as
730                // milliseconds. Approximate; sufficient for bind roundtrip.
731                let days = (*value & 0xFFFF_FFFF_FFFF_FFFF) as i64;
732                let millis = ((*value >> 64) & 0xFFFF_FFFF_FFFF_FFFF) as i64;
733                format!("INTERVAL '{days} days {millis} ms'")
734            }
735            IntervalUnit::MonthDayNano => {
736                let months = (*value & 0xFFFF_FFFF_FFFF_FFFF) as i64;
737                let days = ((*value >> 64) & 0xFFFF_FFFF) as i64;
738                let nanos = ((*value >> 96) & 0xFFFF_FFFF_FFFF_FFFF) as i64;
739                format!("INTERVAL '{months} months {days} days {nanos} ns'")
740            }
741        },
742    }
743}
744
745/// Render an f64 as a SQL literal. NaN and infinity are not valid SQL
746/// literals in any major backend; the portable form is
747/// `CAST('NaN' AS DOUBLE)` / `CAST('Infinity' AS DOUBLE)`.
748fn float_to_sql(value: f64) -> String {
749    if value.is_nan() {
750        "CAST('NaN' AS DOUBLE)".into()
751    } else if value.is_infinite() {
752        if value.is_sign_positive() {
753            "CAST('Infinity' AS DOUBLE)".into()
754        } else {
755            "CAST('-Infinity' AS DOUBLE)".into()
756        }
757    } else {
758        value.to_string()
759    }
760}
761fn type_sql(data_type: &ExprDataType) -> String {
762    match data_type {
763        ExprDataType::Null => "NULL".into(),
764        ExprDataType::Boolean => "BOOLEAN".into(),
765        ExprDataType::Int64 => "BIGINT".into(),
766        ExprDataType::UInt64 => "BIGINT UNSIGNED".into(),
767        ExprDataType::Float64 => "DOUBLE".into(),
768        ExprDataType::Utf8 => "VARCHAR".into(),
769        ExprDataType::Binary => "BINARY".into(),
770        ExprDataType::Decimal128 { precision, scale } => format!("DECIMAL({precision}, {scale})"),
771        ExprDataType::Date32 => "DATE".into(),
772        ExprDataType::Timestamp { .. } => "TIMESTAMP".into(),
773        ExprDataType::Interval { .. } => "INTERVAL".into(),
774        ExprDataType::List(value) => format!("ARRAY<{}>", type_sql(value)),
775        ExprDataType::Map { key, value } => format!("MAP<{}, {}>", type_sql(key), type_sql(value)),
776        ExprDataType::Struct(fields) => format!(
777            "STRUCT<{}>",
778            fields
779                .iter()
780                .map(|field| format!(
781                    "{}: {}",
782                    quote_identifier(&field.name),
783                    type_sql(&field.data_type)
784                ))
785                .collect::<Vec<_>>()
786                .join(", ")
787        ),
788        ExprDataType::Variant => "VARIANT".into(),
789    }
790}
791
792#[cfg(test)]
793mod tests {
794    use super::*;
795    #[test]
796    fn versioned_round_trip_is_stable() {
797        let expr = Expr::column("orders.id")
798            .binary(BinaryOperator::Gt, Expr::literal(ScalarValue::Int64(10)));
799        let bytes = expr.encode_versioned().unwrap();
800        assert_eq!(Expr::decode_versioned(&bytes).unwrap(), expr);
801        assert_eq!(
802            expr,
803            Expr::decode_versioned(&expr.encode_versioned().unwrap()).unwrap()
804        );
805    }
806    #[test]
807    fn rejects_unknown_version() {
808        let bytes = br#"{"version":99,"expression":{"node":"raw_sql","sql":"1"}}"#;
809        assert!(matches!(
810            Expr::decode_versioned(bytes),
811            Err(PlanError::Validation(_))
812        ));
813    }
814    #[test]
815    fn validation_rejects_invalid_decimal_and_empty_raw_sql() {
816        let invalid_decimal = Expr::literal(ScalarValue::Decimal128 {
817            value: 1,
818            precision: 0,
819            scale: 0,
820        });
821        assert!(matches!(
822            invalid_decimal.encode_versioned(),
823            Err(PlanError::Validation(_))
824        ));
825        assert!(matches!(
826            Expr::raw(" ").encode_versioned(),
827            Err(PlanError::Validation(_))
828        ));
829    }
830
831    #[test]
832    fn window_expression_is_structured_and_renderable() {
833        let expression = Expr::function("row_number", vec![]).over(
834            vec![Expr::column("account_id")],
835            vec![Expr::Sort {
836                expression: Box::new(Expr::column("event_time")),
837                direction: SortDirection::Ascending,
838                nulls: NullOrdering::Last,
839            }],
840        );
841        assert_eq!(
842            expression.to_sql(),
843            r#"row_number() OVER (PARTITION BY "account_id" ORDER BY "event_time" ASC NULLS LAST)"#
844        );
845        expression.validate().unwrap();
846    }
847
848    #[test]
849    fn window_frame_renders_rows_between() {
850        let expression = Expr::function("sum", vec![Expr::column("amount")])
851            .over(
852                vec![],
853                vec![Expr::Sort {
854                    expression: Box::new(Expr::column("ts")),
855                    direction: SortDirection::Ascending,
856                    nulls: NullOrdering::First,
857                }],
858            )
859            .frame(WindowFrame::rows(
860                WindowFrameBound::Preceding(3),
861                WindowFrameBound::CurrentRow,
862            ));
863        assert_eq!(
864            expression.to_sql(),
865            r#"sum("amount") OVER (ORDER BY "ts" ASC NULLS FIRST ROWS BETWEEN 3 PRECEDING AND CURRENT ROW)"#
866        );
867        expression.validate().unwrap();
868    }
869
870    #[test]
871    fn window_frame_renders_range_unbounded() {
872        let expression = Expr::function("avg", vec![Expr::column("amount")])
873            .over(vec![], vec![])
874            .frame(WindowFrame::range(
875                WindowFrameBound::UnboundedPreceding,
876                WindowFrameBound::UnboundedFollowing,
877            ));
878        assert_eq!(
879            expression.to_sql(),
880            r#"avg("amount") OVER (RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)"#
881        );
882        expression.validate().unwrap();
883    }
884
885    #[test]
886    fn window_frame_range_with_bounds_requires_order_by() {
887        let expression = Expr::function("sum", vec![Expr::column("amount")])
888            .over(vec![], vec![])
889            .frame(WindowFrame::range(
890                WindowFrameBound::Preceding(1),
891                WindowFrameBound::CurrentRow,
892            ));
893        assert!(matches!(
894            expression.validate(),
895            Err(PlanError::Validation(_))
896        ));
897    }
898
899    #[test]
900    fn frame_is_a_no_op_on_non_window_expressions() {
901        let expression = Expr::column("amount").frame(WindowFrame::rows(
902            WindowFrameBound::CurrentRow,
903            WindowFrameBound::CurrentRow,
904        ));
905        assert_eq!(expression, Expr::column("amount"));
906    }
907
908    #[test]
909    fn window_frame_versioned_round_trip() {
910        let expression = Expr::function("row_number", vec![])
911            .over(
912                vec![],
913                vec![Expr::Sort {
914                    expression: Box::new(Expr::column("ts")),
915                    direction: SortDirection::Ascending,
916                    nulls: NullOrdering::First,
917                }],
918            )
919            .frame(WindowFrame::rows(
920                WindowFrameBound::UnboundedPreceding,
921                WindowFrameBound::CurrentRow,
922            ));
923        let bytes = expression.encode_versioned().unwrap();
924        assert_eq!(Expr::decode_versioned(&bytes).unwrap(), expression);
925    }
926
927    #[test]
928    fn window_without_frame_field_still_decodes() {
929        // Pre-frame-field serialized shape (no "frame" key at all) must still
930        // decode, defaulting to "no frame specified" via #[serde(default)].
931        let json = br#"{"version":1,"expression":{"node":"window","expression":{"node":"function","name":"row_number","arguments":[]},"partition_by":[],"order_by":[]}}"#;
932        let expression = Expr::decode_versioned(json).unwrap();
933        assert_eq!(
934            expression,
935            Expr::function("row_number", vec![]).over(vec![], vec![])
936        );
937    }
938
939    #[test]
940    fn normalized_ast_is_deterministic() {
941        let expr = Expr::column("a").binary(
942            BinaryOperator::Eq,
943            Expr::literal(ScalarValue::Utf8("x".into())),
944        );
945        assert_eq!(
946            expr.normalize_json().unwrap(),
947            expr.normalize_json().unwrap()
948        );
949    }
950
951    #[test]
952    fn variant_type_sql_name() {
953        assert_eq!(type_sql(&ExprDataType::Variant), "VARIANT");
954    }
955
956    #[test]
957    fn variant_type_validates() {
958        assert!(ExprDataType::Variant.validate().is_ok());
959    }
960
961    #[test]
962    fn variant_type_round_trips_via_serde() {
963        let t = ExprDataType::Variant;
964        let json = serde_json::to_string(&t).unwrap();
965        let back: ExprDataType = serde_json::from_str(&json).unwrap();
966        assert_eq!(back, t);
967    }
968}
krishiv_plan/expression.rs

krishiv_plan/
expression.rs
