IgnatiusPang commented on PR #25751:
URL: https://github.com/apache/datafusion/pull/25751#issuecomment-5835310964
```markdown
### 1. Reproducer & Defect Analysis
When merging intermediate aggregate states in distributed query execution
(or streaming partitioned window states):
1. **Slice Underflow in `FirstValueAccumulator` / `LastValueAccumulator`**:
In `datafusion/functions-aggregate/src/first_last.rs`:
```rust
let is_set_idx = states.len() - 1;
let flags = states[is_set_idx].as_boolean();
```
If an empty slice `states = &[]` is passed during an empty partition
merge:
- `states.len() - 1` underflows `usize` to `18446744073709551615`.
- Direct indexing crashes immediately with an out-of-bounds panic.
2. **Unbounded Allocation Underflow in `array_agg`**:
In `datafusion/functions-aggregate/src/array_agg.rs`:
```rust
let sorted_len = self.sorted_runs.iter().map(|run|
run.len()).sum::<usize>();
let mut unsorted_indices = Vec::with_capacity(self.entries.len() -
sorted_len);
```
If `sorted_len > self.entries.len()`, `self.entries.len() - sorted_len`
underflows, causing an immediate process abort attempting to allocate
`usize::MAX` capacity.
---
### 2. Proposed Fix
Add bounds checks:
```rust
let is_set_idx = states.len().checked_sub(1)
.ok_or_else(|| exec_datafusion_err!("Empty states slice in first_last
merge_batch"))?;
```
And use `.saturating_sub()` on slice lengths in `array_agg`.
```
--
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]
