risingwavelabs · mergify · Nov 16, 2022 · Oct 25, 2022 · Oct 25, 2022 · Nov 7, 2022
diff --git a/dashboard/proto/gen/catalog.ts b/dashboard/proto/gen/catalog.ts
diff --git a/dashboard/proto/gen/stream_plan.ts b/dashboard/proto/gen/stream_plan.ts
diff --git a/proto/catalog.proto b/proto/catalog.proto
@@ -98,6 +98,7 @@ message Table {
   // `[0..columns.len()]`.
   repeated int32 value_indices = 19;
   string definition = 20;
+  bool handle_pk_conflict = 21;
 }
 
 message View {

diff --git a/proto/stream_plan.proto b/proto/stream_plan.proto
@@ -155,7 +155,7 @@ message MaterializeNode {
   // Used for internal table states.
   catalog.Table table = 3;
   // Used to control whether doing sanity check, open it when upstream executor is source executor.
-  bool ignore_on_conflict = 4;
+  bool handle_pk_conflict = 4;
 }
 
 message AggCallState {
@@ -349,7 +349,7 @@ message ArrangeNode {
   // Used for internal table states.
   catalog.Table table = 3;
   // Used to control whether doing sanity check, open it when upstream executor is source executor.
-  bool ignore_on_conflict = 4;
+  bool handle_pk_conflict = 4;
 }
 
 // Special node for shared state. LookupNode will join an arrangement with a stream.

diff --git a/src/compute/tests/integration_tests.rs b/src/compute/tests/integration_tests.rs
@@ -160,6 +160,9 @@ async fn test_table_materialize() -> StreamResult<()> {
         vec![OrderPair::new(0, OrderType::Ascending)],
         all_column_ids.clone(),
         2,
+        None,
+        0,
+        false,
     )
     .await
     .boxed()

diff --git a/src/ctl/src/cmd_impl/table/scan.rs b/src/ctl/src/cmd_impl/table/scan.rs
@@ -65,7 +65,7 @@ pub async fn make_state_table<S: StateStore>(hummock: S, table: &TableCatalog) -
         table.pk().iter().map(|x| x.direct.to_order()).collect(),
         table.pk().iter().map(|x| x.index).collect(),
         Distribution::all_vnodes(table.distribution_key().to_vec()), // scan all vnodes
-        table.value_indices.clone(),
+        Some(table.value_indices.clone()),
     )
     .await
 }

diff --git a/src/frontend/src/catalog/table_catalog.rs b/src/frontend/src/catalog/table_catalog.rs
@@ -96,6 +96,8 @@ pub struct TableCatalog {
 
     /// Definition of the materialized view.
     pub definition: String,
+
+    pub handle_pk_conflict: bool,
 }
 
 impl TableCatalog {
@@ -109,6 +111,10 @@ impl TableCatalog {
         self
     }
 
+    pub fn handle_pk_conflict(&self) -> bool {
+        self.handle_pk_conflict
+    }
+
     /// Get the table catalog's associated source id.
     #[must_use]
     pub fn associated_source_id(&self) -> Option<TableId> {
@@ -190,6 +196,7 @@ impl TableCatalog {
                 .map(|i| ProstColumnIndex { index: i as _ }),
             value_indices: self.value_indices.iter().map(|x| *x as _).collect(),
             definition: self.definition.clone(),
+            handle_pk_conflict: self.handle_pk_conflict,
         }
     }
 }
@@ -236,6 +243,7 @@ impl From<ProstTable> for TableCatalog {
             vnode_col_idx: tb.vnode_col_idx.map(|x| x.index as usize),
             value_indices: tb.value_indices.iter().map(|x| *x as _).collect(),
             definition: tb.definition.clone(),
+            handle_pk_conflict: tb.handle_pk_conflict,
         }
     }
 }
@@ -320,6 +328,7 @@ mod tests {
             vnode_col_idx: None,
             value_indices: vec![0],
             definition: "".into(),
+            handle_pk_conflict: false,
         }
         .into();
 
@@ -377,6 +386,7 @@ mod tests {
                 vnode_col_idx: None,
                 value_indices: vec![0],
                 definition: "".into(),
+                handle_pk_conflict: false
             }
         );
         assert_eq!(table, TableCatalog::from(table.to_prost(0, 0)));

diff --git a/src/frontend/src/handler/create_mv.rs b/src/frontend/src/handler/create_mv.rs
@@ -93,7 +93,7 @@ pub fn gen_create_mv_plan(
             .into());
         }
     }
