linear recluster support block stream writer

Author: zhyass

src/query/functions/src/scalars/hilbert.rs

@@ -267,7 +267,7 @@ fn calc_range_partition_id(val: ScalarRef, arr: &Column) -> u64 {
     while low < high {
         let mid = low + ((high - low) / 2);
         let bound = unsafe { arr.index_unchecked(mid) };
-        if val > bound {
+        if val >= bound {
             low = mid + 1;
         } else {
             high = mid;

src/query/service/src/local/display.rs

@@ -259,7 +259,7 @@ impl FormatDisplay<'_> {
                 rows_str,
                 self.start.elapsed().as_secs_f64(),
                 humanize_count(stats.total_rows as f64),
-                HumanBytes(stats.total_rows as u64),
+                HumanBytes(stats.total_bytes as u64),
                 humanize_count(stats.total_rows as f64 / self.start.elapsed().as_secs_f64()),
                 HumanBytes((stats.total_bytes as f64 / self.start.elapsed().as_secs_f64()) as u64),
             );

src/query/service/src/pipelines/builders/builder_hilbert_partition.rs

@@ -71,6 +71,8 @@ impl PipelineBuilder {
             let properties = StreamBlockProperties::try_create(
                 self.ctx.clone(),
                 table,
+                MutationKind::Recluster,
+                None,
                 partition.table_meta_timestamps,
             )?;
 

src/query/service/src/pipelines/builders/builder_recluster.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod builder;
 mod range_bound_sampler;
 mod recluster_partition_exchange;
 mod recluster_partition_strategy;
@@ -25,7 +24,8 @@ pub use range_bound_sampler::RangeBoundSampler;
 pub use recluster_partition_exchange::ReclusterPartitionExchange;
 pub use recluster_partition_strategy::CompactPartitionStrategy;
 pub use recluster_partition_strategy::ReclusterPartitionStrategy;
-pub(crate) use recluster_sample_state::SampleState;
+pub use recluster_sample_state::SampleState;
+pub use transform_add_order_column::TransformAddOrderColumn;
 pub use transform_range_partition_indexer::TransformRangePartitionIndexer;
-pub(crate) use transform_recluster_collect::ReclusterSampleMeta;
+pub use transform_recluster_collect::ReclusterSampleMeta;
 pub use transform_recluster_collect::TransformReclusterCollect;

src/query/service/src/pipelines/processors/transforms/recluster/builder.rs

@@ -137,7 +137,7 @@ where
                 while low < high {
                     let mid = low + ((high - low) / 2);
                     let bound = unsafe { self.bounds.get_unchecked(mid) }.clone();
-                    if val > bound {
+                    if val >= bound {
                         low = mid + 1;
                     } else {
                         high = mid;

src/query/service/src/pipelines/processors/transforms/recluster/mod.rs

@@ -15,7 +15,6 @@
 use databend_common_exception::Result;
 use databend_common_expression::local_block_meta_serde;
 use databend_common_expression::types::ArgType;
-use databend_common_expression::types::ValueType;
 use databend_common_expression::BlockMetaInfo;
 use databend_common_expression::DataBlock;
 use databend_common_expression::Scalar;

src/query/service/src/pipelines/processors/transforms/recluster/transform_range_partition_indexer.rs

@@ -860,7 +860,7 @@ impl DefaultSettings {
                     desc: "Sets the maximum byte size of blocks for recluster",
                     mode: SettingMode::Both,
                     scope: SettingScope::Both,
-                    range: Some(SettingRange::Numeric(0..=u64::MAX)),
+                    range: Some(SettingRange::Numeric(0..=80 * 1024 * 1024 * 1024)),
                 }),
                 ("compact_max_block_selection", DefaultSettingValue {
                     value: UserSettingValue::UInt64(10000),

src/query/service/src/pipelines/processors/transforms/recluster/transform_recluster_collect.rs

@@ -28,13 +28,16 @@ use databend_common_expression::Column;
 use databend_common_expression::ColumnId;
 use databend_common_expression::ComputedExpr;
 use databend_common_expression::DataBlock;
+use databend_common_expression::DataField;
 use databend_common_expression::FieldIndex;
 use databend_common_expression::TableField;
 use databend_common_expression::TableSchema;
 use databend_common_expression::TableSchemaRef;
 use databend_common_expression::ORIGIN_BLOCK_ROW_NUM_COLUMN_ID;
 use databend_common_io::constants::DEFAULT_BLOCK_BUFFER_SIZE;
 use databend_common_native::write::NativeWriter;
+use databend_common_sql::evaluator::BlockOperator;
+use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_storages_common_index::BloomIndex;
 use databend_storages_common_index::BloomIndexBuilder;
 use databend_storages_common_index::Index;
@@ -367,17 +370,24 @@ impl StreamBlockProperties {
     pub fn try_create(
         ctx: Arc<dyn TableContext>,
         table: &FuseTable,
+        kind: MutationKind,
+        level: Option<i32>,
         table_meta_timestamps: TableMetaTimestamps,
     ) -> Result<Arc<Self>> {
         // remove virtual computed fields.
-        let fields = table
+        let mut fields = table
             .schema()
             .fields()
             .iter()
             .filter(|f| !matches!(f.computed_expr(), Some(ComputedExpr::Virtual(_))))
             .cloned()
             .collect::<Vec<_>>();
-
+        if !matches!(kind, MutationKind::Insert | MutationKind::Replace) {
+            // add stream fields.
+            for stream_column in table.stream_columns().iter() {
+                fields.push(stream_column.table_field());
+            }
+        }
         let source_schema = Arc::new(TableSchema {
             fields,
             ..table.schema().as_ref().clone()
@@ -397,7 +407,7 @@ impl StreamBlockProperties {
         let inverted_index_builders = create_inverted_index_builders(&table.table_info.meta);
 
         let cluster_stats_builder =
-            ClusterStatisticsBuilder::try_create(table, ctx.clone(), &source_schema)?;
+            ClusterStatisticsBuilder::try_create(table, ctx.clone(), &source_schema, level)?;
 
         let mut stats_columns = vec![];
         let mut distinct_columns = vec![];
@@ -434,4 +444,16 @@ impl StreamBlockProperties {
         self.block_thresholds
             .check_large_enough(num_rows, data_size)
     }
+
+    pub fn cluster_operators(&self) -> Vec<BlockOperator> {
+        self.cluster_stats_builder.operators()
+    }
+
+    pub fn fields_with_cluster_key(&self) -> Vec<DataField> {
+        self.cluster_stats_builder.out_fields()
+    }
+
+    pub fn cluster_key_index(&self) -> &Vec<usize> {
+        self.cluster_stats_builder.cluster_key_index()
+    }
 }