@Override
        public int compare(Tuple2<String, RowData> k1, Tuple2<String, RowData> k2) {
            if (!Objects.equals(k1.f0, k2.f0)) {
                return Comparator.nullsLast(Comparator.comparing((Tuple2<String, RowData> k) -> k.f0)).compare(k1, k2);
            }

            if (k1.f1 == null || k2.f1 == null) {
                if (Objects.equals(k1.f1, k2.f1)) {
                    return 0;
                } else if (k1.f1 == null) {
                    return -1;
                } else {
                    return 1;
                }
            }

            if (EMPTY_ROW.equals(k1.f1) || EMPTY_ROW.equals(k2.f1)) {
                if (Objects.equals(k1.f1, k2.f1)) {
                    return 0;
                } else if (EMPTY_ROW.equals(k1.f1)) {
                    return 1;
                } else {
                    return -1;
                }
            }

            try {
                keySerializer.serialize(k1.f1, outputBuffer);
                byte[] b1 = outputBuffer.getCopyOfBuffer();
                outputBuffer.clear();
                keySerializer.serialize(k2.f1, outputBuffer);
                byte[] b2 = outputBuffer.getCopyOfBuffer();
                outputBuffer.clear();
                return SortUtil.compareBinary(b1, b2);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
A {@link Convertor} that key represent <partition, primary key>
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/collections/PartitionGroupBuffer.java