[FLINK][VL] Stateful key hash collision: hash value used as key identity causes data corruption

### Backend

VL (Velox)

### 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` 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 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` 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