chore(query): introducing ColumnView provides a new abstraction that unifies access to Column and const Column (#18260)

Author: forsaken628

src/query/ee/tests/it/ngram_index/index_refresh.rs

@@ -115,16 +115,15 @@ async fn get_block_filter(fixture: &TestFixture, columns: &[String]) -> Result<B
         .await
         .transpose()?
         .unwrap();
-    let path = block.columns()[0].to_column().remove_nullable();
-    let path_scalar = path.as_string().unwrap().index(0).unwrap();
-    let length = block.columns()[1].to_column();
-    let length_scalar = length.as_number().unwrap().index(0).unwrap();
+    let path = block.get_by_offset(0).index(0).unwrap();
+    let path = *path.as_string().unwrap();
+    let length = block.get_by_offset(1).index(0).unwrap();
 
     load_bloom_filter_by_columns(
         DataOperator::instance().operator(),
         columns,
-        path_scalar,
-        *length_scalar.as_u_int64().unwrap(),
+        path,
+        *length.as_number().unwrap().as_u_int64().unwrap(),
     )
     .await
 }

src/query/expression/src/aggregate/aggregate_function.rs

@@ -22,11 +22,12 @@ use super::AggrState;
 use super::AggrStateLoc;
 use super::AggrStateRegistry;
 use super::StateAddr;
-use crate::types::BinaryColumn;
+use crate::types::BinaryType;
 use crate::types::DataType;
-use crate::Column;
+use crate::BlockEntry;
 use crate::ColumnBuilder;
-use crate::InputColumns;
+use crate::ColumnView;
+use crate::ProjectedBlock;
 use crate::Scalar;
 
 pub type AggregateFunctionRef = Arc<dyn AggregateFunction>;
