chore: Using an ordered BTreeSet instead of a HashSet as a ColumnSet implementation (#17747)

* update

* fix

Author: forsaken628

src/query/expression/src/block.rs

@@ -26,6 +26,7 @@ use crate::types::AnyType;
 use crate::types::DataType;
 use crate::Column;
 use crate::ColumnBuilder;
+use crate::ColumnSet;
 use crate::DataField;
 use crate::DataSchemaRef;
 use crate::Domain;
@@ -562,7 +563,7 @@ impl DataBlock {
     }
 
     #[inline]
-    pub fn project(mut self, projections: &HashSet<usize>) -> Self {
+    pub fn project(mut self, projections: &ColumnSet) -> Self {
         let mut columns = Vec::with_capacity(projections.len());
         for (index, column) in self.columns.into_iter().enumerate() {
             if !projections.contains(&index) {
@@ -575,7 +576,7 @@ impl DataBlock {
     }
 
     #[inline]
-    pub fn project_with_agg_index(self, projections: &HashSet<usize>, num_evals: usize) -> Self {
+    pub fn project_with_agg_index(self, projections: &ColumnSet, num_evals: usize) -> Self {
         let mut columns = Vec::with_capacity(projections.len());
         let eval_offset = self.columns.len() - num_evals;
         for (index, column) in self.columns.into_iter().enumerate() {

src/query/expression/src/filter/filter_executor.rs

@@ -13,13 +13,13 @@
 // limitations under the License.
 
 use core::ops::Range;
-use std::collections::HashSet;
 
 use databend_common_column::bitmap::MutableBitmap;
 use databend_common_exception::Result;
 
 use crate::filter::SelectExpr;
 use crate::filter::Selector;
+use crate::ColumnSet;
 use crate::DataBlock;
 use crate::Evaluator;
 use crate::Expr;
@@ -35,7 +35,7 @@ pub struct FilterExecutor {
     true_selection: Vec<u32>,
     false_selection: Vec<u32>,
     has_or: bool,
-    projections: Option<HashSet<usize>>,
+    projections: Option<ColumnSet>,
     max_block_size: usize,
     selection_range: Vec<Range<u32>>,
     fn_registry: &'static FunctionRegistry,
@@ -47,7 +47,7 @@ impl FilterExecutor {
         expr: Expr,
         func_ctx: FunctionContext,
         max_block_size: usize,
-        projections: Option<HashSet<usize>>,
+        projections: Option<ColumnSet>,
         fn_registry: &'static FunctionRegistry,
         keep_order: bool,
     ) -> Self {

src/query/expression/src/lib.rs

@@ -98,3 +98,6 @@ pub mod expr {
     pub use super::expression::FunctionCall;
     pub use super::expression::LambdaFunctionCall;
 }
+
+type IndexType = usize;
+type ColumnSet = std::collections::BTreeSet<IndexType>;

src/query/functions/tests/it/aggregates/mod.rs

@@ -78,7 +78,7 @@ pub fn run_agg_ast(
     let used_columns = raw_expr
         .column_refs()
         .keys()
-        .cloned()
+        .copied()
         .sorted()
         .collect::<Vec<_>>();
 

src/query/service/src/interpreters/interpreter_explain.rs

@@ -34,9 +34,9 @@ use databend_common_pipeline_core::ExecutionInfo;
 use databend_common_sql::binder::ExplainConfig;
 use databend_common_sql::executor::format_partial_tree;
 use databend_common_sql::executor::MutationBuildInfo;
-use databend_common_sql::optimizer::ir::ColumnSet;
 use databend_common_sql::plans::Mutation;
 use databend_common_sql::BindContext;
+use databend_common_sql::ColumnSet;
 use databend_common_sql::FormatOptions;
 use databend_common_sql::MetadataRef;
 use databend_common_storages_result_cache::gen_result_cache_key;

src/query/service/src/interpreters/interpreter_optimize_compact_block.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashSet;
 use std::sync::Arc;
 
 use databend_common_base::runtime::GlobalIORuntime;
@@ -23,6 +22,7 @@ use databend_common_pipeline_core::ExecutionInfo;
 use databend_common_sql::executor::PhysicalPlanBuilder;
 use databend_common_sql::optimizer::ir::SExpr;
 use databend_common_sql::plans::OptimizeCompactBlock;
+use databend_common_sql::ColumnSet;
 use databend_common_sql::MetadataRef;
 
 use crate::interpreters::interpreter_optimize_purge::purge;
@@ -83,7 +83,7 @@ impl Interpreter for OptimizeCompactBlockInterpreter {
 
         let mut build_res = PipelineBuildResult::create();
         let mut builder = PhysicalPlanBuilder::new(MetadataRef::default(), self.ctx.clone(), false);
-        match builder.build(&self.s_expr, HashSet::new()).await {
+        match builder.build(&self.s_expr, ColumnSet::new()).await {
             Ok(physical_plan) => {
                 build_res =
                     build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan)

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

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::collections::HashMap;
-use std::collections::HashSet;
 use std::sync::Arc;
 
 use databend_common_catalog::catalog::CatalogManager;
@@ -34,6 +33,7 @@ use databend_common_sql::executor::physical_plans::ChunkFilter;
 use databend_common_sql::executor::physical_plans::ChunkMerge;
 use databend_common_sql::executor::physical_plans::Duplicate;
 use databend_common_sql::executor::physical_plans::Shuffle;
+use databend_common_sql::ColumnSet;
 use databend_common_storages_fuse::operations::CommitMultiTableInsert;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::TableContext;
@@ -60,7 +60,7 @@ impl PipelineBuilder {
             return Ok(());
         }
         let mut f: Vec<DynTransformBuilder> = Vec::with_capacity(plan.predicates.len());
-        let projection: HashSet<_> = (0..plan.input.output_schema()?.fields.len()).collect();
+        let projection: ColumnSet = (0..plan.input.output_schema()?.fields.len()).collect();
         for predicate in plan.predicates.iter() {
             if let Some(predicate) = predicate {
                 f.push(Box::new(self.filter_transform_builder(

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

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashSet;
 use std::sync::Arc;
 
 use databend_common_exception::ErrorCode;
@@ -50,11 +49,12 @@ use crate::pipelines::processors::TransformCastSchema;
 use crate::pipelines::processors::TransformResortAddOn;
 use crate::pipelines::PipelineBuilder;
 use crate::sql::executor::physical_plans::MutationKind;
+
 impl PipelineBuilder {
     pub(crate) fn filter_transform_builder(
         &self,
         predicates: &[RemoteExpr],
-        projections: HashSet<usize>,
+        projections: ColumnSet,
     ) -> Result<impl Fn(Arc<InputPort>, Arc<OutputPort>) -> Result<ProcessorPtr>> {
         let predicate = predicates
             .iter()

src/query/service/src/pipelines/processors/transforms/hash_join/transform_hash_join_probe.rs

@@ -21,8 +21,8 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
 use databend_common_expression::FunctionContext;
-use databend_common_sql::optimizer::ir::ColumnSet;
 use databend_common_sql::plans::JoinType;
+use databend_common_sql::ColumnSet;
 
 use crate::pipelines::processors::transforms::hash_join::transform_hash_join_build::HashTableType;
 use crate::pipelines::processors::transforms::hash_join::HashJoinProbeState;

src/query/service/src/pipelines/processors/transforms/transform_filter.rs

@@ -25,7 +25,7 @@ use databend_common_expression::FunctionContext;
 use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_common_pipeline_transforms::processors::BlockingTransform;
 use databend_common_pipeline_transforms::processors::BlockingTransformer;
-use databend_common_sql::optimizer::ir::ColumnSet;
+use databend_common_sql::ColumnSet;
 
 use crate::pipelines::processors::InputPort;
 use crate::pipelines::processors::OutputPort;