alamb commented on code in PR #25123:
URL: https://github.com/apache/datafusion/pull/25123#discussion_r4066394780
##########
benchmarks/queries/clickbench/README.md:
##########
@@ -301,6 +300,46 @@ LIMIT 10;
+### Q15: Correlation of ad slot dimensions per user
+
+**Question**: "How correlated are the width and height of the screens each user
+browses from?"
+
+**Important Query Properties**: `covar_samp` over a very high cardinality
+`GROUP BY` (about 17.6M distinct `UserID`s). `covar_samp` has no native
Review Comment:
I think the detail about the "no native accumulator" is an implementation
detail that is not an important query property. I recommend removing everything
after "`covar_samp` has no native accumulator" and then just mention Q16 has
the low cardinality version of this
##########
datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs:
##########
@@ -235,14 +235,23 @@ impl GroupsAccumulatorAdapter {
assert_eq!(values[0].len(), group_indices.len());
+ // groups_with_rows holds a list of group indexes that have any rows
+ // that need to be accumulated, stored in order of first appearance in
+ // this batch
+ let mut groups_with_rows = vec![];
Review Comment:
You could potentially avoid this allocation on each call to
`invoke_per_accumulator` by storing it as a field on self
So something like
```rust
let groups_with_rows = &mut self.scratch;
groups_with_rows.clear();
```
##########
datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs:
##########
@@ -380,6 +379,16 @@ impl GroupsAccumulatorAdapter {
})();
self.adjust_allocation(sizes_pre, sizes_post);
+
+ // An error leaves the scratch indexes of the groups after the failed
+ // one in place. The next batch finds the groups it reaches by an empty
+ // `indices`, so clear them. `clear` keeps the capacity, thus the
Review Comment:
I don't think we need to explain how clear works for Vecs here
Also it seems like if one batch returns an error, then the accumulator
shouldn't be invoked again (I would expect the query to error) -- so this code
is probably dead
However, that being said, I don't see any reason it is bad, but I do think
it is unecessary
##########
datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs:
##########
@@ -1134,6 +1143,87 @@ mod tests {
Ok(())
}
+ /// Counts the rows it is handed. `MaxAccumulator` cannot see a row that
+ /// reaches it twice, so the tests need an accumulator that can.
+ #[derive(Debug, Default)]
+ struct RowCountAccumulator {
+ rows: i64,
+ }
+
+ impl Accumulator for RowCountAccumulator {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ self.rows += values[0].len() as i64;
+ Ok(())
+ }
+ fn merge_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ self.update_batch(values)
+ }
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ Ok(ScalarValue::Int64(Some(self.rows)))
+ }
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![ScalarValue::Int64(Some(self.rows))])
+ }
+ fn size(&self) -> usize {
+ size_of::<Self>()
+ }
+ }
+
+ fn row_count_adapter() -> GroupsAccumulatorAdapter {
+ GroupsAccumulatorAdapter::new(|| {
+ Ok(Box::new(RowCountAccumulator::default()) as Box<dyn
Accumulator>)
+ })
+ }
+
+ fn max_adapter() -> GroupsAccumulatorAdapter {
+ GroupsAccumulatorAdapter::new(|| {
+ Ok(Box::new(MaxAccumulator::try_new(&DataType::Int64)?)
+ as Box<dyn Accumulator>)
+ })
+ }
+
+ /// Every row must reach its own group's accumulator, whatever order the
Review Comment:
do we really need new test coverage? I think this code is pretty well
covered by slt tests
##########
datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs:
##########
@@ -253,27 +262,17 @@ impl GroupsAccumulatorAdapter {
}
self.add_allocation(indices_allocation_delta);
- // groups_with_rows holds a list of group indexes that have
- // any rows that need to be accumulated, stored in order of
- // group_index
-
- let mut groups_with_rows = vec![];
-
// batch_indices holds indices into values, each group is contiguous
- let mut batch_indices = vec![];
+ let mut batch_indices = Vec::with_capacity(group_indices.len());
Review Comment:
same thing here -- could reuse the allocations across calls
```rust
struct Scratch {
batch_indices: Vec<usize>,
offsets: Vec<usize>,
groups_with_rows: Vec<usize>
}
```
##########
datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs:
##########
@@ -1159,6 +1249,100 @@ mod tests {
Ok(())
}
+ /// A failed update must not leave rows behind for the next batch: each
Review Comment:
see above -- I am not sure this is an important property (though it is not
incorrect)
##########
datafusion/functions-aggregate/benches/groups_accumulator_adapter.rs:
##########
@@ -0,0 +1,163 @@
+// 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.
+
+//! What it costs to run an [`Accumulator`] through
[`GroupsAccumulatorAdapter`],
+//! over a sweep of `GROUP BY` cardinalities.
+//!
+//! Two accumulators, because the adapter's cost and the aggregate's cost move
+//! in opposite directions as the group count grows:
+//!
+//! * `routing` wraps an accumulator that only counts the rows it is handed, so
+//! what the benchmark measures is the adapter routing the batch and nothing
+//! else. It is the upper bound on what a change to the routing can do.
+//! * `covar_samp` wraps a real aggregate that has no native
`GroupsAccumulator`,
+//! so it shows how much of that upper bound a query actually sees.
+
+use std::hint::black_box;
Review Comment:
I don't think this is necessary as we have end to end coverage, but it isn't
incorrect
--
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]