@@ -46,7 +47,7 @@ pub trait AggregateFunction: fmt::Display + Sync + Send {
     fn accumulate(
         &self,
         place: AggrState,
-        columns: InputColumns,
+        columns: ProjectedBlock,
         validity: Option<&Bitmap>,
         input_rows: usize,
     ) -> Result<()>;
@@ -56,7 +57,7 @@ pub trait AggregateFunction: fmt::Display + Sync + Send {
         &self,
         addrs: &[StateAddr],
         loc: &[AggrStateLoc],
-        columns: InputColumns,
+        columns: ProjectedBlock,
         _input_rows: usize,
     ) -> Result<()> {
         for (row, addr) in addrs.iter().enumerate() {
@@ -66,7 +67,7 @@ pub trait AggregateFunction: fmt::Display + Sync + Send {
     }
 
     // Used in aggregate_null_adaptor
-    fn accumulate_row(&self, place: AggrState, columns: InputColumns, row: usize) -> Result<()>;
+    fn accumulate_row(&self, place: AggrState, columns: ProjectedBlock, row: usize) -> Result<()>;
 
     fn serialize(&self, place: AggrState, writer: &mut Vec<u8>) -> Result<()>;
 
@@ -81,7 +82,7 @@ pub trait AggregateFunction: fmt::Display + Sync + Send {
         &self,
         places: &[StateAddr],
         loc: &[AggrStateLoc],
-        state: &BinaryColumn,
+        state: &ColumnView<BinaryType>,
     ) -> Result<()> {
         for (place, mut data) in places.iter().zip(state.iter()) {
             self.merge(AggrState::new(*place, loc), &mut data)?;
@@ -90,9 +91,9 @@ pub trait AggregateFunction: fmt::Display + Sync + Send {
         Ok(())
     }
 
-    fn batch_merge_single(&self, place: AggrState, state: &Column) -> Result<()> {
-        let c = state.as_binary().unwrap();
-        for mut data in c.iter() {
+    fn batch_merge_single(&self, place: AggrState, state: &BlockEntry) -> Result<()> {
+        let view = state.downcast::<BinaryType>().unwrap();
+        for mut data in view.iter() {
             self.merge(place, &mut data)?;
         }
         Ok(())
@@ -145,7 +146,7 @@ pub trait AggregateFunction: fmt::Display + Sync + Send {
         Ok(None)
     }
 
-    fn get_if_condition(&self, _columns: InputColumns) -> Option<Bitmap> {
+    fn get_if_condition(&self, _columns: ProjectedBlock) -> Option<Bitmap> {
         None
     }
 

src/query/expression/src/aggregate/aggregate_hashtable.rs

@@ -27,13 +27,14 @@ use crate::aggregate::payload_row::row_match_columns;
 use crate::group_hash_columns;
 use crate::new_sel;
 use crate::read;
+use crate::types::BinaryType;
 use crate::types::DataType;
 use crate::AggregateFunctionRef;
-use crate::Column;
+use crate::BlockEntry;
 use crate::ColumnBuilder;
 use crate::HashTableConfig;
-use crate::InputColumns;
 use crate::Payload;
+use crate::ProjectedBlock;
 use crate::StateAddr;
 use crate::BATCH_SIZE;
 use crate::LOAD_FACTOR;
@@ -129,9 +130,9 @@ impl AggregateHashTable {
     pub fn add_groups(
         &mut self,
         state: &mut ProbeState,
-        group_columns: InputColumns,
-        params: &[InputColumns],
-        agg_states: InputColumns,
+        group_columns: ProjectedBlock,
+        params: &[ProjectedBlock],
+        agg_states: ProjectedBlock,
         row_count: usize,
     ) -> Result<usize> {
         if row_count <= BATCH_ADD_SIZE {
@@ -142,13 +143,13 @@ impl AggregateHashTable {
                 let end = (start + BATCH_ADD_SIZE).min(row_count);
                 let step_group_columns = group_columns
                     .iter()
-                    .map(|c| c.slice(start..end))
+                    .map(|entry| entry.slice(start..end))
                     .collect::<Vec<_>>();
 
-                let step_params: Vec<Vec<Column>> = params
+                let step_params: Vec<Vec<BlockEntry>> = params
                     .iter()
                     .map(|c| c.iter().map(|x| x.slice(start..end)).collect())
-                    .collect::<Vec<_>>();
+                    .collect();
                 let step_params = step_params.iter().map(|v| v.into()).collect::<Vec<_>>();
                 let agg_states = agg_states
                     .iter()
@@ -171,9 +172,9 @@ impl AggregateHashTable {
     fn add_groups_inner(
         &mut self,
         state: &mut ProbeState,
-        group_columns: InputColumns,
-        params: &[InputColumns],
-        agg_states: InputColumns,
+        group_columns: ProjectedBlock,
+        params: &[ProjectedBlock],
+        agg_states: ProjectedBlock,
         row_count: usize,
     ) -> Result<usize> {
         state.row_count = row_count;
@@ -218,7 +219,7 @@ impl AggregateHashTable {
                     .zip(agg_states.iter())
                     .zip(states_layout.states_loc.iter())
                 {
-                    func.batch_merge(state_places, loc, state.as_binary().unwrap())?;
+                    func.batch_merge(state_places, loc, &state.downcast::<BinaryType>().unwrap())?;
                 }
             }
         }
@@ -245,7 +246,7 @@ impl AggregateHashTable {
     fn probe_and_create(
         &mut self,
         state: &mut ProbeState,
-        group_columns: InputColumns,
+        group_columns: ProjectedBlock,
         row_count: usize,
     ) -> usize {
         // exceed capacity or should resize
@@ -444,7 +445,7 @@ impl AggregateHashTable {
                     loc,
                     &mut builder,
                 )?;
-                flush_state.aggregate_results.push(builder.build());
+                flush_state.aggregate_results.push(builder.build().into());
             }
         }
         Ok(true)

src/query/expression/src/aggregate/group_hash.rs

@@ -51,20 +51,20 @@ use crate::with_decimal_mapped_type;
 use crate::with_number_mapped_type;
 use crate::with_number_type;
 use crate::Column;
-use crate::InputColumns;
+use crate::ProjectedBlock;
 use crate::Scalar;
 use crate::ScalarRef;
 use crate::Value;
 
 const NULL_HASH_VAL: u64 = 0xd1cefa08eb382d69;
 
-pub fn group_hash_columns(cols: InputColumns, values: &mut [u64]) {
+pub fn group_hash_columns(cols: ProjectedBlock, values: &mut [u64]) {
     debug_assert!(!cols.is_empty());
-    for (i, col) in cols.iter().enumerate() {
+    for (i, entry) in cols.iter().enumerate() {
         if i == 0 {
-            combine_group_hash_column::<true>(col, values);
+            combine_group_hash_column::<true>(&entry.to_column(), values);
         } else {
-            combine_group_hash_column::<false>(col, values);
+            combine_group_hash_column::<false>(&entry.to_column(), values);
         }
     }
 }

src/query/expression/src/aggregate/partitioned_payload.rs

@@ -23,8 +23,8 @@ use crate::get_states_layout;
 use crate::read;
 use crate::types::DataType;
 use crate::AggregateFunctionRef;
-use crate::InputColumns;
 use crate::PayloadFlushState;
+use crate::ProjectedBlock;
 use crate::StatesLayout;
 use crate::BATCH_SIZE;
 
@@ -111,7 +111,7 @@ impl PartitionedPayload {
         &mut self,
         state: &mut ProbeState,
         new_group_rows: usize,
-        group_columns: InputColumns,
+        group_columns: ProjectedBlock,
     ) {
         if self.payloads.len() == 1 {
             self.payloads[0].reserve_append_rows(

src/query/expression/src/aggregate/payload.rs

@@ -17,7 +17,6 @@ use std::sync::Arc;
 
 use bumpalo::Bump;
 use databend_common_base::runtime::drop_guard;
-use itertools::Itertools;
 use log::info;
 use strength_reduce::StrengthReducedU64;
 
@@ -31,8 +30,8 @@ use crate::AggregateFunctionRef;
 use crate::Column;
 use crate::ColumnBuilder;
 use crate::DataBlock;
-use crate::InputColumns;
 use crate::PayloadFlushState;
+use crate::ProjectedBlock;
 use crate::SelectVector;
 use crate::StateAddr;
 use crate::StatesLayout;
@@ -202,7 +201,7 @@ impl Payload {
         address: &mut [*const u8],
         page_index: &mut [usize],
         new_group_rows: usize,
-        group_columns: InputColumns,
+        group_columns: ProjectedBlock,
     ) {
         let tuple_size = self.tuple_size;
         let (mut page, mut page_index_value) = self.writable_page();
@@ -233,13 +232,13 @@ impl Payload {
         address: &mut [*const u8],
         page_index: &mut [usize],
         new_group_rows: usize,
-        group_columns: InputColumns,
+        group_columns: ProjectedBlock,
     ) {
         let mut write_offset = 0;
         // write validity
-        for col in group_columns.iter() {
-            if let Column::Nullable(c) = col {
-                let bitmap = &c.validity;
+        for entry in group_columns.iter() {
+            if let Column::Nullable(c) = entry.to_column() {
+                let bitmap = c.validity();
                 if bitmap.null_count() == 0 || bitmap.null_count() == bitmap.len() {
                     let val: u8 = if bitmap.null_count() == 0 { 1 } else { 0 };
                     // faster path
@@ -262,13 +261,13 @@ impl Payload {
         }
 
         let mut scratch = vec![];
-        for (idx, col) in group_columns.iter().enumerate() {
+        for (idx, entry) in group_columns.iter().enumerate() {
             debug_assert!(write_offset == self.group_offsets[idx]);
 
             unsafe {
                 serialize_column_to_rowformat(
                     &self.arena,
-                    col,
+                    &entry.to_column(),
                     select_vector,
                     new_group_rows,
                     address,
@@ -424,15 +423,19 @@ impl Payload {
 
     pub fn empty_block(&self, fake_rows: Option<usize>) -> DataBlock {
         let fake_rows = fake_rows.unwrap_or(0);
-        let columns = (0..self.aggrs.len())
-            .map(|_| ColumnBuilder::repeat_default(&DataType::Binary, fake_rows).build())
+        let entries = (0..self.aggrs.len())
+            .map(|_| {
+                ColumnBuilder::repeat_default(&DataType::Binary, fake_rows)
+                    .build()
+                    .into()
+            })
             .chain(
                 self.group_types
                     .iter()
-                    .map(|t| ColumnBuilder::repeat_default(t, fake_rows).build()),
+                    .map(|t| ColumnBuilder::repeat_default(t, fake_rows).build().into()),
             )
-            .collect_vec();
-        DataBlock::new_from_columns(columns)
+            .collect();
+        DataBlock::new(entries, fake_rows)
     }
 }
 

src/query/expression/src/aggregate/payload_flush.rs

@@ -40,6 +40,7 @@ use crate::types::NumberType;
 use crate::types::ReturnType;
 use crate::types::TimestampType;
 use crate::with_number_mapped_type;
+use crate::BlockEntry;
 use crate::Column;
 use crate::ColumnBuilder;
 use crate::DataBlock;
@@ -49,8 +50,8 @@ use crate::BATCH_SIZE;
 
 pub struct PayloadFlushState {
     pub probe_state: ProbeState,
-    pub group_columns: Vec<Column>,
-    pub aggregate_results: Vec<Column>,
+    pub group_columns: Vec<BlockEntry>,
+    pub(super) aggregate_results: Vec<BlockEntry>,
     pub row_count: usize,
 
     pub flush_partition: usize,
@@ -88,10 +89,10 @@ impl PayloadFlushState {
         self.flush_page_row = 0;
     }
 
-    pub fn take_group_columns(&mut self) -> Vec<Column> {
+    pub fn take_group_columns(&mut self) -> Vec<BlockEntry> {
         std::mem::take(&mut self.group_columns)
     }
-    pub fn take_aggregate_results(&mut self) -> Vec<Column> {
+    pub fn take_aggregate_results(&mut self) -> Vec<BlockEntry> {
         std::mem::take(&mut self.aggregate_results)
     }
 }
@@ -136,7 +137,7 @@ impl Payload {
 
         let row_count = state.row_count;
 
-        let mut cols = Vec::with_capacity(self.aggrs.len() + self.group_types.len());
+        let mut entries = Vec::with_capacity(self.aggrs.len() + self.group_types.len());
         if let Some(state_layout) = self.states_layout.as_ref() {
             let mut builders = state_layout.serialize_builders(row_count);
 
@@ -155,24 +156,24 @@ impl Payload {
                 }
             }
 
-            cols.extend(
+            entries.extend(
                 builders
                     .into_iter()
-                    .map(|builder| Column::Binary(builder.build())),
+                    .map(|builder| Column::Binary(builder.build()).into()),
             );
         }
 
-        cols.extend_from_slice(&state.take_group_columns());
-        Ok(Some(DataBlock::new_from_columns(cols)))
+        entries.extend_from_slice(&state.take_group_columns());
+        Ok(Some(DataBlock::new(entries, row_count)))
     }
 
     pub fn group_by_flush_all(&self) -> Result<DataBlock> {
         let mut state = PayloadFlushState::default();
         let mut blocks = vec![];
 
         while self.flush(&mut state) {
-            let cols = state.take_group_columns();
-            blocks.push(DataBlock::new_from_columns(cols));
+            let entries = state.take_group_columns();
+            blocks.push(DataBlock::new(entries, state.row_count));
         }
 
         if blocks.is_empty() {
@@ -219,7 +220,7 @@ impl Payload {
 
         for col_index in 0..self.group_types.len() {
             let col = self.flush_column(col_index, state);
-            state.group_columns.push(col);
+            state.group_columns.push(col.into());
         }
 
         state.flush_page_row = end;