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

feldera_types/
program_schema.rs
