stevenzwu commented on code in PR #10823:
URL: https://github.com/apache/iceberg/pull/10823#discussion_r1697716288
##########
flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchRangePartitioner.java:
##########
@@ -47,18 +46,6 @@ class SketchRangePartitioner implements Partitioner<RowData>
{
public int partition(RowData row, int numPartitions) {
// reuse the sortKey and rowDataWrapper
sortKey.wrap(rowDataWrapper.wrap(row));
- int partition = Arrays.binarySearch(rangeBounds, sortKey, comparator);
Review Comment:
this refactoring is to make it easier to write unit test with the static
util method
##########
flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java:
##########
@@ -91,7 +91,7 @@ class DataStatisticsCoordinator implements
OperatorCoordinator {
this.context = context;
this.schema = schema;
this.sortOrder = sortOrder;
- this.comparator = SortOrderComparators.forSchema(schema, sortOrder);
+ this.comparator = Comparators.forType(SortKeyUtil.sortKeySchema(schema,
sortOrder).asStruct());
Review Comment:
this is an important fix. SortKey may have been serialized by subtask and
deserialized by the coordinator. we need to use the schema of the transformed
sort fields for struct comparator.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
