morningman commented on code in PR #65867:
URL: https://github.com/apache/doris/pull/65867#discussion_r3654831386
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -190,6 +202,182 @@ protected void doInitialize() throws UserException {
}
}
+ /**
+ * 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
org.apache.paimon.table.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(...)}, 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, and for the system tables that pick their snapshot on
the BE ({@link
+ * #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.</li>
+ * </ul>
+ */
+ private Table getPaimonTableForBackend() {
+ Table paimonTable = source.getPaimonTable();
+ if (paimonTable instanceof FileStoreTable) {
+ return dropCatalogLoader((FileStoreTable) paimonTable);
+ }
+ if (!(source.getExternalTable() instanceof PaimonSysExternalTable)) {
+ return paimonTable;
+ }
+ PaimonSysExternalTable sysTable = (PaimonSysExternalTable)
source.getExternalTable();
+ // 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 = sysTable.getSysBaseTable();
+ if (dataTable == null) {
+ return paimonTable;
+ }
+ String sysTableType = sysTable.getSysTableType();
+ if (PAIMON_FILES_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)) {
+ authorizeDeferredScan(dataTable);
+ }
+ FileStoreTable baseForBackend = dropCatalogLoader(dataTable);
+ if (resolvesSnapshotOnBackend(sysTableType)) {
+ baseForBackend = pinCatalogSnapshot(baseForBackend, dataTable);
+ }
+ Table catalogLessSysTable = SystemTableLoader.load(sysTableType,
baseForBackend);
+ return catalogLessSysTable == null ? paimonTable : catalogLessSysTable;
+ }
+
+ /**
+ * 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 {@link #pinCatalogSnapshot} cannot
rewind the BE schema.
+ */
+ private static boolean resolvesSnapshotOnBackend(String sysTableType) {
+ return PAIMON_FILES_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)
+ ||
PAIMON_PARTITIONS_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)
+ ||
PAIMON_MANIFESTS_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)
+ ||
PAIMON_STATISTICS_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)
+ ||
PAIMON_TABLE_INDEXES_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType);
+ }
+
+ /**
+ * {@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, so it has no
+ * authorization to transfer.
+ */
+ private static void authorizeDeferredScan(FileStoreTable dataTable) {
+ CoreOptions options = dataTable.coreOptions();
+ if (options.queryAuthEnabled()) {
+ dataTable.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 #resolvesSnapshotOnBackend} system tables. {@code
PaimonJniScanner#initTable}
+ * unconditionally calls {@code table.copy(table.options())}, and every
Paimon system-table
+ * wrapper delegates {@code copy} to {@code FileStoreTable#copy}, which
time-travels the schema
+ * to {@code scan.snapshot-id}. For a wrapper whose row type follows the
base table
+ * ({@code $ro}, {@code $row_tracking}, {@code $audit_log}, {@code
$binlog}) that would rewind
+ * the BE schema: a column added after the latest snapshot is planned by
the FE and then
+ * rejected by {@code PaimonJniScanner#getProjected} with "RequiredField
... not found in
+ * schema". The pinned ones all have a fixed row type, so the pin is safe
there.
+ *
+ * <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: on a table with {@code scan.fallback-branch} this bounds
the main branch only.
+ * {@code FallbackReadFileStoreTable#copyWithoutTimeTravel} derives the
fallback branch's own
+ * bound from the pin ({@code rewriteFallbackOptions} ->
+ * {@code SnapshotManager#earlierOrEqualTimeMills}), and that branch has
no catalog loader left
+ * either, so its bound comes from the snapshot directory too - same
transient window on the same
+ * read-only metadata tables.
+ */
+ private 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(PAIMON_SCAN_SNAPSHOT_ID,
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.
+ */
+ private static FileStoreTable dropCatalogLoader(FileStoreTable dataTable) {
+ if (dataTable.catalogEnvironment().catalogLoader() == null) {
+ return dataTable;
+ }
+ if (dataTable instanceof FallbackReadFileStoreTable) {
Review Comment:
Confirmed and fixed in e63a97be480 (`unwrapToFallbackOrBase`).
Verified the reachability chain against paimon `release-1.3.1` rather than
assuming it: `CatalogFactory#createCatalog:74` calls
`PrivilegedCatalog.tryToCreate(..)`, and all five Doris
`Paimon*MetaStoreProperties` build their catalog through that entry point;
`PrivilegedCatalog#getTable:151-159` then wraps any `FileStoreTable` in
`PrivilegedFileStoreTable`. So with file based privileges enabled the table
does arrive here as `Privileged(FallbackRead(..))` and the direct `instanceof`
looked straight past it.
`dropCatalogLoader` now peels `DelegatedFileStoreTable` layers down to the
fallback pair before matching. The decorators are peeled rather than
re-applied: `PrivilegedFileStoreTable` only asserts on `newScan()` /
`newRead()`, which the FE has already run while planning the query, and a plain
(non fallback) table already loses the wrapper the same way because
`rebuildWithoutCatalogLoader` builds a bare table out of fileIO / location /
schema. `PrivilegedFileStoreTable` overrides none of `catalogEnvironment()` /
`schema()` / `fileIO()` / `location()`, so the rebuild reads the same values
through the delegation. `PrivilegedFileStoreTable` and
`FallbackReadFileStoreTable` are the only two `DelegatedFileStoreTable`
subclasses in paimon-core (`LookupFileStoreTable` is flink-only), so the loop
terminates on the base table in every other case.
Covered by
`PaimonScanNodeTest#testFallbackBranchSurvivesAPaimonTableDecorator`, which
pushes `Privileged(FallbackRead(M1, F1))` through `dropCatalogLoader` with a
newer fallback generation F2 sitting on the filesystem, and asserts the result
is still a `FallbackReadFileStoreTable` carrying M1/F1 with no catalog loader
on either branch. Negative control run: with the peel reverted the test fails
at the `instanceof FallbackReadFileStoreTable` assertion, since the table comes
back rebuilt from the delegated main branch alone. Suite is 22/22 green.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
