feat: add comment on clause support

e2e_test/ddl/show.slt

@@ -7,15 +7,21 @@ create materialized view mv3 as select sum(v1) as sum_v1 from t3;
 statement ok
 create view v3 as select sum(v2) as sum_v2 from t3;
 
-query TTT
+statement ok
+comment on column t3.v2 is 'Lorem ipsum dolor sit amet';
+
+statement ok
+comment on column t3._row_id is 'consectetur adipiscing elit';
+
+query TTTT
 describe t3;
 ----
-v1 integer false
-v2 integer false
-v3 integer false
-_row_id serial true
-primary key _row_id NULL
-distribution key _row_id NULL
+v1               integer false  NULL
+v2               integer false  Lorem ipsum dolor sit amet
+v3               integer false  NULL
+_row_id           serial  true  consectetur adipiscing elit
+primary key      _row_id  NULL  NULL
+distribution key _row_id  NULL  NULL
 
 query TTT
 show columns from t3;
@@ -33,16 +39,19 @@ show indexes from t3;
 ----
 idx1 t3 v1 ASC, v2 ASC v3 v1
 
-query TTT
+statement ok
+comment on column t3.v2 is 'Nemo enim ipsam';
+
+query TTTT
 describe t3;
 ----
-v1 integer false
-v2 integer false
-v3 integer false
-_row_id serial true
-primary key _row_id NULL
-distribution key _row_id NULL
-idx1 index(v1 ASC, v2 ASC) include(v3) distributed by(v1) NULL
+v1               integer false  NULL
+v2               integer false  Nemo enim ipsam
+v3               integer false  NULL
+_row_id           serial  true  consectetur adipiscing elit
+primary key      _row_id  NULL  NULL
+distribution key _row_id  NULL  NULL
+idx1             index(v1 ASC, v2 ASC) include(v3) distributed by(v1) NULL NULL
 
 query TT
 show create index idx1;

e2e_test/extended_mode/basic.slt

@@ -39,15 +39,15 @@ values(round(42.4382));
 statement ok
 create table t3 (v1 int, v2 int, v3 int);
 
-query TTT
+query TTTT
 describe t3;
 ----
-v1 integer false
-v2 integer false
-v3 integer false
-_row_id serial true
-primary key _row_id NULL
-distribution key _row_id NULL
+v1                integer  false  NULL
+v2                integer  false  NULL
+v3                integer  false  NULL
+_row_id           serial    true  NULL
+primary key       _row_id   NULL  NULL
+distribution key  _row_id   NULL  NULL
 
 query TTT
 show columns from t3;

proto/catalog.proto

@@ -280,6 +280,8 @@ message Table {
 
   CreateType create_type = 32;
 
+  optional string description = 33;
+
   // Per-table catalog version, used by schema change. `None` for internal tables and tests.
   // Not to be confused with the global catalog version for notification service.
   TableVersion version = 100;
@@ -317,3 +319,9 @@ message Database {
   string name = 2;
   uint32 owner = 3;
 }
+
+message Comment {
+  uint32 table_id = 1;
+  optional uint32 column_index = 2;
+  optional string description = 3;
+}

proto/ddl_service.proto

@@ -314,6 +314,17 @@ message GetTablesResponse {
   map<uint32, catalog.Table> tables = 1;
 }
 
+message CommentOnRequest {
+  uint32 table_id = 1;
+  optional uint32 column_index = 2;
+  optional string comment = 3;
+}
+
+message CommentOnResponse {
+  common.Status status = 1;
+  uint64 version = 2;
+}
+
 service DdlService {
   rpc CreateDatabase(CreateDatabaseRequest) returns (CreateDatabaseResponse);
   rpc DropDatabase(DropDatabaseRequest) returns (DropDatabaseResponse);
@@ -342,5 +353,6 @@ service DdlService {
   rpc CreateConnection(CreateConnectionRequest) returns (CreateConnectionResponse);
   rpc ListConnections(ListConnectionsRequest) returns (ListConnectionsResponse);
   rpc DropConnection(DropConnectionRequest) returns (DropConnectionResponse);
+  rpc CommentOn(CommentOnRequest) returns (CommentOnResponse);
   rpc GetTables(GetTablesRequest) returns (GetTablesResponse);
 }

proto/plan_common.proto

@@ -37,6 +37,8 @@ message ColumnDesc {
     GeneratedColumnDesc generated_column = 6;
     DefaultColumnDesc default_column = 7;
   }
+
+  optional string description = 8;
 }
 
 message ColumnCatalog {

src/common/src/catalog/column.rs

@@ -101,6 +101,7 @@ pub struct ColumnDesc {
     pub field_descs: Vec<ColumnDesc>,
     pub type_name: String,
     pub generated_or_default_column: Option<GeneratedOrDefaultColumn>,
+    pub description: Option<String>,
 }
 
 impl ColumnDesc {
@@ -112,6 +113,7 @@ impl ColumnDesc {
             field_descs: vec![],
             type_name: String::new(),
             generated_or_default_column: None,
+            description: None,
         }
     }
 
@@ -128,6 +130,7 @@ impl ColumnDesc {
                 .map(|f| f.to_protobuf())
                 .collect_vec(),
             type_name: self.type_name.clone(),
+            description: self.description.clone(),
             generated_or_default_column: self.generated_or_default_column.clone(),
         }
     }
@@ -171,6 +174,7 @@ impl ColumnDesc {
             name: name.to_string(),
             field_descs: vec![],
             type_name: "".to_string(),
+            description: None,
             generated_or_default_column: None,
         }
     }
