diff --git a/src/common/storage/src/metrics/merge_into.rs b/src/common/storage/src/metrics/merge_into.rs index cd1f6a690bf1d..c36f452fbef5c 100644 --- a/src/common/storage/src/metrics/merge_into.rs +++ b/src/common/storage/src/metrics/merge_into.rs @@ -27,10 +27,16 @@ macro_rules! key { lazy_static! { static ref MERGE_INTO_REPLACE_BLOCKS_COUNTER: Counter = register_counter(key!("merge_into_replace_blocks_counter")); + static ref MERGE_INTO_REPLACE_BLOCKS_ROWS_COUNTER: Counter = + register_counter(key!("merge_into_replace_blocks_rows_counter")); static ref MERGE_INTO_DELETED_BLOCKS_COUNTER: Counter = register_counter(key!("merge_into_deleted_blocks_counter")); + static ref MERGE_INTO_DELETED_BLOCKS_ROWS_COUNTER: Counter = + register_counter(key!("merge_into_deleted_blocks_rows_counter")); static ref MERGE_INTO_APPEND_BLOCKS_COUNTER: Counter = register_counter(key!("merge_into_append_blocks_counter")); + static ref MERGE_INTO_APPEND_BLOCKS_ROWS_COUNTER: Counter = + register_counter(key!("merge_into_append_blocks_rows_counter")); static ref MERGE_INTO_MATCHED_ROWS: Counter = register_counter(key!("merge_into_matched_rows")); static ref MERGE_INTO_UNMATCHED_ROWS: Counter = register_counter(key!("merge_into_unmatched_rows")); @@ -50,15 +56,27 @@ pub fn metrics_inc_merge_into_replace_blocks_counter(c: u32) { MERGE_INTO_REPLACE_BLOCKS_COUNTER.inc_by(c as u64); } +pub fn metrics_inc_merge_into_replace_blocks_rows_counter(c: u32) { + MERGE_INTO_REPLACE_BLOCKS_ROWS_COUNTER.inc_by(c as u64); +} + pub fn metrics_inc_merge_into_deleted_blocks_counter(c: u32) { MERGE_INTO_DELETED_BLOCKS_COUNTER.inc_by(c as u64); } +pub fn metrics_inc_merge_into_deleted_blocks_rows_counter(c: u32) { + MERGE_INTO_DELETED_BLOCKS_ROWS_COUNTER.inc_by(c as u64); +} + // used to record append new blocks in matched split and not match insert. pub fn metrics_inc_merge_into_append_blocks_counter(c: u32) { MERGE_INTO_APPEND_BLOCKS_COUNTER.inc_by(c as u64); } +pub fn metrics_inc_merge_into_append_blocks_rows_counter(c: u32) { + MERGE_INTO_APPEND_BLOCKS_ROWS_COUNTER.inc_by(c as u64); +} + // matched_rows and not unmatched_rows is used in the join phase of merge_source. pub fn metrics_inc_merge_into_matched_rows(c: u32) { MERGE_INTO_MATCHED_ROWS.inc_by(c as u64); diff --git a/src/query/storages/fuse/src/lib.rs b/src/query/storages/fuse/src/lib.rs index 14b86d0dacd7e..1713771bc0855 100644 --- a/src/query/storages/fuse/src/lib.rs +++ b/src/query/storages/fuse/src/lib.rs @@ -20,6 +20,7 @@ #![feature(impl_trait_in_assoc_type)] #![feature(int_roundings)] #![feature(result_option_inspect)] +#![recursion_limit = "256"] mod constants; mod fuse_lazy_part; diff --git a/src/query/storages/fuse/src/metrics/fuse_metrics.rs b/src/query/storages/fuse/src/metrics/fuse_metrics.rs index 2ec2ab9e0d15f..2ff3c95112319 100644 --- a/src/query/storages/fuse/src/metrics/fuse_metrics.rs +++ b/src/query/storages/fuse/src/metrics/fuse_metrics.rs @@ -134,6 +134,12 @@ lazy_static! { register_counter(key!("replace_into_block_number_bloom_pruned")); static ref REPLACE_INTO_BLOCK_NUMBER_SOURCE: Counter = register_counter(key!("replace_into_block_number_source")); + static ref REPLACE_INTO_REPLACED_BLOCKS_ROWS: Counter = + register_counter(key!("replace_into_replaced_blocks_rows")); + static ref REPLACE_INTO_DELETED_BLOCKS_ROWS: Counter = + register_counter(key!("replace_into_deleted_blocks_rows")); + static ref REPLACE_INTO_APPEND_BLOCKS_ROWS: Counter = + register_counter(key!("replace_into_append_blocks_rows")); } pub fn metrics_inc_commit_mutation_unresolvable_conflict() { @@ -392,3 +398,18 @@ pub fn metrics_inc_replace_block_number_bloom_pruned(c: u64) { pub fn metrics_inc_replace_block_number_input(c: u64) { REPLACE_INTO_BLOCK_NUMBER_SOURCE.inc_by(c); } + +// rows of blocks that are replaced +pub fn metrics_inc_replace_replaced_blocks_rows(c: u64) { + REPLACE_INTO_REPLACED_BLOCKS_ROWS.inc_by(c); +} + +// rows of blocks that are deleted +pub fn metrics_inc_replace_deleted_blocks_rows(c: u64) { + REPLACE_INTO_DELETED_BLOCKS_ROWS.inc_by(c); +} + +// rows of blocks that are appended +pub fn metrics_inc_replace_append_blocks_rows(c: u64) { + REPLACE_INTO_APPEND_BLOCKS_ROWS.inc_by(c); +} diff --git a/src/query/storages/fuse/src/operations/merge_into/mutator/matched_mutator.rs b/src/query/storages/fuse/src/operations/merge_into/mutator/matched_mutator.rs index b5477b16b70fb..05704c975bff1 100644 --- a/src/query/storages/fuse/src/operations/merge_into/mutator/matched_mutator.rs +++ b/src/query/storages/fuse/src/operations/merge_into/mutator/matched_mutator.rs @@ -39,7 +39,9 @@ use common_expression::TableSchemaRef; use common_storage::metrics::merge_into::metrics_inc_merge_into_accumulate_milliseconds; use common_storage::metrics::merge_into::metrics_inc_merge_into_apply_milliseconds; use common_storage::metrics::merge_into::metrics_inc_merge_into_deleted_blocks_counter; +use common_storage::metrics::merge_into::metrics_inc_merge_into_deleted_blocks_rows_counter; use common_storage::metrics::merge_into::metrics_inc_merge_into_replace_blocks_counter; +use common_storage::metrics::merge_into::metrics_inc_merge_into_replace_blocks_rows_counter; use itertools::Itertools; use log::info; use opendal::Operator; @@ -290,10 +292,10 @@ impl AggregationContext { &self.read_settings, ) .await?; - + let origin_num_rows = origin_data_block.num_rows(); // apply delete let mut bitmap = MutableBitmap::new(); - for row in 0..origin_data_block.num_rows() { + for row in 0..origin_num_rows { if modified_offsets.contains(&row) { bitmap.push(false); } else { @@ -304,6 +306,7 @@ impl AggregationContext { if res_block.is_empty() { metrics_inc_merge_into_deleted_blocks_counter(1); + metrics_inc_merge_into_deleted_blocks_rows_counter(origin_num_rows as u32); return Ok(Some(MutationLogEntry::DeletedBlock { index: BlockMetaIndex { segment_idx, @@ -338,6 +341,7 @@ impl AggregationContext { write_data(new_block_raw_data, &data_accessor, &new_block_location).await?; metrics_inc_merge_into_replace_blocks_counter(1); + metrics_inc_merge_into_replace_blocks_rows_counter(origin_num_rows as u32); // generate log let mutation = MutationLogEntry::ReplacedBlock { index: BlockMetaIndex { diff --git a/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_matched_and_split.rs b/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_matched_and_split.rs index 658b8f1b54557..1bbdc2b8f1a71 100644 --- a/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_matched_and_split.rs +++ b/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_matched_and_split.rs @@ -37,6 +37,7 @@ use common_sql::evaluator::BlockOperator; use common_sql::executor::MatchExpr; use common_storage::metrics::merge_into::merge_into_matched_operation_milliseconds; use common_storage::metrics::merge_into::metrics_inc_merge_into_append_blocks_counter; +use common_storage::metrics::merge_into::metrics_inc_merge_into_append_blocks_rows_counter; use crate::operations::merge_into::mutator::DeleteByExprMutator; use crate::operations::merge_into::mutator::UpdateByExprMutator; @@ -278,6 +279,7 @@ impl Processor for MatchedSplitProcessor { }; current_block = op.execute(&self.ctx.get_function_context()?, current_block)?; metrics_inc_merge_into_append_blocks_counter(1); + metrics_inc_merge_into_append_blocks_rows_counter(current_block.num_rows() as u32); current_block = current_block.add_meta(Some(Box::new(self.target_table_schema.clone())))?; self.output_data_updated_data = Some(current_block); diff --git a/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_not_matched.rs b/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_not_matched.rs index f75d92ef1d360..5816eaeda10d4 100644 --- a/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_not_matched.rs +++ b/src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_not_matched.rs @@ -33,6 +33,7 @@ use common_pipeline_core::processors::Processor; use common_sql::evaluator::BlockOperator; use common_storage::metrics::merge_into::merge_into_not_matched_operation_milliseconds; use common_storage::metrics::merge_into::metrics_inc_merge_into_append_blocks_counter; +use common_storage::metrics::merge_into::metrics_inc_merge_into_append_blocks_rows_counter; use itertools::Itertools; use crate::operations::merge_into::mutator::SplitByExprMutator; @@ -161,6 +162,9 @@ impl Processor for MergeIntoNotMatchedProcessor { .add_meta(Some(Box::new(self.data_schemas.get(&idx).unwrap().clone())))?; if !satisfied_block.is_empty() { metrics_inc_merge_into_append_blocks_counter(1); + metrics_inc_merge_into_append_blocks_rows_counter( + satisfied_block.num_rows() as u32 + ); self.output_data .push(op.op.execute(&self.func_ctx, satisfied_block)?) } diff --git a/src/query/storages/fuse/src/operations/replace_into/mutator/merge_into_mutator.rs b/src/query/storages/fuse/src/operations/replace_into/mutator/merge_into_mutator.rs index 9fa8d0f691af5..66752b93d3b76 100644 --- a/src/query/storages/fuse/src/operations/replace_into/mutator/merge_into_mutator.rs +++ b/src/query/storages/fuse/src/operations/replace_into/mutator/merge_into_mutator.rs @@ -62,8 +62,10 @@ use crate::metrics::metrics_inc_replace_block_number_bloom_pruned; use crate::metrics::metrics_inc_replace_block_number_totally_loaded; use crate::metrics::metrics_inc_replace_block_number_write; use crate::metrics::metrics_inc_replace_block_of_zero_row_deleted; +use crate::metrics::metrics_inc_replace_deleted_blocks_rows; use crate::metrics::metrics_inc_replace_number_accumulated_merge_action; use crate::metrics::metrics_inc_replace_number_apply_deletion; +use crate::metrics::metrics_inc_replace_replaced_blocks_rows; use crate::metrics::metrics_inc_replace_row_number_after_pruning; use crate::metrics::metrics_inc_replace_row_number_totally_loaded; use crate::metrics::metrics_inc_replace_row_number_write; @@ -463,6 +465,7 @@ impl AggregationContext { if delete_nums == block_meta.row_count as usize { info!("whole block deletion"); metrics_inc_replace_whole_block_deletion(1); + metrics_inc_replace_deleted_blocks_rows(num_rows as u64); // whole block deletion // NOTE that if deletion marker is enabled, check the real meaning of `row_count` let mutation = MutationLogEntry::DeletedBlock { @@ -535,6 +538,7 @@ impl AggregationContext { metrics_inc_replace_block_number_write(1); metrics_inc_replace_row_number_write(new_block_meta.row_count); + metrics_inc_replace_replaced_blocks_rows(num_rows as u64); if let Some(index_state) = serialized.bloom_index_state { write_data(index_state.data, &data_accessor, &index_state.location.0).await?; } diff --git a/src/query/storages/fuse/src/operations/replace_into/processors/processor_replace_into.rs b/src/query/storages/fuse/src/operations/replace_into/processors/processor_replace_into.rs index ed98b09342be1..3775356fbc86f 100644 --- a/src/query/storages/fuse/src/operations/replace_into/processors/processor_replace_into.rs +++ b/src/query/storages/fuse/src/operations/replace_into/processors/processor_replace_into.rs @@ -34,6 +34,7 @@ use common_pipeline_core::processors::Processor; use common_sql::executor::OnConflictField; use storages_common_table_meta::meta::ColumnStatistics; +use crate::metrics::metrics_inc_replace_append_blocks_rows; use crate::metrics::metrics_inc_replace_block_number_input; use crate::metrics::metrics_inc_replace_process_input_block_time_ms; use crate::operations::replace_into::mutator::mutator_replace_into::ReplaceIntoMutator; @@ -173,6 +174,7 @@ impl Processor for ReplaceIntoProcessor { self.output_data_merge_into_action = Some(DataBlock::empty_with_meta(Box::new(merge_into_action))); } + metrics_inc_replace_append_blocks_rows(data_block.num_rows() as u64); self.output_data_append = Some(data_block); return Ok(()); }