hudi-agent commented on code in PR #19131:
URL: https://github.com/apache/hudi/pull/19131#discussion_r3503928706
##########
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:
🤖 The original code used `builder.addField(sortIndex, true, true)`, i.e.
ascending with `nullIsLast=true` (nulls sort last). This generated comparator
returns `isNull1 ? -1 : 1`, which puts nulls first. Was flipping to nulls-first
intentional? For clustering/append sort columns that can be null, this changes
the physical row ordering versus the prior planner-generated comparator.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/FlinkTransformationUtils.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.utils;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+
+/**
+ * Utilities for Flink transformations.
+ */
+public final class FlinkTransformationUtils {
+ private FlinkTransformationUtils() {
+ }
+
+ public static <T> void setManagedMemoryWeight(Transformation<T>
transformation, long memoryBytes) {
+ if (memoryBytes <= 0) {
+ return;
+ }
+ int weight = Math.max(1, (int) (memoryBytes >> 20));
Review Comment:
🤖 nit: `>> 20` to convert bytes to MiB is a bit cryptic here — could you use
`/ (1024 * 1024)` (or at least a short inline comment `// bytes → MiB`) so the
unit conversion is immediately obvious to readers?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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, ")
+ .append("org.apache.flink.table.data.RowData row2) {\n");
+ for (int i = 0; i < sortIndices.length; i++) {
+ int sortIndex = sortIndices[i];
+ code.append(" boolean isNull1_").append(i).append(" =
row1.isNullAt(").append(sortIndex).append(");\n")
+ .append(" boolean isNull2_").append(i).append(" =
row2.isNullAt(").append(sortIndex).append(");\n")
+ .append(" if (isNull1_").append(i).append(" ||
isNull2_").append(i).append(") {\n")
+ .append(" if (isNull1_").append(i).append(" &&
isNull2_").append(i).append(") {\n")
+ .append(" } else {\n")
+ .append(" return isNull1_").append(i).append(" ? -1 : 1;\n")
+ .append(" }\n")
+ .append(" } else {\n")
+ .append(" int cmp_").append(i).append(" =
").append(compareExpression(i, sortIndex)).append(";\n")
+ .append(" if (cmp_").append(i).append(" != 0) {\n")
+ .append(" return cmp_").append(i).append(";\n")
+ .append(" }\n")
+ .append(" }\n");
}
- return new SortCodeGenerator(tableConfig,
Thread.currentThread().getContextClassLoader(), rowType, builder.build());
+ code.append(" return 0;\n")
+ .append(" }\n")
+ .append(" private int
compareFallback(org.apache.flink.table.data.RowData row1, ")
+ .append("org.apache.flink.table.data.RowData row2, int referenceIndex)
{\n")
+ .append(" Object value1 =
((org.apache.flink.table.data.RowData.FieldGetter) references[referenceIndex])")
+ .append(".getFieldOrNull(row1);\n")
+ .append(" Object value2 =
((org.apache.flink.table.data.RowData.FieldGetter) references[referenceIndex])")
+ .append(".getFieldOrNull(row2);\n")
+ .append(" return compareValues(value1, value2);\n")
+ .append(" }\n")
+ .append(" private static int compareValues(Object value1, Object
value2) {\n")
+ .append(" if (value1 == value2) {\n")
+ .append(" return 0;\n")
+ .append(" }\n")
+ .append(" if (value1 == null) {\n")
+ .append(" return -1;\n")
+ .append(" }\n")
+ .append(" if (value2 == null) {\n")
+ .append(" return 1;\n")
+ .append(" }\n")
+ .append(" if (value1 instanceof byte[] && value2 instanceof byte[])
{\n")
+ .append(" return compareUnsignedBytes((byte[]) value1, (byte[])
value2);\n")
+ .append(" }\n")
+ .append(" if (value1 instanceof java.lang.Comparable && value2
instanceof java.lang.Comparable) {\n")
+ .append(" return ((java.lang.Comparable)
value1).compareTo(value2);\n")
+ .append(" }\n")
+ .append(" throw new IllegalArgumentException(\"Unsupported sort
field value type: \" ")
+ .append("+ value1.getClass().getName());\n")
+ .append(" }\n")
+ .append(" private static int compareUnsignedBytes(byte[] bytes1,
byte[] bytes2) {\n")
+ .append(" int len = java.lang.Math.min(bytes1.length,
bytes2.length);\n")
+ .append(" for (int i = 0; i < len; i++) {\n")
+ .append(" int result = java.lang.Byte.toUnsignedInt(bytes1[i]) ")
+ .append("- java.lang.Byte.toUnsignedInt(bytes2[i]);\n")
+ .append(" if (result != 0) {\n")
+ .append(" return result;\n")
+ .append(" }\n")
+ .append(" }\n")
+ .append(" return bytes1.length - bytes2.length;\n")
+ .append(" }\n")
+ .append("}\n");
+ return code.toString();
+ }
+
+ private String compareExpression(int referenceIndex, int sortIndex) {
+ LogicalType logicalType = rowType.getTypeAt(sortIndex);
+ switch (logicalType.getTypeRoot()) {
+ case BOOLEAN:
+ return "java.lang.Boolean.compare(row1.getBoolean(" + sortIndex + "),
row2.getBoolean(" + sortIndex + "))";
+ case TINYINT:
+ return "java.lang.Byte.compare(row1.getByte(" + sortIndex + "),
row2.getByte(" + sortIndex + "))";
+ case SMALLINT:
+ return "java.lang.Short.compare(row1.getShort(" + sortIndex + "),
row2.getShort(" + sortIndex + "))";
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ case INTERVAL_YEAR_MONTH:
+ return "java.lang.Integer.compare(row1.getInt(" + sortIndex + "),
row2.getInt(" + sortIndex + "))";
+ case BIGINT:
+ case INTERVAL_DAY_TIME:
+ return "java.lang.Long.compare(row1.getLong(" + sortIndex + "),
row2.getLong(" + sortIndex + "))";
+ case FLOAT:
+ return "java.lang.Float.compare(row1.getFloat(" + sortIndex + "),
row2.getFloat(" + sortIndex + "))";
+ case DOUBLE:
+ return "java.lang.Double.compare(row1.getDouble(" + sortIndex + "),
row2.getDouble(" + sortIndex + "))";
+ case CHAR:
+ case VARCHAR:
+ return "row1.getString(" + sortIndex + ").compareTo(row2.getString(" +
sortIndex + "))";
+ case BINARY:
+ case VARBINARY:
+ return "compareUnsignedBytes(row1.getBinary(" + sortIndex + "),
row2.getBinary(" + sortIndex + "))";
+ case DECIMAL:
+ DecimalType decimalType = (DecimalType) logicalType;
+ return "row1.getDecimal(" + sortIndex + ", " +
decimalType.getPrecision() + ", "
+ + decimalType.getScale() + ").compareTo(row2.getDecimal(" +
sortIndex + ", "
+ + decimalType.getPrecision() + ", " + decimalType.getScale() +
"))";
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return timestampCompareExpression(sortIndex, ((TimestampType)
logicalType).getPrecision());
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return timestampCompareExpression(sortIndex,
((LocalZonedTimestampType) logicalType).getPrecision());
+ case TIMESTAMP_WITH_TIME_ZONE:
+ return timestampCompareExpression(sortIndex, ((ZonedTimestampType)
logicalType).getPrecision());
+ default:
+ return "compareFallback(row1, row2, " + referenceIndex + ")";
+ }
+ }
+
+ private String timestampCompareExpression(int sortIndex, int precision) {
+ return "row1.getTimestamp(" + sortIndex + ", " + precision +
").compareTo(row2.getTimestamp("
+ + sortIndex + ", " + precision + "))";
+ }
+
+ private String generateNormalizedKeyComputerCode(String className) {
+ return "public final class " + className
+ + " implements
org.apache.flink.table.runtime.generated.NormalizedKeyComputer {\n"
+ + " public " + className + "(Object[] references) {\n"
Review Comment:
🤖 The generated NormalizedKeyComputer is effectively a no-op
(getNumKeyBytes()=1, putKey writes 0, compareKey always returns 0,
isKeyFullyDetermines()=false), so every comparison falls back to the full
RecordComparator. Correct, but it disables Flink's binary-prefix sort
optimization that the planner-generated computer provided. Was the performance
impact on large bulk-insert/clustering sorts considered?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
