add mutation serialize transform

zhyass · zhyass · commit 51caf0cd3126 · 2023-02-20T10:31:33.000+08:00
diff --git a/src/common/io/src/constants.rs b/src/common/io/src/constants.rs
@@ -28,6 +28,8 @@ pub const INF_BYTES_LONG: &str = "Infinity";
 pub const DEFAULT_BLOCK_BUFFER_SIZE: usize = 100 * 1024 * 1024;
 // The size of the I/O read/write block index buffer by default.
 pub const DEFAULT_BLOCK_INDEX_BUFFER_SIZE: usize = 300 * 1024;
+// The size of the I/O read/write delete mark buffer by default.
+pub const DEFAULT_BLOCK_DELETE_MARK_SIZE: usize = 1000 * 1000;
 // The max number of a block by default.
 pub const DEFAULT_BLOCK_MAX_ROWS: usize = 1000 * 1000;
 // The min number of a block by default.
diff --git a/src/query/storages/fuse/src/fuse_part.rs b/src/query/storages/fuse/src/fuse_part.rs
@@ -28,6 +28,7 @@ use common_expression::ColumnId;
 use common_expression::Scalar;
 use storages_common_table_meta::meta::ColumnMeta;
 use storages_common_table_meta::meta::Compression;
+use storages_common_table_meta::meta::Location;
 
 #[derive(serde::Serialize, serde::Deserialize, PartialEq, Eq)]
 pub struct FusePartInfo {
@@ -42,7 +43,7 @@ pub struct FusePartInfo {
     pub sort_min_max: Option<(Scalar, Scalar)>,
     /// page range in the file
     pub range: Option<Range<usize>>,
-    pub delete_mark: Option<String>,
+    pub delete_mark: Option<(Location, u64)>,
 }
 
 #[typetag::serde(name = "fuse")]
@@ -75,7 +76,7 @@ impl FusePartInfo {
         compression: Compression,
         sort_min_max: Option<(Scalar, Scalar)>,
         range: Option<Range<usize>>,
-        delete_mark: Option<String>,
+        delete_mark: Option<(Location, u64)>,
     ) -> Arc<Box<dyn PartInfo>> {
         Arc::new(Box::new(FusePartInfo {
             location,
diff --git a/src/query/storages/fuse/src/fuse_table.rs b/src/query/storages/fuse/src/fuse_table.rs
@@ -325,6 +325,10 @@ impl Table for FuseTable {
         true
     }
 
+    fn support_delete_mark(&self) -> bool {
+        false
+    }
+
     fn cluster_keys(&self, ctx: Arc<dyn TableContext>) -> Vec<RemoteExpr<String>> {
         let table_meta = Arc::new(self.clone());
         if let Some((_, order)) = &self.cluster_key_meta {
@@ -552,7 +556,7 @@ impl Table for FuseTable {
         col_indices: Vec<usize>,
         pipeline: &mut Pipeline,
     ) -> Result<()> {
-        self.do_delete(ctx, filter, col_indices, pipeline).await
+        self.do_delete2(ctx, filter, col_indices, pipeline).await
     }
 
     async fn update(
diff --git a/src/query/storages/fuse/src/io/mod.rs b/src/query/storages/fuse/src/io/mod.rs
@@ -23,6 +23,7 @@ pub use files::Files;
 pub use locations::TableMetaLocationGenerator;
 pub use read::BlockReader;
 pub use read::BloomBlockFilterReader;
+pub use read::DeleteMarkReader;
 pub use read::MergeIOReadResult;
 pub use read::MetaReaders;
 pub use read::NativeReaderExt;
diff --git a/src/query/storages/fuse/src/io/read/delete/delete_mark_reader.rs b/src/query/storages/fuse/src/io/read/delete/delete_mark_reader.rs
@@ -47,16 +47,26 @@ type CachedReader = InMemoryItemCacheReader<Bitmap, DeleteMetaLoader>;
 
 #[async_trait::async_trait]
 pub trait DeleteMarkReader {
-    async fn read_delete_mark(&self, dal: Operator, length: u64) -> Result<Arc<Bitmap>>;
+    async fn read_delete_mark(
+        &self,
+        dal: Operator,
+        size: u64,
+        num_rows: usize,
+    ) -> Result<Arc<Bitmap>>;
 }
 
 #[async_trait::async_trait]
 impl DeleteMarkReader for Location {
-    async fn read_delete_mark(&self, dal: Operator, length: u64) -> Result<Arc<Bitmap>> {
+    async fn read_delete_mark(
+        &self,
+        dal: Operator,
+        size: u64,
+        num_rows: usize,
+    ) -> Result<Arc<Bitmap>> {
         let (path, ver) = &self;
         let mask_version = DeleteMaskVersion::try_from(*ver)?;
         if matches!(mask_version, DeleteMaskVersion::V0(_)) {
-            let res = load_delete_mark(dal, path, length).await?;
+            let res = load_delete_mark(dal, path, size, num_rows).await?;
             Ok(res)
         } else {
             unreachable!()
@@ -66,9 +76,14 @@ impl DeleteMarkReader for Location {
 
 /// load delete mark data
 #[tracing::instrument(level = "debug", skip_all)]
-pub async fn load_delete_mark(dal: Operator, path: &str, length: u64) -> Result<Arc<Bitmap>> {
+pub async fn load_delete_mark(
+    dal: Operator,
+    path: &str,
+    size: u64,
+    num_rows: usize,
+) -> Result<Arc<Bitmap>> {
     // 1. load delete mark meta
-    let delete_mark_meta = load_mark_meta(dal.clone(), path, length).await?;
+    let delete_mark_meta = load_mark_meta(dal.clone(), path, size).await?;
     let file_meta = &delete_mark_meta.0;
     if file_meta.row_groups.len() != 1 {
         return Err(ErrorCode::StorageOther(format!(
@@ -81,7 +96,7 @@ pub async fn load_delete_mark(dal: Operator, path: &str, length: u64) -> Result<
     let index_column_chunk_metas = file_meta.row_groups[0].columns();
     assert_eq!(index_column_chunk_metas.len(), 1);
     let column_meta = &index_column_chunk_metas[0];
-    let marks = load_delete_mark_data(column_meta, path, &dal).await?;
+    let marks = load_delete_mark_data(column_meta, path, &dal, num_rows).await?;
 
     Ok(marks)
 }
@@ -92,12 +107,14 @@ async fn load_delete_mark_data<'a>(
     col_chunk_meta: &'a ColumnChunkMetaData,
     path: &'a str,
     dal: &'a Operator,
+    num_rows: usize,
 ) -> Result<Arc<Bitmap>> {
     let storage_runtime = GlobalIORuntime::instance();
     let bytes = {
         let meta = col_chunk_meta.metadata();
         let location = path.to_string();
         let loader = DeleteMetaLoader {
+            num_rows,
             offset: meta.data_page_offset as u64,
             len: meta.total_compressed_size as u64,
             cache_key: location.clone(),
@@ -122,7 +139,7 @@ async fn load_delete_mark_data<'a>(
 /// Loads index meta data
 /// read data from cache, or populate cache items if possible
 #[tracing::instrument(level = "debug", skip_all)]
-async fn load_mark_meta(dal: Operator, path: &str, length: u64) -> Result<Arc<DeleteMarkMeta>> {
+async fn load_mark_meta(dal: Operator, path: &str, size: u64) -> Result<Arc<DeleteMarkMeta>> {
     let path_owned = path.to_owned();
     async move {
         let reader = MetaReaders::delete_mark_meta_reader(dal);
@@ -132,7 +149,7 @@ async fn load_mark_meta(dal: Operator, path: &str, length: u64) -> Result<Arc<De
 
         let load_params = LoadParams {
             location: path_owned,
-            len_hint: Some(length),
+            len_hint: Some(size),
             ver: version,
         };
 
@@ -143,6 +160,7 @@ async fn load_mark_meta(dal: Operator, path: &str, length: u64) -> Result<Arc<De
 }
 
 pub struct DeleteMetaLoader {
+    pub num_rows: usize,
     pub offset: u64,
     pub len: u64,
     pub cache_key: String,
@@ -177,8 +195,13 @@ impl Loader<Bitmap> for DeleteMetaLoader {
         let column_type = self.column_descriptor.descriptor.primitive_type.clone();
         let filed_name = self.column_descriptor.path_in_schema[0].to_owned();
         let field = ArrowField::new(filed_name, DataType::Boolean, false);
-        let mut array_iter =
-            column_iter_to_arrays(vec![decompressor], vec![&column_type], field, None, 1)?;
+        let mut array_iter = column_iter_to_arrays(
+            vec![decompressor],
+            vec![&column_type],
+            field,
+            None,
+            self.num_rows,
+        )?;
         if let Some(array) = array_iter.next() {
             let array = array?;
             let col =
diff --git a/src/query/storages/fuse/src/io/read/delete/mod.rs b/src/query/storages/fuse/src/io/read/delete/mod.rs
@@ -13,3 +13,5 @@
 // limitations under the License.
 
 mod delete_mark_reader;
+
+pub use delete_mark_reader::DeleteMarkReader;
diff --git a/src/query/storages/fuse/src/io/read/mod.rs b/src/query/storages/fuse/src/io/read/mod.rs
@@ -25,6 +25,7 @@ pub use block::NativeReaderExt;
 pub use block::UncompressedBuffer;
 pub use bloom::BloomBlockFilterReader;
 pub use bloom::InRuntime;
+pub use delete::DeleteMarkReader;
 pub use meta::MetaReaders;
 pub use meta::SegmentInfoReader;
 pub use meta::TableSnapshotReader;
diff --git a/src/query/storages/fuse/src/operations/delete.rs b/src/query/storages/fuse/src/operations/delete.rs
@@ -35,6 +35,8 @@ use common_expression::Evaluator;
 use common_expression::Expr;
 use common_expression::FieldIndex;
 use common_expression::RemoteExpr;
+use common_expression::TableDataType;
+use common_expression::TableField;
 use common_expression::TableSchema;
 use common_expression::Value;
 use common_functions::scalars::BUILTIN_FUNCTIONS;
@@ -44,6 +46,8 @@ use common_sql::evaluator::BlockOperator;
 use storages_common_table_meta::meta::Location;
 use storages_common_table_meta::meta::TableSnapshot;
 
+use super::mutation::MutationSerializeTransform;
+use super::mutation::ParquetDeleteSource;
 use crate::operations::mutation::MutationAction;
 use crate::operations::mutation::MutationPartInfo;
 use crate::operations::mutation::MutationSink;
@@ -148,6 +152,107 @@ impl FuseTable {
         Ok(())
     }
 
+    pub async fn do_delete2(
+        &self,
+        ctx: Arc<dyn TableContext>,
+        filter: Option<RemoteExpr<String>>,
+        col_indices: Vec<usize>,
+        pipeline: &mut Pipeline,
+    ) -> Result<()> {
+        let snapshot_opt = self.read_table_snapshot().await?;
+
+        // check if table is empty
+        let snapshot = if let Some(val) = snapshot_opt {
+            val
+        } else {
+            // no snapshot, no deletion
+            return Ok(());
+        };
+
+        if snapshot.summary.row_count == 0 {
+            // empty snapshot, no deletion
+            return Ok(());
+        }
+
+        let scan_progress = ctx.get_scan_progress();
+        // check if unconditional deletion
+        if filter.is_none() {
+            let progress_values = ProgressValues {
+                rows: snapshot.summary.row_count as usize,
+                bytes: snapshot.summary.uncompressed_byte_size as usize,
+            };
+            scan_progress.incr(&progress_values);
+            // deleting the whole table... just a truncate
+            let purge = false;
+            return self.do_truncate(ctx.clone(), purge).await;
+        }
+
+        let filter_expr = filter.unwrap();
+        if col_indices.is_empty() {
+            // here the situation: filter_expr is not null, but col_indices in empty, which
+            // indicates the expr being evaluated is unrelated to the value of rows:
+            //   e.g.
+            //       `delete from t where 1 = 1`, `delete from t where now()`,
+            //       or `delete from t where RANDOM()::INT::BOOLEAN`
+            // if the `filter_expr` is of "constant" nullary :
+            //   for the whole block, whether all of the rows should be kept or dropped,
+            //   we can just return from here, without accessing the block data
+            if self.try_eval_const(ctx.clone(), &self.schema(), &filter_expr)? {
+                let progress_values = ProgressValues {
+                    rows: snapshot.summary.row_count as usize,
+                    bytes: snapshot.summary.uncompressed_byte_size as usize,
+                };
+                scan_progress.incr(&progress_values);
+
+                // deleting the whole table... just a truncate
+                let purge = false;
+                return self.do_truncate(ctx.clone(), purge).await;
+            }
+            // do nothing.
+            return Ok(());
+        }
+
+        let projection = Projection::Columns(col_indices.clone());
+        self.mutation_block_pruning(
+            ctx.clone(),
+            vec![filter_expr.clone()],
+            projection.clone(),
+            &snapshot,
+        )
+        .await?;
+        let block_reader = self.create_block_reader(projection, ctx.clone())?;
+
+        let mut schema = block_reader.schema().as_ref().to_owned();
+        schema.add_columns(&[TableField::new("_row_exists", TableDataType::Boolean)])?;
+        let filter = Arc::new(
+            filter_expr
+                .as_expr(&BUILTIN_FUNCTIONS)
+                .project_column_ref(|name| schema.index_of(name).unwrap()),
+        );
+
+        let max_threads = ctx.get_settings().get_max_threads()? as usize;
+        // Add source pipe.
+        pipeline.add_source(
+            |output| {
+                ParquetDeleteSource::try_create(
+                    ctx.clone(),
+                    output,
+                    self,
+                    filter.clone(),
+                    block_reader.clone(),
+                )
+            },
+            max_threads,
+        )?;
+
+        self.try_add_mutation_transform2(ctx.clone(), snapshot.segments.clone(), pipeline)?;
+
+        pipeline.add_sink(|input| {
+            MutationSink::try_create(self, ctx.clone(), snapshot.clone(), input)
+        })?;
+        Ok(())
+    }
+
     pub fn try_eval_const(
         &self,
         ctx: Arc<dyn TableContext>,
@@ -335,6 +440,44 @@ impl FuseTable {
         }
     }
 
+    pub fn try_add_mutation_transform2(
+        &self,
+        ctx: Arc<dyn TableContext>,
+        base_segments: Vec<Location>,
+        pipeline: &mut Pipeline,
+    ) -> Result<()> {
+        if pipeline.is_empty() {
+            return Err(ErrorCode::Internal("The pipeline is empty."));
+        }
+
+        match pipeline.output_len() {
+            0 => Err(ErrorCode::Internal("The output of the last pipe is 0.")),
+            last_pipe_size => {
+                let mut inputs_port = Vec::with_capacity(last_pipe_size);
+                for _ in 0..last_pipe_size {
+                    inputs_port.push(InputPort::create());
+                }
+                let output_port = OutputPort::create();
+                pipeline.add_pipe(Pipe::create(inputs_port.len(), 1, vec![PipeItem::create(
+                    MutationSerializeTransform::try_create(
+                        ctx,
+                        self.schema(),
+                        inputs_port.clone(),
+                        output_port.clone(),
+                        self.get_operator(),
+                        self.meta_location_generator().clone(),
+                        base_segments,
+                        self.get_block_compact_thresholds(),
+                    )?,
+                    inputs_port,
+                    vec![output_port],
+                )]));
+
+                Ok(())
+            }
+        }
+    }
+
     pub fn cluster_stats_gen(&self, ctx: Arc<dyn TableContext>) -> Result<ClusterStatsGenerator> {
         if self.cluster_key_meta.is_none() {
             return Ok(ClusterStatsGenerator::default());
diff --git a/src/query/storages/fuse/src/operations/mutation/mod.rs b/src/query/storages/fuse/src/operations/mutation/mod.rs
@@ -17,11 +17,11 @@ pub mod base_mutator;
 mod compact;
 pub mod mutation_meta;
 mod mutation_part;
+mod mutation_serialize_transform;
 pub mod mutation_sink;
 mod mutation_source;
 mod mutation_transform;
-#[allow(unused)]
-mod parquet_delete_transform;
+pub mod parquet_delete_source;
 pub mod recluster_mutator;
 mod serialize_data_transform;
 
@@ -39,10 +39,13 @@ pub use mutation_meta::MutationSinkMeta;
 pub use mutation_meta::MutationTransformMeta;
 pub use mutation_meta::SerializeDataMeta;
 pub use mutation_part::MutationPartInfo;
+pub use mutation_serialize_transform::MutationSerializeTransform;
 pub use mutation_sink::MutationSink;
 pub use mutation_source::MutationAction;
 pub use mutation_source::MutationSource;
 pub use mutation_transform::MutationTransform;
+pub use parquet_delete_source::MutationSourceMeta;
+pub use parquet_delete_source::ParquetDeleteSource;
 pub use recluster_mutator::ReclusterMutator;
 pub use serialize_data_transform::SerializeDataTransform;
 pub use serialize_data_transform::SerializeState;
diff --git a/src/query/storages/fuse/src/operations/mutation/mutation_serialize_transform.rs b/src/query/storages/fuse/src/operations/mutation/mutation_serialize_transform.rs
diff --git a/src/query/storages/fuse/src/operations/mutation/parquet_delete_source.rs b/src/query/storages/fuse/src/operations/mutation/parquet_delete_source.rs
diff --git a/src/query/storages/fuse/src/operations/mutation/parquet_delete_transform.rs b/src/query/storages/fuse/src/operations/mutation/parquet_delete_transform.rs
diff --git a/src/query/storages/fuse/src/operations/read_partitions.rs b/src/query/storages/fuse/src/operations/read_partitions.rs
