feat: add comment on clause support

e2e_test/batch/catalog/pg_description.slt.part

@@ -1,18 +1,65 @@
-query IIITT
-SELECT d.*, c.relname FROM pg_catalog.pg_description d join pg_catalog.pg_class c on d.objoid = c.oid ORDER BY d.objoid limit 15;
+statement ok
+create table t(a int, b text, c date);
+
+statement ok
+comment on table t is 'Lorem ipsum';
+
+statement ok
+comment on column t.a is 'Praesent elementum';
+
+statement ok
+comment on column public.t.c is 'Nullam ultricies';
+
+statement ok
+comment on column public.t._row_id is 'facilisis enim';
+
+query TIIIT
+SELECT
+    c.relname,
+    (
+        SELECT relname FROM pg_catalog.pg_class WHERE oid = d.classoid
+    ) AS classoid,
+    d.objsubid,
+    d.description
+FROM
+    pg_catalog.pg_description d
+    JOIN pg_catalog.pg_class c
+        ON d.objoid = c.oid
+ORDER BY d.objsubid;
 ----
-1 NULL 0 NULL pg_type
-2 NULL 0 NULL pg_namespace
-3 NULL 0 NULL pg_cast
-4 NULL 0 NULL pg_matviews
-5 NULL 0 NULL pg_user
-6 NULL 0 NULL pg_class
-7 NULL 0 NULL pg_index
-8 NULL 0 NULL pg_opclass
-9 NULL 0 NULL pg_collation
-10 NULL 0 NULL pg_am
-11 NULL 0 NULL pg_operator
-12 NULL 0 NULL pg_views
-13 NULL 0 NULL pg_attribute
-14 NULL 0 NULL pg_database
-15 NULL 0 NULL pg_description
+t  rw_tables  -1  facilisis enim
+t  rw_tables   0  Lorem ipsum
+t  rw_tables   1  Praesent elementum
+t  rw_tables   3  Nullam ultricies
+
+statement ok
+comment on table public.t is NULL;
+
+statement ok
+comment on column t._row_id is NULL;
+
+statement ok
+comment on column t.c is '';
+
+statement ok
+comment on column public.t.b is 'Vivamus fermentum';
+
+query TIIIT
+SELECT
+    c.relname,
+    (
+        SELECT relname FROM pg_catalog.pg_class WHERE oid = d.classoid
+    ) AS classoid,
+    d.objsubid,
+    d.description
+FROM
+    pg_catalog.pg_description d
+    JOIN pg_catalog.pg_class c
+        ON d.objoid = c.oid
+ORDER BY d.objsubid;
+----
+t  rw_tables  1  Praesent elementum
+t  rw_tables  2  Vivamus fermentum
+
+statement ok
+drop table t;

e2e_test/ddl/show.slt

@@ -7,15 +7,28 @@ 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 table t3 is 'volutpat vitae';
+
+statement ok
+comment on column t3.v1 is 'turpis vehicula';
+
+statement ok
+comment on column t3.v2 is 'Lorem ipsum dolor sit amet';
+
+statement ok
+comment on column public.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  turpis vehicula
+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
+table description       t3   NULL  volutpat vitae
 
 query TTT
 show columns from t3;
@@ -33,16 +46,29 @@ show indexes from t3;
 ----
 idx1 t3 v1 ASC, v2 ASC v3 v1
 
-query TTT
+statement ok
+comment on table public.t3 is 'consectetur turpis';
+
+statement ok
+comment on column t3.v1 is 'Nemo enim ipsam';
+
+statement ok
+comment on column t3.v2 is '';
+
+statement ok
+comment on column t3._row_id is NULL;
+
+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  Nemo enim ipsam
+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
+idx1               index(v1 ASC, v2 ASC) include(v3) distributed by(v1) NULL NULL
+table description       t3   NULL  consectetur turpis
 
 query TT
 show create index idx1;

e2e_test/extended_mode/basic.slt

@@ -39,15 +39,16 @@ 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
+table description t3        NULL  NULL
 
 query TTT
 show columns from t3;

proto/catalog.proto

