This is an automated email from the ASF dual-hosted git repository.
CalvinKirs 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 687500b3358 [fix](paimon) Read Paimon tables on BE without
reconstructing the catalog metastore (#65867)
687500b3358 is described below
commit 687500b3358a3bcc01ca095e1371a8d1f9a51fd4
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Jul 31 09:48:55 2026 +0800
[fix](paimon) Read Paimon tables on BE without reconstructing the catalog
metastore (#65867)
### What problem does this PR solve?
Reading a Paimon table over a metastore-backed or REST catalog (HMS /
DLF) fails on
BE whenever the JNI reader is used — most visibly system tables (e.g.
`$snapshots`,
`$files`), which always go through JNI, and also branch / time-travel /
incremental
reads. Two failures were observed, both caused by the BE rebuilding a
catalog it does
not need:
```
# HMS catalog
[JNI_ERROR] NoClassDefFoundError: org/apache/hadoop/hive/conf/HiveConf
# DLF (REST) catalog
[JNI_ERROR] ClassNotFoundException:
com.aliyun.datalake.metastore.hive2.ProxyMetaStoreClient
```
**Root cause.** A table loaded from a metastore-backed Paimon catalog
carries a Paimon
`CatalogLoader` (e.g. `HiveCatalogLoader`) in its `CatalogEnvironment`.
When FE
serializes that table to BE, `SnapshotManager#latestSnapshotId` resolves
the latest
snapshot through the catalog's `SnapshotLoader`, which on BE
reconstructs the catalog's
metastore client and its whole Hive / DLF-REST stack — even though the
BE only reads
(via FE-resolved splits and the object store) and the snapshots already
live there.
This was previously masked by `java-udf`'s ~122MB `hive-catalog-shade`
on BE's shared
classpath; #65733 replaced it with the slim `hive-udf-shade`, exposing
the dependency.
**Fix 1 — serialize a catalog-less table to the BE (`PaimonScanNode`).**
- data table: rebuild via `FileStoreTableFactory` with an empty
`CatalogEnvironment`.
A `FileStoreTable` is fully defined by fileIO / location / schema, and
its dynamic
options (time travel, incremental) are already merged into the schema by
`copy(...)`,
so nothing is lost except the catalog loader.
- system table: rebuild it over such a catalog-less data table via
`SystemTableLoader`.
With no catalog loader, `SnapshotManager#latestSnapshotId` lists the
snapshot directory
on the filesystem instead of calling the metastore, so the BE never
reconstructs the
catalog and no longer needs any Hive / metastore classes.
**Fix 2 — co-locate the Paimon FileIO plugins with `paimon-scanner`
(packaging).**
Reading the snapshot from the filesystem makes the BE materialize the
object-store
`FileIO` lazily via `FileIO.get()` →
`ServiceLoader.load(FileIOLoader.class, ...)`,
which eagerly instantiates every registered provider. The OSS/S3 plugins
(`paimon-s3` → `S3Loader`, `paimon-jindo` → `JindoLoader`) were bundled
in
`preload-extensions`, on BE's JVM system (app) classpath, but the
`org.apache.paimon.fs.FileIOLoader` interface they implement ships in
`paimon-common`,
bundled only in paimon-scanner's own `JniScannerClassLoader`. That
loader is
parent-first, so the app classloader defines `S3Loader` / `JindoLoader`
and cannot
resolve the child-only `FileIOLoader` → `NoClassDefFoundError:
FileIOLoader`, which
aborts discovery. Moving the two plugins into `paimon-scanner` puts the
whole FileIO
SPI (interface + all providers) in one classloader; the Jindo SDK stays
on the app
classpath (`start_be.sh` adds `jindofs` to `DORIS_CLASSPATH`) and is
still reachable
via parent delegation.
Regression tests to re-run:
`io.trino.tests.product.paimon.TestPaimonSparkCompatibility`
(system-table reads) and
`external_table_p2/paimon/test_paimon_dlf_rest_catalog`.
### Release note
Fix Paimon reads (system tables, branch / time-travel / incremental)
failing on BE over
metastore-backed or REST catalogs (HMS / DLF) with
`NoClassDefFoundError`
(`HiveConf` / `FileIOLoader`) or `ClassNotFoundException:
ProxyMetaStoreClient`.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- Behavior changed:
- [x] No.
- [ ] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
build.sh | 31 +
fe/be-java-extensions/paimon-scanner/pom.xml | 80 +--
fe/be-java-extensions/preload-extensions/pom.xml | 23 +-
.../connector/paimon/PaimonConnectorMetadata.java | 68 +-
.../paimon/PaimonIncrementalScanParams.java | 88 +++
.../doris/connector/paimon/PaimonScanParams.java | 27 +
.../connector/paimon/PaimonScanPlanProvider.java | 253 ++++++-
.../connector/paimon/PaimonTableDecorators.java | 63 ++
.../doris/connector/paimon/PaimonTableHandle.java | 37 +
.../paimon/PaimonBackendBoundTableTest.java | 762 +++++++++++++++++++++
10 files changed, 1342 insertions(+), 90 deletions(-)
diff --git a/build.sh b/build.sh
index d5e2f65178c..36f72ef5622 100755
--- a/build.sh
+++ b/build.sh
@@ -1260,6 +1260,37 @@ EOF
fi
done
+ # Guard the Paimon FileIO SPI classloader boundary on the built artifacts.
The FileIOLoader
+ # interface (paimon-common) and every provider implementing it must sit in
the same jar:
+ # JniScannerClassLoader delegates parent-first, so a provider left on the
shared
+ # preload-extensions (JVM app) classpath cannot resolve the child-only
interface and
+ # ServiceLoader discovery aborts at runtime with "NoClassDefFoundError:
FileIOLoader". Unit
+ # tests all run in one classloader and cannot reproduce that, so assert it
on the packages.
+
PAIMON_SCANNER_JAR="${BE_JAVA_EXTENSIONS_DIR}/paimon-scanner/paimon-scanner-jar-with-dependencies.jar"
+
PRELOAD_JAR="${BE_JAVA_EXTENSIONS_DIR}/preload-extensions/preload-extensions-jar-with-dependencies.jar"
+ if [[ -f "${PAIMON_SCANNER_JAR}" ]]; then
+ # Tolerate a missing entry (unzip exits 11) so the message below is
what the build prints.
+ FILE_IO_SERVICES="$(unzip -p "${PAIMON_SCANNER_JAR}" \
+ META-INF/services/org.apache.paimon.fs.FileIOLoader 2>/dev/null ||
true)"
+ for loader in org.apache.paimon.s3.S3Loader
org.apache.paimon.jindo.JindoLoader; do
+ if ! echo "${FILE_IO_SERVICES}" | grep -q -x "${loader}"; then
+ echo "ERROR: ${loader} is missing from
META-INF/services/org.apache.paimon.fs.FileIOLoader"
+ echo " in ${PAIMON_SCANNER_JAR}. Paimon object-store
reads on BE would fail;"
+ echo " keep the FileIO plugins bundled in
paimon-scanner."
+ exit 1
+ fi
+ done
+ fi
+ # No "grep -q" here: it exits on the first match and SIGPIPEs unzip
halfway through this jar's
+ # 120k-entry listing, which under "set -o pipefail" makes the pipeline 141
and silently skips
+ # the error below - exactly when a provider did leak in. Let grep consume
the whole listing.
+ if [[ -f "${PRELOAD_JAR}" ]] &&
+ unzip -l "${PRELOAD_JAR}" | grep -E 'org/apache/paimon/(s3|jindo)/'
>/dev/null; then
+ echo "ERROR: ${PRELOAD_JAR} bundles a Paimon FileIOLoader provider. It
would be defined by"
+ echo " the JVM app classloader, which carries no FileIOLoader
interface."
+ exit 1
+ fi
+
# Third-party filesystem jars (JuiceFS, JindoFS) are packaged by
post-build.sh
bash "${DORIS_HOME}/post-build.sh" --be --output "${DORIS_OUTPUT}"
diff --git a/fe/be-java-extensions/paimon-scanner/pom.xml
b/fe/be-java-extensions/paimon-scanner/pom.xml
index 33b46aa0704..fa7c27e4e98 100644
--- a/fe/be-java-extensions/paimon-scanner/pom.xml
+++ b/fe/be-java-extensions/paimon-scanner/pom.xml
@@ -56,74 +56,34 @@ under the License.
<artifactId>paimon-hive-connector-3.1</artifactId>
</dependency>
- <!--
- Hive runtime classes required by the Paimon Hive connector on the BE
side.
- paimon-hive-connector-3.1 marks hive-exec as `provided`, and only
bundles
- relocated hive classes under org.apache.paimon.shade.*; the real
- org.apache.hadoop.hive.conf.HiveConf (referenced by
- org.apache.paimon.hive.SerializableHiveConf when a HMS-backed Paimon
table
- is deserialized, and by HiveCatalog.createHiveConf) is NOT packaged.
- It used to be provided accidentally by java-udf's hive-catalog-shade
on the
- shared BE classpath, but #65733 replaced that with the slim
hive-udf-shade,
- so paimon-scanner must now carry HiveConf itself. Version is pinned
to
- ${hive.version} (3.1.x) to match the Hive 3.1 the connector was
built against.
- The transitive trees (jetty/orc/ant/log4j bridges/hadoop/...) are
pruned:
- HiveConf's construction closure lives entirely inside hive-common,
shims/Utils
- (used only by the lazy HiveConf#getUser) lives in hive-shims-common,
and hadoop
- is already on BE's shared classpath.
- -->
<dependency>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-common</artifactId>
- <version>${hive.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.hive.shims</groupId>
- <artifactId>hive-shims-common</artifactId>
- <version>${hive.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-format</artifactId>
</dependency>
+
<!--
- For the same reason (hive-catalog-shade no longer on the shared
classpath),
- paimon-scanner must also carry the Hive metastore API classes. When
BE
- deserializes an HMS-backed Paimon table,
org.apache.paimon.hive.HiveCatalogLoader
- links org.apache.paimon.hive.HiveCatalog, which references a broad
slice of
- org.apache.hadoop.hive.metastore.api.* (Table, Database, Partition,
- StorageDescriptor, SerDeInfo, FieldSchema, NoSuchObjectException,
- UnknownTableException, ...). In Hive 3.x that whole thrift-generated
`api`
- package lives in hive-standalone-metastore (the `hive-metastore`
artifact is a
- thin shim and does NOT contain it). Transitives are pruned like the
two jars
- above: the api classes only extend org.apache.thrift.* (provided by
java-common's
- libthrift on the shared classpath, exactly as under the old
hive-catalog-shade)
- and java stdlib; datanucleus/orc/hadoop/etc. are only needed by the
metastore
- server/client, which BE never runs (it reads the already-resolved
table).
+ Paimon OSS/S3 FileIO plugins. These register
org.apache.paimon.fs.FileIOLoader
+ service providers (paimon-s3 -> S3Loader, paimon-jindo ->
JindoLoader) used to read
+ object-store-backed Paimon tables (e.g. DLF REST catalogs over OSS).
They MUST be
+ bundled here, alongside the FileIOLoader interface that lives in
paimon-common: when a
+ BE-side read resolves the latest snapshot from the filesystem,
RESTTokenFileIO.fileIO()
+ calls FileIO.get() -> ServiceLoader.load(FileIOLoader.class,
FileIOLoader's classloader),
+ which eagerly instantiates every registered provider. A provider and
the FileIOLoader
+ interface it implements must be reachable from the same classloader.
These plugins used
+ to sit on the shared preload-extensions (JVM app) classpath, which
does NOT carry
+ paimon-common's FileIOLoader; since JniScannerClassLoader is
parent-first, the app-loaded
+ S3Loader/JindoLoader could not resolve the child-only interface ->
NoClassDefFoundError:
+ FileIOLoader. Co-locating them here keeps the whole FileIO SPI in
one classloader; the
+ assembly's metaInf-services handler merges their service files with
paimon-common's
+ (local/hadoop) into one complete provider list.
-->
<dependency>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-standalone-metastore</artifactId>
- <version>${hive.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-s3</artifactId>
</dependency>
-
<dependency>
<groupId>org.apache.paimon</groupId>
- <artifactId>paimon-format</artifactId>
+ <artifactId>paimon-jindo</artifactId>
</dependency>
</dependencies>
diff --git a/fe/be-java-extensions/preload-extensions/pom.xml
b/fe/be-java-extensions/preload-extensions/pom.xml
index b359ce15944..0b32dbed8e2 100644
--- a/fe/be-java-extensions/preload-extensions/pom.xml
+++ b/fe/be-java-extensions/preload-extensions/pom.xml
@@ -112,16 +112,19 @@ under the License.
</exclusion>
</exclusions>
</dependency>
- <!-- For BE Paimon OSS/S3 Access -->
- <dependency>
- <groupId>org.apache.paimon</groupId>
- <artifactId>paimon-s3</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.paimon</groupId>
- <artifactId>paimon-jindo</artifactId>
- </dependency>
+ <!--
+ NOTE: the paimon OSS/S3 FileIO plugins (paimon-s3 / paimon-jindo)
are intentionally
+ NOT here. They register org.apache.paimon.fs.FileIOLoader service
providers
+ (S3Loader / JindoLoader), but the FileIOLoader interface itself
lives in paimon-common,
+ which is bundled in paimon-scanner, NOT on this shared/preload
classpath. Because each
+ JniScannerClassLoader delegates parent-first to the JVM app
classloader (where this
+ preload jar sits), a provider defined here could not resolve the
child-only FileIOLoader
+ interface, so ServiceLoader discovery in paimon-scanner's
FileIO.get() failed with
+ NoClassDefFoundError: FileIOLoader. The plugins are therefore
co-located with the
+ FileIOLoader interface in paimon-scanner (its only BE consumer) so
the whole FileIO SPI
+ lives in one classloader. Do not move them back here without also
putting paimon-common
+ on this classpath.
+ -->
<!-- For Avro and Hudi Scanner PreLoad -->
<dependency>
<groupId>org.apache.hadoop</groupId>
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
index e785391776a..386c0aac2fc 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
@@ -47,6 +47,7 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.types.DataField;
@@ -457,25 +458,55 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
// (identical equals/hashCode/toString and the same sys Identifier).
The support check above
// stays case-insensitive; only the canonical stored name is
lowercased.
String sys = sysName.toLowerCase(java.util.Locale.ROOT);
- Identifier sysId = new Identifier(
- base.getDatabaseName(), base.getTableName(), "main", sys);
- // M-11: wrap the remote getTable in executeAuthenticated (D-052).
TableNotExistException is
- // caught INSIDE the lambda (Kerberos UGI.doAs would wrap it
otherwise) and signalled out as a
- // null Table so this method can still short-circuit to
Optional.empty().
- Table sysTable;
- try {
- sysTable = context.executeAuthenticated(() -> {
- try {
- return catalogOps.getTable(sysId);
- } catch (Catalog.TableNotExistException e) {
- return null;
- }
- });
- } catch (Exception e) {
- throw new RuntimeException("Failed to load Paimon system table: "
+ sysId, e);
- }
+ // Build the wrapper over the base Table this handle ALREADY carries,
exactly the way the
+ // Paimon catalog builds it ({@code CatalogUtils#createSystemTable}),
and keep that base on
+ // the sys handle. Loading the wrapper from the catalog instead would
give it its own schema
+ // generation, independent of any base a consumer resolves later: the
FE would then plan on
+ // one generation while PaimonScanPlanProvider serializes to the BE a
system table rebuilt
+ // over the other one, and a system table whose row type follows the
base schema
+ // ($audit_log, $binlog, $ro) could reach the BE without the columns
the FE asked for.
+ //
+ // Strictly the ALREADY-RESOLVED reference, never a reload:
getTableHandle stashes it one
+ // call earlier
(PluginDrivenSysExternalTable#resolveConnectorTableHandle resolves the base
+ // handle immediately before this), so the normal path needs no
round-trip. Resolving it
+ // here instead would enter the authenticator a second time and would
turn a missing base
+ // table into a thrown RuntimeException rather than this method's
Optional.empty() contract.
+ //
+ // "Exactly the way the catalog builds it" includes building it over
the UNDECORATED base:
+ // CatalogUtils#loadTable hands createSystemTable the raw
FileStoreTableFactory#create
+ // result, and the decorator a catalog may add afterwards never
reaches a system table,
+ // because it only wraps a FileStoreTable and no system table is one.
That matters for $ro:
+ // ReadOptimizedTable#newScan builds its two-branch FallbackReadScan
only when its immediate
+ // wrapped object is a FallbackReadFileStoreTable, so with a
PrivilegedFileStoreTable in
+ // between it would plan the main branch alone through the pair's
inherited
+ // newSnapshotReader() and silently drop every fallback-only partition.
+ Table baseTable = base.getPaimonTable();
+ FileStoreTable sysBase = baseTable instanceof FileStoreTable
+ ?
PaimonTableDecorators.unwrapToFallbackOrBase((FileStoreTable) baseTable)
+ : null;
+ Table sysTable = sysBase == null ? null : SystemTableLoader.load(sys,
sysBase);
if (sysTable == null) {
- return Optional.empty();
+ // Not a file store table (format / object table): let the catalog
decide.
+ // M-11: wrap the remote getTable in executeAuthenticated (D-052).
TableNotExistException is
+ // caught INSIDE the lambda (Kerberos UGI.doAs would wrap it
otherwise) and signalled out as a
+ // null Table so this method can still short-circuit to
Optional.empty().
+ Identifier sysId = new Identifier(
+ base.getDatabaseName(), base.getTableName(), "main", sys);
+ try {
+ sysTable = context.executeAuthenticated(() -> {
+ try {
+ return catalogOps.getTable(sysId);
+ } catch (Catalog.TableNotExistException e) {
+ return null;
+ }
+ });
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to load Paimon system
table: " + sysId, e);
+ }
+ if (sysTable == null) {
+ return Optional.empty();
+ }
+ sysBase = null;
}
// #65984 widened the name-forced set to include row_tracking: like
binlog/audit_log its rows
// are materialized by the paimon reader itself, so the native reader
would return wrong rows.
@@ -484,6 +515,7 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
PaimonTableHandle handle = PaimonTableHandle.forSystemTable(
base.getDatabaseName(), base.getTableName(), sys, forceJni);
handle.setPaimonTable(sysTable);
+ handle.setSysBaseTable(sysBase);
return Optional.of(handle);
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonIncrementalScanParams.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonIncrementalScanParams.java
index adc2022057f..4353d30bda2 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonIncrementalScanParams.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonIncrementalScanParams.java
@@ -19,6 +19,11 @@ package org.apache.doris.connector.paimon;
import org.apache.doris.connector.api.DorisConnectorException;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
import java.util.HashMap;
import java.util.Map;
@@ -320,4 +325,87 @@ public final class PaimonIncrementalScanParams {
withResets.putAll(scanOptions);
return withResets;
}
+
+ /**
+ * Bind an incremental relation's range to the catalog-visible snapshot,
for the table the BE
+ * re-plans on a catalog-less copy.
+ *
+ * <p>Only for {@link PaimonScanParams#resolvesSnapshotOnBackend} system
tables, i.e. today
+ * {@code $partitions}, the one table that both re-plans on the BE and
accepts {@code @incr}.
+ * Paimon selects the incremental scanner from {@code
incremental-between*} and never reads
+ * {@code scan.snapshot-id} in that mode, so the snapshot pin
+ * ({@code PaimonScanPlanProvider#pinCatalogSnapshot}) cannot bound this
scan - and
+ * {@link PaimonScanParams#isolateIncrementalRead} clears it anyway, so
one relation's read state
+ * cannot leak into another's.
+ *
+ * <p>The timestamp form is the exposed one. {@code
IncrementalDeltaStartingScanner
+ * #betweenTimestamps} turns both endpoints into snapshot ids through
+ * {@code SnapshotManager#earlierOrEqualTimeMills}, whose binary search
runs up to
+ * {@code latestSnapshotId()}: the catalog's pointer here, but the newest
file in the snapshot
+ * directory on the catalog-less BE. So resolve the endpoints while the
loader is still around
+ * and hand the BE the explicit id range Paimon would have computed
itself. Only the delta
+ * scanner's rule is reachable: {@code incremental-between-scan-mode} is
cleared with the rest of
+ * the read state, and its default {@code AUTO} never selects the diff
scanner.
+ *
+ * <p>Only when the requested end reaches the catalog's snapshot. An older
end already resolves
+ * to the same id on both sides, because every snapshot the catalog has
not published yet is
+ * younger than the one it points at. The snapshot-id form ({@code
startSnapshotId} /
+ * {@code endSnapshotId}) names its endpoints outright and needs nothing.
+ *
+ * <p>Never on a {@code scan.fallback-branch} table, because an id range
cannot be expressed for
+ * the pair. The two branches keep independent snapshot id sequences, and
+ * {@code FallbackReadFileStoreTable#rewriteFallbackOptions} translates
only
+ * {@code scan.snapshot-id} - {@code incremental-between} is copied to the
fallback branch
+ * verbatim. A main-branch id range would then be validated against the
fallback branch's own
+ * range in {@code IncrementalDeltaStartingScanner#betweenSnapshotIds} and
either fail out of
+ * range or select unrelated commits. The timestamp form needs no
translation to begin with: a
+ * wall clock means the same thing on both branches, and each resolves it
against its own
+ * {@code SnapshotManager}, so handing it over untouched is the
branch-correct choice. What that
+ * keeps is the catalog-versus-filesystem endpoint gap {@code
pinCatalogSnapshot} documents for
+ * the fallback branch, which is the same gap on the same table.
+ *
+ * @param scanOptions the relation's resolved scan options
+ * @param dataTable the still catalog-ful base table the FE planned with
+ * @return a NEW map whose timestamp range is replaced by the resolved id
range, or
+ * {@code scanOptions} unchanged (same reference) when the range
is already bound, cannot
+ * be bound, or does not need to be
+ */
+ public static Map<String, String> bindRangeToCatalog(
+ Map<String, String> scanOptions, FileStoreTable dataTable) {
+ String range = scanOptions == null ? null :
scanOptions.get(PAIMON_INCREMENTAL_BETWEEN_TIMESTAMP);
+ if (range == null) {
+ return scanOptions;
+ }
+ // Before any pair-level accessor is read: on a fallback pair both
catalogEnvironment() and
+ // snapshotManager() describe the main branch alone, and the bound
could not be carried to
+ // the fallback branch anyway.
+ if (PaimonTableDecorators.unwrapToFallbackOrBase(dataTable) instanceof
FallbackReadFileStoreTable) {
+ return scanOptions;
+ }
+ if (!dataTable.catalogEnvironment().supportsVersionManagement()) {
+ return scanOptions;
+ }
+ SnapshotManager snapshotManager = dataTable.snapshotManager();
+ Snapshot latest = snapshotManager.latestSnapshot();
+ Snapshot earliest = snapshotManager.earliestSnapshot();
+ if (latest == null || earliest == null) {
+ return scanOptions;
+ }
+ String[] endpoints = range.split(",");
+ long startMillis = Long.parseLong(endpoints[0]);
+ long endMillis = Long.parseLong(endpoints[1]);
+ if (endMillis < latest.timeMillis()) {
+ return scanOptions;
+ }
+ Snapshot start = snapshotManager.earlierOrEqualTimeMills(startMillis);
+ // The starting snapshot is exclusive, hence the id before the
earliest one when the range
+ // opens before it - exactly what betweenTimestamps computes.
+ long startId = (start == null || earliest.timeMillis() > startMillis)
+ ? earliest.id() - 1
+ : start.id();
+ Map<String, String> boundParams = new HashMap<>(scanOptions);
+ boundParams.put(PAIMON_INCREMENTAL_BETWEEN_TIMESTAMP, null);
+ boundParams.put(PAIMON_INCREMENTAL_BETWEEN, startId + "," +
latest.id());
+ return boundParams;
+ }
}
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 4b10b584372..47a6f6dc6eb 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
@@ -103,6 +103,12 @@ public final class PaimonScanParams {
private static final Set<String> PAIMON_READER_SYSTEM_TABLES =
ImmutableSet.of(
"audit_log", "binlog", "row_tracking");
+ // Not a capability the user selects but a property of where the snapshot
is resolved: these five
+ // pick it inside the BE reader rather than reading what the FE planned.
See
+ // resolvesSnapshotOnBackend for the per-table call chain.
+ private static final Set<String> BACKEND_SNAPSHOT_SYSTEM_TABLES =
ImmutableSet.of(
+ "files", "partitions", "manifests", "statistics", "table_indexes");
+
private static final Set<String> OPTIONS_SYSTEM_TABLES = ImmutableSet.of(
// A system table may advertise OPTIONS only when every
row-producing stage observes
// the selected snapshot; files and buckets still consult latest
metadata internally.
@@ -416,6 +422,27 @@ public final class PaimonScanParams {
return
PAIMON_READER_SYSTEM_TABLES.contains(systemTableType.toLowerCase());
}
+ /**
+ * The system tables whose rows are not fixed by what the FE planned: the
FE only sends marker
+ * splits and the snapshot is picked inside the BE reader. Two shapes,
both honoring
+ * {@code scan.snapshot-id}:
+ * <ul>
+ * <li>{@code $files} / {@code $partitions} re-plan the base table on
the BE
+ * ({@code FilesTable.FilesRead#createReader} -> {@code
DataTableScan#plan()},
+ * {@code PartitionsTable.PartitionsRead#createReader} ->
+ * {@code newScan().listPartitionEntries()});</li>
+ * <li>{@code $manifests} / {@code $table_indexes} / {@code $statistics}
resolve it directly
+ * ({@code TimeTravelUtil#tryTravelOrLatest}, reached from {@code
ManifestsTable},
+ * {@code TableIndexesTable} and {@code
AbstractFileStoreTable#statistics}).</li>
+ * </ul>
+ * All five have a fixed row type, so pinning the catalog's snapshot for
them cannot rewind the
+ * BE schema. Consumed by {@code PaimonScanPlanProvider}, which hands the
BE a catalog-less table
+ * and therefore has to do on the FE what the catalog would have done
inside the BE reader.
+ */
+ public static boolean resolvesSnapshotOnBackend(String systemTableType) {
+ return
BACKEND_SNAPSHOT_SYSTEM_TABLES.contains(systemTableType.toLowerCase());
+ }
+
/**
* Rejects the ONE option a system table can never honor. Which system
table accepts {@code @options}
* at all is answered earlier and generically by
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 be76182183e..e24c70723eb 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
@@ -55,12 +55,16 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
import org.apache.paimon.rest.RESTToken;
import org.apache.paimon.rest.RESTTokenFileIO;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.CatalogEnvironment;
+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.DataSplit;
import org.apache.paimon.table.source.DeletionFile;
@@ -72,6 +76,7 @@ import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.table.system.ReadOptimizedTable;
+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;
@@ -863,8 +868,9 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
props.put(ScanNodePropertyKeys.PATH_PARTITION_KEYS,
String.join(",", partitionKeys));
}
- // Serialized table for BE's JNI reader
- String serializedTable = encodeObjectToString(table);
+ // Serialized table for BE's JNI reader, stripped of its catalog
loader (see
+ // tableForBackend) so the BE never has to deserialize the Hive
metastore stack.
+ String serializedTable =
encodeObjectToString(tableForBackend(paimonHandle, table));
props.put(PROP_SERIALIZED_TABLE, serializedTable);
// Serialized predicates for BE's JNI scanner. ALWAYS emit, even for
the no-filter / empty-predicate
@@ -958,6 +964,245 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
return props;
}
+ /**
+ * Build the Paimon table object that is serialized to the BE.
+ *
+ * <p>Every table loaded from a metastore-backed Paimon catalog (HMS /
DLF) carries a Paimon
+ * {@code HiveCatalogLoader} in its {@link CatalogEnvironment}. The BE
only reads — via
+ * FE-resolved splits and the object store — and never needs the catalog,
yet deserializing that
+ * loader forces the whole Hive metastore stack onto the BE classpath:
{@code HiveConf}, the
+ * metastore API, and, when a system table resolves its latest snapshot,
even the metastore
+ * client (DLF's {@code ProxyMetaStoreClient} and its REST stack). So we
serialize a catalog-less
+ * table to the BE:
+ * <ul>
+ * <li>data table: drop the catalog loader. A {@link FileStoreTable} is
fully defined by
+ * fileIO / location / schema / catalogEnvironment, and its dynamic
options (time travel,
+ * incremental) are merged into the schema by {@code copy(...)} —
which
+ * {@link #resolveScanTable} has already applied — so rebuilding from
+ * fileIO / location / schema preserves everything except the
catalog loader.</li>
+ * <li>system table (e.g. {@code $snapshots}): rebuild it over a
catalog-less data table so
+ * {@code SnapshotManager#latestSnapshotId} lists the snapshot
directory on the filesystem
+ * instead of calling the metastore. The base table is the one the
FE-side wrapper was
+ * built over ({@link PaimonTableHandle#getSysBaseTable()}), and for
the system tables that
+ * pick their snapshot on the BE ({@link
PaimonScanParams#resolvesSnapshotOnBackend}) what
+ * the catalog would have done there is done here instead: see
+ * {@link #authorizeDeferredScan} and {@link #pinCatalogSnapshot}.
Every other system table
+ * reads what the FE already planned, so it is handed over
untouched. The relation-scoped
+ * scan params {@link #resolveScanTable} applied to the original
wrapper are re-applied to
+ * the rebuilt one by {@link #reapplyScanParams}.</li>
+ * </ul>
+ */
+ // Package-private for direct unit testing (PaimonBackendBoundTableTest).
+ Table tableForBackend(PaimonTableHandle handle, Table scanTable) {
+ if (scanTable instanceof FileStoreTable) {
+ // resolveScanTable's copy(...) merged the relation's dynamic
options into the schema,
+ // and the rebuild below goes through that schema, so this branch
needs no re-application.
+ return dropCatalogLoader((FileStoreTable) scanTable);
+ }
+ if (!handle.isSystemTable()) {
+ return scanTable;
+ }
+ // The very same base table the FE-side wrapper was built over, so
that the BE never sees a
+ // different schema generation than the one this query was planned
with.
+ FileStoreTable dataTable = handle.getSysBaseTable();
+ if (dataTable == null) {
+ return scanTable;
+ }
+ String sysTableType = handle.getSysTableName();
+ boolean resolvesOnBackend =
PaimonScanParams.resolvesSnapshotOnBackend(sysTableType);
+ if (PAIMON_FILES_SYSTEM_TABLE.equalsIgnoreCase(sysTableType)) {
+ authorizeDeferredScan(dataTable);
+ }
+ FileStoreTable baseForBackend = dropCatalogLoader(dataTable);
+ if (resolvesOnBackend) {
+ baseForBackend = pinCatalogSnapshot(baseForBackend, dataTable);
+ }
+ Table catalogLessSysTable = SystemTableLoader.load(sysTableType,
baseForBackend);
+ if (catalogLessSysTable == null) {
+ return scanTable;
+ }
+ return reapplyScanParams(catalogLessSysTable, dataTable,
resolvesOnBackend,
+ handle.getScanOptions());
+ }
+
+ /**
+ * Re-apply the relation-scoped scan params to a rebuilt system-table
wrapper.
+ *
+ * <p>{@link #resolveScanTable} applies {@code @incr} / {@code @options}
to the wrapper the
+ * handle carries, and every Paimon system table delegates {@code
copy(...)} to the data table it
+ * wraps. The wrapper rebuilt above is a different object, so the same
copy has to be redone on
+ * it, otherwise the BE would materialize its splits against the unpinned
latest state. The
+ * branches mirror {@link #resolveScanTable}'s exactly — one chokepoint's
worth of logic applied
+ * to two different objects.
+ *
+ * <p>This runs last on purpose: an explicit relation option outranks
anything this class pins on
+ * the rebuilt table, and {@code copy(...)} lets the option win. An
incremental relation outranks
+ * {@link #pinCatalogSnapshot} the same way but cannot inherit its bound,
so it is bound to the
+ * catalog's snapshot separately by {@link
PaimonIncrementalScanParams#bindRangeToCatalog}.
+ *
+ * <p>Known gap: the {@code @options} branch reaches tables outside
+ * {@link PaimonScanParams#resolvesSnapshotOnBackend}, {@code $ro} among
them, and on a
+ * {@code scan.fallback-branch} table the {@code copy(...)} here is what
triggers
+ * {@code rewriteFallbackOptions} - on the already catalog-less pair, so
the fallback branch's
+ * {@code scan.snapshot-id} is derived from its snapshot directory rather
than from the catalog
+ * pointer, the same gap {@link #pinCatalogSnapshot} documents. It lands
harder here: unlike the
+ * pin this is {@code copy(...)}, not {@code copyWithoutTimeTravel(...)},
so it also time-travels
+ * the fallback branch's schema, and {@code $ro} is a data table rather
than read-only metadata.
+ */
+ private Table reapplyScanParams(Table rebuiltSysTable, FileStoreTable
dataTable,
+ boolean resolvesOnBackend, Map<String, String> scanOptions) {
+ if (scanOptions == null || scanOptions.isEmpty()) {
+ return rebuiltSysTable;
+ }
+ if (PaimonScanParams.isOptionsPin(scanOptions)) {
+ return PaimonScanParams.applyOptions(rebuiltSysTable, scanOptions);
+ }
+ Map<String, String> params = resolvesOnBackend
+ ? PaimonIncrementalScanParams.bindRangeToCatalog(scanOptions,
dataTable)
+ : scanOptions;
+ return
rebuiltSysTable.copy(PaimonIncrementalScanParams.applyResetsIfIncremental(params));
+ }
+
+ /**
+ * {@code $files} plans only partition-level splits on the FE and re-plans
the base table on the
+ * BE through {@code DataTableScan#plan()} ({@code
FilesTable.FilesRead#createReader}). That
+ * deferred plan normally authorizes itself through the catalog loader
+ * ({@code CatalogEnvironment#tableQueryAuth} -> {@code
Catalog#authTableQuery}); once the
+ * loader is dropped it silently allows everything. So authorize here,
while the loader is still
+ * around. Paimon discards the predicates the call returns (row level
access control is a TODO in
+ * {@code AbstractDataTableScan#authQuery}), so running it on the FE loses
nothing.
+ *
+ * <p>Only {@code $files} may do this: {@code auth(null)} means "every
column" to
+ * {@code Catalog#authTableQuery}, and the system tables that keep
planning on the FE
+ * ({@code $ro}, {@code $row_tracking}, {@code $audit_log}, {@code
$binlog}) already authorize
+ * themselves through {@code DataTableBatchScan} with the slot projection
the query really reads.
+ * Authorizing those again for every column would reject a user allowed to
read only some of the
+ * base columns. {@code $partitions} never reaches {@code plan()} on
either side
+ * ({@code listPartitionEntries} does not authorize), so it has no
authorization to transfer.
+ *
+ * <p>A {@code scan.fallback-branch} table has to be authorized branch by
branch.
+ * {@code FallbackReadFileStoreTable#newScan} builds a {@code
FallbackReadScan} over both
+ * branches' own scans, so each authorizes itself, and {@code
FileStoreTableFactory#create} gives
+ * the fallback branch a {@link CatalogEnvironment} of its own carrying a
branch-qualified
+ * {@code Identifier}. Since the pair delegates {@code
catalogEnvironment()} to its main branch,
+ * authorizing the pair would check the main branch alone and never the
fallback one - and once
+ * its loader is dropped that missing check turns into a permanent allow,
letting a user denied on
+ * the fallback branch read the fallback rows of {@code $files}.
+ */
+ // Package-private for direct unit testing (PaimonBackendBoundTableTest).
+ static void authorizeDeferredScan(FileStoreTable dataTable) {
+ FileStoreTable undecorated =
PaimonTableDecorators.unwrapToFallbackOrBase(dataTable);
+ if (undecorated instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable fallbackReadTable =
(FallbackReadFileStoreTable) undecorated;
+ authorizeBranch(fallbackReadTable.wrapped());
+ authorizeBranch(fallbackReadTable.fallback());
+ return;
+ }
+ authorizeBranch(undecorated);
+ }
+
+ private static void authorizeBranch(FileStoreTable branch) {
+ CoreOptions options = branch.coreOptions();
+ if (options.queryAuthEnabled()) {
+ branch.catalogEnvironment().tableQueryAuth(options).auth(null);
+ }
+ }
+
+ /**
+ * For catalogs that manage versions themselves (Paimon REST / DLF REST)
the committed snapshot
+ * is the one the catalog points at, not the newest file in the snapshot
directory: Paimon
+ * publishes the snapshot file before the pointer moves, and a rollback
leaves newer files
+ * behind. Without the catalog loader {@code SnapshotManager} falls back
to listing that
+ * directory, so the BE could plan on a snapshot the catalog has not
published while the FE
+ * planned on the previous one. Pin the catalog-visible snapshot instead.
+ *
+ * <p>Only for {@link PaimonScanParams#resolvesSnapshotOnBackend} system
tables, because only
+ * those pick their snapshot inside the BE reader. Every other system
table materializes the
+ * splits the FE already planned, so pinning them would bind nothing that
is not already bound.
+ * The pin goes through {@code copyWithoutTimeTravel}, so it bounds which
snapshot the BE plans
+ * on without rewinding the BE's schema to that snapshot.
+ *
+ * <p>Known gap: {@code scan.snapshot-id} only bounds plans that go through
+ * {@code DataTableScan}. {@code $snapshots} ({@code
SnapshotManager#snapshotsWithinRange}) and
+ * {@code $buckets} ({@code SnapshotReader#bucketEntries}) ignore it, so
on the BE they observe
+ * the snapshot directory rather than the catalog's pointer. Both are
read-only metadata tables,
+ * so this shows up only as a transient extra row inside the publication
window of a
+ * version-managed catalog.
+ *
+ * <p>Known gap: {@code $files} is planned in two phases and only the
second one can be pinned.
+ * The FE emits one marker split per partition of {@code
SnapshotManager#latestSnapshot()} at
+ * split generation time ({@code FilesTable.FilesScan#innerPlan} ->
+ * {@code SnapshotReader#partitions}), and that path reads {@code
ManifestsReader#read(null, ..)},
+ * which ignores {@code scan.snapshot-id} - so the marker set cannot be
bound to the pin, and
+ * {@code FilesSplit} is private to {@code FilesTable}, so Doris cannot
emit a snapshot
+ * independent marker either. A commit landing between initialization and
split generation that
+ * drops a partition therefore hides that partition's rows even though the
BE stays on the older
+ * snapshot. Paimon's two-phase plan has this gap regardless: before the
pin the BE resolved the
+ * latest snapshot at read time, i.e. across a strictly wider window, so
this only moves which
+ * side of the window the mismatch falls on.
+ *
+ * <p>Known gap: on a table with {@code scan.fallback-branch} this bounds
the main branch only.
+ * The pin itself is carried across correctly - {@code
copyWithoutTimeTravel} is pair-aware and
+ * {@code rewriteFallbackOptions} converts the main id to the fallback
branch's own through
+ * {@code SnapshotManager#earlierOrEqualTimeMills}. What is lost is the
pointer that conversion
+ * searches against: the rebuilt pair has no catalog loader on either
branch, so the fallback
+ * bound is capped by the newest file in that branch's snapshot directory
instead of by the
+ * catalog. Of the pinned tables only {@code $partitions} and {@code
$files} read the fallback
+ * branch at all ({@code $manifests} / {@code $statistics} / {@code
$table_indexes} reach it
+ * through {@code store()} / {@code statistics()}, which describe the main
branch), so the
+ * effect is confined to their rows. Note this outlasts a publication
window when it is caused
+ * by a rollback: the abandoned snapshot file stays on the filesystem
until it expires, and for
+ * as long as it does the fallback branch can be bound to it.
+ */
+ // Package-private for direct unit testing (PaimonBackendBoundTableTest).
+ static FileStoreTable pinCatalogSnapshot(FileStoreTable catalogLessTable,
FileStoreTable dataTable) {
+ if (!dataTable.catalogEnvironment().supportsVersionManagement()) {
+ return catalogLessTable;
+ }
+ Long snapshotId = dataTable.snapshotManager().latestSnapshotId();
+ if (snapshotId == null) {
+ return catalogLessTable;
+ }
+ // Without time travel: pin which snapshot the BE plans on, leave
schema resolution alone.
+ return catalogLessTable.copyWithoutTimeTravel(
+ Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(),
String.valueOf(snapshotId)));
+ }
+
+ /**
+ * Return an equivalent {@link FileStoreTable} without the catalog loader,
so the BE never
+ * deserializes a {@code HiveCatalogLoader} (and snapshot loading uses the
filesystem instead of
+ * the catalog's metastore). fileIO / location / schema (and the schema's
options) are preserved.
+ *
+ * <p>A {@code scan.fallback-branch} table must be rebuilt branch by
branch.
+ * {@link FallbackReadFileStoreTable} exposes only its main branch through
{@code schema()}, and
+ * the plain {@code FileStoreTableFactory#create} re-expands the fallback
branch from
+ * {@code SchemaManager(fallbackBranch).latest()} instead of the object
the FE captured. That
+ * would ship a main/fallback pair from two different generations: after
external DDL publishes
+ * a new schema on both branches, the FE keeps planning on the cached
M1/F1 while the BE gets
+ * M1/F2 and fails in {@code FallbackReadFileStoreTable#validateSchema}.
So rebuild each branch
+ * from the schema the FE really planned with, and re-wrap.
+ */
+ // Package-private for direct unit testing (PaimonBackendBoundTableTest).
+ static FileStoreTable dropCatalogLoader(FileStoreTable dataTable) {
+ 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.fallback()));
+ }
+ return rebuildWithoutCatalogLoader(undecorated);
+ }
+
+ private static FileStoreTable rebuildWithoutCatalogLoader(FileStoreTable
branch) {
+ return FileStoreTableFactory.createWithoutFallbackBranch(
+ branch.fileIO(), branch.location(), branch.schema(), new
Options(),
+ CatalogEnvironment.empty());
+ }
+
/**
* Resolves the {@link FileStoreTable} whose schema dictionary BE needs to
field-id-match the native
* data files for {@code table}. A normal data table IS the
FileStoreTable. A read-optimized system
@@ -1353,6 +1598,10 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
private static final String PAIMON_PROPERTY_PREFIX = "paimon.";
private static final String PAIMON_BINLOG_SYSTEM_TABLE = "binlog";
+ // The one system table whose deferred BE-side plan must inherit the
catalog's authorization —
+ // see authorizeDeferredScan.
+ private static final String PAIMON_FILES_SYSTEM_TABLE = "files";
+
private static final List<String> BACKEND_PAIMON_JNI_OPTIONS =
Arrays.asList(
"jni.enable_jni_io_manager",
"jni.io_manager.tmp_dir",
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
new file mode 100644
index 00000000000..fbebd7b1989
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.paimon;
+
+import org.apache.paimon.table.DelegatedFileStoreTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+
+/**
+ * The one place that knows how Paimon stacks {@link DelegatedFileStoreTable}
decorators on a loaded
+ * table, and how far they may be peeled off.
+ */
+final class PaimonTableDecorators {
+
+ private PaimonTableDecorators() {
+ }
+
+ /**
+ * 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
+ * right branch.
+ *
+ * <p>{@code PrivilegedCatalog#getTable} wraps whatever {@code
FileStoreTableFactory} built into
+ * a {@code PrivilegedFileStoreTable}, so with file based privileges
enabled a
+ * {@code scan.fallback-branch} table reaches Doris as {@code
Privileged(FallbackRead(...))}.
+ * Paimon itself never lets a decorator sit above the pair - {@code
CatalogUtils#loadTable}
+ * builds system tables straight over the {@code FileStoreTableFactory}
result and only wraps
+ * what it returns - and both Paimon and Doris dispatch on a direct {@code
instanceof
+ * FallbackReadFileStoreTable}, which looks straight past a decorator and
silently falls back to
+ * the delegated main branch alone.
+ *
+ * <p>The decorators are peeled off rather than re-applied, and what that
costs differs by
+ * caller. Rebuilding the table the BE gets loses nothing: the privilege
wrapper only asserts on
+ * {@code newScan()} / {@code newRead()}, which the FE has already run
while planning, and a
+ * plain (non fallback) table loses the wrapper the same way once it is
rebuilt out of
+ * fileIO / location / schema. Building a system table over the peeled
base does drop that
+ * assertion, but that is Paimon's own semantics rather than a relaxation
of it:
+ * {@code PrivilegedCatalog#getTable} wraps only a result that is a {@code
FileStoreTable}, and
+ * no system table is one, so {@code db.tbl$ro} never carries the
decorator there either.
+ */
+ static FileStoreTable unwrapToFallbackOrBase(FileStoreTable dataTable) {
+ FileStoreTable current = dataTable;
+ while (current instanceof DelegatedFileStoreTable && !(current
instanceof FallbackReadFileStoreTable)) {
+ current = ((DelegatedFileStoreTable) current).wrapped();
+ }
+ return current;
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableHandle.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableHandle.java
index 0d4738e211f..4b5b20af6bc 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableHandle.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableHandle.java
@@ -19,6 +19,7 @@ package org.apache.doris.connector.paimon;
import org.apache.doris.connector.api.handle.ConnectorTableHandle;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import java.util.Collections;
@@ -88,6 +89,24 @@ public class PaimonTableHandle implements
ConnectorTableHandle {
/** Transient Paimon Table reference; not serialized. Set by
PaimonConnectorMetadata. */
private transient Table paimonTable;
+ /**
+ * For a SYSTEM handle, the base {@link FileStoreTable} its {@link
#paimonTable} wrapper was built
+ * over - the very same schema generation, because {@link
PaimonConnectorMetadata#getSysTableHandle}
+ * builds both from one load. {@code null} for a normal handle and for the
catalog fallback
+ * (format / object tables, which have no {@code FileStoreTable} base).
+ *
+ * <p>It exists so {@code PaimonScanPlanProvider} can rebuild an
equivalent wrapper over a
+ * catalog-less copy for the BE without ever asking the catalog for a
second, possibly newer,
+ * generation: the FE would then plan on one generation while the BE
materialized the splits on
+ * another, and a system table whose row type follows the base schema
({@code $audit_log},
+ * {@code $binlog}, {@code $ro}) could reach the BE without the columns
the FE asked for.
+ *
+ * <p>Transient like {@link #paimonTable}: after a serialization
round-trip it is null and the BE
+ * simply gets the wrapper as it stands (the pre-#65867 behaviour), never
a wrapper rebuilt over a
+ * base this handle did not resolve.
+ */
+ private transient FileStoreTable sysBaseTable;
+
public PaimonTableHandle(String databaseName, String tableName,
List<String> partitionKeys, List<String> primaryKeys) {
this(databaseName, tableName, partitionKeys, primaryKeys, null, false);
@@ -182,11 +201,16 @@ public class PaimonTableHandle implements
ConnectorTableHandle {
* scanOptions — the snapshot-pinned read variant. The transient Table is
copied over as-is; the
* scan path applies {@code Table.copy(scanOptions)} at resolution time.
branchName is preserved
* because it is part of the handle identity.
+ *
+ * <p>{@link #sysBaseTable} travels with the Table it belongs to: pinning
a system table does not
+ * change which base its wrapper was built over, and dropping it here
would silently send the BE a
+ * catalog-ful wrapper for exactly the pinned reads that need the rebuilt
one most.
*/
public PaimonTableHandle withScanOptions(Map<String, String> options) {
PaimonTableHandle copy = new PaimonTableHandle(databaseName, tableName,
partitionKeys, primaryKeys, sysTableName, forceJni, options,
branchName);
copy.paimonTable = this.paimonTable;
+ copy.sysBaseTable = this.sysBaseTable;
return copy;
}
@@ -216,6 +240,19 @@ public class PaimonTableHandle implements
ConnectorTableHandle {
this.paimonTable = paimonTable;
}
+ /**
+ * The base table this system handle's wrapper was built over, or null
(normal handle, catalog
+ * fallback, or a deserialized handle). See {@link #sysBaseTable}.
+ */
+ public FileStoreTable getSysBaseTable() {
+ return sysBaseTable;
+ }
+
+ /** Sets the base table this system handle's wrapper was built over. See
{@link #sysBaseTable}. */
+ public void setSysBaseTable(FileStoreTable sysBaseTable) {
+ this.sysBaseTable = sysBaseTable;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
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
new file mode 100644
index 00000000000..755d367caf9
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
@@ -0,0 +1,762 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.paimon;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.privilege.AllGrantedPrivilegeChecker;
+import org.apache.paimon.privilege.PrivilegedFileStoreTable;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.CatalogEnvironment;
+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.TableSnapshot;
+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.system.ReadOptimizedTable;
+import org.apache.paimon.table.system.SystemTableLoader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.InstantiationUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * The table {@link PaimonScanPlanProvider#getScanNodeProperties} serializes
to the BE must carry no
+ * Paimon {@code CatalogLoader}: deserializing one drags the whole Hive
metastore stack onto the BE
+ * classpath ({@code HiveConf}, the metastore API, and for a version-managed
catalog even the
+ * metastore client). These tests pin BOTH halves of that: the loader really
is gone, and everything
+ * the loader used to do on the BE — snapshot resolution, query authorization,
the fallback-branch
+ * pair, the relation's own scan params — is done on the FE instead, so
dropping it changes no rows.
+ *
+ * <p>No mocking framework (this module has none by convention): the
divergence a version-managed
+ * catalog creates is reproduced for real by {@link VersionManagedCatalog},
whose {@code loadSnapshot}
+ * answers a pointer the test sets while the snapshot directory on disk holds
something newer —
+ * exactly the publication window / post-rollback shape the pin exists for.
+ */
+public class PaimonBackendBoundTableTest {
+
+ private static final DataField C1 = new DataField(0, "c1",
DataTypes.INT());
+ private static final DataField C2 = new DataField(1, "c2",
DataTypes.INT());
+
+ private static PaimonScanPlanProvider provider() {
+ return new PaimonScanPlanProvider(Collections.emptyMap(), null);
+ }
+
+ private static PaimonConnectorMetadata metadata(RecordingPaimonCatalogOps
ops) {
+ return new PaimonConnectorMetadata(ops, Collections.emptyMap(), new
RecordingConnectorContext());
+ }
+
+ // ==================== the loader itself ====================
+
+ @Test
+ public void dropCatalogLoaderKeepsEverythingButTheLoader(@TempDir Path
warehouse) {
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable table = newTable(warehouse,
catalogEnvironment(catalog), C1);
+
+ FileStoreTable catalogLess =
PaimonScanPlanProvider.dropCatalogLoader(table);
+
+ // The loader is the only thing the BE must not deserialize;
everything a FileStoreTable is
+ // defined by has to survive, or the BE reads a different table than
the FE planned.
+ Assertions.assertNull(catalogLess.catalogEnvironment().catalogLoader(),
+ "the catalog loader must not reach the BE");
+ Assertions.assertEquals(table.rowType(), catalogLess.rowType());
+ Assertions.assertEquals(table.location(), catalogLess.location());
+ Assertions.assertEquals(table.schema().id(),
catalogLess.schema().id());
+ Assertions.assertEquals(table.schema().options(),
catalogLess.schema().options());
+ }
+
+ @Test
+ public void dataTableReachesTheBackendWithoutItsCatalogLoader(@TempDir
Path warehouse) {
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable table = newTable(warehouse,
catalogEnvironment(catalog), C1);
+
+ Table forBackend = provider().tableForBackend(dataHandle(), table);
+
+ Assertions.assertNull(((FileStoreTable)
forBackend).catalogEnvironment().catalogLoader(),
+ "a plain data table must also be stripped, not only system
tables");
+ }
+
+ @Test
+ public void theSerializedTablePropertyIsTheStrippedOne(@TempDir Path
warehouse) throws Exception {
+ // The wiring, end to end: whatever tableForBackend returns is what
actually lands in
+ // paimon.serialized_table and gets Java-deserialized on the BE.
Asserting only on
+ // tableForBackend would leave getScanNodeProperties free to serialize
the catalog-ful table.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable table = newTable(warehouse,
catalogEnvironment(catalog), C1);
+ PaimonTableHandle handle = dataHandle();
+ handle.setPaimonTable(table);
+ Assertions.assertNotNull(table.catalogEnvironment().catalogLoader(),
+ "precondition: the FE-side table really does carry a loader");
+
+ Map<String, String> props = provider().getScanNodeProperties(null,
handle,
+ Collections.emptyList(), Optional.empty());
+
+ FileStoreTable onBackend =
deserializeTable(props.get("paimon.serialized_table"));
+ Assertions.assertNull(onBackend.catalogEnvironment().catalogLoader(),
+ "the BE must never deserialize a CatalogLoader (it drags in
the Hive metastore stack)");
+ Assertions.assertEquals(table.rowType(), onBackend.rowType());
+ }
+
+ // ==================== schema generation ====================
+
+ @Test
+ public void
systemTableForBackendFollowsTheFeWrapperSchemaGeneration(@TempDir Path
warehouse) {
+ // $audit_log / $binlog / $ro derive their row type from the base
table schema. The FE plans
+ // on the wrapper the sys handle carries; rebuilding the table
serialized to the BE over a
+ // DIFFERENT generation of that base makes BE reject the query in
+ // PaimonJniScanner#getProjected ("RequiredField c2 not found in
schema") or read a stale
+ // type under the same name. Both wrappers must come from one
generation.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable planned = newTable(warehouse,
catalogEnvironment(catalog), C1, C2);
+ Table feWrapper = SystemTableLoader.load("audit_log", planned);
+
+ Table forBackend = provider().tableForBackend(sysHandle("audit_log",
planned), feWrapper);
+
+ Assertions.assertEquals(feWrapper.rowType(), forBackend.rowType());
+
Assertions.assertTrue(forBackend.rowType().getFieldNames().contains("c2"),
+ "the BE must see the generation the FE planned with");
+ }
+
+ @Test
+ public void systemTableWithoutACapturedBaseIsHandedOverUntouched(@TempDir
Path warehouse) {
+ // A format / object table has no FileStoreTable base, so
getSysTableHandle falls back to the
+ // catalog and captures nothing. Rebuilding from some other base would
be a guess; the
+ // pre-#65867 behaviour (hand the wrapper over as it stands) is the
safe answer.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable planned = newTable(warehouse,
catalogEnvironment(catalog), C1);
+ Table feWrapper = SystemTableLoader.load("audit_log", planned);
+ PaimonTableHandle handle = PaimonTableHandle.forSystemTable("db",
"tbl", "audit_log", true);
+
+ Assertions.assertSame(feWrapper, provider().tableForBackend(handle,
feWrapper));
+ }
+
+ // ==================== capturing that base in the first place
====================
+
+ @Test
+ public void sysTableHandleCarriesTheBaseItsWrapperWasBuiltOver(@TempDir
Path warehouse) {
+ // The wrapper and the base must come from ONE load, or "the
generation the FE planned with"
+ // is unknowable downstream. Loading the wrapper through the catalog's
4-arg sys Identifier
+ // would give it a generation of its own, so the wrapper is built
here, over the base handle's
+ // own Table, and that base travels with the sys handle.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable base = newTable(warehouse, catalogEnvironment(catalog),
C1, C2);
+ PaimonTableHandle baseHandle = dataHandle();
+ baseHandle.setPaimonTable(base);
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+
+ PaimonTableHandle sysHandle = (PaimonTableHandle) metadata(ops)
+ .getSysTableHandle(null, baseHandle,
"audit_log").orElseThrow(AssertionError::new);
+
+ Assertions.assertSame(base, sysHandle.getSysBaseTable(),
+ "the sys handle must carry the very base its wrapper was built
over");
+ Assertions.assertEquals(base.rowType().getFieldNames().size() + 1,
+ sysHandle.getPaimonTable().rowType().getFieldNames().size(),
+ "$audit_log is the base row type plus rowkind, i.e. built over
that same base");
+ Assertions.assertNull(ops.lastGetTableId,
+ "a FileStoreTable base needs no second catalog round-trip for
its system table");
+ }
+
+ @Test
+ public void
sysTableHandleBuildsTheWrapperOverThePeeledFallbackPair(@TempDir Path warehouse)
+ throws Exception {
+ // With file based privileges enabled the meta cache holds
Privileged(FallbackRead(..)) - a
+ // shape paimon's own catalog never lets a system table see, because
CatalogUtils#loadTable
+ // builds one over the raw FileStoreTableFactory result.
ReadOptimizedTable#newScan dispatches
+ // on a DIRECT instanceof, so with the privilege wrapper still in
between it plans the main
+ // branch alone through the pair's inherited newSnapshotReader() and
silently drops every
+ // fallback-only partition.
+ FileStoreTable[] branches =
fallbackPairWithNewerGenerationOnDisk(warehouse, "fb_sys");
+ FileStoreTable pair = new FallbackReadFileStoreTable(branches[0],
branches[1]);
+ FileStoreTable decorated = PrivilegedFileStoreTable.wrap(pair,
+ new AllGrantedPrivilegeChecker(), Identifier.create("db",
"tbl"));
+ PaimonTableHandle baseHandle = dataHandle();
+ baseHandle.setPaimonTable(decorated);
+
+ PaimonTableHandle sysHandle = (PaimonTableHandle) metadata(new
RecordingPaimonCatalogOps())
+ .getSysTableHandle(null, baseHandle,
"ro").orElseThrow(AssertionError::new);
+
+ Assertions.assertSame(pair, sysHandle.getSysBaseTable(),
+ "the decorator must be peeled, but not the fallback pair under
it");
+ Assertions.assertTrue(((ReadOptimizedTable)
sysHandle.getPaimonTable()).newScan()
+ instanceof FallbackReadFileStoreTable.FallbackReadScan,
+ "the FE must plan tbl$ro over BOTH branches");
+ // What that stands for: leaving the decorator on takes newScan down
its single-branch path.
+ Assertions.assertFalse(((ReadOptimizedTable)
SystemTableLoader.load("ro", decorated)).newScan()
+ instanceof
FallbackReadFileStoreTable.FallbackReadScan);
+ }
+
+ @Test
+ public void pinningASysHandleKeepsItsCapturedBase(@TempDir Path warehouse)
{
+ // applySnapshot rebuilds the handle through withScanOptions for every
@options / @incr read.
+ // Losing the captured base there would send the BE a catalog-ful
wrapper for exactly the
+ // pinned reads that need the rebuilt one most.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable base = newTable(warehouse, catalogEnvironment(catalog),
C1);
+
+ PaimonTableHandle pinned = sysHandle("ro", base).withScanOptions(
+ Collections.singletonMap("scan.snapshot-id", "7"));
+
+ Assertions.assertSame(base, pinned.getSysBaseTable());
+ }
+
+ // ==================== the snapshot the BE plans on ====================
+
+ @Test
+ public void
pinsTheCatalogVisibleSnapshotForAVersionManagedCatalog(@TempDir Path warehouse)
+ throws Exception {
+ // A version-managed (REST / DLF REST) catalog owns the committed
snapshot pointer. Without
+ // the loader the BE resolves "latest" by listing the snapshot
directory, so it can plan on a
+ // snapshot the catalog has not published yet - or one a rollback left
behind - while the FE
+ // planned on the previous one.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ FileStoreTable onDisk = commit(newRealTable(warehouse, "t"), 2);
+ // The catalog still points at snapshot 1 while 2 already sits in the
snapshot directory.
+ catalog.pointAt(onDisk, 1L);
+ FileStoreTable versionManaged = withCatalogEnvironment(onDisk,
catalogEnvironment(catalog));
+
+ FileStoreTable pinned = PaimonScanPlanProvider.pinCatalogSnapshot(
+ PaimonScanPlanProvider.dropCatalogLoader(versionManaged),
versionManaged);
+
+ Assertions.assertEquals("1", pinned.options().get("scan.snapshot-id"),
+ "the BE must be pinned to the catalog's pointer, not to the
newest snapshot file");
+ Assertions.assertEquals(2L,
onDisk.snapshotManager().latestSnapshotIdFromFileSystem(),
+ "the divergence this pins against must be real, not an
artifact of the fixture");
+ }
+
+ @Test
+ public void leavesTheSnapshotUnpinnedForAFilesystemCatalog(@TempDir Path
warehouse) throws Exception {
+ // A catalog that does not manage versions keeps filesystem semantics
on both sides, so there
+ // is nothing to pin and the BE must stay on "latest".
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable plain = withCatalogEnvironment(
+ commit(newRealTable(warehouse, "t"), 1),
catalogEnvironment(catalog));
+
+ Assertions.assertNull(PaimonScanPlanProvider
+
.pinCatalogSnapshot(PaimonScanPlanProvider.dropCatalogLoader(plain), plain)
+ .options().get("scan.snapshot-id"));
+ }
+
+ @Test
+ public void
pinsEverySystemTableThatResolvesItsSnapshotOnTheBackend(@TempDir Path warehouse)
+ throws Exception {
+ // $files / $partitions re-plan the base table inside the BE reader;
$manifests /
+ // $table_indexes / $statistics resolve their snapshot directly there
(TimeTravelUtil
+ // #tryTravelOrLatest). All five honor scan.snapshot-id and have a
fixed row type, so all five
+ // must be bound to what the catalog sees before they leave the FE.
+ for (String sysTableType : Arrays.asList("files", "partitions",
"manifests", "statistics",
+ "table_indexes")) {
+ VersionManagedCatalog catalog = new
VersionManagedCatalog(warehouse, true);
+ FileStoreTable onDisk = commit(newRealTable(warehouse, "t_" +
sysTableType), 1);
+ catalog.pointAt(onDisk, 1L);
+ FileStoreTable versionManaged = withCatalogEnvironment(onDisk,
catalogEnvironment(catalog));
+
+ provider().tableForBackend(sysHandle(sysTableType, versionManaged),
+ SystemTableLoader.load(sysTableType, versionManaged));
+
+ Assertions.assertTrue(catalog.loadSnapshotCalls > 0,
+ "$" + sysTableType + " must be bound to the catalog's
snapshot on the FE");
+ }
+ }
+
+ @Test
+ public void
keepsThePinOffSystemTablesWhoseRowTypeFollowsTheBaseTable(@TempDir Path
warehouse)
+ throws Exception {
+ // PaimonJniScanner#initTable calls table.copy(table.options())
unconditionally, and every
+ // system-table wrapper delegates copy to FileStoreTable#copy, which
time-travels the schema
+ // to scan.snapshot-id. $audit_log / $ro / $binlog / $row_tracking
derive their row type from
+ // the base table, so pinning them would rewind the BE schema: c2,
added after the pinned
+ // snapshot, is planned by the FE and then rejected by getProjected()
with "RequiredField c2
+ // not found in schema".
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ FileStoreTable versionManaged = newTable(warehouse,
catalogEnvironment(catalog), C1, C2);
+
+ Table forBackend = provider().tableForBackend(sysHandle("audit_log",
versionManaged),
+ SystemTableLoader.load("audit_log", versionManaged));
+
+ Assertions.assertNull(forBackend.options().get("scan.snapshot-id"),
+ "a row-type-following system table must not be time-travelled
on the BE");
+
Assertions.assertTrue(forBackend.rowType().getFieldNames().contains("c2"));
+ Assertions.assertEquals(0, catalog.loadSnapshotCalls,
+ "not merely unpinned: the catalog's pointer must never even be
consulted");
+ }
+
+ // ==================== authorization the BE can no longer do
====================
+
+ @Test
+ public void
authorizesTheDeferredScanBeforeDroppingTheCatalogLoader(@TempDir Path
warehouse) {
+ // $files only plans partition-level splits on the FE and re-plans the
base table on the BE,
+ // where Catalog#authTableQuery is what enforces query-auth. A
loader-less table turns that
+ // check into a permanent allow, so a denied query would come back as
a successful metadata
+ // read: authorize here instead, while the loader is still around.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ catalog.denyAuthWith("no privilege for db.tbl");
+ FileStoreTable authorized = newTable(warehouse,
catalogEnvironment(catalog),
+ queryAuthEnabled(), C1);
+
+ RuntimeException denied =
Assertions.assertThrows(RuntimeException.class,
+ () -> provider().tableForBackend(sysHandle("files",
authorized),
+ SystemTableLoader.load("files", authorized)));
+
+ Assertions.assertTrue(rootCauseMessage(denied).contains("no privilege
for db.tbl"),
+ "a denied query must not reach the BE as a catalog-less
table");
+ }
+
+ @Test
+ public void leavesDataSystemTablesTheirOwnProjectedAuthorization(@TempDir
Path warehouse) {
+ // $ro / $row_tracking / $audit_log / $binlog keep planning on the FE
through
+ // DataTableBatchScan, which already calls authTableQuery with the
slot projection
+ // (auth(readType.getFieldNames())). A second auth(null) here means
"every column", so a user
+ // allowed to read only c1 would be rejected for "SELECT c1 FROM
tbl$ro".
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ FileStoreTable dataTable = newTable(warehouse,
catalogEnvironment(catalog),
+ queryAuthEnabled(), C1);
+
+ provider().tableForBackend(sysHandle("ro", dataTable),
SystemTableLoader.load("ro", dataTable));
+
+ Assertions.assertTrue(catalog.authCalls.isEmpty(),
+ "only $files, which loses its authorization by re-planning on
the BE, may transfer it");
+ }
+
+ // ==================== the fallback-branch pair ====================
+
+ @Test
+ public void fallbackBranchKeepsTheGenerationTheFeCaptured(@TempDir Path
warehouse) throws Exception {
+ // FallbackReadFileStoreTable#schema() only exposes its main branch,
so rebuilding it through
+ // FileStoreTableFactory#create re-reads the fallback branch from
+ // SchemaManager(fallbackBranch).latest() instead of the object the FE
planned with. After
+ // external DDL publishes a new generation the BE would get main M1 +
fallback F2 and die in
+ // FallbackReadFileStoreTable#validateSchema.
+ FileStoreTable[] captured =
fallbackPairWithNewerGenerationOnDisk(warehouse, "fb_generation");
+
+ FileStoreTable forBackend = PaimonScanPlanProvider.dropCatalogLoader(
+ new FallbackReadFileStoreTable(captured[0], captured[1]));
+
+ assertPairMatchesTheFeGeneration(forBackend, captured[0], captured[1]);
+ }
+
+ @Test
+ public void fallbackBranchSurvivesAPaimonTableDecorator(@TempDir Path
warehouse) throws Exception {
+ // With file based privileges enabled PrivilegedCatalog#getTable hands
out
+ // Privileged(FallbackRead(..)). A direct instanceof looks straight
past that wrapper and
+ // rebuilds an ordinary table from the delegated main branch, so the
BE loses the
+ // FallbackReadFileStoreTable.Read that dispatches a fallback split to
the fallback branch -
+ // while the FE keeps planning the wrapper and can still emit one.
+ FileStoreTable[] captured =
fallbackPairWithNewerGenerationOnDisk(warehouse, "fb_decorated");
+ FileStoreTable decorated = PrivilegedFileStoreTable.wrap(
+ new FallbackReadFileStoreTable(captured[0], captured[1]),
+ new AllGrantedPrivilegeChecker(), Identifier.create("db",
"tbl"));
+
+
assertPairMatchesTheFeGeneration(PaimonScanPlanProvider.dropCatalogLoader(decorated),
+ captured[0], captured[1]);
+ }
+
+ @Test
+ public void authorizesBothBranchesOfAFallbackPair(@TempDir Path warehouse)
{
+ // FallbackReadFileStoreTable#newScan builds a FallbackReadScan over
both branches' own scans,
+ // so each authorizes itself, and FileStoreTableFactory#create gives
the fallback branch a
+ // CatalogEnvironment of its own carrying a branch-qualified
Identifier. The pair delegates
+ // catalogEnvironment() to its main branch, so authorizing the pair
checks main and silently
+ // skips the fallback branch - and once the loaders are dropped that
missing check is a
+ // permanent allow, letting a user denied on the fallback branch read
the fallback rows.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ Identifier mainIdentifier = Identifier.create("db", "tbl");
+ Identifier fallbackIdentifier = new Identifier("db", "tbl", "fb");
+ Map<String, String> fallbackOptions = queryAuthEnabled();
+ fallbackOptions.put("branch", "fb");
+ FileStoreTable main = newTable(warehouse,
+ new CatalogEnvironment(mainIdentifier, null, () -> catalog,
null, null, true),
+ queryAuthEnabled(), C1);
+ FileStoreTable fallback = newTable(warehouse,
+ new CatalogEnvironment(fallbackIdentifier, null, () ->
catalog, null, null, true),
+ fallbackOptions, C1);
+
+ PaimonScanPlanProvider.authorizeDeferredScan(new
FallbackReadFileStoreTable(main, fallback));
+
+ Assertions.assertEquals(Arrays.asList(mainIdentifier,
fallbackIdentifier), catalog.authCalls,
+ "both branches must be authorized, each against its own
identifier");
+ }
+
+ // ==================== the relation's own scan params ====================
+
+ @Test
+ public void relationOptionsSurviveTheRebuiltSystemTable(@TempDir Path
warehouse) {
+ // The FE applies @options to the wrapper the handle carries
(resolveScanTable ->
+ // PaimonScanParams#applyOptions), but the BE is handed a wrapper
rebuilt over a catalog-less
+ // base - a different object, on which that copy() never ran. Without
re-applying them the
+ // reader that materializes this table's splits falls back to the
unpinned latest state.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable dataTable = newTable(warehouse,
catalogEnvironment(catalog), C1);
+ Map<String, String> pinned =
PaimonScanParams.markAsOptions(Collections.singletonMap(
+ org.apache.paimon.CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
"4"));
+ PaimonTableHandle handle = sysHandle("ro",
dataTable).withScanOptions(pinned);
+
+ Table forBackend = provider().tableForBackend(handle,
+ PaimonScanParams.applyOptions(SystemTableLoader.load("ro",
dataTable), pinned));
+
+ // $ro delegates options() to the data table it wraps, so this reads
the rebuilt base.
+ Assertions.assertEquals("4", forBackend.options()
+
.get(org.apache.paimon.CoreOptions.SCAN_MANIFEST_PARALLELISM.key()));
+ }
+
+ @Test
+ public void incrementalRangeIsResolvedOnTheCatalogVisibleSnapshot(@TempDir
Path warehouse)
+ throws Exception {
+ // Paimon selects the incremental scanner from incremental-between*,
so the snapshot pin
+ // cannot bound this scan - and isolateIncrementalRead clears it
anyway. The timestamp form
+ // would then resolve its endpoints inside the BE reader, through
+ // SnapshotManager#earlierOrEqualTimeMills, whose search runs up to
latestSnapshotId(): the
+ // snapshot directory once the loader is gone. Resolve them here
instead.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ FileStoreTable onDisk = commit(newRealTable(warehouse, "t"), 2);
+ catalog.pointAt(onDisk, 1L);
+ FileStoreTable versionManaged = withCatalogEnvironment(onDisk,
catalogEnvironment(catalog));
+ long firstCommitMillis =
onDisk.snapshotManager().snapshot(1L).timeMillis();
+
+ Map<String, String> unbounded = new HashMap<>();
+ unbounded.put("incremental-between-timestamp", firstCommitMillis + ","
+ Long.MAX_VALUE);
+ Map<String, String> bound =
+ PaimonIncrementalScanParams.bindRangeToCatalog(unbounded,
versionManaged);
+
+ // The range closes on the catalog's snapshot 1, never on snapshot 2
sitting on disk. The
+ // start is the snapshot at or before that wall clock - snapshot 1
itself here - exactly what
+ // IncrementalDeltaStartingScanner#betweenTimestamps would have
resolved on the BE.
+ Assertions.assertEquals("1,1", bound.get("incremental-between"));
+ // Cleared rather than dropped: Paimon's copy() removes a key only
when it maps to null.
+
Assertions.assertTrue(bound.containsKey("incremental-between-timestamp"));
+ Assertions.assertNull(bound.get("incremental-between-timestamp"));
+
+ // And a range that opens BEFORE the earliest snapshot keeps paimon's
exclusive-start rule:
+ // the id before the earliest one, so the earliest snapshot itself is
included.
+ Map<String, String> fromTheStart = new HashMap<>();
+ fromTheStart.put("incremental-between-timestamp", (firstCommitMillis -
1) + "," + Long.MAX_VALUE);
+ Assertions.assertEquals("0,1", PaimonIncrementalScanParams
+ .bindRangeToCatalog(fromTheStart,
versionManaged).get("incremental-between"));
+ }
+
+ @Test
+ public void
incrementalRangeAlreadyOlderThanTheCatalogSnapshotIsLeftAlone(@TempDir Path
warehouse)
+ throws Exception {
+ // An end older than the catalog's snapshot already resolves to the
same id on both sides,
+ // because every snapshot the catalog has not published yet is younger
than it.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ FileStoreTable onDisk = commit(newRealTable(warehouse, "t"), 2);
+ catalog.pointAt(onDisk, 2L);
+ FileStoreTable versionManaged = withCatalogEnvironment(onDisk,
catalogEnvironment(catalog));
+
+ Map<String, String> past = new HashMap<>();
+ past.put("incremental-between-timestamp",
+ "0," + (onDisk.snapshotManager().snapshot(2L).timeMillis() -
1));
+
+ Assertions.assertSame(past,
+ PaimonIncrementalScanParams.bindRangeToCatalog(past,
versionManaged));
+ }
+
+ @Test
+ public void incrementalRangeIsLeftAloneForAFilesystemCatalog(@TempDir Path
warehouse) throws Exception {
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
false);
+ FileStoreTable plain = withCatalogEnvironment(
+ commit(newRealTable(warehouse, "t"), 1),
catalogEnvironment(catalog));
+
+ Map<String, String> range = new HashMap<>();
+ range.put("incremental-between-timestamp", "0," + Long.MAX_VALUE);
+
+ Assertions.assertSame(range,
PaimonIncrementalScanParams.bindRangeToCatalog(range, plain));
+ }
+
+ @Test
+ public void incrementalRangeIsNotBoundOnAFallbackBranchPair(@TempDir Path
warehouse) throws Exception {
+ // The two branches keep independent snapshot id sequences, and
+ // FallbackReadFileStoreTable#rewriteFallbackOptions translates only
scan.snapshot-id, so
+ // incremental-between reaches the fallback branch verbatim. A
main-branch id range bound here
+ // would be validated against the fallback branch's own range in
+ // IncrementalDeltaStartingScanner#betweenSnapshotIds and either fail
out of range or select
+ // unrelated commits. The timestamp form is branch-agnostic and each
branch resolves it
+ // against its own SnapshotManager, so it has to reach the BE
untouched.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ FileStoreTable onDisk = commit(newRealTable(warehouse, "t"), 2);
+ catalog.pointAt(onDisk, 1L);
+ // Stubbed so that the main branch alone WOULD have produced a range:
without the guard this
+ // test sees "0,1" written into incremental-between, not an unchanged
map.
+ FileStoreTable main = withCatalogEnvironment(onDisk,
catalogEnvironment(catalog));
+ Map<String, String> fallbackOptions = new HashMap<>();
+ fallbackOptions.put("branch", "fb");
+ FileStoreTable fallback = newTable(warehouse,
catalogEnvironment(catalog), fallbackOptions, C1);
+ FileStoreTable pair = new FallbackReadFileStoreTable(main, fallback);
+
+ Map<String, String> range = new HashMap<>();
+ range.put("incremental-between-timestamp",
+ onDisk.snapshotManager().snapshot(1L).timeMillis() + "," +
Long.MAX_VALUE);
+
+ Assertions.assertSame(range,
PaimonIncrementalScanParams.bindRangeToCatalog(range, pair));
+ // And through the decorator PrivilegedCatalog adds, since that is the
shape Doris receives.
+ Assertions.assertSame(range,
PaimonIncrementalScanParams.bindRangeToCatalog(range,
+ PrivilegedFileStoreTable.wrap(pair, new
AllGrantedPrivilegeChecker(),
+ Identifier.create("db", "tbl"))));
+ // Not merely "the output equals the input": the main branch's
snapshots must never be read,
+ // because reading them is what produces an id range that cannot be
carried to the fallback.
+ Assertions.assertEquals(0, catalog.loadSnapshotCalls);
+ }
+
+ @Test
+ public void
incrementalPartitionsScanBindsItsRangeBeforeTheBackend(@TempDir Path warehouse)
+ throws Exception {
+ // $partitions is the one system table that both re-plans on the BE
(PartitionsRead
+ // #createReader -> newScan().listPartitionEntries()) and accepts
@incr, so it is the one that
+ // has to reach the BE with an already-resolved range.
+ VersionManagedCatalog catalog = new VersionManagedCatalog(warehouse,
true);
+ FileStoreTable onDisk = commit(newRealTable(warehouse, "t"), 2);
+ catalog.pointAt(onDisk, 1L);
+ FileStoreTable versionManaged = withCatalogEnvironment(onDisk,
catalogEnvironment(catalog));
+ Map<String, String> incremental = new HashMap<>();
+ incremental.put("incremental-between-timestamp",
+ onDisk.snapshotManager().snapshot(1L).timeMillis() + "," +
Long.MAX_VALUE);
+ PaimonTableHandle handle = sysHandle("partitions",
versionManaged).withScanOptions(incremental);
+
+ provider().tableForBackend(handle,
+ SystemTableLoader.load("partitions",
versionManaged).copy(incremental));
+
+ // The range was closed here, on the catalog's snapshot, instead of
inside the BE reader:
+ // pinCatalogSnapshot reads the pointer once and bindRangeToCatalog
reads it again.
+ Assertions.assertTrue(catalog.loadSnapshotCalls >= 2,
+ "an @incr $partitions read must resolve its endpoints against
the catalog");
+ }
+
+ // ==================== fixtures ====================
+
+ private static PaimonTableHandle dataHandle() {
+ return new PaimonTableHandle("db", "tbl", Collections.emptyList(),
Collections.emptyList());
+ }
+
+ private static PaimonTableHandle sysHandle(String sysTableType,
FileStoreTable base) {
+ PaimonTableHandle handle = PaimonTableHandle.forSystemTable("db",
"tbl", sysTableType,
+ PaimonScanParams.requiresPaimonReader(sysTableType));
+ handle.setSysBaseTable(base);
+ return handle;
+ }
+
+ private static Map<String, String> queryAuthEnabled() {
+ Map<String, String> options = new HashMap<>();
+ options.put("query-auth.enabled", "true");
+ return options;
+ }
+
+ private static CatalogEnvironment catalogEnvironment(VersionManagedCatalog
catalog) {
+ return new CatalogEnvironment(Identifier.create("db", "tbl"), null, ()
-> catalog, null, null,
+ catalog.supportsVersionManagement());
+ }
+
+ /** A table object with the given schema; no files are written, so nothing
touches the disk. */
+ private static FileStoreTable newTable(Path warehouse, CatalogEnvironment
env, DataField... fields) {
+ return newTable(warehouse, env, new HashMap<>(), fields);
+ }
+
+ private static FileStoreTable newTable(Path warehouse, CatalogEnvironment
env,
+ Map<String, String> options, DataField... fields) {
+ List<DataField> fieldList = Arrays.asList(fields);
+ TableSchema schema = new TableSchema(0L, fieldList, fieldList.size() -
1,
+ Collections.emptyList(), Collections.emptyList(), options, "");
+ return FileStoreTableFactory.create(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path("file://" + warehouse +
"/db.db/tbl"), 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 =
+ new org.apache.paimon.fs.Path("file://" + warehouse +
"/db.db/" + name);
+ LocalFileIO fileIO = LocalFileIO.create();
+ new SchemaManager(fileIO, tablePath).createTable(new Schema(
+ Collections.singletonList(C1), Collections.emptyList(),
Collections.emptyList(),
+ Collections.emptyMap(), ""));
+ return FileStoreTableFactory.create(fileIO, tablePath);
+ }
+
+ private static FileStoreTable commit(FileStoreTable table, int snapshots)
throws Exception {
+ for (int i = 0; i < snapshots; i++) {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(i));
+ commit.commit(write.prepareCommit());
+ }
+ }
+ return table;
+ }
+
+ /** Re-binds an on-disk table to a catalog environment, the way a catalog
load would. */
+ private static FileStoreTable withCatalogEnvironment(FileStoreTable table,
CatalogEnvironment env) {
+ return
FileStoreTableFactory.createWithoutFallbackBranch(table.fileIO(),
table.location(),
+ table.schema(), new Options(), env);
+ }
+
+ /**
+ * The {@code {main, fallback}} pair the FE captured (M1 / F1), with a
newer fallback generation
+ * (F2, one column wider) already published on the filesystem - so a
rebuild that re-reads the
+ * branch instead of reusing the captured object shows up in the row type.
+ */
+ private static FileStoreTable[] fallbackPairWithNewerGenerationOnDisk(Path
warehouse, String name)
+ throws Exception {
+ org.apache.paimon.fs.Path tablePath =
+ new org.apache.paimon.fs.Path("file://" + warehouse +
"/db.db/" + name);
+ // F2: the generation external DDL has already published on the
fallback branch.
+ new SchemaManager(LocalFileIO.create(), tablePath,
"fb").createTable(new Schema(
+ Arrays.asList(C1, C2), Collections.emptyList(),
Collections.emptyList(),
+ Collections.emptyMap(), ""));
+
+ // F1 / M1: what the FE captured and planned this query with.
+ Map<String, String> mainOptions = new HashMap<>();
+ mainOptions.put("scan.fallback-branch", "fb");
+ Map<String, String> fallbackOptions = new HashMap<>();
+ fallbackOptions.put("branch", "fb");
+ return new FileStoreTable[] {
+ branchTable(warehouse, tablePath, mainOptions),
+ branchTable(warehouse, tablePath, fallbackOptions)};
+ }
+
+ /**
+ * One branch of a {@code scan.fallback-branch} pair, built the way Paimon
builds them: without
+ * re-expanding the fallback branch, so the caller can wrap the two itself.
+ */
+ private static FileStoreTable branchTable(Path warehouse,
org.apache.paimon.fs.Path tablePath,
+ Map<String, String> options) {
+ TableSchema schema = new TableSchema(0L,
Collections.singletonList(C1), 0,
+ Collections.emptyList(), Collections.emptyList(), options, "");
+ return
FileStoreTableFactory.createWithoutFallbackBranch(LocalFileIO.create(),
tablePath,
+ schema, new Options(),
+ catalogEnvironment(new VersionManagedCatalog(warehouse,
false)));
+ }
+
+ private static void assertPairMatchesTheFeGeneration(FileStoreTable
forBackend, FileStoreTable main,
+ FileStoreTable fallback) {
+ Assertions.assertTrue(forBackend instanceof FallbackReadFileStoreTable,
+ "the pair itself must reach the BE, or a fallback split has no
reader there");
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
forBackend;
+ // The fallback branch must stay on F1 - not the c2 generation sitting
on the filesystem.
+ Assertions.assertEquals(fallback.rowType(), pair.fallback().rowType());
+ Assertions.assertEquals(main.rowType(), pair.wrapped().rowType());
+ // And neither branch may carry the loader that drags the metastore
stack onto the BE.
+
Assertions.assertNull(pair.wrapped().catalogEnvironment().catalogLoader());
+
Assertions.assertNull(pair.fallback().catalogEnvironment().catalogLoader());
+ }
+
+ /** Decodes {@code paimon.serialized_table} the way {@code
PaimonJniScanner#initTable} does. */
+ private static FileStoreTable deserializeTable(String encoded) throws
Exception {
+ byte[] raw = encoded.getBytes(StandardCharsets.UTF_8);
+ byte[] bytes;
+ try {
+ bytes = Base64.getUrlDecoder().decode(raw);
+ } catch (IllegalArgumentException urlReject) {
+ bytes = Base64.getDecoder().decode(raw);
+ }
+ return InstantiationUtil.deserializeObject(bytes,
+ PaimonBackendBoundTableTest.class.getClassLoader());
+ }
+
+ private static String rootCauseMessage(Throwable t) {
+ Throwable cause = t;
+ StringBuilder messages = new StringBuilder();
+ while (cause != null) {
+ messages.append(cause.getMessage()).append('\n');
+ cause = cause.getCause();
+ }
+ return messages.toString();
+ }
+
+ /**
+ * A real {@link FileSystemCatalog} that additionally answers the two
calls the BE would have made
+ * through the catalog loader: {@code loadSnapshot} (the
committed-snapshot pointer a
+ * version-managed catalog owns, which the test sets independently of the
snapshot directory) and
+ * {@code authTableQuery} (recorded, and optionally denied). Both are
{@code
+ * UnsupportedOperationException} on the base class, so overriding them is
what makes an offline
+ * version-managed catalog possible at all.
+ */
+ private static final class VersionManagedCatalog extends FileSystemCatalog
{
+
+ private final boolean supportsVersionManagement;
+ private final Map<String, TableSnapshot> pointers = new HashMap<>();
+ private final List<Identifier> authCalls = new ArrayList<>();
+ private String authDenialMessage;
+ private int loadSnapshotCalls;
+
+ private VersionManagedCatalog(Path warehouse, boolean
supportsVersionManagement) {
+ super(LocalFileIO.create(), new
org.apache.paimon.fs.Path("file://" + warehouse));
+ this.supportsVersionManagement = supportsVersionManagement;
+ }
+
+ /** Publishes {@code snapshotId} as the committed pointer, whatever
the directory holds. */
+ private void pointAt(FileStoreTable table, long snapshotId) {
+ Snapshot snapshot = table.snapshotManager().snapshot(snapshotId);
+ pointers.put(table.location().getName(), new
TableSnapshot(snapshot, 0L, 0L, 0L, 0L));
+ }
+
+ private void denyAuthWith(String message) {
+ this.authDenialMessage = message;
+ }
+
+ @Override
+ public boolean supportsVersionManagement() {
+ return supportsVersionManagement;
+ }
+
+ @Override
+ public Optional<TableSnapshot> loadSnapshot(Identifier identifier) {
+ loadSnapshotCalls++;
+ // Keyed by the on-disk table directory rather than the
Identifier: the fixtures bind one
+ // catalog to tables built straight from a path, so the Identifier
is the same for all.
+ return pointers.isEmpty()
+ ? Optional.empty()
+ : Optional.of(pointers.values().iterator().next());
+ }
+
+ @Override
+ public List<String> authTableQuery(Identifier identifier, List<String>
select) {
+ authCalls.add(identifier);
+ if (authDenialMessage != null) {
+ throw new RuntimeException(authDenialMessage);
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void close() {
+ // CatalogEnvironment#tableQueryAuth and SnapshotLoaderImpl#load
both close the catalog
+ // they load; this instance is shared across those calls, so
closing must not disarm it.
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
