feldera_types/
program_schema.rs

1pub use feldera_ir::SourcePosition;
2use serde::{Deserialize, Deserializer, Serialize, Serializer};
3use std::cmp::Ordering;
4use std::collections::BTreeMap;
5use std::fmt::Display;
6use std::hash::{Hash, Hasher};
7use utoipa::ToSchema;
8use utoipa::openapi::{ObjectBuilder, RefOr, Schema, SchemaType};
9
10#[cfg(feature = "testing")]
11use proptest::{collection::vec, prelude::any};
12
13/// Returns canonical form of a SQL identifier:
14///
15/// - If id is _not_ quoted, then it is interpreted as a case-insensitive
16///   identifier and is converted to the lowercase representation
17/// - If id _is_ quoted, then it is a case-sensitive identifier and is returned
18///   as is, without quotes. No other processing is done on the inner string,
19///   e.g., un-escaping quotes.
20pub fn canonical_identifier(id: &str) -> String {
21    if id.starts_with('"') && id.ends_with('"') && id.len() >= 2 {
22        id[1..id.len() - 1].to_string()
23    } else {
24        id.to_lowercase()
25    }
26}
27
28/// An SQL identifier.
29///
30/// This struct is used to represent SQL identifiers in a canonical form.
31/// We store table names or field names as identifiers in the schema.
32#[derive(Serialize, Deserialize, ToSchema, Debug, Clone)]
33#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
34pub struct SqlIdentifier {
35    #[cfg_attr(feature = "testing", proptest(regex = "relation1|relation2|relation3"))]
36    name: String,
37    pub case_sensitive: bool,
38}
39
40impl SqlIdentifier {
41    pub fn new<S: AsRef<str>>(name: S, case_sensitive: bool) -> Self {
42        Self {
43            name: name.as_ref().to_string(),
44            case_sensitive,
45        }
46    }
47
48    /// Return the name of the identifier in canonical form.
49    /// The result is the true case-sensitive identifying name of the table,
50    /// and can be used for example to detect duplicate table names.
51    ///
52    /// Example return values for this function:
53    /// - `CREATE TABLE t1` -> `t1`
54    /// - `CREATE TABLE T1` -> `t1`
55    /// - `CREATE TABLE "t1"` -> `t1`
56    /// - `CREATE TABLE "T1"` -> `T1`
57    pub fn name(&self) -> String {
58        if self.case_sensitive {
59            self.name.clone()
60        } else {
61            self.name.to_lowercase()
62        }
63    }
64
65    /// Return the name of the identifier as it appeared originally in SQL.
66    /// This method should only be used for log or error messages as it is what
67    /// the user originally wrote, however it should not be used for identification
68    /// or disambiguation (use `name()` for that instead).
69    ///
70    /// Example return values for this function:
71    /// - `CREATE TABLE t1` -> `t1`
72    /// - `CREATE TABLE T1` -> `T1`
73    /// - `CREATE TABLE "t1"` -> `"t1"`
74    /// - `CREATE TABLE "T1"` -> `"T1"`
75    pub fn sql_name(&self) -> String {
76        if self.case_sensitive {
77            format!("\"{}\"", self.name)
78        } else {
79            self.name.clone()
80        }
81    }
82}
83
84impl Hash for SqlIdentifier {
85    fn hash<H: Hasher>(&self, state: &mut H) {
86        self.name().hash(state);
87    }
88}
89
90impl PartialEq for SqlIdentifier {
91    fn eq(&self, other: &Self) -> bool {
92        match (self.case_sensitive, other.case_sensitive) {
93            (true, true) => self.name == other.name,
94            (false, false) => self.name.to_lowercase() == other.name.to_lowercase(),
95            (true, false) => self.name == other.name,
96            (false, true) => self.name == other.name,
97        }
98    }
99}
100
101impl Ord for SqlIdentifier {
102    fn cmp(&self, other: &Self) -> Ordering {
103        self.name().cmp(&other.name())
104    }
105}
106
107impl PartialOrd for SqlIdentifier {
108    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
109        Some(self.cmp(other))
110    }
111}
112
113impl<S: AsRef<str>> PartialEq<S> for SqlIdentifier {
114    fn eq(&self, other: &S) -> bool {
115        self == &SqlIdentifier::from(other.as_ref())
116    }
117}
118
119impl Eq for SqlIdentifier {}
120
121impl<S: AsRef<str>> From<S> for SqlIdentifier {
122    fn from(name: S) -> Self {
123        if name.as_ref().starts_with('"')
124            && name.as_ref().ends_with('"')
125            && name.as_ref().len() >= 2
126        {
127            Self {
128                name: name.as_ref()[1..name.as_ref().len() - 1].to_string(),
129                case_sensitive: true,
130            }
131        } else {
132            Self::new(name, false)
133        }
134    }
135}
136
137impl From<SqlIdentifier> for String {
138    fn from(id: SqlIdentifier) -> String {
139        id.name()
140    }
141}
142
143impl From<&SqlIdentifier> for String {
144    fn from(id: &SqlIdentifier) -> String {
145        id.name()
146    }
147}
148
149impl Display for SqlIdentifier {
150    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
151        write!(f, "{}", self.name())
152    }
153}
154
155/// A struct containing the tables (inputs) and views for a program.
156///
157/// Parse from the JSON data-type of the DDL generated by the SQL compiler.
158#[derive(Default, Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone)]
159#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
160pub struct ProgramSchema {
161    #[cfg_attr(
162        feature = "testing",
163        proptest(strategy = "vec(any::<Relation>(), 0..2)")
164    )]
165    pub inputs: Vec<Relation>,
166    #[cfg_attr(
167        feature = "testing",
168        proptest(strategy = "vec(any::<Relation>(), 0..2)")
169    )]
170    pub outputs: Vec<Relation>,
171}
172
173impl ProgramSchema {
174    pub fn relations_with_lateness(&self) -> Vec<SqlIdentifier> {
175        self.inputs
176            .iter()
177            .chain(self.outputs.iter())
178            .filter(|rel| rel.has_lateness())
179            .map(|rel| rel.name.clone())
180            .collect()
181    }
182}
183
184/// A version of `ProgramSchema` that contains only the name and properties of the relations
185/// by making use of `RelationPropertiesOnly` instead of `Relation` for its inputs and outputs.
186/// This is used to avoid parsing the entire `Relation` object, including SQL schema,
187/// which can change across runtime versions.
188#[derive(Debug, Deserialize)]
189pub struct ProgramSchemaPropertiesOnly {
190    #[serde(default)]
191    pub inputs: Vec<RelationPropertiesOnly>,
192    #[serde(default)]
193    pub outputs: Vec<RelationPropertiesOnly>,
194}
195
196#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone)]
197#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
198pub struct PropertyValue {
199    pub value: String,
200    pub key_position: SourcePosition,
201    pub value_position: SourcePosition,
202}
203
204/// A SQL table or view. It has a name and a list of fields.
205///
206/// Matches the Calcite JSON format.
207#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone)]
208#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
209pub struct Relation {
210    #[serde(flatten)]
211    pub name: SqlIdentifier,
212    #[cfg_attr(feature = "testing", proptest(value = "Vec::new()"))]
213    pub fields: Vec<Field>,
214    #[serde(default)]
215    pub materialized: bool,
216    #[serde(default)]
217    pub properties: BTreeMap<String, PropertyValue>,
218    pub primary_key: Option<Vec<String>>,
219}
220
221impl Relation {
222    pub fn empty() -> Self {
223        Self {
224            name: SqlIdentifier::from("".to_string()),
225            fields: Vec::new(),
226            materialized: false,
227            properties: BTreeMap::new(),
228            primary_key: None,
229        }
230    }
231
232    pub fn new(
233        name: SqlIdentifier,
234        fields: Vec<Field>,
235        materialized: bool,
236        properties: BTreeMap<String, PropertyValue>,
237    ) -> Self {
238        Self {
239            name,
240            fields,
241            materialized,
242            properties,
243            primary_key: None,
244        }
245    }
246
247    /// Lookup field by name.
248    pub fn field(&self, name: &str) -> Option<&Field> {
249        let name = canonical_identifier(name);
250        self.fields.iter().find(|f| f.name == name)
251    }
252
253    pub fn has_lateness(&self) -> bool {
254        self.fields.iter().any(|f| f.lateness.is_some())
255    }
256
257    pub fn get_property(&self, name: &str) -> Option<&str> {
258        self.properties.get(name).map(|p| p.value.as_str())
259    }
260
261    /// `true` if the table declaration sets the `skip_unused_columns`
262    /// property, which asks connectors not to read declared-but-unused
263    /// columns.
264    pub fn skip_unused_columns(&self) -> bool {
265        self.get_property("skip_unused_columns") == Some("true")
266    }
267
268    pub fn with_primary_key<'a>(
269        mut self,
270        primary_key: impl IntoIterator<Item = &'a SqlIdentifier>,
271    ) -> Self {
272        self.primary_key = Some(primary_key.into_iter().map(|id| id.name()).collect());
273        self
274    }
275}
276
277/// A version of `Relation` that only contains the name and properties.
278/// This is used to avoid parsing the entire `Relation` object, including
279/// SQL schema, which can change across runtime versions.
280#[derive(Debug, Deserialize)]
281pub struct RelationPropertiesOnly {
282    #[serde(flatten)]
283    pub name: SqlIdentifier,
284    #[serde(default)]
285    pub properties: BTreeMap<String, PropertyValue>,
286}
287
288/// A SQL field.
289///
290/// Matches the SQL compiler JSON format.
291#[derive(Serialize, ToSchema, Debug, Eq, PartialEq, Clone)]
292#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
293pub struct Field {
294    #[serde(flatten)]
295    pub name: SqlIdentifier,
296    pub columntype: ColumnType,
297    pub lateness: Option<String>,
298    pub default: Option<String>,
299    pub unused: bool,
300}
301
302impl Field {
303    pub fn new(name: SqlIdentifier, columntype: ColumnType) -> Self {
304        Self {
305            name,
306            columntype,
307            lateness: None,
308            default: None,
309            unused: false,
310        }
311    }
312
313    pub fn with_lateness(mut self, lateness: &str) -> Self {
314        self.lateness = Some(lateness.to_string());
315        self
316    }
317
318    pub fn with_unused(mut self, unused: bool) -> Self {
319        self.unused = unused;
320        self
321    }
322}
323
324/// Thanks to the brain-dead Calcite schema, if we are deserializing a field, the type options
325/// end up inside the Field struct.
326///
327/// This helper struct is used to deserialize the Field struct.
328impl<'de> Deserialize<'de> for Field {
329    fn deserialize<D>(deserializer: D) -> Result<Field, D::Error>
330    where
331        D: Deserializer<'de>,
332    {
333        const fn default_is_struct() -> Option<SqlType> {
334            Some(SqlType::Struct)
335        }
336
337        #[derive(Debug, Clone, Deserialize)]
338        struct FieldHelper {
339            name: Option<String>,
340            #[serde(default)]
341            case_sensitive: bool,
342            columntype: Option<ColumnType>,
343            #[serde(rename = "type")]
344            #[serde(default = "default_is_struct")]
345            typ: Option<SqlType>,
346            nullable: Option<bool>,
347            precision: Option<i64>,
348            scale: Option<i64>,
349            component: Option<Box<ColumnType>>,
350            fields: Option<serde_json::Value>,
351            key: Option<Box<ColumnType>>,
352            value: Option<Box<ColumnType>>,
353            default: Option<String>,
354            #[serde(default)]
355            unused: bool,
356            lateness: Option<String>,
357        }
358
359        fn helper_to_field(helper: FieldHelper) -> Field {
360            let columntype = if let Some(ctype) = helper.columntype {
361                ctype
362            } else if let Some(serde_json::Value::Array(fields)) = helper.fields {
363                let fields = fields
364                    .into_iter()
365                    .map(|field| {
366                        let field: FieldHelper = serde_json::from_value(field).unwrap();
367                        helper_to_field(field)
368                    })
369                    .collect::<Vec<Field>>();
370
371                ColumnType {
372                    typ: helper.typ.unwrap_or(SqlType::Null),
373                    nullable: helper.nullable.unwrap_or(false),
374                    precision: helper.precision,
375                    scale: helper.scale,
376                    component: helper.component,
377                    fields: Some(fields),
378                    key: None,
379                    value: None,
380                }
381            } else if let Some(serde_json::Value::Object(obj)) = helper.fields {
382                serde_json::from_value(serde_json::Value::Object(obj))
383                    .expect("Failed to deserialize object")
384            } else {
385                ColumnType {
386                    typ: helper.typ.unwrap_or(SqlType::Null),
387                    nullable: helper.nullable.unwrap_or(false),
388                    precision: helper.precision,
389                    scale: helper.scale,
390                    component: helper.component,
391                    fields: None,
392                    key: helper.key,
393                    value: helper.value,
394                }
395            };
396
397            Field {
398                name: SqlIdentifier::new(helper.name.unwrap(), helper.case_sensitive),
399                columntype,
400                default: helper.default,
401                unused: helper.unused,
402                lateness: helper.lateness,
403            }
404        }
405
406        let helper = FieldHelper::deserialize(deserializer)?;
407        Ok(helper_to_field(helper))
408    }
409}
410
411/// The specified units for SQL Interval types.
412///
413/// `INTERVAL 1 DAY`, `INTERVAL 1 DAY TO HOUR`, `INTERVAL 1 DAY TO MINUTE`,
414/// would yield `Day`, `DayToHour`, `DayToMinute`, as the `IntervalUnit` respectively.
415#[derive(ToSchema, Debug, Eq, PartialEq, Clone, Copy)]
416#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
417pub enum IntervalUnit {
418    /// Unit for `INTERVAL ... DAY`.
419    Day,
420    /// Unit for `INTERVAL ... DAY TO HOUR`.
421    DayToHour,
422    /// Unit for `INTERVAL ... DAY TO MINUTE`.
423    DayToMinute,
424    /// Unit for `INTERVAL ... DAY TO SECOND`.
425    DayToSecond,
426    /// Unit for `INTERVAL ... HOUR`.
427    Hour,
428    /// Unit for `INTERVAL ... HOUR TO MINUTE`.
429    HourToMinute,
430    /// Unit for `INTERVAL ... HOUR TO SECOND`.
431    HourToSecond,
432    /// Unit for `INTERVAL ... MINUTE`.
433    Minute,
434    /// Unit for `INTERVAL ... MINUTE TO SECOND`.
435    MinuteToSecond,
436    /// Unit for `INTERVAL ... MONTH`.
437    Month,
438    /// Unit for `INTERVAL ... SECOND`.
439    Second,
440    /// Unit for `INTERVAL ... YEAR`.
441    Year,
442    /// Unit for `INTERVAL ... YEAR TO MONTH`.
443    YearToMonth,
444}
445
446/// The available SQL column types.
447///
448/// The OpenAPI schema for this type is implemented by hand (see the `ToSchema`
449/// impl below) rather than derived, because `SqlType` has hand-written
450/// `Serialize`/`Deserialize` impls that emit Feldera's uppercase wire spellings
451/// (`"BIGINT"`, `"INTEGER"`, `"INTERVAL_DAY"`, …) instead of the Rust variant
452/// names. These wire strings are the platform's encoding of the types on the
453/// JSON API; they are not valid SQL type syntax (SQL writes `INTERVAL DAY`, with
454/// a space). A derived schema would advertise the variant names and thus
455/// disagree with the bytes actually on the wire, breaking generated clients.
456#[derive(Debug, Eq, PartialEq, Clone, Copy)]
457#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
458pub enum SqlType {
459    /// SQL `BOOLEAN` type.
460    Boolean,
461    /// SQL `TINYINT` type.
462    TinyInt,
463    /// SQL `SMALLINT` or `INT2` type.
464    SmallInt,
465    /// SQL `INTEGER`, `INT`, `SIGNED`, `INT4` type.
466    Int,
467    /// SQL `BIGINT` or `INT64` type.
468    BigInt,
469    /// SQL `TINYINT UNSIGNED` type.
470    UTinyInt,
471    /// SQL `SMALLINT UNSIGNED` type.
472    USmallInt,
473    /// SQL `UNSIGNED`, `INTEGER UNSIGNED`, `INT UNSIGNED` type.
474    UInt,
475    /// SQL `BIGINT UNSIGNED` type.
476    UBigInt,
477    /// SQL `REAL` or `FLOAT4` or `FLOAT32` type.
478    Real,
479    /// SQL `DOUBLE` or `FLOAT8` or `FLOAT64` type.
480    Double,
481    /// SQL `DECIMAL` or `DEC` or `NUMERIC` type.
482    Decimal,
483    /// SQL `CHAR(n)` or `CHARACTER(n)` type.
484    Char,
485    /// SQL `VARCHAR`, `CHARACTER VARYING`, `TEXT`, or `STRING` type.
486    Varchar,
487    /// SQL `BINARY(n)` type.
488    Binary,
489    /// SQL `VARBINARY` or `BYTEA` type.
490    Varbinary,
491    /// SQL `TIME` type.
492    Time,
493    /// SQL `DATE` type.
494    Date,
495    /// SQL `TIMESTAMP` type.
496    Timestamp,
497    /// SQL `TIMESTAMP WITH TIME ZONE` type.
498    TimestampTz,
499    /// SQL `INTERVAL ... X` type where `X` is a unit.
500    Interval(IntervalUnit),
501    /// SQL `ARRAY` type.
502    Array,
503    /// A complex SQL struct type (`CREATE TYPE x ...`).
504    Struct,
505    /// SQL `MAP` type.
506    Map,
507    /// SQL `NULL` type.
508    Null,
509    /// SQL `UUID` type.
510    Uuid,
511    /// SQL `VARIANT` type.
512    Variant,
513}
514
515/// Every wire string `SqlType` can serialize to, in variant order.
516///
517/// This is the single source of truth shared by `Serialize` (via
518/// `SqlType::as_wire_str`) and the OpenAPI schema, so the generated clients see
519/// exactly the values that appear on the wire. The
520/// `sql_type_schema_matches_serialization` test guards that every variant's
521/// serialized form is listed here.
522const SQL_TYPE_VALUES: &[&str] = &[
523    "BOOLEAN",
524    "TINYINT",
525    "SMALLINT",
526    "INTEGER",
527    "BIGINT",
528    "UTINYINT",
529    "USMALLINT",
530    "UINTEGER",
531    "UBIGINT",
532    "REAL",
533    "DOUBLE",
534    "DECIMAL",
535    "CHAR",
536    "VARCHAR",
537    "BINARY",
538    "VARBINARY",
539    "TIME",
540    "DATE",
541    "TIMESTAMP",
542    "TIMESTAMP_TZ",
543    "INTERVAL_DAY",
544    "INTERVAL_DAY_HOUR",
545    "INTERVAL_DAY_MINUTE",
546    "INTERVAL_DAY_SECOND",
547    "INTERVAL_HOUR",
548    "INTERVAL_HOUR_MINUTE",
549    "INTERVAL_HOUR_SECOND",
550    "INTERVAL_MINUTE",
551    "INTERVAL_MINUTE_SECOND",
552    "INTERVAL_MONTH",
553    "INTERVAL_SECOND",
554    "INTERVAL_YEAR",
555    "INTERVAL_YEAR_MONTH",
556    "ARRAY",
557    "STRUCT",
558    "MAP",
559    "NULL",
560    "UUID",
561    "VARIANT",
562];
563
564impl SqlType {
565    /// The uppercase wire spelling of this type — the canonical serialized form
566    /// the platform emits on the JSON API (`"BIGINT"`, `"INTEGER"`,
567    /// `"INTERVAL_DAY"`, …). Note these are wire strings, not valid SQL type
568    /// syntax.
569    pub fn as_wire_str(&self) -> &'static str {
570        match self {
571            SqlType::Boolean => "BOOLEAN",
572            SqlType::TinyInt => "TINYINT",
573            SqlType::SmallInt => "SMALLINT",
574            SqlType::Int => "INTEGER",
575            SqlType::BigInt => "BIGINT",
576            SqlType::UTinyInt => "UTINYINT",
577            SqlType::USmallInt => "USMALLINT",
578            SqlType::UInt => "UINTEGER",
579            SqlType::UBigInt => "UBIGINT",
580            SqlType::Real => "REAL",
581            SqlType::Double => "DOUBLE",
582            SqlType::Decimal => "DECIMAL",
583            SqlType::Char => "CHAR",
584            SqlType::Varchar => "VARCHAR",
585            SqlType::Binary => "BINARY",
586            SqlType::Varbinary => "VARBINARY",
587            SqlType::Time => "TIME",
588            SqlType::Date => "DATE",
589            SqlType::Timestamp => "TIMESTAMP",
590            SqlType::TimestampTz => "TIMESTAMP_TZ",
591            SqlType::Interval(interval_unit) => match interval_unit {
592                IntervalUnit::Day => "INTERVAL_DAY",
593                IntervalUnit::DayToHour => "INTERVAL_DAY_HOUR",
594                IntervalUnit::DayToMinute => "INTERVAL_DAY_MINUTE",
595                IntervalUnit::DayToSecond => "INTERVAL_DAY_SECOND",
596                IntervalUnit::Hour => "INTERVAL_HOUR",
597                IntervalUnit::HourToMinute => "INTERVAL_HOUR_MINUTE",
598                IntervalUnit::HourToSecond => "INTERVAL_HOUR_SECOND",
599                IntervalUnit::Minute => "INTERVAL_MINUTE",
600                IntervalUnit::MinuteToSecond => "INTERVAL_MINUTE_SECOND",
601                IntervalUnit::Month => "INTERVAL_MONTH",
602                IntervalUnit::Second => "INTERVAL_SECOND",
603                IntervalUnit::Year => "INTERVAL_YEAR",
604                IntervalUnit::YearToMonth => "INTERVAL_YEAR_MONTH",
605            },
606            SqlType::Array => "ARRAY",
607            SqlType::Struct => "STRUCT",
608            SqlType::Map => "MAP",
609            SqlType::Null => "NULL",
610            SqlType::Uuid => "UUID",
611            SqlType::Variant => "VARIANT",
612        }
613    }
614}
615
616impl ToSchema<'_> for SqlType {
617    fn schema() -> (&'static str, RefOr<Schema>) {
618        (
619            "SqlType",
620            RefOr::T(Schema::Object(
621                ObjectBuilder::new()
622                    .schema_type(SchemaType::String)
623                    .description(Some(
624                        "The available SQL column type names. Each value is the platform's wire \
625                         encoding of the type (e.g. `BIGINT`, `INTEGER`, `INTERVAL_DAY`), not \
626                         valid SQL type syntax.",
627                    ))
628                    .enum_values(Some(SQL_TYPE_VALUES.iter().copied()))
629                    .build(),
630            )),
631        )
632    }
633}
634
635impl Display for SqlType {
636    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
637        f.write_str(&serde_json::to_string(self).unwrap())
638    }
639}
640
641impl<'de> Deserialize<'de> for SqlType {
642    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
643    where
644        D: Deserializer<'de>,
645    {
646        let value: String = Deserialize::deserialize(deserializer)?;
647        match value.to_lowercase().as_str() {
648            "interval_day" => Ok(SqlType::Interval(IntervalUnit::Day)),
649            "interval_day_hour" => Ok(SqlType::Interval(IntervalUnit::DayToHour)),
650            "interval_day_minute" => Ok(SqlType::Interval(IntervalUnit::DayToMinute)),
651            "interval_day_second" => Ok(SqlType::Interval(IntervalUnit::DayToSecond)),
652            "interval_hour" => Ok(SqlType::Interval(IntervalUnit::Hour)),
653            "interval_hour_minute" => Ok(SqlType::Interval(IntervalUnit::HourToMinute)),
654            "interval_hour_second" => Ok(SqlType::Interval(IntervalUnit::HourToSecond)),
655            "interval_minute" => Ok(SqlType::Interval(IntervalUnit::Minute)),
656            "interval_minute_second" => Ok(SqlType::Interval(IntervalUnit::MinuteToSecond)),
657            "interval_month" => Ok(SqlType::Interval(IntervalUnit::Month)),
658            "interval_second" => Ok(SqlType::Interval(IntervalUnit::Second)),
659            "interval_year" => Ok(SqlType::Interval(IntervalUnit::Year)),
660            "interval_year_month" => Ok(SqlType::Interval(IntervalUnit::YearToMonth)),
661            "boolean" => Ok(SqlType::Boolean),
662            "tinyint" => Ok(SqlType::TinyInt),
663            "smallint" => Ok(SqlType::SmallInt),
664            "integer" => Ok(SqlType::Int),
665            "bigint" => Ok(SqlType::BigInt),
666            "utinyint" => Ok(SqlType::UTinyInt),
667            "usmallint" => Ok(SqlType::USmallInt),
668            "uinteger" => Ok(SqlType::UInt),
669            "ubigint" => Ok(SqlType::UBigInt),
670            "real" => Ok(SqlType::Real),
671            "double" => Ok(SqlType::Double),
672            "decimal" => Ok(SqlType::Decimal),
673            "char" => Ok(SqlType::Char),
674            "varchar" => Ok(SqlType::Varchar),
675            "binary" => Ok(SqlType::Binary),
676            "varbinary" => Ok(SqlType::Varbinary),
677            "variant" => Ok(SqlType::Variant),
678            "time" => Ok(SqlType::Time),
679            "date" => Ok(SqlType::Date),
680            "timestamp" => Ok(SqlType::Timestamp),
681            "timestamp_tz" => Ok(SqlType::TimestampTz),
682            "array" => Ok(SqlType::Array),
683            "struct" => Ok(SqlType::Struct),
684            "map" => Ok(SqlType::Map),
685            "null" => Ok(SqlType::Null),
686            "uuid" => Ok(SqlType::Uuid),
687            _ => Err(serde::de::Error::custom(format!(
688                "Unknown SQL type: {}",
689                value
690            ))),
691        }
692    }
693}
694
695impl Serialize for SqlType {
696    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
697    where
698        S: Serializer,
699    {
700        serializer.serialize_str(self.as_wire_str())
701    }
702}
703
704impl SqlType {
705    /// Is this a string type?
706    pub fn is_string(&self) -> bool {
707        matches!(self, Self::Char | Self::Varchar)
708    }
709
710    pub fn is_varchar(&self) -> bool {
711        matches!(self, Self::Varchar)
712    }
713
714    pub fn is_varbinary(&self) -> bool {
715        matches!(self, Self::Varbinary)
716    }
717}
718
719/// It so happens that when the type field is missing in the Calcite schema, it's a struct,
720/// so we use it as the default.
721const fn default_is_struct() -> SqlType {
722    SqlType::Struct
723}
724
725/// A SQL column type description.
726///
727/// Matches the Calcite JSON format.
728#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone)]
729#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
730pub struct ColumnType {
731    /// Identifier for the type (e.g., `VARCHAR`, `BIGINT`, `ARRAY` etc.)
732    #[serde(rename = "type")]
733    #[serde(default = "default_is_struct")]
734    pub typ: SqlType,
735    /// Does the type accept NULL values?
736    pub nullable: bool,
737    /// Precision of the type.
738    ///
739    /// # Examples
740    /// - `VARCHAR` sets precision to `-1`.
741    /// - `VARCHAR(255)` sets precision to `255`.
742    /// - `BIGINT`, `DATE`, `FLOAT`, `DOUBLE`, `GEOMETRY`, etc. sets precision
743    ///   to None
744    /// - `TIME`, `TIMESTAMP` set precision to `0`.
745    pub precision: Option<i64>,
746    /// The scale of the type.
747    ///
748    /// # Example
749    /// - `DECIMAL(1,2)` sets scale to `2`.
750    pub scale: Option<i64>,
751    /// A component of the type (if available).
752    ///
753    /// This is in a `Box` because it makes it a recursive types.
754    ///
755    /// For example, this would specify the `VARCHAR(20)` in the `VARCHAR(20)
756    /// ARRAY` type.
757    #[cfg_attr(feature = "testing", proptest(value = "None"))]
758    pub component: Option<Box<ColumnType>>,
759    /// The fields of the type (if available).
760    ///
761    /// For example this would specify the fields of a `CREATE TYPE` construct.
762    ///
763    /// ```sql
764    /// CREATE TYPE person_typ AS (
765    ///   firstname       VARCHAR(30),
766    ///   lastname        VARCHAR(30),
767    ///   address         ADDRESS_TYP
768    /// );
769    /// ```
770    ///
771    /// Would lead to the following `fields` value:
772    ///
773    /// ```sql
774    /// [
775    ///  ColumnType { name: "firstname, ... },
776    ///  ColumnType { name: "lastname", ... },
777    ///  ColumnType { name: "address", fields: [ ... ] }
778    /// ]
779    /// ```
780    #[cfg_attr(feature = "testing", proptest(value = "Some(Vec::new())"))]
781    pub fields: Option<Vec<Field>>,
782    /// Key type; must be set when `type == "MAP"`.
783    #[cfg_attr(feature = "testing", proptest(value = "None"))]
784    pub key: Option<Box<ColumnType>>,
785    /// Value type; must be set when `type == "MAP"`.
786    #[cfg_attr(feature = "testing", proptest(value = "None"))]
787    pub value: Option<Box<ColumnType>>,
788}
789
790impl ColumnType {
791    pub fn boolean(nullable: bool) -> Self {
792        ColumnType {
793            typ: SqlType::Boolean,
794            nullable,
795            precision: None,
796            scale: None,
797            component: None,
798            fields: None,
799            key: None,
800            value: None,
801        }
802    }
803
804    pub fn uuid(nullable: bool) -> Self {
805        ColumnType {
806            typ: SqlType::Uuid,
807            nullable,
808            precision: None,
809            scale: None,
810            component: None,
811            fields: None,
812            key: None,
813            value: None,
814        }
815    }
816
817    pub fn tinyint(nullable: bool) -> Self {
818        ColumnType {
819            typ: SqlType::TinyInt,
820            nullable,
821            precision: None,
822            scale: None,
823            component: None,
824            fields: None,
825            key: None,
826            value: None,
827        }
828    }
829
830    pub fn smallint(nullable: bool) -> Self {
831        ColumnType {
832            typ: SqlType::SmallInt,
833            nullable,
834            precision: None,
835            scale: None,
836            component: None,
837            fields: None,
838            key: None,
839            value: None,
840        }
841    }
842
843    pub fn int(nullable: bool) -> Self {
844        ColumnType {
845            typ: SqlType::Int,
846            nullable,
847            precision: None,
848            scale: None,
849            component: None,
850            fields: None,
851            key: None,
852            value: None,
853        }
854    }
855
856    pub fn bigint(nullable: bool) -> Self {
857        ColumnType {
858            typ: SqlType::BigInt,
859            nullable,
860            precision: None,
861            scale: None,
862            component: None,
863            fields: None,
864            key: None,
865            value: None,
866        }
867    }
868
869    pub fn utinyint(nullable: bool) -> Self {
870        ColumnType {
871            typ: SqlType::UTinyInt,
872            nullable,
873            precision: None,
874            scale: None,
875            component: None,
876            fields: None,
877            key: None,
878            value: None,
879        }
880    }
881
882    pub fn usmallint(nullable: bool) -> Self {
883        ColumnType {
884            typ: SqlType::USmallInt,
885            nullable,
886            precision: None,
887            scale: None,
888            component: None,
889            fields: None,
890            key: None,
891            value: None,
892        }
893    }
894
895    pub fn uint(nullable: bool) -> Self {
896        ColumnType {
897            typ: SqlType::UInt,
898            nullable,
899            precision: None,
900            scale: None,
901            component: None,
902            fields: None,
903            key: None,
904            value: None,
905        }
906    }
907
908    pub fn ubigint(nullable: bool) -> Self {
909        ColumnType {
910            typ: SqlType::UBigInt,
911            nullable,
912            precision: None,
913            scale: None,
914            component: None,
915            fields: None,
916            key: None,
917            value: None,
918        }
919    }
920
921    pub fn double(nullable: bool) -> Self {
922        ColumnType {
923            typ: SqlType::Double,
924            nullable,
925            precision: None,
926            scale: None,
927            component: None,
928            fields: None,
929            key: None,
930            value: None,
931        }
932    }
933
934    pub fn real(nullable: bool) -> Self {
935        ColumnType {
936            typ: SqlType::Real,
937            nullable,
938            precision: None,
939            scale: None,
940            component: None,
941            fields: None,
942            key: None,
943            value: None,
944        }
945    }
946
947    pub fn decimal(precision: i64, scale: i64, nullable: bool) -> Self {
948        ColumnType {
949            typ: SqlType::Decimal,
950            nullable,
951            precision: Some(precision),
952            scale: Some(scale),
953            component: None,
954            fields: None,
955            key: None,
956            value: None,
957        }
958    }
959
960    pub fn varchar(nullable: bool) -> Self {
961        ColumnType {
962            typ: SqlType::Varchar,
963            nullable,
964            precision: None,
965            scale: None,
966            component: None,
967            fields: None,
968            key: None,
969            value: None,
970        }
971    }
972
973    pub fn varbinary(nullable: bool) -> Self {
974        ColumnType {
975            typ: SqlType::Varbinary,
976            nullable,
977            precision: None,
978            scale: None,
979            component: None,
980            fields: None,
981            key: None,
982            value: None,
983        }
984    }
985
986    pub fn fixed(width: i64, nullable: bool) -> Self {
987        ColumnType {
988            typ: SqlType::Binary,
989            nullable,
990            precision: Some(width),
991            scale: None,
992            component: None,
993            fields: None,
994            key: None,
995            value: None,
996        }
997    }
998
999    pub fn date(nullable: bool) -> Self {
1000        ColumnType {
1001            typ: SqlType::Date,
1002            nullable,
1003            precision: None,
1004            scale: None,
1005            component: None,
1006            fields: None,
1007            key: None,
1008            value: None,
1009        }
1010    }
1011
1012    pub fn time(nullable: bool) -> Self {
1013        ColumnType {
1014            typ: SqlType::Time,
1015            nullable,
1016            precision: None,
1017            scale: None,
1018            component: None,
1019            fields: None,
1020            key: None,
1021            value: None,
1022        }
1023    }
1024
1025    pub fn timestamp(nullable: bool) -> Self {
1026        ColumnType {
1027            typ: SqlType::Timestamp,
1028            nullable,
1029            precision: None,
1030            scale: None,
1031            component: None,
1032            fields: None,
1033            key: None,
1034            value: None,
1035        }
1036    }
1037
1038    pub fn timestamp_tz(nullable: bool) -> Self {
1039        ColumnType {
1040            typ: SqlType::TimestampTz,
1041            nullable,
1042            precision: None,
1043            scale: None,
1044            component: None,
1045            fields: None,
1046            key: None,
1047            value: None,
1048        }
1049    }
1050
1051    pub fn variant(nullable: bool) -> Self {
1052        ColumnType {
1053            typ: SqlType::Variant,
1054            nullable,
1055            precision: None,
1056            scale: None,
1057            component: None,
1058            fields: None,
1059            key: None,
1060            value: None,
1061        }
1062    }
1063
1064    pub fn array(nullable: bool, element: ColumnType) -> Self {
1065        ColumnType {
1066            typ: SqlType::Array,
1067            nullable,
1068            precision: None,
1069            scale: None,
1070            component: Some(Box::new(element)),
1071            fields: None,
1072            key: None,
1073            value: None,
1074        }
1075    }
1076
1077    pub fn structure(nullable: bool, fields: &[Field]) -> Self {
1078        ColumnType {
1079            typ: SqlType::Struct,
1080            nullable,
1081            precision: None,
1082            scale: None,
1083            component: None,
1084            fields: Some(fields.to_vec()),
1085            key: None,
1086            value: None,
1087        }
1088    }
1089
1090    pub fn map(nullable: bool, key: ColumnType, val: ColumnType) -> Self {
1091        ColumnType {
1092            typ: SqlType::Map,
1093            nullable,
1094            precision: None,
1095            scale: None,
1096            component: None,
1097            fields: None,
1098            key: Some(Box::new(key)),
1099            value: Some(Box::new(val)),
1100        }
1101    }
1102
1103    pub fn is_integral_type(&self) -> bool {
1104        matches!(
1105            &self.typ,
1106            SqlType::TinyInt
1107                | SqlType::SmallInt
1108                | SqlType::Int
1109                | SqlType::BigInt
1110                | SqlType::UTinyInt
1111                | SqlType::USmallInt
1112                | SqlType::UInt
1113                | SqlType::UBigInt
1114        )
1115    }
1116
1117    pub fn is_fp_type(&self) -> bool {
1118        matches!(&self.typ, SqlType::Double | SqlType::Real)
1119    }
1120
1121    pub fn is_decimal_type(&self) -> bool {
1122        matches!(&self.typ, SqlType::Decimal)
1123    }
1124
1125    pub fn is_numeric_type(&self) -> bool {
1126        self.is_integral_type() || self.is_fp_type() || self.is_decimal_type()
1127    }
1128}
1129
1130#[cfg(test)]
1131mod tests {
1132    use super::{IntervalUnit, SqlIdentifier};
1133    use crate::program_schema::{ColumnType, Field, SQL_TYPE_VALUES, SqlType};
1134
1135    /// Every `SqlType` variant, with intervals expanded to one entry per unit.
1136    fn all_sql_types() -> Vec<SqlType> {
1137        let mut types = vec![
1138            SqlType::Boolean,
1139            SqlType::TinyInt,
1140            SqlType::SmallInt,
1141            SqlType::Int,
1142            SqlType::BigInt,
1143            SqlType::UTinyInt,
1144            SqlType::USmallInt,
1145            SqlType::UInt,
1146            SqlType::UBigInt,
1147            SqlType::Real,
1148            SqlType::Double,
1149            SqlType::Decimal,
1150            SqlType::Char,
1151            SqlType::Varchar,
1152            SqlType::Binary,
1153            SqlType::Varbinary,
1154            SqlType::Time,
1155            SqlType::Date,
1156            SqlType::Timestamp,
1157            SqlType::TimestampTz,
1158        ];
1159        types.extend(
1160            [
1161                IntervalUnit::Day,
1162                IntervalUnit::DayToHour,
1163                IntervalUnit::DayToMinute,
1164                IntervalUnit::DayToSecond,
1165                IntervalUnit::Hour,
1166                IntervalUnit::HourToMinute,
1167                IntervalUnit::HourToSecond,
1168                IntervalUnit::Minute,
1169                IntervalUnit::MinuteToSecond,
1170                IntervalUnit::Month,
1171                IntervalUnit::Second,
1172                IntervalUnit::Year,
1173                IntervalUnit::YearToMonth,
1174            ]
1175            .map(SqlType::Interval),
1176        );
1177        types.extend([
1178            SqlType::Array,
1179            SqlType::Struct,
1180            SqlType::Map,
1181            SqlType::Null,
1182            SqlType::Uuid,
1183            SqlType::Variant,
1184        ]);
1185        types
1186    }
1187
1188    /// The OpenAPI schema is generated from `SQL_TYPE_VALUES`, while the wire
1189    /// format comes from `Serialize`. This guards that the two never drift:
1190    /// every variant serializes to a value the schema advertises, and the
1191    /// schema lists exactly those values (no stale or missing entries).
1192    #[test]
1193    fn sql_type_schema_matches_serialization() {
1194        let serialized: Vec<String> = all_sql_types()
1195            .iter()
1196            .map(|t| {
1197                serde_json::to_value(t)
1198                    .unwrap()
1199                    .as_str()
1200                    .unwrap()
1201                    .to_owned()
1202            })
1203            .collect();
1204
1205        // Every serialized value is advertised by the schema, and vice versa.
1206        assert_eq!(
1207            serialized,
1208            SQL_TYPE_VALUES
1209                .iter()
1210                .map(|s| s.to_string())
1211                .collect::<Vec<_>>(),
1212            "SQL_TYPE_VALUES (the OpenAPI enum) must match `Serialize` output exactly"
1213        );
1214
1215        // `as_wire_str` and `Serialize` must agree.
1216        for t in all_sql_types() {
1217            assert_eq!(
1218                serde_json::to_value(t).unwrap().as_str().unwrap(),
1219                t.as_wire_str()
1220            );
1221        }
1222    }
1223
1224    /// Only the exact property value "true" enables `skip_unused_columns`;
1225    /// connectors have always compared the value verbatim.
1226    #[test]
1227    fn relation_skip_unused_columns_property() {
1228        use super::{PropertyValue, Relation, SourcePosition};
1229        use std::collections::BTreeMap;
1230
1231        let zero = SourcePosition {
1232            start_line_number: 0,
1233            start_column: 0,
1234            end_line_number: 0,
1235            end_column: 0,
1236        };
1237        let relation = |value: Option<&str>| {
1238            let mut properties = BTreeMap::new();
1239            if let Some(value) = value {
1240                properties.insert(
1241                    "skip_unused_columns".to_string(),
1242                    PropertyValue {
1243                        value: value.to_string(),
1244                        key_position: zero,
1245                        value_position: zero,
1246                    },
1247                );
1248            }
1249            Relation::new(
1250                SqlIdentifier::new("t", false),
1251                Vec::new(),
1252                false,
1253                properties,
1254            )
1255        };
1256
1257        assert!(relation(Some("true")).skip_unused_columns());
1258        assert!(!relation(Some("false")).skip_unused_columns());
1259        assert!(!relation(Some("TRUE")).skip_unused_columns());
1260        assert!(!relation(None).skip_unused_columns());
1261    }
1262
1263    #[test]
1264    fn serde_sql_type() {
1265        for (sql_str_base, expected_value) in [
1266            ("Boolean", SqlType::Boolean),
1267            ("Uuid", SqlType::Uuid),
1268            ("TinyInt", SqlType::TinyInt),
1269            ("SmallInt", SqlType::SmallInt),
1270            ("Integer", SqlType::Int),
1271            ("BigInt", SqlType::BigInt),
1272            ("UTinyInt", SqlType::UTinyInt),
1273            ("USmallInt", SqlType::USmallInt),
1274            ("UInteger", SqlType::UInt),
1275            ("UBigInt", SqlType::UBigInt),
1276            ("Real", SqlType::Real),
1277            ("Double", SqlType::Double),
1278            ("Decimal", SqlType::Decimal),
1279            ("Char", SqlType::Char),
1280            ("Varchar", SqlType::Varchar),
1281            ("Binary", SqlType::Binary),
1282            ("Varbinary", SqlType::Varbinary),
1283            ("Time", SqlType::Time),
1284            ("Date", SqlType::Date),
1285            ("Timestamp", SqlType::Timestamp),
1286            ("Timestamp_Tz", SqlType::TimestampTz),
1287            ("Interval_Day", SqlType::Interval(IntervalUnit::Day)),
1288            (
1289                "Interval_Day_Hour",
1290                SqlType::Interval(IntervalUnit::DayToHour),
1291            ),
1292            (
1293                "Interval_Day_Minute",
1294                SqlType::Interval(IntervalUnit::DayToMinute),
1295            ),
1296            (
1297                "Interval_Day_Second",
1298                SqlType::Interval(IntervalUnit::DayToSecond),
1299            ),
1300            ("Interval_Hour", SqlType::Interval(IntervalUnit::Hour)),
1301            (
1302                "Interval_Hour_Minute",
1303                SqlType::Interval(IntervalUnit::HourToMinute),
1304            ),
1305            (
1306                "Interval_Hour_Second",
1307                SqlType::Interval(IntervalUnit::HourToSecond),
1308            ),
1309            ("Interval_Minute", SqlType::Interval(IntervalUnit::Minute)),
1310            (
1311                "Interval_Minute_Second",
1312                SqlType::Interval(IntervalUnit::MinuteToSecond),
1313            ),
1314            ("Interval_Month", SqlType::Interval(IntervalUnit::Month)),
1315            ("Interval_Second", SqlType::Interval(IntervalUnit::Second)),
1316            ("Interval_Year", SqlType::Interval(IntervalUnit::Year)),
1317            (
1318                "Interval_Year_Month",
1319                SqlType::Interval(IntervalUnit::YearToMonth),
1320            ),
1321            ("Array", SqlType::Array),
1322            ("Struct", SqlType::Struct),
1323            ("Map", SqlType::Map),
1324            ("Null", SqlType::Null),
1325            ("Variant", SqlType::Variant),
1326        ] {
1327            for sql_str in [
1328                sql_str_base,                 // Capitalized
1329                &sql_str_base.to_lowercase(), // lowercase
1330                &sql_str_base.to_uppercase(), // UPPERCASE
1331            ] {
1332                let value1: SqlType = serde_json::from_str(&format!("\"{}\"", sql_str))
1333                    .unwrap_or_else(|e| {
1334                        panic!(
1335                            "\"{sql_str}\" should deserialize into its SQL type: {}",
1336                            e.to_string()
1337                        )
1338                    });
1339                assert_eq!(value1, expected_value);
1340                let serialized_str =
1341                    serde_json::to_string(&value1).expect("Value should serialize into JSON");
1342                let value2: SqlType = serde_json::from_str(&serialized_str).unwrap_or_else(|_| {
1343                    panic!(
1344                        "{} should deserialize back into its SQL type",
1345                        serialized_str
1346                    )
1347                });
1348                assert_eq!(value1, value2);
1349            }
1350        }
1351    }
1352
1353    #[test]
1354    fn deserialize_interval_types() {
1355        use super::IntervalUnit::*;
1356        use super::SqlType::*;
1357
1358        let schema = r#"
1359{
1360  "inputs" : [ {
1361    "name" : "sales",
1362    "case_sensitive" : false,
1363    "fields" : [ {
1364      "name" : "sales_id",
1365      "case_sensitive" : false,
1366      "columntype" : {
1367        "type" : "INTEGER",
1368        "nullable" : true
1369      }
1370    }, {
1371      "name" : "customer_id",
1372      "case_sensitive" : false,
1373      "columntype" : {
1374        "type" : "INTEGER",
1375        "nullable" : true
1376      }
1377    }, {
1378      "name" : "age",
1379      "case_sensitive" : false,
1380      "columntype" : {
1381        "type" : "UINTEGER",
1382        "nullable" : true
1383      }
1384    }, {
1385      "name" : "amount",
1386      "case_sensitive" : false,
1387      "columntype" : {
1388        "type" : "DECIMAL",
1389        "nullable" : true,
1390        "precision" : 10,
1391        "scale" : 2
1392      }
1393    }, {
1394      "name" : "sale_date",
1395      "case_sensitive" : false,
1396      "columntype" : {
1397        "type" : "DATE",
1398        "nullable" : true
1399      }
1400    } ],
1401    "primary_key" : [ "sales_id" ]
1402  } ],
1403  "outputs" : [ {
1404    "name" : "salessummary",
1405    "case_sensitive" : false,
1406    "fields" : [ {
1407      "name" : "customer_id",
1408      "case_sensitive" : false,
1409      "columntype" : {
1410        "type" : "INTEGER",
1411        "nullable" : true
1412      }
1413    }, {
1414      "name" : "total_sales",
1415      "case_sensitive" : false,
1416      "columntype" : {
1417        "type" : "DECIMAL",
1418        "nullable" : true,
1419        "precision" : 38,
1420        "scale" : 2
1421      }
1422    }, {
1423      "name" : "interval_day",
1424      "case_sensitive" : false,
1425      "columntype" : {
1426        "type" : "INTERVAL_DAY",
1427        "nullable" : false,
1428        "precision" : 2,
1429        "scale" : 6
1430      }
1431    }, {
1432      "name" : "interval_day_to_hour",
1433      "case_sensitive" : false,
1434      "columntype" : {
1435        "type" : "INTERVAL_DAY_HOUR",
1436        "nullable" : false,
1437        "precision" : 2,
1438        "scale" : 6
1439      }
1440    }, {
1441      "name" : "interval_day_to_minute",
1442      "case_sensitive" : false,
1443      "columntype" : {
1444        "type" : "INTERVAL_DAY_MINUTE",
1445        "nullable" : false,
1446        "precision" : 2,
1447        "scale" : 6
1448      }
1449    }, {
1450      "name" : "interval_day_to_second",
1451      "case_sensitive" : false,
1452      "columntype" : {
1453        "type" : "INTERVAL_DAY_SECOND",
1454        "nullable" : false,
1455        "precision" : 2,
1456        "scale" : 6
1457      }
1458    }, {
1459      "name" : "interval_hour",
1460      "case_sensitive" : false,
1461      "columntype" : {
1462        "type" : "INTERVAL_HOUR",
1463        "nullable" : false,
1464        "precision" : 2,
1465        "scale" : 6
1466      }
1467    }, {
1468      "name" : "interval_hour_to_minute",
1469      "case_sensitive" : false,
1470      "columntype" : {
1471        "type" : "INTERVAL_HOUR_MINUTE",
1472        "nullable" : false,
1473        "precision" : 2,
1474        "scale" : 6
1475      }
1476    }, {
1477      "name" : "interval_hour_to_second",
1478      "case_sensitive" : false,
1479      "columntype" : {
1480        "type" : "INTERVAL_HOUR_SECOND",
1481        "nullable" : false,
1482        "precision" : 2,
1483        "scale" : 6
1484      }
1485    }, {
1486      "name" : "interval_minute",
1487      "case_sensitive" : false,
1488      "columntype" : {
1489        "type" : "INTERVAL_MINUTE",
1490        "nullable" : false,
1491        "precision" : 2,
1492        "scale" : 6
1493      }
1494    }, {
1495      "name" : "interval_minute_to_second",
1496      "case_sensitive" : false,
1497      "columntype" : {
1498        "type" : "INTERVAL_MINUTE_SECOND",
1499        "nullable" : false,
1500        "precision" : 2,
1501        "scale" : 6
1502      }
1503    }, {
1504      "name" : "interval_month",
1505      "case_sensitive" : false,
1506      "columntype" : {
1507        "type" : "INTERVAL_MONTH",
1508        "nullable" : false
1509      }
1510    }, {
1511      "name" : "interval_second",
1512      "case_sensitive" : false,
1513      "columntype" : {
1514        "type" : "INTERVAL_SECOND",
1515        "nullable" : false,
1516        "precision" : 2,
1517        "scale" : 6
1518      }
1519    }, {
1520      "name" : "interval_year",
1521      "case_sensitive" : false,
1522      "columntype" : {
1523        "type" : "INTERVAL_YEAR",
1524        "nullable" : false
1525      }
1526    }, {
1527      "name" : "interval_year_to_month",
1528      "case_sensitive" : false,
1529      "columntype" : {
1530        "type" : "INTERVAL_YEAR_MONTH",
1531        "nullable" : false
1532      }
1533    } ]
1534  } ]
1535}
1536"#;
1537
1538        let schema: super::ProgramSchema = serde_json::from_str(schema).unwrap();
1539        let types = schema
1540            .outputs
1541            .iter()
1542            .flat_map(|r| r.fields.iter().map(|f| f.columntype.typ));
1543        let expected_types = [
1544            Int,
1545            Decimal,
1546            Interval(Day),
1547            Interval(DayToHour),
1548            Interval(DayToMinute),
1549            Interval(DayToSecond),
1550            Interval(Hour),
1551            Interval(HourToMinute),
1552            Interval(HourToSecond),
1553            Interval(Minute),
1554            Interval(MinuteToSecond),
1555            Interval(Month),
1556            Interval(Second),
1557            Interval(Year),
1558            Interval(YearToMonth),
1559        ];
1560
1561        assert_eq!(types.collect::<Vec<_>>(), &expected_types);
1562    }
1563
1564    #[test]
1565    fn serialize_struct_schemas() {
1566        let schema = r#"{
1567  "inputs" : [ {
1568    "name" : "PERS",
1569    "case_sensitive" : false,
1570    "fields" : [ {
1571      "name" : "P0",
1572      "case_sensitive" : false,
1573      "columntype" : {
1574        "fields" : [ {
1575          "type" : "VARCHAR",
1576          "nullable" : true,
1577          "precision" : 30,
1578          "name" : "FIRSTNAME"
1579        }, {
1580          "type" : "VARCHAR",
1581          "nullable" : true,
1582          "precision" : 30,
1583          "name" : "LASTNAME"
1584        }, {
1585          "type" : "UINTEGER",
1586          "nullable" : true,
1587          "name" : "AGE"
1588        }, {
1589          "fields" : {
1590            "fields" : [ {
1591              "type" : "VARCHAR",
1592              "nullable" : true,
1593              "precision" : 30,
1594              "name" : "STREET"
1595            }, {
1596              "type" : "VARCHAR",
1597              "nullable" : true,
1598              "precision" : 30,
1599              "name" : "CITY"
1600            }, {
1601              "type" : "CHAR",
1602              "nullable" : true,
1603              "precision" : 2,
1604              "name" : "STATE"
1605            }, {
1606              "type" : "VARCHAR",
1607              "nullable" : true,
1608              "precision" : 6,
1609              "name" : "POSTAL_CODE"
1610            } ],
1611            "nullable" : false
1612          },
1613          "nullable" : false,
1614          "name" : "ADDRESS"
1615        } ],
1616        "nullable" : false
1617      }
1618    }]
1619  } ],
1620  "outputs" : [ ]
1621}
1622"#;
1623        let schema: super::ProgramSchema = serde_json::from_str(schema).unwrap();
1624        eprintln!("{:#?}", schema);
1625        let pers = schema.inputs.iter().find(|r| r.name == "PERS").unwrap();
1626        let p0 = pers.fields.iter().find(|f| f.name == "P0").unwrap();
1627        assert_eq!(p0.columntype.typ, SqlType::Struct);
1628        let p0_fields = p0.columntype.fields.as_ref().unwrap();
1629        assert_eq!(p0_fields[0].columntype.typ, SqlType::Varchar);
1630        assert_eq!(p0_fields[1].columntype.typ, SqlType::Varchar);
1631        assert_eq!(p0_fields[2].columntype.typ, SqlType::UInt);
1632        assert_eq!(p0_fields[3].columntype.typ, SqlType::Struct);
1633        assert_eq!(p0_fields[3].name, "ADDRESS");
1634        let address = &p0_fields[3].columntype.fields.as_ref().unwrap();
1635        assert_eq!(address.len(), 4);
1636        assert_eq!(address[0].name, "STREET");
1637        assert_eq!(address[0].columntype.typ, SqlType::Varchar);
1638        assert_eq!(address[1].columntype.typ, SqlType::Varchar);
1639        assert_eq!(address[2].columntype.typ, SqlType::Char);
1640        assert_eq!(address[3].columntype.typ, SqlType::Varchar);
1641    }
1642
1643    #[test]
1644    fn sql_identifier_cmp() {
1645        assert_eq!(SqlIdentifier::from("foo"), SqlIdentifier::from("foo"));
1646        assert_ne!(SqlIdentifier::from("foo"), SqlIdentifier::from("bar"));
1647        assert_eq!(SqlIdentifier::from("bar"), SqlIdentifier::from("BAR"));
1648        assert_eq!(SqlIdentifier::from("foo"), SqlIdentifier::from("\"foo\""));
1649        assert_eq!(SqlIdentifier::from("bar"), SqlIdentifier::from("\"bar\""));
1650        assert_eq!(SqlIdentifier::from("bAr"), SqlIdentifier::from("\"bAr\""));
1651        assert_eq!(
1652            SqlIdentifier::new("bAr", true),
1653            SqlIdentifier::from("\"bAr\"")
1654        );
1655
1656        assert_eq!(SqlIdentifier::from("bAr"), "bar");
1657        assert_eq!(SqlIdentifier::from("bAr"), "bAr");
1658    }
1659
1660    #[test]
1661    fn sql_identifier_ord() {
1662        let mut btree = std::collections::BTreeSet::new();
1663        assert!(btree.insert(SqlIdentifier::from("foo")));
1664        assert!(btree.insert(SqlIdentifier::from("bar")));
1665        assert!(!btree.insert(SqlIdentifier::from("BAR")));
1666        assert!(!btree.insert(SqlIdentifier::from("\"foo\"")));
1667        assert!(!btree.insert(SqlIdentifier::from("\"bar\"")));
1668    }
1669
1670    #[test]
1671    fn sql_identifier_hash() {
1672        let mut hs = std::collections::HashSet::new();
1673        assert!(hs.insert(SqlIdentifier::from("foo")));
1674        assert!(hs.insert(SqlIdentifier::from("bar")));
1675        assert!(!hs.insert(SqlIdentifier::from("BAR")));
1676        assert!(!hs.insert(SqlIdentifier::from("\"foo\"")));
1677        assert!(!hs.insert(SqlIdentifier::from("\"bar\"")));
1678    }
1679
1680    #[test]
1681    fn sql_identifier_name() {
1682        assert_eq!(SqlIdentifier::from("foo").name(), "foo");
1683        assert_eq!(SqlIdentifier::from("bAr").name(), "bar");
1684        assert_eq!(SqlIdentifier::from("\"bAr\"").name(), "bAr");
1685        assert_eq!(SqlIdentifier::from("foo").sql_name(), "foo");
1686        assert_eq!(SqlIdentifier::from("bAr").sql_name(), "bAr");
1687        assert_eq!(SqlIdentifier::from("\"bAr\"").sql_name(), "\"bAr\"");
1688    }
1689
1690    #[test]
1691    fn issue3277() {
1692        let schema = r#"{
1693      "name" : "j",
1694      "case_sensitive" : false,
1695      "columntype" : {
1696        "fields" : [ {
1697          "key" : {
1698            "nullable" : false,
1699            "precision" : -1,
1700            "type" : "VARCHAR"
1701          },
1702          "name" : "s",
1703          "nullable" : true,
1704          "type" : "MAP",
1705          "value" : {
1706            "nullable" : true,
1707            "precision" : -1,
1708            "type" : "VARCHAR"
1709          }
1710        } ],
1711        "nullable" : true
1712      }
1713    }"#;
1714        let field: Field = serde_json::from_str(schema).unwrap();
1715        println!("field: {:#?}", field);
1716        assert_eq!(
1717            field,
1718            Field {
1719                name: SqlIdentifier {
1720                    name: "j".to_string(),
1721                    case_sensitive: false,
1722                },
1723                columntype: ColumnType {
1724                    typ: SqlType::Struct,
1725                    nullable: true,
1726                    precision: None,
1727                    scale: None,
1728                    component: None,
1729                    fields: Some(vec![Field {
1730                        name: SqlIdentifier {
1731                            name: "s".to_string(),
1732                            case_sensitive: false,
1733                        },
1734                        columntype: ColumnType {
1735                            typ: SqlType::Map,
1736                            nullable: true,
1737                            precision: None,
1738                            scale: None,
1739                            component: None,
1740                            fields: None,
1741                            key: Some(Box::new(ColumnType {
1742                                typ: SqlType::Varchar,
1743                                nullable: false,
1744                                precision: Some(-1),
1745                                scale: None,
1746                                component: None,
1747                                fields: None,
1748                                key: None,
1749                                value: None,
1750                            })),
1751                            value: Some(Box::new(ColumnType {
1752                                typ: SqlType::Varchar,
1753                                nullable: true,
1754                                precision: Some(-1),
1755                                scale: None,
1756                                component: None,
1757                                fields: None,
1758                                key: None,
1759                                value: None,
1760                            })),
1761                        },
1762                        lateness: None,
1763                        default: None,
1764                        unused: false,
1765                    }]),
1766                    key: None,
1767                    value: None,
1768                },
1769                lateness: None,
1770                default: None,
1771                unused: false,
1772            }
1773        );
1774    }
1775}
feldera_types/program_schema.rs

feldera_types/
program_schema.rs
