From 31e9ad02d4b729da3d158a88035fb65f6064d761 Mon Sep 17 00:00:00 2001 From: Freejww <103876282+Freejww@users.noreply.github.com> Date: Wed, 13 Mar 2024 13:37:59 +0800 Subject: [PATCH] feat(query): support spill for new agg hashtable (#14905) * test * test * finish singleton spill * cluster spill * Update src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_meta.rs Co-authored-by: Winter Zhang * add Result<> for func return type * fix bug * fix decimal bug * remove output columns(AggregatePartial) to pass ci * revert set --------- Co-authored-by: jw Co-authored-by: Winter Zhang --- .../src/aggregate/aggregate_hashtable.rs | 33 +++- .../src/aggregate/partitioned_payload.rs | 25 ++- src/query/expression/src/aggregate/payload.rs | 35 ++-- .../expression/src/aggregate/payload_flush.rs | 103 +++++++++--- .../expression/src/aggregate/payload_row.rs | 45 ++--- src/query/expression/src/kernels/utils.rs | 13 +- .../tests/it/aggregates/agg_hashtable.rs | 19 ++- .../aggregator/aggregate_exchange_injector.rs | 71 +++++++- .../transforms/aggregator/aggregate_meta.rs | 65 ++++++++ .../transforms/aggregator/serde/serde_meta.rs | 23 ++- .../serde/transform_aggregate_serializer.rs | 47 +----- .../serde/transform_aggregate_spill_writer.rs | 156 +++++++++++++++-- .../serde/transform_deserializer.rs | 1 + ...transform_exchange_aggregate_serializer.rs | 152 +++++++++++++++++ .../transform_exchange_group_by_serializer.rs | 150 +++++++++++++++++ .../serde/transform_group_by_serializer.rs | 33 +--- .../serde/transform_group_by_spill_writer.rs | 157 ++++++++++++++++-- .../serde/transform_spill_reader.rs | 4 +- .../aggregator/transform_aggregate_final.rs | 28 ++++ .../aggregator/transform_aggregate_partial.rs | 38 ++++- .../aggregator/transform_group_by_final.rs | 28 ++++ .../aggregator/transform_group_by_partial.rs | 29 +++- .../aggregator/transform_partition_bucket.rs | 97 ++++------- src/query/sql/src/executor/format.rs | 4 - .../02_0000_async_agg_index_base.test | 3 - .../suites/mode/cluster/exchange.test | 6 - .../suites/mode/cluster/explain_v2.test | 2 - .../suites/mode/cluster/group_shuffle.test | 2 - .../mode/standalone/ee/explain_agg_index.test | 20 --- .../mode/standalone/explain/aggregate.test | 14 -- .../mode/standalone/explain/explain.test | 28 ++-- .../explain/explain_grouping_sets.test | 2 - .../mode/standalone/explain/fold_count.test | 2 - .../standalone/explain/join_reorder/mark.test | 1 - .../mode/standalone/explain/lateral.test | 2 - .../suites/mode/standalone/explain/limit.test | 4 - .../mode/standalone/explain/project_set.test | 1 - .../mode/standalone/explain/prune_column.test | 6 - .../standalone/explain/push_down_filter.test | 4 - .../mode/standalone/explain/subquery.test | 4 - .../standalone/explain_native/aggregate.test | 10 -- .../standalone/explain_native/explain.test | 6 +- .../explain_native/explain_grouping_sets.test | 2 - .../standalone/explain_native/fold_count.test | 1 - .../mode/standalone/explain_native/limit.test | 4 - .../explain_native/project_set.test | 1 - .../explain_native/prune_column.test | 6 - .../explain_native/push_down_filter.test | 4 - .../standalone/explain_native/subquery.test | 4 - 49 files changed, 1108 insertions(+), 387 deletions(-) diff --git a/src/query/expression/src/aggregate/aggregate_hashtable.rs b/src/query/expression/src/aggregate/aggregate_hashtable.rs index 279df546121b..7fa14fde74e7 100644 --- a/src/query/expression/src/aggregate/aggregate_hashtable.rs +++ b/src/query/expression/src/aggregate/aggregate_hashtable.rs @@ -15,7 +15,9 @@ // A new AggregateHashtable which inspired by duckdb's https://duckdb.org/2022/03/07/aggregate-hashtable.html use std::sync::atomic::Ordering; +use std::sync::Arc; +use bumpalo::Bump; use databend_common_exception::Result; use super::partitioned_payload::PartitionedPayload; @@ -24,6 +26,7 @@ use super::probe_state::ProbeState; use crate::aggregate::payload_row::row_match_columns; use crate::group_hash_columns; use crate::new_sel; +use crate::read; use crate::types::DataType; use crate::AggregateFunctionRef; use crate::Column; @@ -44,7 +47,7 @@ pub struct AggregateHashTable { pub payload: PartitionedPayload, // use for append rows directly during deserialize pub direct_append: bool, - config: HashTableConfig, + pub config: HashTableConfig, current_radix_bits: u64, entries: Vec, count: usize, @@ -59,9 +62,10 @@ impl AggregateHashTable { group_types: Vec, aggrs: Vec, config: HashTableConfig, + arena: Arc, ) -> Self { let capacity = Self::initial_capacity(); - Self::new_with_capacity(group_types, aggrs, config, capacity) + Self::new_with_capacity(group_types, aggrs, config, capacity, arena) } pub fn new_with_capacity( @@ -69,13 +73,19 @@ impl AggregateHashTable { aggrs: Vec, config: HashTableConfig, capacity: usize, + arena: Arc, ) -> Self { Self { entries: vec![0u64; capacity], count: 0, direct_append: false, current_radix_bits: config.initial_radix_bits, - payload: PartitionedPayload::new(group_types, aggrs, 1 << config.initial_radix_bits), + payload: PartitionedPayload::new( + group_types, + aggrs, + 1 << config.initial_radix_bits, + vec![arena], + ), capacity, config, } @@ -150,8 +160,8 @@ impl AggregateHashTable { if !self.payload.aggrs.is_empty() { for i in 0..row_count { state.state_places[i] = unsafe { - StateAddr::new(core::ptr::read::( - state.addresses[i].add(self.payload.state_offset) as _, + StateAddr::new(read::( + state.addresses[i].add(self.payload.state_offset) as _ ) as usize) }; } @@ -356,7 +366,7 @@ impl AggregateHashTable { if !self.payload.aggrs.is_empty() { for i in 0..row_count { flush_state.probe_state.state_places[i] = unsafe { - StateAddr::new(core::ptr::read::( + StateAddr::new(read::( flush_state.probe_state.addresses[i].add(self.payload.state_offset) as _, ) as usize) @@ -446,6 +456,7 @@ impl AggregateHashTable { self.payload.group_types.clone(), self.payload.aggrs.clone(), 1, + vec![Arc::new(Bump::new())], ); let payload = std::mem::replace(&mut self.payload, temp_payload); let mut state = PayloadFlushState::default(); @@ -530,6 +541,16 @@ impl AggregateHashTable { pub fn reset_count(&mut self) { self.count = 0; } + + pub fn allocated_bytes(&self) -> usize { + self.payload.memory_size() + + self + .payload + .arenas + .iter() + .map(|arena| arena.allocated_bytes()) + .sum::() + } } /// Upper 16 bits are salt diff --git a/src/query/expression/src/aggregate/partitioned_payload.rs b/src/query/expression/src/aggregate/partitioned_payload.rs index fdcee199e571..507e203684ab 100644 --- a/src/query/expression/src/aggregate/partitioned_payload.rs +++ b/src/query/expression/src/aggregate/partitioned_payload.rs @@ -20,6 +20,7 @@ use itertools::Itertools; use super::payload::Payload; use super::probe_state::ProbeState; +use crate::read; use crate::types::DataType; use crate::AggregateFunctionRef; use crate::Column; @@ -54,14 +55,13 @@ impl PartitionedPayload { group_types: Vec, aggrs: Vec, partition_count: u64, + arenas: Vec>, ) -> Self { let radix_bits = partition_count.trailing_zeros() as u64; debug_assert_eq!(1 << radix_bits, partition_count); - let arena = Arc::new(Bump::new()); - let payloads = (0..partition_count) - .map(|_| Payload::new(arena.clone(), group_types.clone(), aggrs.clone())) + .map(|_| Payload::new(arenas[0].clone(), group_types.clone(), aggrs.clone())) .collect_vec(); let group_sizes = payloads[0].group_sizes.clone(); @@ -85,7 +85,7 @@ impl PartitionedPayload { state_layout, partition_count, - arenas: vec![arena], + arenas, mask_v: mask(radix_bits), shift_v: shift(radix_bits), } @@ -145,13 +145,14 @@ impl PartitionedPayload { self.group_types.clone(), self.aggrs.clone(), new_partition_count as u64, + self.arenas.clone(), ); new_partition_payload.combine(self, state); new_partition_payload } - pub fn combine(&mut self, mut other: PartitionedPayload, state: &mut PayloadFlushState) { + pub fn combine(&mut self, other: PartitionedPayload, state: &mut PayloadFlushState) { if other.partition_count == self.partition_count { for (l, r) in self.payloads.iter_mut().zip(other.payloads.into_iter()) { l.combine(r); @@ -163,7 +164,6 @@ impl PartitionedPayload { self.combine_single(payload, state) } } - self.arenas.append(&mut other.arenas); } pub fn combine_single(&mut self, mut other: Payload, state: &mut PayloadFlushState) { @@ -218,8 +218,7 @@ impl PartitionedPayload { for idx in 0..rows { state.addresses[idx] = other.data_ptr(page, idx + state.flush_page_row); - let hash = - unsafe { core::ptr::read::(state.addresses[idx].add(self.hash_offset) as _) }; + let hash = unsafe { read::(state.addresses[idx].add(self.hash_offset) as _) }; let partition_idx = ((hash & self.mask_v) >> self.shift_v) as usize; @@ -250,6 +249,16 @@ impl PartitionedPayload { pub fn memory_size(&self) -> usize { self.payloads.iter().map(|x| x.memory_size()).sum() } + + pub fn include_arena(&self, other: &Arc) -> bool { + for arena in self.arenas.iter() { + if Arc::ptr_eq(arena, other) { + return true; + } + } + + false + } } #[inline] diff --git a/src/query/expression/src/aggregate/payload.rs b/src/query/expression/src/aggregate/payload.rs index d045ce41ee1a..02ac9d8a597c 100644 --- a/src/query/expression/src/aggregate/payload.rs +++ b/src/query/expression/src/aggregate/payload.rs @@ -18,15 +18,19 @@ use std::sync::Arc; use bumpalo::Bump; use databend_common_base::runtime::drop_guard; +use itertools::Itertools; use strength_reduce::StrengthReducedU64; use super::payload_row::rowformat_size; use super::payload_row::serialize_column_to_rowformat; use crate::get_layout_offsets; +use crate::read; use crate::store; use crate::types::DataType; use crate::AggregateFunctionRef; use crate::Column; +use crate::ColumnBuilder; +use crate::DataBlock; use crate::PayloadFlushState; use crate::SelectVector; use crate::StateAddr; @@ -41,7 +45,6 @@ use crate::MAX_PAGE_SIZE; // [STATE_ADDRS] is the state_addrs of the aggregate functions, 8 bytes each pub struct Payload { pub arena: Arc, - pub arenas: Vec>, // if true, the states are moved out of the payload into other payload, and will not be dropped pub state_move_out: bool, pub group_types: Vec, @@ -124,8 +127,7 @@ impl Payload { let row_per_page = (u16::MAX as usize).min(MAX_PAGE_SIZE / tuple_size).max(1); Self { - arena: arena.clone(), - arenas: vec![arena], + arena, state_move_out: false, pages: vec![], current_write_page: 0, @@ -236,14 +238,14 @@ impl Payload { for idx in select_vector.iter().take(new_group_rows).copied() { unsafe { let dst = address[idx].add(write_offset); - store(val, dst as *mut u8); + store::(&val, dst as *mut u8); } } } else { for idx in select_vector.iter().take(new_group_rows).copied() { unsafe { let dst = address[idx].add(write_offset); - store(bitmap.get_bit(idx) as u8, dst as *mut u8); + store::(&(bitmap.get_bit(idx) as u8), dst as *mut u8); } } } @@ -274,7 +276,7 @@ impl Payload { for idx in select_vector.iter().take(new_group_rows).copied() { unsafe { let dst = address[idx].add(write_offset); - store(group_hashes[idx], dst as *mut u8); + store::(&group_hashes[idx], dst as *mut u8); } } @@ -286,7 +288,7 @@ impl Payload { let place = self.arena.alloc_layout(layout); unsafe { let dst = address[idx].add(write_offset); - store(place.as_ptr() as u64, dst as *mut u8); + store::(&(place.as_ptr() as u64), dst as *mut u8); } let place = StateAddr::from(place); @@ -364,8 +366,7 @@ impl Payload { for idx in 0..rows { state.addresses[idx] = self.data_ptr(page, idx + state.flush_page_row); - let hash = - unsafe { core::ptr::read::(state.addresses[idx].add(self.hash_offset) as _) }; + let hash = unsafe { read::(state.addresses[idx].add(self.hash_offset) as _) }; let partition_idx = (hash % mods) as usize; @@ -376,6 +377,20 @@ impl Payload { state.flush_page_row = end; true } + + pub fn empty_block(&self) -> DataBlock { + let columns = self + .aggrs + .iter() + .map(|f| ColumnBuilder::with_capacity(&f.return_type().unwrap(), 0).build()) + .chain( + self.group_types + .iter() + .map(|t| ColumnBuilder::with_capacity(t, 0).build()), + ) + .collect_vec(); + DataBlock::new_from_columns(columns) + } } impl Drop for Payload { @@ -388,7 +403,7 @@ impl Drop for Payload { for page in self.pages.iter() { for row in 0..page.rows { unsafe { - let state_place = StateAddr::new(core::ptr::read::( + let state_place = StateAddr::new(read::( self.data_ptr(page, row).add(self.state_offset) as _, ) as usize); diff --git a/src/query/expression/src/aggregate/payload_flush.rs b/src/query/expression/src/aggregate/payload_flush.rs index a5c961c8fb1e..8a7c4099dca9 100644 --- a/src/query/expression/src/aggregate/payload_flush.rs +++ b/src/query/expression/src/aggregate/payload_flush.rs @@ -12,12 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +use databend_common_exception::Result; use databend_common_io::prelude::bincode_deserialize_from_slice; use ethnum::i256; use super::partitioned_payload::PartitionedPayload; use super::payload::Payload; use super::probe_state::ProbeState; +use crate::read; use crate::types::binary::BinaryColumn; use crate::types::binary::BinaryColumnBuilder; use crate::types::decimal::Decimal; @@ -34,8 +36,10 @@ use crate::types::NumberType; use crate::types::TimestampType; use crate::types::ValueType; use crate::with_number_mapped_type; +use crate::AggregateFunctionRef; use crate::Column; use crate::ColumnBuilder; +use crate::DataBlock; use crate::Scalar; use crate::StateAddr; use crate::BATCH_SIZE; @@ -108,6 +112,67 @@ impl PartitionedPayload { } impl Payload { + pub fn aggregate_flush_all(&self) -> Result { + let mut state = PayloadFlushState::default(); + let mut blocks = vec![]; + + while self.flush(&mut state) { + let row_count = state.row_count; + + let mut state_builders: Vec = self + .aggrs + .iter() + .map(|agg| state_serializer(agg, row_count)) + .collect(); + + for place in state.state_places.as_slice()[0..row_count].iter() { + for (idx, (addr_offset, aggr)) in self + .state_addr_offsets + .iter() + .zip(self.aggrs.iter()) + .enumerate() + { + let arg_place = place.next(*addr_offset); + aggr.serialize(arg_place, &mut state_builders[idx].data) + .unwrap(); + state_builders[idx].commit_row(); + } + } + + let mut cols = Vec::with_capacity(self.aggrs.len() + self.group_types.len()); + for builder in state_builders.into_iter() { + let col = Column::Binary(builder.build()); + cols.push(col); + } + + cols.extend_from_slice(&state.take_group_columns()); + + blocks.push(DataBlock::new_from_columns(cols)); + } + + if blocks.is_empty() { + return Ok(self.empty_block()); + } + + DataBlock::concat(&blocks) + } + + pub fn group_by_flush_all(&self) -> Result { + 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)); + } + + if blocks.is_empty() { + return Ok(self.empty_block()); + } + + DataBlock::concat(&blocks) + } + pub fn flush(&self, state: &mut PayloadFlushState) -> bool { if state.flush_page >= self.pages.len() { return false; @@ -132,13 +197,13 @@ impl Payload { for idx in 0..rows { state.addresses[idx] = self.data_ptr(page, idx + state.flush_page_row); state.probe_state.group_hashes[idx] = - unsafe { core::ptr::read::(state.addresses[idx].add(self.hash_offset) as _) }; + unsafe { read::(state.addresses[idx].add(self.hash_offset) as _) }; if !self.aggrs.is_empty() { state.state_places[idx] = unsafe { - StateAddr::new(core::ptr::read::( - state.addresses[idx].add(self.state_offset) as _, - ) as usize) + StateAddr::new( + read::(state.addresses[idx].add(self.state_offset) as _) as usize, + ) }; } } @@ -204,9 +269,8 @@ impl Payload { state: &mut PayloadFlushState, ) -> Column { let len = state.probe_state.row_count; - let iter = (0..len).map(|idx| unsafe { - core::ptr::read::(state.addresses[idx].add(col_offset) as _) - }); + let iter = (0..len) + .map(|idx| unsafe { read::(state.addresses[idx].add(col_offset) as _) }); let col = T::column_from_iter(iter, &[]); T::upcast_column(col) } @@ -219,8 +283,8 @@ impl Payload { ) -> Column { let len = state.probe_state.row_count; let iter = (0..len).map(|idx| unsafe { - core::ptr::read::< as ValueType>::Scalar>( - state.addresses[idx].add(col_offset) as _, + read::< as ValueType>::Scalar>( + state.addresses[idx].add(col_offset) as _ ) }); let col = DecimalType::::column_from_iter(iter, &[]); @@ -237,11 +301,9 @@ impl Payload { unsafe { for idx in 0..len { - let str_len = - core::ptr::read::(state.addresses[idx].add(col_offset) as _) as usize; - let data_address = - core::ptr::read::(state.addresses[idx].add(col_offset + 4) as _) as usize - as *const u8; + let str_len = read::(state.addresses[idx].add(col_offset) as _) as usize; + let data_address = read::(state.addresses[idx].add(col_offset + 4) as _) + as usize as *const u8; let scalar = std::slice::from_raw_parts(data_address, str_len); @@ -271,11 +333,9 @@ impl Payload { unsafe { for idx in 0..len { - let str_len = - core::ptr::read::(state.addresses[idx].add(col_offset) as _) as usize; - let data_address = - core::ptr::read::(state.addresses[idx].add(col_offset + 4) as _) as usize - as *const u8; + let str_len = read::(state.addresses[idx].add(col_offset) as _) as usize; + let data_address = read::(state.addresses[idx].add(col_offset + 4) as _) + as usize as *const u8; let scalar = std::slice::from_raw_parts(data_address, str_len); let scalar: Scalar = bincode_deserialize_from_slice(scalar).unwrap(); @@ -286,3 +346,8 @@ impl Payload { builder.build() } } + +fn state_serializer(func: &AggregateFunctionRef, row: usize) -> BinaryColumnBuilder { + let size = func.serialize_size_per_row().unwrap_or(4); + BinaryColumnBuilder::with_capacity(row, row * size) +} diff --git a/src/query/expression/src/aggregate/payload_row.rs b/src/query/expression/src/aggregate/payload_row.rs index 9ccf28b74671..fb2dc158e3ce 100644 --- a/src/query/expression/src/aggregate/payload_row.rs +++ b/src/query/expression/src/aggregate/payload_row.rs @@ -18,6 +18,7 @@ use databend_common_io::prelude::bincode_deserialize_from_slice; use databend_common_io::prelude::bincode_serialize_into_buf; use ethnum::i256; +use crate::read; use crate::store; use crate::types::binary::BinaryColumn; use crate::types::decimal::DecimalColumn; @@ -76,7 +77,7 @@ pub unsafe fn serialize_column_to_rowformat( Column::Number(v) => with_number_mapped_type!(|NUM_TYPE| match v { NumberColumn::NUM_TYPE(buffer) => { for index in select_vector.iter().take(rows).copied() { - store(buffer[index], address[index].add(offset) as *mut u8); + store(&buffer[index], address[index].add(offset) as *mut u8); } } }), @@ -84,7 +85,7 @@ pub unsafe fn serialize_column_to_rowformat( with_decimal_mapped_type!(|DECIMAL_TYPE| match v { DecimalColumn::DECIMAL_TYPE(buffer, _) => { for index in select_vector.iter().take(rows).copied() { - store(buffer[index], address[index].add(offset) as *mut u8); + store(&buffer[index], address[index].add(offset) as *mut u8); } } }) @@ -94,12 +95,12 @@ pub unsafe fn serialize_column_to_rowformat( let val: u8 = if v.unset_bits() == 0 { 1 } else { 0 }; // faster path for index in select_vector.iter().take(rows).copied() { - store(val, address[index].add(offset) as *mut u8); + store(&val, address[index].add(offset) as *mut u8); } } else { for index in select_vector.iter().take(rows).copied() { store( - v.get_bit(index) as u8, + &(v.get_bit(index) as u8), address[index].add(offset) as *mut u8, ); } @@ -108,9 +109,9 @@ pub unsafe fn serialize_column_to_rowformat( Column::Binary(v) | Column::Bitmap(v) | Column::Variant(v) | Column::Geometry(v) => { for index in select_vector.iter().take(rows).copied() { let data = arena.alloc_slice_copy(v.index_unchecked(index)); - store(data.len() as u32, address[index].add(offset) as *mut u8); + store(&(data.len() as u32), address[index].add(offset) as *mut u8); store( - data.as_ptr() as u64, + &(data.as_ptr() as u64), address[index].add(offset + 4) as *mut u8, ); } @@ -118,21 +119,21 @@ pub unsafe fn serialize_column_to_rowformat( Column::String(v) => { for index in select_vector.iter().take(rows).copied() { let data = arena.alloc_str(v.index_unchecked(index)); - store(data.len() as u32, address[index].add(offset) as *mut u8); + store(&(data.len() as u32), address[index].add(offset) as *mut u8); store( - data.as_ptr() as u64, + &(data.as_ptr() as u64), address[index].add(offset + 4) as *mut u8, ); } } Column::Timestamp(buffer) => { for index in select_vector.iter().take(rows).copied() { - store(buffer[index], address[index].add(offset) as *mut u8); + store(&buffer[index], address[index].add(offset) as *mut u8); } } Column::Date(buffer) => { for index in select_vector.iter().take(rows).copied() { - store(buffer[index], address[index].add(offset) as *mut u8); + store(&buffer[index], address[index].add(offset) as *mut u8); } } Column::Nullable(c) => serialize_column_to_rowformat( @@ -153,9 +154,9 @@ pub unsafe fn serialize_column_to_rowformat( bincode_serialize_into_buf(scratch, &s).unwrap(); let data = arena.alloc_slice_copy(scratch); - store(data.len() as u32, address[index].add(offset) as *mut u8); + store(&(data.len() as u32), address[index].add(offset) as *mut u8); store( - data.as_ptr() as u64, + &(data.as_ptr() as u64), address[index].add(offset + 4) as *mut u8, ); } @@ -362,19 +363,19 @@ unsafe fn row_match_binary_column( for idx in select_vector[..*count].iter() { let idx = *idx; let validity_address = address[idx].add(validity_offset); - let is_set2 = core::ptr::read::(validity_address as _) != 0; + let is_set2 = read::(validity_address as _) != 0; let is_set = is_all_set || validity.get_bit_unchecked(idx); if is_set && is_set2 { let len_address = address[idx].add(col_offset); let address = address[idx].add(col_offset + 4); - let len = core::ptr::read::(len_address as _) as usize; + let len = read::(len_address as _) as usize; let value = BinaryType::index_column_unchecked(col, idx); if len != value.len() { equal = false; } else { - let data_address = core::ptr::read::(address as _) as usize as *const u8; + let data_address = read::(address as _) as usize as *const u8; let scalar = std::slice::from_raw_parts(data_address, len); equal = databend_common_hashtable::fast_memcmp(scalar, value); } @@ -396,13 +397,13 @@ unsafe fn row_match_binary_column( let len_address = address[idx].add(col_offset); let address = address[idx].add(col_offset + 4); - let len = core::ptr::read::(len_address as _) as usize; + let len = read::(len_address as _) as usize; let value = BinaryType::index_column_unchecked(col, idx); if len != value.len() { equal = false; } else { - let data_address = core::ptr::read::(address as _) as usize as *const u8; + let data_address = read::(address as _) as usize as *const u8; let scalar = std::slice::from_raw_parts(data_address, len); equal = databend_common_hashtable::fast_memcmp(scalar, value); @@ -444,11 +445,11 @@ unsafe fn row_match_column_type( for idx in select_vector[..*count].iter() { let idx = *idx; let validity_address = address[idx].add(validity_offset); - let is_set2 = core::ptr::read::(validity_address as _) != 0; + let is_set2 = read::(validity_address as _) != 0; let is_set = is_all_set || validity.get_bit_unchecked(idx); if is_set && is_set2 { let address = address[idx].add(col_offset); - let scalar = core::ptr::read::<::Scalar>(address as _); + let scalar = read::<::Scalar>(address as _); let value = T::index_column_unchecked(&col, idx); let value = T::to_owned_scalar(value); @@ -470,7 +471,7 @@ unsafe fn row_match_column_type( let idx = *idx; let value = T::index_column_unchecked(&col, idx); let address = address[idx].add(col_offset); - let scalar = core::ptr::read::<::Scalar>(address as _); + let scalar = read::<::Scalar>(address as _); let value = T::to_owned_scalar(value); if scalar.eq(&value) { @@ -502,12 +503,12 @@ unsafe fn row_match_generic_column( for idx in select_vector[..*count].iter() { let idx = *idx; let len_address = address[idx].add(col_offset); - let len = core::ptr::read::(len_address as _) as usize; + let len = read::(len_address as _) as usize; let address = address[idx].add(col_offset + 4); let value = AnyType::index_column_unchecked(col, idx); - let data_address = core::ptr::read::(address as _) as usize as *const u8; + let data_address = read::(address as _) as usize as *const u8; let scalar = std::slice::from_raw_parts(data_address, len); let scalar: Scalar = bincode_deserialize_from_slice(scalar).unwrap(); diff --git a/src/query/expression/src/kernels/utils.rs b/src/query/expression/src/kernels/utils.rs index 2b594fe7ad61..d74bb15ba2d9 100644 --- a/src/query/expression/src/kernels/utils.rs +++ b/src/query/expression/src/kernels/utils.rs @@ -66,10 +66,17 @@ pub unsafe fn set_vec_len_by_ptr(vec: &mut Vec, ptr: *const T) { } /// # Safety -/// # As: core::ptr::write +/// # As: core::ptr::copy_nonoverlapping #[inline] -pub unsafe fn store(val: T, ptr: *mut u8) { - core::ptr::write(ptr as _, val) +pub unsafe fn store(val: &T, ptr: *mut u8) { + core::ptr::copy_nonoverlapping(val as *const T as *const u8, ptr, std::mem::size_of::()); +} + +/// # Safety +/// # As: core::ptr::read_unaligned +#[inline] +pub unsafe fn read(ptr: *const u8) -> T { + core::ptr::read_unaligned::(ptr as _) } /// Iterates over an arbitrarily aligned byte buffer diff --git a/src/query/functions/tests/it/aggregates/agg_hashtable.rs b/src/query/functions/tests/it/aggregates/agg_hashtable.rs index 9fc7a270abd5..6f00b119c004 100644 --- a/src/query/functions/tests/it/aggregates/agg_hashtable.rs +++ b/src/query/functions/tests/it/aggregates/agg_hashtable.rs @@ -26,6 +26,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::sync::Arc; + +use bumpalo::Bump; use databend_common_expression::block_debug::assert_block_value_sort_eq; use databend_common_expression::types::ArgType; use databend_common_expression::types::BooleanType; @@ -87,16 +90,24 @@ fn test_agg_hashtable() { let params: Vec> = aggrs.iter().map(|_| vec![columns[1].clone()]).collect(); let config = HashTableConfig::default(); - let mut hashtable = - AggregateHashTable::new(group_types.clone(), aggrs.clone(), config.clone()); + let mut hashtable = AggregateHashTable::new( + group_types.clone(), + aggrs.clone(), + config.clone(), + Arc::new(Bump::new()), + ); let mut state = ProbeState::default(); let _ = hashtable .add_groups(&mut state, &group_columns, ¶ms, &[], n) .unwrap(); - let mut hashtable2 = - AggregateHashTable::new(group_types.clone(), aggrs.clone(), config.clone()); + let mut hashtable2 = AggregateHashTable::new( + group_types.clone(), + aggrs.clone(), + config.clone(), + Arc::new(Bump::new()), + ); let mut state2 = ProbeState::default(); let _ = hashtable2 diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_exchange_injector.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_exchange_injector.rs index 1232679ae8ac..93cd9b5039b3 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_exchange_injector.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_exchange_injector.rs @@ -21,6 +21,7 @@ use databend_common_exception::ErrorCode; use databend_common_exception::Result; use databend_common_expression::BlockMetaInfoDowncast; use databend_common_expression::DataBlock; +use databend_common_expression::PartitionedPayload; use databend_common_expression::Payload; use databend_common_expression::PayloadFlushState; use databend_common_hashtable::FastHash; @@ -81,7 +82,8 @@ impl ExchangeSorting AggregateMeta::HashTable(v) => Ok(v.bucket), AggregateMeta::AggregateHashTable(_) => unreachable!(), AggregateMeta::AggregatePayload(v) => Ok(v.bucket), - AggregateMeta::Spilled(_) + AggregateMeta::AggregateSpilling(_) + | AggregateMeta::Spilled(_) | AggregateMeta::Spilling(_) | AggregateMeta::BucketSpilled(_) => Ok(-1), }, @@ -151,16 +153,12 @@ fn scatter_paylaod(mut payload: Payload, buckets: usize) -> Result> for _ in 0..buckets.capacity() { buckets.push(Payload::new( - Arc::new(Bump::new()), + payload.arena.clone(), group_types.clone(), aggrs.clone(), )); } - for bucket in buckets.iter_mut() { - bucket.arenas.extend_from_slice(&payload.arenas); - } - // scatter each page of the payload. while payload.scatter(&mut state, buckets.len()) { // copy to the corresponding bucket. @@ -178,6 +176,60 @@ fn scatter_paylaod(mut payload: Payload, buckets: usize) -> Result> Ok(buckets) } +fn scatter_partitioned_payload( + partitioned_payload: PartitionedPayload, + buckets: usize, +) -> Result> { + let mut buckets = Vec::with_capacity(buckets); + + let group_types = partitioned_payload.group_types.clone(); + let aggrs = partitioned_payload.aggrs.clone(); + let partition_count = partitioned_payload.partition_count() as u64; + let mut state = PayloadFlushState::default(); + + for _ in 0..buckets.capacity() { + buckets.push(PartitionedPayload::new( + group_types.clone(), + aggrs.clone(), + partition_count, + partitioned_payload.arenas.clone(), + )); + } + + let mut payloads = Vec::with_capacity(buckets.len()); + + for _ in 0..payloads.capacity() { + payloads.push(Payload::new( + Arc::new(Bump::new()), + group_types.clone(), + aggrs.clone(), + )); + } + + for mut payload in partitioned_payload.payloads.into_iter() { + // scatter each page of the payload. + while payload.scatter(&mut state, buckets.len()) { + // copy to the corresponding bucket. + for (idx, bucket) in payloads.iter_mut().enumerate() { + let count = state.probe_state.partition_count[idx]; + + if count > 0 { + let sel = &state.probe_state.partition_entries[idx]; + bucket.copy_rows(sel, count, &state.addresses); + } + } + } + state.clear(); + payload.state_move_out = true; + } + + for (idx, payload) in payloads.into_iter().enumerate() { + buckets[idx].combine_single(payload, &mut state); + } + + Ok(buckets) +} + impl FlightScatter for HashTableHashScatter { @@ -201,6 +253,13 @@ impl FlightScatter }); } } + AggregateMeta::AggregateSpilling(payload) => { + for p in scatter_partitioned_payload(payload, self.buckets)? { + blocks.push(DataBlock::empty_with_meta( + AggregateMeta::::create_agg_spilling(p), + )) + } + } AggregateMeta::HashTable(payload) => { let bucket = payload.bucket; for hashtable_cell in scatter(payload, self.buckets, &self.method)? { diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_meta.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_meta.rs index b6f944a4908a..1be7d09960c6 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_meta.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_meta.rs @@ -15,13 +15,21 @@ use std::fmt::Debug; use std::fmt::Formatter; use std::ops::Range; +use std::sync::Arc; +use bumpalo::Bump; +use databend_common_exception::Result; +use databend_common_expression::types::DataType; +use databend_common_expression::AggregateFunction; +use databend_common_expression::AggregateHashTable; use databend_common_expression::BlockMetaInfo; use databend_common_expression::BlockMetaInfoPtr; use databend_common_expression::Column; use databend_common_expression::DataBlock; +use databend_common_expression::HashTableConfig; use databend_common_expression::PartitionedPayload; use databend_common_expression::Payload; +use databend_common_expression::ProbeState; use crate::pipelines::processors::transforms::aggregator::HashTableCell; use crate::pipelines::processors::transforms::group_by::HashMethodBounds; @@ -44,6 +52,53 @@ impl SerializedPayload { let entry = self.data_block.columns().last().unwrap(); entry.value.as_column().unwrap() } + + pub fn convert_to_partitioned_payload( + &self, + group_types: Vec, + aggrs: Vec>, + ) -> Result { + let rows_num = self.data_block.num_rows(); + let radix_bits = self.max_partition_count.trailing_zeros() as u64; + let config = HashTableConfig::default().with_initial_radix_bits(radix_bits); + let mut state = ProbeState::default(); + let agg_len = aggrs.len(); + let group_len = group_types.len(); + let mut hashtable = AggregateHashTable::new_with_capacity( + group_types, + aggrs, + config, + rows_num, + Arc::new(Bump::new()), + ); + hashtable.direct_append = true; + + let agg_states = (0..agg_len) + .map(|i| { + self.data_block + .get_by_offset(i) + .value + .as_column() + .unwrap() + .clone() + }) + .collect::>(); + let group_columns = (agg_len..(agg_len + group_len)) + .map(|i| { + self.data_block + .get_by_offset(i) + .value + .as_column() + .unwrap() + .clone() + }) + .collect::>(); + + let _ = + hashtable.add_groups(&mut state, &group_columns, &[vec![]], &agg_states, rows_num)?; + + Ok(hashtable.payload) + } } pub struct BucketSpilledPayload { @@ -51,11 +106,13 @@ pub struct BucketSpilledPayload { pub location: String, pub data_range: Range, pub columns_layout: Vec, + pub max_partition_count: usize, } pub struct AggregatePayload { pub bucket: isize, pub payload: Payload, + // use for new agg_hashtable pub max_partition_count: usize, } @@ -64,6 +121,7 @@ pub enum AggregateMeta { HashTable(HashTablePayload), AggregateHashTable(PartitionedPayload), AggregatePayload(AggregatePayload), + AggregateSpilling(PartitionedPayload), BucketSpilled(BucketSpilledPayload), Spilled(Vec), Spilling(HashTablePayload, V>), @@ -97,6 +155,10 @@ impl AggregateMeta BlockMetaInfoPtr { + Box::new(AggregateMeta::::AggregateSpilling(payload)) + } + pub fn create_serialized( bucket: isize, block: DataBlock, @@ -168,6 +230,9 @@ impl Debug for AggregateMeta AggregateMeta::AggregatePayload(_) => { f.debug_struct("AggregateMeta:AggregatePayload").finish() } + AggregateMeta::AggregateSpilling(_) => { + f.debug_struct("AggregateMeta:AggregateSpilling").finish() + } } } } diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/serde_meta.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/serde_meta.rs index abec404babfe..731ec4e1b104 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/serde_meta.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/serde_meta.rs @@ -29,8 +29,7 @@ pub struct AggregateSerdeMeta { pub location: Option, pub data_range: Option>, pub columns_layout: Vec, - // use for new agg_hashtable - pub is_agg_payload: bool, + // use for new agg hashtable pub max_partition_count: usize, } @@ -42,7 +41,6 @@ impl AggregateSerdeMeta { location: None, data_range: None, columns_layout: vec![], - is_agg_payload: false, max_partition_count: 0, }) } @@ -54,7 +52,6 @@ impl AggregateSerdeMeta { location: None, data_range: None, columns_layout: vec![], - is_agg_payload: true, max_partition_count, }) } @@ -71,10 +68,26 @@ impl AggregateSerdeMeta { columns_layout, location: Some(location), data_range: Some(data_range), - is_agg_payload: false, max_partition_count: 0, }) } + + pub fn create_agg_spilled( + bucket: isize, + location: String, + data_range: Range, + columns_layout: Vec, + max_partition_count: usize, + ) -> BlockMetaInfoPtr { + Box::new(AggregateSerdeMeta { + typ: SPILLED_TYPE, + bucket, + columns_layout, + location: Some(location), + data_range: Some(data_range), + max_partition_count, + }) + } } #[typetag::serde(name = "aggregate_serde")] diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_serializer.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_serializer.rs index 95f770ba1adc..eb9a50b092cf 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_serializer.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_serializer.rs @@ -22,7 +22,6 @@ use databend_common_expression::types::binary::BinaryColumnBuilder; use databend_common_expression::BlockMetaInfoDowncast; use databend_common_expression::Column; use databend_common_expression::DataBlock; -use databend_common_expression::PayloadFlushState; use databend_common_functions::aggregates::StateAddr; use databend_common_hashtable::HashtableEntryRefLike; use databend_common_hashtable::HashtableLike; @@ -34,7 +33,6 @@ use databend_common_pipeline_core::processors::ProcessorPtr; use super::SerializePayload; use crate::pipelines::processors::transforms::aggregator::create_state_serializer; -use crate::pipelines::processors::transforms::aggregator::empty_block; use crate::pipelines::processors::transforms::aggregator::estimated_key_size; use crate::pipelines::processors::transforms::aggregator::AggregateMeta; use crate::pipelines::processors::transforms::aggregator::AggregateSerdeMeta; @@ -139,6 +137,7 @@ impl TransformAggregateSerializer { AggregateMeta::BucketSpilled(_) => unreachable!(), AggregateMeta::Partitioned { .. } => unreachable!(), AggregateMeta::AggregateHashTable(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), AggregateMeta::HashTable(payload) => { self.input_data = Some(SerializeAggregateStream::create( &self.method, @@ -300,52 +299,10 @@ impl SerializeAggregateStream { self.finish(state_builders, group_key_builder) } SerializePayload::AggregatePayload(p) => { - let mut state = PayloadFlushState::default(); - let mut blocks = vec![]; - - while p.payload.flush(&mut state) { - let row_count = state.row_count; - - let mut state_builders: Vec = p - .payload - .aggrs - .iter() - .map(|agg| create_state_serializer(agg, row_count)) - .collect(); - - for place in state.state_places.as_slice()[0..row_count].iter() { - for (idx, (addr_offset, aggr)) in p - .payload - .state_addr_offsets - .iter() - .zip(p.payload.aggrs.iter()) - .enumerate() - { - let arg_place = place.next(*addr_offset); - aggr.serialize(arg_place, &mut state_builders[idx].data)?; - state_builders[idx].commit_row(); - } - } - - let mut cols = - Vec::with_capacity(p.payload.aggrs.len() + p.payload.group_types.len()); - for builder in state_builders.into_iter() { - let col = Column::Binary(builder.build()); - cols.push(col); - } - - cols.extend_from_slice(&state.take_group_columns()); - - blocks.push(DataBlock::new_from_columns(cols)); - } + let data_block = p.payload.aggregate_flush_all()?; self.end_iter = true; - let data_block = if blocks.is_empty() { - empty_block(p) - } else { - DataBlock::concat(&blocks).unwrap() - }; Ok(Some(data_block.add_meta(Some( AggregateSerdeMeta::create_agg_payload(p.bucket, p.max_partition_count), ))?)) diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_spill_writer.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_spill_writer.rs index 3ee101315447..1e2211fea2ad 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_spill_writer.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_spill_writer.rs @@ -26,6 +26,7 @@ use databend_common_exception::Result; use databend_common_expression::arrow::serialize_column; use databend_common_expression::BlockMetaInfoDowncast; use databend_common_expression::DataBlock; +use databend_common_expression::PartitionedPayload; use databend_common_hashtable::HashtableLike; use databend_common_metrics::transform::*; use databend_common_pipeline_core::processors::Event; @@ -135,6 +136,13 @@ impl Processor for TransformAggregateSpillWriter::downcast_from(block_meta); return Ok(Event::Sync); } + + if matches!(block_meta, AggregateMeta::AggregateSpilling(_)) { + self.input.set_not_need_data(); + let block_meta = data_block.take_meta().unwrap(); + self.spilling_meta = AggregateMeta::::downcast_from(block_meta); + return Ok(Event::Sync); + } } self.output.push_data(Ok(data_block)); @@ -152,20 +160,34 @@ impl Processor for TransformAggregateSpillWriter Result<()> { if let Some(spilling_meta) = self.spilling_meta.take() { - if let AggregateMeta::Spilling(payload) = spilling_meta { - self.spilling_future = Some(spilling_aggregate_payload( - self.ctx.clone(), - self.operator.clone(), - &self.method, - &self.location_prefix, - &self.params, - payload, - )?); - - return Ok(()); - } + match spilling_meta { + AggregateMeta::Spilling(payload) => { + self.spilling_future = Some(spilling_aggregate_payload( + self.ctx.clone(), + self.operator.clone(), + &self.method, + &self.location_prefix, + &self.params, + payload, + )?); + + return Ok(()); + } + AggregateMeta::AggregateSpilling(payload) => { + self.spilling_future = Some(agg_spilling_aggregate_payload::( + self.ctx.clone(), + self.operator.clone(), + &self.location_prefix, + payload, + )?); + + return Ok(()); + } - return Err(ErrorCode::Internal("")); + _ => { + return Err(ErrorCode::Internal("")); + } + } } Ok(()) @@ -181,6 +203,113 @@ impl Processor for TransformAggregateSpillWriter( + ctx: Arc, + operator: Operator, + location_prefix: &str, + partitioned_payload: PartitionedPayload, +) -> Result>> { + let unique_name = GlobalUniqName::unique(); + let location = format!("{}/{}", location_prefix, unique_name); + + let mut write_size = 0; + let partition_count = partitioned_payload.partition_count(); + let mut write_data = Vec::with_capacity(partition_count); + let mut spilled_buckets_payloads = Vec::with_capacity(partition_count); + // Record how many rows are spilled. + let mut rows = 0; + for (bucket, payload) in partitioned_payload.payloads.into_iter().enumerate() { + if payload.len() == 0 { + continue; + } + + let now = Instant::now(); + let data_block = payload.aggregate_flush_all()?; + rows += data_block.num_rows(); + + let begin = write_size; + let columns = data_block.columns().to_vec(); + let mut columns_data = Vec::with_capacity(columns.len()); + let mut columns_layout = Vec::with_capacity(columns.len()); + for column in columns.into_iter() { + let column = column.value.as_column().unwrap(); + let column_data = serialize_column(column); + write_size += column_data.len() as u64; + columns_layout.push(column_data.len() as u64); + columns_data.push(column_data); + } + + // perf + { + metrics_inc_aggregate_spill_data_serialize_milliseconds( + now.elapsed().as_millis() as u64 + ); + } + + write_data.push(columns_data); + spilled_buckets_payloads.push(BucketSpilledPayload { + bucket: bucket as isize, + location: location.clone(), + data_range: begin..write_size, + columns_layout, + max_partition_count: partition_count, + }); + } + + Ok(Box::pin(async move { + let instant = Instant::now(); + + let mut write_bytes = 0; + + if !write_data.is_empty() { + let mut writer = operator + .writer_with(&location) + .buffer(8 * 1024 * 1024) + .await?; + for write_bucket_data in write_data.into_iter() { + for data in write_bucket_data.into_iter() { + write_bytes += data.len(); + writer.write(data).await?; + } + } + + writer.close().await?; + } + + // perf + { + metrics_inc_aggregate_spill_write_count(); + metrics_inc_aggregate_spill_write_bytes(write_bytes as u64); + metrics_inc_aggregate_spill_write_milliseconds(instant.elapsed().as_millis() as u64); + + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteCount, 1); + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteBytes, write_bytes); + Profile::record_usize_profile( + ProfileStatisticsName::SpillWriteTime, + instant.elapsed().as_millis() as usize, + ); + } + + { + let progress_val = ProgressValues { + rows, + bytes: write_bytes, + }; + ctx.get_aggregate_spill_progress().incr(&progress_val); + } + + info!( + "Write aggregate spill {} successfully, elapsed: {:?}", + location, + instant.elapsed() + ); + + Ok(DataBlock::empty_with_meta( + AggregateMeta::::create_spilled(spilled_buckets_payloads), + )) + })) +} + pub fn spilling_aggregate_payload( ctx: Arc, operator: Operator, @@ -232,6 +361,7 @@ pub fn spilling_aggregate_payload( location: location.clone(), data_range: begin..write_size, columns_layout, + max_partition_count: 0, }); } diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_deserializer.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_deserializer.rs index e8dafbb02358..514e1a037aee 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_deserializer.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_deserializer.rs @@ -164,6 +164,7 @@ impl TransformDeserializer BlockMetaTransform }, )); } + Some(AggregateMeta::AggregateSpilling(payload)) => { + serialized_blocks.push(FlightSerialized::Future( + match index == self.local_pos { + true => local_agg_spilling_aggregate_payload::( + self.ctx.clone(), + self.operator.clone(), + &self.location_prefix, + payload, + )?, + false => agg_spilling_aggregate_payload::( + self.ctx.clone(), + self.operator.clone(), + &self.location_prefix, + payload, + )?, + }, + )); + } Some(AggregateMeta::HashTable(payload)) => { if index == self.local_pos { serialized_blocks.push(FlightSerialized::DataBlock(block.add_meta( @@ -211,6 +231,138 @@ impl BlockMetaTransform } } +fn agg_spilling_aggregate_payload( + ctx: Arc, + operator: Operator, + location_prefix: &str, + partitioned_payload: PartitionedPayload, +) -> Result>> { + let unique_name = GlobalUniqName::unique(); + let location = format!("{}/{}", location_prefix, unique_name); + + let partition_count = partitioned_payload.partition_count(); + let mut write_size = 0; + let mut write_data = Vec::with_capacity(partition_count); + let mut buckets_column_data = Vec::with_capacity(partition_count); + let mut data_range_start_column_data = Vec::with_capacity(partition_count); + let mut data_range_end_column_data = Vec::with_capacity(partition_count); + let mut columns_layout_column_data = Vec::with_capacity(partition_count); + // Record how many rows are spilled. + let mut rows = 0; + + for (bucket, payload) in partitioned_payload.payloads.into_iter().enumerate() { + if payload.len() == 0 { + continue; + } + + let now = Instant::now(); + let data_block = payload.aggregate_flush_all()?; + rows += data_block.num_rows(); + + let old_write_size = write_size; + let columns = data_block.columns().to_vec(); + let mut columns_data = Vec::with_capacity(columns.len()); + let mut columns_layout = Vec::with_capacity(columns.len()); + + for column in columns.into_iter() { + let column = column.value.as_column().unwrap(); + let column_data = serialize_column(column); + write_size += column_data.len() as u64; + columns_layout.push(column_data.len() as u64); + columns_data.push(column_data); + } + + // perf + { + metrics_inc_aggregate_spill_data_serialize_milliseconds( + now.elapsed().as_millis() as u64 + ); + } + + write_data.push(columns_data); + buckets_column_data.push(bucket as i64); + data_range_end_column_data.push(write_size); + columns_layout_column_data.push(columns_layout); + data_range_start_column_data.push(old_write_size); + } + + Ok(Box::pin(async move { + if !write_data.is_empty() { + let instant = Instant::now(); + + let mut write_bytes = 0; + let mut writer = operator + .writer_with(&location) + .buffer(8 * 1024 * 1024) + .await?; + for write_bucket_data in write_data.into_iter() { + for data in write_bucket_data.into_iter() { + write_bytes += data.len(); + writer.write(data).await?; + } + } + + writer.close().await?; + + // perf + { + metrics_inc_aggregate_spill_write_count(); + metrics_inc_aggregate_spill_write_bytes(write_bytes as u64); + metrics_inc_aggregate_spill_write_milliseconds(instant.elapsed().as_millis() as u64); + + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteCount, 1); + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteBytes, write_bytes); + Profile::record_usize_profile( + ProfileStatisticsName::SpillWriteTime, + instant.elapsed().as_millis() as usize, + ); + } + + { + { + let progress_val = ProgressValues { + rows, + bytes: write_bytes, + }; + ctx.get_aggregate_spill_progress().incr(&progress_val); + } + } + + info!( + "Write aggregate spill {} successfully, elapsed: {:?}", + location, + instant.elapsed() + ); + + let data_block = DataBlock::new_from_columns(vec![ + Int64Type::from_data(buckets_column_data), + UInt64Type::from_data(data_range_start_column_data), + UInt64Type::from_data(data_range_end_column_data), + ArrayType::upcast_column(ArrayType::::column_from_iter( + columns_layout_column_data + .into_iter() + .map(|x| UInt64Type::column_from_iter(x.into_iter(), &[])), + &[], + )), + ]); + + let data_block = data_block.add_meta(Some(AggregateSerdeMeta::create_agg_spilled( + -1, + location.clone(), + 0..0, + vec![], + partition_count, + )))?; + + let ipc_fields = exchange_defines::spilled_ipc_fields(); + let write_options = exchange_defines::spilled_write_options(); + return serialize_block(-1, data_block, ipc_fields, write_options); + } + + Ok(DataBlock::empty()) + })) +} + fn spilling_aggregate_payload( ctx: Arc, operator: Operator, diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_exchange_group_by_serializer.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_exchange_group_by_serializer.rs index 5990abaa102c..6190be88fb01 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_exchange_group_by_serializer.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_exchange_group_by_serializer.rs @@ -41,6 +41,7 @@ use databend_common_expression::BlockMetaInfoPtr; use databend_common_expression::DataBlock; use databend_common_expression::DataSchemaRef; use databend_common_expression::FromData; +use databend_common_expression::PartitionedPayload; use databend_common_hashtable::HashtableLike; use databend_common_metrics::transform::*; use databend_common_pipeline_core::processors::InputPort; @@ -57,6 +58,7 @@ use opendal::Operator; use super::SerializePayload; use crate::api::serialize_block; use crate::api::ExchangeShuffleMeta; +use crate::pipelines::processors::transforms::aggregator::agg_spilling_group_by_payload as local_agg_spilling_group_by_payload; use crate::pipelines::processors::transforms::aggregator::exchange_defines; use crate::pipelines::processors::transforms::aggregator::serialize_group_by; use crate::pipelines::processors::transforms::aggregator::spilling_group_by_payload as local_spilling_group_by_payload; @@ -209,6 +211,24 @@ impl BlockMetaTransform }, )); } + Some(AggregateMeta::AggregateSpilling(payload)) => { + serialized_blocks.push(FlightSerialized::Future( + match index == self.local_pos { + true => local_agg_spilling_group_by_payload::( + self.ctx.clone(), + self.operator.clone(), + &self.location_prefix, + payload, + )?, + false => agg_spilling_group_by_payload::( + self.ctx.clone(), + self.operator.clone(), + &self.location_prefix, + payload, + )?, + }, + )); + } Some(AggregateMeta::HashTable(payload)) => { if index == self.local_pos { serialized_blocks.push(FlightSerialized::DataBlock(block.add_meta( @@ -262,6 +282,136 @@ fn get_columns(data_block: DataBlock) -> Vec { data_block.columns().to_vec() } +fn agg_spilling_group_by_payload( + ctx: Arc, + operator: Operator, + location_prefix: &str, + partitioned_payload: PartitionedPayload, +) -> Result>> { + let unique_name = GlobalUniqName::unique(); + let location = format!("{}/{}", location_prefix, unique_name); + + let mut write_size = 0; + let partition_count = partitioned_payload.partition_count(); + let mut write_data = Vec::with_capacity(partition_count); + let mut buckets_column_data = Vec::with_capacity(partition_count); + let mut data_range_start_column_data = Vec::with_capacity(partition_count); + let mut data_range_end_column_data = Vec::with_capacity(partition_count); + let mut columns_layout_column_data = Vec::with_capacity(partition_count); + // Record how many rows are spilled + let mut rows = 0; + + for (bucket, payload) in partitioned_payload.payloads.into_iter().enumerate() { + if payload.len() == 0 { + continue; + } + + let now = Instant::now(); + let data_block = payload.group_by_flush_all()?; + rows += data_block.num_rows(); + + let old_write_size = write_size; + let columns = get_columns(data_block); + let mut columns_data = Vec::with_capacity(columns.len()); + let mut columns_layout = Vec::with_capacity(columns.len()); + + for column in columns.into_iter() { + let column = column.value.as_column().unwrap(); + let column_data = serialize_column(column); + write_size += column_data.len() as u64; + columns_layout.push(column_data.len() as u64); + columns_data.push(column_data); + } + + // perf + { + metrics_inc_aggregate_spill_data_serialize_milliseconds( + now.elapsed().as_millis() as u64 + ); + } + + write_data.push(columns_data); + buckets_column_data.push(bucket as i64); + data_range_end_column_data.push(write_size); + columns_layout_column_data.push(columns_layout); + data_range_start_column_data.push(old_write_size); + } + + Ok(Box::pin(async move { + let instant = Instant::now(); + + if !write_data.is_empty() { + let mut write_bytes = 0; + let mut writer = operator + .writer_with(&location) + .buffer(8 * 1024 * 1024) + .await?; + for write_bucket_data in write_data.into_iter() { + for data in write_bucket_data.into_iter() { + write_bytes += data.len(); + writer.write(data).await?; + } + } + + writer.close().await?; + + // perf + { + metrics_inc_group_by_spill_write_count(); + metrics_inc_group_by_spill_write_bytes(write_bytes as u64); + metrics_inc_group_by_spill_write_milliseconds(instant.elapsed().as_millis() as u64); + + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteCount, 1); + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteBytes, write_bytes); + Profile::record_usize_profile( + ProfileStatisticsName::SpillWriteTime, + instant.elapsed().as_millis() as usize, + ); + } + + { + let progress_val = ProgressValues { + rows, + bytes: write_bytes, + }; + ctx.get_group_by_spill_progress().incr(&progress_val); + } + + info!( + "Write aggregate spill {} successfully, elapsed: {:?}", + location, + instant.elapsed() + ); + + let data_block = DataBlock::new_from_columns(vec![ + Int64Type::from_data(buckets_column_data), + UInt64Type::from_data(data_range_start_column_data), + UInt64Type::from_data(data_range_end_column_data), + ArrayType::upcast_column(ArrayType::::column_from_iter( + columns_layout_column_data + .into_iter() + .map(|x| UInt64Type::column_from_iter(x.into_iter(), &[])), + &[], + )), + ]); + + let data_block = data_block.add_meta(Some(AggregateSerdeMeta::create_agg_spilled( + -1, + location.clone(), + 0..0, + vec![], + partition_count, + )))?; + + let ipc_fields = exchange_defines::spilled_ipc_fields(); + let write_options = exchange_defines::spilled_write_options(); + return serialize_block(-1, data_block, ipc_fields, write_options); + } + + Ok(DataBlock::empty()) + })) +} + fn spilling_group_by_payload( ctx: Arc, operator: Operator, diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_serializer.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_serializer.rs index 2d87c2c81ad9..7d952d7cf1d8 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_serializer.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_serializer.rs @@ -19,9 +19,7 @@ use std::sync::Arc; use databend_common_exception::Result; use databend_common_expression::BlockMetaInfoDowncast; -use databend_common_expression::ColumnBuilder; use databend_common_expression::DataBlock; -use databend_common_expression::PayloadFlushState; use databend_common_hashtable::HashtableEntryRefLike; use databend_common_hashtable::HashtableLike; use databend_common_pipeline_core::processors::Event; @@ -29,7 +27,6 @@ use databend_common_pipeline_core::processors::InputPort; use databend_common_pipeline_core::processors::OutputPort; use databend_common_pipeline_core::processors::Processor; use databend_common_pipeline_core::processors::ProcessorPtr; -use itertools::Itertools; use crate::pipelines::processors::transforms::aggregator::estimated_key_size; use crate::pipelines::processors::transforms::aggregator::AggregateMeta; @@ -131,6 +128,7 @@ impl TransformGroupBySerializer { AggregateMeta::BucketSpilled(_) => unreachable!(), AggregateMeta::Partitioned { .. } => unreachable!(), AggregateMeta::AggregateHashTable(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), AggregateMeta::AggregatePayload(p) => { self.input_data = Some(SerializeGroupByStream::create( &self.method, @@ -246,21 +244,10 @@ impl Iterator for SerializeGroupByStream { Some(data_block.add_meta(Some(AggregateSerdeMeta::create(bucket)))) } SerializePayload::AggregatePayload(p) => { - let mut state = PayloadFlushState::default(); - let mut blocks = vec![]; - - while p.payload.flush(&mut state) { - let col = state.take_group_columns(); - blocks.push(DataBlock::new_from_columns(col)); - } + let data_block = p.payload.group_by_flush_all().ok()?; self.end_iter = true; - let data_block = if blocks.is_empty() { - empty_block(p) - } else { - DataBlock::concat(&blocks).unwrap() - }; Some( data_block.add_meta(Some(AggregateSerdeMeta::create_agg_payload( p.bucket, @@ -271,19 +258,3 @@ impl Iterator for SerializeGroupByStream { } } } - -pub fn empty_block(p: &AggregatePayload) -> DataBlock { - let columns = p - .payload - .aggrs - .iter() - .map(|f| ColumnBuilder::with_capacity(&f.return_type().unwrap(), 0).build()) - .chain( - p.payload - .group_types - .iter() - .map(|t| ColumnBuilder::with_capacity(t, 0).build()), - ) - .collect_vec(); - DataBlock::new_from_columns(columns) -} diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_spill_writer.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_spill_writer.rs index 126d6bee2f5f..2ec44425fbd3 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_spill_writer.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_group_by_spill_writer.rs @@ -26,6 +26,7 @@ use databend_common_exception::Result; use databend_common_expression::arrow::serialize_column; use databend_common_expression::BlockMetaInfoDowncast; use databend_common_expression::DataBlock; +use databend_common_expression::PartitionedPayload; use databend_common_hashtable::HashtableLike; use databend_common_metrics::transform::*; use databend_common_pipeline_core::processors::Event; @@ -131,6 +132,13 @@ impl Processor for TransformGroupBySpillWriter self.spilling_meta = AggregateMeta::::downcast_from(block_meta); return Ok(Event::Sync); } + + if matches!(block_meta, AggregateMeta::AggregateSpilling(_)) { + self.input.set_not_need_data(); + let block_meta = data_block.take_meta().unwrap(); + self.spilling_meta = AggregateMeta::::downcast_from(block_meta); + return Ok(Event::Sync); + } } self.output.push_data(Ok(data_block)); @@ -148,21 +156,35 @@ impl Processor for TransformGroupBySpillWriter fn process(&mut self) -> Result<()> { if let Some(spilling_meta) = self.spilling_meta.take() { - if let AggregateMeta::Spilling(payload) = spilling_meta { - self.spilling_future = Some(spilling_group_by_payload( - self.ctx.clone(), - self.operator.clone(), - &self.method, - &self.location_prefix, - payload, - )?); - - return Ok(()); - } + match spilling_meta { + AggregateMeta::Spilling(payload) => { + self.spilling_future = Some(spilling_group_by_payload( + self.ctx.clone(), + self.operator.clone(), + &self.method, + &self.location_prefix, + payload, + )?); + + return Ok(()); + } + AggregateMeta::AggregateSpilling(payload) => { + self.spilling_future = Some(agg_spilling_group_by_payload::( + self.ctx.clone(), + self.operator.clone(), + &self.location_prefix, + payload, + )?); + + return Ok(()); + } - return Err(ErrorCode::Internal( - "TransformGroupBySpillWriter only recv AggregateMeta", - )); + _ => { + return Err(ErrorCode::Internal( + "TransformGroupBySpillWriter only recv AggregateMeta", + )); + } + } } Ok(()) @@ -178,6 +200,112 @@ impl Processor for TransformGroupBySpillWriter } } +pub fn agg_spilling_group_by_payload( + ctx: Arc, + operator: Operator, + location_prefix: &str, + partitioned_payload: PartitionedPayload, +) -> Result>> { + let unique_name = GlobalUniqName::unique(); + let location = format!("{}/{}", location_prefix, unique_name); + + let mut write_size = 0; + let partition_count = partitioned_payload.partition_count(); + let mut write_data = Vec::with_capacity(partition_count); + let mut spilled_buckets_payloads = Vec::with_capacity(partition_count); + let mut rows = 0; + + for (bucket, payload) in partitioned_payload.payloads.into_iter().enumerate() { + if payload.len() == 0 { + continue; + } + + let now = Instant::now(); + let data_block = payload.group_by_flush_all()?; + rows += data_block.num_rows(); + + let begin = write_size; + let columns = data_block.columns().to_vec(); + let mut columns_data = Vec::with_capacity(columns.len()); + let mut columns_layout = Vec::with_capacity(columns.len()); + for column in columns.into_iter() { + let column = column.value.as_column().unwrap(); + let column_data = serialize_column(column); + write_size += column_data.len() as u64; + columns_layout.push(column_data.len() as u64); + columns_data.push(column_data); + } + + // perf + { + metrics_inc_aggregate_spill_data_serialize_milliseconds( + now.elapsed().as_millis() as u64 + ); + } + + write_data.push(columns_data); + spilled_buckets_payloads.push(BucketSpilledPayload { + bucket: bucket as isize, + location: location.clone(), + data_range: begin..write_size, + columns_layout, + max_partition_count: partition_count, + }); + } + + Ok(Box::pin(async move { + let instant = Instant::now(); + + let mut write_bytes = 0; + if !write_data.is_empty() { + let mut writer = operator + .writer_with(&location) + .buffer(8 * 1024 * 1024) + .await?; + for write_bucket_data in write_data.into_iter() { + for data in write_bucket_data.into_iter() { + write_bytes += data.len(); + writer.write(data).await?; + } + } + + writer.close().await?; + } + + // perf + { + metrics_inc_group_by_spill_write_count(); + metrics_inc_group_by_spill_write_bytes(write_bytes as u64); + metrics_inc_group_by_spill_write_milliseconds(instant.elapsed().as_millis() as u64); + + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteCount, 1); + Profile::record_usize_profile(ProfileStatisticsName::SpillWriteBytes, write_bytes); + Profile::record_usize_profile( + ProfileStatisticsName::SpillWriteTime, + instant.elapsed().as_millis() as usize, + ); + } + + { + let progress_val = ProgressValues { + rows, + bytes: write_bytes, + }; + ctx.get_group_by_spill_progress().incr(&progress_val); + } + + info!( + "Write aggregate spill {} successfully, elapsed: {:?}", + location, + instant.elapsed() + ); + + Ok(DataBlock::empty_with_meta( + AggregateMeta::::create_spilled(spilled_buckets_payloads), + )) + })) +} + pub fn spilling_group_by_payload( ctx: Arc, operator: Operator, @@ -226,6 +354,7 @@ pub fn spilling_group_by_payload( location: location.clone(), data_range: begin..write_size, columns_layout, + max_partition_count: 0, }); } diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_spill_reader.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_spill_reader.rs index 8d69e57b0322..c2a7ac732e6f 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_spill_reader.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_spill_reader.rs @@ -138,6 +138,7 @@ impl Processor AggregateMeta::Spilling(_) => unreachable!(), AggregateMeta::AggregateHashTable(_) => unreachable!(), AggregateMeta::AggregatePayload(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), AggregateMeta::HashTable(_) => unreachable!(), AggregateMeta::Serialized(_) => unreachable!(), AggregateMeta::BucketSpilled(payload) => { @@ -181,6 +182,7 @@ impl Processor AggregateMeta::HashTable(_) => unreachable!(), AggregateMeta::AggregateHashTable(_) => unreachable!(), AggregateMeta::AggregatePayload(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), AggregateMeta::Serialized(_) => unreachable!(), AggregateMeta::BucketSpilled(payload) => { let instant = Instant::now(); @@ -300,7 +302,7 @@ impl TransformSpillReader::Serialized(SerializedPayload { bucket: payload.bucket, data_block: DataBlock::new_from_columns(columns), - max_partition_count: 0, + max_partition_count: payload.max_partition_count, }) } } diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_final.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_final.rs index 556062bb53dc..e024c86121e9 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_final.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_final.rs @@ -84,6 +84,33 @@ impl TransformFinalAggregate { self.params.aggregate_functions.clone(), HashTableConfig::default().with_initial_radix_bits(0), capacity, + Arc::new(Bump::new()), + ); + hashtable.combine_payloads(&payload, &mut self.flush_state)?; + agg_hashtable = Some(hashtable); + } + }, + AggregateMeta::Serialized(payload) => match agg_hashtable.as_mut() { + Some(ht) => { + let payload = payload.convert_to_partitioned_payload( + self.params.group_data_types.clone(), + self.params.aggregate_functions.clone(), + )?; + ht.combine_payloads(&payload, &mut self.flush_state)?; + } + None => { + let payload = payload.convert_to_partitioned_payload( + self.params.group_data_types.clone(), + self.params.aggregate_functions.clone(), + )?; + let capacity = + AggregateHashTable::get_capacity_for_count(payload.len()); + let mut hashtable = AggregateHashTable::new_with_capacity( + self.params.group_data_types.clone(), + self.params.aggregate_functions.clone(), + HashTableConfig::default().with_initial_radix_bits(0), + capacity, + Arc::new(Bump::new()), ); hashtable.combine_payloads(&payload, &mut self.flush_state)?; agg_hashtable = Some(hashtable); @@ -241,6 +268,7 @@ where Method: HashMethodBounds }, AggregateMeta::AggregateHashTable(_) => unreachable!(), AggregateMeta::AggregatePayload(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), } } diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_partial.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_partial.rs index 3094d21c28e7..75085f0400e8 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_partial.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_aggregate_partial.rs @@ -136,10 +136,12 @@ impl TransformPartialAggregate { )?), } } else { + let arena = Arc::new(Bump::new()); HashTable::AggregateHashTable(AggregateHashTable::new( params.group_data_types.clone(), params.aggregate_functions.clone(), config, + arena, )) }; @@ -383,6 +385,39 @@ impl AccumulatingTransform for TransformPartialAggrega unreachable!() } + + if matches!(&self.hash_table, HashTable::AggregateHashTable(cell) if cell.allocated_bytes() > self.settings.spilling_bytes_threshold_per_proc + || GLOBAL_MEM_STAT.get_memory_usage() as usize >= self.settings.max_memory_usage) + { + if let HashTable::AggregateHashTable(v) = std::mem::take(&mut self.hash_table) { + // perf + { + metrics_inc_aggregate_partial_spill_count(); + metrics_inc_aggregate_partial_spill_cell_count(1); + metrics_inc_aggregate_partial_hashtable_allocated_bytes( + v.allocated_bytes() as u64, + ); + } + + let group_types = v.payload.group_types.clone(); + let aggrs = v.payload.aggrs.clone(); + let config = v.config.clone(); + let blocks = vec![DataBlock::empty_with_meta( + AggregateMeta::::create_agg_spilling(v.payload), + )]; + + let arena = Arc::new(Bump::new()); + self.hash_table = HashTable::AggregateHashTable(AggregateHashTable::new( + group_types, + aggrs, + config, + arena, + )); + return Ok(blocks); + } + + unreachable!() + } } Ok(vec![]) @@ -422,8 +457,7 @@ impl AccumulatingTransform for TransformPartialAggrega HashTable::AggregateHashTable(hashtable) => { let partition_count = hashtable.payload.partition_count(); let mut blocks = Vec::with_capacity(partition_count); - for (bucket, mut payload) in hashtable.payload.payloads.into_iter().enumerate() { - payload.arenas.extend_from_slice(&hashtable.payload.arenas); + for (bucket, payload) in hashtable.payload.payloads.into_iter().enumerate() { blocks.push(DataBlock::empty_with_meta( AggregateMeta::::create_agg_payload( bucket as isize, diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_final.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_final.rs index 1e3653bee01e..6ed56f8a2b2d 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_final.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_final.rs @@ -77,6 +77,33 @@ impl TransformFinalGroupBy { self.params.aggregate_functions.clone(), HashTableConfig::default().with_initial_radix_bits(0), capacity, + Arc::new(Bump::new()), + ); + hashtable.combine_payloads(&payload, &mut self.flush_state)?; + agg_hashtable = Some(hashtable); + } + }, + AggregateMeta::Serialized(payload) => match agg_hashtable.as_mut() { + Some(ht) => { + let payload = payload.convert_to_partitioned_payload( + self.params.group_data_types.clone(), + self.params.aggregate_functions.clone(), + )?; + ht.combine_payloads(&payload, &mut self.flush_state)?; + } + None => { + let payload = payload.convert_to_partitioned_payload( + self.params.group_data_types.clone(), + self.params.aggregate_functions.clone(), + )?; + let capacity = + AggregateHashTable::get_capacity_for_count(payload.len()); + let mut hashtable = AggregateHashTable::new_with_capacity( + self.params.group_data_types.clone(), + self.params.aggregate_functions.clone(), + HashTableConfig::default().with_initial_radix_bits(0), + capacity, + Arc::new(Bump::new()), ); hashtable.combine_payloads(&payload, &mut self.flush_state)?; agg_hashtable = Some(hashtable); @@ -162,6 +189,7 @@ where Method: HashMethodBounds }, AggregateMeta::AggregateHashTable(_) => unreachable!(), AggregateMeta::AggregatePayload(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), } } diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_partial.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_partial.rs index 584e15570436..6e3253b59a04 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_partial.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_group_by_partial.rs @@ -124,10 +124,12 @@ impl TransformPartialGroupBy { let _dropper = GroupByHashTableDropper::::create(); HashTable::HashTable(HashTableCell::create(hashtable, _dropper)) } else { + let arena = Arc::new(Bump::new()); HashTable::AggregateHashTable(AggregateHashTable::new( params.group_data_types.clone(), params.aggregate_functions.clone(), config, + arena, )) }; @@ -227,6 +229,30 @@ impl AccumulatingTransform for TransformPartialGroupBy unreachable!() } + + if matches!(&self.hash_table, HashTable::AggregateHashTable(cell) if cell.allocated_bytes() > self.settings.spilling_bytes_threshold_per_proc + || GLOBAL_MEM_STAT.get_memory_usage() as usize >= self.settings.max_memory_usage) + { + if let HashTable::AggregateHashTable(v) = std::mem::take(&mut self.hash_table) { + let group_types = v.payload.group_types.clone(); + let aggrs = v.payload.aggrs.clone(); + let config = v.config.clone(); + let blocks = vec![DataBlock::empty_with_meta( + AggregateMeta::::create_agg_spilling(v.payload), + )]; + + let arena = Arc::new(Bump::new()); + self.hash_table = HashTable::AggregateHashTable(AggregateHashTable::new( + group_types, + aggrs, + config, + arena, + )); + return Ok(blocks); + } + + unreachable!() + } } } @@ -267,8 +293,7 @@ impl AccumulatingTransform for TransformPartialGroupBy HashTable::AggregateHashTable(hashtable) => { let partition_count = hashtable.payload.partition_count(); let mut blocks = Vec::with_capacity(partition_count); - for (bucket, mut payload) in hashtable.payload.payloads.into_iter().enumerate() { - payload.arenas.extend_from_slice(&hashtable.payload.arenas); + for (bucket, payload) in hashtable.payload.payloads.into_iter().enumerate() { blocks.push(DataBlock::empty_with_meta( AggregateMeta::::create_agg_payload( bucket as isize, diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_partition_bucket.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_partition_bucket.rs index 0ef1d2552989..585bdd6bb474 100644 --- a/src/query/service/src/pipelines/processors/transforms/aggregator/transform_partition_bucket.rs +++ b/src/query/service/src/pipelines/processors/transforms/aggregator/transform_partition_bucket.rs @@ -19,15 +19,13 @@ use std::marker::PhantomData; use std::mem::take; use std::sync::Arc; +use bumpalo::Bump; use databend_common_exception::ErrorCode; use databend_common_exception::Result; -use databend_common_expression::AggregateHashTable; use databend_common_expression::BlockMetaInfoDowncast; use databend_common_expression::DataBlock; -use databend_common_expression::HashTableConfig; use databend_common_expression::PartitionedPayload; use databend_common_expression::PayloadFlushState; -use databend_common_expression::ProbeState; use databend_common_hashtable::hash2bucket; use databend_common_hashtable::HashtableLike; use databend_common_pipeline_core::processors::Event; @@ -146,7 +144,7 @@ impl } let data_block = self.inputs[index].port.pull_data().unwrap()?; - self.inputs[index].bucket = self.add_bucket(data_block); + self.inputs[index].bucket = self.add_bucket(data_block)?; if self.inputs[index].bucket <= SINGLE_LEVEL_BUCKET_NUM || self.max_partition_count > 0 { @@ -158,13 +156,14 @@ impl Ok(self.initialized_all_inputs) } - fn add_bucket(&mut self, mut data_block: DataBlock) -> isize { + fn add_bucket(&mut self, mut data_block: DataBlock) -> Result { if let Some(block_meta) = data_block.get_meta() { if let Some(block_meta) = AggregateMeta::::downcast_ref_from(block_meta) { let (bucket, res) = match block_meta { AggregateMeta::Spilling(_) => unreachable!(), AggregateMeta::Partitioned { .. } => unreachable!(), AggregateMeta::AggregateHashTable(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), AggregateMeta::BucketSpilled(payload) => { (payload.bucket, SINGLE_LEVEL_BUCKET_NUM) } @@ -201,7 +200,7 @@ impl }; } - return SINGLE_LEVEL_BUCKET_NUM; + return Ok(SINGLE_LEVEL_BUCKET_NUM); } unreachable!() @@ -223,7 +222,7 @@ impl } }; - return res; + return Ok(res); } } } @@ -235,56 +234,17 @@ impl AggregateMeta::AggregatePayload(p) => { let res = p.bucket; self.agg_payloads.push(p); - res + Ok(res) } AggregateMeta::Serialized(p) => { - let rows_num = p.data_block.num_rows(); - let radix_bits = p.max_partition_count.trailing_zeros() as u64; - let config = HashTableConfig::default().with_initial_radix_bits(radix_bits); - let mut state = ProbeState::default(); - let capacity = AggregateHashTable::get_capacity_for_count(rows_num); - let mut hashtable = AggregateHashTable::new_with_capacity( - self.params.group_data_types.clone(), - self.params.aggregate_functions.clone(), - config, - capacity, - ); - hashtable.direct_append = true; - - let agg_len = self.params.aggregate_functions.len(); - let group_len = self.params.group_columns.len(); - let agg_states = (0..agg_len) - .map(|i| { - p.data_block - .get_by_offset(i) - .value - .as_column() - .unwrap() - .clone() - }) - .collect::>(); - let group_columns = (agg_len..(agg_len + group_len)) - .map(|i| { - p.data_block - .get_by_offset(i) - .value - .as_column() - .unwrap() - .clone() - }) - .collect::>(); - - let _ = hashtable - .add_groups( - &mut state, - &group_columns, - &[vec![]], - &agg_states, - rows_num, - ) - .unwrap(); - - for (bucket, payload) in hashtable.payload.payloads.into_iter().enumerate() + for (bucket, payload) in p + .convert_to_partitioned_payload( + self.params.group_data_types.clone(), + self.params.aggregate_functions.clone(), + )? + .payloads + .into_iter() + .enumerate() { self.agg_payloads.push(AggregatePayload { bucket: bucket as isize, @@ -293,16 +253,16 @@ impl }); } - p.bucket + Ok(p.bucket) } _ => unreachable!(), }; } - return 0; + return Ok(0); } self.unsplitted_blocks.push(data_block); - SINGLE_LEVEL_BUCKET_NUM + Ok(SINGLE_LEVEL_BUCKET_NUM) } fn try_push_data_block(&mut self) -> bool { @@ -468,7 +428,7 @@ impl Processor } let data_block = self.inputs[index].port.pull_data().unwrap()?; - self.inputs[index].bucket = self.add_bucket(data_block); + self.inputs[index].bucket = self.add_bucket(data_block)?; debug_assert!(self.unsplitted_blocks.is_empty()); if self.inputs[index].bucket <= self.working_bucket { @@ -511,12 +471,16 @@ impl Processor group_types.clone(), aggrs.clone(), self.max_partition_count as u64, + vec![Arc::new(Bump::new())], ); for agg_payload in self.agg_payloads.drain(0..) { - partitioned_payload - .arenas - .extend_from_slice(&agg_payload.payload.arenas); + if !partitioned_payload.include_arena(&agg_payload.payload.arena) { + partitioned_payload + .arenas + .extend_from_slice(&[agg_payload.payload.arena.clone()]); + } + if agg_payload.max_partition_count != self.max_partition_count { debug_assert!(agg_payload.max_partition_count < self.max_partition_count); partitioned_payload.combine_single(agg_payload.payload, &mut self.flush_state); @@ -527,8 +491,12 @@ impl Processor } for (bucket, payload) in partitioned_payload.payloads.into_iter().enumerate() { - let mut part = PartitionedPayload::new(group_types.clone(), aggrs.clone(), 1); - part.arenas.extend_from_slice(&partitioned_payload.arenas); + let mut part = PartitionedPayload::new( + group_types.clone(), + aggrs.clone(), + 1, + partitioned_payload.arenas.clone(), + ); part.combine_single(payload, &mut self.flush_state); if part.len() != 0 { @@ -562,6 +530,7 @@ impl Processor AggregateMeta::HashTable(payload) => self.partition_hashtable(payload)?, AggregateMeta::AggregateHashTable(_) => unreachable!(), AggregateMeta::AggregatePayload(_) => unreachable!(), + AggregateMeta::AggregateSpilling(_) => unreachable!(), }; for (bucket, block) in data_blocks.into_iter().enumerate() { diff --git a/src/query/sql/src/executor/format.rs b/src/query/sql/src/executor/format.rs index 28965dc0d549..19320a161131 100644 --- a/src/query/sql/src/executor/format.rs +++ b/src/query/sql/src/executor/format.rs @@ -575,10 +575,6 @@ fn aggregate_partial_to_format_tree( .join(", "); let mut children = vec![ - FormatTreeNode::new(format!( - "output columns: [{}]", - format_output_columns(plan.output_schema()?, metadata, true) - )), FormatTreeNode::new(format!("group by: [{group_by}]")), FormatTreeNode::new(format!("aggregate functions: [{agg_funcs}]")), ]; diff --git a/tests/sqllogictests/suites/ee/02_ee_aggregating_index/02_0000_async_agg_index_base.test b/tests/sqllogictests/suites/ee/02_ee_aggregating_index/02_0000_async_agg_index_base.test index c0e28689c35a..d152378d5cc4 100644 --- a/tests/sqllogictests/suites/ee/02_ee_aggregating_index/02_0000_async_agg_index_base.test +++ b/tests/sqllogictests/suites/ee/02_ee_aggregating_index/02_0000_async_agg_index_base.test @@ -386,7 +386,6 @@ EvalScalar ├── aggregate functions: [max(user_id), sum(id), count(id)] ├── estimated rows: 8.00 └── AggregatePartial - ├── output columns: [max(user_id) (#3), sum(id) (#4), count(id) (#5), #_group_by_key] ├── group by: [event_name, user_id] ├── aggregate functions: [max(user_id), sum(id), count(id)] ├── estimated rows: 8.00 @@ -427,7 +426,6 @@ EvalScalar ├── aggregate functions: [max(user_id), sum(id), count(id)] ├── estimated rows: 7.11 └── AggregatePartial - ├── output columns: [max(user_id) (#3), sum(id) (#4), count(id) (#5), #_group_by_key] ├── group by: [event_name, user_id] ├── aggregate functions: [max(user_id), sum(id), count(id)] ├── estimated rows: 7.11 @@ -474,7 +472,6 @@ EvalScalar ├── aggregate functions: [max(user_id), sum(id), count(id)] ├── estimated rows: 7.11 └── AggregatePartial - ├── output columns: [max(user_id) (#3), sum(id) (#4), count(id) (#5), #_group_by_key] ├── group by: [event_name, user_id] ├── aggregate functions: [max(user_id), sum(id), count(id)] ├── estimated rows: 7.11 diff --git a/tests/sqllogictests/suites/mode/cluster/exchange.test b/tests/sqllogictests/suites/mode/cluster/exchange.test index abefb62726df..b1fa94daaa0a 100644 --- a/tests/sqllogictests/suites/mode/cluster/exchange.test +++ b/tests/sqllogictests/suites/mode/cluster/exchange.test @@ -170,7 +170,6 @@ Exchange │ ├── output columns: [sum(number) (#2), #_group_by_key] │ ├── exchange type: Hash(_group_by_key) │ └── AggregatePartial - │ ├── output columns: [sum(number) (#2), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [sum(number)] │ ├── estimated rows: 1.00 @@ -202,7 +201,6 @@ Fragment 0: ├── output columns: [sum(number) (#2), #_group_by_key] ├── destination fragment: [1] └── AggregatePartial - ├── output columns: [sum(number) (#2), #_group_by_key] ├── group by: [number] ├── aggregate functions: [sum(number)] ├── estimated rows: 1.00 @@ -285,7 +283,6 @@ AggregateFinal ├── output columns: [sum(a) (#4)] ├── exchange type: Merge └── AggregatePartial - ├── output columns: [sum(a) (#4)] ├── group by: [] ├── aggregate functions: [sum(number)] ├── estimated rows: 1.00 @@ -368,7 +365,6 @@ AggregateFinal ├── aggregate functions: [sum(a)] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [sum(a) (#6)] ├── group by: [] ├── aggregate functions: [sum(a)] ├── estimated rows: 1.00 @@ -419,7 +415,6 @@ AggregateFinal ├── output columns: [sum(t1.a) (#4)] ├── exchange type: Merge └── AggregatePartial - ├── output columns: [sum(t1.a) (#4)] ├── group by: [] ├── aggregate functions: [sum(number)] ├── estimated rows: 1.00 @@ -473,7 +468,6 @@ AggregateFinal ├── output columns: [sum(a) (#3)] ├── exchange type: Merge └── AggregatePartial - ├── output columns: [sum(a) (#3)] ├── group by: [] ├── aggregate functions: [sum(number)] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/cluster/explain_v2.test b/tests/sqllogictests/suites/mode/cluster/explain_v2.test index 1960044971f9..7708ea83dc06 100644 --- a/tests/sqllogictests/suites/mode/cluster/explain_v2.test +++ b/tests/sqllogictests/suites/mode/cluster/explain_v2.test @@ -186,7 +186,6 @@ Limit ├── output columns: [count(1) (#2), max(a) (#4)] ├── exchange type: Merge └── AggregatePartial - ├── output columns: [count(1) (#2), max(a) (#4)] ├── group by: [] ├── aggregate functions: [count(), max(a)] ├── estimated rows: 1.00 @@ -339,7 +338,6 @@ AggregateFinal ├── output columns: [sum(a) (#2)] ├── exchange type: Merge └── AggregatePartial - ├── output columns: [sum(a) (#2)] ├── group by: [] ├── aggregate functions: [sum(number)] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/cluster/group_shuffle.test b/tests/sqllogictests/suites/mode/cluster/group_shuffle.test index ab6149d1b812..55535eaf1d65 100644 --- a/tests/sqllogictests/suites/mode/cluster/group_shuffle.test +++ b/tests/sqllogictests/suites/mode/cluster/group_shuffle.test @@ -13,7 +13,6 @@ Exchange ├── aggregate functions: [] ├── estimated rows: 100000.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 100000.00 @@ -49,7 +48,6 @@ Exchange ├── output columns: [#_group_by_key] ├── exchange type: Hash(_group_by_key) └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 100000.00 diff --git a/tests/sqllogictests/suites/mode/standalone/ee/explain_agg_index.test b/tests/sqllogictests/suites/mode/standalone/ee/explain_agg_index.test index 699468527146..6cedb6329335 100644 --- a/tests/sqllogictests/suites/mode/standalone/ee/explain_agg_index.test +++ b/tests/sqllogictests/suites/mode/standalone/ee/explain_agg_index.test @@ -42,7 +42,6 @@ AggregateFinal ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [SUM(a) (#2), #_group_by_key] ├── group by: [b] ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 @@ -71,7 +70,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [b] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -104,7 +102,6 @@ EvalScalar ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [SUM(a) (#2), #_group_by_key] ├── group by: [b] ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 @@ -137,7 +134,6 @@ EvalScalar ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [SUM(a) (#2), #_group_by_key] ├── group by: [b] ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 @@ -177,7 +173,6 @@ Filter │ ├── aggregate functions: [sum(a)] │ ├── estimated rows: 0.00 │ └── AggregatePartial - │ ├── output columns: [SUM(a) (#5), #_group_by_key] │ ├── group by: [b] │ ├── aggregate functions: [sum(a)] │ ├── estimated rows: 0.00 @@ -202,7 +197,6 @@ Filter ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [SUM(a) (#2), #_group_by_key] ├── group by: [b] ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 @@ -229,7 +223,6 @@ AggregateFinal ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [SUM(a) (#2), #_group_by_key] ├── group by: [b] ├── aggregate functions: [sum(a)] ├── estimated rows: 0.00 @@ -269,7 +262,6 @@ EvalScalar ├── aggregate functions: [sum(sum_arg_0), count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [sum((a + 1)) (#3), count((a + 1)) (#4)] ├── group by: [] ├── aggregate functions: [sum(sum_arg_0), count()] ├── estimated rows: 1.00 @@ -308,7 +300,6 @@ EvalScalar ├── aggregate functions: [sum(a), count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [sum(a) (#2), count(a) (#3)] ├── group by: [] ├── aggregate functions: [sum(a), count()] ├── estimated rows: 1.00 @@ -346,7 +337,6 @@ EvalScalar ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -385,7 +375,6 @@ EvalScalar ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -424,7 +413,6 @@ Sort ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -463,7 +451,6 @@ Sort ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -503,7 +490,6 @@ Sort ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name, measurement] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -547,7 +533,6 @@ Sort ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -591,7 +576,6 @@ Sort ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name, measurement] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -635,7 +619,6 @@ Sort ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [MIN(measurement) (#5), sum(measurement) (#6), count(measurement) (#7), MAX(measurement) (#8), #_group_by_key] ├── group by: [station_name] ├── aggregate functions: [min(measurement), sum(measurement), count(measurement), max(measurement)] ├── estimated rows: 0.00 @@ -670,7 +653,6 @@ AggregateFinal ├── aggregate functions: [count(measurement)] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count(measurement) (#2)] ├── group by: [] ├── aggregate functions: [count(measurement)] ├── estimated rows: 1.00 @@ -705,7 +687,6 @@ AggregateFinal ├── aggregate functions: [min(a), max(a), count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [min(a) (#2), max(a) (#3), COUNT(*) (#4), #_group_by_key] ├── group by: [b] ├── aggregate functions: [min(a), max(a), count()] ├── estimated rows: 0.00 @@ -746,7 +727,6 @@ EvalScalar ├── aggregate functions: [sum(sum_arg_0), count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [sum((a + 1)) (#3), count((a + 1)) (#4)] ├── group by: [] ├── aggregate functions: [sum(sum_arg_0), count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/aggregate.test b/tests/sqllogictests/suites/mode/standalone/explain/aggregate.test index 07efdaa8214a..cfe54466d471 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/aggregate.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/aggregate.test @@ -7,7 +7,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 10.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 10.00 @@ -31,7 +30,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 10.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 10.00 @@ -78,7 +76,6 @@ EvalScalar ├── aggregate functions: [] ├── estimated rows: 10.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 10.00 @@ -106,7 +103,6 @@ EvalScalar ├── aggregate functions: [count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [count(3) (#13), #_group_by_key] ├── group by: [name, type] ├── aggregate functions: [count()] ├── estimated rows: 0.00 @@ -133,7 +129,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -160,7 +155,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -187,7 +181,6 @@ AggregateFinal ├── aggregate functions: [max(b)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [max(b) (#2), #_group_by_key] ├── group by: [a] ├── aggregate functions: [max(b)] ├── estimated rows: 0.00 @@ -222,7 +215,6 @@ EvalScalar ├── aggregate functions: [sum(b), count(), max(b)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), max(b) (#5), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count(), max(b)] ├── estimated rows: 0.00 @@ -261,7 +253,6 @@ EvalScalar ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 @@ -297,7 +288,6 @@ EvalScalar ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 @@ -332,7 +322,6 @@ EvalScalar ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 @@ -367,7 +356,6 @@ EvalScalar ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 @@ -402,7 +390,6 @@ EvalScalar ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count()] ├── estimated rows: 0.00 @@ -440,7 +427,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count() (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/explain.test b/tests/sqllogictests/suites/mode/standalone/explain/explain.test index 47ac380d5960..a2a3ca00ed47 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/explain.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/explain.test @@ -866,7 +866,7 @@ Limit ├── push downs: [filters: [t2.a (#2) > 2 OR t2.b (#3) < 4], limit: NONE] └── estimated rows: 5.00 -query +query explain select * from t1,t2 where (t1.a > 1 or t1.b < 2) and (t1.a > 1 or t1.b < 2) ---- HashJoin @@ -901,7 +901,7 @@ HashJoin ├── push downs: [filters: [], limit: NONE] └── estimated rows: 5.00 -query +query explain select count(distinct a) from t1; ---- AggregateFinal @@ -910,7 +910,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count(_1) (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 @@ -920,7 +919,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -935,7 +933,7 @@ AggregateFinal ├── push downs: [filters: [], limit: NONE] └── estimated rows: 1.00 -query +query explain select count_distinct(a) from t1; ---- AggregateFinal @@ -944,7 +942,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count(_1) (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 @@ -954,7 +951,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -969,7 +965,7 @@ AggregateFinal ├── push downs: [filters: [], limit: NONE] └── estimated rows: 1.00 -query +query explain select * from (values(1, 'a'),(2, 'b')) t(c1,c2) ---- ConstantTableScan @@ -983,28 +979,28 @@ drop table t1 statement ok drop table t2 -query +query explain syntax select * from read_parquet('p1', 'p2', 'p3'); ---- SELECT * FROM read_parquet('p1', 'p2', 'p3') -query +query explain syntax select * from read_parquet(prune_page=>true, refresh_meta_cache=>true); ---- SELECT * FROM read_parquet(prune_page=>TRUE, refresh_meta_cache=>TRUE) -query +query explain syntax select * from read_parquet('p1', 'p2', 'p3', prune_page=>true, refresh_meta_cache=>true); ---- SELECT * FROM read_parquet('p1', 'p2', 'p3', prune_page=>TRUE, refresh_meta_cache=>TRUE) -query +query explain ast select * from read_parquet('p1', 'p2', 'p3', prune_page=>true, refresh_meta_cache=>true); ---- Query (children 1) @@ -1026,7 +1022,7 @@ drop table if exists t4 statement ok create table t4(a int, b string); -query +query explain select * from t4 where a = 1 and try_cast(get(try_parse_json(b),'bb') as varchar) = 'xx'; ---- Filter @@ -1049,7 +1045,7 @@ drop view if exists v4 statement ok create view v4 as select a as a, try_cast(get(try_parse_json(b), 'bb') as varchar) as b from t4; -query +query explain select * from v4 where b = 'xx'; ---- EvalScalar @@ -1313,7 +1309,6 @@ HashJoin ├── aggregate functions: [] ├── estimated rows: 1294.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [col0] ├── aggregate functions: [] ├── estimated rows: 1294.00 @@ -1351,7 +1346,6 @@ Filter ├── aggregate functions: [] ├── estimated rows: 1294.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [col0] ├── aggregate functions: [] ├── estimated rows: 1294.00 @@ -1626,7 +1620,6 @@ Filter │ ├── aggregate functions: [count(a)] │ ├── estimated rows: 0.00 │ └── AggregatePartial - │ ├── output columns: [count(a) (#6), #_group_by_key] │ ├── group by: [c, a] │ ├── aggregate functions: [count(a)] │ ├── estimated rows: 0.00 @@ -1681,7 +1674,6 @@ Filter │ ├── aggregate functions: [count()] │ ├── estimated rows: 0.00 │ └── AggregatePartial - │ ├── output columns: [COUNT(*) (#6), #_group_by_key] │ ├── group by: [c, a] │ ├── aggregate functions: [count()] │ ├── estimated rows: 0.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/explain_grouping_sets.test b/tests/sqllogictests/suites/mode/standalone/explain/explain_grouping_sets.test index e378af9d4f79..bd0b11ab01db 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/explain_grouping_sets.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/explain_grouping_sets.test @@ -11,7 +11,6 @@ EvalScalar ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a, b, c, _grouping_id] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -46,7 +45,6 @@ EvalScalar ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a, b, c, _grouping_id] ├── aggregate functions: [] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/fold_count.test b/tests/sqllogictests/suites/mode/standalone/explain/fold_count.test index 4ef08cea2112..9c5ee96a5490 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/fold_count.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/fold_count.test @@ -34,7 +34,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [COUNT(*) (#1)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 @@ -62,7 +61,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1000.00 └── AggregatePartial - ├── output columns: [COUNT(*) (#1), #_group_by_key] ├── group by: [number] ├── aggregate functions: [count()] ├── estimated rows: 1000.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/mark.test b/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/mark.test index 2722382e60e1..cb71ec673453 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/mark.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/mark.test @@ -115,7 +115,6 @@ EvalScalar │ │ │ ├── aggregate functions: [] │ │ │ ├── estimated rows: 0.00 │ │ │ └── AggregatePartial - │ │ │ ├── output columns: [#_group_by_key] │ │ │ ├── group by: [c] │ │ │ ├── aggregate functions: [] │ │ │ ├── estimated rows: 0.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/lateral.test b/tests/sqllogictests/suites/mode/standalone/explain/lateral.test index e4466bada019..9232f46356f7 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/lateral.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/lateral.test @@ -54,7 +54,6 @@ HashJoin │ │ ├── aggregate functions: [] │ │ ├── estimated rows: 0.00 │ │ └── AggregatePartial -│ │ ├── output columns: [#_group_by_key] │ │ ├── group by: [number] │ │ ├── aggregate functions: [] │ │ ├── estimated rows: 0.00 @@ -113,7 +112,6 @@ HashJoin │ │ ├── aggregate functions: [] │ │ ├── estimated rows: 0.00 │ │ └── AggregatePartial -│ │ ├── output columns: [#_group_by_key] │ │ ├── group by: [number] │ │ ├── aggregate functions: [] │ │ ├── estimated rows: 0.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/limit.test b/tests/sqllogictests/suites/mode/standalone/explain/limit.test index 12c1a26d4a41..4311fbfdf7ec 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/limit.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/limit.test @@ -91,7 +91,6 @@ Limit ├── aggregate functions: [] ├── estimated rows: 0.20 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 0.20 @@ -112,7 +111,6 @@ Limit │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [COUNT(*) (#4), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -192,7 +190,6 @@ Limit │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [count(t1.number) (#2), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -211,7 +208,6 @@ Limit ├── aggregate functions: [count()] ├── estimated rows: 2.00 └── AggregatePartial - ├── output columns: [count(t.number) (#5), #_group_by_key] ├── group by: [number] ├── aggregate functions: [count()] ├── estimated rows: 2.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/project_set.test b/tests/sqllogictests/suites/mode/standalone/explain/project_set.test index d1fab8567732..839152101083 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/project_set.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/project_set.test @@ -19,7 +19,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count() (#4)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/prune_column.test b/tests/sqllogictests/suites/mode/standalone/explain/prune_column.test index 02e89b358f33..8d26af19ccb6 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/prune_column.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/prune_column.test @@ -20,7 +20,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -52,7 +51,6 @@ Limit ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number, number, number, number] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -137,7 +135,6 @@ HashJoin │ ├── limit: 1 │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [_count_scalar_subquery (#13), _any_scalar_subquery (#14)] │ ├── group by: [] │ ├── aggregate functions: [count(), any(COUNT(*))] │ ├── estimated rows: 1.00 @@ -147,7 +144,6 @@ HashJoin │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [COUNT(*) (#12)] │ ├── group by: [] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -239,7 +235,6 @@ HashJoin │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [count(*) (#2)] │ ├── group by: [] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -286,7 +281,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [COUNT(*) (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/push_down_filter.test b/tests/sqllogictests/suites/mode/standalone/explain/push_down_filter.test index 9a26f18a33d6..71a345670f3d 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/push_down_filter.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/push_down_filter.test @@ -137,7 +137,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [id] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -154,7 +153,6 @@ AggregateFinal │ ├── aggregate functions: [sum(sum(coalesce(t3.val, 0)))] │ ├── estimated rows: 0.00 │ └── AggregatePartial - │ ├── output columns: [sum(tb.de) (#7), #_group_by_key] │ ├── group by: [id] │ ├── aggregate functions: [sum(sum(coalesce(t3.val, 0)))] │ ├── estimated rows: 0.00 @@ -171,7 +169,6 @@ AggregateFinal │ │ ├── aggregate functions: [sum(sum_arg_0)] │ │ ├── estimated rows: 0.00 │ │ └── AggregatePartial - │ │ ├── output columns: [sum(coalesce(t3.val, 0)) (#5), #_group_by_key] │ │ ├── group by: [sid] │ │ ├── aggregate functions: [sum(sum_arg_0)] │ │ ├── estimated rows: 0.00 @@ -215,7 +212,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [id] ├── aggregate functions: [] ├── estimated rows: 0.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain/subquery.test b/tests/sqllogictests/suites/mode/standalone/explain/subquery.test index c292637b56df..c863fa595f64 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/subquery.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/subquery.test @@ -18,7 +18,6 @@ Filter │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [COUNT(*) (#4), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -126,7 +125,6 @@ HashJoin │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [count(*) (#2)] │ ├── group by: [] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -184,7 +182,6 @@ HashJoin │ ├── limit: 1 │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [_count_scalar_subquery (#2), _any_scalar_subquery (#3)] │ ├── group by: [] │ ├── aggregate functions: [count(), any(number)] │ ├── estimated rows: 1.00 @@ -417,7 +414,6 @@ Filter │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [COUNT(*) (#3), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/aggregate.test b/tests/sqllogictests/suites/mode/standalone/explain_native/aggregate.test index 6f66bf98afec..7606c8f7bccc 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/aggregate.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/aggregate.test @@ -7,7 +7,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 10.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 10.00 @@ -31,7 +30,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 10.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 10.00 @@ -78,7 +76,6 @@ EvalScalar ├── aggregate functions: [] ├── estimated rows: 10.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 10.00 @@ -106,7 +103,6 @@ EvalScalar ├── aggregate functions: [count()] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [count(3) (#13), #_group_by_key] ├── group by: [name, type] ├── aggregate functions: [count()] ├── estimated rows: 0.00 @@ -133,7 +129,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -156,7 +151,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -179,7 +173,6 @@ AggregateFinal ├── aggregate functions: [max(b)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [max(b) (#2), #_group_by_key] ├── group by: [a] ├── aggregate functions: [max(b)] ├── estimated rows: 0.00 @@ -210,7 +203,6 @@ EvalScalar ├── aggregate functions: [sum(b), count(), max(b)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), max(b) (#5), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count(), max(b)] ├── estimated rows: 0.00 @@ -241,7 +233,6 @@ EvalScalar ├── aggregate functions: [sum(b), count(), max(b)] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [sum(b) (#2), count(b) (#3), max(b) (#5), #_group_by_key] ├── group by: [a] ├── aggregate functions: [sum(b), count(), max(b)] ├── estimated rows: 0.00 @@ -279,7 +270,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count() (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/explain.test b/tests/sqllogictests/suites/mode/standalone/explain_native/explain.test index a83fca002c4b..a6f228fc7936 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/explain.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/explain.test @@ -862,7 +862,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count(_1) (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 @@ -872,7 +871,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -896,7 +894,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count(_1) (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 @@ -906,7 +903,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -921,7 +917,7 @@ AggregateFinal ├── push downs: [filters: [], limit: NONE] └── estimated rows: 1.00 -query +query explain select * from (values(1, 'a'),(2, 'b')) t(c1,c2) ---- ConstantTableScan diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/explain_grouping_sets.test b/tests/sqllogictests/suites/mode/standalone/explain_native/explain_grouping_sets.test index e378af9d4f79..bd0b11ab01db 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/explain_grouping_sets.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/explain_grouping_sets.test @@ -11,7 +11,6 @@ EvalScalar ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a, b, c, _grouping_id] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -46,7 +45,6 @@ EvalScalar ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [a, b, c, _grouping_id] ├── aggregate functions: [] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/fold_count.test b/tests/sqllogictests/suites/mode/standalone/explain_native/fold_count.test index ed75e0538f98..d321ac50bad9 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/fold_count.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/fold_count.test @@ -34,7 +34,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [COUNT(*) (#1)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/limit.test b/tests/sqllogictests/suites/mode/standalone/explain_native/limit.test index 12c1a26d4a41..4311fbfdf7ec 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/limit.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/limit.test @@ -91,7 +91,6 @@ Limit ├── aggregate functions: [] ├── estimated rows: 0.20 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 0.20 @@ -112,7 +111,6 @@ Limit │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [COUNT(*) (#4), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -192,7 +190,6 @@ Limit │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [count(t1.number) (#2), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -211,7 +208,6 @@ Limit ├── aggregate functions: [count()] ├── estimated rows: 2.00 └── AggregatePartial - ├── output columns: [count(t.number) (#5), #_group_by_key] ├── group by: [number] ├── aggregate functions: [count()] ├── estimated rows: 2.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/project_set.test b/tests/sqllogictests/suites/mode/standalone/explain_native/project_set.test index 59de879ece6a..1cdf3e351e22 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/project_set.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/project_set.test @@ -19,7 +19,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [count() (#4)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/prune_column.test b/tests/sqllogictests/suites/mode/standalone/explain_native/prune_column.test index cbe6ae5d4af1..1648c00bf518 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/prune_column.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/prune_column.test @@ -20,7 +20,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number] ├── aggregate functions: [] ├── estimated rows: 1.00 @@ -52,7 +51,6 @@ Limit ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [number, number, number, number] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -137,7 +135,6 @@ HashJoin │ ├── limit: 1 │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [_count_scalar_subquery (#13), _any_scalar_subquery (#14)] │ ├── group by: [] │ ├── aggregate functions: [count(), any(COUNT(*))] │ ├── estimated rows: 1.00 @@ -147,7 +144,6 @@ HashJoin │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [COUNT(*) (#12)] │ ├── group by: [] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -239,7 +235,6 @@ HashJoin │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [count(*) (#2)] │ ├── group by: [] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -286,7 +281,6 @@ AggregateFinal ├── aggregate functions: [count()] ├── estimated rows: 1.00 └── AggregatePartial - ├── output columns: [COUNT(*) (#2)] ├── group by: [] ├── aggregate functions: [count()] ├── estimated rows: 1.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/push_down_filter.test b/tests/sqllogictests/suites/mode/standalone/explain_native/push_down_filter.test index 22648067dc47..f43c65eff041 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/push_down_filter.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/push_down_filter.test @@ -129,7 +129,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [id] ├── aggregate functions: [] ├── estimated rows: 0.00 @@ -146,7 +145,6 @@ AggregateFinal │ ├── aggregate functions: [sum(sum(coalesce(t3.val, 0)))] │ ├── estimated rows: 0.00 │ └── AggregatePartial - │ ├── output columns: [sum(tb.de) (#7), #_group_by_key] │ ├── group by: [id] │ ├── aggregate functions: [sum(sum(coalesce(t3.val, 0)))] │ ├── estimated rows: 0.00 @@ -163,7 +161,6 @@ AggregateFinal │ │ ├── aggregate functions: [sum(sum_arg_0)] │ │ ├── estimated rows: 0.00 │ │ └── AggregatePartial - │ │ ├── output columns: [sum(coalesce(t3.val, 0)) (#5), #_group_by_key] │ │ ├── group by: [sid] │ │ ├── aggregate functions: [sum(sum_arg_0)] │ │ ├── estimated rows: 0.00 @@ -203,7 +200,6 @@ AggregateFinal ├── aggregate functions: [] ├── estimated rows: 0.00 └── AggregatePartial - ├── output columns: [#_group_by_key] ├── group by: [id] ├── aggregate functions: [] ├── estimated rows: 0.00 diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/subquery.test b/tests/sqllogictests/suites/mode/standalone/explain_native/subquery.test index 23a0fbaa259e..6b60f5c5565c 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain_native/subquery.test +++ b/tests/sqllogictests/suites/mode/standalone/explain_native/subquery.test @@ -18,7 +18,6 @@ Filter │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [COUNT(*) (#4), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -126,7 +125,6 @@ HashJoin │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [count(*) (#2)] │ ├── group by: [] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 @@ -184,7 +182,6 @@ HashJoin │ ├── limit: 1 │ ├── estimated rows: 1.00 │ └── AggregatePartial -│ ├── output columns: [_count_scalar_subquery (#2), _any_scalar_subquery (#3)] │ ├── group by: [] │ ├── aggregate functions: [count(), any(number)] │ ├── estimated rows: 1.00 @@ -417,7 +414,6 @@ Filter │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00 │ └── AggregatePartial - │ ├── output columns: [COUNT(*) (#3), #_group_by_key] │ ├── group by: [number] │ ├── aggregate functions: [count()] │ ├── estimated rows: 1.00