feldera_types/

program_schema.rs

use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::cmp::Ordering;
use std::collections::BTreeMap;
use std::fmt::Display;
use std::hash::{Hash, Hasher};
use utoipa::ToSchema;

#[cfg(feature = "testing")]
use proptest::{collection::vec, prelude::any};

/// Returns canonical form of a SQL identifier:
///
/// - If id is _not_ quoted, then it is interpreted as a case-insensitive
///   identifier and is converted to the lowercase representation
/// - If id _is_ quoted, then it is a case-sensitive identifier and is returned
///   as is, without quotes. No other processing is done on the inner string,
///   e.g., un-escaping quotes.
pub fn canonical_identifier(id: &str) -> String {
    if id.starts_with('"') && id.ends_with('"') && id.len() >= 2 {
        id[1..id.len() - 1].to_string()
    } else {
        id.to_lowercase()
    }
}

/// An SQL identifier.
///
/// This struct is used to represent SQL identifiers in a canonical form.
/// We store table names or field names as identifiers in the schema.
#[derive(Serialize, Deserialize, ToSchema, Debug, Clone)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub struct SqlIdentifier {
    #[cfg_attr(feature = "testing", proptest(regex = "relation1|relation2|relation3"))]
    name: String,
    pub case_sensitive: bool,
}

impl SqlIdentifier {
    pub fn new<S: AsRef<str>>(name: S, case_sensitive: bool) -> Self {
        Self {
            name: name.as_ref().to_string(),
            case_sensitive,
        }
    }

    /// Return the name of the identifier in canonical form.
    pub fn name(&self) -> String {
        if self.case_sensitive {
            self.name.clone()
        } else {
            self.name.to_lowercase()
        }
    }

    /// Return the name of the identifier as it would appear in SQL.
    pub fn sql_name(&self) -> String {
        if self.case_sensitive {
            format!("\"{}\"", self.name)
        } else {
            self.name.clone()
        }
    }
}

impl Hash for SqlIdentifier {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.name().hash(state);
    }
}

impl PartialEq for SqlIdentifier {
    fn eq(&self, other: &Self) -> bool {
        match (self.case_sensitive, other.case_sensitive) {
            (true, true) => self.name == other.name,
            (false, false) => self.name.to_lowercase() == other.name.to_lowercase(),
            (true, false) => self.name == other.name,
            (false, true) => self.name == other.name,
        }
    }
}

impl Ord for SqlIdentifier {
    fn cmp(&self, other: &Self) -> Ordering {
        self.name().cmp(&other.name())
    }
}

impl PartialOrd for SqlIdentifier {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl<S: AsRef<str>> PartialEq<S> for SqlIdentifier {
    fn eq(&self, other: &S) -> bool {
        self == &SqlIdentifier::from(other.as_ref())
    }
}

impl Eq for SqlIdentifier {}

impl<S: AsRef<str>> From<S> for SqlIdentifier {
    fn from(name: S) -> Self {
        if name.as_ref().starts_with('"')
            && name.as_ref().ends_with('"')
            && name.as_ref().len() >= 2
        {
            Self {
                name: name.as_ref()[1..name.as_ref().len() - 1].to_string(),
                case_sensitive: true,
            }
        } else {
            Self::new(name, false)
        }
    }
}

impl From<SqlIdentifier> for String {
    fn from(id: SqlIdentifier) -> String {
        id.name()
    }
}

impl From<&SqlIdentifier> for String {
    fn from(id: &SqlIdentifier) -> String {
        id.name()
    }
}

impl Display for SqlIdentifier {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.name())
    }
}

/// A struct containing the tables (inputs) and views for a program.
///
/// Parse from the JSON data-type of the DDL generated by the SQL compiler.
#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone, Default)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub struct ProgramSchema {
    #[cfg_attr(
        feature = "testing",
        proptest(strategy = "vec(any::<Relation>(), 0..2)")
    )]
    pub inputs: Vec<Relation>,
    #[cfg_attr(
        feature = "testing",
        proptest(strategy = "vec(any::<Relation>(), 0..2)")
    )]
    pub outputs: Vec<Relation>,
}

