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 fc2037fcdb0 [fix](paimon) Keep chain groups in Paimon fallback
rebuilds and reject chain table scans clearly (#68897)
fc2037fcdb0 is described below
commit fc2037fcdb04063104c40e1700cb9c1e45a1ecc6
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Tue Oct 13 15:02:06 2026 +0800
[fix](paimon) Keep chain groups in Paimon fallback rebuilds and reject
chain table scans clearly (#68897)
### What problem does this PR solve?
Issue Number: None
Related PR: #68885
Problem Summary:
**In short.** Master cannot read a Paimon 1.4 chain table: every data
scan fails in planning with Paimon's internal `This is a bug, perhaps
there is a recursive call.` This PR keeps the chain group intact
wherever the Paimon connector and the BE JNI scanner rebuild a fallback
pair, and makes a data scan of a chain table fail with a clear "not
supported yet" message, because the rest of the scan path does not plan
chain partitions and chain splits yet. The review of #68885 (closed
since) found the BE half of the problem.
**Background.** Paimon 1.4 builds a chain table (`chain-table.enabled`
with `scan.fallback-snapshot-branch` and `scan.fallback-delta-branch`)
as `FallbackReadFileStoreTable(main, ChainGroupReadTable(snapshot,
delta), true)`. `ChainGroupReadTable` is itself a fallback pair. It
plans and reads the chain: a partition missing from the snapshot branch
is read as the latest earlier snapshot partition merged with the delta
branch. Paimon's `FallbackReadFileStoreTable` constructor rejects any
other pair nested inside a pair. The Paimon connector and the BE JNI
scanner rebuild fallback pairs branch by branch in six places:
- the manifest-parallelism bound on the BE (`PaimonJniScanner`, taken
when the FE cap is below the BE's core count) and on the FE
(`PaimonReaderOptions`);
- the catalog-loader strip (`dropCatalogLoader`) and the catalog
snapshot pin (`pinCatalogSnapshot`) of the table sent to the BE;
- the restore of a statement pin: pinned fallback snapshots
(`copyWithPinnedFallback`) and bound schemas (`restoreBoundSchema`).
**The problem, and what it cost.** Each of these recursed into the chain
group and rebuilt it either as a plain pair, which the outer constructor
rejects, or as a plain table of the snapshot branch, which silently
drops the delta branch. On a local FE + BE with a file-system Paimon
catalog and a chain table created and written through the Paimon API:
| Read | Paimon's own read | master |
|---|---|---|
| `SELECT * FROM chain_t` | 5 rows: 2 from the snapshot branch, 3
chained from the snapshot and delta branches | fails: `This is a bug,
perhaps there is a recursive call.` (first from
`PaimonScanParams.restoreBoundSchema`) |
| a `$files` scan when the FE cap is below the BE's core count | | the
BE manifest bound fails the same way (unit test; a single host does not
take that path) |
Keeping the chain group is not enough to read a chain table. Doris plans
neither the partitions a chain table reads from its branches nor the
chain splits. With only the rebuilds fixed, a chain table whose main
branch is empty reads as empty (0 partitions, 0 splits), and one with
main data fails with `Fail to parse columnsFromPath, expected: [dt],
filePath: connector://virtual`. branch-4.x has the same manifest-bound
rebuild and no chain support either.
**How this PR fixes it.**
- `PaimonTableDecorators.rebuildFallbackPair` rebuilds a
`ChainGroupReadTable` as a group and any other pair as before. The FE
manifest bound, `copyWithPinnedFallback` and `restoreBoundSchema` use
it, and `dropCatalogLoader` and `pinCatalogSnapshot` now recurse into
the nested pair. The BE manifest bound keeps the group the same way.
- `PaimonScanPlanProvider.resolveScanTable` rejects a data scan of a
chain table (a fallback pair over a `ChainGroupReadTable`) with `Reading
Paimon chain table db.tbl (chain-table.enabled) is not supported yet`.
System tables and a branch read on its own (`@branch(...)`) are plain
tables and are not affected.
**Results.** Same cluster and table:
| Read | before | after |
|---|---|---|
| `SELECT ... FROM chain_t` | `This is a bug, perhaps there is a
recursive call.` | `Reading Paimon chain table d.chain_t
(chain-table.enabled) is not supported yet` |
| ``chain_t$files`` | reads the main branch's files | same |
| `chain_t@branch(snapshot)`, `chain_t@branch(delta)` | read that
branch's rows | same |
| an ordinary Paimon table | reads | same |
System tables over a chain table now read through the chain group too.
Paimon resolves `$partitions` on the backend, and for a chain table
whose delta branch alone holds partition 20240102 it returns `[20231231,
20240101]` from master's backend payload and `[20231231, 20240101,
20240102]`, Paimon's own answer, from this PR's (Paimon 1.4.2 in jshell
on the same payloads, not on the cluster). `chain_t$ro` is unchanged: it
returns Paimon's own read-optimized rows, which leave out a partition
only the delta branch holds.
---
.../org/apache/doris/paimon/PaimonJniScanner.java | 6 ++
.../apache/doris/paimon/PaimonJniScannerTest.java | 67 +++++++++++++++++++++
.../connector/paimon/PaimonReaderOptions.java | 3 +-
.../doris/connector/paimon/PaimonScanParams.java | 11 ++--
.../connector/paimon/PaimonScanPlanProvider.java | 39 ++++++++-----
.../connector/paimon/PaimonTableDecorators.java | 28 +++++++++
.../paimon/PaimonBackendBoundTableTest.java | 58 ++++++++++++++++++
.../connector/paimon/PaimonReaderOptionsTest.java | 35 +++++++++++
.../paimon/PaimonStatementSchemaTest.java | 68 ++++++++++++++++++++++
9 files changed, 293 insertions(+), 22 deletions(-)
diff --git
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
index 6b349f1556d..c7b410c3102 100644
---
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
+++
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
@@ -32,6 +32,7 @@ import org.apache.paimon.options.MemorySize;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.FileRecordIterator;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.DelegatedFileStoreTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
@@ -778,6 +779,11 @@ public class PaimonJniScanner extends JniScanner {
}
// Each branch owns an independent planner setting; a smaller
sibling is not an
// execution ceiling and must never throttle the other branch.
+ if (pair instanceof ChainGroupReadTable) {
+ // A chain table nests this group over its snapshot and delta
branches. The group plans
+ // and reads the chain itself and is the only pair Paimon lets
one nest inside another.
+ return new ChainGroupReadTable(main, other);
+ }
return new FallbackReadFileStoreTable(main, other,
isWrappedFirst(pair));
}
diff --git
a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
index 4cdc78dea2e..b4cabd0f424 100644
---
a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
+++
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
@@ -28,13 +28,16 @@ import org.apache.paimon.disk.BufferFileWriter;
import org.apache.paimon.disk.FileIOChannel;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.privilege.PrivilegeChecker;
import org.apache.paimon.privilege.PrivilegedFileStoreTable;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.DelegatedFileStoreTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.TableRead;
@@ -315,6 +318,60 @@ public class PaimonJniScannerTest {
.options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
}
+ @Test
+ public void testBackendManifestCapKeepsTheChainGroupOfAChainTable(@TempDir
Path dir) {
+ // Paimon builds a chain table as a fallback pair whose other branch
is a ChainGroupReadTable, and the outer
+ // pair accepts no other nested fallback. Bounding the manifest
parallelism rebuilds every branch, so the
+ // chain group has to come back as one. MUTATION: rebuilding it as a
plain fallback pair -> red.
+ Table chain = new FallbackReadFileStoreTable(primaryKeyTable(dir,
"main"),
+ new ChainGroupReadTable(primaryKeyTable(dir, "snapshot"),
primaryKeyTable(dir, "delta")), true);
+
+ Table safe = PaimonJniScanner.applyBackendManifestParallelism(chain,
"2", 4);
+
+ FileStoreTable group = ((FallbackReadFileStoreTable) safe).other();
+ Assertions.assertTrue(group instanceof ChainGroupReadTable,
group.getClass().getName());
+ Assertions.assertEquals("2", ((ChainGroupReadTable)
group).wrapped().options()
+ .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
+ Assertions.assertEquals("2", ((ChainGroupReadTable)
group).other().options()
+ .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
+ }
+
+ @Test
+ public void
testBackendCapKeepsTheChainGroupOfAChainTableSystemSource(@TempDir Path dir)
throws Exception {
+ // A system table over a chain table ($partitions, $files, ...) is
rebuilt from the transported source, and
+ // the rebuilt wrapper must still read through the chain group: the
backend resolves $partitions, and
+ // without the delta branch a partition only it holds goes missing.
MUTATION: rebuilding the group as a
+ // plain fallback pair -> red.
+ FileStoreTable chain = new
FallbackReadFileStoreTable(primaryKeyTable(dir, "main"),
+ new ChainGroupReadTable(primaryKeyTable(dir, "snapshot"),
primaryKeyTable(dir, "delta")), true);
+
+ Table safe = PaimonJniScanner.applyBackendManifestParallelism(
+ SystemTableLoader.load("partitions", chain), "2", 4, chain,
"partitions");
+
+ Field storeTable = safe.getClass().getDeclaredField("storeTable");
+ storeTable.setAccessible(true);
+ FileStoreTable group = ((FallbackReadFileStoreTable)
storeTable.get(safe)).other();
+ Assertions.assertTrue(group instanceof ChainGroupReadTable,
group.getClass().getName());
+ Assertions.assertEquals("2", ((ChainGroupReadTable)
group).other().options()
+ .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
+ }
+
+ @Test
+ public void
testBackendCapStillReadsTheChainTablePayloadOfAnOlderFrontend(@TempDir Path
dir) {
+ // An older FE sends a chain table with its group replaced by a plain
table at the snapshot branch. A backend
+ // upgraded before the FE keeps reading that payload as it did.
MUTATION: rebuilding a pair as a chain group
+ // whatever it holds -> red.
+ FileStoreTable flattened = new
FallbackReadFileStoreTable(primaryKeyTable(dir, "main"),
+ primaryKeyTable(dir, "snapshot"), true);
+
+ Table safe =
PaimonJniScanner.applyBackendManifestParallelism(flattened, "2", 4);
+
+ Assertions.assertEquals(FallbackReadFileStoreTable.class,
safe.getClass());
+ FileStoreTable other = ((FallbackReadFileStoreTable) safe).other();
+ Assertions.assertFalse(other instanceof FallbackReadFileStoreTable,
other.getClass().getName());
+ Assertions.assertEquals("2",
other.options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
+ }
+
@Test
public void testAdvertisedFeCapStillChecksSerializedChildren() {
FileStoreTable main =
serializableFileStoreTable(Collections.emptyMap());
@@ -516,6 +573,16 @@ public class PaimonJniScannerTest {
.get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
}
+ /** A primary-key table object, as each branch of a chain table is; no
files are written. */
+ private static FileStoreTable primaryKeyTable(Path dir, String name) {
+ TableSchema schema = new TableSchema(0L,
+ Arrays.asList(new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "v", DataTypes.STRING())),
+ 1, Collections.emptyList(), Collections.singletonList("id"),
Collections.emptyMap(), "");
+ return FileStoreTableFactory.create(LocalFileIO.create(),
+ new
org.apache.paimon.fs.Path(dir.resolve(name).toUri().toString()), schema);
+ }
+
private static FileStoreTable serializableFileStoreTable(Map<String,
String> options) {
return (FileStoreTable)
Proxy.newProxyInstance(FileStoreTable.class.getClassLoader(),
new Class[] {FileStoreTable.class}, new
SerializableTableHandler(options));
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
index 94b05c8563f..de09d4e9132 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
@@ -223,8 +223,7 @@ public final class PaimonReaderOptions {
FileStoreTable other = normalizeManifestParallelism(
pair.other(), safeBound, materializeAbsent);
return main == pair.wrapped() && other == pair.other()
- ? table : new FallbackReadFileStoreTable(
- main, other, isWrappedFirst(pair));
+ ? table : PaimonTableDecorators.rebuildFallbackPair(pair,
main, other);
}
if (table instanceof DelegatedFileStoreTable) {
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
index 829ebd4cddc..5e4c59f33f7 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
@@ -224,10 +224,9 @@ public final class PaimonScanParams {
fallbackOptions.remove(CoreOptions.BUCKET.key());
fallbackOptions.put(CoreOptions.BRANCH.key(),
pair.other().coreOptions().branch());
fallbackOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
snapshotId);
- return new FallbackReadFileStoreTable(
+ return PaimonTableDecorators.rebuildFallbackPair(pair,
copyWithPinnedFallback(pair.wrapped(), dynamicOptions,
coordinates, path),
- copyWithPinnedFallback(pair.other(), fallbackOptions,
coordinates, fallbackPath),
- PaimonReaderOptions.isWrappedFirst(pair));
+ copyWithPinnedFallback(pair.other(), fallbackOptions,
coordinates, fallbackPath));
}
}
if (table instanceof DelegatedFileStoreTable && !(table instanceof
FallbackReadFileStoreTable)) {
@@ -246,9 +245,9 @@ public final class PaimonScanParams {
String fallbackPath = path + "fallback.";
long fallbackId = PaimonSchemaPin.fallbackSchemaId(options,
fallbackPath,
() ->
pair.other().schemaManager().latest().orElseThrow(IllegalStateException::new).id());
- return new
FallbackReadFileStoreTable(restoreBoundSchema(pair.wrapped(), schemaId,
options, path),
- restoreBoundSchema(pair.other(), fallbackId, options,
fallbackPath),
- PaimonReaderOptions.isWrappedFirst(pair));
+ return PaimonTableDecorators.rebuildFallbackPair(pair,
+ restoreBoundSchema(pair.wrapped(), schemaId, options,
path),
+ restoreBoundSchema(pair.other(), fallbackId, options,
fallbackPath));
}
if (table instanceof DelegatedFileStoreTable) {
FileStoreTable wrapped = ((DelegatedFileStoreTable)
table).wrapped();
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 3d912bbebd2..2a2157e1bdc 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
@@ -346,6 +346,12 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
*/
Table resolveScanTable(PaimonTableHandle paimonHandle) {
Table table = resolveTable(paimonHandle);
+ if (!paimonHandle.isSystemTable() &&
PaimonTableDecorators.isChainTable(table)) {
+ // A chain table reads a partition its main branch lacks from its
snapshot and delta branches.
+ // Doris plans neither those partitions nor the chain splits yet,
so it would read them as empty.
+ throw new DorisConnectorException("Reading Paimon chain table " +
paimonHandle.getDatabaseName() + "."
+ + paimonHandle.getTableName() + " (chain-table.enabled) is
not supported yet");
+ }
return withBoundSchemaAuthentication(paimonHandle, () ->
applyScanOptions(paimonHandle, table));
}
@@ -1504,18 +1510,21 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
*/
// Package-private for direct unit testing (PaimonBackendBoundTableTest).
static FileStoreTable pinCatalogSnapshot(FileStoreTable catalogLessTable,
FileStoreTable dataTable) {
- FileStoreTable source =
PaimonTableDecorators.unwrapToFallbackOrBase(dataTable);
- FileStoreTable target =
PaimonTableDecorators.unwrapToFallbackOrBase(catalogLessTable);
+ return
pinCatalogSnapshotOfBranches(PaimonTableDecorators.unwrapToFallbackOrBase(catalogLessTable),
+ PaimonTableDecorators.unwrapToFallbackOrBase(dataTable));
+ }
+
+ private static FileStoreTable pinCatalogSnapshotOfBranches(FileStoreTable
target, FileStoreTable source) {
if (source instanceof FallbackReadFileStoreTable) {
if (!(target instanceof FallbackReadFileStoreTable)) {
throw new IllegalStateException("Catalog-less Paimon table
lost its fallback branch");
}
FallbackReadFileStoreTable sourcePair =
(FallbackReadFileStoreTable) source;
FallbackReadFileStoreTable targetPair =
(FallbackReadFileStoreTable) target;
- return new FallbackReadFileStoreTable(
- pinCatalogSnapshotBranch(targetPair.wrapped(),
sourcePair.wrapped()),
- pinCatalogSnapshotBranch(targetPair.other(),
sourcePair.other()),
- PaimonReaderOptions.isWrappedFirst(targetPair));
+ // A chain table nests its snapshot and delta branches in a pair
of their own; pin those too.
+ return PaimonTableDecorators.rebuildFallbackPair(targetPair,
+ pinCatalogSnapshotOfBranches(targetPair.wrapped(),
sourcePair.wrapped()),
+ pinCatalogSnapshotOfBranches(targetPair.other(),
sourcePair.other()));
}
return pinCatalogSnapshotBranch(target, source);
}
@@ -1553,15 +1562,17 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
if (dataTable.catalogEnvironment().catalogLoader() == null) {
return dataTable;
}
- FileStoreTable undecorated =
PaimonTableDecorators.unwrapToFallbackOrBase(dataTable);
- if (undecorated instanceof FallbackReadFileStoreTable) {
- FallbackReadFileStoreTable fallbackReadTable =
(FallbackReadFileStoreTable) undecorated;
- return new FallbackReadFileStoreTable(
- rebuildWithoutCatalogLoader(fallbackReadTable.wrapped()),
- rebuildWithoutCatalogLoader(fallbackReadTable.other()),
- PaimonReaderOptions.isWrappedFirst(fallbackReadTable));
+ return
dropCatalogLoaderOfBranches(PaimonTableDecorators.unwrapToFallbackOrBase(dataTable));
+ }
+
+ private static FileStoreTable dropCatalogLoaderOfBranches(FileStoreTable
table) {
+ if (table instanceof FallbackReadFileStoreTable) {
+ // A chain table nests its snapshot and delta branches in a pair
of their own; rebuild those too.
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
+ return PaimonTableDecorators.rebuildFallbackPair(pair,
+ dropCatalogLoaderOfBranches(pair.wrapped()),
dropCatalogLoaderOfBranches(pair.other()));
}
- return rebuildWithoutCatalogLoader(undecorated);
+ return rebuildWithoutCatalogLoader(table);
}
private static FileStoreTable rebuildWithoutCatalogLoader(FileStoreTable
branch) {
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
index b762ff35e4f..fd82806e085 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
@@ -20,9 +20,11 @@ package org.apache.doris.connector.paimon;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.privilege.PrivilegeChecker;
import org.apache.paimon.privilege.PrivilegedFileStoreTable;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.DelegatedFileStoreTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
import java.lang.reflect.Field;
@@ -54,6 +56,32 @@ final class PaimonTableDecorators {
}
}
+ /**
+ * Rebuild a fallback-branch pair around new branches. A chain table is a
pair whose other branch is a
+ * {@link ChainGroupReadTable} over its snapshot and delta branches: the
group plans and reads the chain
+ * itself, and it is the only pair Paimon lets one nest inside another, so
it is rebuilt as a group.
+ */
+ static FileStoreTable rebuildFallbackPair(FallbackReadFileStoreTable pair,
FileStoreTable wrapped,
+ FileStoreTable other) {
+ if (pair instanceof ChainGroupReadTable) {
+ return new ChainGroupReadTable(wrapped, other);
+ }
+ return new FallbackReadFileStoreTable(wrapped, other,
PaimonReaderOptions.isWrappedFirst(pair));
+ }
+
+ /**
+ * Whether {@code table} reads as a chain table: a fallback pair whose
other branch is a
+ * {@link ChainGroupReadTable}. A branch of a chain table read on its own
is a plain table.
+ */
+ static boolean isChainTable(Table table) {
+ if (!(table instanceof FileStoreTable)) {
+ return false;
+ }
+ FileStoreTable base = unwrapToFallbackOrBase((FileStoreTable) table);
+ return base instanceof FallbackReadFileStoreTable
+ && ((FallbackReadFileStoreTable) base).other() instanceof
ChainGroupReadTable;
+ }
+
/**
* Peel the decorators Paimon may have stacked on top of the table, down
to the fallback-branch
* pair - the one layer that must stay on top, because that is what
dispatches a read to the
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
index 03e4703f117..bca0c1544e0 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
@@ -17,6 +17,8 @@
package org.apache.doris.connector.paimon;
+import org.apache.doris.connector.spi.DorisConnectorException;
+
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.FileSystemCatalog;
@@ -32,6 +34,7 @@ import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
@@ -144,6 +147,51 @@ public class PaimonBackendBoundTableTest {
sqlNull.getStaticPartition().get("pt"));
}
+ @Test
+ public void chainTableKeepsItsChainGroupWithoutTheCatalogLoader(@TempDir
Path warehouse) {
+ // Paimon builds a chain table as a fallback pair whose other branch
is a ChainGroupReadTable over the
+ // snapshot and the delta branch. Every branch is rebuilt without the
loader, so the chain group has to
+ // come back as one, with both of its branches. MUTATION: rebuilding
the chain group as one plain table
+ // -> red.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ CatalogEnvironment env = catalogEnvironment(catalog);
+ FileStoreTable snapshot = primaryKeyTable(warehouse, "snapshot", env);
+ FileStoreTable delta = primaryKeyTable(warehouse, "delta", env);
+ FileStoreTable chain = new
FallbackReadFileStoreTable(primaryKeyTable(warehouse, "main", env),
+ new ChainGroupReadTable(snapshot, delta), true);
+
+ FileStoreTable forBackend =
PaimonScanPlanProvider.dropCatalogLoader(chain);
+
+ FileStoreTable group = ((FallbackReadFileStoreTable)
forBackend).other();
+ Assertions.assertTrue(group instanceof ChainGroupReadTable,
group.getClass().getName());
+ ChainGroupReadTable chainGroup = (ChainGroupReadTable) group;
+ Assertions.assertEquals(snapshot.location(),
chainGroup.wrapped().location());
+ Assertions.assertEquals(delta.location(),
chainGroup.other().location());
+
Assertions.assertNull(chainGroup.wrapped().catalogEnvironment().catalogLoader());
+
Assertions.assertNull(chainGroup.other().catalogEnvironment().catalogLoader());
+ }
+
+ @Test
+ public void scanOfAChainTableFailsClearly(@TempDir Path warehouse) {
+ // Doris plans neither the partitions a chain table reads from its
snapshot and delta branches nor the chain
+ // splits, so a scan would read them as empty; it fails instead. A
branch read on its own is a plain table
+ // and still reads. MUTATION: dropping the check -> red.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ CatalogEnvironment env = catalogEnvironment(catalog);
+ FileStoreTable snapshot = primaryKeyTable(warehouse, "snapshot", env);
+ FileStoreTable chain = new
FallbackReadFileStoreTable(primaryKeyTable(warehouse, "main", env),
+ new ChainGroupReadTable(snapshot, primaryKeyTable(warehouse,
"delta", env)), true);
+ PaimonTableHandle handle = dataHandle();
+ handle.setPaimonTable(chain);
+ PaimonTableHandle branch = dataHandle();
+ branch.setPaimonTable(snapshot);
+
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider().resolveScanTable(handle));
+ Assertions.assertTrue(error.getMessage().contains("Paimon chain table
db.tbl"), error.getMessage());
+ Assertions.assertDoesNotThrow(() ->
provider().resolveScanTable(branch));
+ }
+
@Test
public void dataTableReachesTheBackendWithoutItsCatalogLoader(@TempDir
Path warehouse) {
VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
@@ -801,6 +849,16 @@ public class PaimonBackendBoundTableTest {
new org.apache.paimon.fs.Path("file://" + warehouse +
"/db.db/tbl"), schema, env);
}
+ /** A primary-key table object, as each branch of a chain table is; no
files are written. */
+ private static FileStoreTable primaryKeyTable(Path warehouse, String name,
CatalogEnvironment env) {
+ TableSchema schema = new TableSchema(0L,
+ Arrays.asList(new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "v", DataTypes.STRING())),
+ 1, Collections.emptyList(), Collections.singletonList("id"),
Collections.emptyMap(), "");
+ return FileStoreTableFactory.create(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path("file://" + warehouse +
"/db.db/" + name), schema, env);
+ }
+
/** A real, on-disk table so its {@code SnapshotManager} answers from real
snapshot files. */
private static FileStoreTable newRealTable(Path warehouse, String name)
throws Exception {
org.apache.paimon.fs.Path tablePath =
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonReaderOptionsTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonReaderOptionsTest.java
index 8ad3796dc67..ac11a315f98 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonReaderOptionsTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonReaderOptionsTest.java
@@ -28,8 +28,10 @@ import org.apache.paimon.privilege.PrivilegedFileStoreTable;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.PrimaryKeyFileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.IntType;
@@ -265,6 +267,26 @@ public class PaimonReaderOptionsTest {
pair.other().options().get("scan.manifest.parallelism"));
}
+ @Test
+ void testRuntimeSafeTableKeepsTheChainGroupOfAChainTable() {
+ // Paimon builds a chain table as a fallback pair whose other branch
is a ChainGroupReadTable, and the outer
+ // pair accepts no other nested fallback. Capping a branch rebuilds
the pairs above it, so the chain group
+ // has to come back as one. MUTATION: rebuilding it as a plain
fallback pair -> red.
+ FileStoreTable snapshot = newPrimaryKeyTable("chain_snapshot",
+ ImmutableMap.of(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
"8"));
+ FileStoreTable delta = newPrimaryKeyTable("chain_delta",
Collections.emptyMap());
+ Table chain = new
FallbackReadFileStoreTable(newPrimaryKeyTable("chain_main",
Collections.emptyMap()),
+ new ChainGroupReadTable(snapshot, delta), true);
+
+ Table safe = PaimonReaderOptions.runtimeSafeTable(chain, 4);
+
+ FileStoreTable group = ((FallbackReadFileStoreTable) safe).other();
+ Assertions.assertInstanceOf(ChainGroupReadTable.class, group);
+ Assertions.assertEquals("4", ((ChainGroupReadTable)
group).wrapped().options()
+ .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
+ Assertions.assertSame(delta, ((ChainGroupReadTable) group).other());
+ }
+
@Test
void testBackendManifestCapUsesExecutionCeilingNotSmallestBranch() {
FileStoreTable main = newFileStoreTable("main_cap", ImmutableMap.of(
@@ -311,6 +333,19 @@ public class PaimonReaderOptionsTest {
LocalFileIO.create(), new Path("memory://" + name), schema,
CatalogEnvironment.empty());
}
+ private FileStoreTable newPrimaryKeyTable(String name, Map<String, String>
options) {
+ TableSchema schema = new TableSchema(
+ 0,
+ Collections.singletonList(new DataField(0, "id", new
IntType(false))),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ null);
+ return new PrimaryKeyFileStoreTable(
+ LocalFileIO.create(), new Path("memory://" + name), schema,
CatalogEnvironment.empty());
+ }
+
private FakePaimonTable fakeTable(String name, Map<String, String>
options) {
FakePaimonTable table = new FakePaimonTable(
name, org.apache.paimon.types.RowType.builder().field("id",
new IntType()).build(),
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java
index a488e37cd02..25bc3525853 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java
@@ -30,6 +30,7 @@ import org.apache.doris.thrift.TFileScanRangeParams;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.FileSystemCatalog;
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.fs.FileIO;
@@ -40,6 +41,7 @@ import org.apache.paimon.privilege.PrivilegedFileStoreTable;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.DelegatedFileStoreTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
@@ -722,6 +724,72 @@ public class PaimonStatementSchemaTest {
}
}
+ @Test
+ public void statementPinKeepsTheChainOfAChainTable(@TempDir Path
warehouse) throws Exception {
+ // A chain table reads a partition its snapshot branch lacks as the
latest earlier snapshot partition
+ // merged with the delta branch, through a ChainGroupReadTable nested
in its fallback pair. Restoring the
+ // pinned fallback snapshots and schemas rebuilds every pair, so the
group has to stay a group and still
+ // read the chain. MUTATION: rebuilding the chain group as a plain
fallback pair -> red.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "t");
+ catalog.createTable(id, Schema.newBuilder().column("id",
DataTypes.INT().notNull())
+ .column("value", DataTypes.STRING()).column("seq",
DataTypes.BIGINT())
+ .column("dt",
DataTypes.STRING().notNull()).primaryKey("dt", "id").partitionKeys("dt")
+ .option("bucket", "1").option("sequence.field",
"seq").option("file.format", "parquet")
+ .option("partition.timestamp-pattern",
"$dt").option("partition.timestamp-formatter", "yyyyMMdd")
+ .build(), false);
+ FileStoreTable main = (FileStoreTable) catalog.getTable(id);
+ main.createBranch("snapshot");
+ main.createBranch("delta");
+ // A pin maps each branch to its latest snapshot no later than the
one above it (main, then the snapshot
+ // branch), as the SDK does; commit in that order so every row is
visible at the pinned main snapshot.
+ append(main.switchToBranch("delta"), chainRow(2, "b2", 2L,
"20240102"));
+ append(main.switchToBranch("delta"), chainRow(3, "c", 2L,
"20240102"));
+ append(main.switchToBranch("snapshot"), chainRow(1, "a", 1L,
"20240101"));
+ append(main.switchToBranch("snapshot"), chainRow(2, "b", 1L,
"20240101"));
+ append(main, chainRow(9, "z", 1L, "20231231"));
+ List<SchemaChange> chainOptions =
Arrays.asList(SchemaChange.setOption("chain-table.enabled", "true"),
+ SchemaChange.setOption("scan.fallback-snapshot-branch",
"snapshot"),
+ SchemaChange.setOption("scan.fallback-delta-branch",
"delta"));
+ catalog.alterTable(id, chainOptions, false);
+ catalog.alterTable(new Identifier("db", "t", "snapshot"),
chainOptions, false);
+ catalog.alterTable(new Identifier("db", "t", "delta"),
chainOptions, false);
+ FileStoreTable table = (FileStoreTable) catalog.getTable(id);
+ Assertions.assertInstanceOf(ChainGroupReadTable.class,
((FallbackReadFileStoreTable) table).other());
+ long snapshotId = table.snapshotManager().latestSnapshotId();
+ Map<String, String> pinned = new
HashMap<>(PaimonScanParams.withBoundSchema(
+
PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), snapshotId),
table.schema().id()));
+ pinned.putAll(PaimonSchemaPin.capture(table, table.schema().id(),
snapshotId));
+
+ FileStoreTable restored =
PaimonScanParams.applyOptionsWithoutTimeTravel(table, pinned);
+
+ Assertions.assertInstanceOf(ChainGroupReadTable.class,
((FallbackReadFileStoreTable) restored).other());
+ List<String> expected = Arrays.asList("20231231 9 z", "20240101 1
a", "20240101 2 b",
+ "20240102 1 a", "20240102 2 b2", "20240102 3 c");
+ Assertions.assertEquals(expected, readChainRows(restored));
+ Assertions.assertEquals(expected,
readChainRows(table.copy(Collections.singletonMap(
+ "scan.snapshot-id", String.valueOf(snapshotId)))));
+ }
+ }
+
+ private static GenericRow chainRow(int id, String value, long seq, String
dt) {
+ return GenericRow.of(id, BinaryString.fromString(value), seq,
BinaryString.fromString(dt));
+ }
+
+ private static List<String> readChainRows(Table table) throws Exception {
+ List<String> rows = new ArrayList<>();
+ for (Split split : table.newReadBuilder().newScan().plan().splits()) {
+ try (RecordReader<InternalRow> reader =
table.newReadBuilder().newRead().createReader(split)) {
+ reader.forEachRemaining(row -> rows.add(row.getString(3) + " "
+ row.getInt(0) + " "
+ + row.getString(1)));
+ }
+ }
+ Collections.sort(rows);
+ return rows;
+ }
+
private void checkSchemaMutation(Path warehouse, boolean options, boolean
system) throws Exception {
try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
new org.apache.paimon.fs.Path(warehouse.toUri()))) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
