Port Arroyo patches to sqlparser 0.62

examples/cli.rs

@@ -71,9 +71,9 @@ $ cargo run --example cli - [--dialectname]
             .expect("failed to read from stdin");
         String::from_utf8(buf).expect("stdin content wasn't valid utf8")
     } else {
-        println!("Parsing from file '{}' using {:?}", &filename, dialect);
+        println!("Parsing from file '{}' using {:?}", filename, dialect);
         fs::read_to_string(&filename)
-            .unwrap_or_else(|_| panic!("Unable to read the file {}", &filename))
+            .unwrap_or_else(|_| panic!("Unable to read the file {}", filename))
     };
     let without_bom = if contents.chars().next().unwrap() as u64 != 0xfeff {
         contents.as_str()

src/ast/ddl.rs

@@ -1939,6 +1939,20 @@ pub enum ColumnOption {
     Comment(String),
     /// `ON UPDATE <expr>` column option
     OnUpdate(Expr),
+    /// `METADATA FROM 'key'`
+    ///
+    /// A special type of column that gets its value from metadata
+    /// associated with the record.
+    ///
+    /// Example:
+    /// ```sql
+    /// CREATE TABLE logs (
+    ///   id TEXT,
+    ///   kafka_topic STRING METADATA FROM 'topic',
+    ///   log TEXT
+    /// )
+    /// ```
+    MetadataField(String, Span),
     /// `Generated`s are modifiers that follow a column definition in a `CREATE
     /// TABLE` statement.
     Generated {
@@ -2085,6 +2099,9 @@ impl fmt::Display for ColumnOption {
             Collation(n) => write!(f, "COLLATE {n}"),
             Comment(v) => write!(f, "COMMENT '{}'", escape_single_quote_string(v)),
             OnUpdate(expr) => write!(f, "ON UPDATE {expr}"),
+            MetadataField(key, _) => {
+                write!(f, "METADATA FROM '{}'", escape_single_quote_string(key))
+            }
             Generated {
                 generated_as,
                 sequence_options,
@@ -3060,6 +3077,10 @@ pub struct CreateTable {
     /// Redshift `BACKUP` option: `BACKUP { YES | NO }`
     /// <https://docs.aws.amazon.com/redshift/latest/dg/r_CREATE_TABLE_NEW.html>
     pub backup: Option<bool>,
+    /// Arroyo-specific: Iceberg partition transforms
+    /// Syntax: PARTITIONED BY (hour(ts), bucket(32, id), truncate(8, color))
+    /// <https://iceberg.apache.org/spec/#partitioning>
+    pub arroyo_partitions: Option<Vec<Expr>>,
 }
 
 impl fmt::Display for CreateTable {
@@ -3250,8 +3271,25 @@ impl fmt::Display for CreateTable {
         if let Some(cluster_by) = self.cluster_by.as_ref() {
             write!(f, " CLUSTER BY {cluster_by}")?;
         }
-        if let options @ CreateTableOptions::Options(_) = &self.table_options {
-            write!(f, " {options}")?;
+        // Connector partitions are parsed after table options. Keep `OPTIONS`
+        // before them when both are present so GenericDialect does not reparse
+        // the partition expressions as Hive partition columns.
+        if self.arroyo_partitions.is_some() {
+            if let options @ CreateTableOptions::Options(_) = &self.table_options {
+                write!(f, " {options}")?;
+            }
+        }
+        if let Some(partitions) = &self.arroyo_partitions {
+            write!(
+                f,
+                " PARTITIONED BY ({})",
+                display_comma_separated(partitions)
+            )?;
+        }
+        if self.arroyo_partitions.is_none() {
+            if let options @ CreateTableOptions::Options(_) = &self.table_options {
+                write!(f, " {options}")?;
+            }
         }
         if let Some(external_volume) = self.external_volume.as_ref() {
             write!(f, " EXTERNAL_VOLUME='{external_volume}'")?;
@@ -4698,7 +4736,7 @@ impl fmt::Display for AlterTable {
         if self.only {
             write!(f, "ONLY ")?;
         }
-        write!(f, "{} ", &self.name)?;
+        write!(f, "{} ", self.name)?;
         if let Some(cluster) = &self.on_cluster {
             write!(f, "ON CLUSTER {cluster} ")?;
         }

src/ast/helpers/stmt_create_table.rs

@@ -183,6 +183,8 @@ pub struct CreateTableBuilder {
     pub sortkey: Option<Vec<Expr>>,
     /// Redshift `BACKUP` option.
     pub backup: Option<bool>,
+    /// Arroyo connector partition expressions.
+    pub arroyo_partitions: Option<Vec<Expr>>,
 }
 
 impl CreateTableBuilder {
@@ -248,6 +250,7 @@ impl CreateTableBuilder {
             distkey: None,
             sortkey: None,
             backup: None,
+            arroyo_partitions: None,
         }
     }
     /// Set `OR REPLACE` for the CREATE TABLE statement.
@@ -556,6 +559,12 @@ impl CreateTableBuilder {
         self.backup = backup;
         self
     }
+    /// Set Arroyo connector partition expressions.
+    pub fn arroyo_partitions(mut self, partitions: Option<Vec<Expr>>) -> Self {
+        self.arroyo_partitions = partitions;
+        self
+    }
+
     /// Consume the builder and produce a `CreateTable`.
     pub fn build(self) -> CreateTable {
         CreateTable {
@@ -618,6 +627,7 @@ impl CreateTableBuilder {
             distkey: self.distkey,
             sortkey: self.sortkey,
             backup: self.backup,
+            arroyo_partitions: self.arroyo_partitions,
         }
     }
 }
@@ -699,6 +709,7 @@ impl From<CreateTable> for CreateTableBuilder {
             distkey: table.distkey,
             sortkey: table.sortkey,
             backup: table.backup,
+            arroyo_partitions: table.arroyo_partitions,
         }
     }
 }

src/ast/mod.rs

@@ -11748,7 +11748,7 @@ impl fmt::Display for AlterUser {
         let has_props = !self.set_props.options.is_empty();
         if has_props {
             write!(f, " SET")?;
-            write!(f, " {}", &self.set_props)?;
+            write!(f, " {}", self.set_props)?;
         }
         if !self.unset_props.is_empty() {
             write!(f, " UNSET {}", display_comma_separated(&self.unset_props))?;

src/ast/query.rs

@@ -3520,7 +3520,7 @@ pub struct LockClause {
 
 impl fmt::Display for LockClause {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "FOR {}", &self.lock_type)?;
+        write!(f, "FOR {}", self.lock_type)?;
         if let Some(ref of) = self.of {
             write!(f, " OF {of}")?;
         }

src/ast/spans.rs

@@ -604,6 +604,7 @@ impl Spanned for CreateTable {
             distkey: _,
             sortkey: _,
             backup: _,
+            arroyo_partitions,
         } = self;
 
         union_spans(
@@ -614,7 +615,8 @@ impl Spanned for CreateTable {
                 .chain(query.iter().map(|i| i.span()))
                 .chain(clone.iter().map(|i| i.span()))
                 .chain(partition_of.iter().map(|i| i.span()))
-                .chain(for_values.iter().map(|i| i.span())),
+                .chain(for_values.iter().map(|i| i.span()))
+                .chain(arroyo_partitions.iter().flatten().map(Spanned::span)),
         )
     }
 }
@@ -650,6 +652,12 @@ impl Spanned for TableConstraint {
             TableConstraint::FulltextOrSpatial(constraint) => constraint.span(),
             TableConstraint::PrimaryKeyUsingIndex(constraint)
             | TableConstraint::UniqueUsingIndex(constraint) => constraint.span(),
+            TableConstraint::Watermark {
+                column_name,
+                watermark_expr,
+            } => column_name
+                .span
+                .union_opt(&watermark_expr.as_ref().map(Spanned::span)),
         }
     }
 }
@@ -832,6 +840,7 @@ impl Spanned for ColumnOption {
             ColumnOption::Collation(object_name) => object_name.span(),
             ColumnOption::Comment(_) => Span::empty(),
             ColumnOption::OnUpdate(expr) => expr.span(),
+            ColumnOption::MetadataField(_, span) => *span,
             ColumnOption::Generated { .. } => Span::empty(),
             ColumnOption::Options(vec) => union_spans(vec.iter().map(|i| i.span())),
             ColumnOption::Identity(..) => Span::empty(),

src/ast/table_constraints.rs

@@ -117,6 +117,21 @@ pub enum TableConstraint {
     ///
     /// [1]: https://www.postgresql.org/docs/current/sql-altertable.html
     UniqueUsingIndex(ConstraintUsingIndex),
+    /// Arroyo specific: Watermark definition for streaming tables
+    /// Syntax:
+    /// ```sql
+    /// WATERMARK FOR timestamp AS timestamp - INTERVAL '5 seconds'
+    /// ```
+    /// or without an expression
+    /// ```sql
+    /// WATERMARK FOR timestamp
+    /// ```
+    Watermark {
+        /// Column name to be used for the watermark
+        column_name: Ident,
+        /// Optional watermark expression
+        watermark_expr: Option<Expr>,
+    },
 }
 
 impl From<UniqueConstraint> for TableConstraint {
@@ -166,6 +181,16 @@ impl fmt::Display for TableConstraint {
             TableConstraint::FulltextOrSpatial(constraint) => constraint.fmt(f),
             TableConstraint::PrimaryKeyUsingIndex(c) => c.fmt_with_keyword(f, "PRIMARY KEY"),
             TableConstraint::UniqueUsingIndex(c) => c.fmt_with_keyword(f, "UNIQUE"),
+            TableConstraint::Watermark {
+                column_name,
+                watermark_expr,
+            } => {
+                write!(f, "WATERMARK FOR {column_name}")?;
+                if let Some(expr) = watermark_expr {
+                    write!(f, " AS {expr}")?;
+                }
+                Ok(())
+            }
         }
     }
 }