#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone, Copy)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub struct SourcePosition {
    pub start_line_number: usize,
    pub start_column: usize,
    pub end_line_number: usize,
    pub end_column: usize,
}

#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub struct PropertyValue {
    pub value: String,
    pub key_position: SourcePosition,
    pub value_position: SourcePosition,
}

/// A SQL table or view. It has a name and a list of fields.
///
/// Matches the Calcite JSON format.
#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub struct Relation {
    // This field should only be accessed via the `name()` method.
    #[serde(flatten)]
    pub name: SqlIdentifier,
    #[cfg_attr(feature = "testing", proptest(value = "Vec::new()"))]
    pub fields: Vec<Field>,
    #[serde(default)]
    pub materialized: bool,
    #[serde(default)]
    pub properties: BTreeMap<String, PropertyValue>,
}

impl Relation {
    pub fn empty() -> Self {
        Self {
            name: SqlIdentifier::from("".to_string()),
            fields: Vec::new(),
            materialized: false,
            properties: BTreeMap::new(),
        }
    }

    pub fn new(
        name: SqlIdentifier,
        fields: Vec<Field>,
        materialized: bool,
        properties: BTreeMap<String, PropertyValue>,
    ) -> Self {
        Self {
            name,
            fields,
            materialized,
            properties,
        }
    }

    /// Lookup field by name.
    pub fn field(&self, name: &str) -> Option<&Field> {
        let name = canonical_identifier(name);
        self.fields.iter().find(|f| f.name == name)
    }
}

/// A SQL field.
///
/// Matches the SQL compiler JSON format.
#[derive(Serialize, ToSchema, Debug, Eq, PartialEq, Clone)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub struct Field {
    #[serde(flatten)]
    pub name: SqlIdentifier,
    pub columntype: ColumnType,
}

/// Thanks to the brain-dead Calcite schema, if we are deserializing a field, the type options
/// end up inside the Field struct.
///
/// This helper struct is used to deserialize the Field struct.
impl<'de> Deserialize<'de> for Field {
    fn deserialize<D>(deserializer: D) -> Result<Field, D::Error>
    where
        D: Deserializer<'de>,
    {
        const fn default_is_struct() -> Option<SqlType> {
            Some(SqlType::Struct)
        }

        #[derive(Debug, Clone, Deserialize)]
        struct FieldHelper {
            name: Option<String>,
            #[serde(default)]
            case_sensitive: bool,
            columntype: Option<ColumnType>,
            #[serde(rename = "type")]
            #[serde(default = "default_is_struct")]
            typ: Option<SqlType>,
            nullable: Option<bool>,
            precision: Option<i64>,
            scale: Option<i64>,
            component: Option<Box<ColumnType>>,
            fields: Option<serde_json::Value>,
        }

        fn helper_to_field(helper: FieldHelper) -> Field {
            let columntype = if let Some(ctype) = helper.columntype {
                ctype
            } else if let Some(serde_json::Value::Array(fields)) = helper.fields {
                let fields = fields
                    .into_iter()
                    .map(|field| {
                        let field: FieldHelper = serde_json::from_value(field).unwrap();
                        helper_to_field(field)
                    })
                    .collect::<Vec<Field>>();

                ColumnType {
                    typ: helper.typ.unwrap_or(SqlType::Null),
                    nullable: helper.nullable.unwrap_or(false),
                    precision: helper.precision,
                    scale: helper.scale,
                    component: helper.component,
                    fields: Some(fields),
                    key: None,
                    value: None,
                }
            } else if let Some(serde_json::Value::Object(obj)) = helper.fields {
                serde_json::from_value(serde_json::Value::Object(obj))
                    .expect("Failed to deserialize object")
            } else {
                ColumnType {
                    typ: helper.typ.unwrap_or(SqlType::Null),
                    nullable: helper.nullable.unwrap_or(false),
                    precision: helper.precision,
                    scale: helper.scale,
                    component: helper.component,
                    fields: None,
                    key: None,
                    value: None,
                }
            };

            Field {
                name: SqlIdentifier::new(helper.name.unwrap(), helper.case_sensitive),
                columntype,
            }
        }

        let helper = FieldHelper::deserialize(deserializer)?;
        Ok(helper_to_field(helper))
    }
}

