Repository navigation
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23695 +/- ##
=======================================
Coverage 80.67% 80.67%
=======================================
Files 1095 1095
Lines 372416 372504 +88
Branches 372416 372504 +88
=======================================
+ Hits 300434 300517 +83
+ Misses 54055 54052 -3
- Partials 17927 17935 +8 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
ed48d69 to
ef9b92c
Compare
|
hi @maxburke |
Done |
There was a problem hiding this comment.
@maxburke
Thanks for the update.
I think the overall direction makes sense, especially moving the internal group value storage to 64-bit offsets. I found one blocking issue around output emission that can still hit the narrow Arrow offset limit. I also have one performance suggestion that may be worth considering separately.
| // schema types. Each slice holds at most `batch_size` rows, so unlike | ||
| // the full materialized batch, the narrow representation fits. | ||
| let batch = output | ||
| .next_batch(batch_size) |
There was a problem hiding this comment.
I think there is still one important edge case here. The output is split by row count before the LargeUtf8 / LargeBinary group keys are narrowed back to the declared Utf8 / Binary schema.
The Arrow limit for narrow string and binary arrays is based on the total value buffer size, not the number of rows. That means a batch_size slice can still exceed i32::MAX bytes if it contains many large distinct keys accumulated across input batches.
In that case this change moves the overflow from group value accumulation to the later narrow_group_key_columns cast, so the query can still fail.
I noticed the same assumption in datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs:827 and datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs:432.
Could we instead split emitted output based on the narrow offset byte limit, or otherwise guarantee that each emitted slice is below that limit? It would also be great to add a regression that exercises this boundary.
|
|
||
| /// Widens string/binary group key arrays to the internal representation | ||
| /// interned by `group_values` (see `PhysicalGroupBy::group_schema`). | ||
| pub(crate) fn widen_group_key_arrays(arrays: Vec<ArrayRef>) -> Result<Vec<ArrayRef>> { |
There was a problem hiding this comment.
One performance thought. ByteGroupValueBuilder now accepts both narrow and large input arrays, so I wonder if the per input batch arrow::compute::cast in widen_group_key_arrays is still necessary.
If the hashing, ordering, spilling and merge paths do not actually require the grouping arrays to be widened ahead of time, it may be possible to remove this cast entirely. That would avoid some extra offset and value buffer work on the hot aggregation path.
|
Marked as draft after 14 days without activity: merge conflicts; changes requested on the latest push. Please address these items and mark Ready for review when ready. |
Store Utf8 and Binary group keys with 64-bit offsets. This removes the 2 GiB limit on the combined key data in an aggregate table. Divide the output into batches, then change the keys to their declared types. Each output column must still fit the 32-bit offset limit. Keys inside lists or dictionaries keep their types. Use the same schema helper for each aggregate table. Use the existing stream adapter to divide large batches and restore the output types.
Check that each aggregate stream keeps the declared key types and returns the expected results. Check sorted and unsorted input, one and two stages, grouping sets, and groups that the stream writes to disk. The memory limit allows all four final streams to start. Check output under a memory limit before the input ends. Check streams that can and cannot reserve memory for the output batch. Also check empty batches and batches that need no type conversion.
A batch can fit the row limit and still contain too much key data for 32-bit offsets. Limit both rows and key bytes before output conversion. Use the same limit check for aggregate tables and streams. Search the offsets only when the requested rows do not fit. Test the 32-bit boundary, array slices, and keys that cannot fit in one output batch. Addresses apache#23695 (comment)
Let group storage with 64-bit offsets read Utf8 and Binary input arrays directly. Reuse the existing comparison and storage methods without a new offset buffer for each input column. Keep input conversion in GroupValuesRows because its row format requires the schema types. Reject extra columns before conversion. Test strings, binary values, nulls, sorted groups, and both offset sizes. Addresses apache#23695 (comment)
Output batches need 32-bit offsets but can use the stored key bytes. Subtract the first offset and share the required part of the value buffer. This removes the copy of the key data for each output batch. The output keeps the original value buffer in memory. Check that output keys share this buffer, including keys beyond the 32-bit offset limit.
efbf9db to
8d24422
Compare
github-actions
Bot
added
physical-expr
and downcast back to i32-offset-types as necessary.
Which issue does this PR close?
Closes #23694 (and possibly apache/datafusion-comet#4718)
Rationale for this change
Fixes an error when grouping on large string/binary data
What changes are included in this PR?
Inner core groups using i64 offsets and then downcasts to correct type when emitting batch slices
Are these changes tested?
Yes
Are there any user-facing changes?
No