fix(source): REFRESH SCHEMA shall keep INCLUDE pk for UPSERT

risingwavelabs · Nov 13, 2024 · af3516d · af3516d
1 parent 1aed314
commit af3516d
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 1 deletion.
diff --git a/src/frontend/src/handler/alter_table_column.rs b/src/frontend/src/handler/alter_table_column.rs
@@ -180,6 +180,7 @@ pub async fn get_replace_table_plan(
         wildcard_idx,
         cdc_table_info,
         format_encode,
+        include_column_options,
         ..
     } = new_definition
     else {
@@ -206,6 +207,7 @@ pub async fn get_replace_table_plan(
         with_version_column,
         cdc_table_info,
         new_version_columns,
+        include_column_options,
     )
     .await?;
 

diff --git a/src/frontend/src/handler/create_sink.rs b/src/frontend/src/handler/create_sink.rs
@@ -667,6 +667,7 @@ pub(crate) async fn reparse_table_for_sink(
         append_only,
         on_conflict,
         with_version_column,
+        include_column_options,
         ..
     } = definition
     else {
@@ -689,6 +690,7 @@ pub(crate) async fn reparse_table_for_sink(
         with_version_column,
         None,
         None,
+        include_column_options,
     )
     .await?;
 

diff --git a/src/frontend/src/handler/create_table.rs b/src/frontend/src/handler/create_table.rs
@@ -1309,6 +1309,7 @@ pub async fn generate_stream_graph_for_replace_table(
     with_version_column: Option<String>,
     cdc_table_info: Option<CdcTableInfo>,
     new_version_columns: Option<Vec<ColumnCatalog>>,
+    include_column_options: IncludeOption,
 ) -> Result<(StreamFragmentGraph, Table, Option<PbSource>, TableJobType)> {
     use risingwave_pb::catalog::table::OptionalAssociatedSourceId;
 
@@ -1327,7 +1328,7 @@ pub async fn generate_stream_graph_for_replace_table(
                 append_only,
                 on_conflict,
                 with_version_column,
-                vec![],
+                include_column_options,
             )
             .await?,
             TableJobType::General,