refactor: only support comment on table and col

reduce the changes impact radius.

proto/catalog.proto

@@ -94,7 +94,6 @@ 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;
@@ -141,7 +140,6 @@ message Sink {
   string sink_from_name = 18;
   StreamJobStatus stream_job_status = 19;
   SinkFormatDesc format_desc = 20;
-  optional string description = 21;
 }
 
 message Connection {
@@ -166,7 +164,6 @@ message Connection {
     PrivateLinkService private_link_service = 5;
   }
   uint32 owner = 6;
-  optional string description = 7;
 }
 
 message Index {
@@ -185,7 +182,6 @@ 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 {
@@ -208,8 +204,6 @@ message Function {
   message ScalarFunction {}
   message TableFunction {}
   message AggregateFunction {}
-
-  optional string description = 14;
 }
 
 // See `TableCatalog` struct in frontend crate for more information.
@@ -311,26 +305,23 @@ 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 uint32 column_index = 2;
   optional string description = 3;
 }

proto/ddl_service.proto

@@ -314,13 +314,13 @@ message GetTablesResponse {
   map<uint32, catalog.Table> tables = 1;
 }
 
-message CreateCommentRequest {
+message CommentOnRequest {
   uint32 table_id = 1;
-  uint32 column_index = 2;
+  optional uint32 column_index = 2;
   optional string comment = 3;
 }
 
-message CreateCommentResponse {
+message CommentOnResponse {
   common.Status status = 1;
   uint64 version = 2;
 }
@@ -353,6 +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 CommentOn(CommentOnRequest) returns (CommentOnResponse);
   rpc GetTables(GetTablesRequest) returns (GetTablesResponse);
 }

proto/stream_plan.proto

@@ -181,8 +181,6 @@ 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/connector/src/sink/catalog/desc.rs

@@ -64,8 +64,6 @@ 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 {
@@ -97,7 +95,6 @@ impl SinkDesc {
             initialized_at_epoch: None,
             db_name: self.db_name,
             sink_from_name: self.sink_from_name,
-            description: None,
         }
     }
 
@@ -119,7 +116,6 @@ 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,8 +293,6 @@ pub struct SinkCatalog {
 
     /// Name for the table info for Debezium sink
     pub sink_from_name: String,
-
-    pub description: Option<String>,
 }
 
 impl SinkCatalog {
@@ -332,7 +330,6 @@ 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(),
         }
     }
 
@@ -421,7 +418,6 @@ 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/frontend/src/catalog/catalog_service.rs

@@ -111,10 +111,10 @@ pub trait CatalogWriter: Send + Sync {
         connection: create_connection_request::Payload,
     ) -> Result<()>;
 
-    async fn create_comment(
+    async fn comment_on(
         &self,
         table_id: TableId,
-        column_index: u32,
+        column_index: Option<u32>,
         comment: Option<String>,
     ) -> Result<()>;
 
@@ -169,7 +169,6 @@ impl CatalogWriter for CatalogWriterImpl {
                 name: db_name.to_string(),
                 id: 0,
                 owner,
-                description: None,
             })
             .await?;
         self.wait_version(version).await
@@ -188,7 +187,6 @@ impl CatalogWriter for CatalogWriterImpl {
                 name: schema_name.to_string(),
                 database_id: db_id,
                 owner,
-                description: None,
             })
             .await?;
         self.wait_version(version).await
@@ -291,15 +289,15 @@ impl CatalogWriter for CatalogWriterImpl {
         self.wait_version(version).await
     }
 
