CalvinKirs commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3835776735
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +114,293 @@ public Table getPaimonTable(NameMapping nameMapping) {
public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable)
{
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ MetaCacheEntry<NameMapping, PaimonTableCacheValue> tables =
tableEntry.get(nameMapping.getCtlId());
+ PaimonTableCacheValue tableValue = tables.get(nameMapping);
+ PaimonSnapshot fence = loadLatestSnapshotFence(nameMapping,
tableValue).getSnapshot();
+ if (!tables.isEffectivelyEnabled()) {
+ // Projections are keyed by the synthetic generation of a
published table handle. An
+ // ineffective table entry publishes nothing, so nothing keyed by
this load could ever
+ // be looked up again: serve it directly instead of churning the
snapshot entry.
+ return executeForGeneration(tableValue, nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration()))
+ .bindCapturedAuthenticator(tableValue.getAuthenticator());
+ }
+ // Order fence observations, not snapshot ids: a rollback moves the
latest snapshot
+ // backwards, and a concurrent call may finish after a later
observation (reversed
+ // completion). Either way the most recently observed fence is the one
future lookups read.
+ long observation = fenceObservations.incrementAndGet();
+ PaimonSnapshotEntryKey key = PaimonSnapshotEntryKey.of(
+ nameMapping, fence, tableValue.getGeneration());
+ MetaCacheEntry<PaimonSnapshotEntryKey, PaimonSnapshotCacheValue> entry
=
+ snapshotEntry.get(nameMapping.getCtlId());
+ AtomicBoolean loaded = new AtomicBoolean();
+ PaimonSnapshotCacheValue snapshotValue = entry.get(key,
+ ignored -> executeForGeneration(tableValue, nameMapping, () ->
{
+ loaded.set(true);
+ return latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration())
+
.bindCapturedAuthenticator(tableValue.getAuthenticator());
+ }));
+ LatestFenceOwner owner = new LatestFenceOwner(nameMapping,
tableValue.getGeneration());
+ ObservedFence latest = latestObservedFences.compute(owner, (ignored,
current) ->
+ current == null || current.observation < observation ? new
ObservedFence(observation, key) : current);
+ if (loaded.get()) {
+ retireSupersededLatestProjections(entry, owner, latest.key);
+ }
+ if (!isCurrentTableGeneration(nameMapping,
tableValue.getGeneration())) {
+ entry.invalidateKeyIfSame(key, snapshotValue);
+ // A generation that is not published (rejected admission,
replaced or invalidated
+ // mid-load) can never be observed again; drop the owner this call
registered so
+ // persistently rejected tables cannot grow the map, and so a
delayed old-generation
+ // load cannot resurrect an owner that catalog cleanup already
removed.
+ latestObservedFences.remove(owner);
+ }
+ return snapshotValue;
+ }
+
+ /**
+ * Only the projection of the most recently observed latest fence of a
table generation is
+ * reachable: every later call re-reads the fence and looks up that key.
After a load, retire
+ * every other projection of the generation, including this load itself
when a concurrent call
+ * observed a later fence and finished first, so a busy table never
accumulates projections.
+ */
+ private static void retireSupersededLatestProjections(
+ MetaCacheEntry<PaimonSnapshotEntryKey, PaimonSnapshotCacheValue>
entry,
+ LatestFenceOwner owner, PaimonSnapshotEntryKey latestKey) {
+ entry.invalidateIf(key -> owner.owns(key) && !key.equals(latestKey));
+ }
+
+ private void forgetObservedFences(NameMapping nameMapping,
java.util.function.LongPredicate retiredGeneration) {
+ latestObservedFences.keySet().removeIf(owner ->
owner.nameMapping.equals(nameMapping)
+ && retiredGeneration.test(owner.generation));
+ }
+
+ private static final class LatestFenceOwner {
+ private final NameMapping nameMapping;
+ private final long generation;
+
+ private LatestFenceOwner(NameMapping nameMapping, long generation) {
+ this.nameMapping = nameMapping;
+ this.generation = generation;
+ }
+
+ private boolean owns(PaimonSnapshotEntryKey key) {
+ return key.getTableGeneration() == generation &&
key.getNameMapping().equals(nameMapping);
+ }
+
+ @Override
+ public boolean equals(Object object) {
+ if (!(object instanceof LatestFenceOwner)) {
+ return false;
+ }
+ LatestFenceOwner that = (LatestFenceOwner) object;
+ return generation == that.generation &&
nameMapping.equals(that.nameMapping);
+ }
+
+ @Override
+ public int hashCode() {
+ return java.util.Objects.hash(nameMapping, generation);
+ }
+ }
+
+ private static final class ObservedFence {
+ private final long observation;
+ private final PaimonSnapshotEntryKey key;
+
+ private ObservedFence(long observation, PaimonSnapshotEntryKey key) {
+ this.observation = observation;
+ this.key = key;
+ }
+ }
+
+ /**
+ * Load a projection for a relation-scoped effective table. The caller
derived
+ * {@code effectiveTable} from {@code tableGeneration}'s base handle, so
the load and any later
+ * hydration of the returned value must run under that generation's
captured execution context:
+ * re-resolving the catalog here could pair the retained handle with the
resources of a
+ * concurrent property/credential ALTER, or fail while reset closes the
old generation.
+ */
+ public PaimonSnapshotCacheValue loadSnapshotProjection(ExternalTable
dorisTable, Table effectiveTable,
+ PaimonTableCacheValue tableGeneration) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return executeForGeneration(tableGeneration, nameMapping,
+ () -> latestSnapshotProjectionLoader.load(nameMapping,
effectiveTable))
+ .bindCapturedAuthenticator(tableGeneration.getAuthenticator());
}
- public PaimonSnapshotCacheValue loadSnapshotProjection(ExternalTable
dorisTable, Table effectiveTable) {
- return
latestSnapshotProjectionLoader.load(dorisTable.getOrBuildNameMapping(),
effectiveTable);
+ /** Resolve the current table generation, exposing the handle and its
captured context together. */
+ public PaimonTableCacheValue getTableCacheValue(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
}
public PaimonSnapshotCacheValue loadLatestSnapshotFence(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- Table table =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getPaimonTable();
- return latestSnapshotProjectionLoader.loadFence(nameMapping, table);
+ PaimonTableCacheValue tableValue =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ return loadLatestSnapshotFence(nameMapping, tableValue);
}
+ /**
+ * Hydrate a statement fence into a full projection. The fence retains the
physical table of
+ * the generation it was captured from, so hydration runs under the
fence's captured execution
+ * context instead of re-resolving the catalog's current one, and the
returned value carries
+ * the same context forward for later schema hydration.
+ */
public PaimonSnapshotCacheValue loadSnapshotAtFence(
- ExternalTable dorisTable, PaimonSnapshot fence) {
+ ExternalTable dorisTable, PaimonSnapshotCacheValue fenceValue) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return latestSnapshotProjectionLoader.loadAtFence(nameMapping, fence);
+ return
executeForCapturedAuthenticator(fenceValue.getCapturedAuthenticator(),
nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(nameMapping,
fenceValue.getSnapshot()))
+
.bindCapturedAuthenticator(fenceValue.getCapturedAuthenticator());
}
+ /** Effective-table sibling of {@link #loadSnapshotAtFence(ExternalTable,
PaimonSnapshotCacheValue)}. */
public PaimonSnapshotCacheValue loadSnapshotAtFence(
- ExternalTable dorisTable, Table effectiveTable, PaimonSnapshot
fence) {
- return latestSnapshotProjectionLoader.loadEffectiveAtFence(
- dorisTable.getOrBuildNameMapping(), effectiveTable, fence);
+ ExternalTable dorisTable, Table effectiveTable,
PaimonSnapshotCacheValue fenceValue) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return
executeForCapturedAuthenticator(fenceValue.getCapturedAuthenticator(),
nameMapping,
+ () -> latestSnapshotProjectionLoader.loadEffectiveAtFence(
+ nameMapping, effectiveTable, fenceValue.getSnapshot()))
+
.bindCapturedAuthenticator(fenceValue.getCapturedAuthenticator());
}
public PaimonSchemaCacheValue getPaimonSchemaCacheValue(NameMapping
nameMapping, long schemaId) {
- SchemaCacheValue schemaCacheValue =
schemaEntry.get(nameMapping.getCtlId())
- .get(new PaimonSchemaCacheKey(nameMapping, schemaId));
+ PaimonTableCacheValue tableValue =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ return getPaimonSchemaCacheValue(nameMapping, schemaId,
tableValue.getGeneration(),
+ tableValue.getPaimonTable(), tableValue.getAuthenticator());
+ }
+
+ PaimonSchemaCacheValue getPaimonSchemaCacheValue(
+ NameMapping nameMapping, long schemaId, long tableGeneration,
Table retainedTable) {
+ return getPaimonSchemaCacheValue(nameMapping, schemaId,
tableGeneration, retainedTable, null);
+ }
+
+ PaimonSchemaCacheValue getPaimonSchemaCacheValue(NameMapping nameMapping,
long schemaId,
+ long tableGeneration, Table retainedTable,
+ @Nullable
org.apache.doris.common.security.authentication.ExecutionAuthenticator
authenticator) {
+ return getPaimonSchemaCacheValueInternal(nameMapping, schemaId,
tableGeneration, retainedTable,
+ load -> executeForCapturedAuthenticator(authenticator,
nameMapping, load));
+ }
+
+ /**
+ * Schema resolution for a projection load that is already running inside
the caller's
+ * captured authenticated context: re-resolving the catalog here could
observe a concurrent
+ * property reset, so the load executes directly.
+ */
+ private PaimonSchemaCacheValue getPaimonSchemaCacheValuePreAuthenticated(
+ NameMapping nameMapping, long schemaId, long tableGeneration,
Table retainedTable) {
+ return getPaimonSchemaCacheValueInternal(nameMapping, schemaId,
tableGeneration, retainedTable,
+ load -> {
+ try {
+ return load.call();
+ } catch (Exception e) {
+ throw new CacheException("failed to load paimon schema
%s.%s.%s: %s",
+ e, nameMapping.getCtlId(),
nameMapping.getLocalDbName(),
+ nameMapping.getLocalTblName(), e.getMessage());
+ }
+ });
+ }
+
+ private PaimonSchemaCacheValue
getPaimonSchemaCacheValueInternal(NameMapping nameMapping, long schemaId,
+ long tableGeneration, Table retainedTable,
+
java.util.function.Function<java.util.concurrent.Callable<SchemaCacheValue>,
+ SchemaCacheValue> executor) {
+ java.util.concurrent.Callable<SchemaCacheValue> load =
+ () -> loadSchemaCacheValue(new PaimonSchemaCacheKey(
+ nameMapping, tableGeneration, schemaId),
retainedTable);
+ java.util.function.Supplier<SchemaCacheValue> authenticated = () ->
executor.apply(load);
+ PaimonSchemaCacheKey key = new PaimonSchemaCacheKey(nameMapping,
tableGeneration, schemaId);
+ if (tableGeneration <= 0L ||
!tableEntry.get(nameMapping.getCtlId()).isEffectivelyEnabled()) {
+ // See getSnapshotCache: without a published table handle no
generation-keyed
+ // projection is reachable again.
+ return (PaimonSchemaCacheValue) authenticated.get();
+ }
+ MetaCacheEntry<PaimonSchemaCacheKey, SchemaCacheValue> entry =
schemaEntry.get(nameMapping.getCtlId());
+ SchemaCacheValue schemaCacheValue = entry.get(key, ignored ->
authenticated.get());
+ if (!isCurrentTableGeneration(nameMapping, tableGeneration)) {
+ entry.invalidateKeyIfSame(key, schemaCacheValue);
+ }
return (PaimonSchemaCacheValue) schemaCacheValue;
}
+ /**
+ * Snapshot and schema projections are keyed by the synthetic generation
of the base table
+ * handle they were derived from. A generation that is no longer published
(replaced, expired,
+ * or never admitted because its weight estimate was rejected) can never
be looked up again,
+ * so its projections must not stay behind in the child entries.
+ */
+ private boolean isCurrentTableGeneration(NameMapping nameMapping, long
tableGeneration) {
+ // Revalidation must never re-prepare a retired catalog group (or
throw out of a lookup
+ // that already holds a valid value): an absent group simply means
"not current".
+ MetaCacheEntry<NameMapping, PaimonTableCacheValue> tables =
+ tableEntry.getIfInitialized(nameMapping.getCtlId());
+ PaimonTableCacheValue currentTable = tables == null ? null :
tables.peekIfPresent(nameMapping);
+ return currentTable != null && currentTable.getGeneration() ==
tableGeneration;
+ }
+
private PaimonTableCacheValue loadTableCacheValue(NameMapping nameMapping)
{
- Table paimonTable = tableLoader.load(nameMapping);
- return new PaimonTableCacheValue(paimonTable,
- () -> latestSnapshotProjectionLoader.load(nameMapping,
paimonTable));
+ try {
+ // One catalog generation must supply the loaded table and the
captured execution
+ // context together: reading the authenticator only after the load
could pair the old
+ // generation's table with the context a concurrent ALTER
installed mid-flight (an
+ // ordinary credential ALTER does not retire this engine's cache
group). Capture
+ // first, load, then re-validate before publication; a mid-flight
reset fails the
+ // miss (the caller retries against the reinitialized catalog)
instead of publishing
+ // a spliced synthetic generation that every later fence and
schema hydration would
+ // deliberately trust.
+ PaimonExternalCatalog catalog = tableLoader.catalog(nameMapping);
+
org.apache.doris.common.security.authentication.ExecutionAuthenticator
authenticator =
+ catalog.getExecutionAuthenticator();
Review Comment:
Fixed in 8cb50adebc0: loadTableCacheValue now runs
catalog.makeSureInitialized() before capturing the execution authenticator,
restoring the lazy-initialization the load used to trigger implicitly; the
post-load stability check still reads without re-initializing so a mid-flight
reset stays a retryable failure. Regression
testResetCatalogReinitializesBeforeCapturingAuthenticator covers
reset-completes-before-miss through a retained external table.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -108,13 +136,114 @@ public Table getIcebergTable(ExternalTable dorisTable) {
return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
}
+ public Table getWritableIcebergTable(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ CatalogIf catalog = getCatalog(nameMapping.getCtlId());
+ if (catalog == null) {
+ throw new RuntimeException("Cannot find catalog " +
nameMapping.getCtlId()
+ + " when loading a writable Iceberg table");
+ }
+ // DDL/actions must start from the live catalog generation. DML that
was planned against a
+ // retained read generation wraps this live table separately in
IcebergTransaction. The
+ // authenticator, ops and loaded handle must all come from that one
generation, so the
+ // acquisition is re-validated afterwards: a concurrent
property/credential ALTER can reset
+ // and reinitialize the catalog mid-flight without retiring this
engine's cache group.
+ ExecutionAuthenticator authenticator =
requireExecutionAuthenticator(catalog);
Review Comment:
Fixed in 8cb50adebc0: requireExecutionAuthenticator() now runs
makeSureInitialized() before capturing, so both the table-entry miss loader and
the writable accessor reinitialize a reset catalog before G2 acquisition
starts; the stability re-validation deliberately reads without initializing so
a mid-flight reset still counts as unstable. Regression
testResetCatalogReinitializesBeforeCaptureAndWritableStaysOnDispatchGeneration
covers reset-before-miss and reset-before-writable.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java:
##########
@@ -483,7 +483,7 @@ public void truncateTableImpl(ExternalTable dorisTable,
List<String> partitions)
@Override
public void createOrReplaceBranchImpl(ExternalTable dorisTable,
CreateOrReplaceBranchInfo branchInfo)
throws UserException {
- Table icebergTable = IcebergUtils.getIcebergTable(dorisTable);
+ Table icebergTable = IcebergUtils.getWritableIcebergTable(dorisTable);
Review Comment:
Fixed in 8cb50adebc0: getWritableIcebergTable gained a dispatch-generation
anchor - getWritableIcebergTable(dorisTable, expectedOps) fails the acquisition
(retryable) when the catalog no longer serves the generation the caller
retains. Every DDL update site in IcebergMetadataOps now passes its own ops
instance, so the acquired table, the ops and the retained authenticator that
later runs manageSnapshots/commit are provably from one generation (the ops
instance and its captured authenticator are created together per
initialization).
--
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]