@@ -191,6 +195,7 @@ impl ColumnDesc {
             name: name.to_string(),
             field_descs: fields,
             type_name: type_name.to_string(),
+            description: None,
             generated_or_default_column: None,
         }
     }
@@ -206,6 +211,7 @@ impl ColumnDesc {
                 .map(Self::from_field_without_column_id)
                 .collect_vec(),
             type_name: field.type_name.clone(),
+            description: None,
             generated_or_default_column: None,
         }
     }
@@ -242,6 +248,7 @@ impl From<PbColumnDesc> for ColumnDesc {
             name: prost.name,
             type_name: prost.type_name,
             field_descs,
+            description: prost.description.clone(),
             generated_or_default_column: prost.generated_or_default_column,
         }
     }
@@ -261,6 +268,7 @@ impl From<&ColumnDesc> for PbColumnDesc {
             name: c.name.clone(),
             field_descs: c.field_descs.iter().map(ColumnDesc::to_protobuf).collect(),
             type_name: c.type_name.clone(),
+            description: c.description.clone(),
             generated_or_default_column: c.generated_or_default_column.clone(),
         }
     }

src/common/src/catalog/mod.rs

@@ -109,6 +109,7 @@ pub fn row_id_column_desc() -> ColumnDesc {
         name: row_id_column_name(),
         field_descs: vec![],
         type_name: "".to_string(),
+        description: None,
         generated_or_default_column: None,
     }
 }
@@ -130,6 +131,7 @@ pub fn offset_column_desc() -> ColumnDesc {
         name: offset_column_name(),
         field_descs: vec![],
         type_name: "".to_string(),
+        description: None,
         generated_or_default_column: None,
     }
 }

src/common/src/catalog/test_utils.rs

@@ -56,6 +56,7 @@ impl ColumnDescTestExt for ColumnDesc {
             name: name.to_string(),
             type_name: type_name.to_string(),
             field_descs: fields,
+            description: None,
             generated_or_default_column: None,
         }
     }

src/compute/tests/integration_tests.rs

@@ -158,6 +158,7 @@ async fn test_table_materialize() -> StreamResult<()> {
             name: field.name,
             field_descs: vec![],
             type_name: "".to_string(),
+            description: None,
             generated_or_default_column: None,
         })
         .collect_vec();

src/connector/src/parser/avro/util.rs

@@ -56,6 +56,7 @@ fn avro_field_to_column_desc(
                 name: name.to_owned(),
                 field_descs: vec_column,
                 type_name: schema_name.to_string(),
+                description: None,
                 generated_or_default_column: None,
             })
         }

src/connector/src/parser/protobuf/parser.rs

@@ -196,6 +196,7 @@ impl ProtobufParserConfig {
                 column_type: Some(field_type.to_protobuf()),
                 field_descs,
                 type_name: m.full_name().to_string(),
+                description: None,
                 generated_or_default_column: None,
             })
         } else {

src/connector/src/source/manager.rs

@@ -122,6 +122,7 @@ impl From<&SourceColumnDesc> for ColumnDesc {
             name: s.name.clone(),
             field_descs: s.fields.clone(),
             type_name: "".to_string(),
+            description: None,
             generated_or_default_column: None,
         }
     }

src/frontend/src/binder/expr/mod.rs

