feat(datafusion): support table and column comment DDL

crates/integrations/datafusion/src/sql_context.rs

@@ -66,10 +66,11 @@ use datafusion::prelude::{DataFrame, SessionContext};
 use datafusion::sql::planner::IdentNormalizer;
 use datafusion::sql::sqlparser::ast::{
     AlterColumnOperation, AlterTableOperation, BinaryLength, CharacterLength, ColumnDef,
-    ColumnOption, CreateFunction, CreateFunctionBody, CreateTable, CreateTableOptions, CreateView,
-    Delete, Expr as SqlExpr, FromTable, FunctionBehavior, FunctionReturnType, Ident, Insert, Merge,
-    ObjectName, ObjectType, RenameTableNameKind, Reset, ResetStatement, Set, ShowCreateObject,
-    SqlOption, Statement, TableFactor, TableObject, Truncate, Update, Use, Value as SqlValue,
+    ColumnOption, CommentDef, CommentObject, CreateFunction, CreateFunctionBody, CreateTable,
+    CreateTableOptions, CreateView, Delete, Expr as SqlExpr, FromTable, FunctionBehavior,
+    FunctionReturnType, Ident, Insert, Merge, ObjectName, ObjectType, RenameTableNameKind, Reset,
+    ResetStatement, Set, ShowCreateObject, SqlOption, Statement, TableFactor, TableObject,
+    Truncate, Update, Use, Value as SqlValue,
 };
 use datafusion::sql::sqlparser::dialect::GenericDialect;
 use datafusion::sql::sqlparser::keywords::Keyword;
@@ -477,7 +478,13 @@ impl SQLContext {
             .await;
         }
 
