github-actions[bot] commented on code in PR #65867:
URL: https://github.com/apache/doris/pull/65867#discussion_r3652926090
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -190,6 +196,149 @@ 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 re-plan on the BE
({@link
+ * #replansOnBackend}) what the catalog would have done there is
done here instead: see
+ * {@link #authorizeDeferredScan} and {@link #pinCatalogSnapshot}.
Every other system
+ * table reads the splits 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 (replansOnBackend(sysTableType)) {
Review Comment:
[P1] Pin every metadata table that selects its snapshot on BE
`replansOnBackend` is too narrow: `$manifests`, `$statistics`, and
`$table_indexes` also send only singleton marker splits from FE and choose
their snapshot inside the BE reader. In Paimon 1.3.1, `ManifestsRead` and
`IndexesRead` call `TimeTravelUtil.tryTravelOrLatest(dataTable)`, while
`StatisticRead` calls `dataTable.statistics()`, which uses the same helper.
Since this branch leaves them unpinned after removing the loader, a REST/DLF
catalog that still points to N while snapshot N+1 is already on the filesystem
(or points back to N after rollback) returns unpublished/rollback-retained
metadata from N+1.
This is distinct from the existing `$snapshots`/`$buckets` thread: those
readers ignore `scan.snapshot-id`, whereas these three honor it but never
receive it. Please classify every fixed-row system table that selects a
snapshot during BE reading separately from `replansOnBackend`, pin these three
to the catalog-visible snapshot, and cover publication/rollback cases.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -190,6 +196,149 @@ 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 re-plan on the BE
({@link
+ * #replansOnBackend}) what the catalog would have done there is
done here instead: see
+ * {@link #authorizeDeferredScan} and {@link #pinCatalogSnapshot}.
Every other system
+ * table reads the splits 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 (replansOnBackend(sysTableType)) {
+ baseForBackend = pinCatalogSnapshot(baseForBackend, dataTable);
+ }
+ Table catalogLessSysTable = SystemTableLoader.load(sysTableType,
baseForBackend);
+ return catalogLessSysTable == null ? paimonTable : catalogLessSysTable;
+ }
+
+ /**
+ * {@code $files} and {@code $partitions} are the only system tables that
do not read what the FE
+ * planned: they carry partition-level splits and call {@code
FileStoreTable#newScan} again on
+ * the BE ({@code FilesTable.FilesSplit#splits},
+ * {@code PartitionsTable.PartitionsRead#createReader}).
+ */
+ private static boolean replansOnBackend(String sysTableType) {
+ return PAIMON_FILES_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)
+ ||
PAIMON_PARTITIONS_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 #replansOnBackend} 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". {@code $files} / {@code $partitions} 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.
+ */
+ private static FileStoreTable dropCatalogLoader(FileStoreTable dataTable) {
+ if (dataTable.catalogEnvironment().catalogLoader() == null) {
+ return dataTable;
+ }
+ return FileStoreTableFactory.create(
Review Comment:
[P1] Preserve the captured fallback-table generation
When `dataTable` is a `FallbackReadFileStoreTable`, `schema()` exposes only
its wrapped/main schema. `FileStoreTableFactory.create(...)` sees the retained
`scan.fallback-branch` option and independently calls `SchemaManager(...,
fallbackBranch).latest()` instead of reusing the fallback object already
captured by FE. For example, after the cache captures compatible main/fallback
M1/F1 and external DDL publishes M2/F2, this serializes a hybrid M1/F2 while FE
schema and split planning still use M1/F1. `$files`/`$partitions` then reach
`FallbackReadFileStoreTable.validateSchema()` on BE and fail on the mismatched
row types.
This is separate from the existing fallback snapshot-bound thread: the
schema generation is reread here before pin translation. Please rebuild both
the captured wrapped and fallback objects from their exact schemas/options
while removing their loaders, and add a cache-skew schema-evolution test.
--
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]
