github-actions[bot] commented on code in PR #65867:
URL: https://github.com/apache/doris/pull/65867#discussion_r3671163149
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -203,7 +215,310 @@ protected void doInitialize() throws UserException {
private void serializeProcessedTable() throws UserException {
// System-table splits are materialized by the BE JNI reader, so it
must receive the same
// option-bearing table copy that FE uses to plan the split.
- serializedTable = PaimonUtil.encodeObjectToString(getProcessedTable());
+ serializedTable =
PaimonUtil.encodeObjectToString(getPaimonTableForBackend());
+ }
+
+ /**
+ * 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.
+ * The relation-scoped scan params the FE applied to the original
wrapper are re-applied
+ * to the rebuilt one by {@link #reapplyScanParams(Table)}.</li>
+ * </ul>
+ */
+ private Table getPaimonTableForBackend() throws UserException {
+ Table paimonTable = getProcessedTable();
+ if (paimonTable instanceof FileStoreTable) {
+ // copy(...) merges 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) 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();
+ boolean resolvesOnBackend = resolvesSnapshotOnBackend(sysTableType);
+ if (PAIMON_FILES_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)) {
+ authorizeDeferredScan(dataTable);
+ }
+ FileStoreTable baseForBackend = dropCatalogLoader(dataTable);
+ if (resolvesOnBackend) {
+ baseForBackend = pinCatalogSnapshot(baseForBackend, dataTable);
+ }
+ Table catalogLessSysTable = SystemTableLoader.load(sysTableType,
baseForBackend);
+ if (catalogLessSysTable == null) {
+ return paimonTable;
+ }
+ return reapplyScanParams(catalogLessSysTable, dataTable,
resolvesOnBackend);
+ }
+
+ /**
+ * Re-apply the relation-scoped scan params to a rebuilt system-table
wrapper.
+ *
+ * <p>{@link #getProcessedTable()} applies {@code @incr} / {@code
@options} to the wrapper the
+ * meta cache holds, 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.
+ *
+ * <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
#bindIncrementalRangeToCatalog}.
+ */
+ private Table reapplyScanParams(Table rebuiltSysTable, FileStoreTable
dataTable, boolean resolvesOnBackend)
+ throws UserException {
+ TableScanParams theScanParams = getScanParams();
+ if (theScanParams == null) {
+ return rebuiltSysTable;
+ }
+ if (theScanParams.incrementalRead()) {
+ Map<String, String> incrementalParams = getIncrReadParams();
+ if (resolvesOnBackend) {
+ incrementalParams =
bindIncrementalRangeToCatalog(incrementalParams, dataTable);
+ }
+ return rebuiltSysTable.copy(incrementalParams);
+ }
+ if (theScanParams.isOptions()) {
+ return PaimonScanParams.applyOptions(rebuiltSysTable,
+
theScanParams.getResolvedMapParams().orElse(Collections.emptyMap()));
+ }
+ return rebuiltSysTable;
+ }
+
+ /**
+ * Bind an incremental relation's range to the catalog-visible snapshot.
+ *
+ * <p>Only for {@link #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 {@link #pinCatalogSnapshot}'s pin cannot bound this
scan - and
+ * {@code 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.
+ */
+ private static Map<String, String> bindIncrementalRangeToCatalog(
+ Map<String, String> incrementalParams, FileStoreTable dataTable) {
+ String range =
incrementalParams.get(PAIMON_INCREMENTAL_BETWEEN_TIMESTAMP);
+ if (range == null ||
!dataTable.catalogEnvironment().supportsVersionManagement()) {
+ return incrementalParams;
+ }
+ SnapshotManager snapshotManager = dataTable.snapshotManager();
Review Comment:
[P1] Bind this incremental range against each fallback branch
This helper makes both the rewrite decision and endpoint resolution through
`dataTable.snapshotManager()`, which a `FallbackReadFileStoreTable` delegates
to its main branch. If the requested end is older than the main latest, line
352 leaves the timestamp range unchanged even when it extends past the fallback
catalog pointer; after loader removal, that branch can include an unpublished
or rollback-retained snapshot. If this method does rewrite, Paimon's fallback
option rewrite copies the resulting main `incremental-between` IDs unchanged
because it translates only `scan.snapshot-id`, so disjoint branch histories
fail out of range and overlapping IDs select the wrong commits. Please
resolve/cap the range separately with both captured branch managers (or reject
this composition), and test different branch IDs and catalog times.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -203,7 +215,310 @@ protected void doInitialize() throws UserException {
private void serializeProcessedTable() throws UserException {
// System-table splits are materialized by the BE JNI reader, so it
must receive the same
// option-bearing table copy that FE uses to plan the split.
- serializedTable = PaimonUtil.encodeObjectToString(getProcessedTable());
+ serializedTable =
PaimonUtil.encodeObjectToString(getPaimonTableForBackend());
+ }
+
+ /**
+ * 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.
+ * The relation-scoped scan params the FE applied to the original
wrapper are re-applied
+ * to the rebuilt one by {@link #reapplyScanParams(Table)}.</li>
+ * </ul>
+ */
+ private Table getPaimonTableForBackend() throws UserException {
+ Table paimonTable = getProcessedTable();
+ if (paimonTable instanceof FileStoreTable) {
+ // copy(...) merges 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) 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();
+ boolean resolvesOnBackend = resolvesSnapshotOnBackend(sysTableType);
+ if (PAIMON_FILES_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)) {
+ authorizeDeferredScan(dataTable);
+ }
+ FileStoreTable baseForBackend = dropCatalogLoader(dataTable);
+ if (resolvesOnBackend) {
+ baseForBackend = pinCatalogSnapshot(baseForBackend, dataTable);
+ }
+ Table catalogLessSysTable = SystemTableLoader.load(sysTableType,
baseForBackend);
+ if (catalogLessSysTable == null) {
+ return paimonTable;
+ }
+ return reapplyScanParams(catalogLessSysTable, dataTable,
resolvesOnBackend);
+ }
+
+ /**
+ * Re-apply the relation-scoped scan params to a rebuilt system-table
wrapper.
+ *
+ * <p>{@link #getProcessedTable()} applies {@code @incr} / {@code
@options} to the wrapper the
+ * meta cache holds, 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.
+ *
+ * <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
#bindIncrementalRangeToCatalog}.
+ */
+ private Table reapplyScanParams(Table rebuiltSysTable, FileStoreTable
dataTable, boolean resolvesOnBackend)
+ throws UserException {
+ TableScanParams theScanParams = getScanParams();
+ if (theScanParams == null) {
+ return rebuiltSysTable;
+ }
+ if (theScanParams.incrementalRead()) {
+ Map<String, String> incrementalParams = getIncrReadParams();
+ if (resolvesOnBackend) {
+ incrementalParams =
bindIncrementalRangeToCatalog(incrementalParams, dataTable);
+ }
+ return rebuiltSysTable.copy(incrementalParams);
+ }
+ if (theScanParams.isOptions()) {
+ return PaimonScanParams.applyOptions(rebuiltSysTable,
Review Comment:
[P1] Preserve the fallback selector already resolved on FE
The processed FE wrapper has already applied this OPTIONS selector to the
catalog-backed fallback pair, where Paimon maps the main snapshot to the
fallback branch through that branch's catalog-visible `SnapshotManager`. Here
the code rebuilds an unoptioned pair with `CatalogEnvironment.empty()` and
applies only the main selector again, so `FallbackReadFileStoreTable.copy`
repeats the translation against filesystem state. After fallback
rollback/publication skew, FE can plan F7 while the serialized BE table selects
retained F8; this also reaches OPTIONS-capable FE-planned system tables such as
`$ro`. Please preserve each selected branch before removing its loader, and add
an encode/decode fallback-skew OPTIONS 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]
