Merge branch 'main' of https://github.com/singularity-data/risingwave …

…into li0k/storage_ttl_selector

proto/batch_plan.proto

@@ -77,6 +77,8 @@ message FilterNode {
 message InsertNode {
   // Id of the table to perform inserting.
   uint32 table_id = 1;
+  // Version of the table.
+  uint64 table_version_id = 5;
   repeated uint32 column_indices = 2;
   // An optional field and will be `None` for tables without user-defined pk.
   // The `BatchInsertExecutor` should add a column with NULL value which will
@@ -88,12 +90,16 @@ message InsertNode {
 message DeleteNode {
   // Id of the table to perform deleting.
   uint32 table_id = 1;
+  // Version of the table.
+  uint64 table_version_id = 3;
   bool returning = 2;
 }
 
 message UpdateNode {
   // Id of the table to perform updating.
   uint32 table_id = 1;
+  // Version of the table.
+  uint64 table_version_id = 4;
   repeated expr.ExprNode exprs = 2;
   bool returning = 3;
 }

proto/stream_plan.proto

@@ -471,6 +471,8 @@ message SortNode {
 message DmlNode {
   // Id of the table on which DML performs.
   uint32 table_id = 1;
+  // Version of the table.
+  uint64 table_version_id = 3;
   // Column descriptions of the table.
   repeated plan_common.ColumnDesc column_descs = 2;
 }

src/batch/src/executor/delete.rs

@@ -16,7 +16,7 @@ use anyhow::anyhow;
 use futures::future::try_join_all;
 use futures_async_stream::try_stream;
 use risingwave_common::array::{ArrayBuilder, DataChunk, Op, PrimitiveArrayBuilder, StreamChunk};
-use risingwave_common::catalog::{Field, Schema, TableId};
+use risingwave_common::catalog::{Field, Schema, TableId, TableVersionId};
 use risingwave_common::error::{Result, RwError};
 use risingwave_common::types::DataType;
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
@@ -35,6 +35,7 @@ use crate::task::BatchTaskContext;
 pub struct DeleteExecutor {
     /// Target table id.
     table_id: TableId,
+    table_version_id: TableVersionId,
     dml_manager: DmlManagerRef,
     child: BoxedExecutor,
     chunk_size: usize,
@@ -46,6 +47,7 @@ pub struct DeleteExecutor {
 impl DeleteExecutor {
     pub fn new(
         table_id: TableId,
+        table_version_id: TableVersionId,
         dml_manager: DmlManagerRef,
         child: BoxedExecutor,
         chunk_size: usize,
@@ -55,6 +57,7 @@ impl DeleteExecutor {
         let table_schema = child.schema().clone();
         Self {
             table_id,
+            table_version_id,
             dml_manager,
             child,
             chunk_size,
@@ -98,7 +101,9 @@ impl DeleteExecutor {
             let cap = chunk.capacity();
             let stream_chunk = StreamChunk::from_parts(vec![Op::Delete; cap], chunk);
 
-            let notifier = self.dml_manager.write_chunk(&self.table_id, stream_chunk)?;
+            let notifier =
+                self.dml_manager
+                    .write_chunk(self.table_id, self.table_version_id, stream_chunk)?;
             notifiers.push(notifier);
 
             Ok(())
@@ -155,6 +160,7 @@ impl BoxedExecutorBuilder for DeleteExecutor {
 
         Ok(Box::new(Self::new(
             table_id,
+            delete_node.table_version_id,
             source.context().dml_manager(),
             child,
             source.context.get_config().developer.batch_chunk_size,
@@ -171,7 +177,9 @@ mod tests {
     use futures::StreamExt;
     use itertools::Itertools;
     use risingwave_common::array::Array;
-    use risingwave_common::catalog::{schema_test_utils, ColumnDesc, ColumnId};
+    use risingwave_common::catalog::{
+        schema_test_utils, ColumnDesc, ColumnId, INITIAL_TABLE_VERSION_ID,
+    };
     use risingwave_common::test_prelude::DataChunkTestExt;
     use risingwave_source::dml_manager::DmlManager;
 
@@ -210,13 +218,14 @@ mod tests {
         // We must create a variable to hold this `Arc<TableDmlHandle>` here, or it will be dropped
         // due to the `Weak` reference in `DmlManager`.
         let reader = dml_manager
-            .register_reader(table_id, &column_descs)
+            .register_reader(table_id, INITIAL_TABLE_VERSION_ID, &column_descs)
             .unwrap();
         let mut reader = reader.stream_reader().into_stream();
 
         // Delete
         let delete_executor = Box::new(DeleteExecutor::new(
             table_id,
+            INITIAL_TABLE_VERSION_ID,
             dml_manager,
             Box::new(mock_executor),
             1024,

src/batch/src/executor/insert.rs

@@ -20,7 +20,7 @@ use futures_async_stream::try_stream;
 use risingwave_common::array::{
     ArrayBuilder, DataChunk, I64Array, Op, PrimitiveArrayBuilder, StreamChunk,
 };
-use risingwave_common::catalog::{Field, Schema, TableId};
+use risingwave_common::catalog::{Field, Schema, TableId, TableVersionId};
 use risingwave_common::error::{Result, RwError};
 use risingwave_common::types::DataType;
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
@@ -36,6 +36,7 @@ use crate::task::BatchTaskContext;
 pub struct InsertExecutor {
     /// Target table id.
     table_id: TableId,
+    table_version_id: TableVersionId,
     dml_manager: DmlManagerRef,
 
     child: BoxedExecutor,
@@ -52,6 +53,7 @@ impl InsertExecutor {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         table_id: TableId,
+        table_version_id: TableVersionId,
         dml_manager: DmlManagerRef,
         child: BoxedExecutor,
         chunk_size: usize,
@@ -63,6 +65,7 @@ impl InsertExecutor {
         let table_schema = child.schema().clone();
         Self {
             table_id,
+            table_version_id,
             dml_manager,
             child,
             chunk_size,
@@ -127,7 +130,9 @@ impl InsertExecutor {
             let stream_chunk =
                 StreamChunk::new(vec![Op::Insert; cap], columns, vis.into_visibility());
 
-            let notifier = self.dml_manager.write_chunk(&self.table_id, stream_chunk)?;
+            let notifier =
+                self.dml_manager
+                    .write_chunk(self.table_id, self.table_version_id, stream_chunk)?;
             notifiers.push(notifier);
 
             Ok(())
@@ -190,6 +195,7 @@ impl BoxedExecutorBuilder for InsertExecutor {
 
         Ok(Box::new(Self::new(
             table_id,
+            insert_node.table_version_id,
             source.context().dml_manager(),
             child,
             source.context.get_config().developer.batch_chunk_size,
@@ -212,7 +218,9 @@ mod tests {
     use futures::StreamExt;
     use itertools::Itertools;
     use risingwave_common::array::{Array, ArrayImpl, I32Array, StructArray};
-    use risingwave_common::catalog::{schema_test_utils, ColumnDesc, ColumnId};
+    use risingwave_common::catalog::{
+        schema_test_utils, ColumnDesc, ColumnId, INITIAL_TABLE_VERSION_ID,
+    };
     use risingwave_common::column_nonnull;
     use risingwave_common::types::DataType;
     use risingwave_source::dml_manager::DmlManager;
@@ -274,13 +282,14 @@ mod tests {
         // We must create a variable to hold this `Arc<TableDmlHandle>` here, or it will be dropped
         // due to the `Weak` reference in `DmlManager`.
         let reader = dml_manager
-            .register_reader(table_id, &column_descs)
+            .register_reader(table_id, INITIAL_TABLE_VERSION_ID, &column_descs)
             .unwrap();
         let mut reader = reader.stream_reader().into_stream();
 
         // Insert
         let insert_executor = Box::new(InsertExecutor::new(
             table_id,
+            INITIAL_TABLE_VERSION_ID,
             dml_manager,
             Box::new(mock_executor),
             1024,

src/batch/src/executor/join/hash_join.rs

@@ -29,7 +29,7 @@ use risingwave_common::hash::{HashKey, HashKeyDispatcher, PrecomputedBuildHasher
 use risingwave_common::row::{repeat_n, RowExt};
 use risingwave_common::types::{DataType, Datum};
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
-use risingwave_common::util::iter_util::ZipEqDebug;
+use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_expr::expr::{build_from_prost, BoxedExpression, Expression};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 
@@ -1530,7 +1530,7 @@ impl DataChunkMutator {
     ) -> Self {
         let mut new_visibility = BitmapBuilder::zeroed(self.0.capacity());
         for (output_row_id, (output_row_non_null, &build_row_id)) in
-            filter.iter().zip_eq_debug(build_row_ids.iter()).enumerate()
+            filter.iter().zip_eq_fast(build_row_ids.iter()).enumerate()
         {
             if output_row_non_null {
                 build_row_matched[build_row_id] = true;
@@ -1550,7 +1550,7 @@ impl DataChunkMutator {
         build_row_ids: &mut Vec<RowId>,
         build_row_matched: &mut ChunkedData<bool>,
     ) {
-        for (output_row_non_null, &build_row_id) in filter.iter().zip_eq_debug(build_row_ids.iter())
+        for (output_row_non_null, &build_row_id) in filter.iter().zip_eq_fast(build_row_ids.iter())
         {
             if output_row_non_null {
                 build_row_matched[build_row_id] = true;
@@ -1607,7 +1607,7 @@ impl DataChunkMutator {
         first_output_row_id.clear();
 
         for (output_row_id, (output_row_non_null, &build_row_id)) in
-            filter.iter().zip_eq_debug(build_row_ids.iter()).enumerate()
+            filter.iter().zip_eq_fast(build_row_ids.iter()).enumerate()
         {
             if output_row_non_null {
                 build_row_matched[build_row_id] = true;