PaimonWriteSchema.java
// 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.doris.paimon;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.casting.CastExecutor;
import org.apache.paimon.casting.CastExecutors;
import org.apache.paimon.casting.DefaultValueRow;
import org.apache.paimon.crosspartition.IndexBootstrap;
import org.apache.paimon.data.BinaryRow;
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.data.serializer.InternalRowSerializer;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.sink.RowKindGenerator;
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.DataTypeRoot;
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.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.UnaryOperator;
/**
* Immutable mapping from Doris input columns to a Paimon table row.
*
* <p>The input may contain a subset of table columns in a different order. This class
* resolves their types and table positions once, then converts each input row to the
* table-schema layout expected by the Paimon writer.
*/
final class PaimonWriteSchema {
static final String ROW_KIND_COLUMN = "__DORIS_PAIMON_ROW_KIND__";
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 static final BinaryString SPACE = BinaryString.fromString(" ");
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;
/**
* Per input column, the length of a padded CHAR primary-key column of a changelog write, or 0. Padding a key
* that MERGE inserts can change it after MERGE matched it and checked it for duplicates; see {@link #padRow}.
*/
private final int[] insertedKeyCharLengths;
/** Per input column with an inserted key length, how Doris reads its stored keys back; see {@link #padRow}. */
private final KeyReadBack[] insertedKeyReadBacks;
/** Which data files of the table are ORC files ({@link #orcFiles}); see {@link #padsRow}. */
private final OrcFiles orcFiles;
/** Decodes the kind of a row from the table's rowkind.field as Paimon does when it writes the row, or null. */
private final RowKindGenerator rowKindGenerator;
/**
* Fills a NULL field with its default value, as Paimon does before it decodes the row kind; null when the table
* has no default value.
*/
private final DefaultValueRow defaultValueRow;
/** Whether the write overwrites the partitions it writes; see {@link #padsRow}. */
private final boolean overwrite;
private final RowType inputType;
/** Maps Doris input-column position ��� Paimon table-schema position. */
private final int[] tableFieldIndexes;
/** Paimon defaults for table fields omitted from the Doris input. */
private final int[] omittedDefaultFieldIndexes;
private final Object[] omittedDefaultValues;
private final int tableFieldCount;
private PaimonWriteSchema(int[] tableFieldIndexes, int[] omittedDefaultFieldIndexes,
Object[] omittedDefaultValues, int tableFieldCount, RowType inputType, CharPadder[] charPadders,
int[] insertedKeyCharLengths, KeyReadBack[] insertedKeyReadBacks, OrcFiles orcFiles,
RowKindGenerator rowKindGenerator, DefaultValueRow defaultValueRow, boolean overwrite) {
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.insertedKeyCharLengths = insertedKeyCharLengths;
this.insertedKeyReadBacks = insertedKeyReadBacks;
this.orcFiles = orcFiles;
this.rowKindGenerator = rowKindGenerator;
this.defaultValueRow = defaultValueRow;
this.overwrite = overwrite;
this.inputType = inputType;
this.tableFieldIndexes = tableFieldIndexes;
this.omittedDefaultFieldIndexes = omittedDefaultFieldIndexes;
this.omittedDefaultValues = omittedDefaultValues;
this.tableFieldCount = tableFieldCount;
}
/**
* Create the write schema by resolving {@code columnNames} against the
* Paimon table schema.
*
* @param tableType full Paimon table row type (all columns in table order)
* @param columnNames output column names from BE (in Doris output order)
* @return immutable schema metadata for this writer session
* @throws IllegalArgumentException if any column name is not found in the table schema
*/
static PaimonWriteSchema create(RowType tableType, String[] columnNames) {
return create(tableType, columnNames, false);
}
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);
}
static PaimonWriteSchema create(RowType tableType, String[] columnNames, boolean changelogWrite,
boolean charPadding, String rowKindField) {
return create(tableType, columnNames, changelogWrite, charPadding, rowKindField, Collections.emptyList());
}
static PaimonWriteSchema create(RowType tableType, String[] columnNames, boolean changelogWrite,
boolean charPadding, String rowKindField, List<String> primaryKeys) {
return create(tableType, columnNames, changelogWrite, charPadding, rowKindField, primaryKeys,
Collections.emptyList(), OrcFiles.NONE, false);
}
/**
* @param charPadding whether to pad CHAR values as Paimon's cast to CHAR does; see {@link #charPadder}, and
* {@link #padsRow} for the rows that are padded
* @param rowKindField the table's {@code rowkind.field}, or null. Its value is not padded: Paimon decodes the row
* kind from it ({@code RowKind.fromShortString}), and a padded '+I ' is no row kind.
* @param primaryKeys the table's primary keys. In a changelog write, a CHAR key that MERGE inserts fails the
* row when padding could make it a key that MERGE did not see; see {@link #padRow}.
* @param partitionKeys the table's partition keys, whose values Doris reads back as they are stored
* @param orcFiles which data files of the table are ORC files ({@link #orcFiles}); every row of a table
* that writes some is padded, see {@link #padsRow}
* @param overwrite whether the write overwrites the partitions it writes; see {@link #padsRow}
*/
static PaimonWriteSchema create(RowType tableType, String[] columnNames, boolean changelogWrite,
boolean charPadding, String rowKindField, List<String> primaryKeys, List<String> partitionKeys,
OrcFiles orcFiles, boolean overwrite) {
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];
int[] insertedKeyCharLengths = new int[columnNames.length];
KeyReadBack[] insertedKeyReadBacks = new KeyReadBack[columnNames.length];
List<DataField> inputFields = new ArrayList<>(columnNames.length);
boolean[] specifiedFields = new boolean[tableType.getFieldCount()];
for (int i = 0; i < columnNames.length; i++) {
if (changelogWrite && i == 0) {
if (!ROW_KIND_COLUMN.equals(columnNames[i])) {
throw new IllegalArgumentException(
"Paimon changelog write requires row kind as the first column");
}
tableFieldIndexes[i] = -1;
inputFields.add(new DataField(
Integer.MIN_VALUE, ROW_KIND_COLUMN, new TinyIntType(true)));
continue;
}
int tableIndex = tableType.getFieldIndex(columnNames[i]);
if (tableIndex < 0) {
throw new IllegalArgumentException(
"Paimon column '" + columnNames[i] + "' not found in table schema");
}
if (specifiedFields[tableIndex]) {
throw new IllegalArgumentException(
"Duplicate Paimon write column '" + columnNames[i] + "'");
}
specifiedFields[tableIndex] = true;
DataField field = tableType.getFields().get(tableIndex);
tableFieldIndexes[i] = tableIndex;
// Input nullability is independent from the target table constraint. Doris can still
// produce an explicit NULL for a NOT NULL target; carrying it through Arrow lets the
// 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(), field.name());
if (changelogWrite && field.type().getTypeRoot() == DataTypeRoot.CHAR
&& primaryKeys.contains(field.name())) {
insertedKeyCharLengths[i] = ((CharType) field.type()).getLength();
insertedKeyReadBacks[i] = partitionKeys.contains(field.name()) || orcFiles == OrcFiles.NONE
? KeyReadBack.PADDED
: orcFiles == OrcFiles.ALL ? KeyReadBack.TRIMMED : KeyReadBack.PADDED_OR_TRIMMED;
}
}
}
int[] omittedDefaultFieldIndexes = new int[tableType.getFieldCount()];
Object[] omittedDefaultValues = new Object[tableType.getFieldCount()];
int omittedDefaultCount = 0;
for (int tableIndex = 0; tableIndex < tableType.getFieldCount(); tableIndex++) {
DataField field = tableType.getFields().get(tableIndex);
if (specifiedFields[tableIndex] || field.defaultValue() == null) {
continue;
}
omittedDefaultFieldIndexes[omittedDefaultCount] = tableIndex;
omittedDefaultValues[omittedDefaultCount] =
DefaultValueUtils.convertDefaultValue(field.type(), field.defaultValue());
omittedDefaultCount++;
}
return new PaimonWriteSchema(
tableFieldIndexes,
Arrays.copyOf(omittedDefaultFieldIndexes, omittedDefaultCount),
Arrays.copyOf(omittedDefaultValues, omittedDefaultCount),
tableType.getFieldCount(),
new RowType(inputFields),
charPadders,
insertedKeyCharLengths,
insertedKeyReadBacks,
orcFiles,
rowKindField == null ? null : new RowKindGenerator(rowKindField, tableType),
DefaultValueRow.create(tableType),
overwrite);
}
/**
* 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));
}
/**
* Which data files of the table are ORC files: Paimon writes a file of a level in the format that
* {@code file.format.per.level} names for the level, or else in {@code file.format}.
*/
static OrcFiles orcFiles(Map<String, String> tableOptions) {
CoreOptions options = CoreOptions.fromMap(tableOptions);
Map<Integer, String> formatPerLevel = options.fileFormatPerLevel();
boolean orc = false;
boolean other = false;
for (int level = 0; level < options.numLevels(); level++) {
if (CoreOptions.FILE_FORMAT_ORC.equals(formatPerLevel.getOrDefault(level, options.formatType()))) {
orc = true;
} else {
other = true;
}
}
return !orc ? OrcFiles.NONE : other ? OrcFiles.SOME : OrcFiles.ALL;
}
/** Paimon input fields in the exact order transported by Arrow C Data. */
RowType inputType() {
return inputType;
}
/** Expand one input row to the full Paimon table-schema layout. */
GenericRow tableRow(InternalRow columnValues) {
if (columnValues.getFieldCount() != tableFieldIndexes.length) {
throw new IllegalArgumentException(
"Paimon input value count does not match write schema");
}
GenericRow row = new GenericRow(tableFieldCount);
for (int i = 0; i < omittedDefaultFieldIndexes.length; i++) {
row.setField(omittedDefaultFieldIndexes[i], omittedDefaultValues[i]);
}
for (int i = 0; i < tableFieldIndexes.length; i++) {
Object value = fieldGetters[i].getFieldOrNull(columnValues);
if (tableFieldIndexes[i] < 0) {
row.setRowKind(toRowKind(value));
continue;
}
// 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);
}
if (padsRow(row)) {
padRow(row);
}
return row;
}
/**
* Returns what gives a row that Paimon's IndexBootstrap reads from the key-dynamic table of {@code schema}, to
* index the keys stored in it, those keys as this writer writes them. Such a row holds the primary keys that are
* no partition keys as the data files give them back, then the partition and the bucket
* ({@code IndexBootstrap#bootstrapType}). ORC gives a CHAR key back without the trailing spaces that every row
* this writer writes into a table that writes ORC files carries ({@link #padsRow}), so an index of the keys as
* read would hold 'k' for the stored 'k ', and a row of 'k' written into another partition would miss it and
* leave the stored row in place, two rows of one key. Those keys are padded as the rows are; the other files give
* the keys back as they are stored, and the partition is the stored one.
*/
UnaryOperator<InternalRow> bootstrapKeyPadder(TableSchema schema) {
List<String> keys = schema.trimmedPrimaryKeys();
CharPadder[] keyPadders = new CharPadder[keys.size()];
boolean padsKeys = false;
if (orcFiles != OrcFiles.NONE) {
for (int i = 0; i < keyPadders.length; i++) {
// A key omitted from the input takes its default value unpadded, as the rows written do.
int column = inputType.getFieldIndex(keys.get(i));
keyPadders[i] = column < 0 ? null : charPadders[column];
padsKeys |= keyPadders[i] != null;
}
}
if (!padsKeys) {
return UnaryOperator.identity();
}
RowType bootstrapType = IndexBootstrap.bootstrapType(schema);
InternalRow.FieldGetter[] bootstrapFieldGetters = new InternalRow.FieldGetter[bootstrapType.getFieldCount()];
for (int i = 0; i < bootstrapFieldGetters.length; i++) {
bootstrapFieldGetters[i] = InternalRow.createFieldGetter(bootstrapType.getTypeAt(i), i);
}
return bootstrapRow -> {
GenericRow padded = new GenericRow(bootstrapFieldGetters.length);
for (int i = 0; i < bootstrapFieldGetters.length; i++) {
Object value = bootstrapFieldGetters[i].getFieldOrNull(bootstrapRow);
padded.setField(i, i < keyPadders.length && keyPadders[i] != null ? keyPadders[i].pad(value) : value);
}
return padded;
};
}
/**
* Whether to pad {@code row}. Every row of a table that writes ORC files, at some levels or at all, is padded.
* ORC pads every CHAR value it writes and strips the trailing spaces from one it reads, so Doris reads back 'k'
* for the 'k ' that Spark and Flink write. Paimon puts a row in the bucket of its key as the writer holds it,
* and merges only the data files whose key ranges, also kept as the writer held the keys, overlap: an UPDATE or
* DELETE row that wrote that key back as 'k' would land in another bucket, or below the range of the file
* holding 'k ', and miss the row.
*
* <p>On any other table, a row is padded when it inserts, outside an overwrite. Its row kind, if any, is the
* first input column. An UPDATE or DELETE row carries the key Doris read back from the table as it is stored,
* which has to stay byte-equal to the stored key for Paimon to match it, padded or not. A table with a
* rowkind.field takes the kind of a row from that field's value when Paimon writes it
* ({@code TableWriteImpl#writeAndReturn}), after filling a NULL with the field's default and ignoring case:
* '+I', '+i', or NULL under DEFAULT '+I' inserts, while a '-D', '-U' or '+U' row carries a key to match, as an
* UPDATE or DELETE row does. A value Paimon cannot decode, such as NULL without a default, fails the write in
* Paimon with the same error, after Paimon's own NOT NULL check, so such a row is left as it is. An overwrite
* replaces the partitions it writes, so none of its rows matches a stored key, and all of them are padded, like
* the static partition value FE pads into the overwrite filter: a '+U' row of INSERT OVERWRITE ... PARTITION
* (p = 'x') has to land in 'x ', and a '+I' and a '-D' of one key in the statement still cancel out.
*/
private boolean padsRow(GenericRow row) {
if (orcFiles != OrcFiles.NONE) {
return true;
}
if (row.getRowKind() != RowKind.INSERT) {
return false;
}
if (overwrite || rowKindGenerator == null) {
return true;
}
try {
return rowKindGenerator.generate(defaultValueRow == null ? row : defaultValueRow.replaceRow(row))
== RowKind.INSERT;
} catch (RuntimeException e) {
return false;
}
}
private void padRow(GenericRow row) {
for (int i = 0; i < charPadders.length; i++) {
Object value = charPadders[i] == null ? null : row.getField(tableFieldIndexes[i]);
if (value == null) {
continue;
}
if (insertedKeyCharLengths[i] > 0 && row.getRowKind() == RowKind.INSERT) {
checkInsertedKey((BinaryString) value, i);
}
row.setField(tableFieldIndexes[i], charPadders[i].pad(value));
}
}
/**
* In a changelog write only MERGE inserts rows, and MERGE has matched them to the table and checked them for
* duplicate keys by the key as it is: Doris compares CHAR values unpadded. So a key that padding could make a
* stored key that MERGE did not match, or the same key as another one MERGE inserts, fails the row, and which
* keys those are depends on how Doris reads the stored keys back ({@link KeyReadBack}). Read back padded, as the
* values of a partition column and the keys of a table that writes no ORC files are, a key shorter than the
* column, such as 'k', matched no stored 'k ' yet pads to it, and pads to the key 'k ' pads to; a key as long as
* the column is not changed by padding. ORC gives the keys back without their trailing spaces, so there MERGE
* matched 'k' to the stored 'k ', and only a key that ends with a space, such as 'k ', matches no stored key yet
* pads to one. A table that writes ORC files at some levels only gives a key back in either form, by the file
* that holds it, so both such keys fail the row.
*/
private void checkInsertedKey(BinaryString key, int column) {
int length = insertedKeyCharLengths[column];
String name = inputType.getFields().get(column).name();
KeyReadBack readBack = insertedKeyReadBacks[column];
if (readBack == KeyReadBack.PADDED_OR_TRIMMED) {
if (key.numChars() < length || key.endsWith(SPACE)) {
// The remedy comes first: the client gets a long error message of the writer cut short.
throw new IllegalArgumentException(String.format(
"MERGE cannot insert '%s' into the CHAR(%d) primary key column '%s' of a Paimon table "
+ "that writes ORC files at some levels only. Insert keys as long as the column "
+ "that do not end with a space, or set enable_paimon_write_char_padding = false: "
+ "the table gives its keys back padded from its other files and without their "
+ "trailing spaces from its ORC files, so MERGE matched this value against keys in "
+ "either form, and padding it can turn it into a stored key that MERGE did not "
+ "match",
key, length, name));
}
return;
}
if (readBack == KeyReadBack.TRIMMED) {
if (key.endsWith(SPACE)) {
// The remedy comes first: the client gets a long error message of the writer cut short.
throw new IllegalArgumentException(String.format(
"MERGE cannot insert '%s' into the CHAR(%d) primary key column '%s' of a Paimon table "
+ "that writes ORC files. Remove the trailing spaces of the key in the MERGE, with "
+ "rtrim(<key>) in both the ON condition and the inserted values, or set "
+ "enable_paimon_write_char_padding = false: ORC drops the trailing spaces of the "
+ "keys it stores, so MERGE matched this value against keys without them, and "
+ "padding it can turn it into a stored key that MERGE did not match",
key, length, name));
}
return;
}
if (key.numChars() < length) {
throw new IllegalArgumentException(String.format(
"MERGE cannot insert '%s' into the CHAR(%d) primary key column '%s' of a Paimon table: "
+ "the value is shorter than the column, and padding it to the column "
+ "length would change the key that MERGE matched. Pad the key in the "
+ "MERGE, with rpad(<key>, %d, ' ') in both the ON condition and the "
+ "inserted values, or set enable_paimon_write_char_padding = false",
key, length, name, length));
}
}
/**
* Returns the padder of a value of {@code type} in {@code column}, 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. A map whose keys padding would make equal fails the row; see {@link #padMapKeys}.
*/
private static CharPadder charPadder(DataType type, String column) {
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, column);
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(), column);
CharPadder valuePadder = charPadder(mapType.getValueType(), column);
if (keyPadder == null && valuePadder == null) {
return null;
}
InternalArray.ElementGetter keyGetter =
InternalArray.createElementGetter(mapType.getKeyType().nullable());
InternalArray.ElementGetter valueGetter =
InternalArray.createElementGetter(mapType.getValueType().nullable());
MapKeyFormat keyFormat = keyPadder == null ? null : new MapKeyFormat(mapType.getKeyType());
return value -> {
InternalMap map = (InternalMap) value;
return new PaddedMap(
keyPadder == null ? map.keyArray() : padMapKeys(map.keyArray(), keyGetter, keyPadder,
keyFormat, column),
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(), column);
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 keys of one map, failing when padding makes two of its distinct keys one: a map holds one entry per
* key, and Paimon's Avro reader, for one, keeps only one of them. Keys equal before padding are kept as they are.
* Keys are equal as Paimon compares them once its Avro reader has read them: a map nested in a key is compared
* regardless of the order of its entries and as holding only the last of the entries with equal keys, and a NaN
* regardless of its bits. Keys are compared and printed through {@code keyFormat}, never by their own hashCode,
* equals or toString: a key of a nested type holds the columnar rows and arrays of the Arrow batch, which support
* none of them (ColumnarArray keeps identity equality), so hashing the key of a
* MAP<ROW<c CHAR(2), r ROW<i INT>>, INT> itself would fail even a map of one entry.
*/
private static InternalArray padMapKeys(InternalArray keys, InternalArray.ElementGetter keyGetter,
CharPadder keyPadder, MapKeyFormat keyFormat, String column) {
Object[] padded = new Object[keys.size()];
// Bytes of a padded key -> position of the first key padded into it.
Map<BinaryRow, Integer> firstKeyOfPadded = new HashMap<>();
for (int i = 0; i < padded.length; i++) {
Object key = keyGetter.getElementOrNull(keys, i);
padded[i] = key == null ? null : keyPadder.pad(key);
Integer first = firstKeyOfPadded.putIfAbsent(keyFormat.bytes(padded[i]), i);
if (first == null) {
continue;
}
Object firstKey = keyGetter.getElementOrNull(keys, first);
if (!keyFormat.bytes(firstKey).equals(keyFormat.bytes(key))) {
throw new IllegalArgumentException(String.format(
"Padding the CHAR keys of a map in column '%s' of a Paimon table to the CHAR length turns "
+ "the keys '%s' and '%s' into the same key '%s', and a map holds one entry per key. "
+ "Make the keys distinct, or set enable_paimon_write_char_padding = false",
column, keyFormat.text(firstKey), keyFormat.text(key), keyFormat.text(padded[i])));
}
}
return new GenericArray(padded);
}
/**
* Returns the normalizer of a value of {@code type}, or null when the type holds no MAP, FLOAT or DOUBLE. The
* normalizer copies the value with each NaN in it, at any depth, made the canonical NaN, and the entries of each
* MAP in it sorted by the bytes of their keys, keeping only the last of the entries whose keys are then equal:
* two values that Paimon takes as equal, though their NaNs differ in their bits or their maps hold the same
* entries in another order, then have the same bytes. A NaN in a ROW is serialized with its bits, while Paimon
* compares a FLOAT or DOUBLE as {@code Double.equals} does, taking every NaN as one value; -0.0 and 0.0 stay
* apart, as they are for Paimon. Paimon's Avro reader puts the entries of a map in a hash map, which keeps the
* value of the last of the entries with equal keys: the map {NaN: 1, -NaN: 2}, which Doris's map() keeps whole,
* since its keys differ in their bits, is read back as {NaN: 2}.
*/
private static Normalizer normalizer(DataType type) {
switch (type.getTypeRoot()) {
case FLOAT:
return value -> Float.isNaN((Float) value) ? Float.NaN : value;
case DOUBLE:
return value -> Double.isNaN((Double) value) ? Double.NaN : value;
case ARRAY: {
DataType elementType = ((ArrayType) type).getElementType();
Normalizer elementNormalizer = normalizer(elementType);
if (elementNormalizer == null) {
return null;
}
InternalArray.ElementGetter elementGetter = InternalArray.createElementGetter(elementType.nullable());
return value -> new GenericArray(
normalizeElements((InternalArray) value, elementGetter, elementNormalizer));
}
case MAP: {
MapType mapType = (MapType) type;
Normalizer keyNormalizer = normalizer(mapType.getKeyType());
Normalizer valueNormalizer = normalizer(mapType.getValueType());
InternalArray.ElementGetter keyGetter =
InternalArray.createElementGetter(mapType.getKeyType().nullable());
InternalArray.ElementGetter valueGetter =
InternalArray.createElementGetter(mapType.getValueType().nullable());
InternalRowSerializer keySerializer = new InternalRowSerializer(mapType.getKeyType());
return value -> {
InternalMap map = (InternalMap) value;
Object[] keys = normalizeElements(map.keyArray(), keyGetter, keyNormalizer);
Object[] values = normalizeElements(map.valueArray(), valueGetter, valueNormalizer);
byte[][] keyBytes = new byte[keys.length][];
Integer[] order = new Integer[keys.length];
for (int i = 0; i < keys.length; i++) {
// Copied first: the serializer reuses its buffer, and toBytes returns that buffer itself when
// the row fills it.
keyBytes[i] = keySerializer.toBinaryRow(GenericRow.of(keys[i])).copy().toBytes();
order[i] = i;
}
// The sort is stable, so the entries of equal keys keep their order.
Arrays.sort(order, (left, right) -> compareBytes(keyBytes[left], keyBytes[right]));
List<Object> sortedKeys = new ArrayList<>(keys.length);
List<Object> sortedValues = new ArrayList<>(keys.length);
for (int i = 0; i < keys.length; i++) {
// Of the entries whose keys are equal, only the last one is kept, as Paimon's Avro reader
// keeps its value.
if (i + 1 < keys.length && Arrays.equals(keyBytes[order[i]], keyBytes[order[i + 1]])) {
continue;
}
sortedKeys.add(keys[order[i]]);
sortedValues.add(values[order[i]]);
}
return new PaddedMap(new GenericArray(sortedKeys.toArray()),
new GenericArray(sortedValues.toArray()));
};
}
case ROW: {
List<DataField> fields = ((RowType) type).getFields();
Normalizer[] fieldNormalizers = new Normalizer[fields.size()];
InternalRow.FieldGetter[] fieldGetters = new InternalRow.FieldGetter[fields.size()];
boolean normalized = false;
for (int i = 0; i < fields.size(); i++) {
fieldNormalizers[i] = normalizer(fields.get(i).type());
fieldGetters[i] = InternalRow.createFieldGetter(fields.get(i).type().nullable(), i);
normalized |= fieldNormalizers[i] != null;
}
if (!normalized) {
return null;
}
return value -> {
InternalRow nested = (InternalRow) value;
GenericRow copy = new GenericRow(fieldGetters.length);
for (int i = 0; i < fieldGetters.length; i++) {
Object field = fieldGetters[i].getFieldOrNull(nested);
copy.setField(i, field == null || fieldNormalizers[i] == null
? field : fieldNormalizers[i].normalize(field));
}
return copy;
};
}
default:
return null;
}
}
/**
* The elements of {@code array}, each non-null one normalized by {@code normalizer} unless the normalizer is
* null.
*/
private static Object[] normalizeElements(InternalArray array, InternalArray.ElementGetter elementGetter,
Normalizer normalizer) {
Object[] elements = new Object[array.size()];
for (int i = 0; i < elements.length; i++) {
Object element = elementGetter.getElementOrNull(array, i);
elements[i] = element == null || normalizer == null ? element : normalizer.normalize(element);
}
return elements;
}
/** Orders byte strings by their bytes, compared unsigned, then by their length. */
private static int compareBytes(byte[] left, byte[] right) {
for (int i = 0; i < Math.min(left.length, right.length); i++) {
int compared = Integer.compare(left[i] & 0xFF, right[i] & 0xFF);
if (compared != 0) {
return compared;
}
}
return Integer.compare(left.length, right.length);
}
/**
* Compares the keys of a map by their bytes and prints them as Paimon casts them to STRING, for any key type.
* Paimon compares maps regardless of the order of their entries ({@code GenericMap.equals}) and a FLOAT or DOUBLE
* regardless of the bits of a NaN ({@code GenericRow.equals}), and its Avro reader puts the entries of a map in a
* hash map, keeping one entry per key, the last one, at any depth. So a key is normalized before it is compared;
* see {@link #normalizer}.
*/
private static final class MapKeyFormat {
private final InternalRowSerializer serializer;
// Null when the key type holds no MAP, FLOAT or DOUBLE.
private final Normalizer normalizer;
private final CastExecutor<Object, Object> toText;
@SuppressWarnings("unchecked")
private MapKeyFormat(DataType keyType) {
this.serializer = new InternalRowSerializer(keyType);
this.normalizer = normalizer(keyType);
this.toText = (CastExecutor<Object, Object>) CastExecutors.resolveToString(keyType);
}
/**
* The bytes of {@code key}, normalized, as the only field of a row, copied since the serializer reuses its
* buffer.
*/
private BinaryRow bytes(Object key) {
Object normalized = key == null || normalizer == null ? key : normalizer.normalize(key);
return serializer.toBinaryRow(GenericRow.of(normalized)).copy();
}
/** {@code key} as Paimon casts it to STRING: a ROW key (k, 1) prints as {k, 1}. */
private String text(Object key) {
return String.valueOf(toText.cast(key));
}
}
/** Pads the CHAR values held by one non-null value of an input column. */
private interface CharPadder {
Object pad(Object value);
}
/**
* Which data files of a table are ORC files. ORC pads every CHAR value it writes and strips the trailing spaces
* from one it reads, while the other formats give a value back as it is stored.
*/
enum OrcFiles {
/** No level writes ORC files. */
NONE,
/** Some levels write ORC files, and the others another format. */
SOME,
/** Every level writes ORC files. */
ALL
}
/** How Doris reads back the stored values of a CHAR primary-key column, against which MERGE matched its keys. */
private enum KeyReadBack {
/**
* As they are stored, padded: the values of a partition column, which Paimon keeps in the partition, and the
* keys of a table that writes no ORC files.
*/
PADDED,
/** Without their trailing spaces: the keys of a table that writes only ORC files. */
TRIMMED,
/** Padded from the files of some levels and trimmed from the ORC files of the others. */
PADDED_OR_TRIMMED
}
/**
* Copies one non-null value of a type that holds a MAP, FLOAT or DOUBLE, with each NaN in it made the canonical
* NaN and the entries of each MAP in it sorted, one per key.
*/
private interface Normalizer {
Object normalize(Object value);
}
/**
* A map over the given key and value arrays, in their order: the padded entries of an input map, which keep the
* order of its entries, or the sorted entries, one per key, of a map nested in a key that is compared.
*/
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");
}
switch ((Byte) operation) {
case INSERT_OPERATION:
return RowKind.INSERT;
case UPDATE_OPERATION:
return RowKind.UPDATE_AFTER;
case DELETE_OPERATION:
return RowKind.DELETE;
default:
throw new IllegalArgumentException(
"Unknown Paimon row change operation: " + operation);
}
}
}