/// The specified units for SQL Interval types.
///
/// `INTERVAL 1 DAY`, `INTERVAL 1 DAY TO HOUR`, `INTERVAL 1 DAY TO MINUTE`,
/// would yield `Day`, `DayToHour`, `DayToMinute`, as the `IntervalUnit` respectively.
#[derive(ToSchema, Debug, Eq, PartialEq, Clone, Copy)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub enum IntervalUnit {
    /// Unit for `INTERVAL ... DAY`.
    Day,
    /// Unit for `INTERVAL ... DAY TO HOUR`.
    DayToHour,
    /// Unit for `INTERVAL ... DAY TO MINUTE`.
    DayToMinute,
    /// Unit for `INTERVAL ... DAY TO SECOND`.
    DayToSecond,
    /// Unit for `INTERVAL ... HOUR`.
    Hour,
    /// Unit for `INTERVAL ... HOUR TO MINUTE`.
    HourToMinute,
    /// Unit for `INTERVAL ... HOUR TO SECOND`.
    HourToSecond,
    /// Unit for `INTERVAL ... MINUTE`.
    Minute,
    /// Unit for `INTERVAL ... MINUTE TO SECOND`.
    MinuteToSecond,
    /// Unit for `INTERVAL ... MONTH`.
    Month,
    /// Unit for `INTERVAL ... SECOND`.
    Second,
    /// Unit for `INTERVAL ... YEAR`.
    Year,
    /// Unit for `INTERVAL ... YEAR TO MONTH`.
    YearToMonth,
}

/// The available SQL types as specified in `CREATE` statements.
#[derive(ToSchema, Debug, Eq, PartialEq, Clone, Copy)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub enum SqlType {
    /// SQL `BOOLEAN` type.
    Boolean,
    /// SQL `TINYINT` type.
    TinyInt,
    /// SQL `SMALLINT` or `INT2` type.
    SmallInt,
    /// SQL `INTEGER`, `INT`, `SIGNED`, `INT4` type.
    Int,
    /// SQL `BIGINT` or `INT64` type.
    BigInt,
    /// SQL `REAL` or `FLOAT4` or `FLOAT32` type.
    Real,
    /// SQL `DOUBLE` or `FLOAT8` or `FLOAT64` type.
    Double,
    /// SQL `DECIMAL` or `DEC` or `NUMERIC` type.
    Decimal,
    /// SQL `CHAR(n)` or `CHARACTER(n)` type.
    Char,
    /// SQL `VARCHAR`, `CHARACTER VARYING`, `TEXT`, or `STRING` type.
    Varchar,
    /// SQL `BINARY(n)` type.
    Binary,
    /// SQL `VARBINARY` or `BYTEA` type.
    Varbinary,
    /// SQL `TIME` type.
    Time,
    /// SQL `DATE` type.
    Date,
    /// SQL `TIMESTAMP` type.
    Timestamp,
    /// SQL `INTERVAL ... X` type where `X` is a unit.
    Interval(IntervalUnit),
    /// SQL `ARRAY` type.
    Array,
    /// A complex SQL struct type (`CREATE TYPE x ...`).
    Struct,
    /// SQL `MAP` type.
    Map,
    /// SQL `NULL` type.
    Null,
    /// SQL `VARIANT` type.
    Variant,
}

impl Display for SqlType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&serde_json::to_string(self).unwrap())
    }
}

