This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ff7ba1faa79 [fix](paimon) Let DML reach the CHAR keys Spark writes
into Paimon ORC tables (#69295)
ff7ba1faa79 is described below
commit ff7ba1faa79c2a91e694ac86a70fd029d78eb56c
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Oct 25 11:01:21 2026 +0000
[fix](paimon) Let DML reach the CHAR keys Spark writes into Paimon ORC
tables (#69295)
### What problem does this PR solve?
Issue Number: None
Related PR: #69083
Problem Summary:
**In short.** Doris's UPDATE, DELETE and MERGE on a Paimon primary-key
table that stores ORC files miss the rows whose CHAR keys Spark or Flink
wrote, when the key is shorter than the column: the statement commits,
and the rows stay, or a second row with the same key appears. #69083
made Doris pad CHAR values as Spark and Flink do, but left the rows of
ORC tables unpadded, because an UPDATE or DELETE reads its keys back
from ORC without the padding. This PR pads every row written to an ORC
table, UPDATE and DELETE rows included, so that Doris writes the keys as
Spark and Flink do.
**Background.** ORC stores a CHAR(n) value padded and strips the
trailing spaces when it reads it, so Doris reads back `'k'` for the `'k
'` that Spark wrote. Paimon uses the key bytes as the writer held them
in two places: the bucket of a row is a hash of them, and every data
file records its smallest and largest key, by which reads and
compactions decide which files to merge. The value ORC gives back tells
neither.
**The problem, and what it cost.** On a local cluster with a file://
Paimon catalog, tables with a CHAR(3) primary key whose rows were
written through the Paimon API as Spark writes them (`'k '`, `'ab '`,
…), on master:
| Table | `DELETE … WHERE v IN (1, 3, 4, 5)`, then `UPDATE … SET v = v *
10 WHERE v IN (2, 6)` |
|---|---|
| ORC, `bucket = 4` | both COMMITTED, none of the 4 rows deleted |
| ORC, `bucket = -1` (dynamic bucket, one key per bucket) | none
deleted, and `ab` read back twice (2 and 20) |
| ORC, `bucket = 1`: `DELETE … WHERE c = 'ab'`, the smallest key of its
file | not deleted; `UPDATE … WHERE c = 'ab'` leaves two `ab` rows |
| Parquet, all of the above | correct |
Writing the change to the right bucket would not be enough. Through the
Paimon API, a `-D 'k'` written by itself to the bucket that holds `'k '`
deletes nothing: the key range `['k', 'k']` of its file lies below the
range `['k ', 'x ']` of the file holding the row, and Paimon does not
merge files whose ranges do not overlap. A `-D 'x'` in the same bucket,
inside that range, does delete. The key bytes have to be the padded
ones. The same happens to keys with trailing spaces that Doris itself
writes, and an `INSERT` of a key Spark wrote gives a second row with
that key.
branch-4.2 writes CHAR values unpadded and buckets UPDATE and DELETE
rows the same way, so it very likely has the same problem; this was not
run there.
**How this PR fixes it.**
- `PaimonWriteSchema` (the BE Java writer): on a table that writes ORC
files, by `file.format` or at any level of `file.format.per.level`,
every row is padded, whatever its row kind: INSERT, UPDATE, DELETE and
MERGE rows, and every row of a `rowkind.field` table (whose
`rowkind.field` value itself is still left alone). Other tables keep the
rule of #69083: inserted rows are padded, while UPDATE and DELETE rows
keep the bytes Doris read back, which Parquet and Avro return as stored.
- Cross-partition tables (`bucket = -1` with a primary key that leaves
out a partition column): each writer bootstraps its global key index
from the keys the files hold, which ORC gives back without their
trailing spaces. The writer now pads those keys as it pads the rows it
writes, and keeps the partition values and buckets as they are stored. A
key written into another partition then finds its stored row and deletes
it. Paimon's own index for Spark and Flink bootstraps the same trimmed
keys: Spark writing `'k'` into `p1` and then `p2` of such an ORC table
keeps two rows.
- MERGE: Doris compares a source key with a target key as it reads it.
The writer fails an inserted CHAR key that padding could turn into a
stored key MERGE did not match. Which keys those are depends on how the
target keys come back:
- Partition columns (Paimon keeps the value in the partition) and tables
without ORC files return the keys padded, as stored. There a key shorter
than the column fails, as in #69083.
- A table whose every level writes ORC returns the keys without their
trailing spaces. A short key is fine there: MERGE would have matched a
stored `'k '` to `'k'`, so `'k'` is new, and it is padded. A key that
ends with a space, such as `'ab '`, matched no target key, yet padded it
is the stored `'ab '`, so it fails; master silently replaced the stored
row.
- A table that writes ORC at some levels only (`file.format.per.level`,
such as `'orc'` with `'0:avro'`) returns a key in either form, depending
on the file. Only a key as long as the column that does not end with a
space is safe there.
The messages point to `rpad(<key>, n, ' ')`, `rtrim(<key>)` or
full-length keys, and to `enable_paimon_write_char_padding`. The formats
are read per level, from 0 to `num-levels - 1`, as Paimon's writer picks
them.
- FE: `PaimonWriteBinding.charPaddedFields` no longer exempts ORC
tables, so the CHAR bucket key of an ORC table falls back to a single
writer, as on other formats (BE would hash the unpadded value).
**Results.** Rows written by Spark into ORC tables, in the regression
suite:
| Statement | master | this PR |
|---|---|---|
| DELETE and UPDATE, `bucket = 4` | 4 rows not deleted | correct |
| DELETE and UPDATE, `bucket = -1` | not deleted, two `ab` rows |
correct |
| UPDATE, then DELETE of the smallest key, `bucket = 1` | two `ab` rows,
then one not deleted | correct |
| `INSERT` of keys Spark wrote | two `k` rows and two `ab` rows | one
row each |
| `INSERT … SELECT c, '-D', v` into a `rowkind.field` table | not
deleted | deleted |
| MERGE matching `'k'` | two `k` rows | updated |
| MERGE inserting `'ab '` | replaces the `ab` row Spark wrote | fails,
writes nothing |
| Cross-partition table: `'m'`, which Spark wrote into `p1`, written
into `p2` | two `m` rows | one row, in `p2` |
| Cross-partition table: `'k'` written into `p1`, then into `p2` | one
row | one row |
| `'orc'` + `'0:avro'` table: MERGE inserting `'k'`, which Spark wrote
into level 0 | a second `k` row | fails, writes nothing |
| ORC table: MERGE inserting the CHAR partition value `'x'` next to the
stored `'x '` | fails | fails |
Compatibility: Doris up to 4.2 wrote such keys unpadded. On an ORC
primary-key table whose CHAR key is shorter than the column, DML from
this version may miss those rows, as Spark and Flink do. `SET
enable_paimon_write_char_padding = false` writes as before, and `INSERT
OVERWRITE t SELECT * FROM t` rewrites such a table padded. An ORC table
bucketed by a CHAR column is written by one writer, as tables in other
formats are.
One limit belongs to Paimon itself. On a table that writes ORC at some
levels only, a CHAR key compacted into an ORC file does not merge with a
newer version of it in a level of another format, because Paimon's merge
reader compares the keys as each format returns them. Spark reads back
two rows after upserting such a key, and so does Doris.
---
.../org/apache/doris/paimon/PaimonJniWriter.java | 9 +-
.../org/apache/doris/paimon/PaimonWriteSchema.java | 242 +++++++++++++----
.../apache/doris/paimon/PaimonWriteSchemaTest.java | 301 ++++++++++++++++++---
.../doris/connector/paimon/PaimonWriteBinding.java | 15 +-
.../connector/paimon/PaimonWritePlanProvider.java | 3 +-
.../connector/paimon/PaimonWriteBindingTest.java | 6 +-
.../paimon/PaimonWritePlanProviderTest.java | 21 +-
.../write/test_paimon_write_char_padding.out | 53 +++-
.../write/test_paimon_write_char_padding.groovy | 240 +++++++++++++++-
9 files changed, 772 insertions(+), 118 deletions(-)
diff --git
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
index b94b11ad22a..8d28e82c58a 100644
---
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
+++
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
@@ -64,6 +64,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.function.UnaryOperator;
/**
* JNI entry point for Paimon write operations.
@@ -185,8 +186,8 @@ public class PaimonJniWriter extends JniWriter {
CoreOptions coreOptions =
CoreOptions.fromMap(table.options());
this.writeSchema =
PaimonWriteSchema.create(table.rowType(), columnNames, changelogWrite,
PaimonWriteSchema.charPadding(table.options()),
coreOptions.rowkindField().orElse(null),
- table.primaryKeys(), table.partitionKeys(),
- PaimonWriteSchema.writesOrcFiles(table.options()),
overwrite);
+ table.primaryKeys(), table.partitionKeys(),
PaimonWriteSchema.orcFiles(table.options()),
+ overwrite);
this.arrowAdapter = new PaimonArrowBatchAdapter(
writeSchema.inputType(), ZoneId.of(timeZone),
allocator);
validateWriteColumnsForMergeEngine(
@@ -471,8 +472,10 @@ public class PaimonJniWriter extends JniWriter {
long writerMemoryLimitBytes) throws Exception {
globalIndexAssigner = new GlobalIndexAssigner(table,
writerMemoryLimitBytes);
globalIndexAssigner.open(1, 0, this::writeAssignedRow);
+ // The index has to hold the stored keys as the rows written carry
them, which ORC does not give back.
+ UnaryOperator<InternalRow> bootstrapKeyPadder =
writeSchema.bootstrapKeyPadder(table.schema());
new IndexBootstrap(table).bootstrap(
- 1, 0, this::bootstrapGlobalIndexKey);
+ 1, 0, row ->
bootstrapGlobalIndexKey(bootstrapKeyPadder.apply(row)));
globalIndexAssigner.finishBootstrap();
}
diff --git
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java
index 739cdff442d..285b0de60b8 100644
---
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java
+++
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java
@@ -21,6 +21,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.casting.CastExecutor;
import org.apache.paimon.casting.CastExecutors;
import org.apache.paimon.casting.DefaultValueRow;
+import org.apache.paimon.crosspartition.IndexBootstrap;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericArray;
@@ -29,6 +30,7 @@ import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.sink.RowKindGenerator;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.CharType;
@@ -48,6 +50,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.UnaryOperator;
/**
* Immutable mapping from Doris input columns to a Paimon table row.
@@ -67,19 +70,23 @@ final class PaimonWriteSchema {
* fe-connector-paimon.
*/
static final String DORIS_WRITE_CHAR_PADDING = "doris.write.char-padding";
+ private static final BinaryString SPACE = BinaryString.fromString(" ");
private final InternalRow.FieldGetter[] fieldGetters;
/**
- * Per input column, pads the CHAR values it holds; null when it holds
none, when padding is off, for the
- * table's rowkind.field, or for a column other than a partition column of
a table that writes ORC files.
+ * Per input column, pads the CHAR values it holds; null when it holds
none, when padding is off, or for the
+ * table's rowkind.field.
*/
private final CharPadder[] charPadders;
/**
- * Per input column, the length of a padded CHAR primary-key column of a
changelog write, or 0. Padding a
- * shorter key that MERGE inserts would change the key after MERGE matched
it and checked it for duplicates;
- * see {@link #padRow}.
+ * Per input column, the length of a padded CHAR primary-key column of a
changelog write, or 0. Padding a key
+ * that MERGE inserts can change it after MERGE matched it and checked it
for duplicates; see {@link #padRow}.
*/
private final int[] insertedKeyCharLengths;
+ /** Per input column with an inserted key length, how Doris reads its
stored keys back; see {@link #padRow}. */
+ private final KeyReadBack[] insertedKeyReadBacks;
+ /** Which data files of the table are ORC files ({@link #orcFiles}); see
{@link #padsRow}. */
+ private final OrcFiles orcFiles;
/** Decodes the kind of a row from the table's rowkind.field as Paimon
does when it writes the row, or null. */
private final RowKindGenerator rowKindGenerator;
/**
@@ -99,8 +106,8 @@ final class PaimonWriteSchema {
private PaimonWriteSchema(int[] tableFieldIndexes, int[]
omittedDefaultFieldIndexes,
Object[] omittedDefaultValues, int tableFieldCount, RowType
inputType, CharPadder[] charPadders,
- int[] insertedKeyCharLengths, RowKindGenerator rowKindGenerator,
DefaultValueRow defaultValueRow,
- boolean overwrite) {
+ int[] insertedKeyCharLengths, KeyReadBack[] insertedKeyReadBacks,
OrcFiles orcFiles,
+ RowKindGenerator rowKindGenerator, DefaultValueRow
defaultValueRow, boolean overwrite) {
this.fieldGetters = new
InternalRow.FieldGetter[inputType.getFieldCount()];
for (int i = 0; i < fieldGetters.length; i++) {
// Getters describe the transport row, not the target table.
Transport fields are
@@ -110,6 +117,8 @@ final class PaimonWriteSchema {
}
this.charPadders = charPadders;
this.insertedKeyCharLengths = insertedKeyCharLengths;
+ this.insertedKeyReadBacks = insertedKeyReadBacks;
+ this.orcFiles = orcFiles;
this.rowKindGenerator = rowKindGenerator;
this.defaultValueRow = defaultValueRow;
this.overwrite = overwrite;
@@ -151,27 +160,24 @@ final class PaimonWriteSchema {
static PaimonWriteSchema create(RowType tableType, String[] columnNames,
boolean changelogWrite,
boolean charPadding, String rowKindField, List<String>
primaryKeys) {
return create(tableType, columnNames, changelogWrite, charPadding,
rowKindField, primaryKeys,
- Collections.emptyList(), false, false);
+ Collections.emptyList(), OrcFiles.NONE, false);
}
/**
- * @param charPadding whether to pad the CHAR values of inserted rows as
Paimon's cast to CHAR does; see
- * {@link #charPadder} and {@link #padsRow}
+ * @param charPadding whether to pad CHAR values as Paimon's cast to
CHAR does; see {@link #charPadder}, and
+ * {@link #padsRow} for the rows that are padded
* @param rowKindField the table's {@code rowkind.field}, or null. Its
value is not padded: Paimon decodes the row
* kind from it ({@code RowKind.fromShortString}),
and a padded '+I ' is no row kind.
- * @param primaryKeys the table's primary keys. In a changelog write, a
CHAR key that padding would change
- * fails the row; see {@link #padRow}.
- * @param partitionKeys the table's partition keys
- * @param orcFiles whether the table writes ORC files ({@link
#writesOrcFiles}). ORC pads every CHAR value
- * it writes and strips the trailing spaces from one
it reads, so padding changes nothing
- * ORC stores, only the bucket of a CHAR key, which
an UPDATE or DELETE that reads the key
- * back unpadded would then miss. Only the partition
columns of such a table are padded: a
- * partition value is read back as it is stored.
+ * @param primaryKeys the table's primary keys. In a changelog write, a
CHAR key that MERGE inserts fails the
+ * row when padding could make it a key that MERGE
did not see; see {@link #padRow}.
+ * @param partitionKeys the table's partition keys, whose values Doris
reads back as they are stored
+ * @param orcFiles which data files of the table are ORC files
({@link #orcFiles}); every row of a table
+ * that writes some is padded, see {@link #padsRow}
* @param overwrite whether the write overwrites the partitions it
writes; see {@link #padsRow}
*/
static PaimonWriteSchema create(RowType tableType, String[] columnNames,
boolean changelogWrite,
boolean charPadding, String rowKindField, List<String>
primaryKeys, List<String> partitionKeys,
- boolean orcFiles, boolean overwrite) {
+ OrcFiles orcFiles, boolean overwrite) {
if (columnNames == null || columnNames.length == 0) {
throw new IllegalArgumentException(
"PaimonJniWriter requires explicit column names");
@@ -180,6 +186,7 @@ final class PaimonWriteSchema {
int[] tableFieldIndexes = new int[columnNames.length];
CharPadder[] charPadders = new CharPadder[columnNames.length];
int[] insertedKeyCharLengths = new int[columnNames.length];
+ KeyReadBack[] insertedKeyReadBacks = new
KeyReadBack[columnNames.length];
List<DataField> inputFields = new ArrayList<>(columnNames.length);
boolean[] specifiedFields = new boolean[tableType.getFieldCount()];
for (int i = 0; i < columnNames.length; i++) {
@@ -210,12 +217,14 @@ final class PaimonWriteSchema {
// Paimon writer return its normal constraint diagnostic instead
of reading a default
// primitive value from a non-null getter.
inputFields.add(field.newType(field.type().nullable()));
- if (charPadding && !field.name().equals(rowKindField)
- && (!orcFiles || partitionKeys.contains(field.name()))) {
+ if (charPadding && !field.name().equals(rowKindField)) {
charPadders[i] = charPadder(field.type(), field.name());
if (changelogWrite && field.type().getTypeRoot() ==
DataTypeRoot.CHAR
&& primaryKeys.contains(field.name())) {
insertedKeyCharLengths[i] = ((CharType)
field.type()).getLength();
+ insertedKeyReadBacks[i] =
partitionKeys.contains(field.name()) || orcFiles == OrcFiles.NONE
+ ? KeyReadBack.PADDED
+ : orcFiles == OrcFiles.ALL ? KeyReadBack.TRIMMED :
KeyReadBack.PADDED_OR_TRIMMED;
}
}
}
@@ -242,6 +251,8 @@ final class PaimonWriteSchema {
new RowType(inputFields),
charPadders,
insertedKeyCharLengths,
+ insertedKeyReadBacks,
+ orcFiles,
rowKindField == null ? null : new
RowKindGenerator(rowKindField, tableType),
DefaultValueRow.create(tableType),
overwrite);
@@ -255,11 +266,23 @@ final class PaimonWriteSchema {
return
Boolean.parseBoolean(tableOptions.get(DORIS_WRITE_CHAR_PADDING));
}
- /** Whether the table writes ORC files, by {@code file.format} or at some
level by {@code file.format.per.level}. */
- static boolean writesOrcFiles(Map<String, String> tableOptions) {
+ /**
+ * Which data files of the table are ORC files: Paimon writes a file of a
level in the format that
+ * {@code file.format.per.level} names for the level, or else in {@code
file.format}.
+ */
+ static OrcFiles orcFiles(Map<String, String> tableOptions) {
CoreOptions options = CoreOptions.fromMap(tableOptions);
- return CoreOptions.FILE_FORMAT_ORC.equals(options.formatType())
- ||
options.fileFormatPerLevel().containsValue(CoreOptions.FILE_FORMAT_ORC);
+ Map<Integer, String> formatPerLevel = options.fileFormatPerLevel();
+ boolean orc = false;
+ boolean other = false;
+ for (int level = 0; level < options.numLevels(); level++) {
+ if
(CoreOptions.FILE_FORMAT_ORC.equals(formatPerLevel.getOrDefault(level,
options.formatType()))) {
+ orc = true;
+ } else {
+ other = true;
+ }
+ }
+ return !orc ? OrcFiles.NONE : other ? OrcFiles.SOME : OrcFiles.ALL;
}
/** Paimon input fields in the exact order transported by Arrow C Data. */
@@ -294,19 +317,69 @@ final class PaimonWriteSchema {
}
/**
- * 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.
+ * Returns what gives a row that Paimon's IndexBootstrap reads from the
key-dynamic table of {@code schema}, to
+ * index the keys stored in it, those keys as this writer writes them.
Such a row holds the primary keys that are
+ * no partition keys as the data files give them back, then the partition
and the bucket
+ * ({@code IndexBootstrap#bootstrapType}). ORC gives a CHAR key back
without the trailing spaces that every row
+ * this writer writes into a table that writes ORC files carries ({@link
#padsRow}), so an index of the keys as
+ * read would hold 'k' for the stored 'k ', and a row of 'k' written into
another partition would miss it and
+ * leave the stored row in place, two rows of one key. Those keys are
padded as the rows are; the other files give
+ * the keys back as they are stored, and the partition is the stored one.
+ */
+ UnaryOperator<InternalRow> bootstrapKeyPadder(TableSchema schema) {
+ List<String> keys = schema.trimmedPrimaryKeys();
+ CharPadder[] keyPadders = new CharPadder[keys.size()];
+ boolean padsKeys = false;
+ if (orcFiles != OrcFiles.NONE) {
+ for (int i = 0; i < keyPadders.length; i++) {
+ // A key omitted from the input takes its default value
unpadded, as the rows written do.
+ int column = inputType.getFieldIndex(keys.get(i));
+ keyPadders[i] = column < 0 ? null : charPadders[column];
+ padsKeys |= keyPadders[i] != null;
+ }
+ }
+ if (!padsKeys) {
+ return UnaryOperator.identity();
+ }
+ RowType bootstrapType = IndexBootstrap.bootstrapType(schema);
+ InternalRow.FieldGetter[] bootstrapFieldGetters = new
InternalRow.FieldGetter[bootstrapType.getFieldCount()];
+ for (int i = 0; i < bootstrapFieldGetters.length; i++) {
+ bootstrapFieldGetters[i] =
InternalRow.createFieldGetter(bootstrapType.getTypeAt(i), i);
+ }
+ return bootstrapRow -> {
+ GenericRow padded = new GenericRow(bootstrapFieldGetters.length);
+ for (int i = 0; i < bootstrapFieldGetters.length; i++) {
+ Object value =
bootstrapFieldGetters[i].getFieldOrNull(bootstrapRow);
+ padded.setField(i, i < keyPadders.length && keyPadders[i] !=
null ? keyPadders[i].pad(value) : value);
+ }
+ return padded;
+ };
+ }
+
+ /**
+ * Whether to pad {@code row}. Every row of a table that writes ORC files,
at some levels or at all, is padded.
+ * ORC pads every CHAR value it writes and strips the trailing spaces from
one it reads, so Doris reads back 'k'
+ * for the 'k ' that Spark and Flink write. Paimon puts a row in the
bucket of its key as the writer holds it,
+ * and merges only the data files whose key ranges, also kept as the
writer held the keys, overlap: an UPDATE or
+ * DELETE row that wrote that key back as 'k' would land in another
bucket, or below the range of the file
+ * holding 'k ', and miss the row.
+ *
+ * <p>On any other table, a row is padded when 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 as it is stored,
+ * 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 (orcFiles != OrcFiles.NONE) {
+ return true;
+ }
if (row.getRowKind() != RowKind.INSERT) {
return false;
}
@@ -327,24 +400,69 @@ final class PaimonWriteSchema {
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]));
+ if (insertedKeyCharLengths[i] > 0 && row.getRowKind() ==
RowKind.INSERT) {
+ checkInsertedKey((BinaryString) value, i);
}
row.setField(tableFieldIndexes[i], charPadders[i].pad(value));
}
}
+ /**
+ * 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. So a key that padding could make a
+ * stored key that MERGE did not match, or the same key as another one
MERGE inserts, fails the row, and which
+ * keys those are depends on how Doris reads the stored keys back ({@link
KeyReadBack}). Read back padded, as the
+ * values of a partition column and the keys of a table that writes no ORC
files are, a key shorter than the
+ * column, such as 'k', matched no stored 'k ' yet pads to it, and pads
to the key 'k ' pads to; a key as long as
+ * the column is not changed by padding. ORC gives the keys back without
their trailing spaces, so there MERGE
+ * matched 'k' to the stored 'k ', and only a key that ends with a space,
such as 'k ', matches no stored key yet
+ * pads to one. A table that writes ORC files at some levels only gives a
key back in either form, by the file
+ * that holds it, so both such keys fail the row.
+ */
+ private void checkInsertedKey(BinaryString key, int column) {
+ int length = insertedKeyCharLengths[column];
+ String name = inputType.getFields().get(column).name();
+ KeyReadBack readBack = insertedKeyReadBacks[column];
+ if (readBack == KeyReadBack.PADDED_OR_TRIMMED) {
+ if (key.numChars() < length || key.endsWith(SPACE)) {
+ // The remedy comes first: the client gets a long error
message of the writer cut short.
+ throw new IllegalArgumentException(String.format(
+ "MERGE cannot insert '%s' into the CHAR(%d) primary
key column '%s' of a Paimon table "
+ + "that writes ORC files at some levels only.
Insert keys as long as the column "
+ + "that do not end with a space, or set
enable_paimon_write_char_padding = false: "
+ + "the table gives its keys back padded from
its other files and without their "
+ + "trailing spaces from its ORC files, so
MERGE matched this value against keys in "
+ + "either form, and padding it can turn it
into a stored key that MERGE did not "
+ + "match",
+ key, length, name));
+ }
+ return;
+ }
+ if (readBack == KeyReadBack.TRIMMED) {
+ if (key.endsWith(SPACE)) {
+ // The remedy comes first: the client gets a long error
message of the writer cut short.
+ throw new IllegalArgumentException(String.format(
+ "MERGE cannot insert '%s' into the CHAR(%d) primary
key column '%s' of a Paimon table "
+ + "that writes ORC files. Remove the trailing
spaces of the key in the MERGE, with "
+ + "rtrim(<key>) in both the ON condition and
the inserted values, or set "
+ + "enable_paimon_write_char_padding = false:
ORC drops the trailing spaces of the "
+ + "keys it stores, so MERGE matched this value
against keys without them, and "
+ + "padding it can turn it into a stored key
that MERGE did not match",
+ key, length, name));
+ }
+ return;
+ }
+ if (key.numChars() < length) {
+ 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",
+ key, length, name, length));
+ }
+ }
+
/**
* 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
@@ -625,6 +743,32 @@ final class PaimonWriteSchema {
Object pad(Object value);
}
+ /**
+ * Which data files of a table are ORC files. ORC pads every CHAR value it
writes and strips the trailing spaces
+ * from one it reads, while the other formats give a value back as it is
stored.
+ */
+ enum OrcFiles {
+ /** No level writes ORC files. */
+ NONE,
+ /** Some levels write ORC files, and the others another format. */
+ SOME,
+ /** Every level writes ORC files. */
+ ALL
+ }
+
+ /** How Doris reads back the stored values of a CHAR primary-key column,
against which MERGE matched its keys. */
+ private enum KeyReadBack {
+ /**
+ * As they are stored, padded: the values of a partition column, which
Paimon keeps in the partition, and the
+ * keys of a table that writes no ORC files.
+ */
+ PADDED,
+ /** Without their trailing spaces: the keys of a table that writes
only ORC files. */
+ TRIMMED,
+ /** Padded from the files of some levels and trimmed from the ORC
files of the others. */
+ PADDED_OR_TRIMMED
+ }
+
/**
* Copies one non-null value of a type that holds a MAP, FLOAT or DOUBLE,
with each NaN in it made the canonical
* NaN and the entries of each MAP in it sorted, one per key.
diff --git
a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonWriteSchemaTest.java
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonWriteSchemaTest.java
index 45ef5baf00e..4051f53e316 100644
---
a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonWriteSchemaTest.java
+++
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonWriteSchemaTest.java
@@ -17,11 +17,17 @@
package org.apache.doris.paimon;
+import org.apache.doris.paimon.PaimonWriteSchema.OrcFiles;
+
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowStreamReader;
import org.apache.paimon.arrow.writer.ArrowFieldWriterFactoryVisitor;
import org.apache.paimon.casting.DefaultValueRow;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.crosspartition.IndexBootstrap;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericMap;
@@ -29,6 +35,13 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.types.CharType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
@@ -40,15 +53,20 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.ByteArrayInputStream;
+import java.nio.file.Path;
import java.time.ZoneId;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
+import java.util.function.UnaryOperator;
public class PaimonWriteSchemaTest {
@@ -438,8 +456,8 @@ public class PaimonWriteSchemaTest {
// An overwrite replaces the partitions it writes, so none of its rows
matches a stored key. FE pads the
// static partition value of INSERT OVERWRITE ... PARTITION (p = 'x')
into the overwrite filter, so a '+U'
// or '-D' row has to land in 'x ' too, or Paimon rejects its file as
outside the overwritten partition;
- // and a '+I' and a '-D' of one key in the statement still cancel out.
An ORC table pads only its partition
- // values, in an overwrite too. MUTATION: padding by the decoded row
kind in an overwrite -> red.
+ // and a '+I' and a '-D' of one key in the statement still cancel out.
An ORC table pads every row anyway.
+ // MUTATION: padding by the decoded row kind in an overwrite -> red.
RowType tableType = new RowType(Arrays.asList(
new DataField(0, "id", new IntType(false)),
new DataField(1, "p", new CharType(false, 3)),
@@ -447,7 +465,7 @@ public class PaimonWriteSchemaTest {
new DataField(3, "code", new CharType(3))));
String[] columns = {"id", "p", "kind", "code"};
PaimonWriteSchema overwrite = PaimonWriteSchema.create(tableType,
columns, false, true, "kind",
- Arrays.asList("id", "p"), Collections.singletonList("p"),
false, true);
+ Arrays.asList("id", "p"), Collections.singletonList("p"),
OrcFiles.NONE, true);
for (String kind : Arrays.asList("+I", "+U", "-D", "-U")) {
InternalRow row = tableRow(overwrite, 1, string("x"),
string(kind), string("k"));
Assertions.assertEquals("x ", row.getString(1).toString(), kind);
@@ -456,13 +474,13 @@ public class PaimonWriteSchemaTest {
}
PaimonWriteSchema orcOverwrite = PaimonWriteSchema.create(tableType,
columns, false, true, "kind",
- Arrays.asList("id", "p"), Collections.singletonList("p"),
true, true);
+ Arrays.asList("id", "p"), Collections.singletonList("p"),
OrcFiles.ALL, true);
InternalRow orcRow = tableRow(orcOverwrite, 1, string("x"),
string("+U"), string("k"));
Assertions.assertEquals("x ", orcRow.getString(1).toString());
- Assertions.assertEquals("k", orcRow.getString(3).toString());
+ Assertions.assertEquals("k ", orcRow.getString(3).toString());
PaimonWriteSchema insert = PaimonWriteSchema.create(tableType,
columns, false, true, "kind",
- Arrays.asList("id", "p"), Collections.singletonList("p"),
false, false);
+ Arrays.asList("id", "p"), Collections.singletonList("p"),
OrcFiles.NONE, false);
Assertions.assertEquals("x",
tableRow(insert, 1, string("x"), string("+U"),
string("k")).getString(1).toString());
}
@@ -697,49 +715,255 @@ public class PaimonWriteSchemaTest {
}
@Test
- public void testOrcTablePadsOnlyItsPartitionValues() {
- // ORC pads every CHAR value it writes and strips the trailing spaces
from one it reads, so padding changes
- // nothing that ORC stores, only the bucket of a CHAR key: an UPDATE
or DELETE that reads 'k' back would
- // hash another bucket than the padded 'k ' went to, and miss the
row. A partition value is read back as
- // it is stored, so it is padded, and a MERGE that inserts a short one
fails as for any padded key, while a
- // short key that is not padded does not. MUTATION: padding every
column of an ORC table -> red.
+ public void testOrcTablePadsTheCharValuesOfEveryRow() {
+ // ORC pads every CHAR value it writes and strips the trailing spaces
from one it reads, so Doris reads back
+ // 'k' for the 'k ' that Spark and Flink write. Paimon puts a row in
the bucket of its key as the writer
+ // holds it, and merges only the data files whose key ranges, also
kept as the writer held the keys,
+ // overlap: an UPDATE or DELETE that wrote that key as 'k' would land
in another bucket, or below the range
+ // of the file holding 'k ', and miss the row. So every row of an ORC
table is padded, including UPDATE and
+ // DELETE rows, and so are the keys of its maps, which fail the row
when padding makes them equal.
+ // MUTATION: padding only the partition columns, or only the inserted
rows, of an ORC table -> red.
RowType tableType = new RowType(Arrays.asList(
new DataField(0, "p", new CharType(false, 3)),
new DataField(1, "code", new CharType(false, 3)),
new DataField(2, "labels", DataTypes.MAP(DataTypes.CHAR(2),
DataTypes.INT()))));
PaimonWriteSchema insert = PaimonWriteSchema.create(tableType, new
String[] {"p", "code", "labels"},
- false, true, null, Arrays.asList("p", "code"),
Collections.singletonList("p"), true, false);
+ false, true, null, Arrays.asList("p", "code"),
Collections.singletonList("p"), OrcFiles.ALL, false);
InternalRow row = tableRow(insert, string("x"), string("k"),
- map(new Object[] {string("k"), string("k ")}, new Object[] {1,
2}));
+ map(new Object[] {string("k")}, new Object[] {1}));
Assertions.assertEquals("x ", row.getString(0).toString());
- Assertions.assertEquals("k", row.getString(1).toString());
- Assertions.assertEquals(2, row.getMap(2).size());
- Assertions.assertEquals("k",
row.getMap(2).keyArray().getString(0).toString());
+ Assertions.assertEquals("k ", row.getString(1).toString());
+ Assertions.assertEquals("k ",
row.getMap(2).keyArray().getString(0).toString());
+ IllegalArgumentException e =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> tableRow(insert, string("x"), string("k"),
+ map(new Object[] {string("k"), string("k ")}, new
Object[] {1, 2})));
+ Assertions.assertTrue(e.getMessage().contains("into the same key"),
e.getMessage());
- PaimonWriteSchema merge = PaimonWriteSchema.create(tableType,
+ PaimonWriteSchema changelog = PaimonWriteSchema.create(tableType,
new String[] {PaimonWriteSchema.ROW_KIND_COLUMN, "p", "code"},
true, true, null,
- Arrays.asList("p", "code"), Collections.singletonList("p"),
true, false);
- Assertions.assertEquals("k", tableRow(merge,
PaimonWriteSchema.INSERT_OPERATION, string("abc"),
- string("k")).getString(1).toString());
- IllegalArgumentException e =
Assertions.assertThrows(IllegalArgumentException.class,
- () -> tableRow(merge, PaimonWriteSchema.INSERT_OPERATION,
string("x"), string("k")));
- Assertions.assertTrue(e.getMessage().contains("CHAR(3) primary key
column 'p'"), e.getMessage());
+ Arrays.asList("p", "code"), Collections.singletonList("p"),
OrcFiles.ALL, false);
+ for (byte operation : new byte[] {PaimonWriteSchema.UPDATE_OPERATION,
PaimonWriteSchema.DELETE_OPERATION}) {
+ InternalRow changed = tableRow(changelog, operation, string("x"),
string("k"));
+ Assertions.assertEquals("x ", changed.getString(0).toString());
+ Assertions.assertEquals("k ", changed.getString(1).toString());
+ }
+ }
+
+ @Test
+ public void testOrcTablePadsTheRowsOfEveryRowKindFieldValue() {
+ // A '-D', '-U' or '+U' row of a rowkind.field table carries a key to
match, which Doris reads back from an
+ // ORC table without its trailing spaces, so on such a table those
rows are padded as the '+I' rows are,
+ // and a value Paimon cannot decode is padded and left for Paimon to
reject. The rowkind.field value itself
+ // stays as it is. MUTATION: padding by the decoded row kind on an ORC
table, or padding the rowkind.field
+ // -> red.
+ RowType tableType = new RowType(Arrays.asList(
+ new DataField(0, "code", new CharType(false, 3)),
+ new DataField(1, "kind", new CharType(3)),
+ new DataField(2, "label", new CharType(3))));
+ PaimonWriteSchema schema = PaimonWriteSchema.create(tableType, new
String[] {"code", "kind", "label"},
+ false, true, "kind", Collections.singletonList("code"),
Collections.emptyList(), OrcFiles.ALL, false);
+
+ for (String kind : Arrays.asList("+I", "-D", "-U", "+U", "+X")) {
+ InternalRow row = tableRow(schema, string("k"), string(kind),
string("a"));
+ Assertions.assertEquals("k ", row.getString(0).toString(), kind);
+ Assertions.assertEquals(kind, row.getString(1).toString(), kind);
+ Assertions.assertEquals("a ", row.getString(2).toString(), kind);
+ }
+ }
+
+ @Test
+ public void
testMergeIntoAnOrcTableFailsOnlyForInsertedKeysEndingWithASpace() {
+ // Doris reads the keys of an ORC table back without their trailing
spaces, and MERGE matches its source
+ // rows to them and checks the keys it inserts for duplicates by the
key as Doris holds it. A short key
+ // such as 'k' is then a key no stored row has, as the stored 'k '
would have matched it, and it is padded
+ // as Spark and Flink pad it. A key that ends with a space, such as 'k
', matches no stored key, yet padding
+ // makes it the stored 'k ' that MERGE did not match, and 'k' and 'k
' one key, so such a row fails. UPDATE
+ // and DELETE rows carry a key read from the table and are padded, and
one read from a level that
+ // file.format.per.level writes in another format keeps the padding
Spark wrote. MUTATION: no check, the
+ // check of a table that writes no ORC files, or checking UPDATE /
DELETE rows -> red.
+ PaimonWriteSchema schema = PaimonWriteSchema.create(charKeyTableType(),
+ new String[] {PaimonWriteSchema.ROW_KIND_COLUMN, "code",
"label"}, true, true, null,
+ Collections.singletonList("code"), Collections.emptyList(),
OrcFiles.ALL, false);
+
+ Assertions.assertEquals("k ", tableRow(schema,
PaimonWriteSchema.INSERT_OPERATION, string("k"),
+ string("x")).getString(0).toString());
+ Assertions.assertEquals("中中 ", tableRow(schema,
PaimonWriteSchema.INSERT_OPERATION, string("中中"),
+ null).getString(0).toString());
+ Assertions.assertEquals("abc", tableRow(schema,
PaimonWriteSchema.INSERT_OPERATION, string("abc"),
+ null).getString(0).toString());
+ for (String key : Arrays.asList("k ", "ab ", "k ", " ")) {
+ IllegalArgumentException e =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> tableRow(schema, PaimonWriteSchema.INSERT_OPERATION,
string(key), string("x")));
+ Assertions.assertTrue(e.getMessage().contains("CHAR(3) primary key
column 'code'"), e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("rtrim(<key>)"),
e.getMessage());
+
Assertions.assertTrue(e.getMessage().contains("enable_paimon_write_char_padding"),
e.getMessage());
+ }
+ for (byte operation : new byte[] {PaimonWriteSchema.UPDATE_OPERATION,
PaimonWriteSchema.DELETE_OPERATION}) {
+ Assertions.assertEquals("k ", tableRow(schema, operation,
string("k"), string("x"))
+ .getString(0).toString());
+ Assertions.assertEquals("k ", tableRow(schema, operation,
string("k "), string("x"))
+ .getString(0).toString());
+ }
+ }
+
+ @Test
+ public void testMergeIntoAnOrcTableChecksItsPartitionKeysAsTheyAreStored()
{
+ // Paimon keeps the value of a partition column in the partition, so
Doris reads it back as it is stored, with
+ // its padding, from an ORC table too. MERGE then did not match 'x' to
the stored 'x ', and padding it would
+ // let a NOT MATCHED row replace the stored row of key ('x ', 'k '),
so a partition key shorter than its
+ // column fails the row as on a table that writes no ORC files, while
the other key keeps the check of an ORC
+ // table. MUTATION: checking the partition key as the keys of an ORC
table -> red.
+ RowType tableType = new RowType(Arrays.asList(
+ new DataField(0, "p", new CharType(false, 3)),
+ new DataField(1, "code", new CharType(false, 3)),
+ new DataField(2, "label", new CharType(3))));
+ PaimonWriteSchema schema = PaimonWriteSchema.create(tableType,
+ new String[] {PaimonWriteSchema.ROW_KIND_COLUMN, "p", "code",
"label"}, true, true, null,
+ Arrays.asList("p", "code"), Collections.singletonList("p"),
OrcFiles.ALL, false);
+
+ IllegalArgumentException shortPartition =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> tableRow(schema, PaimonWriteSchema.INSERT_OPERATION,
string("x"), string("k"), string("a")));
+ Assertions.assertTrue(shortPartition.getMessage().contains("CHAR(3)
primary key column 'p'"),
+ shortPartition.getMessage());
+
Assertions.assertTrue(shortPartition.getMessage().contains("rpad(<key>, 3, '
')"), shortPartition.getMessage());
+ InternalRow row = tableRow(schema, PaimonWriteSchema.INSERT_OPERATION,
string("x "), string("k"),
+ string("a"));
+ Assertions.assertEquals("x ", row.getString(0).toString());
+ Assertions.assertEquals("k ", row.getString(1).toString());
+ Assertions.assertEquals("a ", row.getString(2).toString());
+ IllegalArgumentException trailingSpace =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> tableRow(schema, PaimonWriteSchema.INSERT_OPERATION,
string("x "), string("k "),
+ string("a")));
+ Assertions.assertTrue(trailingSpace.getMessage().contains("CHAR(3)
primary key column 'code'"),
+ trailingSpace.getMessage());
+
Assertions.assertTrue(trailingSpace.getMessage().contains("rtrim(<key>)"),
trailingSpace.getMessage());
+ }
+
+ @Test
+ public void
testMergeIntoATableWithOrcFilesAtSomeLevelsFailsForShortKeysAndKeysEndingWithASpace()
{
+ // A table that writes ORC files at some levels only, such as
file.format = 'orc' with
+ // file.format.per.level = '0:avro', gives Doris a stored 'k ' back
as 'k ' from its other files and as 'k'
+ // from its ORC files. A key that MERGE inserts is safe only when
padding does not change it and no stored key
+ // reads back as it does without its trailing spaces: as long as the
column, and not ending with a space.
+ // UPDATE and DELETE rows are padded as on an ORC table. MUTATION: the
check of a table that writes only ORC
+ // files, or of one that writes none -> red.
+ PaimonWriteSchema schema = PaimonWriteSchema.create(charKeyTableType(),
+ new String[] {PaimonWriteSchema.ROW_KIND_COLUMN, "code",
"label"}, true, true, null,
+ Collections.singletonList("code"), Collections.emptyList(),
OrcFiles.SOME, false);
+
+ for (String key : Arrays.asList("k", "k ", "ab ", " ", "中中")) {
+ IllegalArgumentException e =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> tableRow(schema, PaimonWriteSchema.INSERT_OPERATION,
string(key), string("x")));
+ Assertions.assertTrue(e.getMessage().contains("CHAR(3) primary key
column 'code'"), e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("at some levels
only"), e.getMessage());
+
Assertions.assertTrue(e.getMessage().contains("enable_paimon_write_char_padding"),
e.getMessage());
+ }
+ Assertions.assertEquals("abc", tableRow(schema,
PaimonWriteSchema.INSERT_OPERATION, string("abc"),
+ string("x")).getString(0).toString());
+ Assertions.assertEquals("abcd", tableRow(schema,
PaimonWriteSchema.INSERT_OPERATION, string("abcd"),
+ null).getString(0).toString());
+ for (byte operation : new byte[] {PaimonWriteSchema.UPDATE_OPERATION,
PaimonWriteSchema.DELETE_OPERATION}) {
+ Assertions.assertEquals("k ", tableRow(schema, operation,
string("k"), string("x"))
+ .getString(0).toString());
+ }
}
@Test
public void testOrcFilesAreReadFromTheTableFormatOptions() {
- // MUTATION: ignoring file.format.per.level -> red.
-
Assertions.assertTrue(PaimonWriteSchema.writesOrcFiles(Collections.singletonMap("file.format",
"orc")));
-
Assertions.assertTrue(PaimonWriteSchema.writesOrcFiles(Collections.singletonMap("file.format",
"ORC")));
- Map<String, String> perLevel = new HashMap<>();
- perLevel.put("file.format", "parquet");
- perLevel.put("file.format.per.level", "0:avro,5:orc");
- Assertions.assertTrue(PaimonWriteSchema.writesOrcFiles(perLevel));
-
Assertions.assertFalse(PaimonWriteSchema.writesOrcFiles(Collections.emptyMap()));
-
Assertions.assertFalse(PaimonWriteSchema.writesOrcFiles(Collections.singletonMap("file.format",
"avro")));
- Assertions.assertFalse(PaimonWriteSchema.writesOrcFiles(
- Collections.singletonMap("file.format.per.level", "0:avro")));
+ // Paimon writes the files of a level, 0 to num-levels - 1, in the
format that file.format.per.level names for
+ // the level, or else in file.format. MUTATION: ignoring
file.format.per.level, or the levels that it does not
+ // name, or the number of levels -> red.
+ Assertions.assertEquals(OrcFiles.ALL,
+
PaimonWriteSchema.orcFiles(Collections.singletonMap("file.format", "orc")));
+ Assertions.assertEquals(OrcFiles.ALL,
+
PaimonWriteSchema.orcFiles(Collections.singletonMap("file.format", "ORC")));
+ Assertions.assertEquals(OrcFiles.NONE,
PaimonWriteSchema.orcFiles(Collections.emptyMap()));
+ Assertions.assertEquals(OrcFiles.NONE,
+
PaimonWriteSchema.orcFiles(Collections.singletonMap("file.format", "avro")));
+ Assertions.assertEquals(OrcFiles.NONE,
+
PaimonWriteSchema.orcFiles(Collections.singletonMap("file.format.per.level",
"0:avro")));
+ Assertions.assertEquals(OrcFiles.SOME,
PaimonWriteSchema.orcFiles(formatOptions("parquet", "0:avro,5:orc")));
+ Assertions.assertEquals(OrcFiles.SOME,
PaimonWriteSchema.orcFiles(formatOptions("orc", "0:avro")));
+ Assertions.assertEquals(OrcFiles.ALL,
+ PaimonWriteSchema.orcFiles(formatOptions("parquet",
"0:orc,1:orc,2:orc,3:orc,4:orc,5:orc")));
+ Assertions.assertEquals(OrcFiles.NONE,
PaimonWriteSchema.orcFiles(formatOptions("parquet", "6:orc")));
+ Map<String, String> threeLevels = formatOptions("parquet",
"0:orc,1:orc,2:orc");
+ threeLevels.put("num-levels", "3");
+ Assertions.assertEquals(OrcFiles.ALL,
PaimonWriteSchema.orcFiles(threeLevels));
+ }
+
+ @Test
+ public void testBootstrapKeysOfAnOrcTableArePaddedAsTheWrittenKeys() {
+ // The global index of a key-dynamic table is bootstrapped from rows
of the primary keys that are no partition
+ // keys, as the data files give them back, then the partition and the
bucket. ORC gives back 'k' for the
+ // stored 'k ', so on a table that writes ORC files those keys are
padded as the rows written are, while the
+ // partition, the stored one, is kept: 'x' stays 'x'. Other files give
the keys back as they are stored, and
+ // with padding off the writer pads nothing. MUTATION: padding no key,
or the partition -> red.
+ TableSchema schema = new TableSchema(0L, Arrays.asList(
+ new DataField(0, "c", new CharType(false, 3)),
+ new DataField(1, "v", new IntType()),
+ new DataField(2, "p", new CharType(false, 3))),
+ 2, Collections.singletonList("p"),
Collections.singletonList("c"), Collections.emptyMap(), "");
+ String[] columns = {"v", "p", "c"};
+ InternalRow bootstrapRow = GenericRow.of(string("k"), string("x"), 7);
+
+ InternalRow padded = PaimonWriteSchema.create(schema.logicalRowType(),
columns, false, true, null,
+ schema.primaryKeys(), schema.partitionKeys(), OrcFiles.SOME,
false)
+ .bootstrapKeyPadder(schema).apply(bootstrapRow);
+ Assertions.assertEquals("k ", padded.getString(0).toString());
+ Assertions.assertEquals("x", padded.getString(1).toString());
+ Assertions.assertEquals(7, padded.getInt(2));
+ for (PaimonWriteSchema unpadded : Arrays.asList(
+ PaimonWriteSchema.create(schema.logicalRowType(), columns,
false, true, null, schema.primaryKeys(),
+ schema.partitionKeys(), OrcFiles.NONE, false),
+ PaimonWriteSchema.create(schema.logicalRowType(), columns,
false, false, null, schema.primaryKeys(),
+ schema.partitionKeys(), OrcFiles.ALL, false))) {
+ Assertions.assertSame(bootstrapRow,
unpadded.bootstrapKeyPadder(schema).apply(bootstrapRow));
+ }
+ }
+
+ @Test
+ public void
testBootstrappedIndexOfAnOrcTableFindsTheKeyThatTheWriterPads(@TempDir Path
warehouse)
+ throws Exception {
+ // A row of 'k' that a second writer writes into partition p2 of a
cross-partition ORC table, padded to 'k ',
+ // has to find the 'k ' stored in p1 in the index that this writer
bootstraps from the files, so that the
+ // stored row is deleted from p1, or the table keeps two rows of one
key. MUTATION: bootstrapping the keys
+ // as ORC gives them back -> red.
+ FileStoreTable table;
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "cross_partition");
+ catalog.createTable(id, Schema.newBuilder()
+ .column("c", DataTypes.CHAR(3).notNull())
+ .column("v", DataTypes.INT())
+ .column("p", DataTypes.STRING().notNull())
+ .partitionKeys("p")
+ .primaryKey("c")
+ .option("bucket", "-1")
+ .option("file.format", "orc")
+ .build(), false);
+ table = (FileStoreTable) catalog.getTable(id);
+ }
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite(); BatchTableCommit
commit = builder.newCommit()) {
+ write.write(GenericRow.of(string("k "), 1, string("p1")), 0);
+ commit.commit(write.prepareCommit());
+ }
+
+ PaimonWriteSchema writeSchema =
PaimonWriteSchema.create(table.rowType(), new String[] {"c", "v", "p"},
+ false, true, null, table.primaryKeys(), table.partitionKeys(),
+ PaimonWriteSchema.orcFiles(table.options()), false);
+ List<String> written = new ArrayList<>();
+ GlobalIndexAssigner assigner = new GlobalIndexAssigner(table, 1 << 20);
+ assigner.open(1, 0, (row, bucket) -> written.add(String.format("%s
[%s] %s %d",
+ row.getRowKind().shortString(), row.getString(0),
row.getString(2), bucket)));
+ UnaryOperator<InternalRow> bootstrapKeyPadder =
writeSchema.bootstrapKeyPadder(table.schema());
+ new IndexBootstrap(table).bootstrap(1, 0, row ->
assigner.bootstrapKey(bootstrapKeyPadder.apply(row)));
+ assigner.finishBootstrap();
+ assigner.processInput(writeSchema.tableRow(GenericRow.of(string("k"),
2, string("p2"))));
+ Assertions.assertEquals(Arrays.asList("-D [k ] p1 0", "+I [k ] p2
0"), written);
}
@Test
@@ -847,6 +1071,13 @@ public class PaimonWriteSchemaTest {
return BinaryString.fromString(value);
}
+ private static Map<String, String> formatOptions(String format, String
formatPerLevel) {
+ Map<String, String> options = new HashMap<>();
+ options.put("file.format", format);
+ options.put("file.format.per.level", formatPerLevel);
+ return options;
+ }
+
/** A map over key and value arrays, which unlike GenericMap can hold one
key twice. */
private static InternalMap map(Object[] keys, Object[] values) {
InternalArray keyArray = new GenericArray(keys);
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
index 66e7c6febf6..38ccc73488d 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
@@ -43,7 +43,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HexFormat;
import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.TreeMap;
@@ -205,21 +204,15 @@ final class PaimonWriteBinding {
/**
* The fields whose CHAR values the BE writer pads, as
PaimonWriteSchema.create decides it: with {@code charPadding}
- * on, every field but the table's {@code rowkind.field}, from whose value
Paimon decodes the row kind, and of a
- * table that writes ORC files only the partition fields. ORC pads every
CHAR value it writes and strips the
- * trailing spaces from one it reads, so the writer leaves the other
fields of such a table alone. Keep in sync
- * with PaimonWriteSchema.create and PaimonWriteSchema.writesOrcFiles in
be-java-extensions/paimon-scanner.
+ * on, every field but the table's {@code rowkind.field}, from whose value
Paimon decodes the row kind. Keep in
+ * sync with PaimonWriteSchema.create in be-java-extensions/paimon-scanner.
*/
static Predicate<String> charPaddedFields(TableSchema schema, boolean
charPadding) {
if (!charPadding) {
return field -> false;
}
- CoreOptions options = CoreOptions.fromMap(schema.options());
- String rowKindField = options.rowkindField().orElse(null);
- boolean orcFiles =
CoreOptions.FILE_FORMAT_ORC.equals(options.formatType())
- ||
options.fileFormatPerLevel().containsValue(CoreOptions.FILE_FORMAT_ORC);
- List<String> partitionKeys = schema.partitionKeys();
- return field -> !field.equals(rowKindField) && (!orcFiles ||
partitionKeys.contains(field));
+ String rowKindField =
CoreOptions.fromMap(schema.options()).rowkindField().orElse(null);
+ return field -> !field.equals(rowKindField);
}
/**
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
index bb8e12f4a0f..4d327c215fb 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
@@ -372,8 +372,7 @@ public class PaimonWritePlanProvider implements
ConnectorWritePlanProvider {
* Paimon computes the partition and the bucket of the row, while BE would
hash the unpadded values, and several
* writers would write and compact the same bucket. Such a routing field
cannot be routed natively: this returns
* null, and the table gets the distribution of a table whose routing
fields BE cannot hash, a single writer for
- * a primary-key table. A CHAR field that the writer leaves alone, such as
the bucket key of a table that writes
- * ORC files, is routed natively.
+ * a primary-key table. A CHAR field that the writer leaves alone, the
table's rowkind.field, is routed natively.
*/
private ConnectorWriteDistribution fixedBucketDistribution(FileStoreTable
table, boolean charPadding) {
if (table.bucketMode() != BucketMode.HASH_FIXED) {
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
index 63d6fe5f564..9cf92cac527 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
@@ -325,9 +325,9 @@ public class PaimonWriteBindingTest {
@Test
public void staticCharPartitionOfAnOrcTableIsPaddedAsTheRowsAre(@TempDir
Path warehouse) {
- // The BE writer leaves the CHAR values of a table that writes ORC
files unpadded, since ORC pads them
- // itself, but for the partition values, which Paimon reads back as
they are stored. MUTATION: leaving every
- // field of an ORC table unpadded -> red.
+ // The BE writer pads the CHAR values of a table that writes ORC files
as it pads those of any table, so the
+ // static partition value of such a table is padded as its rows are.
MUTATION: leaving the fields of an ORC
+ // table unpadded -> red.
FileStoreTable table = partitionedTable(warehouse,
Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "orc"),
new DataField(1, "part", DataTypes.CHAR(3)));
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
index 997f9ccf72a..c2791bd7d1d 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
@@ -289,25 +289,28 @@ public class PaimonWritePlanProviderTest {
}
@Test
- public void charRouteFieldsThatTheWriterLeavesUnpaddedAreRoutedNatively() {
- // ORC pads every CHAR value it writes, so the writer pads only the
partition values of a table that writes
- // ORC files, and it never pads the table's rowkind.field. BE hashes
the bytes the writer writes for such a
- // field, so the table keeps its parallel bucket writers while padding
is on. MUTATION: falling back for
- // every CHAR routing field while padding is on, or ignoring
file.format.per.level, or the rowkind.field
- // -> red; routing the CHAR partition field of an ORC table natively
-> red.
+ public void
charRouteFieldsOfOrcTablesFallBackToSingleWriterButTheRowKindFieldIsRoutedNatively()
{
+ // The writer pads the CHAR values of a table that writes ORC files as
it pads those of any table, so that
+ // Doris writes the padded keys Spark and Flink write, and its UPDATE
and DELETE rows reach them. Such a
+ // routing field falls back to a single writer, whether file.format or
file.format.per.level names ORC.
+ // The writer never pads the table's rowkind.field, so BE routes that
one natively. MUTATION: routing the
+ // CHAR bucket key or partition field of an ORC table natively, or
falling back for the rowkind.field -> red.
List<DataField> charKey = Arrays.asList(
new DataField(0, "c", DataTypes.CHAR(3).notNull()),
new DataField(1, "v", DataTypes.INT()));
Map<String, String> orc = new HashMap<>();
orc.put(CoreOptions.BUCKET.key(), "4");
orc.put(CoreOptions.FILE_FORMAT.key(), "orc");
+ Assertions.assertEquals(ConnectorWriteDistribution.Mode.GATHER,
writeDistribution(charKey,
+ Collections.emptyList(), Collections.singletonList("c"), orc,
BucketMode.HASH_FIXED, true).getMode());
Assertions.assertEquals("paimon_fixed_bucket",
writeDistribution(charKey, Collections.emptyList(),
- Collections.singletonList("c"), orc, BucketMode.HASH_FIXED,
true).getPartitionFunction());
+ Collections.singletonList("c"), orc, BucketMode.HASH_FIXED,
false).getPartitionFunction());
Map<String, String> orcAtALevel = new HashMap<>();
orcAtALevel.put(CoreOptions.BUCKET.key(), "4");
orcAtALevel.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(), "5:orc");
- Assertions.assertEquals("paimon_fixed_bucket",
writeDistribution(charKey, Collections.emptyList(),
- Collections.singletonList("c"), orcAtALevel,
BucketMode.HASH_FIXED, true).getPartitionFunction());
+ Assertions.assertEquals(ConnectorWriteDistribution.Mode.GATHER,
writeDistribution(charKey,
+ Collections.emptyList(), Collections.singletonList("c"),
orcAtALevel, BucketMode.HASH_FIXED, true)
+ .getMode());
// A table partitioned by a CHAR field and bucketed by id.
List<DataField> charPartition = Arrays.asList(
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
index e1a699d7a60..c809ba7400c 100644
---
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
@@ -50,6 +50,57 @@
[ab] 2 20
[abc] 3 60
+-- !orc_spark_dml --
+[ab] 2 20
+[abc] 3 60
+
+-- !orc_spark_upsert --
+[ab] 2 7
+[abc] 3 60
+[k] 1 8
+
+-- !orc_spark_one_bucket_update --
+[ab] 2 20
+[k] 1 1
+[x] 1 3
+
+-- !orc_spark_one_bucket_delete --
+[k] 1 1
+[x] 1 3
+
+-- !orc_spark_dynamic_bucket --
+[ab] 2 20
+[abc] 3 60
+
+-- !orc_spark_rowkind_delete --
+[ab] +I 2
+
+-- !orc_spark_merge --
+[ab] 2 2
+[k] 1 10
+[q] 1 30
+
+-- !orc_spark_merge_trailing_space_fails --
+[ab] 2 2
+[k] 1 10
+[q] 1 30
+
+-- !orc_cross_partition_upsert --
+[k] 1 2 p2
+[m] 1 2 p2
+
+-- !orc_partition_key_merge --
+[k] 9 [x ]
+
+-- !avro_level0_keys_read_back --
+[k ] 3 2
+[q] 1 1
+
+-- !avro_level0_merge --
+[abc] 3 30
+[k ] 3 20
+[q] 1 1
+
-- !bucketed_route_plan --
PhysicalConnectorTableSink
--PhysicalDistribute[DistributionSpecGather]
@@ -65,7 +116,7 @@ PhysicalConnectorTableSink
-- !orc_bucketed_route_plan --
PhysicalConnectorTableSink
---PhysicalDistribute[DistributionSpecExternalTableSinkHashPartitioned]
+--PhysicalDistribute[DistributionSpecGather]
----PhysicalProject
------PhysicalTVFRelation
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
index cdb8602d454..bea00877274 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
@@ -118,6 +118,91 @@ suite("test_paimon_write_char_padding",
"p0,external,paimon") {
'changelog-producer' = 'full-compaction'
);
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_orc_spark;
+ CREATE TABLE paimon.${dbName}.t_char_orc_spark (
+ c CHAR(3), v INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'c',
+ 'bucket' = '4',
+ 'file.format' = 'orc'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_orc_spark_one_bucket;
+ CREATE TABLE paimon.${dbName}.t_char_orc_spark_one_bucket (
+ c CHAR(3), v INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'c',
+ 'bucket' = '1',
+ 'file.format' = 'orc'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_orc_spark_dynamic;
+ CREATE TABLE paimon.${dbName}.t_char_orc_spark_dynamic (
+ c CHAR(3), v INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'c',
+ 'bucket' = '-1',
+ 'dynamic-bucket.target-row-num' = '1',
+ 'file.format' = 'orc'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_orc_spark_merge;
+ CREATE TABLE paimon.${dbName}.t_char_orc_spark_merge (
+ c CHAR(3), v INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'c',
+ 'bucket' = '4',
+ 'file.format' = 'orc'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_orc_cross_partition;
+ CREATE TABLE paimon.${dbName}.t_char_orc_cross_partition (
+ c CHAR(3), v INT, p STRING
+ ) USING paimon
+ PARTITIONED BY (p)
+ TBLPROPERTIES (
+ 'primary-key' = 'c',
+ 'bucket' = '-1',
+ 'file.format' = 'orc'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_orc_partition_key_merge;
+ CREATE TABLE paimon.${dbName}.t_char_orc_partition_key_merge (
+ c CHAR(3), v INT, p CHAR(3)
+ ) USING paimon
+ PARTITIONED BY (p)
+ TBLPROPERTIES (
+ 'primary-key' = 'c,p',
+ 'bucket' = '1',
+ 'file.format' = 'orc'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_avro_level0_merge;
+ CREATE TABLE paimon.${dbName}.t_char_avro_level0_merge (
+ c CHAR(3), v INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'c',
+ 'bucket' = '1',
+ 'file.format' = 'orc',
+ 'file.format.per.level' = '0:avro'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_char_orc_spark_rowkind;
+ CREATE TABLE paimon.${dbName}.t_char_orc_spark_rowkind (
+ c CHAR(3), rk STRING, v INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'c',
+ 'bucket' = '4',
+ 'rowkind.field' = 'rk',
+ 'file.format' = 'orc'
+ );
+
DROP TABLE IF EXISTS paimon.${dbName}.t_char_rowkind_key;
CREATE TABLE paimon.${dbName}.t_char_rowkind_key (
c CHAR(3), rk STRING, v INT
@@ -229,6 +314,13 @@ suite("test_paimon_write_char_padding",
"p0,external,paimon") {
) duplicate key(pos) distributed by hash(pos) buckets 1
properties ('replication_num' = '1')
"""
+ sql """drop table if exists internal.${dbName}.char_orc_merge_source"""
+ sql """
+ create table internal.${dbName}.char_orc_merge_source (
+ c char(3), v int
+ ) duplicate key(c) distributed by hash(c) buckets 1
+ properties ('replication_num' = '1')
+ """
try {
// A primary key Doris writes is padded, and Spark's write of the same
key replaces it.
@@ -358,14 +450,153 @@ suite("test_paimon_write_char_padding",
"p0,external,paimon") {
sql """UPDATE t_char_pk SET v = 40 WHERE v = 4"""
order_qt_dml_matches_unpadded_keys """SELECT concat('[', c, ']'),
char_length(c), v FROM t_char_pk"""
- // ORC pads every CHAR value it writes and strips the trailing spaces
from one it reads, so only the
- // partition values of an ORC table are padded: padding a key would
only change its bucket, and the DELETE
- // and UPDATE below, which read the keys back without the spaces,
would hash other buckets and miss them.
+ // ORC pads every CHAR value it writes and strips the trailing spaces
from one it reads, so Doris reads back
+ // 'k' for the 'k ' that it, Spark and Flink write. Paimon puts a row
in the bucket of its key as the writer
+ // holds it, and merges only the data files whose key ranges, kept as
the writer held the keys, overlap.
+ // So the rows of an ORC table are padded whatever they do, and the
DELETE and UPDATE below, which read the
+ // keys back without the spaces, write them padded and reach them.
sql """INSERT INTO t_char_orc_bucketed VALUES ('k', 1), ('ab', 2),
('x', 3), ('m', 4), ('zz', 5), ('abc', 6)"""
sql """DELETE FROM t_char_orc_bucketed WHERE v IN (1, 3, 4, 5)"""
sql """UPDATE t_char_orc_bucketed SET v = v * 10 WHERE v IN (2, 6)"""
order_qt_orc_dml_reaches_keys """SELECT concat('[', c, ']'),
char_length(c), v FROM t_char_orc_bucketed"""
+ // The same reaches the padded keys Spark writes into ORC tables:
written back unpadded, the key of a DELETE
+ // or UPDATE would hash another of 4 buckets, ask the dynamic bucket
index for a key it does not hold, or, as
+ // the smallest key 'ab' of a one-bucket table, sort below the range
['ab ', 'x '] of the file holding it.
+ // A key Doris writes again replaces the one Spark wrote.
+ spark_paimon_multi """
+ INSERT INTO paimon.${dbName}.t_char_orc_spark
+ VALUES ('k', 1), ('ab', 2), ('x', 3), ('m', 4), ('zz', 5),
('abc', 6);
+ INSERT INTO paimon.${dbName}.t_char_orc_spark_one_bucket VALUES
('k', 1), ('ab', 2), ('x', 3);
+ INSERT INTO paimon.${dbName}.t_char_orc_spark_dynamic
+ VALUES ('k', 1), ('ab', 2), ('x', 3), ('m', 4), ('zz', 5),
('abc', 6);
+ INSERT INTO paimon.${dbName}.t_char_orc_spark_merge VALUES ('k',
1), ('ab', 2);
+ INSERT INTO paimon.${dbName}.t_char_orc_spark_rowkind VALUES ('k',
'+I', 1), ('ab', '+I', 2);
+ """
+ sql """refresh table t_char_orc_spark"""
+ sql """refresh table t_char_orc_spark_one_bucket"""
+ sql """refresh table t_char_orc_spark_dynamic"""
+ sql """refresh table t_char_orc_spark_merge"""
+ sql """refresh table t_char_orc_spark_rowkind"""
+ sql """DELETE FROM t_char_orc_spark WHERE v IN (1, 3, 4, 5)"""
+ sql """UPDATE t_char_orc_spark SET v = v * 10 WHERE v IN (2, 6)"""
+ order_qt_orc_spark_dml """SELECT concat('[', c, ']'), char_length(c),
v FROM t_char_orc_spark"""
+ sql """INSERT INTO t_char_orc_spark VALUES ('ab', 7), ('k', 8)"""
+ order_qt_orc_spark_upsert """SELECT concat('[', c, ']'),
char_length(c), v FROM t_char_orc_spark"""
+ sql """UPDATE t_char_orc_spark_one_bucket SET v = 20 WHERE c = 'ab'"""
+ order_qt_orc_spark_one_bucket_update """
+ SELECT concat('[', c, ']'), char_length(c), v FROM
t_char_orc_spark_one_bucket
+ """
+ sql """DELETE FROM t_char_orc_spark_one_bucket WHERE c = 'ab'"""
+ order_qt_orc_spark_one_bucket_delete """
+ SELECT concat('[', c, ']'), char_length(c), v FROM
t_char_orc_spark_one_bucket
+ """
+ sql """DELETE FROM t_char_orc_spark_dynamic WHERE v IN (1, 3, 4, 5)"""
+ sql """UPDATE t_char_orc_spark_dynamic SET v = v * 10 WHERE v IN (2,
6)"""
+ order_qt_orc_spark_dynamic_bucket """
+ SELECT concat('[', c, ']'), char_length(c), v FROM
t_char_orc_spark_dynamic
+ """
+
+ // A '-D' row of a rowkind.field table carries the key Doris read
back, and is padded on an ORC table too.
+ sql """INSERT INTO t_char_orc_spark_rowkind SELECT c, '-D', v FROM
t_char_orc_spark_rowkind WHERE v = 1"""
+ order_qt_orc_spark_rowkind_delete """SELECT concat('[', c, ']'), rk, v
FROM t_char_orc_spark_rowkind"""
+
+ // MERGE matches 'k' to the 'k ' Spark wrote, as Doris reads it back
as 'k', and inserts 'q' padded. A
+ // source key that ends with a space, 'ab ', matches none of the keys
read back without their spaces, yet
+ // padding would make it the stored 'ab ' that MERGE did not match, so
that MERGE fails and writes nothing.
+ sql """INSERT INTO internal.${dbName}.char_orc_merge_source VALUES
('k', 10), ('q', 30)"""
+ sql """
+ MERGE INTO t_char_orc_spark_merge t
+ USING internal.${dbName}.char_orc_merge_source s ON t.c = s.c
+ WHEN MATCHED THEN UPDATE SET v = s.v
+ WHEN NOT MATCHED THEN INSERT (c, v) VALUES (s.c, s.v)
+ """
+ order_qt_orc_spark_merge """SELECT concat('[', c, ']'),
char_length(c), v FROM t_char_orc_spark_merge"""
+ sql """INSERT INTO internal.${dbName}.char_orc_merge_source VALUES
('ab ', 40)"""
+ test {
+ sql """
+ MERGE INTO t_char_orc_spark_merge t
+ USING internal.${dbName}.char_orc_merge_source s ON t.c = s.c
+ WHEN NOT MATCHED THEN INSERT (c, v) VALUES (s.c, s.v)
+ """
+ exception "rtrim(<key>)"
+ }
+ order_qt_orc_spark_merge_trailing_space_fails """
+ SELECT concat('[', c, ']'), char_length(c), v FROM
t_char_orc_spark_merge
+ """
+
+ // Each writer of a cross-partition table bootstraps its global key
index from the keys the files hold, which
+ // ORC gives back without the trailing spaces that the keys Doris and
Spark write carry. The writer indexes
+ // them padded, so a key written again into another partition, 'k'
written by Doris and 'm' by Spark, finds
+ // the stored row and deletes it from its old partition: each key
keeps one row, in its last partition.
+ spark_paimon """INSERT INTO
paimon.${dbName}.t_char_orc_cross_partition VALUES ('m', 1, 'p1')"""
+ sql """refresh table t_char_orc_cross_partition"""
+ sql """INSERT INTO t_char_orc_cross_partition VALUES ('k', 1, 'p1')"""
+ sql """INSERT INTO t_char_orc_cross_partition VALUES ('k', 2, 'p2'),
('m', 2, 'p2')"""
+ order_qt_orc_cross_partition_upsert """
+ SELECT concat('[', c, ']'), char_length(c), v, p FROM
t_char_orc_cross_partition
+ """
+
+ // Paimon keeps a partition value in the partition, so Doris reads it
back as it is stored, padded, from an ORC
+ // table too. MERGE does not match 'x' to the stored 'x ', and
padding 'x' would let it replace the row it did
+ // not match, so that MERGE fails as on a table that writes no ORC
files; one that pads the partition key in
+ // the MERGE matches the row and updates it.
+ spark_paimon """INSERT INTO
paimon.${dbName}.t_char_orc_partition_key_merge VALUES ('k', 1, 'x')"""
+ sql """refresh table t_char_orc_partition_key_merge"""
+ test {
+ sql """
+ MERGE INTO t_char_orc_partition_key_merge t
+ USING (SELECT 'k' AS c, 9 AS v, 'x' AS p) s ON t.c = s.c AND
t.p = s.p
+ WHEN MATCHED THEN UPDATE SET v = s.v
+ WHEN NOT MATCHED THEN INSERT (c, v, p) VALUES (s.c, s.v, s.p)
+ """
+ exception "rpad(<key>, 3, ' ')"
+ }
+ sql """
+ MERGE INTO t_char_orc_partition_key_merge t
+ USING (SELECT 'k' AS c, 9 AS v, 'x' AS p) s ON t.c = s.c AND t.p =
rpad(s.p, 3, ' ')
+ WHEN MATCHED THEN UPDATE SET v = s.v
+ WHEN NOT MATCHED THEN INSERT (c, v, p) VALUES (s.c, s.v, rpad(s.p,
3, ' '))
+ """
+ order_qt_orc_partition_key_merge """
+ SELECT concat('[', c, ']'), v, concat('[', p, ']') FROM
t_char_orc_partition_key_merge
+ """
+
+ // A table that writes ORC files at some levels only, here all but
level 0, which writes Avro, gives a key back
+ // padded from its Avro files and without its trailing spaces from its
ORC files: 'q', whose two files the
+ // compaction merges into an ORC file (it would only move a single
file up a level), comes back as 'q', and
+ // 'k', written after, as 'k '. MERGE does not match 'k' to the
stored 'k ', nor 'q ' to the stored 'q', and
+ // padding either would let it replace the row it did not match, so
both fail; a key as long as the column,
+ // which padding does not change, is inserted, and 'k ', which
matches, is updated.
+ spark_paimon_multi """
+ INSERT INTO paimon.${dbName}.t_char_avro_level0_merge VALUES ('q',
0);
+ INSERT INTO paimon.${dbName}.t_char_avro_level0_merge VALUES ('q',
1);
+ CALL paimon.sys.compact(table =>
'${dbName}.t_char_avro_level0_merge');
+ INSERT INTO paimon.${dbName}.t_char_avro_level0_merge VALUES ('k',
2);
+ """
+ sql """refresh table t_char_avro_level0_merge"""
+ order_qt_avro_level0_keys_read_back """
+ SELECT concat('[', c, ']'), char_length(c), v FROM
t_char_avro_level0_merge
+ """
+ for (String key : ["k", "q "]) {
+ test {
+ sql """
+ MERGE INTO t_char_avro_level0_merge t
+ USING (SELECT '${key}' AS c, 9 AS v) s ON t.c = s.c
+ WHEN MATCHED THEN UPDATE SET v = s.v
+ WHEN NOT MATCHED THEN INSERT (c, v) VALUES (s.c, s.v)
+ """
+ exception "at some levels only"
+ }
+ }
+ sql """
+ MERGE INTO t_char_avro_level0_merge t
+ USING (SELECT 'k ' AS c, 20 AS v UNION ALL SELECT 'abc', 30) s ON
t.c = s.c
+ WHEN MATCHED THEN UPDATE SET v = s.v
+ WHEN NOT MATCHED THEN INSERT (c, v) VALUES (s.c, s.v)
+ """
+ order_qt_avro_level0_merge """SELECT concat('[', c, ']'),
char_length(c), v FROM t_char_avro_level0_merge"""
+
// BE cannot route a CHAR bucket key that the writer pads by Paimon's
bucket, since it would hash the
// unpadded value. Rows of one bucket would then reach several of the
four writers, each would compact
// the bucket, and Paimon would give up committing the compaction, and
with it the changelog, after the
@@ -385,8 +616,7 @@ suite("test_paimon_write_char_padding",
"p0,external,paimon") {
SELECT snapshot_id, commit_kind, changelog_record_count FROM
t_char_bucketed\$snapshots
"""
- // The writer leaves the CHAR bucket key of an ORC table unpadded, as
ORC pads it itself, so BE routes the
- // rows to the four bucket writers by the bytes the writer writes:
each bucket still has one writer, and both
+ // The writer pads the CHAR bucket key of an ORC table too, so that
table also gets one writer, and both
// snapshots of the full-compaction table are committed.
sql """set parallel_pipeline_task_num = 4"""
qt_orc_bucketed_route_plan """
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
