Rachelint commented on code in PR #15591: URL: https://github.com/apache/datafusion/pull/15591#discussion_r3353185195
########## datafusion/physical-plan/src/aggregates/row_hash.rs: ########## Review Comment: Serialize and spill the batch to disk. ########## datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/blocks.rs: ########## @@ -0,0 +1,387 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Aggregation intermediate results blocks in blocked approach + +use std::{ + fmt::Debug, + iter, mem, + ops::{Index, IndexMut}, +}; + +use datafusion_expr_common::groups_accumulator::EmitTo; + +/// Structure used to store aggregation intermediate results in `blocked approach` +/// +/// Aggregation intermediate results will be stored as multiple [`Block`]s +/// (simply you can think a [`Block`] as a `Vec`). And `Blocks` is the structure +/// to represent such multiple [`Block`]s. +/// +/// Internally uses a `Vec<B>` with a `start` offset to track the first active +/// block. When blocks are popped via [`Self::pop_block`], the block is swapped +/// out in O(1) using `mem::replace` and the `start` cursor advances, avoiding +/// the O(n) shift cost of `Vec::remove(0)`. Review Comment: > This avoids the O(n) shift but it doesn't free the memory, I'm wondering whether a VecDeque would work better here, so inner: VecDeque instead of Vec The memory usage of `Vec` may not be large (8bytes(pointer) * num_blocks). But index op of `VecDeque ` is very expansive compared with `Vec`, so I think it may be worth to waste few memory to get better performance? ########## datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/block_store/vec_block_store.rs: ########## @@ -0,0 +1,298 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! [`BlockStore`] wrapper specialized for stores whose blocks are `Vec<T>`. + +use std::fmt::Debug; +use std::iter; +use std::marker::PhantomData; +use std::ops::{Index, IndexMut}; + +use datafusion_common::utils::split_vec_min_alloc; +use datafusion_common::{Result, internal_datafusion_err, internal_err}; +use datafusion_expr_common::groups_accumulator::EmitTo; + +use crate::aggregate::groups_accumulator::block_store::{Block, BlockStore}; + +/// Thin wrapper around a [`BlockStore<Vec<T>>`] that adds vector-specific +/// emit semantics on top of the generic block store API. +/// +/// Emitting `EmitTo::First` requires vector-specific split semantics, so it +/// lives on this wrapper rather than on [`BlockStore`] itself. The wrapper +/// re-exposes every [`BlockStore`] method by delegation, plus an [`emit`] +/// method implemented purely via [`BlockStore::push_block`] / +/// [`BlockStore::pop_block`] so it works uniformly over flat and blocked +/// storage. +/// +/// [`emit`]: VecBlockStore::emit +#[derive(Debug)] +pub struct VecBlockStore<T, S> +where + T: Clone + Debug, + S: BlockStore<Vec<T>>, +{ + inner: S, + _phantom: PhantomData<T>, +} + +impl<T, S> VecBlockStore<T, S> +where + T: Clone + Debug, + S: BlockStore<Vec<T>>, +{ + /// Wrap an existing block store. + pub fn new(inner: S) -> Self { + Self { + inner, + _phantom: PhantomData, + } + } + + // ---- BlockStore method delegation ---------------------------------- + pub fn reserve_blocks(&mut self) { + self.inner.reserve_blocks(); + } + + pub fn resize(&mut self, total_num_groups: usize, default_value: T) { + self.inner.resize(total_num_groups, default_value); + } + + pub fn num_blocks(&self) -> usize { + self.inner.num_blocks() + } + + pub fn block_size(&self) -> Option<usize> { + self.inner.block_size() + } + + pub fn clear(&mut self) { + self.inner.clear(); + } + + // ---- Emit ---------------------------------------------------------- + + /// Emit values according to `emit_to`, expressed only in terms of + /// [`BlockStore::push_block`] and [`BlockStore::pop_block`]. + /// + /// - [`EmitTo::All`]: drains every block via repeated `pop_block` and + /// concatenates the results into a single `Vec<T>`. + /// - [`EmitTo::First`]`(n)`: pops the first block, splits off the first + /// `n` values, and pushes the remainder back. Only meaningful when the + /// first block holds at least `n` values (true for flat storage); the + /// call returns an internal error otherwise. + /// - [`EmitTo::NextBlock`]: pops a single block. + pub fn emit(&mut self, emit_to: EmitTo) -> Result<Vec<T>> { + match emit_to { + EmitTo::All => self.inner.pop_block().ok_or_else(|| { Review Comment: Maybe a bit confusing now due to: - `EmitTo::First` and `EmitTo::All` only meaningful in `flat mode` - `EmitTo::NextBlock` only meaningful in `blocked mode` Here (`vec_block_store.rs`) maybe more suitable to return error when encountering `EmitTo::First` and `EmitTo::All`? ########## datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/block_store/mod.rs: ########## @@ -0,0 +1,60 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Storage abstraction for aggregation intermediate result blocks. + +use std::fmt::Debug; +use std::ops::{Index, IndexMut}; + +use crate::aggregate::groups_accumulator::blocks::Block; + +pub mod blocked; +pub mod flat; +pub mod vec_values; + +pub use blocked::BlockedBlockStore; +pub use flat::FlatBlockStore; +pub use vec_values::{VecValues, VecValuesBlockStore}; + +/// Storage abstraction for aggregation intermediate result blocks. +/// +/// [`BlockStore`] lets flat and blocked group state share the same accumulation +/// flow while using different physical layouts. Implementations should keep +/// block lookup cheap because it is used by per-row accumulator update paths. +pub trait BlockStore<B: Block>: Review Comment: @Dandandan @ariel-miculas @2010YOUY01 @alamb I think the greatest difficulty currently being faced is that how to abstract the common trait (BlockStore here) to help conveniently impl blocked accumulator and group values Current one can work and zero cost when we disable blocked mode, but I think it seems too complex... ########## datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/block_store/blocked.rs: ########## @@ -0,0 +1,436 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Blocked [`BlockStore`] implementation. + +use std::{ + fmt::Debug, + mem, + ops::{Index, IndexMut}, +}; + +use crate::aggregate::groups_accumulator::block_store::{Block, BlockStore}; + +/// Structure used to store aggregation intermediate results in `blocked approach` +/// +/// Aggregation intermediate results will be stored as multiple [`Block`]s +/// (simply you can think a [`Block`] as a `Vec`). And `Blocks` is the structure +/// to represent such multiple [`Block`]s. +/// +/// Blocks are popped in FIFO order by keeping a cursor into `inner`. Popping a +/// block swaps it out in O(1) using `mem::take` and advances the cursor, +/// avoiding the O(n) shift cost of `Vec::remove(0)`. +/// +/// More details about `blocked approach` can see in: [`GroupsAccumulator::supports_blocked_groups`]. +/// +/// [`GroupsAccumulator::supports_blocked_groups`]: datafusion_expr_common::groups_accumulator::GroupsAccumulator::supports_blocked_groups +/// +#[derive(Debug)] +pub struct BlockedBlockStore<B: Block> { + inner: Vec<B>, + /// Index of the next active block. + cursor: usize, + block_size: usize, +} + +impl<B: Block> BlockedBlockStore<B> { + /// Create a new blocked store with the given fixed block size. + pub fn new(block_size: usize) -> Self { + Self { + inner: Vec::new(), + cursor: 0, + block_size, + } + } +} + +impl<B: Block> BlockStore<B> for BlockedBlockStore<B> { + fn push_block(&mut self, block: B) { + self.inner.push(block); + } + + fn pop_block(&mut self) -> Option<B> { + if self.cursor >= self.inner.len() { + return None; + } + + let block = mem::take(&mut self.inner[self.cursor]); Review Comment: `VecDeque` is used at the beginning, but the `index op` in `VecDeque` actually not really trivial as `Vec`, and switch to `Vec` later suggested by reviews. But I agree with that `VecDeque` may be more suitable, due to: - its main target is saving memory (although Vec<Empty block> will not occupy much) - the logic can become simple, and I think as a initial pr, it should keep as simple as possible. ########## datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/block_store/blocked.rs: ########## @@ -0,0 +1,436 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Blocked [`BlockStore`] implementation. + +use std::{ + fmt::Debug, + mem, + ops::{Index, IndexMut}, +}; + +use crate::aggregate::groups_accumulator::block_store::{Block, BlockStore}; + +/// Structure used to store aggregation intermediate results in `blocked approach` +/// +/// Aggregation intermediate results will be stored as multiple [`Block`]s +/// (simply you can think a [`Block`] as a `Vec`). And `Blocks` is the structure +/// to represent such multiple [`Block`]s. +/// +/// Blocks are popped in FIFO order by keeping a cursor into `inner`. Popping a +/// block swaps it out in O(1) using `mem::take` and advances the cursor, +/// avoiding the O(n) shift cost of `Vec::remove(0)`. +/// +/// More details about `blocked approach` can see in: [`GroupsAccumulator::supports_blocked_groups`]. +/// +/// [`GroupsAccumulator::supports_blocked_groups`]: datafusion_expr_common::groups_accumulator::GroupsAccumulator::supports_blocked_groups +/// +#[derive(Debug)] +pub struct BlockedBlockStore<B: Block> { Review Comment: Methods in `Block` trait is mainly abstracted from logic in `prim_op.rs`, and may only useful for `accumulators`. Maybe we should define some trait different for `groups` case? For me, how to define the suitable traits is actually a main block point... -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
