chore: remove SortColumnDescription.is_nullable (#16699)

forsaken628 · web-flow · commit 134976a209b5 · 2024-10-28T15:49:32.000+08:00
remove is_nullable

Signed-off-by: coldWater &lt;forsaken628@gmail.com&gt;
diff --git a/src/query/expression/src/kernels/sort.rs b/src/query/expression/src/kernels/sort.rs
@@ -36,7 +36,6 @@ pub struct SortColumnDescription {
     pub offset: usize,
     pub asc: bool,
     pub nulls_first: bool,
-    pub is_nullable: bool,
 }
 
 #[derive(Copy, Clone, Debug)]
@@ -140,11 +139,10 @@ pub fn compare_scalars(rows: Vec<Vec<Scalar>>, data_types: &[DataType]) -> Resul
     let descriptions = order_columns
         .iter()
         .enumerate()
-        .map(|(idx, array)| SortColumnDescription {
+        .map(|(idx, _)| SortColumnDescription {
             offset: idx,
             asc: true,
             nulls_first: false,
-            is_nullable: array.data_type().is_nullable(),
         })
         .collect::<Vec<_>>();
 
diff --git a/src/query/expression/tests/it/sort.rs b/src/query/expression/tests/it/sort.rs
@@ -43,7 +43,6 @@ fn test_block_sort() -> Result<()> {
                 offset: 0,
                 asc: true,
                 nulls_first: false,
-                is_nullable: false,
             }],
             None,
             vec![
@@ -56,7 +55,6 @@ fn test_block_sort() -> Result<()> {
                 offset: 0,
                 asc: true,
                 nulls_first: false,
-                is_nullable: false,
             }],
             Some(4),
             vec![
@@ -69,7 +67,6 @@ fn test_block_sort() -> Result<()> {
                 offset: 1,
                 asc: false,
                 nulls_first: false,
-                is_nullable: false,
             }],
             None,
             vec![
@@ -83,13 +80,11 @@ fn test_block_sort() -> Result<()> {
                     offset: 0,
                     asc: true,
                     nulls_first: false,
-                    is_nullable: false,
                 },
                 SortColumnDescription {
                     offset: 1,
                     asc: false,
                     nulls_first: false,
-                    is_nullable: false,
                 },
             ],
             None,
@@ -129,7 +124,6 @@ fn test_block_sort() -> Result<()> {
                 offset: 0,
                 asc: true,
                 nulls_first: false,
-                is_nullable: false,
             }],
             None,
             vec![
@@ -142,7 +136,6 @@ fn test_block_sort() -> Result<()> {
                 offset: 0,
                 asc: true,
                 nulls_first: false,
-                is_nullable: false,
             }],
             Some(4),
             vec![
@@ -155,7 +148,6 @@ fn test_block_sort() -> Result<()> {
                 offset: 1,
                 asc: false,
                 nulls_first: false,
-                is_nullable: false,
             }],
             None,
             vec![
@@ -169,13 +161,11 @@ fn test_block_sort() -> Result<()> {
                     offset: 0,
                     asc: true,
                     nulls_first: false,
-                    is_nullable: false,
                 },
                 SortColumnDescription {
                     offset: 1,
                     asc: false,
                     nulls_first: false,
-                    is_nullable: false,
                 },
             ],
             None,
@@ -227,7 +217,6 @@ fn sort_concat() {
                 offset: *i,
                 asc: rng.gen_bool(0.5),
                 nulls_first: rng.gen_bool(0.5),
-                is_nullable: rng.gen_bool(0.5),
             })
             .collect_vec();
 
diff --git a/src/query/functions/src/scalars/array.rs b/src/query/functions/src/scalars/array.rs
@@ -868,7 +868,6 @@ fn register_array_aggr(registry: &mut FunctionRegistry) {
                     offset: 0,
                     asc: sort_desc.0,
                     nulls_first: sort_desc.1,
-                    is_nullable: false,  // This information is not needed here.
                 }];
                 let columns = vec![BlockEntry{
                     data_type: arr.data_type(),
diff --git a/src/query/pipeline/transforms/src/processors/transforms/sort/rows/simple.rs b/src/query/pipeline/transforms/src/processors/transforms/sort/rows/simple.rs
@@ -194,7 +194,6 @@ impl<T: ArgType> SimpleRowConverter<T> {
             offset: 0,
             asc,
             nulls_first: false,
-            is_nullable: false,
         }];
 
         R::from_column(&col, &desc)
diff --git a/src/query/pipeline/transforms/tests/it/merger.rs b/src/query/pipeline/transforms/tests/it/merger.rs
@@ -150,7 +150,6 @@ fn create_test_merger<A: SortAlgorithm>(
         offset: 0,
         asc: true,
         nulls_first: true,
-        is_nullable: false,
     }]);
     let streams = input
         .into_iter()
