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 ea34d1c85b3 [fix](paimon) Fail scans of binary partitions with legacy
partition names (#69571)
ea34d1c85b3 is described below
commit ea34d1c85b3854291358dff74bac039075146bbe
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 26 07:25:28 2026 +0000
[fix](paimon) Fail scans of binary partitions with legacy partition names
(#69571)
### What problem does this PR solve?
Issue Number: None
Related PR: #69567
Problem Summary:
**In short.** A query of a Paimon table with a BINARY / VARBINARY
partition column now fails with an error naming
`'partition.legacy-name'` when it would read a non-NULL binary partition
whose files Paimon cannot locate, instead of silently returning only the
rows of the other partitions. This is the read-side counterpart of
#69567, which rejects writes to such tables.
**Background.** Paimon looks for the files of a partition under the
directory its partition values name, computed when the split is planned;
a writer puts the files under the name it computes when it writes. With
`partition.legacy-name = true`, the default, the name of a value is its
`toString()`, which for a binary value is the Java array object
(`[B@1b6d3586`) rather than its bytes, so every resolution differs. With
`partition.legacy-name = false` it is the text of the bytes (`p=AB` for
`X'4142'`). The option can be changed on an existing table, which
changes how existing partitions are looked up but does not move their
files. The BE skips a file of an external table that no longer exists
(`ignore_not_found_file_in_external_table`, default true), which is
meant for files removed between planning and reading.
**The problem, and what it cost.** Measured on master with Paimon 2.1.0,
on a `file://` Paimon catalog of a local FE + BE, with tables created
and written through the Paimon API:
| Table | Paimon API read | Doris before this PR |
|---|---|---|
| legacy names (default), rows in `X'4142'`, `X'EF'`, NULL | fails:
`FileNotFoundException` (`p=%5BB@...`) | `SELECT *` returns only the
NULL row; `COUNT(*)`, answered from metadata, returns 3 |
| legacy names when `X'4142'` was written, then `'partition.legacy-name'
= 'false'` and rows in `X'4344'`, NULL | fails: looks for `p=AB`, the
file is under `p=%5BB@...` | returns 2 of 3 rows, the `X'4142'` row is
missing |
| new names when `X'4142'` and NULL were written, then
`'partition.legacy-name' = 'true'` | fails | returns 1 of 2 rows |
| legacy names, rows only in the NULL partition | reads | reads
correctly |
So the query succeeds with a partial result and nothing tells the user.
Following the obvious fix, setting `'partition.legacy-name' = 'false'`,
does not make the old partitions readable either; it only moves the loss
to the files written before the change.
**How this PR fixes it.** `PaimonScanPlanProvider.planScanInternal`
checks every data split that will read files (after the splits whose
`COUNT(*)` is answered from metadata) whose binary partition value is
not NULL:
- if the split is looked up under legacy names (the options of its
table, or of its branch for a fallback read), planning fails;
- otherwise, if one of its data files was written under a schema whose
options use legacy names (the file's `schemaId`, looked up once per
schema and branch, inside the catalog authentication) and is also
missing at its planned path (`bucketPath/fileName`, the path the native
reader is given), planning fails and names that file. The schema alone
does not prove how a file was named: a writer's dynamic options
(`spark.paimon.*`, Flink hints, `Table.copy`) keep the schema ID, so a
file written with a dynamic `'partition.legacy-name' = 'false'` on a
table that keeps the default refers to a schema with legacy names while
it sits under the bytes of its value, and it is read once the table uses
those names;
- and, on a table with `'index-file-in-data-file-dir' = 'true'`, every
deletion vector the split carries must exist, whether or not its data
file has an external path, or planning fails and names the deletion
vector file and its data file. Such a table keeps its deletion vectors
with its data files: one written while no external data path is
configured records only its file name and is looked up under the
directory of its partition, which a legacy name never finds, while its
data file keeps its external path, and the BE skips the whole data file
when it cannot open the deletion vector. The option is immutable, so
every writer used the reading table's value. Deletion vectors under the
table's `index/` directory do not depend on partition names and are not
checked.
A NULL partition is named after `partition.default-name` under both
namings, and a data file with an external path is read from the path its
metadata records, so neither is rejected; the deletion vectors of a file
with an external path are still checked. The system tables read through
JNI (`$audit_log`, `$row_tracking`) report the options of the main
branch, while a fallback split is planned and read under its own branch,
so the check resolves the branch that produced each split from the
pinned source of the wrapper, as it does for data tables. `$binlog`
exposes its partition columns as arrays and is not checked; Paimon's own
reader fails on a file it cannot find. A
`scan.file-creation-time-millis` read of a fallback table plans the main
branch through its `SnapshotReader`, and its plain splits are judged
under the main branch. The error names the table, the partition and the
option. The existence checks run inside the catalog authentication, once
per path and scan: a table that keeps its index files under `index/` and
whose files were written under its current naming makes no extra
filesystem calls.
**Not covered.** A file written with a dynamic `'partition.legacy-name'
= 'true'` (or by a Paimon writer too old to know the option) while its
schema says `'false'` sits under a legacy name but is looked up under
its bytes, and the BE still skips it, as on master and branch-4.2.
Catching it would take an existence call for every file of every
non-NULL binary partition on every scan.
**Results.** On the same local cluster, after this PR:
| Query | Doris after this PR |
|---|---|
| `SELECT *` of the legacy table (also with `force_jni_scanner = true`)
| fails: `Cannot read Paimon table d.api_legacy: the table uses legacy
partition names ('partition.legacy-name' = 'true'), under which Paimon
names partition p = 0xEF after a Java array object rather than its
bytes, so the files of the partition are never found. Paimon cannot read
the partition either; ...` |
| `SELECT COUNT(*)` of the legacy table | 3, answered from metadata as
before |
| `SELECT * ... WHERE p IS NULL` of the legacy table | the NULL row;
Paimon prunes the other partitions |
| `SELECT *` of the table switched to new names | fails: `... data file
data-b3abc825-...-0.parquet of partition p = 0x4142 was written while
the table used legacy partition names ...`, the file the Paimon API read
reports as not found |
| `SELECT *` of the table switched back to legacy names | fails: `...
the table uses legacy partition names ...` |
| `SELECT *` of the legacy table with only NULL partitions, and of
tables with new names | all rows, the same as the Paimon API read |
`PaimonScanPlanProviderTest` adds eight tests: a scan under legacy names
fails while the NULL partition and `COUNT(*)` still plan; a file written
under legacy names fails after the option changes to false, and a file
written under the new names fails after it changes back to true; a file
at an external path still plans and points at an existing file; a file
at an external path whose deletion vector was written after the external
paths were removed (persisted the way
`PaimonSparkWriter.persistDeletionVectors` does it for a DELETE) fails
and names the deletion vector, while a deletion vector written with the
external paths in place still plans and points at an existing file; a
file written through `table.copy({"partition.legacy-name": "false"})` on
a table that keeps the default fails as a legacy lookup, then plans once
the option is altered to `false`; a fallback table whose main branch
holds only a NULL partition and whose fallback branch holds a non-NULL
binary partition plans the data table and `$audit_log` when the fallback
branch names partitions by bytes, and the planned `$audit_log` splits
read both rows back through Paimon, while both fail at planning with the
legacy-name error when only the main branch names them by bytes; and a
`scan.file-creation-time-millis` read of a fallback table with a
non-NULL binary partition plans the main branch's file.
`nativeBinaryPartitionsKeepAllPathValues`, added by #68786 for the
rendering of binary partition values, now creates its table with
`'partition.legacy-name' = 'false'`: under legacy names the files it
planned did not exist. The new tests fail when the lookup check, the
check of the files' schemas, the exemption of external paths or of NULL
values is removed, when the check moves before the `COUNT(*)` splits,
when the deletion vectors are not checked, when every deletion vector
fails without its existence being checked, when a file whose schema uses
legacy names fails without its existence being checked, when a system
table read through JNI is checked under the wrapper's options, when
every split is judged under the main branch, or when a split that does
not come from the fallback scan is cast to a fallback split (the
file-creation-time read then failed with a `ClassCastException`). The
853 tests of fe-connector-paimon run without failures (one is skipped,
as before).
No regression test is added: Doris cannot create a binary column, so the
tables would have to come from the Spark container of the Paimon docker
environment, which was not run locally.
---
.../connector/paimon/PaimonScanPlanProvider.java | 125 +++++++-
.../paimon/PaimonScanPlanProviderTest.java | 326 +++++++++++++++++++++
2 files changed, 450 insertions(+), 1 deletion(-)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
index 548907b5249..074639d7f3e 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
@@ -59,6 +59,7 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
import org.apache.paimon.options.Options;
@@ -86,6 +87,7 @@ import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.MapType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.InstantiationUtil;
@@ -989,6 +991,15 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
// $ro wraps the pinned file-store table; resolve its schema
dictionary once, only if native is considered.
java.util.function.Supplier<Table> legacyOrcSchemaTable =
com.google.common.base.Suppliers.memoize(
() -> resolveSchemaDictTable(table, paimonHandle));
+ // The wrappers read through JNI report the options of the main
branch, but their splits come from a branch
+ // of the pinned source, which the timestamp-history checks resolve
the same way.
+ java.util.function.Supplier<Table> binaryPartitionSourceTable =
paimonHandle.isForceJni()
+ ? com.google.common.base.Suppliers.memoize(() ->
withSchemaHistoryAuthentication(
+ () -> resolveTimestampHistoryTable(table,
paimonHandle)))
+ : legacyOrcSchemaTable;
+ int[] binaryPartitionFields = binaryPartitionFields(table);
+ Map<FileStoreTable, Map<Long, Boolean>> legacyPartitionNameSchemas =
new java.util.IdentityHashMap<>();
+ Map<String, Boolean> fileExistence = new HashMap<>();
// Process DataSplits
for (DataSplit dataSplit : dataSplits) {
if (isCountPushdownSplit(countPushdown, dataSplit)) {
@@ -998,6 +1009,10 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
}
continue;
}
+ if (binaryPartitionFields.length > 0) {
+ checkBinaryPartitionFiles(paimonHandle, table, dataSplit,
binaryPartitionFields,
+ binaryPartitionSourceTable,
legacyPartitionNameSchemas, fileExistence);
+ }
Map<String, String> partitionValues = getPartitionInfoMap(
table, dataSplit.partition(), session.getTimeZone());
@@ -1428,7 +1443,7 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
- throw new RuntimeException("Failed to read Paimon timestamp schema
history", e);
+ throw new RuntimeException("Failed to access Paimon table files",
e);
}
}
@@ -2286,6 +2301,11 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
FileStoreTable base =
PaimonTableDecorators.unwrapToFallbackOrBase(table);
if (base instanceof FallbackReadFileStoreTable) {
FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
base;
+ if (!(split instanceof FallbackReadFileStoreTable.FallbackSplit)) {
+ // A split planned through the pair's SnapshotReader (a
file-creation-time read) comes from its
+ // wrapped branch.
+ return pair.wrapped();
+ }
boolean fallback = ((FallbackReadFileStoreTable.FallbackSplit)
split).isFallback();
// Match the SDK reader's branch selection before reading either
options or historical schemas.
return PaimonReaderOptions.isWrappedFirst(pair) != fallback ?
pair.wrapped() : pair.other();
@@ -2293,6 +2313,109 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
return base;
}
+ /** Positions in the partition row of the partition columns that hold
binary values. */
+ private static int[] binaryPartitionFields(Table table) {
+ RowType partitionType = table.rowType().project(table.partitionKeys());
+ return java.util.stream.IntStream.range(0,
partitionType.getFieldCount()).filter(field -> {
+ DataTypeRoot typeRoot =
partitionType.getTypeAt(field).getTypeRoot();
+ return typeRoot == DataTypeRoot.BINARY || typeRoot ==
DataTypeRoot.VARBINARY;
+ }).toArray();
+ }
+
+ /**
+ * Paimon finds the files of a partition under the directory its values
name. With {@code partition.legacy-name}
+ * = true, the default, it names a partition after the string of each
value, which for a non-NULL binary value
+ * is the Java array object ({@code [B@1b6d3586}) rather than its bytes,
and every resolution of the name
+ * differs. A file written under such a name, or looked up under one, is
never found: Paimon's own read fails,
+ * while the BE skips a missing file of an external table and would
silently drop the partition's rows. A file
+ * with an external path is found by the path its metadata records. A
system wrapper reports the options of the
+ * main branch while a fallback split is planned and read under its own
branch, so the options and schemas come
+ * from the branch that produced the split; for the wrappers read through
JNI that branch is resolved from the
+ * pinned source of the wrapper.
+ *
+ * <p>The schema a file refers to does not prove how the file was named,
because dynamic options
+ * ({@code spark.paimon.*}, Flink hints, {@code Table.copy}) keep the
schema ID: a file whose schema uses legacy
+ * names fails only if it is also missing at its planned path. With {@code
index-file-in-data-file-dir} = true, a
+ * deletion vector written without an external data path is looked up
under the directory of its partition too,
+ * whatever the path of its data file, and the BE skips the whole data
file when it cannot open the deletion
+ * vector, so every deletion vector must exist. Each path is checked once
per scan.
+ */
+ private void checkBinaryPartitionFiles(PaimonTableHandle handle, Table
table, DataSplit split,
+ int[] binaryPartitionFields, Supplier<Table> schemaTable,
+ Map<FileStoreTable, Map<Long, Boolean>>
legacyPartitionNameSchemas, Map<String, Boolean> fileExistence) {
+ BinaryRow partition = split.partition();
+ OptionalInt nonNullField = Arrays.stream(binaryPartitionFields)
+ .filter(field -> !partition.isNullAt(field)).findFirst();
+ if (!nonNullField.isPresent()) {
+ return;
+ }
+ int binaryField = nonNullField.getAsInt();
+ FileStoreTable splitTable = schemaTable.get() == null
+ ? null : legacyOrcSplitTable((FileStoreTable)
schemaTable.get(), split);
+ Table optionsTable = splitTable == null ? table : splitTable;
+ CoreOptions options = CoreOptions.fromMap(optionsTable.options());
+ boolean legacyLookup = options.legacyPartitionName();
+ // Deletion vectors under the table's index directory do not depend on
partition names. The option is
+ // immutable, so every writer of the table used this value.
+ List<DeletionFile> deletionFiles = options.indexFileInDataFileDir() ?
split.deletionFiles().orElse(null) : null;
+ List<DataFileMeta> files = split.dataFiles();
+ for (int i = 0; i < files.size(); i++) {
+ DataFileMeta file = files.get(i);
+ if (!file.externalPath().isPresent()) {
+ if (legacyLookup) {
+ throw binaryPartitionReadError(handle, "the table uses
legacy partition names"
+ + " ('partition.legacy-name' = 'true'), under
which Paimon names partition "
+ + binaryPartitionName(table, partition,
binaryField) + " after a Java array object"
+ + " rather than its bytes, so the files of the
partition are never found");
+ }
+ if (splitTable != null && legacyPartitionNameSchemas
+ .computeIfAbsent(splitTable, ignored -> new
HashMap<>())
+ .computeIfAbsent(file.schemaId(), id ->
withSchemaHistoryAuthentication(() -> CoreOptions
+
.fromMap(splitTable.schemaManager().schema(id).options()).legacyPartitionName()))
+ && !fileExists(optionsTable.fileIO(),
split.bucketPath() + "/" + file.fileName(),
+ fileExistence)) {
+ throw binaryPartitionReadError(handle, "data file " +
file.fileName() + " of partition "
+ + binaryPartitionName(table, partition,
binaryField) + " was written while the table"
+ + " used legacy partition names
('partition.legacy-name' = 'true'), under which Paimon"
+ + " names the partition after a Java array object
rather than its bytes, so the file is"
+ + " not under the directory of the partition");
+ }
+ }
+ DeletionFile deletionFile = deletionFiles == null ? null :
deletionFiles.get(i);
+ if (deletionFile != null && !fileExists(optionsTable.fileIO(),
deletionFile.path(), fileExistence)) {
+ throw binaryPartitionReadError(handle, "deletion vector file "
+ + new Path(deletionFile.path()).getName() + " of data
file " + file.fileName()
+ + " of partition " + binaryPartitionName(table,
partition, binaryField)
+ + " does not exist: the table keeps index files with
its data files"
+ + " ('index-file-in-data-file-dir' = 'true'), so a
deletion vector written without an external"
+ + " path is looked up under the directory of its
partition, and legacy partition names"
+ + " ('partition.legacy-name' = 'true') name the
partition after a Java array object rather"
+ + " than its bytes");
+ }
+ }
+ }
+
+ private static String binaryPartitionName(Table table, BinaryRow
partition, int binaryField) {
+ return table.partitionKeys().get(binaryField) + " = 0x"
+ +
java.util.HexFormat.of().withUpperCase().formatHex(partition.getBinary(binaryField));
+ }
+
+ private static DorisConnectorException
binaryPartitionReadError(PaimonTableHandle handle, String reason) {
+ return new DorisConnectorException("Cannot read Paimon table " +
handle.getDatabaseName() + "."
+ + handle.getTableName() + ": " + reason + ". Paimon cannot
read the partition either; a table with a"
+ + " binary partition column needs 'partition.legacy-name' =
'false' before its data is written");
+ }
+
+ private boolean fileExists(FileIO fileIO, String path, Map<String,
Boolean> fileExistence) {
+ return fileExistence.computeIfAbsent(path, ignored ->
withSchemaHistoryAuthentication(() -> {
+ try {
+ return fileIO.exists(new Path(path));
+ } catch (IOException e) {
+ throw new DorisConnectorException("Failed to check whether
Paimon file " + path + " exists", e);
+ }
+ }));
+ }
+
private static boolean requiresCrossFamilyTimestampReader(FileStoreTable
table, Optional<List<RawFile>> rawFiles,
Set<Integer> readFieldIds, Map<Long, Boolean> schemaTimestamps) {
RowType readType = new RowType(table.rowType().getFields().stream()
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
index d29cae9ba8d..bbb0513a29a 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
@@ -50,9 +50,14 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.io.DataInputViewStreamWrapper;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
@@ -69,11 +74,13 @@ 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.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.RawFile;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.system.ReadOptimizedTable;
+import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -892,9 +899,11 @@ public class PaimonScanPlanProviderTest {
new
org.apache.paimon.fs.Path(warehouse.resolve(format).toUri()))) {
catalog.createDatabase("db", false);
Identifier id = Identifier.create("db", "binary_parts");
+ // Legacy partition names cannot name a binary partition
(legacyNamedBinaryPartitionsFailScanPlanning).
catalog.createTable(id, Schema.newBuilder().column("id",
DataTypes.INT())
.column("key", DataTypes.BYTES()).column("region",
DataTypes.STRING())
.partitionKeys("key", "region").option("bucket", "-1")
+ .option("partition.legacy-name", "false")
.option("file.format", format).build(), false);
Table table = catalog.getTable(id);
BatchWriteBuilder writer = table.newBatchWriteBuilder();
@@ -925,6 +934,323 @@ public class PaimonScanPlanProviderTest {
}
}
+ @Test
+ public void legacyNamedBinaryPartitionsFailScanPlanning(@TempDir Path
warehouse) throws Exception {
+ // With partition.legacy-name = true, the default, Paimon looks a
non-NULL binary partition up under the name
+ // of a Java array object, so the planned files never exist and the BE
would skip them as missing files.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Table legacy = binaryPartitionTable(catalog, "legacy",
Collections.emptyMap(),
+ new byte[] {0x41, 0x42}, null);
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> planBinaryPartitionScan(legacy, false));
+ Assertions.assertTrue(error.getMessage().startsWith("Cannot read
Paimon table db.legacy: the table uses"
+ + " legacy partition names ('partition.legacy-name' =
'true'), under which Paimon names partition"
+ + " key = 0x4142 after a Java array object"),
error.getMessage());
+
+ // COUNT(*) answers from the metadata row counts and reads no file.
+ List<ConnectorScanRange> count = planBinaryPartitionScan(legacy,
true);
+ Assertions.assertEquals(1, count.size());
+ Assertions.assertEquals(2, ((PaimonScanRange)
count.get(0)).getPushDownRowCount());
+
+ // The NULL partition is named after partition.default-name, which
Paimon finds again.
+ Table nullOnly = binaryPartitionTable(catalog, "null_only",
Collections.emptyMap(), (byte[]) null);
+ Assertions.assertEquals(1, planBinaryPartitionScan(nullOnly,
false).size());
+ }
+ }
+
+ @Test
+ public void binaryPartitionFilesOfAnotherNamingFailScanPlanning(@TempDir
Path warehouse) throws Exception {
+ // Paimon writes a file under the name its partition has when written
and looks it up under the name the
+ // partition has when read, so a file written under a legacy name
stays lost after the option changes.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Table legacy = binaryPartitionTable(catalog, "renamed",
Collections.emptyMap(), new byte[] {0x41, 0x42});
+ String legacyFile = ((DataSplit)
legacy.newReadBuilder().newScan().plan().splits().get(0))
+ .dataFiles().get(0).fileName();
+ catalog.alterTable(Identifier.create("db", "renamed"),
Collections.singletonList(
+ SchemaChange.setOption("partition.legacy-name", "false")),
false);
+ writeBinaryPartitions(catalog.getTable(Identifier.create("db",
"renamed")), new byte[] {0x43, 0x44});
+ Table current = catalog.getTable(Identifier.create("db",
"renamed"));
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> planBinaryPartitionScan(current, false));
+ Assertions.assertTrue(error.getMessage().startsWith("Cannot read
Paimon table db.renamed: data file "
+ + legacyFile + " of partition key = 0x4142 was written
while the table used legacy partition"
+ + " names ('partition.legacy-name' = 'true')"),
error.getMessage());
+
+ // A file written under the bytes of its value is looked up under
a legacy name once those are back on.
+ binaryPartitionTable(catalog, "restored",
Collections.singletonMap("partition.legacy-name", "false"),
+ new byte[] {0x41, 0x42});
+ Table byBytes = catalog.getTable(Identifier.create("db",
"restored"));
+ Assertions.assertEquals(1, planBinaryPartitionScan(byBytes,
false).size());
+ catalog.alterTable(Identifier.create("db", "restored"),
Collections.singletonList(
+ SchemaChange.setOption("partition.legacy-name", "true")),
false);
+ Table restored = catalog.getTable(Identifier.create("db",
"restored"));
+ error = Assertions.assertThrows(DorisConnectorException.class,
+ () -> planBinaryPartitionScan(restored, false));
+ Assertions.assertTrue(error.getMessage().startsWith("Cannot read
Paimon table db.restored: the table"
+ + " uses legacy partition names"), error.getMessage());
+ }
+ }
+
+ @Test
+ public void
binaryPartitionFilesWrittenWithDynamicNamesStayReadable(@TempDir Path
warehouse) throws Exception {
+ // Dynamic options keep the schema ID, so a file written with a
dynamic partition.legacy-name = false refers
+ // to a schema with legacy names while it sits under the bytes of its
value, where a lookup by bytes finds it.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "dynamic_names");
+ Table legacy = binaryPartitionTable(catalog, "dynamic_names",
Collections.emptyMap());
+
writeBinaryPartitions(legacy.copy(Collections.singletonMap("partition.legacy-name",
"false")),
+ new byte[] {0x41, 0x42});
+ Table current = catalog.getTable(id);
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> planBinaryPartitionScan(current, false));
+ Assertions.assertTrue(error.getMessage().startsWith("Cannot read
Paimon table db.dynamic_names: the"
+ + " table uses legacy partition names"),
error.getMessage());
+
+ catalog.alterTable(id, Collections.singletonList(
+ SchemaChange.setOption("partition.legacy-name", "false")),
false);
+ List<ConnectorScanRange> ranges =
planBinaryPartitionScan(catalog.getTable(id), false);
+ Assertions.assertEquals(1, ranges.size());
+ String path =
ranges.get(0).getPath().orElseThrow(AssertionError::new);
+ Assertions.assertTrue(path.contains("/key=AB/"), path);
+ Assertions.assertTrue(LocalFileIO.create().exists(new
org.apache.paimon.fs.Path(path)), path);
+ }
+ }
+
+ @Test
+ public void
legacyNamedBinaryPartitionFilesAtExternalPathsStayReadable(@TempDir Path
warehouse) throws Exception {
+ // A file at an external path is read from the path its metadata
records, whatever its partition is named.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new
org.apache.paimon.fs.Path(warehouse.resolve("warehouse").toUri()))) {
+ catalog.createDatabase("db", false);
+ String externalRoot =
warehouse.resolve("external").toUri().toString();
+ Map<String, String> options = new HashMap<>();
+ options.put("data-file.external-paths", externalRoot);
+ options.put("data-file.external-paths.strategy", "round-robin");
+ Table legacy = binaryPartitionTable(catalog, "external_files",
options, new byte[] {0x41, 0x42});
+ List<ConnectorScanRange> ranges = planBinaryPartitionScan(legacy,
false);
+ Assertions.assertEquals(1, ranges.size());
+ String path =
ranges.get(0).getPath().orElseThrow(AssertionError::new);
+
Assertions.assertTrue(path.contains(warehouse.resolve("external").toString()),
path);
+ Assertions.assertTrue(LocalFileIO.create().exists(new
org.apache.paimon.fs.Path(path)), path);
+ }
+ }
+
+ @Test
+ public void
externalBinaryPartitionFileWithLocalDeletionVectorFailsScanPlanning(@TempDir
Path warehouse)
+ throws Exception {
+ // With index-file-in-data-file-dir, a deletion vector written while
no external data path is configured
+ // records only its name and is looked up under the directory of its
partition, which a legacy name never
+ // finds, while its data file keeps its external path. The BE skips
the whole data file when it cannot open
+ // the deletion vector, rows that were not deleted included.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new
org.apache.paimon.fs.Path(warehouse.resolve("warehouse").toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "local_dv");
+ deletionVectorBinaryPartitionTable(catalog, "local_dv", warehouse);
+ catalog.alterTable(id,
Arrays.asList(SchemaChange.removeOption("data-file.external-paths"),
+
SchemaChange.removeOption("data-file.external-paths.strategy")), false);
+ Table table = catalog.getTable(id);
+ DataFileMeta file = ((DataSplit)
table.newReadBuilder().newScan().plan().splits().get(0))
+ .dataFiles().get(0);
+ Assertions.assertTrue(file.externalPath().isPresent());
+ String deletionVector = deleteFirstRow(table);
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> planBinaryPartitionScan(table, false));
+ Assertions.assertTrue(error.getMessage().startsWith("Cannot read
Paimon table db.local_dv: deletion vector"
+ + " file " + deletionVector + " of data file " +
file.fileName() + " of partition key = 0x4142"
+ + " does not exist"), error.getMessage());
+ }
+ }
+
+ @Test
+ public void
externalBinaryPartitionDeletionVectorAtExternalPathStaysReadable(@TempDir Path
warehouse)
+ throws Exception {
+ // A deletion vector written while an external data path is configured
records its absolute path.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new
org.apache.paimon.fs.Path(warehouse.resolve("warehouse").toUri()))) {
+ catalog.createDatabase("db", false);
+ Table table = deletionVectorBinaryPartitionTable(catalog,
"external_dv", warehouse);
+ deleteFirstRow(table);
+ List<ConnectorScanRange> ranges = planBinaryPartitionScan(table,
false);
+ Assertions.assertEquals(1, ranges.size());
+ String deletionVector =
ranges.get(0).getProperties().get("paimon.deletion_file.path");
+
Assertions.assertTrue(deletionVector.contains(warehouse.resolve("external").toString()),
deletionVector);
+ Assertions.assertTrue(LocalFileIO.create().exists(new
org.apache.paimon.fs.Path(deletionVector)),
+ deletionVector);
+ }
+ }
+
+ @Test
+ public void fallbackBinaryPartitionsUseTheNamingOfTheirBranch(@TempDir
Path warehouse) throws Exception {
+ // A partition missing on the main branch is planned and read under
the options of the fallback branch,
+ // while the table and its $audit_log wrapper report the options of
the main branch.
+ for (boolean fallbackUsesBytes : new boolean[] {true, false}) {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new
org.apache.paimon.fs.Path(warehouse.resolve("fallback_bytes_" +
fallbackUsesBytes).toUri()))) {
+ catalog.createDatabase("db", false);
+ Map<String, String> byBytes =
Collections.singletonMap("partition.legacy-name", "false");
+ // The main branch only holds the NULL partition, which every
naming finds.
+ FileStoreTable main = (FileStoreTable)
binaryPartitionTable(catalog, "main",
+ fallbackUsesBytes ? Collections.emptyMap() : byBytes,
(byte[]) null);
+ FileStoreTable fallback = (FileStoreTable)
binaryPartitionTable(catalog, "fallback",
+ fallbackUsesBytes ? byBytes : Collections.emptyMap(),
new byte[] {0x41, 0x42});
+ FallbackReadFileStoreTable pair = new
FallbackReadFileStoreTable(main, fallback, true);
+ if (!fallbackUsesBytes) {
+ Map<String, Supplier<List<ConnectorScanRange>>> plans =
new LinkedHashMap<>();
+ plans.put("table", () -> planBinaryPartitionScan(pair,
false));
+ plans.put("$audit_log", () ->
planSystemTableScan("audit_log", pair));
+ plans.forEach((name, plan) -> {
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ plan::get, name);
+
Assertions.assertTrue(error.getMessage().startsWith("Cannot read Paimon table
db.main: the"
+ + " table uses legacy partition names
('partition.legacy-name' = 'true'), under which"
+ + " Paimon names partition key = 0x4142"),
error.getMessage());
+ });
+ continue;
+ }
+ List<ConnectorScanRange> ranges =
planBinaryPartitionScan(pair, false);
+ Assertions.assertEquals(2, ranges.size());
+ for (ConnectorScanRange range : ranges) {
+ String path =
range.getPath().orElseThrow(AssertionError::new);
+ Assertions.assertTrue(LocalFileIO.create().exists(new
org.apache.paimon.fs.Path(path)), path);
+ }
+ Assertions.assertTrue(ranges.stream().anyMatch(range ->
range.getPath().get().contains("/key=AB/")));
+
+ List<ConnectorScanRange> auditLog =
planSystemTableScan("audit_log", pair);
+ Assertions.assertEquals(2, auditLog.size());
+ List<String> rows = new ArrayList<>();
+ SystemTableLoader.load("audit_log",
pair).newReadBuilder().newRead()
+ .createReader(deserializeJniSplits(auditLog))
+ .forEachRemaining(row -> rows.add(row.getString(0) + "
" + (row.isNullAt(2) ? "NULL"
+ :
java.util.HexFormat.of().withUpperCase().formatHex(row.getBinary(2)))));
+ rows.sort(Comparator.naturalOrder());
+ Assertions.assertEquals(Arrays.asList("+I 4142", "+I NULL"),
rows);
+ }
+ }
+ }
+
+ @Test
+ public void
fileCreationTimeScanOfFallbackTableChecksBinaryPartitionsOfTheMainBranch(@TempDir
Path warehouse)
+ throws Exception {
+ // A scan.file-creation-time-millis read plans a fallback table
through its SnapshotReader, which reads the
+ // main branch and returns plain DataSplits rather than the
FallbackSplits of the fallback scan.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Map<String, String> byBytes =
Collections.singletonMap("partition.legacy-name", "false");
+ FileStoreTable main = (FileStoreTable)
binaryPartitionTable(catalog, "main", byBytes,
+ new byte[] {0x41, 0x42});
+ FileStoreTable fallback = (FileStoreTable)
binaryPartitionTable(catalog, "fallback", byBytes,
+ new byte[] {0x43, 0x44});
+ FallbackReadFileStoreTable pair = new
FallbackReadFileStoreTable(main, fallback, true);
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = pair;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle("db", "main",
Collections.emptyList(),
+
Collections.emptyList()).withScanOptions(PaimonScanParams.markAsOptions(
+ PaimonScanParams.resolveOptions(pair,
Collections.singletonMap(
+
CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), "0"))));
+ List<ConnectorScanRange> ranges = provider.planScan(
+ sessionWithProps(Collections.emptyMap(), new
TestStatementScope()),
+ ConnectorScanRequest.builder(handle, Arrays.asList(new
PaimonColumnHandle("id", 0),
+ new PaimonColumnHandle("key", 1))).build());
+ Assertions.assertEquals(1, ranges.size());
+ String path =
ranges.get(0).getPath().orElseThrow(AssertionError::new);
+ Assertions.assertTrue(path.contains("/key=AB/"), path);
+
+ DataSplit split = (DataSplit)
pair.newSnapshotReader().read().splits().get(0);
+ Assertions.assertFalse(split instanceof
FallbackReadFileStoreTable.FallbackSplit);
+ Assertions.assertSame(main,
PaimonScanPlanProvider.legacyOrcSplitTable(pair, split));
+ }
+ }
+
+ private static Table binaryPartitionTable(Catalog catalog, String name,
Map<String, String> options,
+ byte[]... keys) throws Exception {
+ Identifier id = Identifier.create("db", name);
+ Schema.Builder schema = Schema.newBuilder().column("id",
DataTypes.INT()).column("key", DataTypes.BYTES())
+ .partitionKeys("key").option("bucket", "-1");
+ options.forEach(schema::option);
+ catalog.createTable(id, schema.build(), false);
+ writeBinaryPartitions(catalog.getTable(id), keys);
+ return catalog.getTable(id);
+ }
+
+ private static void writeBinaryPartitions(Table table, byte[]... keys)
throws Exception {
+ BatchWriteBuilder writer = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writer.newWrite()) {
+ for (int i = 0; i < keys.length; i++) {
+ write.write(GenericRow.of(i, keys[i]));
+ }
+ try (BatchTableCommit commit = writer.newCommit()) {
+ commit.commit(write.prepareCommit());
+ }
+ }
+ }
+
+ private static List<ConnectorScanRange> planBinaryPartitionScan(Table
table, boolean countPushdown) {
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = table;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle("db", table.name(),
+ Collections.emptyList(), Collections.emptyList());
+ return provider.planScan(sessionWithProps(Collections.emptyMap(), new
TestStatementScope()),
+ ConnectorScanRequest.builder(handle, Arrays.asList(new
PaimonColumnHandle("id", 0),
+ new PaimonColumnHandle("key",
1))).countPushdown(countPushdown).build());
+ }
+
+ /** Plans a scan of {@code source$kind} the way the connector metadata
builds its system table handle. */
+ private static List<ConnectorScanRange> planSystemTableScan(String kind,
FileStoreTable source) {
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = source;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = PaimonTableHandle.forSystemTable("db",
source.name(), kind,
+ PaimonScanParams.requiresPaimonReader(kind));
+ handle.setPaimonTable(SystemTableLoader.load(kind, source));
+ handle.setSysBaseTable(source);
+ handle.setSystemTableSource(source);
+ return provider.planScan(sessionWithProps(Collections.emptyMap(), new
TestStatementScope()),
+ ConnectorScanRequest.builder(handle,
Collections.emptyList()).build());
+ }
+
+ private static Table deletionVectorBinaryPartitionTable(Catalog catalog,
String name, Path warehouse)
+ throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put("deletion-vectors.enabled", "true");
+ options.put("index-file-in-data-file-dir", "true");
+ options.put("data-file.external-paths",
warehouse.resolve("external").toUri().toString());
+ options.put("data-file.external-paths.strategy", "round-robin");
+ return binaryPartitionTable(catalog, name, options, new byte[] {0x41,
0x42}, new byte[] {0x41, 0x42});
+ }
+
+ /** Deletes the first row of the first data file the way a Spark DELETE
persists its deletion vectors. */
+ private static String deleteFirstRow(Table table) throws Exception {
+ DataSplit split = (DataSplit)
table.newReadBuilder().newScan().plan().splits().get(0);
+ // PaimonSparkWriter.persistDeletionVectors writes the index files
under the table's current options.
+ BaseAppendDeleteFileMaintainer maintainer =
BaseAppendDeleteFileMaintainer.forUnawareAppend(
+ ((FileStoreTable) table).store().newIndexFileHandler(),
table.latestSnapshot().get(),
+ split.partition());
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(0);
+
maintainer.notifyNewDeletionVector(split.dataFiles().get(0).fileName(),
deletionVector);
+ List<IndexFileMeta> indexFiles = new ArrayList<>();
+ maintainer.persist().forEach(entry ->
indexFiles.add(entry.indexFile()));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(new
CommitMessageImpl(maintainer.getPartition(),
+ maintainer.getBucket(), null,
DataIncrement.indexIncrement(indexFiles),
+ CompactIncrement.emptyIncrement())));
+ }
+ return indexFiles.get(0).fileName();
+ }
+
@Test
public void planScanPushesLimitIntoPaimonSplitPlanning(@TempDir Path
warehouse) throws Exception {
try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
