morningman commented on code in PR #69083:
URL: https://github.com/apache/doris/pull/69083#discussion_r4299008914
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java:
##########
@@ -126,10 +153,13 @@ static Map<String, String>
resolveStaticPartition(FileStoreTable table, Connecto
}
String value = Objects.requireNonNull(castValues.get(key),
() -> "missing the cast value of static partition column "
+ key);
- if (table.rowType().getField(canonicalName).type().getTypeRoot()
- == DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
+ DataType type = table.rowType().getField(canonicalName).type();
+ if (type.getTypeRoot() ==
DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
value = toSdkLocalTime(canonicalName, value);
}
+ if (charPadding && type.getTypeRoot() == DataTypeRoot.CHAR) {
+ value = padChar(value, ((CharType) type).getLength());
Review Comment:
A correction to the second paragraph of my reply above. With `1012317776a`
this table is no longer written by one writer: its CHAR partition column is the
`rowkind.field`, which the writer never pads, so it is routed natively. And
since #69010, fe-core runs the transaction of an external `INSERT OVERWRITE`
even when planning proves its source empty. On a cluster with both, `INSERT
OVERWRITE TABLE t PARTITION (rk = '+I') SELECT 3 LIMIT 0` now commits an
OVERWRITE snapshot and clears the partition, as a source that is empty at run
time does; the regression still uses the latter.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java:
##########
@@ -167,9 +287,369 @@ 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, 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: 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.
+ */
+ 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;
+ }
+ Arrays.sort(order, (left, right) ->
compareBytes(keyBytes[left], keyBytes[right]));
+ Object[] sortedKeys = new Object[keys.length];
Review Comment:
Fixed in `41d351b3b90`. This reproduces as described. In an Avro table
created through the Paimon API with `m MAP<ROW<c CHAR(2), mm MAP<DOUBLE, INT>>,
INT>`, `INSERT ... SELECT 1, map(named_struct('c', 'k', 'mm', map(cast(pos AS
DOUBLE), 1, cast(neg AS DOUBLE), 2)), 10, named_struct('c', 'k ', 'mm',
map(cast(pos AS DOUBLE), 2)), 20)` from a Doris table holding `'NaN'` and
`'-NaN'` succeeded, and Doris read back the single entry `{"c":"k ",
"mm":{NaN:2}}: 20`; a Parquet table kept both entries, two keys that Paimon
takes as equal. With padding off both entries are kept, as before this PR.
The copy of a nested map that the check compares now keeps, once its NaNs
are canonical and its entries sorted by their key bytes, only the last of the
entries whose keys are equal, the one whose value Paimon's Avro reader keeps
(`ArrayMapReader` puts the entries into a `HashMap` in order). Both tables now
fail that write with `turns the keys '{k, {NaN -> 1, NaN -> 2}}' and '{k , {NaN
-> 2}}' into the same key '{k , {NaN -> 2}}'`. Keys whose nested maps keep
other last values, such as `(k, {NaN: 1, -NaN: 2})` and `(k , {NaN: 1})`, are
padded and written, and read back as two entries; the nested maps are written
whole, as before.
Tests:
`PaimonWriteSchemaTest.testMapsNestedInMapKeysAreComparedByTheLastEntryOfEachKey`
feeds rows read back from an Arrow batch: the pair above collides, keys that
were one key before padding are kept, and keys whose nested maps keep other
last values are padded, with their nested maps whole. Not reducing the nested
entries, keeping the first of them instead of the last, or comparing the nested
maps by their keys alone turns it red. `test_paimon_write_char_padding` creates
such an Avro table through Spark, fails the pair cast at run time from a Doris
table and pads the other pair; with the paimon-scanner jar from before this
commit the suite fails at that write.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java:
##########
@@ -409,16 +428,17 @@ private List<Integer> appendRouteFields(List<String>
fieldNames,
return indexes;
}
- private boolean supportsNativeRouting(DataTypeRoot type) {
+ private boolean supportsNativeRouting(DataTypeRoot type, boolean
charPadding) {
switch (type) {
+ case CHAR:
+ return !charPadding;
Review Comment:
Fixed in `1012317776a`. Confirmed: for a primary-key ORC table `(c CHAR(3),
v INT)` with `bucket = 4`, `EXPLAIN SHAPE PLAN INSERT ... SELECT ... FROM
numbers("number" = "100")` planned `DistributionSpecGather` with padding on and
`DistributionSpecExternalTableSinkHashPartitioned` with it off, though the
writer leaves that key unpadded.
FE now decides it per routing field by the writer's rule, in
`PaimonWriteBinding.charPaddedFields`: with padding on, the writer pads every
CHAR field but the table's `rowkind.field`, and of a table that writes ORC
files, by `file.format` or at some level of `file.format.per.level`, only the
partition fields. A CHAR routing field it pads still falls back, as before, and
the others are routed natively. The static partition values follow the same
rule, which leaves them as they were. The remark in r4279897764 that FE does
not look at the file format no longer holds. With padding on, the ORC table
above and one with `file.format.per.level = '5:orc'` now plan
`DistributionSpecExternalTableSinkHashPartitioned`, while a Parquet table and
an ORC table partitioned by a CHAR field still plan `DistributionSpecGather`.
Writing the ORC table with `changelog-producer = full-compaction` from
`numbers()` with four writers commits an APPEND and a COMPACT snapshot with 100
changelog records, with n
o LSM conflict, and a DELETE and an UPDATE then reach its keys.
Tests:
`PaimonWritePlanProviderTest.charRouteFieldsThatTheWriterLeavesUnpaddedAreRoutedNatively`
routes, with padding on, the CHAR key of an ORC table, by `file.format` and by
`file.format.per.level`, and a CHAR partition that is the `rowkind.field`
natively, and the CHAR partition of an ORC table through one writer;
`PaimonWriteBindingTest.staticCharPartitionOfAnOrcTableIsPaddedAsTheRowsAre`
keeps the static value of that partition padded. Falling back for every CHAR
routing field, ignoring `file.format.per.level` or the `rowkind.field`, or
routing the partition of an ORC table natively and leaving it unpadded turns
them red. `test_paimon_write_char_padding` checks the plan of such a
full-compaction ORC table, next to the Parquet one's, and its `$snapshots`
after a four-writer write; with the fe-connector-paimon jar from before this
commit the suite fails at that plan.
--
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]
