Implement METADATA FROM syntax

Author: mwylde

src/ast/ddl.rs

@@ -1655,6 +1655,20 @@ pub enum ColumnOption {
         /// false if 'GENERATED ALWAYS' is skipped (option starts with AS)
         generated_keyword: bool,
     },
+    /// `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),
     /// BigQuery specific: Explicit column options in a view [1] or table [2]
     /// Syntax
     /// ```sql
@@ -1745,6 +1759,7 @@ 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,

src/parser/mod.rs

@@ -7084,6 +7084,17 @@ impl<'a> Parser<'a> {
             Ok(Some(ColumnOption::Null))
         } else if self.parse_keyword(Keyword::DEFAULT) {
             Ok(Some(ColumnOption::Default(self.parse_expr()?)))
+        } else if self.parse_keywords(&[Keyword::METADATA, Keyword::FROM])
+            && dialect_of!(self is ArroyoDialect | GenericDialect)
+        {
+            // Parse metadata field syntax: METADATA FROM 'key'
+            let next_token = self.next_token();
+            match next_token.token {
+                Token::SingleQuotedString(value, ..) => {
+                    Ok(Some(ColumnOption::MetadataField(value)))
+                }
+                _ => self.expected("string literal for metadata key", next_token),
+            }
         } else if dialect_of!(self is ClickHouseDialect| GenericDialect)
             && self.parse_keyword(Keyword::MATERIALIZED)
         {

tests/sqlparser_arroyo.rs

@@ -12,7 +12,7 @@
 
 #![warn(clippy::all)]
 
-use sqlparser::ast::{BinaryOperator, Expr, Ident, Statement, TableConstraint};
+use sqlparser::ast::{BinaryOperator, ColumnOption, Expr, Ident, Statement, TableConstraint};
 use sqlparser::dialect::ArroyoDialect;
 use sqlparser::parser::Parser;
 use sqlparser::test_utils;
@@ -81,3 +81,37 @@ fn test_watermark_without_expr() {
         }]
     );
 }
+
+#[test]
+fn test_metadata_field() {
+    let sql = "CREATE TABLE logs (
+        id TEXT,
+        kafka_topic STRING METADATA FROM 'topic',
+        log TEXT
+    )";
+
+    let parse = Parser::parse_sql(&ArroyoDialect {}, sql).unwrap();
+    let Statement::CreateTable(ct) = parse.get(0).unwrap() else {
+        panic!("not create table")
+    };
+
+    assert_eq!(ct.columns.len(), 3);
+
+    // Check the middle column with METADATA FROM
+    let column = &ct.columns[1];
+    assert_eq!(column.name, Ident::new("kafka_topic"));
+
+    // Check for the METADATA FROM option
+    let mut found_metadata = false;
+    for option_def in &column.options {
+        if let ColumnOption::MetadataField(key) = &option_def.option {
+            found_metadata = true;
+            assert_eq!(key, "topic");
+        }
+    }
+
+    assert!(
+        found_metadata,
+        "Expected METADATA FROM option in column definition"
+    );
+}