@@ -280,6 +280,9 @@ message Table {
 
   CreateType create_type = 32;
 
+  // This field is used to store the description set by the `comment on` clause.
+  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 +320,11 @@ message Database {
   string name = 2;
   uint32 owner = 3;
 }
+
+message Comment {
+  uint32 table_id = 1;
+  uint32 schema_id = 2;
+  uint32 database_id = 3;
+  optional uint32 column_index = 4;
+  optional string description = 5;
+}

proto/ddl_service.proto

@@ -318,6 +318,15 @@ message WaitRequest {}
 
 message WaitResponse {}
 
+message CommentOnRequest {
+  catalog.Comment comment = 1;
+}
+
+message CommentOnResponse {
+  common.Status status = 1;
+  uint64 version = 2;
+}
+
 service DdlService {
   rpc CreateDatabase(CreateDatabaseRequest) returns (CreateDatabaseResponse);
   rpc DropDatabase(DropDatabaseRequest) returns (DropDatabaseResponse);
@@ -348,4 +357,5 @@ service DdlService {
   rpc DropConnection(DropConnectionRequest) returns (DropConnectionResponse);
   rpc GetTables(GetTablesRequest) returns (GetTablesResponse);
   rpc Wait(WaitRequest) returns (WaitResponse);
+  rpc CommentOn(CommentOnRequest) returns (CommentOnResponse);
 }

proto/plan_common.proto

@@ -37,6 +37,9 @@ message ColumnDesc {
     GeneratedColumnDesc generated_column = 6;
     DefaultColumnDesc default_column = 7;
   }
+
+  // This field is used to store the description set by the `comment on` clause.
+  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,
         }
     }
 
@@ -129,6 +131,7 @@ impl ColumnDesc {
                 .collect_vec(),
             type_name: self.type_name.clone(),
             generated_or_default_column: self.generated_or_default_column.clone(),
+            description: self.description.clone(),
         }
     }
 
@@ -172,6 +175,7 @@ impl ColumnDesc {
             field_descs: vec![],
             type_name: "".to_string(),
             generated_or_default_column: None,
+            description: None,
         }
     }
 
@@ -192,6 +196,7 @@ impl ColumnDesc {
             field_descs: fields,
             type_name: type_name.to_string(),
             generated_or_default_column: None,
+            description: 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,
         }
     }
@@ -243,6 +249,7 @@ impl From<PbColumnDesc> for ColumnDesc {
             type_name: prost.type_name,
             field_descs,
             generated_or_default_column: prost.generated_or_default_column,
+            description: prost.description.clone(),
         }
     }
 }
@@ -262,6 +269,7 @@ impl From<&ColumnDesc> for PbColumnDesc {
             field_descs: c.field_descs.iter().map(ColumnDesc::to_protobuf).collect(),
             type_name: c.type_name.clone(),
             generated_or_default_column: c.generated_or_default_column.clone(),
+            description: c.description.clone(),
         }
     }
 }

src/common/src/catalog/mod.rs

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

src/common/src/catalog/test_utils.rs

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

src/compute/tests/integration_tests.rs

@@ -159,6 +159,7 @@ async fn test_table_materialize() -> StreamResult<()> {
             field_descs: vec![],
             type_name: "".to_string(),
             generated_or_default_column: None,
+            description: None,
         })
         .collect_vec();
     let (barrier_tx, barrier_rx) = unbounded_channel();

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

@@ -57,6 +57,7 @@ fn avro_field_to_column_desc(
                 field_descs: vec_column,
                 type_name: schema_name.to_string(),
                 generated_or_default_column: None,
+                description: None,
             })
         }
         _ => {

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

@@ -199,6 +199,7 @@ impl ProtobufParserConfig {
                 field_descs,
                 type_name: m.full_name().to_string(),
                 generated_or_default_column: None,
+                description: None,
             })
         } else {
             *index += 1;

src/connector/src/source/manager.rs

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

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

@@ -576,6 +576,7 @@ pub fn bind_struct_field(column_def: &StructField) -> Result<ColumnDesc> {
                     field_descs: vec![],
                     type_name: "".to_string(),
                     generated_or_default_column: None,
+                    description: None,
                 })
             })
             .collect::<Result<Vec<_>>>()?
@@ -589,6 +590,7 @@ pub fn bind_struct_field(column_def: &StructField) -> Result<ColumnDesc> {
         field_descs,
         type_name: "".to_string(),
         generated_or_default_column: None,
+        description: None,
     })
 }