danny0405 commented on code in PR #19131:
URL: https://github.com/apache/hudi/pull/19131#discussion_r3510412745
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/SortOperatorGen.java:
##########
@@ -34,26 +38,201 @@
public class SortOperatorGen {
private final int[] sortIndices;
private final RowType rowType;
- private final TableConfig tableConfig = TableConfig.getDefault();
+ private final RowData.FieldGetter[] fieldGetters;
public SortOperatorGen(RowType rowType, String[] sortFields) {
- this.sortIndices =
Arrays.stream(sortFields).mapToInt(rowType::getFieldIndex).toArray();
this.rowType = rowType;
+ this.sortIndices = Arrays.stream(sortFields).mapToInt(field -> {
+ int index = rowType.getFieldIndex(field);
+ if (index < 0) {
+ throw new IllegalArgumentException("Can not find sort field '" + field
+ "' in row type " + rowType);
+ }
+ return index;
+ }).toArray();
+ this.fieldGetters = Arrays.stream(sortIndices)
+ .mapToObj(index -> RowData.createFieldGetter(rowType.getTypeAt(index),
index))
+ .toArray(RowData.FieldGetter[]::new);
}
public OneInputStreamOperator<RowData, RowData>
createSortOperator(Configuration conf) {
- SortCodeGenerator codeGen = createSortCodeGenerator();
return new SortOperator(
- codeGen.generateNormalizedKeyComputer("SortComputer"),
- codeGen.generateRecordComparator("SortComparator"),
+ generateNormalizedKeyComputer("SortComputer"),
+ generateRecordComparator("SortComparator"),
conf);
}
- public SortCodeGenerator createSortCodeGenerator() {
- SortSpec.SortSpecBuilder builder = SortSpec.builder();
- for (int sortIndex : sortIndices) {
- builder.addField(sortIndex, true, true);
+ public GeneratedNormalizedKeyComputer generateNormalizedKeyComputer(String
name) {
+ String className = generatedClassName(name);
+ return new GeneratedNormalizedKeyComputer(className,
generateNormalizedKeyComputerCode(className));
+ }
+
+ public GeneratedRecordComparator generateRecordComparator(String name) {
+ String className = generatedClassName(name);
+ return new GeneratedRecordComparator(className,
generateRecordComparatorCode(className), fieldGetters);
+ }
+
+ private String generatedClassName(String name) {
+ String normalizedName = name.replaceAll("[^A-Za-z0-9_$]", "_");
+ if (normalizedName.isEmpty() ||
!Character.isJavaIdentifierStart(normalizedName.charAt(0))) {
+ normalizedName = "_" + normalizedName;
+ }
+ int hash = 31 * Arrays.hashCode(sortIndices) +
rowType.asSerializableString().hashCode();
+ return normalizedName + "_" + Integer.toUnsignedString(hash);
+ }
+
+ private String generateRecordComparatorCode(String className) {
+ StringBuilder code = new StringBuilder();
+ code.append("public final class ").append(className)
+ .append(" implements
org.apache.flink.table.runtime.generated.RecordComparator {\n")
+ .append(" private final Object[] references;\n")
+ .append(" public ").append(className).append("(Object[] references)
{\n")
+ .append(" this.references = references;\n")
+ .append(" }\n")
+ .append(" @Override\n")
+ .append(" public int compare(org.apache.flink.table.data.RowData
row1, ")
Review Comment:
Good catch. I restored the comparator to match the original
`SortSpec.addField(sortIndex, true, true)` behavior: ascending sort with nulls
last. I also added `TestSortOperatorGen` coverage to assert nulls-last in both
compare directions.
--
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]