-        let statements = parse_sql_statements(&rewritten_sql)?;
+        let alter_comment = parse_alter_column_comment(&rewritten_sql)?;
+        let from_alter_comment = alter_comment.is_some();
+        let statements = if let Some(statement) = alter_comment {
+            vec![statement]
+        } else {
+            parse_sql_statements(&rewritten_sql)?
+        };
 
         if statements.len() != 1 {
             return Err(DataFusionError::Plan(
@@ -486,6 +493,65 @@ impl SQLContext {
         }
 
         match &statements[0] {
+            Statement::Comment {
+                object_type,
+                object_name,
+                comment,
+                if_exists,
+            } if matches!(object_type, CommentObject::Table | CommentObject::Column) => {
+                if *if_exists && !from_alter_comment {
+                    return Err(DataFusionError::NotImplemented(
+                        "COMMENT IF EXISTS is not supported; ALTER TABLE IF EXISTS is supported"
+                            .into(),
+                    ));
+                }
+                let mut table_name = object_name.clone();
+                let column = if *object_type == CommentObject::Column {
+                    if table_name.0.len() < 2 {
+                        return Err(DataFusionError::Plan(
+                            "COMMENT ON COLUMN requires table.column".into(),
+                        ));
+                    }
+                    let part = table_name.0.pop().unwrap();
+                    let ident = part
+                        .as_ident()
+                        .ok_or_else(|| DataFusionError::Plan("Invalid column name".into()))?;
+                    Some(normalize_schema_identifier(
+                        ident,
+                        enable_ident_normalization,
+                    ))
+                } else {
+                    None
+                };
+                Self::ensure_main_branch_write_target(&table_name, "COMMENT ON")?;
+                for part in &mut table_name.0 {
+                    if let datafusion::sql::sqlparser::ast::ObjectNamePart::Identifier(ident) = part
+                    {
+                        ident.value =
+                            normalize_schema_identifier(ident, enable_ident_normalization);
+                    }
+                }
+                if self.temp_table_for(&table_name).await?.is_some() {
+                    return Err(DataFusionError::NotImplemented(
+                        "Comments on temporary tables or views are not supported".into(),
+                    ));
+                }
+                let (catalog, _, identifier) = self.resolve_catalog_and_table(&table_name)?;
+                let change = match column {
+                    Some(column) => SchemaChange::update_column_comment(
+                        column,
+                        comment.clone().unwrap_or_default(),
+                    ),
+                    None => SchemaChange::UpdateComment {
+                        comment: comment.clone(),
+                    },
+                };
+                catalog
+                    .alter_table(&identifier, vec![change], *if_exists)
+                    .await
+                    .map_err(to_datafusion_error)?;
+                ok_result(&self.ctx)
+            }
             statement @ (Statement::ShowDatabases { .. }
             | Statement::ShowSchemas { .. }
             | Statement::ShowCatalogs { .. }
@@ -1033,6 +1099,30 @@ impl SQLContext {
         let identifier = self.resolve_table_name(&ct.name)?;
 
         let mut builder = paimon::spec::Schema::builder();
+        let mut comments = Vec::new();
+        if let Some(comment) = &ct.comment {
+            comments.push(comment);
+        }
+        let options = match &ct.table_options {
+            CreateTableOptions::With(options)
+            | CreateTableOptions::Options(options)
+            | CreateTableOptions::TableProperties(options)
+            | CreateTableOptions::Plain(options) => options.as_slice(),
+            CreateTableOptions::None => &[],
+        };
+        for option in options {
+            if let SqlOption::Comment(comment) = option {
+                comments.push(comment);
+            }
+        }
+        if comments.len() > 1 {
+            return Err(DataFusionError::Plan("Duplicate table COMMENT".into()));
+        }
+        if let Some(CommentDef::WithEq(comment) | CommentDef::WithoutEq(comment)) =
+            comments.first().copied()
+        {
+            builder = builder.comment(Some(comment.clone()));
+        }
         let table_options = extract_options(&ct.table_options)?;
         let column_names: Vec<String> = ct
             .columns
@@ -2654,6 +2744,42 @@ fn parse_sql_statements(sql: &str) -> DFResult<Vec<Statement>> {
     Ok(statements)
 }
 
+/// Parse the Paimon/Flink column-comment spelling into the standard COMMENT AST.
+/// Token parsing preserves quoted names and literals and rejects trailing statements.
+fn parse_alter_column_comment(sql: &str) -> DFResult<Option<Statement>> {
+    let parse = || -> Result<Option<Statement>, datafusion::sql::sqlparser::parser::ParserError> {
+        let mut parser = Parser::new(&GenericDialect {}).try_with_sql(sql)?;
+        if !parser.parse_keywords(&[Keyword::ALTER, Keyword::TABLE]) {
+            return Ok(None);
+        }
+        let if_exists = parser.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
+        let mut object_name = parser.parse_object_name(true)?;
+        if !parser.parse_keyword(Keyword::ALTER) {
+            return Ok(None);
+        }
+        let _ = parser.parse_keyword(Keyword::COLUMN);
+        let column = parser.parse_identifier()?;
+        if !parser.parse_keyword(Keyword::COMMENT) {
+            return Ok(None);
+        }
+        let comment = parser.parse_literal_string()?;
+        object_name
+            .0
+            .push(datafusion::sql::sqlparser::ast::ObjectNamePart::Identifier(
+                column,
+            ));
+        let _ = parser.consume_token(&Token::SemiColon);
+        parser.expect_token(&Token::EOF)?;
+        Ok(Some(Statement::Comment {
+            object_type: CommentObject::Column,
+            object_name,
+            comment: Some(comment),
+            if_exists,
+        }))
+    };
+    parse().map_err(|e| DataFusionError::Plan(format!("SQL parse error: {e}")))
+}
+
 fn validate_persistent_create_view(create_view: &CreateView) -> DFResult<()> {
     let unsupported = if create_view.or_alter {
         Some("CREATE OR ALTER VIEW is not supported")
@@ -3318,6 +3444,7 @@ fn extract_options(opts: &CreateTableOptions) -> DFResult<Vec<(String, String)>>
     };
     sql_options
         .iter()
+        .filter(|option| !matches!(option, SqlOption::Comment(_)))
         .map(|opt| match opt {
             SqlOption::KeyValue { key, value } => {
                 let v = value.to_string();

crates/integrations/datafusion/tests/comment_ddl.rs

@@ -0,0 +1,155 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+use paimon::{catalog::Identifier, Catalog};
+
+#[tokio::test]
+async fn comments_round_trip_without_changing_fields_or_data() {
+    let (_tmp, catalog) = common::create_test_env();
+    let ctx = common::create_sql_context(catalog.clone()).await;
+    ctx.sql("CREATE TABLE default.c (id INT COMMENT 'old') COMMENT 'table ''quote'''")
+        .await
+        .unwrap();
+    ctx.sql("INSERT INTO default.c VALUES (7)")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    let id = Identifier::new("default", "c");
+    let before = catalog.get_table(&id).await.unwrap().schema().clone();
+    assert_eq!(before.comment(), Some("table 'quote'"));
+    ctx.sql("ALTER TABLE default.c ALTER COLUMN id COMMENT '列 ''quote'''")
+        .await
+        .unwrap();
+    ctx.sql("COMMENT ON TABLE default.c IS 'new'")
+        .await
+        .unwrap();
+    let after = catalog.get_table(&id).await.unwrap().schema().clone();
+    assert_eq!(after.comment(), Some("new"));
+    assert_eq!(after.fields()[0].description(), Some("列 'quote'"));
+    assert_eq!(after.fields()[0].id(), before.fields()[0].id());
+    assert_eq!(
+        after.fields()[0].data_type(),
+        before.fields()[0].data_type()
+    );
+    ctx.sql("COMMENT ON COLUMN default.c.id IS NULL")
+        .await
+        .unwrap();
+    ctx.sql("COMMENT ON TABLE default.c IS NULL").await.unwrap();
+    let cleared = catalog.get_table(&id).await.unwrap().schema().clone();
+    assert_eq!(cleared.comment(), None);
+    assert_eq!(cleared.fields()[0].description(), Some(""));
+    assert_eq!(
+        common::collect_id_value(&ctx, "SELECT id, id AS value FROM default.c").await,
+        vec![(7, 7)]
+    );
+}
+
+#[tokio::test]
+async fn comments_resolve_names_and_reject_invalid_targets_without_mutation() {
+    let (_tmp, catalog) = common::create_test_env();
+    let ctx = common::create_sql_context(catalog.clone()).await;
+    ctx.sql("CREATE TABLE default.c (id INT) COMMENT 'original'")
+        .await
+        .unwrap();
+    ctx.sql("CREATE TABLE default.\"MixedCase\" (id INT) COMMENT 'upper'")
+        .await
+        .unwrap();
+    ctx.sql("COMMENT ON COLUMN DEFAULT.C.ID IS 'lower column'")
+        .await
+        .unwrap();
+    ctx.sql("COMMENT ON TABLE DEFAULT.C IS 'lower table'")
+        .await
+        .unwrap();
+    assert_eq!(
+        catalog
+            .get_table(&Identifier::new("default", "c"))
+            .await
+            .unwrap()
+            .schema()
+            .comment(),
+        Some("lower table")
+    );
+    assert_eq!(
+        catalog
+            .get_table(&Identifier::new("default", "MixedCase"))
+            .await
+            .unwrap()
+            .schema()
+            .comment(),
+        Some("upper")
+    );
+    ctx.sql("COMMENT ON TABLE default.\"MixedCase\" IS 'quoted'")
+        .await
+        .unwrap();
+    assert_eq!(
+        catalog
+            .get_table(&Identifier::new("default", "MixedCase"))
+            .await
+            .unwrap()
+            .schema()
+            .comment(),
+        Some("quoted")
+    );
+    let before = catalog
+        .get_table(&Identifier::new("default", "c"))
+        .await
+        .unwrap()
+        .schema()
+        .clone();
+    for sql in [
+        "COMMENT ON COLUMN default.c.missing IS 'bad'",
+        "COMMENT IF EXISTS ON COLUMN default.c.missing IS 'bad'",
+        "ALTER TABLE IF EXISTS default.c ALTER COLUMN missing COMMENT 'bad'",
+        "ALTER TABLE default.c ALTER COLUMN id COMMENT 'bad'; DROP TABLE default.c",
+        "CREATE TABLE default.duplicate(id INT) COMMENT 'a' COMMENT 'b'",
+        "COMMENT ON TABLE default.\"c$branch_test\" IS 'bad'",
+    ] {
+        assert!(ctx.sql(sql).await.is_err(), "{sql}");
+        assert_eq!(
+            catalog
+                .get_table(&Identifier::new("default", "c"))
+                .await
+                .unwrap()
+                .schema()
+                .id(),
+            before.id(),
+            "{sql}"
+        );
+    }
+    ctx.sql("ALTER TABLE IF EXISTS default.missing ALTER COLUMN id COMMENT 'ok'")
+        .await
+        .unwrap();
+    ctx.sql("CREATE TEMPORARY VIEW default.c AS SELECT 1 AS id")
+        .await
+        .unwrap();
+    assert!(ctx
+        .sql("COMMENT ON TABLE default.c IS 'hidden mutation'")
+        .await
+        .is_err());
+    assert_eq!(
+        catalog
+            .get_table(&Identifier::new("default", "c"))
+            .await
+            .unwrap()
+            .schema()
+            .comment(),
+        Some("lower table")
+    );
+}

docs/src/sql.md

@@ -794,6 +794,27 @@ ALTER TABLE paimon.my_db.users SET TBLPROPERTIES('data-evolution.enabled' = 'tru
 ALTER TABLE IF EXISTS paimon.my_db.users ADD COLUMN age INT;
 ```
 
+### Table and column comments
+
+```sql
+CREATE TABLE db.t (id INT COMMENT 'identifier') COMMENT 'description';
+ALTER TABLE db.t ALTER COLUMN id COMMENT 'new description';
+COMMENT ON TABLE db.t IS 'new table description';
+COMMENT ON COLUMN db.t.id IS 'new column description';
+COMMENT ON TABLE db.t IS NULL;
+COMMENT ON COLUMN db.t.id IS NULL;
+```
+
+These statements update catalog metadata without changing field IDs, types or
+stored rows. Quoted identifiers retain their case; unquoted names follow the
+session's identifier-normalization setting. Column comments address top-level
+columns. Clearing a column comment uses Paimon's empty-description encoding.
+
+`ALTER TABLE IF EXISTS ... ALTER COLUMN ... COMMENT` ignores a missing table,
+but still reports a missing column. The nonstandard `COMMENT IF EXISTS` form
+is rejected. Comments on temporary objects, including temporary objects that
+shadow persistent tables, and writes to branch references are rejected.
+
 ## DML
 
 The table type determines which row-level DML operations are supported: