fix include column (hacky)

risingwavelabs · Jun 19, 2024 · 97e0a4d · 97e0a4d
1 parent 6acfac5
commit 97e0a4d
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 5 deletions.
diff --git a/e2e_test/schema_registry/alter_sr.slt b/e2e_test/schema_registry/alter_sr.slt
@@ -9,7 +9,7 @@ python3 e2e_test/schema_registry/pb.py "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${R
 
 statement ok
 CREATE SOURCE src_user
-INCLUDE timestamp
+INCLUDE timestamp -- include explicitly here to test a bug found in https://github.com/risingwavelabs/risingwave/pull/17293
 WITH (
     ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
     topic = 'sr_pb_test',

diff --git a/e2e_test/source_inline/kafka/avro/alter_source.slt b/e2e_test/source_inline/kafka/avro/alter_source.slt
@@ -59,6 +59,8 @@ ABC	1
 statement ok
 create materialized view mv as select * from s;
 
+sleep 2s
+
 query ??
 select * from mv
 ----

diff --git a/src/common/src/catalog/column.rs b/src/common/src/catalog/column.rs
@@ -350,6 +350,11 @@ impl ColumnCatalog {
         self.column_desc.is_default()
     }
 
+    /// If the columns is an `INCLUDE ... AS ...` connector column.
+    pub fn is_connector_additional_column(&self) -> bool {
+        self.column_desc.additional_column.column_type.is_some()
+    }
+
     /// Get a reference to the column desc's data type.
     pub fn data_type(&self) -> &DataType {
         &self.column_desc.data_type

diff --git a/src/frontend/src/handler/alter_source_with_sr.rs b/src/frontend/src/handler/alter_source_with_sr.rs
@@ -69,14 +69,19 @@ fn encode_type_to_encode(from: EncodeType) -> Option<Encode> {
     })
 }
 
-/// Returns the columns in `columns_a` but not in `columns_b`,
-/// where the comparison is done by name and data type,
-/// and hidden columns are ignored.
+/// Returns the columns in `columns_a` but not in `columns_b`.
+///
+/// Note:
+/// - The comparison is done by name and data type, without checking `ColumnId`.
+/// - Hidden columns and `INCLUDE ... AS ...` columns are ignored. Because it's only for the special handling of alter sr.
+///   For the newly resolved `columns_from_resolve_source` (created by [`bind_columns_from_source`]), it doesn't contain hidden columns (`_row_id`) and `INCLUDE ... AS ...` columns.
+///   This is fragile and we should really refactor it later.
 fn columns_minus(columns_a: &[ColumnCatalog], columns_b: &[ColumnCatalog]) -> Vec<ColumnCatalog> {
     columns_a
         .iter()
         .filter(|col_a| {
             !col_a.is_hidden()
+                && !col_a.is_connector_additional_column()
                 && !columns_b.iter().any(|col_b| {
                     col_a.name() == col_b.name() && col_a.data_type() == col_b.data_type()
                 })
@@ -172,7 +177,7 @@ pub async fn refresh_sr_and_get_columns_diff(
     }
     let dropped_columns = columns_minus(&original_source.columns, &columns_from_resolve_source);
     tracing::debug!(
-        added_columns = ?added_columns,
+        ?added_columns,
         ?dropped_columns,
         ?columns_from_resolve_source,
         original_source = ?original_source.columns