impl<'de> Deserialize<'de> for SqlType {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value: String = Deserialize::deserialize(deserializer)?;
        match value.to_lowercase().as_str() {
            "interval_day" => Ok(SqlType::Interval(IntervalUnit::Day)),
            "interval_day_hour" => Ok(SqlType::Interval(IntervalUnit::DayToHour)),
            "interval_day_minute" => Ok(SqlType::Interval(IntervalUnit::DayToMinute)),
            "interval_day_second" => Ok(SqlType::Interval(IntervalUnit::DayToSecond)),
            "interval_hour" => Ok(SqlType::Interval(IntervalUnit::Hour)),
            "interval_hour_minute" => Ok(SqlType::Interval(IntervalUnit::HourToMinute)),
            "interval_hour_second" => Ok(SqlType::Interval(IntervalUnit::HourToSecond)),
            "interval_minute" => Ok(SqlType::Interval(IntervalUnit::Minute)),
            "interval_minute_second" => Ok(SqlType::Interval(IntervalUnit::MinuteToSecond)),
            "interval_month" => Ok(SqlType::Interval(IntervalUnit::Month)),
            "interval_second" => Ok(SqlType::Interval(IntervalUnit::Second)),
            "interval_year" => Ok(SqlType::Interval(IntervalUnit::Year)),
            "interval_year_month" => Ok(SqlType::Interval(IntervalUnit::YearToMonth)),
            "boolean" => Ok(SqlType::Boolean),
            "tinyint" => Ok(SqlType::TinyInt),
            "smallint" => Ok(SqlType::SmallInt),
            "integer" => Ok(SqlType::Int),
            "bigint" => Ok(SqlType::BigInt),
            "real" => Ok(SqlType::Real),
            "double" => Ok(SqlType::Double),
            "decimal" => Ok(SqlType::Decimal),
            "char" => Ok(SqlType::Char),
            "varchar" => Ok(SqlType::Varchar),
            "binary" => Ok(SqlType::Binary),
            "varbinary" => Ok(SqlType::Varbinary),
            "variant" => Ok(SqlType::Variant),
            "time" => Ok(SqlType::Time),
            "date" => Ok(SqlType::Date),
            "timestamp" => Ok(SqlType::Timestamp),
            "array" => Ok(SqlType::Array),
            "struct" => Ok(SqlType::Struct),
            "map" => Ok(SqlType::Map),
            "null" => Ok(SqlType::Null),
            _ => Err(serde::de::Error::custom(format!(
                "Unknown SQL type: {}",
                value
            ))),
        }
    }
}

impl Serialize for SqlType {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let type_str = match self {
            SqlType::Boolean => "BOOLEAN",
            SqlType::TinyInt => "TINYINT",
            SqlType::SmallInt => "SMALLINT",
            SqlType::Int => "INTEGER",
            SqlType::BigInt => "BIGINT",
            SqlType::Real => "REAL",
            SqlType::Double => "DOUBLE",
            SqlType::Decimal => "DECIMAL",
            SqlType::Char => "CHAR",
            SqlType::Varchar => "VARCHAR",
            SqlType::Binary => "BINARY",
            SqlType::Varbinary => "VARBINARY",
            SqlType::Time => "TIME",
            SqlType::Date => "DATE",
            SqlType::Timestamp => "TIMESTAMP",
            SqlType::Interval(interval_unit) => match interval_unit {
                IntervalUnit::Day => "INTERVAL_DAY",
                IntervalUnit::DayToHour => "INTERVAL_DAY_HOUR",
                IntervalUnit::DayToMinute => "INTERVAL_DAY_MINUTE",
                IntervalUnit::DayToSecond => "INTERVAL_DAY_SECOND",
                IntervalUnit::Hour => "INTERVAL_HOUR",
                IntervalUnit::HourToMinute => "INTERVAL_HOUR_MINUTE",
                IntervalUnit::HourToSecond => "INTERVAL_HOUR_SECOND",
                IntervalUnit::Minute => "INTERVAL_MINUTE",
                IntervalUnit::MinuteToSecond => "INTERVAL_MINUTE_SECOND",
                IntervalUnit::Month => "INTERVAL_MONTH",
                IntervalUnit::Second => "INTERVAL_SECOND",
                IntervalUnit::Year => "INTERVAL_YEAR",
                IntervalUnit::YearToMonth => "INTERVAL_YEAR_MONTH",
            },
            SqlType::Array => "ARRAY",
            SqlType::Struct => "STRUCT",
            SqlType::Map => "MAP",
            SqlType::Null => "NULL",
            SqlType::Variant => "VARIANT",
        };
        serializer.serialize_str(type_str)
    }
}

