morningman commented on code in PR #69083:
URL: https://github.com/apache/doris/pull/69083#discussion_r4291870773
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java:
##########
@@ -167,9 +287,355 @@ GenericRow tableRow(InternalRow columnValues) {
// from an omitted field and retains Paimon's writer-side
semantics.
row.setField(tableFieldIndexes[i], value);
}
+ if (padsRow(row)) {
+ padRow(row);
+ }
return row;
}
+ /**
+ * Whether to pad {@code row}: whether 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, 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 (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;
+ }
+ // 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. Padding a shorter key
+ // here would make 'k' and 'k ' one key, or let a NOT MATCHED row
replace the stored 'k ' it did not
+ // match. A key as long as the column is not changed by padding.
+ if (insertedKeyCharLengths[i] > 0 && ((BinaryString)
value).numChars() < insertedKeyCharLengths[i]) {
+ 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",
+ value, insertedKeyCharLengths[i],
inputType.getFields().get(i).name(),
+ insertedKeyCharLengths[i]));
+ }
+ row.setField(tableFieldIndexes[i], charPadders[i].pad(value));
+ }
+ }
+
+ /**
+ * 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: a map nested in a key is
compared regardless of the order of its
+ * entries. 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 sorter of a value of {@code type}, or null when the type
holds no MAP. The sorter copies the value
+ * with the entries of each MAP in it, at any depth, sorted by the bytes
of their keys: two values whose maps hold
+ * the same entries in another order, which Paimon takes as equal, then
have the same bytes.
+ */
+ private static EntrySorter entrySorter(DataType type) {
+ switch (type.getTypeRoot()) {
+ case ARRAY: {
+ DataType elementType = ((ArrayType) type).getElementType();
+ EntrySorter elementSorter = entrySorter(elementType);
+ if (elementSorter == null) {
+ return null;
+ }
+ InternalArray.ElementGetter elementGetter =
InternalArray.createElementGetter(elementType.nullable());
+ return value -> new GenericArray(sortElements((InternalArray)
value, elementGetter, elementSorter));
+ }
+ case MAP: {
+ MapType mapType = (MapType) type;
+ EntrySorter keySorter = entrySorter(mapType.getKeyType());
+ EntrySorter valueSorter = entrySorter(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 = sortElements(map.keyArray(), keyGetter,
keySorter);
+ Object[] values = sortElements(map.valueArray(),
valueGetter, valueSorter);
+ 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;
+ }
+ Arrays.sort(order, (left, right) ->
compareBytes(keyBytes[left], keyBytes[right]));
+ Object[] sortedKeys = new Object[keys.length];
+ Object[] sortedValues = new Object[keys.length];
+ for (int i = 0; i < keys.length; i++) {
+ sortedKeys[i] = keys[order[i]];
+ sortedValues[i] = values[order[i]];
+ }
+ return new PaddedMap(new GenericArray(sortedKeys), new
GenericArray(sortedValues));
+ };
+ }
+ case ROW: {
+ List<DataField> fields = ((RowType) type).getFields();
+ EntrySorter[] fieldSorters = new EntrySorter[fields.size()];
+ InternalRow.FieldGetter[] fieldGetters = new
InternalRow.FieldGetter[fields.size()];
+ boolean holdsMap = false;
+ for (int i = 0; i < fields.size(); i++) {
+ fieldSorters[i] = entrySorter(fields.get(i).type());
+ fieldGetters[i] =
InternalRow.createFieldGetter(fields.get(i).type().nullable(), i);
+ holdsMap |= fieldSorters[i] != null;
+ }
+ if (!holdsMap) {
+ return null;
+ }
+ return value -> {
+ InternalRow nested = (InternalRow) value;
+ GenericRow sorted = new GenericRow(fieldGetters.length);
+ for (int i = 0; i < fieldGetters.length; i++) {
+ Object field = fieldGetters[i].getFieldOrNull(nested);
+ sorted.setField(i, field == null || fieldSorters[i] ==
null
+ ? field : fieldSorters[i].sort(field));
+ }
+ return sorted;
+ };
+ }
+ default:
+ return null;
+ }
+ }
+
+ /** The elements of {@code array}, each non-null one sorted by {@code
sorter} unless the sorter is null. */
+ private static Object[] sortElements(InternalArray array,
InternalArray.ElementGetter elementGetter,
+ EntrySorter sorter) {
+ Object[] elements = new Object[array.size()];
+ for (int i = 0; i < elements.length; i++) {
+ Object element = elementGetter.getElementOrNull(array, i);
+ elements[i] = element == null || sorter == null ? element :
sorter.sort(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 its Avro reader
+ * keeps one of two keys whose nested maps differ only in that order, so
the entries of the maps nested in a key
+ * are sorted before the key is compared.
+ */
+ private static final class MapKeyFormat {
+ private final InternalRowSerializer serializer;
+ // Null when the key type holds no MAP.
+ private final EntrySorter entrySorter;
+ private final CastExecutor<Object, Object> toText;
+
+ @SuppressWarnings("unchecked")
+ private MapKeyFormat(DataType keyType) {
+ this.serializer = new InternalRowSerializer(keyType);
+ this.entrySorter = entrySorter(keyType);
+ this.toText = (CastExecutor<Object, Object>)
CastExecutors.resolveToString(keyType);
+ }
+
+ /**
+ * The bytes of {@code key}, with the entries of its nested maps
sorted, as the only field of a row, copied
+ * since the serializer reuses its buffer.
+ */
+ private BinaryRow bytes(Object key) {
+ Object sorted = key == null || entrySorter == null ? key :
entrySorter.sort(key);
+ return serializer.toBinaryRow(GenericRow.of(sorted)).copy();
Review Comment:
Fixed in `c4acf1b45e1`. This reproduces as described. In an Avro table
created through the Paimon API with `m MAP<ROW<c CHAR(2), d DOUBLE>, INT>`,
`INSERT ... SELECT 1, map(named_struct('c', 'k', 'd', cast(pos AS DOUBLE)), 10,
named_struct('c', 'k ', 'd', cast(neg AS DOUBLE)), 20)` from a Doris table
holding `'NaN'` and `'-NaN'` succeeded, and Doris (native and JNI reads) read
back the single entry `{"c":"k ", "d":NaN}: 20`; a FLOAT field behaves the
same. A Parquet table kept both entries, two keys that Paimon takes as equal.
With padding off both keys are kept, as before this PR, and the same pair with
two positive NaNs already failed with the intended error.
Before the check serializes a key, it now also makes each FLOAT and DOUBLE
NaN in the key, at any depth, the canonical NaN, besides sorting the entries of
its nested maps. -0.0 stays apart from 0.0, as `Double.equals` keeps them apart
for Paimon. All three tables now fail that write with `turns the keys '{k,
NaN}' and '{k , NaN}' into the same key '{k , NaN}'`. Keys that were one key
before padding, such as `('k ', NaN)` and `('k ', -NaN)`, are still written as
they are; an Avro read keeps one of them with padding on or off, as on master.
Tests:
`PaimonWriteSchemaTest.testNaNsInMapKeysAreComparedRegardlessOfTheirBits` feeds
rows read back from an Arrow batch: NaNs of opposite signs collide in a DOUBLE
field, a FLOAT field and a nested ROW, keys equal before padding are kept, and
`('k', 0.0)` and `('k ', -0.0)` are padded and kept. Not normalizing the NaNs,
or only the DOUBLE ones, or making -0.0 into 0.0, turns it red.
`test_paimon_write_char_padding` now creates such an Avro table through Spark,
fails the pair cast at run time from a Doris table and pads a single key
holding a NaN; with the paimon-scanner jar from before this commit the suite
fails at that write.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