@@ -575,6 +575,7 @@ pub fn bind_struct_field(column_def: &StructField) -> Result<ColumnDesc> {
                     name: f.name.real_value(),
                     field_descs: vec![],
                     type_name: "".to_string(),
+                    description: None,
                     generated_or_default_column: None,
                 })
             })
@@ -588,6 +589,7 @@ pub fn bind_struct_field(column_def: &StructField) -> Result<ColumnDesc> {
         name: column_def.name.real_value(),
         field_descs,
         type_name: "".to_string(),
+        description: None,
         generated_or_default_column: None,
     })
 }

src/frontend/src/catalog/catalog_service.rs

@@ -111,6 +111,13 @@ pub trait CatalogWriter: Send + Sync {
         connection: create_connection_request::Payload,
     ) -> Result<()>;
 
+    async fn comment_on(
+        &self,
+        table_id: TableId,
+        column_index: Option<u32>,
+        comment: Option<String>,
+    ) -> Result<()>;
+
     async fn drop_table(
         &self,
         source_id: Option<u32>,
@@ -282,6 +289,19 @@ impl CatalogWriter for CatalogWriterImpl {
         self.wait_version(version).await
     }
 
+    async fn comment_on(
+        &self,
+        TableId { table_id }: TableId,
+        column_index: Option<u32>,
+        comment: Option<String>,
+    ) -> Result<()> {
+        let version = self
+            .meta_client
+            .comment_on(table_id, column_index, comment)
+            .await?;
+        self.wait_version(version).await
+    }
+
     async fn drop_table(
         &self,
         source_id: Option<u32>,

src/frontend/src/catalog/root_catalog.rs

@@ -370,6 +370,14 @@ impl Catalog {
         Ok(self.get_database_by_name(db_name)?.iter_schemas())
     }
 
+    pub fn iter_schemas_except_rw_catalog(
+        &self,
+        db_name: &str,
+    ) -> CatalogResult<impl Iterator<Item = &SchemaCatalog>> {
+        self.iter_schemas(db_name)
+            .map(|scs| scs.filter(|sc| !sc.is_rw_catalog()))
+    }
+
     pub fn get_all_database_names(&self) -> Vec<String> {
         self.database_by_name.keys().cloned().collect_vec()
     }

src/frontend/src/catalog/schema_catalog.rs

@@ -432,6 +432,10 @@ impl SchemaCatalog {
         self.system_table_by_name.values()
     }
 
+    pub fn is_rw_catalog(&self) -> bool {
+        !self.system_table_by_name.is_empty()
+    }
+
     pub fn get_table_by_name(&self, table_name: &str) -> Option<&Arc<TableCatalog>> {
         self.table_by_name.get(table_name)
     }

src/frontend/src/catalog/system_catalog/mod.rs

@@ -57,6 +57,8 @@ pub struct SystemTableCatalog {
 
     // owner of table, should always be default super user, keep it for compatibility.
     pub owner: u32,
+
+    pub description: Option<String>,
 }
 
 impl SystemTableCatalog {
@@ -165,6 +167,7 @@ impl From<&BuiltinTable> for SystemTableCatalog {
                 .collect(),
             pk: val.pk.to_vec(),
             owner: DEFAULT_SUPER_USER_ID,
+            description: None,
         }
     }
 }
@@ -412,6 +415,7 @@ prepare_sys_catalog! {
     { BuiltinCatalog::Table(&RW_HUMMOCK_BRANCHED_OBJECTS), read_hummock_branched_objects await },
     { BuiltinCatalog::Table(&RW_HUMMOCK_COMPACTION_GROUP_CONFIGS), read_hummock_compaction_group_configs await },
     { BuiltinCatalog::Table(&RW_HUMMOCK_META_CONFIGS), read_hummock_meta_configs await},
+    { BuiltinCatalog::Table(&RW_DESCRIPTION), read_rw_description },
 }
 
 #[cfg(test)]

src/frontend/src/catalog/system_catalog/rw_catalog/mod.rs

@@ -17,6 +17,7 @@ mod rw_columns;
 mod rw_connections;
 mod rw_databases;
 mod rw_ddl_progress;
+mod rw_description;
 mod rw_fragments;
 mod rw_functions;
 mod rw_hummock_branched_objects;
@@ -50,6 +51,7 @@ pub use rw_columns::*;
 pub use rw_connections::*;
 pub use rw_databases::*;
 pub use rw_ddl_progress::*;
+pub use rw_description::*;
 pub use rw_fragments::*;
 pub use rw_functions::*;
 pub use rw_hummock_branched_objects::*;