fix(spark): make binary clustering/bulk-insert sort keys Comparable - #19622
fix(spark): make binary clustering/bulk-insert sort keys Comparable#19622lokeshj1703 wants to merge 2 commits into
Conversation
Clustering and bulk-insert sort keys flow through
SortUtils.getComparableSortColumns -> FlatLists.ofComparableArray, which
casts every sort value to Comparable. On the Spark record path a binary
(BINARY/bytes) sort column arrives as a raw byte[], which is not
Comparable, so ofComparableArray threw:
java.lang.ClassCastException: [B cannot be cast to java.lang.Comparable
at FlatLists.ofComparableArray(FlatLists.java:55)
The Avro path never hit this because getNestedFieldVal yields a
java.nio.ByteBuffer for the same column (and BaseSparkInternalRecordContext
already wraps byte[] as ByteBuffer elsewhere). Wrap byte[] into
ByteBuffer.wrap(...) inside the shared HoodieUTF8StringFactory hook that
both RDDCustomColumnsSortPartitioner and RDDBucketIndexPartitioner use.
ByteBuffer is Comparable + Serializable (survives the sortBy shuffle) and
preserves the exact byte-lexicographic ordering the Avro path produced, so
this is behavior-preserving with no FlatLists change.
Adds TestBulkInsertInternalPartitioner#testSortColumnsWithBinaryValueAreComparable,
which reproduces the CCE without the fix and asserts correct ordering with it.
aa4e6e0 to
4b1cbed
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19622 +/- ##
============================================
+ Coverage 77.09% 77.80% +0.70%
- Complexity 32490 33059 +569
============================================
Files 2522 2525 +3
Lines 139112 139645 +533
Branches 16714 17075 +361
============================================
+ Hits 107243 108645 +1402
+ Misses 24291 23382 -909
- Partials 7578 7618 +40
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR fixes a ClassCastException when clustering or bulk-inserting with a BINARY sort column on the Spark record path, by wrapping raw byte[] into a ByteBuffer in the shared wrapArrayOfObjects hook so it becomes Comparable with the same byte-lexicographic ordering the Avro path already produces. I traced the sort-key path (SortUtils → FlatLists.ofComparableArray) and the serialization path (ComparableListImpl is KryoSerializable, matching how the Avro ByteBuffer keys already shuffle), and the ordering and edge-case behavior line up with the existing Avro path. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR fixes a ClassCastException on the Spark clustering/bulk-insert path when sorting by a BINARY column by wrapping raw byte[] into ByteBuffer.wrap(...) in the shared wrapArrayOfObjects hook, matching the ordering the Avro path already produces. I traced the reachable call paths (only the Spark wrapUTF8StringFunc overload hits this; the Java/Avro overload is unaffected), the aliasing safety of ByteBuffer.wrap (the byte[] is a per-record value from getBinary, not a reused buffer), and the signed byte-lexicographic ordering equivalence with the Avro ByteBuffer path — all consistent. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
4 participants
Describe the issue this Pull Request addresses
Closes #19621
Summary and Changelog
Clustering and bulk-insert sort keys flow through
SortUtils.getComparableSortColumns->FlatLists.ofComparableArray, which casts every sort value toComparable. On the Spark record path a BINARY sort column arrives as a rawbyte[](notComparable), soofComparableArraythrewClassCastException: [B cannot be cast to java.lang.Comparable, failing every clustering / bulk-insert on that column. The Avro path was unaffected becauseHoodieAvroUtils.getNestedFieldValreturns ajava.nio.ByteBufferfor the same column.This wraps
byte[]intoByteBuffer.wrap(...)inside the sharedHoodieUTF8StringFactory.wrapArrayOfObjectshook used by bothRDDCustomColumnsSortPartitionerandRDDBucketIndexPartitioner.ByteBufferisComparable, and it survives thesortByshuffle because the sort keyFlatLists.ComparableListisKryoSerializableand Hudi always runs the write path with Kryo (HoodieSparkSqlWriterrejects any otherspark.serializer) -- so the wrappedbyte[]rides Kryo exactly as the Avro binary path already does. (ByteBufferis notjava.io.Serializable, and neither is the key; the shuffle uses Kryo, not Java serialization.) It preserves the exact byte-lexicographic ordering the Avro path already produced, so the change is behavior-preserving and needs no change toFlatLists. Adds two tests toTestBulkInsertInternalPartitioner:testSortColumnsWithBinaryValueAreComparablereproduces the CCE without the fix and asserts the ordering with it, andtestBinarySortKeySurvivesKryoRoundTripround-trips the wrapped key through Kryo (viaSerializationUtils) and asserts the element stays aByteBufferwith its ordering preserved (verified to fail without the fix with the sameClassCastException).Impact
Clustering and
bulk_insertcan sort by BINARY columns on the Spark record path. No behavior change for existing (string / primitive) sort columns.Risk Level
low
Behavior-preserving: the byte-lexicographic ordering is identical to the pre-existing Avro path, and it is covered by a new unit test that fails without the change. The wrapped
ByteBuffersort key is shuffled by the sort partitioner exactly as the Avro binary path already shuffles it (the sameByteBufferin the sameFlatLists.ComparableList), so there is no new serialization behavior (the key rides Kryo, same as the Avro path).Documentation Update
none
Contributor's checklist