morningman commented on code in PR #69083:
URL: https://github.com/apache/doris/pull/69083#discussion_r4285192374
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java:
##########
@@ -167,9 +271,189 @@ GenericRow tableRow(InternalRow columnValues) {
// from an omitted field and retains Paimon's writer-side
semantics.
row.setField(tableFieldIndexes[i], value);
}
+ if (insertsRow(row)) {
+ padInsertedRow(row);
+ }
return row;
}
+ /**
+ * Whether {@code row} inserts, the only kind of row that 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. A table with a
rowkind.field takes the kind of a row
+ * from that field's value when Paimon writes it ({@code TableWriteImpl}),
so only its '+I' rows insert: a '-D',
+ * '-U' or '+U' row carries a key to match, as an UPDATE or DELETE row
does.
+ */
+ private boolean insertsRow(GenericRow row) {
+ return row.getRowKind() == RowKind.INSERT
+ && (rowKindFieldIndex < 0 ||
INSERT_ROW_KIND.equals(row.getField(rowKindFieldIndex)));
Review Comment:
Fixed in `1740cfa4637`. This reproduces: in a Parquet table keyed by `c
CHAR(3)` with `rowkind.field = rk`, `('k', '+i', 1)` was written unpadded, and
so was `('j', NULL, 2)` when `rk` has `DEFAULT '+I'`, so a later `'+I'` of each
key added a second row `'k '` or `'j '`. Paimon fills a NULL with the field's
default (`DefaultValueRow`) and decodes the value ignoring case
(`RowKind.fromShortString`) before it takes the kind of the row
(`TableWriteImpl#writeAndReturn`), while the writer compared the value with
`'+I'` byte by byte. It now decodes the kind with Paimon's own
`DefaultValueRow` and `RowKindGenerator`, so `'+i'`, and a NULL or an omitted
value under a `'+I'` or `'+i'` default, insert and are padded. A value Paimon
cannot decode is left for Paimon's write to reject, as before.
Tests:
`PaimonWriteSchemaTest.testRowKindFieldValueDecidesWhetherARowIsPadded` now
covers `'+i'`, `'-d'`, an unknown value, and an explicit and an omitted NULL
under `'+I'`, `'+i'` and `'-D'` defaults; not filling the default turns it red.
`test_paimon_write_char_padding` writes `'+i'` and a NULL under `DEFAULT '+I'`,
then `'+I'` rows of the same keys, and gets one padded row per key.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java:
##########
@@ -167,9 +271,189 @@ GenericRow tableRow(InternalRow columnValues) {
// from an omitted field and retains Paimon's writer-side
semantics.
row.setField(tableFieldIndexes[i], value);
}
+ if (insertsRow(row)) {
+ padInsertedRow(row);
+ }
return row;
}
+ /**
+ * Whether {@code row} inserts, the only kind of row that 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. A table with a
rowkind.field takes the kind of a row
+ * from that field's value when Paimon writes it ({@code TableWriteImpl}),
so only its '+I' rows insert: a '-D',
+ * '-U' or '+U' row carries a key to match, as an UPDATE or DELETE row
does.
+ */
+ private boolean insertsRow(GenericRow row) {
+ return row.getRowKind() == RowKind.INSERT
+ && (rowKindFieldIndex < 0 ||
INSERT_ROW_KIND.equals(row.getField(rowKindFieldIndex)));
+ }
+
+ private void padInsertedRow(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());
+ return value -> {
+ InternalMap map = (InternalMap) value;
+ return new PaddedMap(
+ keyPadder == null ? map.keyArray() :
padMapKeys(map.keyArray(), keyGetter, keyPadder,
+ 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.
+ */
+ private static InternalArray padMapKeys(InternalArray keys,
InternalArray.ElementGetter keyGetter,
+ CharPadder keyPadder, String column) {
+ Object[] padded = new Object[keys.size()];
+ Map<Object, Object> keyOfPadded = new HashMap<>();
+ for (int i = 0; i < padded.length; i++) {
+ Object key = keyGetter.getElementOrNull(keys, i);
+ padded[i] = key == null ? null : keyPadder.pad(key);
+ if (!keyOfPadded.containsKey(padded[i])) {
+ keyOfPadded.put(padded[i], key);
Review Comment:
Fixed in `1740cfa4637`. This reproduces, and for more than this shape: the
writer's rows always come from the Arrow batch, so a ROW key holding a ROW is a
columnar row whose `hashCode` throws, and even a map of one such key in
`MAP<ROW<c CHAR(2), r ROW<i INT>>, INT>` failed the row with `ColumnarRowData
do not support hashCode`, whether the source was a Paimon table read natively
or through JNI, or a Doris literal. Two ROW or ARRAY keys that padding made
equal hit the unsupported `equals` instead of the intended error, and two ROW
keys holding arrays were not caught at all, since `ColumnarArray` hashes by
identity; an Avro table then kept only one of them. Keys are now compared by
their bytes as the only field of a `BinaryRow` (`InternalRowSerializer`), and
the error prints them as Paimon casts them to STRING, since `toString` of a
columnar row calls the same `hashCode`: `turns the keys '{k, 1}' and '{k , 1}'
into the same key '{k , 1}'`.
Tests: `PaimonWriteSchemaTest.testMapKeysOfArrowRowsAreComparedByTheirBytes`
feeds rows read back from an Arrow batch: a ROW-of-ROW key is padded, and
colliding ROW, ARRAY and ROW-with-ARRAY keys fail with the intended message.
Hashing the key objects, or printing them with `toString`, turns it red.
`test_paimon_write_char_padding` writes a map with a ROW-of-ROW key into a
Spark-created table and fails a pair of such keys that padding makes equal;
with the paimon-scanner jar from before this commit the suite fails at that
first write.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java:
##########
@@ -126,10 +156,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 &&
!canonicalName.equals(rowKindField)) {
+ value = padChar(value, ((CharType) type).getLength());
Review Comment:
Fixed in `1740cfa4637`. This reproduces, and for more than `'+U'`: in a
table keyed by `(id, p)`, partitioned by `p CHAR(3)` and with `rowkind.field =
rk`, a static overwrite of `p = 'x'` failed with `Trying to overwrite partition
{p=x }, but the changes in p=x/ does not belong to this partition` for a
`'+U'` row, and likewise for a `'-D'` row, in an ORC table, and for a `'+I'`
and a `'-D'` of one key in one statement; with padding off all of them worked.
An overwrite replaces the partitions it writes (`withIgnorePreviousFiles`), so
none of its rows matches a stored key, and the reason the previous commit gave
for leaving `'-D'`, `'-U'` and `'+U'` rows unpadded does not hold there. The
writer now pads every row of an overwrite, as FE pads the static partition
value of its filter: the `'+U'` row lands in `'x '`, and a `'+I'` and a `'-D'`
of one key in the overwrite still cancel out. Writes other than overwrites are
unchanged.
Tests: `PaimonWriteSchemaTest.testOverwritePadsEveryRowOfARowKindFieldTable`
(the `'+I'`, `'+U'`, `'-D'` and `'-U'` rows of an overwrite are padded, only
the partition value of an ORC table, and the `'+U'` row of a plain INSERT is
not); padding by the decoded row kind in an overwrite turns it red.
`test_paimon_write_char_padding` overwrites `p = 'x'` with a `'+U'` row, which
reads back as `[x ]`, and `p = 'y'` with a `'+I'` and a `'-D'` of one key,
which leaves the partition empty.
--
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]