impl SqlType {
    /// Is this a string type?
    pub fn is_string(&self) -> bool {
        matches!(self, Self::Char | Self::Varchar)
    }
}

/// It so happens that when the type field is missing in the Calcite schema, it's a struct,
/// so we use it as the default.
const fn default_is_struct() -> SqlType {
    SqlType::Struct
}

/// A SQL column type description.
///
/// Matches the Calcite JSON format.
#[derive(Serialize, Deserialize, ToSchema, Debug, Eq, PartialEq, Clone)]
#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
pub struct ColumnType {
    /// Identifier for the type (e.g., `VARCHAR`, `BIGINT`, `ARRAY` etc.)
    #[serde(rename = "type")]
    #[serde(default = "default_is_struct")]
    pub typ: SqlType,
    /// Does the type accept NULL values?
    pub nullable: bool,
    /// Precision of the type.
    ///
    /// # Examples
    /// - `VARCHAR` sets precision to `-1`.
    /// - `VARCHAR(255)` sets precision to `255`.
    /// - `BIGINT`, `DATE`, `FLOAT`, `DOUBLE`, `GEOMETRY`, etc. sets precision
    ///   to None
    /// - `TIME`, `TIMESTAMP` set precision to `0`.
    pub precision: Option<i64>,
    /// The scale of the type.
    ///
    /// # Example
    /// - `DECIMAL(1,2)` sets scale to `2`.
    pub scale: Option<i64>,
    /// A component of the type (if available).
    ///
    /// This is in a `Box` because it makes it a recursive types.
    ///
    /// For example, this would specify the `VARCHAR(20)` in the `VARCHAR(20)
    /// ARRAY` type.
    #[cfg_attr(feature = "testing", proptest(value = "None"))]
    pub component: Option<Box<ColumnType>>,
    /// The fields of the type (if available).
    ///
    /// For example this would specify the fields of a `CREATE TYPE` construct.
    ///
    /// ```sql
    /// CREATE TYPE person_typ AS (
    ///   firstname       VARCHAR(30),
    ///   lastname        VARCHAR(30),
    ///   address         ADDRESS_TYP
    /// );
    /// ```
    ///
    /// Would lead to the following `fields` value:
    ///
    /// ```sql
    /// [
    ///  ColumnType { name: "firstname, ... },
    ///  ColumnType { name: "lastname", ... },
    ///  ColumnType { name: "address", fields: [ ... ] }
    /// ]
    /// ```
    #[cfg_attr(feature = "testing", proptest(value = "Some(Vec::new())"))]
    pub fields: Option<Vec<Field>>,
    /// Key type; must be set when `type == "MAP"`.
    #[cfg_attr(feature = "testing", proptest(value = "None"))]
    pub key: Option<Box<ColumnType>>,
    /// Value type; must be set when `type == "MAP"`.
    #[cfg_attr(feature = "testing", proptest(value = "None"))]
    pub value: Option<Box<ColumnType>>,
}

impl ColumnType {
    pub fn boolean(nullable: bool) -> Self {
        ColumnType {
            typ: SqlType::Boolean,
            nullable,
            precision: None,
            scale: None,
            component: None,
            fields: None,
            key: None,
            value: None,
        }
    }

    pub fn tinyint(nullable: bool) -> Self {
        ColumnType {
            typ: SqlType::TinyInt,
            nullable,
            precision: None,
            scale: None,
            component: None,
            fields: None,
            key: None,
            value: None,
        }
    }

