[fix](paimon) Pad CHAR values written to Paimon tables as Spark and Flink do

fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java

@@ -183,8 +183,8 @@ public void open(String serializedTable, Map<String, String> hadoopConfig,
                     this.bucketMode = table.bucketMode();
 
                     CoreOptions coreOptions = CoreOptions.fromMap(table.options());
-                    this.writeSchema = PaimonWriteSchema.create(
-                            table.rowType(), columnNames, changelogWrite);
+                    this.writeSchema = PaimonWriteSchema.create(table.rowType(), columnNames, changelogWrite,
+                            PaimonWriteSchema.charPadding(table.options()), coreOptions.rowkindField().orElse(null));
                     this.arrowAdapter = new PaimonArrowBatchAdapter(
                             writeSchema.inputType(), ZoneId.of(timeZone), allocator);
                     validateWriteColumnsForMergeEngine(

fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java

@@ -17,17 +17,27 @@
 
 package org.apache.doris.paimon;
 
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
 import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.CharType;
 import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
 import org.apache.paimon.types.RowKind;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.utils.BinaryStringUtils;
 import org.apache.paimon.utils.DefaultValueUtils;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Immutable mapping from Doris input columns to a Paimon table row.
@@ -41,8 +51,19 @@ final class PaimonWriteSchema {
     static final byte INSERT_OPERATION = 0;
     static final byte UPDATE_OPERATION = 1;
     static final byte DELETE_OPERATION = 2;
+    /**
+     * Table option through which FE asks the writer to pad CHAR values, set from the session variable
+     * enable_paimon_write_char_padding. Keep in sync with PaimonWriteBinding.DORIS_WRITE_CHAR_PADDING in
+     * fe-connector-paimon.
+     */
+    static final String DORIS_WRITE_CHAR_PADDING = "doris.write.char-padding";
 
     private final InternalRow.FieldGetter[] fieldGetters;
+    /**
+     * Per input column, pads the CHAR values it holds; null when it holds none, when padding is off, or for the
+     * table's rowkind.field.
+     */
+    private final CharPadder[] charPadders;
     private final RowType inputType;
     /** Maps Doris input-column position → Paimon table-schema position. */
     private final int[] tableFieldIndexes;
@@ -52,14 +73,15 @@ final class PaimonWriteSchema {
     private final int tableFieldCount;
 
     private PaimonWriteSchema(int[] tableFieldIndexes, int[] omittedDefaultFieldIndexes,
-            Object[] omittedDefaultValues, int tableFieldCount, RowType inputType) {
+            Object[] omittedDefaultValues, int tableFieldCount, RowType inputType, CharPadder[] charPadders) {
         this.fieldGetters = new InternalRow.FieldGetter[inputType.getFieldCount()];
         for (int i = 0; i < fieldGetters.length; i++) {
             // Getters describe the transport row, not the target table. Transport fields are
             // nullable so an invalid Doris NULL survives projection and reaches Paimon's
             // authoritative table-schema validation.
             this.fieldGetters[i] = InternalRow.createFieldGetter(inputType.getTypeAt(i), i);
         }
+        this.charPadders = charPadders;
         this.inputType = inputType;
         this.tableFieldIndexes = tableFieldIndexes;
         this.omittedDefaultFieldIndexes = omittedDefaultFieldIndexes;
@@ -82,12 +104,29 @@ static PaimonWriteSchema create(RowType tableType, String[] columnNames) {
 
     static PaimonWriteSchema create(
             RowType tableType, String[] columnNames, boolean changelogWrite) {
+        return create(tableType, columnNames, changelogWrite, false);
+    }
+
+    static PaimonWriteSchema create(
+            RowType tableType, String[] columnNames, boolean changelogWrite, boolean charPadding) {
+        return create(tableType, columnNames, changelogWrite, charPadding, null);
+    }
+
+    /**
+     * @param charPadding  whether to pad the CHAR values of inserted rows as Paimon's cast to CHAR does; see
+     *                     {@link #charPadder}
+     * @param rowKindField the table's {@code rowkind.field}, or null. Its value is not padded: Paimon takes the row
+     *                     kind from it as it is ({@code RowKind.fromShortString}), and a padded '+I' is no row kind.
+     */
+    static PaimonWriteSchema create(RowType tableType, String[] columnNames, boolean changelogWrite,
+            boolean charPadding, String rowKindField) {
         if (columnNames == null || columnNames.length == 0) {
             throw new IllegalArgumentException(
                     "PaimonJniWriter requires explicit column names");
         }
 
         int[] tableFieldIndexes = new int[columnNames.length];
+        CharPadder[] charPadders = new CharPadder[columnNames.length];
         List<DataField> inputFields = new ArrayList<>(columnNames.length);
         boolean[] specifiedFields = new boolean[tableType.getFieldCount()];
         for (int i = 0; i < columnNames.length; i++) {
@@ -118,6 +157,9 @@ static PaimonWriteSchema create(
             // Paimon writer return its normal constraint diagnostic instead of reading a default
             // primitive value from a non-null getter.
             inputFields.add(field.newType(field.type().nullable()));
+            if (charPadding && !field.name().equals(rowKindField)) {
+                charPadders[i] = charPadder(field.type());
+            }
         }
 
         int[] omittedDefaultFieldIndexes = new int[tableType.getFieldCount()];
@@ -139,7 +181,16 @@ static PaimonWriteSchema create(
                 Arrays.copyOf(omittedDefaultFieldIndexes, omittedDefaultCount),
                 Arrays.copyOf(omittedDefaultValues, omittedDefaultCount),
                 tableType.getFieldCount(),
-                new RowType(inputFields));
+                new RowType(inputFields),
+                charPadders);
+    }
+
+    /**
+     * Whether FE asked for CHAR padding through {@link #DORIS_WRITE_CHAR_PADDING}. An FE that predates the option
+     * does not set it, and does not pad its static partition values either.
+     */
+    static boolean charPadding(Map<String, String> tableOptions) {
+        return Boolean.parseBoolean(tableOptions.get(DORIS_WRITE_CHAR_PADDING));
     }
 
     /** Paimon input fields in the exact order transported by Arrow C Data. */
@@ -163,13 +214,135 @@ GenericRow tableRow(InternalRow columnValues) {
                 row.setRowKind(toRowKind(value));
                 continue;
             }
+            // Only an inserted row is padded; its row kind, if any, is the first input column. An UPDATE or
+            // DELETE row carries the key Doris read back from the table, which has to stay byte-equal to the
+            // stored key for Paimon to match it, padded or not.
+            if (charPadders[i] != null && value != null && row.getRowKind() == RowKind.INSERT) {
+                value = charPadders[i].pad(value);
+            }
             // Actual Doris input is applied last so an explicit NULL remains distinct
             // from an omitted field and retains Paimon's writer-side semantics.
             row.setField(tableFieldIndexes[i], value);
         }
         return row;
     }
 
+    /**
+     * Returns the padder of a value of {@code type}, or null when the type holds no CHAR. A CHAR value is padded
+     * with spaces to the type length in code points, as Paimon's cast to CHAR pads it
+     * ({@code BinaryStringUtils.toCharacterString}), and so are the CHAR values nested in an ARRAY, a MAP or a
+     * ROW. Unlike that cast, a longer value is kept: Doris has already cut it to the length or kept it, as the
+     * session asks.
+     */
+    private static CharPadder charPadder(DataType type) {
+        switch (type.getTypeRoot()) {
+            case CHAR: {
+                int length = ((CharType) type).getLength();
+                return value -> padChar((BinaryString) value, length);
+            }
+            case ARRAY: {
+                DataType elementType = ((ArrayType) type).getElementType();
+                CharPadder elementPadder = charPadder(elementType);
+                if (elementPadder == null) {
+                    return null;
+                }
+                InternalArray.ElementGetter elementGetter = InternalArray.createElementGetter(elementType.nullable());
+                return value -> padArray((InternalArray) value, elementGetter, elementPadder);
+            }
+            case MAP: {
+                MapType mapType = (MapType) type;
+                CharPadder keyPadder = charPadder(mapType.getKeyType());
+                CharPadder valuePadder = charPadder(mapType.getValueType());
+                if (keyPadder == null && valuePadder == null) {
+                    return null;
+                }
+                InternalArray.ElementGetter keyGetter =
+                        InternalArray.createElementGetter(mapType.getKeyType().nullable());
+                InternalArray.ElementGetter valueGetter =
+                        InternalArray.createElementGetter(mapType.getValueType().nullable());
+                return value -> {
+                    InternalMap map = (InternalMap) value;
+                    return new PaddedMap(
+                            keyPadder == null ? map.keyArray() : padArray(map.keyArray(), keyGetter, keyPadder),
+                            valuePadder == null
+                                    ? map.valueArray() : padArray(map.valueArray(), valueGetter, valuePadder));
+                };
+            }
+            case ROW: {
+                List<DataField> fields = ((RowType) type).getFields();
+                CharPadder[] fieldPadders = new CharPadder[fields.size()];
+                InternalRow.FieldGetter[] fieldGetters = new InternalRow.FieldGetter[fields.size()];
+                boolean holdsChar = false;
+                for (int i = 0; i < fields.size(); i++) {
+                    fieldPadders[i] = charPadder(fields.get(i).type());
+                    fieldGetters[i] = InternalRow.createFieldGetter(fields.get(i).type().nullable(), i);
+                    holdsChar |= fieldPadders[i] != null;
+                }
+                if (!holdsChar) {
+                    return null;
+                }
+                return value -> {
+                    InternalRow nested = (InternalRow) value;
+                    GenericRow padded = new GenericRow(fieldGetters.length);
+                    for (int i = 0; i < fieldGetters.length; i++) {
+                        Object field = fieldGetters[i].getFieldOrNull(nested);
+                        padded.setField(i, field == null || fieldPadders[i] == null
+                                ? field : fieldPadders[i].pad(field));
+                    }
+                    return padded;
+                };
+            }
+            default:
+                return null;
+        }
+    }
+
+    private static BinaryString padChar(BinaryString value, int length) {
+        int padLength = length - value.numChars();
+        return padLength > 0 ? BinaryStringUtils.concat(value, BinaryString.blankString(padLength)) : value;
+    }
+
+    private static InternalArray padArray(InternalArray array, InternalArray.ElementGetter elementGetter,
+            CharPadder elementPadder) {
+        Object[] elements = new Object[array.size()];
+        for (int i = 0; i < elements.length; i++) {
+            Object element = elementGetter.getElementOrNull(array, i);
+            elements[i] = element == null ? null : elementPadder.pad(element);
+        }
+        return new GenericArray(elements);
+    }
+
+    /** Pads the CHAR values held by one non-null value of an input column. */
+    private interface CharPadder {
+        Object pad(Object value);
+    }
+
+    /** A map over padded key and value arrays. Unlike GenericMap it keeps every entry of the input map. */
+    private static final class PaddedMap implements InternalMap {
+        private final InternalArray keys;
+        private final InternalArray values;
+
+        private PaddedMap(InternalArray keys, InternalArray values) {
+            this.keys = keys;
+            this.values = values;
+        }
+
+        @Override
+        public int size() {
+            return keys.size();
+        }
+
+        @Override
+        public InternalArray keyArray() {
+            return keys;
+        }
+
+        @Override
+        public InternalArray valueArray() {
+            return values;
+        }
+    }
+
     private static RowKind toRowKind(Object operation) {
         if (!(operation instanceof Byte)) {
             throw new IllegalArgumentException("Paimon row change operation must be a TINYINT");