danny0405 commented on code in PR #19727:
URL: https://github.com/apache/hudi/pull/19727#discussion_r3870192169
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/BucketPartitionUtils.scala:
##########
@@ -49,10 +57,35 @@ object BucketPartitionUtils extends SparkAdapterSupport {
val getPartitionKey = getPartitionKeyExtractor()
// use internalRow to avoid extra convert.
- val reRdd = df.queryExecution.toRdd
- .keyBy(row => getPartitionKey(row))
- .repartitionAndSortWithinPartitions(partitioner)
- .values
+ val internalRows = df.queryExecution.toRdd
+ val reRdd = if (sortByRecordKey) {
+ val utf8StringFactory = sparkAdapter.getUTF8StringFactory
+ // Use (bucket route, record key) as the shuffle key. Tuple ordering
groups rows by
+ // (partition path, bucket id) first, then orders each bucket by the
full record key.
+ // Scala derives the record-key ordering from HoodieUTF8String's
Comparable implementation,
+ // which uses binary UTF-8 ordering for both Spark 3 and Spark 4.
+ val keyedRows = internalRows.keyBy(row => {
+ // InternalRow may reuse its mutable backing buffer, so keep an
independent key for shuffle.
+ val recordKey = utf8StringFactory.wrapUTF8String(
+ row.getUTF8String(HoodieRecord.RECORD_KEY_META_FIELD_ORD).copy())
Review Comment:
Could we avoid this per-record copy? Hudi requires Kryo, and Spark’s
serialized shuffle serializes both the key and value before advancing the input
iterator, so the reduce-side sorter sees independently deserialized objects.
Also, the shuffled value is still the original `row`; if this path retained a
reused `InternalRow` buffer, copying only the key would preserve ordering but
not the row contents. Please remove `.copy()` and the preceding comment. If
this utility is intended to support a non-serialized shuffle path, it would
need to copy the entire row once and derive the key from that copied row
instead.
--
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]
