cshuo commented on code in PR #19727:
URL: https://github.com/apache/hudi/pull/19727#discussion_r3871456388
##########
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:
Makes sense. The Kryo shuffle serializes both key and value before advancing
the iterator. Removed the redundant key copy and comment.
--
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]