-    let materialize = plan_root.gen_create_mv_plan(table_name, definition, col_names)?;
+    let materialize = plan_root.gen_create_mv_plan(table_name, definition, col_names, false)?;
     let mut table = materialize.table().to_prost(schema_id, database_id);
     if session.config().get_create_compaction_group_for_mv() {
         table.properties.insert(

diff --git a/src/frontend/src/handler/create_table.rs b/src/frontend/src/handler/create_table.rs
@@ -232,6 +232,9 @@ pub(crate) fn gen_materialized_source_plan(
         // Manually assemble the materialization plan for the table.
         let source_node: PlanRef = LogicalSource::new(Rc::new((&source).into()), context).into();
         let row_id_index = source.row_id_index.as_ref().map(|index| index.index as _);
+        // row_id_index is Some means that the user has not specified pk, then we will add a hidden
+        // column to store pk, and materialize executor do not need to handle pk conflict.
+        let handle_pk_conflict = row_id_index.is_none();
         let mut required_cols = FixedBitSet::with_capacity(source_node.schema().len());
         required_cols.toggle_range(..);
         let mut out_names = source_node.schema().names();
@@ -247,7 +250,7 @@ pub(crate) fn gen_materialized_source_plan(
             required_cols,
             out_names,
         )
-        .gen_create_mv_plan(source.name.clone(), "".into(), None)?
+        .gen_create_mv_plan(source.name.clone(), "".into(), None, handle_pk_conflict)?
     };
     let mut table = materialize
         .table()

diff --git a/src/frontend/src/optimizer/mod.rs b/src/frontend/src/optimizer/mod.rs
@@ -482,6 +482,7 @@ impl PlanRoot {
         mv_name: String,
         definition: String,
         col_names: Option<Vec<String>>,
+        handle_pk_conflict: bool,
     ) -> Result<StreamMaterialize> {
         let out_names = if let Some(col_names) = col_names {
             col_names
@@ -498,6 +499,7 @@ impl PlanRoot {
             out_names,
             false,
             definition,
+            handle_pk_conflict,
         )
     }
 
@@ -513,6 +515,7 @@ impl PlanRoot {
             self.out_names.clone(),
             true,
             "".into(),
+            false,
         )
     }
 

diff --git a/src/frontend/src/optimizer/plan_node/stream.rs b/src/frontend/src/optimizer/plan_node/stream.rs
@@ -658,7 +658,7 @@ pub fn to_stream_prost_body(
                 table_id: 0,
                 column_orders: me.table.pk().iter().map(FieldOrder::to_protobuf).collect(),
                 table: Some(me.table.to_internal_table_prost()),
-                ignore_on_conflict: true,
+                handle_pk_conflict: false,
             })
         }
         Node::ProjectSet(me) => {

diff --git a/src/frontend/src/optimizer/plan_node/stream_materialize.rs b/src/frontend/src/optimizer/plan_node/stream_materialize.rs
@@ -84,6 +84,7 @@ impl StreamMaterialize {
         out_names: Vec<String>,
         is_index: bool,
         definition: String,
+        handle_pk_conflict: bool,
     ) -> Result<Self> {
         let required_dist = match input.distribution() {
             Distribution::Single => RequiredDist::single(),
@@ -167,7 +168,6 @@ impl StreamMaterialize {
 
         let ctx = input.ctx();
         let properties = ctx.inner().with_options.internal_table_subset();
-
         let table = TableCatalog {
             id: TableId::placeholder(),
             associated_source_id: None,
@@ -185,6 +185,7 @@ impl StreamMaterialize {
             vnode_col_idx: None,
             value_indices,
             definition,
+            handle_pk_conflict,
         };
 
         Ok(Self { base, input, table })
@@ -265,7 +266,7 @@ impl StreamNode for StreamMaterialize {
                 .map(FieldOrder::to_protobuf)
                 .collect(),
             table: Some(self.table().to_internal_table_prost()),
-            ignore_on_conflict: true,
+            handle_pk_conflict: self.table.handle_pk_conflict(),
         })
     }
 }
diff --git a/src/frontend/src/optimizer/plan_node/utils.rs b/src/frontend/src/optimizer/plan_node/utils.rs
@@ -123,6 +123,7 @@ impl TableCatalogBuilder {
                 .value_indices
                 .unwrap_or_else(|| (0..self.columns.len()).collect_vec()),
             definition: "".into(),
+            handle_pk_conflict: false,
         }
     }
 

diff --git a/src/meta/src/stream/test_fragmenter.rs b/src/meta/src/stream/test_fragmenter.rs
@@ -336,7 +336,7 @@ fn make_stream_fragments() -> Vec<StreamFragment> {
             table_id: 1,
             table: Some(make_internal_table(4, true)),
             column_orders: vec![make_column_order(1), make_column_order(2)],
-            ignore_on_conflict: true,
+            handle_pk_conflict: false,
         })),
         fields: vec![], // TODO: fill this later
         operator_id: 7,

diff --git a/src/storage/hummock_sdk/src/filter_key_extractor.rs b/src/storage/hummock_sdk/src/filter_key_extractor.rs
@@ -476,6 +476,7 @@ mod tests {
             vnode_col_idx: None,
             value_indices: vec![0],
             definition: "".into(),
+            handle_pk_conflict: false,
         }
     }
 

diff --git a/src/storage/src/table/mod.rs b/src/storage/src/table/mod.rs
@@ -125,7 +125,11 @@ fn compute_vnode(row: &Row, indices: &[usize], vnodes: &Bitmap) -> VirtualNode {
 }
 
 /// Get vnode values with `indices` on the given `chunk`.
-fn compute_chunk_vnode(chunk: &DataChunk, indices: &[usize], vnodes: &Bitmap) -> Vec<VirtualNode> {
+pub fn compute_chunk_vnode(
+    chunk: &DataChunk,
+    indices: &[usize],
+    vnodes: &Bitmap,
+) -> Vec<VirtualNode> {
     if indices.is_empty() {
         vec![DEFAULT_VNODE; chunk.capacity()]
     } else {
-Original file line number
+Diff line change
@@ Expand Up / @@ -160,6 +160,9 @@ async fn test_table_materialize() -> StreamResult<()> { @@
             vec![OrderPair::new(0, OrderType::Ascending)],
             all_column_ids.clone(),
 ,
+            None,
+,
+            false,
         )
         .await
         .boxed()
@@ Expand Down @@