ggjh-159 opened a new issue, #12805:
URL: https://github.com/apache/gluten/issues/12805
### Backend
VL (Velox)
### Bug description
## Bug description
The Velox stateful streaming path uses `HashPartitionFunction` to compute a
hash of the key columns, then uses this **hash value as the key identity** for
state storage — discarding the original key values entirely. When two different
keys produce the same hash (hash collision), their data is merged into the same
state, causing silent data corruption in aggregation, join, and other keyed
operations.
Additionally, the current pre-grouping approach (`KeySelector::partition` +
per-key state get/put) prevents the use of Velox's native batch vectorized
processing (`Aggregate::addRawInput`), negating the performance advantage that
Velox is supposed to provide.
## Root Cause
### How keys are currently processed
1. `KeySelector::partition()` calls `HashPartitionFunction::partition()`
which computes a 64-bit hash of the key columns via `VectorHasher`, then takes
`hash % numPartitions` (where `numPartitions = INT_MAX`) to produce a
`uint32_t` partition number.
2. This partition number is used as the **map key** in `std::map<int64_t,
RowVectorPtr>` to group rows — different keys with the same hash are merged
into the same group.
3. The partition number is passed to `setCurrentKey(uint32_t)` as the state
key identity. All subsequent state operations (`state_->value(key, window)`,
`state_->update(key, window, acc)`) use this hash value, not the original key.
**Source code references:**
```
KeySelector.cpp:44-52 — partition value used as map key
KeySelector.cpp:85-91 — map<int64_t, RowVectorPtr> grouped by partition
number
KeyedStateBackend.h:60 — setCurrentKey(uint32_t key)
GroupWindowAggregator.cpp:87-90 — partition result used as key for
setCurrentKey
GroupWindowAggregator.cpp:99 — windowState_->value(key, window) using
hash as key
HashPartitionFunction.cpp:115-117 — partitions[i] = hashes_[i] %
numPartitions_
```
### Why this causes data corruption
`HashPartitionFunction` is designed for **data partitioning** (deciding
which subtask receives data), not for **key identity**. In Velox's native batch
`HashAggregation`, the hash is used only for bucket lookup, and
`HashTable::compareKeys` performs a precise comparison of the original key
values to distinguish keys that hash to the same bucket. The stateful path
**bypasses this safeguard** by using the hash value directly as the key — there
is no original-key comparison anywhere in the state access path.
### Contrast with Flink
Flink never uses hash values as key identity:
- **Heap StateBackend**: keys are compared using `equals()` on the original
Java objects (`CopyOnWriteStateMap.java:278`)
- **RocksDB StateBackend**: keys are serialized to byte arrays and compared
using lexicographic byte comparison (`BytewiseComparator`)
- Hash functions (`murmurHash(key.hashCode())`) are used **only** for key
group assignment (deciding which parallel subtask owns a key), never for key
equality
## Collision Can Definitely Occur
The hash value is `uint64_t % INT_MAX`, stored in `uint32_t`. Since the
range is finite (~2.1 billion) while the number of possible distinct keys is
unbounded, collisions are mathematically guaranteed to occur given enough keys.
The concrete example above (key=391 and key=32728) demonstrates that collisions
exist even among small integer keys — no adversarial input is needed.
## Reproduction Scenario
### Minimal example
```sql
-- A simple GROUP BY aggregation with SUM
CREATE TABLE source (
key BIGINT,
value BIGINT
) WITH ('connector' = 'from-elements');
INSERT INTO sink
SELECT key, SUM(value) AS total
FROM source
GROUP BY key;
```
### Concrete collision example
Velox's `VectorHasher` uses `folly::hasher<int64_t>` for BIGINT keys, which
applies `fmix64` (MurmurHash3 finalizer). Two different BIGINT keys `391` and
`32728` produce different 64-bit hashes but collide after `% INT_MAX`:
```
key=391: fmix64(391) = 8035620079152987179, % INT_MAX = 250707955
key=32728: fmix64(32728) = 11952965464888349291, % INT_MAX = 250707955
```
Both keys hash to partition number `250707955`. Since this partition number
is used as the key identity for state operations, the state for `key=391` and
`key=32728` is shared — their data is silently merged.
### Reproduction SQL
```sql
-- Source with two keys that collide
CREATE TABLE source (
key BIGINT,
value BIGINT
) WITH ('connector' = 'from-elements', 'data' = '[
{"key": 391, "value": 100},
{"key": 32728, "value": 200}
]');
-- GROUP BY SUM should produce separate results per key
SELECT key, SUM(value) AS total
FROM source
GROUP BY key;
```
### Expected vs actual behavior
**Expected output**:
```
key=391, total=100
key=32728, total=200
```
**Actual output** (one of the following, depending on which key is processed
first):
```
key=391, total=300 ← includes value from key=32728
key=32728, total=300 ← or both show the merged sum
```
Or one key's result is missing entirely. **No error is raised** — the
corruption is silent.
In practice, with 65,000+ distinct keys in a streaming aggregation,
collisions are statistically likely to occur naturally without any crafted
input.
## Impact
### Affected operators
All operators that use `KeySelector` and `setCurrentKey(uint32_t)`:
| Operator | File | Impact |
|---|---|---|
| GroupAggregate | `StreamKeyedOperator.cpp:53` | Aggregation results
include wrong keys' data |
| GroupWindowAggregate | `GroupWindowAggregator.cpp:87` | Window aggregation
merges different keys' windows |
| WindowJoin | `WindowJoin.cpp:94` | Join matches incorrect keys |
| LocalWindowAggregate | `LocalWindowAggregator.cpp:52` | Same as
GroupWindowAggregate |
| WindowAggregate | `WindowAggregator.cpp:76` | Same as GroupWindowAggregate
|
| StreamRank (TopN) | `AppendOnlyTopNRanker.cpp:83` | Rank includes wrong
keys |
| StreamJoin | `StreamJoin.cpp:69` | Join on wrong keys |
### Checkpoint corruption
State is checkpointed with the hash value as the key. After recovery, the
corrupted state persists — there is no way to distinguish which original keys
were merged.
### Cannot leverage Velox batch processing
The current design pre-groups data by partition number via
`KeySelector::partition()`, then processes each group sequentially with per-key
`state.get() → accumulate → state.put()`. This two-pass approach (group, then
process) prevents the use of Velox's native `Aggregate::addRawInput`, which can
process an entire `RowVector` in a single vectorized batch via
`HashTable::groupProbe` + `compareKeys`.
Velox's batch `HashAggregation` achieves high throughput by:
1. Single-pass `groupProbe`: hash lookup + `compareKeys` (original key
comparison) in one traversal
2. `Aggregate::addRawInput`: vectorized SIMD accumulation across all rows
simultaneously
3. No per-key state get/put in the hot path
The current stateful path gains none of these benefits because:
- The pre-grouping step (`KeySelector::partition`) is an extra O(n) traversal
- Per-key `accumulate()` (currently an empty implementation in
`GroupWindowAggsHandler.cpp`) does not use vectorized accumulation
- Per-key `state.get()/state.put()` adds O(distinct_keys) state backend
operations per batch
## Key Source Files(bigo-sg/velox)
| File | Path (relative to velox root) | Role |
|---|---|---|
| `KeySelector.h` | `velox/experimental/stateful/KeySelector.h` | Key
grouping (problem origin) |
| `KeySelector.cpp` | `velox/experimental/stateful/KeySelector.cpp:29-92` |
`partition()` uses hash as map key |
| `HashPartitionFunction.cpp` |
`velox/exec/HashPartitionFunction.cpp:76-122` | Hash computation + modulo |
| `KeyedStateBackend.h` |
`velox/experimental/stateful/state/KeyedStateBackend.h:60-69` |
`setCurrentKey(uint32_t)` + `currentKey_` type |
| `GroupWindowAggregator.cpp` |
`velox/experimental/stateful/GroupWindowAggregator.cpp:87-90` | Partition →
setCurrentKey chain |
| `StateMap.h` | `velox/experimental/stateful/state/StateMap.h:132-152` |
`e->key_ == key` compares hash values, not original keys |
| `StateTable.h` | `velox/experimental/stateful/state/StateTable.h:37-50` |
Key group assignment by hash |
| `GroupWindowAggsHandler.cpp` |
`velox/experimental/stateful/window/GroupWindowAggsHandler.cpp` | All methods
empty (no vectorized accumulation) |
### Gluten version
main branch
### Spark version
None
### Spark configurations
_No response_
### System information
_No response_
### Relevant logs
```bash
```
--
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]
