github-actions[bot] commented on code in PR #69083:
URL: https://github.com/apache/doris/pull/69083#discussion_r4291993215
##########
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:
[P2] Reduce duplicate descendant keys before comparing padded map keys
For an Avro `MAP<ROW<c CHAR(2), m MAP<DOUBLE,INT>>,INT>`, insert outer keys
`ROW('k',map(CAST(pos AS DOUBLE),1,CAST(neg AS DOUBLE),2)) -> 10` and `ROW('k
',map(CAST(pos AS DOUBLE),2)) -> 20`, reading VARCHAR `pos='NaN', neg='-NaN'`
at run time. Doris's `map()` deduplicates serialized bytes, so the signed NaNs
remain separate through Arrow. This normalizer canonicalizes their bits but
keeps both entries, so the padded keys still compare differently (two inner
entries versus one). [Paimon's Avro
reader](https://github.com/apache/paimon/blob/release-1.4.2/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java#L452-L483)
reduces the first inner map to `{NaN:2}`, then considers the outer keys equal
and drops value 10. With padding off their CHAR fields differ and both outer
values survive. This differs from the existing scalar-NaN and entry-order
threads: equal descendant keys need last-wins reduction in the comparison copy
before sorting/serialization, or eq
uivalent collision rejection. Cover this two-entry versus one-entry case
through Arrow and Avro.
##########
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:
[P2] Preserve native routing for unpadded ORC CHAR bucket keys
For an unpartitioned ORC table `(c CHAR(3), v INT)` with primary key `c`,
`bucket=4` and no key default, `PaimonWriteSchema.create` installs no padders:
ORC's nonpartition fields are exempt. The new default-on flag nevertheless
rejects its CHAR bucket key here, making `fixedBucketDistribution` return null
and `requiresSingleWriter` select GATHER. A distributed `INSERT ... SELECT`
therefore goes through one sink instance instead of the existing parallel
bucket writers, even though none of its routing bytes changes. This differs
from the earlier routing thread, whose padded fields actually changed bucket
identity. Make routing eligibility follow the writer's per-field padding
eligibility, preserving native routing for this unchanged ORC case, and add an
ORC padding-on distribution test.
--
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]