-    async fn create_comment(
+    async fn comment_on(
         &self,
         TableId { table_id }: TableId,
-        column_index: u32,
+        column_index: Option<u32>,
         comment: Option<String>,
     ) -> Result<()> {
         let version = self
             .meta_client
-            .create_comment(table_id, column_index, comment)
+            .comment_on(table_id, column_index, comment)
             .await?;
         self.wait_version(version).await
     }

src/frontend/src/catalog/connection_catalog.rs

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

src/frontend/src/catalog/database_catalog.rs

@@ -28,7 +28,6 @@ pub struct DatabaseCatalog {
     schema_by_name: HashMap<String, SchemaCatalog>,
     schema_name_by_id: HashMap<SchemaId, String>,
     owner: u32,
-    description: Option<String>,
 }
 
 impl DatabaseCatalog {
@@ -66,7 +65,6 @@ impl DatabaseCatalog {
                 database_id: self.id,
                 name: schema.name(),
                 owner: schema.owner(),
-                description: None,
             })
             .collect_vec()
     }
@@ -110,10 +108,6 @@ impl DatabaseCatalog {
     pub fn owner(&self) -> u32 {
         self.owner
     }
-
-    pub fn description(&self) -> Option<&str> {
-        self.description.as_deref()
-    }
 }
 impl From<&PbDatabase> for DatabaseCatalog {
     fn from(db: &PbDatabase) -> Self {
@@ -123,7 +117,6 @@ impl From<&PbDatabase> for DatabaseCatalog {
             schema_by_name: HashMap::new(),
             schema_name_by_id: HashMap::new(),
             owner: db.owner,
-            description: db.description.clone(),
         }
     }
 }

src/frontend/src/catalog/function_catalog.rs

@@ -31,7 +31,6 @@ pub struct FunctionCatalog {
     pub language: String,
     pub identifier: String,
     pub link: String,
-    pub description: Option<String>,
 }
 
 #[derive(Clone, Display, PartialEq, Eq, Hash, Debug)]
@@ -65,7 +64,6 @@ impl From<&PbFunction> for FunctionCatalog {
             language: prost.language.clone(),
             identifier: prost.identifier.clone(),
             link: prost.link.clone(),
-            description: prost.description.clone(),
         }
     }
 }

src/frontend/src/catalog/index_catalog.rs

@@ -63,8 +63,6 @@ pub struct IndexCatalog {
     pub created_at_epoch: Option<Epoch>,
 
     pub initialized_at_epoch: Option<Epoch>,
-
-    pub description: Option<String>,
 }
 
 impl IndexCatalog {
@@ -126,7 +124,6 @@ impl IndexCatalog {
             original_columns,
             created_at_epoch: index_prost.created_at_epoch.map(Epoch::from),
             initialized_at_epoch: index_prost.initialized_at_epoch.map(Epoch::from),
-            description: index_prost.description.clone(),
         }
     }
 
@@ -188,7 +185,6 @@ impl IndexCatalog {
             initialized_at_epoch: self.initialized_at_epoch.map(|e| e.0),
             created_at_epoch: self.created_at_epoch.map(|e| e.0),
             stream_job_status: PbStreamJobStatus::Creating.into(),
-            description: self.description.clone(),
         }
     }
 

src/frontend/src/catalog/source_catalog.rs

@@ -43,7 +43,6 @@ pub struct SourceCatalog {
     pub connection_id: Option<ConnectionId>,
     pub created_at_epoch: Option<Epoch>,
     pub initialized_at_epoch: Option<Epoch>,
-    pub description: Option<String>,
     pub version: SourceVersionId,
 }
 
@@ -73,7 +72,6 @@ impl SourceCatalog {
             optional_associated_table_id: self
                 .associated_table_id
                 .map(|id| OptionalAssociatedTableId::AssociatedTableId(id.table_id)),
-            description: self.description.clone(),
             version: self.version,
         }
     }
@@ -129,7 +127,6 @@ impl From<&PbSource> for SourceCatalog {
             connection_id,
             created_at_epoch: prost.created_at_epoch.map(Epoch::from),
             initialized_at_epoch: prost.initialized_at_epoch.map(Epoch::from),
-            description: prost.description.clone(),
             version,
         }
     }

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

@@ -185,7 +185,6 @@ impl From<&BuiltinView> for ViewCatalog {
             sql: val.sql.to_string(),
             owner: DEFAULT_SUPER_USER_ID,
             properties: Default::default(),
-            description: None,
         }
     }
 }