add block statistics

Author: zhyass

src/query/service/src/interpreters/common/table_option_validation.rs

@@ -25,6 +25,7 @@ use databend_common_expression::TableSchemaRef;
 use databend_common_io::constants::DEFAULT_BLOCK_ROW_COUNT;
 use databend_common_settings::Settings;
 use databend_common_sql::BloomIndexColumns;
+use databend_common_sql::DistinctColumns;
 use databend_common_storages_fuse::FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
 use databend_common_storages_fuse::FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP;
@@ -35,12 +36,14 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_BLOCK;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_PAGE;
 use databend_storages_common_index::BloomIndex;
+use databend_storages_common_index::RangeIndex;
 use databend_storages_common_table_meta::table::OPT_KEY_BLOOM_INDEX_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_CHANGE_TRACKING;
 use databend_storages_common_table_meta::table::OPT_KEY_CLUSTER_TYPE;
 use databend_storages_common_table_meta::table::OPT_KEY_COMMENT;
 use databend_storages_common_table_meta::table::OPT_KEY_CONNECTION_NAME;
 use databend_storages_common_table_meta::table::OPT_KEY_DATABASE_ID;
+use databend_storages_common_table_meta::table::OPT_KEY_DISTINCT_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH;
 use databend_storages_common_table_meta::table::OPT_KEY_ENGINE;
 use databend_storages_common_table_meta::table::OPT_KEY_LOCATION;
@@ -68,6 +71,7 @@ pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(
     r.insert(FUSE_OPT_KEY_ENABLE_AUTO_VACUUM);
 
     r.insert(OPT_KEY_BLOOM_INDEX_COLUMNS);
+    r.insert(OPT_KEY_DISTINCT_COLUMNS);
     r.insert(OPT_KEY_TABLE_COMPRESSION);
     r.insert(OPT_KEY_STORAGE_FORMAT);
     r.insert(OPT_KEY_DATABASE_ID);
@@ -213,6 +217,16 @@ pub fn is_valid_bloom_index_columns(
     Ok(())
 }
 
+pub fn is_valid_distinct_columns(
+    options: &BTreeMap<String, String>,
+    schema: TableSchemaRef,
+) -> databend_common_exception::Result<()> {
+    if let Some(value) = options.get(OPT_KEY_DISTINCT_COLUMNS) {
+        DistinctColumns::verify_definition(value, schema, RangeIndex::supported_table_type)?;
+    }
+    Ok(())
+}
+
 pub fn is_valid_change_tracking(
     options: &BTreeMap<String, String>,
 ) -> databend_common_exception::Result<()> {

src/query/service/src/interpreters/interpreter_table_create.rs

@@ -71,6 +71,7 @@ use crate::interpreters::common::table_option_validation::is_valid_bloom_index_c
 use crate::interpreters::common::table_option_validation::is_valid_change_tracking;
 use crate::interpreters::common::table_option_validation::is_valid_create_opt;
 use crate::interpreters::common::table_option_validation::is_valid_data_retention_period;
+use crate::interpreters::common::table_option_validation::is_valid_distinct_columns;
 use crate::interpreters::common::table_option_validation::is_valid_option_of_type;
 use crate::interpreters::common::table_option_validation::is_valid_random_seed;
 use crate::interpreters::common::table_option_validation::is_valid_row_per_block;
@@ -467,7 +468,8 @@ impl CreateTableInterpreter {
         is_valid_block_per_segment(&table_meta.options)?;
         is_valid_row_per_block(&table_meta.options)?;
         // check bloom_index_columns.
-        is_valid_bloom_index_columns(&table_meta.options, schema)?;
+        is_valid_bloom_index_columns(&table_meta.options, schema.clone())?;
+        is_valid_distinct_columns(&table_meta.options, schema)?;
         is_valid_change_tracking(&table_meta.options)?;
         // check random seed
         is_valid_random_seed(&table_meta.options)?;

src/query/service/src/interpreters/interpreter_table_set_options.rs

@@ -53,6 +53,7 @@ use crate::interpreters::common::table_option_validation::is_valid_block_per_seg
 use crate::interpreters::common::table_option_validation::is_valid_bloom_index_columns;
 use crate::interpreters::common::table_option_validation::is_valid_create_opt;
 use crate::interpreters::common::table_option_validation::is_valid_data_retention_period;
+use crate::interpreters::common::table_option_validation::is_valid_distinct_columns;
 use crate::interpreters::common::table_option_validation::is_valid_option_of_type;
 use crate::interpreters::common::table_option_validation::is_valid_row_per_block;
 use crate::interpreters::Interpreter;
@@ -163,6 +164,7 @@ impl Interpreter for SetOptionsInterpreter {
 
         // check bloom_index_columns.
         is_valid_bloom_index_columns(&self.plan.set_options, table.schema())?;
+        is_valid_distinct_columns(&self.plan.set_options, table.schema())?;
 
         if let Some(new_snapshot_location) =
             set_segment_format(self.ctx.clone(), table.clone(), &self.plan.set_options).await?

src/query/service/src/test_kits/block_writer.rs

@@ -112,6 +112,7 @@ impl<'a> BlockWriter<'a> {
             None,
             None,
             None,
+            None,
             Compression::Lz4Raw,
             Some(Utc::now()),
         );

src/query/service/tests/it/storages/fuse/bloom_index_meta_size.rs

@@ -340,6 +340,7 @@ fn build_test_segment_info(
         vector_index_size: None,
         vector_index_location: None,
         virtual_block_meta: None,
+        block_stats_meta: None,
         compression: Compression::Lz4,
         create_on: Some(Utc::now()),
     };

src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs

@@ -81,6 +81,7 @@ async fn test_recluster_mutator_block_select() -> Result<()> {
             None,
             None,
             None,
+            None,
             meta::Compression::Lz4Raw,
             Some(Utc::now()),
         ));

src/query/service/tests/it/storages/fuse/operations/mutation/segments_compact_mutator.rs

@@ -781,6 +781,7 @@ impl CompactSegmentTestFixture {
                         None,
                         None,
                         None,
+            None,
                         Compression::Lz4Raw,
                         Some(Utc::now()),
                     );

src/query/service/tests/it/storages/fuse/operations/read_plan.rs

@@ -107,6 +107,7 @@ fn test_to_partitions() -> Result<()> {
         None,
         None,
         None,
+        None,
         meta::Compression::Lz4Raw,
         Some(Utc::now()),
     ));

src/query/service/tests/it/storages/fuse/statistics.rs

@@ -637,6 +637,7 @@ fn test_reduce_block_meta() -> databend_common_exception::Result<()> {
             None,
             None,
             None,
+            None,
             Compression::Lz4Raw,
             Some(Utc::now()),
         );

src/query/settings/src/settings_default.rs

@@ -1319,6 +1319,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: None,
                 }),
+                ("enable_block_stream_write", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(1),
+                    desc: "Enables block stream write",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
                 ("trace_sample_rate", DefaultSettingValue {
                     value: UserSettingValue::UInt64(1),
                     desc: "Setting the trace sample rate. The value should be between '0' and '100'",