danny0405 commented on PR #20138:
URL: https://github.com/apache/hudi/pull/20138#issuecomment-5902985136
An alternative worth considering is to use the existing serialized-record
offset as the arrival-order tie-breaker, instead of allocating
`originalPositions`.
I tested this in an isolated copy of Flink 2.2.1's buffer classes. In the
append-only `BinaryInMemorySortBuffer`, each sort-index entry already stores
the record's byte offset. Those offsets increase with arrival order, and
existing swaps move the offsets together with their index entries. Comparing
offsets when complete record keys are equal therefore preserves arrival order
without an additional `int[N]` or position updates on every swap.
The change is in the segment-addressed `BinaryIndexedSortable.compare(...)`
method, after obtaining `segI` and `segJ`:
```java
int val = normalizedKeyComputer.compareKey(
segI, segmentOffsetI + OFFSET_LEN,
segJ, segmentOffsetJ + OFFSET_LEN);
if (val != 0) {
return useNormKeyUninverted ? val : -val;
}
long pointerI = segI.getLong(segmentOffsetI);
long pointerJ = segJ.getLong(segmentOffsetJ);
if (!normalizedKeyFullyDetermines) {
val = compareRecords(pointerI, pointerJ);
if (val != 0) {
return val;
}
}
return Long.compare(pointerI, pointerJ);
```
The offset comparison must happen only after full-key equality is
established, including when the normalized key fully determines the result. The
index-addressed overload delegates here, so this also covers the HeapSort
fallback. QuickSort and the existing swap implementation do not change.
In local sort-only measurements on 100,000 records:
| Workload | Position-array wrapper | Offset tie-breaker |
| --- | ---: | ---: |
| Shared-prefix string keys, 17 distinct keys | 150.1 ms | 144.1 ms |
| Short string keys, 17 distinct keys | 106.3 ms | 100.2 ms |
| Encoded keys, 17 distinct keys | 219.8 ms | 201.9 ms |
Across nine cases covering 1, 17, and 100,000 distinct keys, median sort
time decreased by approximately 1–8%; the smaller differences overlap
run-to-run variation. The clear benefit is memory: sorting-thread allocation
decreased by exactly 400,048 bytes per 100,000-record sort, eliminating the
position array and tracking wrapper. Existing comparator allocations remain.
Measurements used Java 17 on macOS arm64, Flink 2.2.1, deterministic
shuffled inputs, three independent JVM forks per implementation, four warmups
and seven measured iterations per case per fork. Timings include wrapper/array
initialization and exclude buffer filling and verification. No already-sorted
fast path or comparison-count instrumentation was added. The wrapper benchmark
used commit `90dacf96c973`; its implementation is unchanged at `afd135362242`.
The isolated prototype passed 50 correctness cases covering generated/encoded
comparators, QuickSort/HeapSort, page boundaries, variable-length rows, row
kinds, page release, and fully determining normalized keys.
This does **not** remove the duplicate-heavy CPU cost of imposing an
arrival-order tie-breaker: original key-only QuickSort took 29.1 ms on the
17-key shared-prefix workload, versus 144.1 ms for the offset version.
The integration tradeoff is that Flink's current public buffer API does not
expose these pointers, and `BinaryInMemorySortBuffer` is final. This would need
an upstream buffer API/implementation change or a maintained Hudi-owned buffer
implementation. For production, I would make stability opt-in for the
append-only buffer rather than change every user of the shared
`BinaryIndexedSortable` base class. The prototype is an allocation-saving
alternative, not a drop-in replacement through the current wrapper API.
--
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]
