morningman commented on code in PR #69083:
URL: https://github.com/apache/doris/pull/69083#discussion_r4274463612
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -183,8 +183,8 @@ public void open(String serializedTable, Map<String,
String> hadoopConfig,
this.bucketMode = table.bucketMode();
CoreOptions coreOptions =
CoreOptions.fromMap(table.options());
- this.writeSchema = PaimonWriteSchema.create(
- table.rowType(), columnNames, changelogWrite);
+ this.writeSchema =
PaimonWriteSchema.create(table.rowType(), columnNames, changelogWrite,
+ PaimonWriteSchema.charPadding(table.options()));
Review Comment:
Fixed in b58434116f0. You are right, and it reproduces. On a local cluster
with a file:// catalog I wrote a CHAR(3) primary-key table with `bucket = 4`
and `changelog-producer = full-compaction` through `INSERT ... SELECT ... FROM
numbers("number" = "100")` with four writers. Every bucket received files from
all four writers. Paimon committed the APPEND snapshot and then gave up the
compaction with `LSM conflicts detected` (two level-5 files of bucket 2 from
different writers); the following INSERTs gave it up with `File deletion
conflicts detected`. Each INSERT still returned COMMITTED, and the table had
only APPEND snapshots, so no changelog was written.
The fix takes your second route. With padding on, `PaimonWritePlanProvider`
no longer routes a CHAR partition or bucket field natively
(`supportsNativeRouting`), so `fixedBucketDistribution` returns null and the
table gets the distribution of a routing field that BE cannot hash: GATHER to
one writer for a primary-key table or one that compacts. With padding off the
route is unchanged. Normalizing the exchanged values would mean padding in BE's
route function, and only for the rows the writer pads, which are the INSERT
rows and not the UPDATE and DELETE rows of a MERGE; that is a larger change
than this PR should carry.
Tests:
-
`PaimonWritePlanProviderTest.charRouteFieldsFallBackToSingleWriterWhenCharPaddingIsOn`:
a CHAR(3) primary key and a CHAR(3) partition key give GATHER with padding on
and `paimon_fixed_bucket` with it off.
- `test_paimon_write_char_padding` now writes a CHAR(3) primary-key table
with `bucket = 4` and `changelog-producer = full-compaction` from `numbers()`
with `parallel_pipeline_task_num = 4`, and checks `$snapshots`: an APPEND, then
a COMPACT with 100 changelog records. With the jars from before this fix it
fails there, with no COMPACT snapshot.
- On the local cluster, EXPLAIN of that INSERT now shows
`DistributionSpecGather`, and every INSERT commits an APPEND and a COMPACT
snapshot.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java:
##########
@@ -163,13 +201,135 @@ GenericRow tableRow(InternalRow columnValues) {
row.setRowKind(toRowKind(value));
continue;
}
+ // Only an inserted row 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.
+ if (charPadders[i] != null && value != null && row.getRowKind() ==
RowKind.INSERT) {
Review Comment:
Fixed in b58434116f0. `PaimonWriteSchema` now leaves the column that the
table names as `rowkind.field` unpadded (`PaimonJniWriter` passes
`CoreOptions.rowkindField()`), and pads the other CHAR columns as before.
Before the fix, `INSERT INTO t VALUES (3, '+I', 'c', 30)` into a table with `rk
CHAR(3)` and `'rowkind.field' = 'rk'` failed with `Unsupported short string '+I
' for row kind`.
Tests: `PaimonWriteSchemaTest.testCharPaddingLeavesTheRowKindFieldAlone`;
and `test_paimon_write_char_padding` writes two `'+I'` rows and then a `'-D'`
row into such a table, and checks that the `'-D'` row deletes key 1 while the
other CHAR column is padded.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java:
##########
@@ -163,13 +201,135 @@ GenericRow tableRow(InternalRow columnValues) {
row.setRowKind(toRowKind(value));
continue;
}
+ // Only an inserted row 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.
+ if (charPadders[i] != null && value != null && row.getRowKind() ==
RowKind.INSERT) {
+ value = charPadders[i].pad(value);
+ }
// 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);
}
return row;
}
+ /**
+ * Returns the padder of a value of {@code type}, 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.
+ */
+ private static CharPadder charPadder(DataType type) {
+ 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);
+ 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());
+ CharPadder valuePadder = charPadder(mapType.getValueType());
+ 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() :
padArray(map.keyArray(), keyGetter, keyPadder),
+ 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());
+ 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);
Review Comment:
Not changing this. The padded value is the value Paimon stores: an
`ARRAY<CHAR(65535)>` with 4096 `'a'` elements is a 256 MiB row once padded,
Paimon serializes that row when it writes it, and Spark and Flink build the
same row in their JVMs. The writer pads one row at a time, and the padded
values are garbage once `writer.write` has taken the row. The amplification
comes from the CHAR length the table was declared with; Doris itself does not
create a CHAR longer than 255 and reads a longer Paimon CHAR as STRING.
Charging the padded bytes to the query would need a new bridge from these Java
heap allocations to Doris's memory tracker, which is out of proportion to this
case.
--
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]
