feat: add comment on clause support

proto/catalog.proto

@@ -94,6 +94,7 @@ message Source {
 
   optional uint64 initialized_at_epoch = 15;
   optional uint64 created_at_epoch = 16;
+  optional string description = 17;
 
   // Per-source catalog version, used by schema change.
   uint64 version = 100;
@@ -140,6 +141,7 @@ message Sink {
   string sink_from_name = 18;
   StreamJobStatus stream_job_status = 19;
   SinkFormatDesc format_desc = 20;
+  optional string description = 21;
 }
 
 message Connection {
@@ -164,6 +166,7 @@ message Connection {
     PrivateLinkService private_link_service = 5;
   }
   uint32 owner = 6;
+  optional string description = 7;
 }
 
 message Index {
@@ -182,6 +185,7 @@ message Index {
   optional uint64 initialized_at_epoch = 10;
   optional uint64 created_at_epoch = 11;
   StreamJobStatus stream_job_status = 12;
+  optional string description = 13;
 }
 
 message Function {
@@ -204,6 +208,8 @@ message Function {
   message ScalarFunction {}
   message TableFunction {}
   message AggregateFunction {}
+
+  optional string description = 14;
 }
 
 // See `TableCatalog` struct in frontend crate for more information.
@@ -280,6 +286,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;
@@ -303,17 +311,26 @@ message View {
   repeated uint32 dependent_relations = 8;
   // User-specified column names.
   repeated plan_common.Field columns = 9;
+  optional string description = 10;
 }
 
 message Schema {
   uint32 id = 1;
   uint32 database_id = 2;
   string name = 3;
   uint32 owner = 4;
+  optional string description = 5;
 }
 
 message Database {
   uint32 id = 1;
   string name = 2;
   uint32 owner = 3;
+  optional string description = 4;
+}
+
+message Comment {
+  uint32 table_id = 1;
+  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 CreateCommentRequest {
+  uint32 table_id = 1;
+  uint32 column_index = 2;
+  optional string comment = 3;
+}
+
+message CreateCommentResponse {
+  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 CreateComment(CreateCommentRequest) returns (CreateCommentResponse);
   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 {

proto/stream_plan.proto

@@ -181,6 +181,8 @@ message SinkDesc {
   // it is the name of the sink itself.
   string sink_from_name = 12;
   catalog.SinkFormatDesc format_desc = 13;
+
+  optional string description = 14;
 }
 
 enum SinkLogStoreType {

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/sink/catalog/desc.rs

@@ -64,6 +64,8 @@ pub struct SinkDesc {
     /// Name of the "table" field for Debezium. If the sink is from table or mv,
     /// it is the name of table/mv. Otherwise, it is the name of the sink.
     pub sink_from_name: String,
+
+    pub description: Option<String>,
 }
 
 impl SinkDesc {
@@ -95,6 +97,7 @@ impl SinkDesc {
             initialized_at_epoch: None,
             db_name: self.db_name,
             sink_from_name: self.sink_from_name,
+            description: None,
         }
     }
 
@@ -116,6 +119,7 @@ impl SinkDesc {
             format_desc: self.format_desc.as_ref().map(|f| f.to_proto()),
             db_name: self.db_name.clone(),
             sink_from_name: self.sink_from_name.clone(),
+            description: self.description.clone(),
         }
     }
 }

src/connector/src/sink/catalog/mod.rs

@@ -293,6 +293,8 @@ pub struct SinkCatalog {
 
     /// Name for the table info for Debezium sink
     pub sink_from_name: String,
+
+    pub description: Option<String>,
 }
 
 impl SinkCatalog {
@@ -330,6 +332,7 @@ impl SinkCatalog {
             db_name: self.db_name.clone(),
             sink_from_name: self.sink_from_name.clone(),
             stream_job_status: PbStreamJobStatus::Creating.into(),
+            description: self.description.clone(),
         }
     }
 
@@ -418,6 +421,7 @@ impl From<PbSink> for SinkCatalog {
             initialized_at_epoch: pb.initialized_at_epoch.map(Epoch::from),
             db_name: pb.db_name,
             sink_from_name: pb.sink_from_name,
+            description: pb.description.clone(),
         }
     }
 }

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

@@ -589,6 +589,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,
                 })
             })
@@ -602,6 +603,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 create_comment(
+        &self,
+        table_id: TableId,
+        column_index: u32,
+        comment: Option<String>,
+    ) -> Result<()>;
+
     async fn drop_table(
         &self,
         source_id: Option<u32>,
@@ -162,6 +169,7 @@ impl CatalogWriter for CatalogWriterImpl {
                 name: db_name.to_string(),
                 id: 0,
                 owner,
+                description: None,
             })
             .await?;
         self.wait_version(version).await
@@ -180,6 +188,7 @@ impl CatalogWriter for CatalogWriterImpl {
                 name: schema_name.to_string(),
                 database_id: db_id,
                 owner,
+                description: None,
             })
             .await?;
         self.wait_version(version).await
@@ -282,6 +291,19 @@ impl CatalogWriter for CatalogWriterImpl {
         self.wait_version(version).await
     }
 
+    async fn create_comment(
+        &self,
+        TableId { table_id }: TableId,
+        column_index: u32,
+        comment: Option<String>,
+    ) -> Result<()> {
+        let version = self
+            .meta_client
+            .create_comment(table_id, column_index, comment)
+            .await?;
+        self.wait_version(version).await
+    }
+
     async fn drop_table(
         &self,
         source_id: Option<u32>,

src/frontend/src/catalog/connection_catalog.rs

@@ -32,6 +32,7 @@ pub struct ConnectionCatalog {
     pub name: String,
     pub info: connection::Info,
     pub owner: UserId,
+    pub description: Option<String>,
 }
 
 impl ConnectionCatalog {
@@ -55,6 +56,7 @@ impl From<&PbConnection> for ConnectionCatalog {
             name: prost.name.clone(),
             info: prost.info.clone().unwrap(),
             owner: prost.owner,
+            description: prost.description.clone(),
         }
     }
 }