diff --git a/src/query/service/src/pipelines/builders/builder_aggregate.rs b/src/query/service/src/pipelines/builders/builder_aggregate.rs
@@ -141,18 +141,17 @@ impl PipelineBuilder {
         // For rank limit, we can filter data using sort with rank before partial
         if let Some(rank_limit) = &aggregate.rank_limit {
             let sort_desc = rank_limit
-            .0
-            .iter()
-            .map(|desc| {
-                let offset = schema_before_group_by.index_of(&desc.order_by.to_string())?;
-                Ok(SortColumnDescription {
-                    offset,
-                    asc: desc.asc,
-                    nulls_first: desc.nulls_first,
-                    is_nullable: schema_before_group_by.field(offset).is_nullable(),  // This information is not needed here.
+                .0
+                .iter()
+                .map(|desc| {
+                    let offset = schema_before_group_by.index_of(&desc.order_by.to_string())?;
+                    Ok(SortColumnDescription {
+                        offset,
+                        asc: desc.asc,
+                        nulls_first: desc.nulls_first,
+                    })
                 })
-            })
-            .collect::<Result<Vec<_>>>()?;
+                .collect::<Result<Vec<_>>>()?;
             let sort_desc = Arc::new(sort_desc);
 
             self.main_pipeline.add_transformer(|| {
diff --git a/src/query/service/src/pipelines/builders/builder_insert_multi_table.rs b/src/query/service/src/pipelines/builders/builder_insert_multi_table.rs
@@ -222,7 +222,6 @@ impl PipelineBuilder {
                         offset: *index,
                         asc: true,
                         nulls_first: false,
-                        is_nullable: false, // This information is not needed here.
                     })
                     .collect();
                 let sort_desc = Arc::new(sort_desc);
diff --git a/src/query/service/src/pipelines/builders/builder_recluster.rs b/src/query/service/src/pipelines/builders/builder_recluster.rs
@@ -126,7 +126,6 @@ impl PipelineBuilder {
                         offset: *offset,
                         asc: true,
                         nulls_first: false,
-                        is_nullable: false, // This information is not needed here.
                     })
                     .collect();
 
diff --git a/src/query/service/src/pipelines/builders/builder_sort.rs b/src/query/service/src/pipelines/builders/builder_sort.rs
@@ -83,7 +83,6 @@ impl PipelineBuilder {
                     offset,
                     asc: desc.asc,
                     nulls_first: desc.nulls_first,
-                    is_nullable: plan_schema.field(offset).is_nullable(),  // This information is not needed here.
                 })
             })
             .collect::<Result<Vec<_>>>()?;
diff --git a/src/query/service/src/pipelines/builders/builder_window.rs b/src/query/service/src/pipelines/builders/builder_window.rs
@@ -30,11 +30,12 @@ use opendal::services::Fs;
 use opendal::Operator;
 
 use crate::pipelines::processors::transforms::FrameBound;
+use crate::pipelines::processors::transforms::TransformWindow;
 use crate::pipelines::processors::transforms::TransformWindowPartitionCollect;
 use crate::pipelines::processors::transforms::WindowFunctionInfo;
 use crate::pipelines::processors::transforms::WindowPartitionExchange;
+use crate::pipelines::processors::transforms::WindowSortDesc;
 use crate::pipelines::processors::transforms::WindowSpillSettings;
-use crate::pipelines::processors::TransformWindow;
 use crate::pipelines::PipelineBuilder;
 use crate::spillers::SpillerDiskConfig;
 
@@ -56,11 +57,11 @@ impl PipelineBuilder {
             .iter()
             .map(|o| {
                 let offset = input_schema.index_of(&o.order_by.to_string())?;
-                Ok(SortColumnDescription {
+                Ok(WindowSortDesc {
                     offset,
                     asc: o.asc,
                     nulls_first: o.nulls_first,
-                    is_nullable: input_schema.field(offset).is_nullable(), // Used for check null frame.
+                    is_nullable: input_schema.field(offset).is_nullable(),
                 })
             })
             .collect::<Result<Vec<_>>>()?;
@@ -164,7 +165,6 @@ impl PipelineBuilder {
                     offset,
                     asc: desc.asc,
                     nulls_first: desc.nulls_first,
-                    is_nullable: plan_schema.field(offset).is_nullable(),
                 })
             })
             .collect::<Result<Vec<_>>>()?;
diff --git a/src/query/service/src/pipelines/processors/mod.rs b/src/query/service/src/pipelines/processors/mod.rs
@@ -25,4 +25,3 @@ pub use transforms::TransformLimit;
 pub use transforms::TransformNullIf;
 pub use transforms::TransformResortAddOn;
 pub use transforms::TransformResortAddOnWithoutSourceSchema;
-pub use transforms::TransformWindow;
diff --git a/src/query/service/src/pipelines/processors/transforms/range_join/ie_join_state.rs b/src/query/service/src/pipelines/processors/transforms/range_join/ie_join_state.rs
@@ -82,19 +82,16 @@ impl IEJoinState {
                 offset: 0,
                 asc: l1_order,
                 nulls_first: false,
-                is_nullable: l1_data_type.is_nullable(),
             },
             SortColumnDescription {
                 offset: 1,
                 asc: l2_order,
                 nulls_first: false,
-                is_nullable: l2_data_type.is_nullable(),
             },
             SortColumnDescription {
                 offset: 2,
                 asc: false,
                 nulls_first: false,
-                is_nullable: false,
             },
         ];
 