    pub fn smallint(nullable: bool) -> Self {
        ColumnType {
            typ: SqlType::SmallInt,
            nullable,
            precision: None,
            scale: None,
            component: None,
            fields: None,
            key: None,
            value: None,
        }
    }

    pub fn int(nullable: bool) -> Self {
        ColumnType {
            typ: SqlType::Int,
            nullable,
            precision: None,
            scale: None,
            component: None,
            fields: None,
            key: None,
            value: None,
        }
    }

    pub fn bigint(nullable: bool) -> Self {
        ColumnType {
            typ: SqlType::BigInt,
            nullable,
            precision: None,
            scale: None,
            component: None,
            fields: None,
            key: None,
            value: None,
        }
    }

    pub fn varchar(nullable: bool) -> Self {
        ColumnType {
            typ: SqlType::Varchar,
            nullable,
            precision: None,
            scale: None,
            component: None,
            fields: None,
            key: None,
            value: None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{IntervalUnit, SqlIdentifier};
    use crate::program_schema::SqlType;

    #[test]
    fn serde_sql_type() {
        for (sql_str_base, expected_value) in [
            ("Boolean", SqlType::Boolean),
            ("TinyInt", SqlType::TinyInt),
            ("SmallInt", SqlType::SmallInt),
            ("Integer", SqlType::Int),
            ("BigInt", SqlType::BigInt),
            ("Real", SqlType::Real),
            ("Double", SqlType::Double),
            ("Decimal", SqlType::Decimal),
            ("Char", SqlType::Char),
            ("Varchar", SqlType::Varchar),
            ("Binary", SqlType::Binary),
            ("Varbinary", SqlType::Varbinary),
            ("Time", SqlType::Time),
            ("Date", SqlType::Date),
            ("Timestamp", SqlType::Timestamp),
            ("Interval_Day", SqlType::Interval(IntervalUnit::Day)),
            (
                "Interval_Day_Hour",
                SqlType::Interval(IntervalUnit::DayToHour),
            ),
            (
                "Interval_Day_Minute",
                SqlType::Interval(IntervalUnit::DayToMinute),
            ),
            (
                "Interval_Day_Second",
                SqlType::Interval(IntervalUnit::DayToSecond),
            ),
            ("Interval_Hour", SqlType::Interval(IntervalUnit::Hour)),
            (
                "Interval_Hour_Minute",
                SqlType::Interval(IntervalUnit::HourToMinute),
            ),
            (
                "Interval_Hour_Second",
                SqlType::Interval(IntervalUnit::HourToSecond),
            ),
            ("Interval_Minute", SqlType::Interval(IntervalUnit::Minute)),
            (
                "Interval_Minute_Second",
                SqlType::Interval(IntervalUnit::MinuteToSecond),
            ),
            ("Interval_Month", SqlType::Interval(IntervalUnit::Month)),
            ("Interval_Second", SqlType::Interval(IntervalUnit::Second)),
            ("Interval_Year", SqlType::Interval(IntervalUnit::Year)),
            (
                "Interval_Year_Month",
                SqlType::Interval(IntervalUnit::YearToMonth),
            ),
            ("Array", SqlType::Array),
            ("Struct", SqlType::Struct),
            ("Map", SqlType::Map),
            ("Null", SqlType::Null),
            ("Variant", SqlType::Variant),
        ] {
            for sql_str in [
                sql_str_base,                 // Capitalized
                &sql_str_base.to_lowercase(), // lowercase
                &sql_str_base.to_uppercase(), // UPPERCASE
            ] {
                let value1: SqlType = serde_json::from_str(&format!("\"{}\"", sql_str))
                    .unwrap_or_else(|_| {
                        panic!("\"{sql_str}\" should deserialize into its SQL type")
                    });
                assert_eq!(value1, expected_value);
                let serialized_str =
                    serde_json::to_string(&value1).expect("Value should serialize into JSON");
                let value2: SqlType = serde_json::from_str(&serialized_str).unwrap_or_else(|_| {
                    panic!(
                        "{} should deserialize back into its SQL type",
                        serialized_str
                    )
                });
                assert_eq!(value1, value2);
            }
        }
    }

    #[test]
    fn deserialize_interval_types() {
        use super::IntervalUnit::*;
        use super::SqlType::*;

        let schema = r#"
{
  "inputs" : [ {
    "name" : "sales",
    "case_sensitive" : false,
    "fields" : [ {
      "name" : "sales_id",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTEGER",
        "nullable" : true
      }
    }, {
      "name" : "customer_id",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTEGER",
        "nullable" : true
      }
    }, {
      "name" : "amount",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "DECIMAL",
        "nullable" : true,
        "precision" : 10,
        "scale" : 2
      }
    }, {
      "name" : "sale_date",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "DATE",
        "nullable" : true
      }
    } ],
    "primary_key" : [ "sales_id" ]
  } ],
  "outputs" : [ {
    "name" : "salessummary",
    "case_sensitive" : false,
    "fields" : [ {
      "name" : "customer_id",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTEGER",
        "nullable" : true
      }
    }, {
      "name" : "total_sales",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "DECIMAL",
        "nullable" : true,
        "precision" : 38,
        "scale" : 2
      }
    }, {
      "name" : "interval_day",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_DAY",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_day_to_hour",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_DAY_HOUR",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_day_to_minute",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_DAY_MINUTE",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_day_to_second",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_DAY_SECOND",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_hour",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_HOUR",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_hour_to_minute",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_HOUR_MINUTE",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_hour_to_second",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_HOUR_SECOND",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_minute",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_MINUTE",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_minute_to_second",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_MINUTE_SECOND",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_month",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_MONTH",
        "nullable" : false
      }
    }, {
      "name" : "interval_second",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_SECOND",
        "nullable" : false,
        "precision" : 2,
        "scale" : 6
      }
    }, {
      "name" : "interval_year",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_YEAR",
        "nullable" : false
      }
    }, {
      "name" : "interval_year_to_month",
      "case_sensitive" : false,
      "columntype" : {
        "type" : "INTERVAL_YEAR_MONTH",
        "nullable" : false
      }
    } ]
  } ]
}
"#;

        let schema: super::ProgramSchema = serde_json::from_str(schema).unwrap();
        let types = schema
            .outputs
            .iter()
            .flat_map(|r| r.fields.iter().map(|f| f.columntype.typ));
        let expected_types = [
            Int,
            Decimal,
            Interval(Day),
            Interval(DayToHour),
            Interval(DayToMinute),
            Interval(DayToSecond),
            Interval(Hour),
            Interval(HourToMinute),
            Interval(HourToSecond),
            Interval(Minute),
            Interval(MinuteToSecond),
            Interval(Month),
            Interval(Second),
            Interval(Year),
            Interval(YearToMonth),
        ];

        assert_eq!(types.collect::<Vec<_>>(), &expected_types);
    }

    #[test]
    fn serialize_struct_schemas() {
        let schema = r#"{
  "inputs" : [ {
    "name" : "PERS",
    "case_sensitive" : false,
    "fields" : [ {
      "name" : "P0",
      "case_sensitive" : false,
      "columntype" : {
        "fields" : [ {
          "type" : "VARCHAR",
          "nullable" : true,
          "precision" : 30,
          "name" : "FIRSTNAME"
        }, {
          "type" : "VARCHAR",
          "nullable" : true,
          "precision" : 30,
          "name" : "LASTNAME"
        }, {
          "fields" : {
            "fields" : [ {
              "type" : "VARCHAR",
              "nullable" : true,
              "precision" : 30,
              "name" : "STREET"
            }, {
              "type" : "VARCHAR",
              "nullable" : true,
              "precision" : 30,
              "name" : "CITY"
            }, {
              "type" : "CHAR",
              "nullable" : true,
              "precision" : 2,
              "name" : "STATE"
            }, {
              "type" : "VARCHAR",
              "nullable" : true,
              "precision" : 6,
              "name" : "POSTAL_CODE"
            } ],
            "nullable" : false
          },
          "nullable" : false,
          "name" : "ADDRESS"
        } ],
        "nullable" : false
      }
    }]
  } ],
  "outputs" : [ ]
}
"#;
        let schema: super::ProgramSchema = serde_json::from_str(schema).unwrap();
        eprintln!("{:#?}", schema);
        let pers = schema.inputs.iter().find(|r| r.name == "PERS").unwrap();
        let p0 = pers.fields.iter().find(|f| f.name == "P0").unwrap();
        assert_eq!(p0.columntype.typ, SqlType::Struct);
        let p0_fields = p0.columntype.fields.as_ref().unwrap();
        assert_eq!(p0_fields[0].columntype.typ, SqlType::Varchar);
        assert_eq!(p0_fields[1].columntype.typ, SqlType::Varchar);
        assert_eq!(p0_fields[2].columntype.typ, SqlType::Struct);
        assert_eq!(p0_fields[2].name, "ADDRESS");
        let address = &p0_fields[2].columntype.fields.as_ref().unwrap();
        assert_eq!(address.len(), 4);
        assert_eq!(address[0].name, "STREET");
        assert_eq!(address[0].columntype.typ, SqlType::Varchar);
        assert_eq!(address[1].columntype.typ, SqlType::Varchar);
        assert_eq!(address[2].columntype.typ, SqlType::Char);
        assert_eq!(address[3].columntype.typ, SqlType::Varchar);
    }

    #[test]
    fn sql_identifier_cmp() {
        assert_eq!(SqlIdentifier::from("foo"), SqlIdentifier::from("foo"));
        assert_ne!(SqlIdentifier::from("foo"), SqlIdentifier::from("bar"));
        assert_eq!(SqlIdentifier::from("bar"), SqlIdentifier::from("BAR"));
        assert_eq!(SqlIdentifier::from("foo"), SqlIdentifier::from("\"foo\""));
        assert_eq!(SqlIdentifier::from("bar"), SqlIdentifier::from("\"bar\""));
        assert_eq!(SqlIdentifier::from("bAr"), SqlIdentifier::from("\"bAr\""));
        assert_eq!(
            SqlIdentifier::new("bAr", true),
            SqlIdentifier::from("\"bAr\"")
        );

        assert_eq!(SqlIdentifier::from("bAr"), "bar");
        assert_eq!(SqlIdentifier::from("bAr"), "bAr");
    }

    #[test]
    fn sql_identifier_ord() {
        let mut btree = std::collections::BTreeSet::new();
        assert!(btree.insert(SqlIdentifier::from("foo")));
        assert!(btree.insert(SqlIdentifier::from("bar")));
        assert!(!btree.insert(SqlIdentifier::from("BAR")));
        assert!(!btree.insert(SqlIdentifier::from("\"foo\"")));
        assert!(!btree.insert(SqlIdentifier::from("\"bar\"")));
    }

    #[test]
    fn sql_identifier_hash() {
        let mut hs = std::collections::HashSet::new();
        assert!(hs.insert(SqlIdentifier::from("foo")));
        assert!(hs.insert(SqlIdentifier::from("bar")));
        assert!(!hs.insert(SqlIdentifier::from("BAR")));
        assert!(!hs.insert(SqlIdentifier::from("\"foo\"")));
        assert!(!hs.insert(SqlIdentifier::from("\"bar\"")));
    }

    #[test]
    fn sql_identifier_name() {
        assert_eq!(SqlIdentifier::from("foo").name(), "foo");
        assert_eq!(SqlIdentifier::from("bAr").name(), "bar");
        assert_eq!(SqlIdentifier::from("\"bAr\"").name(), "bAr");
        assert_eq!(SqlIdentifier::from("foo").sql_name(), "foo");
        assert_eq!(SqlIdentifier::from("bAr").sql_name(), "bAr");
        assert_eq!(SqlIdentifier::from("\"bAr\"").sql_name(), "\"bAr\"");
    }
}