refactor(hash-aggr): Use EmitTo to output#23055
Conversation
Amazing |
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
This is amazing @2010YOUY01 -- thank you. I found this code really easy to follow and understand. While it is complicated, I think it much more closely mirrors the complexity of the problem being solved now and setting up the control flow logic in this way means we will be in a much better place to improve the performance / featuers going forward
👏
cc @Rachelint
| AggregateExec, PhysicalGroupBy, aggregate_expressions, evaluate_group_by, | ||
| }; | ||
|
|
||
| /// Marker for raw rows -> partial state aggregation. |
There was a problem hiding this comment.
I like this structure and how it makes it clearer what is going on with the state here
There was a problem hiding this comment.
Minor is that the structuis called final but the module is called final_table.rs -- should we keep it consistent with final.rs?
There was a problem hiding this comment.
No, that is the marker struct for hash table aggregation mode. I have renamed it AggregateHashTable<Final> -> AggregateHashTable<FinalMarker> to make it more explicit
There was a problem hiding this comment.
likewise here, the struct is named Partial but the module partial_table.rs -- recommend partial.rs to be consistent
There was a problem hiding this comment.
Same as above ⬆️
| ) -> Result<Option<RecordBatch>> { | ||
| let output_schema = Arc::clone(&self.output_schema); | ||
| let batch_size = self.batch_size; | ||
| match &mut self.state { |
There was a problem hiding this comment.
this state match and some of the outputtting state is duplicated across the types of tables, but I think it is ok
There was a problem hiding this comment.
There are several small differences like metrics tracking, so probably it's clearer to keep them separated 🤔
| /// In skip-partial-aggregation optimization, when a decision has made to skip | ||
| /// partial stage, build a typed hash table only for aggregation state conversion | ||
| /// row-by-row. | ||
| pub(in crate::aggregates) fn partial_skip_table( |
There was a problem hiding this comment.
I wonder if we could avoid some clones below if this consumed self rather than took it by reference
Maybe it doesn't matter
There was a problem hiding this comment.
Yes, it's doable, and I think we can further simplify it into a much smaller struct since for partial aggregation skip stage, only a bunch of GroupsAccmulators are used.
This requires a separate PR, but I agree it's more of a idea to polish the code, not super important for now, I'll try to address it when the refactor is mostly done
| .building() | ||
| .accumulators | ||
| .iter() | ||
| .all(|acc| acc.supports_convert_to_state()) |
There was a problem hiding this comment.
I think we should try and remove this "supports_convert_to_state" API (as a follow on PR / project) to simplify the hash aggregate code and ensure all our groups accumulators have the high performance APIs.
I filed a ticket
…artial_table.rs Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
|
All comments have been addressed, thank you for the careful reviews! @alamb @Rachelint
I only figured this out very recently. The split-stream approach is somewhat counterintuitive: it does introduce a lot of duplicated code, but it can make the code easier to work with. The key idea, I think, is problem decomposition. If we can break a large problem into smaller subproblems, we can tackle each of them individually. |
3 participants
Which issue does this PR close?
Part of #22710
Rationale for this change
Regarding the EPIC issue: I have drafted all the migrations locally, and verified that after deleting the old implementation, UTs are passing.
We are now about 4 feature migration PRs away from completing the EPIC. Before continuing with those migrations, this PR performs some cleanup and refactoring.
What changes are included in this PR?
This PR can be read commit by commit:
hash_table.rsinto small filesAre these changes tested?
Are there any user-facing changes?