@@ -103,20 +100,17 @@ impl IEJoinState {
                 offset: 1,
                 asc: l2_order,
                 nulls_first: false,
-                is_nullable: l2_data_type.is_nullable(),
             },
             SortColumnDescription {
                 offset: 0,
                 asc: l1_order,
                 nulls_first: false,
-                is_nullable: l1_data_type.is_nullable(),
             },
             // `_tuple_id` column
             SortColumnDescription {
                 offset: 2,
                 asc: false,
                 nulls_first: false,
-                is_nullable: false,
             },
         ];
 
diff --git a/src/query/service/src/pipelines/processors/transforms/range_join/merge_join_state.rs b/src/query/service/src/pipelines/processors/transforms/range_join/merge_join_state.rs
@@ -141,31 +141,17 @@ impl RangeJoinState {
     }
 
     // Used by merge join
-    fn sort_descriptions(&self, left: bool) -> Vec<SortColumnDescription> {
+    fn sort_descriptions(&self, _: bool) -> Vec<SortColumnDescription> {
         let op = &self.conditions[0].operator;
         let asc = match op.as_str() {
             "gt" | "gte" => false,
             "lt" | "lte" => true,
             _ => unreachable!(),
         };
-        let is_nullable = if left {
-            self.conditions[0]
-                .left_expr
-                .as_expr(&BUILTIN_FUNCTIONS)
-                .data_type()
-                .is_nullable()
-        } else {
-            self.conditions[0]
-                .right_expr
-                .as_expr(&BUILTIN_FUNCTIONS)
-                .data_type()
-                .is_nullable()
-        };
         vec![SortColumnDescription {
             offset: 0,
             asc,
             nulls_first: true,
-            is_nullable,
         }]
     }
 }
diff --git a/src/query/service/src/pipelines/processors/transforms/transform_sort_spill.rs b/src/query/service/src/pipelines/processors/transforms/transform_sort_spill.rs
@@ -512,7 +512,6 @@ mod tests {
             offset: 0,
             asc: true,
             nulls_first: true,
-            is_nullable: false,
         }]);
 
         let transform = TransformSortSpill::<A>::create(
diff --git a/src/query/service/src/pipelines/processors/transforms/window/mod.rs b/src/query/service/src/pipelines/processors/transforms/window/mod.rs
@@ -19,5 +19,5 @@ mod window_function;
 
 pub use frame_bound::FrameBound;
 pub use partition::*;
-pub use transform_window::TransformWindow;
+pub use transform_window::*;
 pub use window_function::WindowFunctionInfo;
diff --git a/src/query/service/src/pipelines/processors/transforms/window/transform_window.rs b/src/query/service/src/pipelines/processors/transforms/window/transform_window.rs
diff --git a/src/query/service/tests/it/pipelines/transforms/sort/k_way.rs b/src/query/service/tests/it/pipelines/transforms/sort/k_way.rs
diff --git a/src/query/service/tests/it/pipelines/transforms/sort/spill.rs b/src/query/service/tests/it/pipelines/transforms/sort/spill.rs
diff --git a/src/query/storages/fuse/src/operations/append.rs b/src/query/storages/fuse/src/operations/append.rs
