CalvinKirs commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3830215208
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -139,45 +291,62 @@ public ManifestCacheValue
getManifestCacheValue(ExternalTable dorisTable,
MetaCacheEntry<IcebergManifestEntryKey, ManifestCacheValue>
manifestEntry =
this.manifestEntry.get(nameMapping.getCtlId());
IcebergManifestEntryKey key = IcebergManifestEntryKey.of(manifest);
- boolean hit = manifestEntry.getIfPresent(key) != null;
+ boolean hit = manifestEntry.peekIfPresent(key) != null;
if (cacheHitRecorder != null) {
cacheHitRecorder.accept(hit);
}
- return manifestEntry.get(key, ignored ->
loadManifestCacheValue(manifest, icebergTable, key.getContent()));
+ return manifestEntry.get(key,
+ ignored -> loadManifestCacheValue(
+ manifest, icebergTable, key.getContent(),
manifestEntry.isWeightAccounting()));
}
@Override
public void invalidateCatalog(long catalogId) {
- dropManifestFileIoCacheForCatalog(catalogId);
+ // Collect while the entries are still enumerable, drop only after the
entries are
+ // detached: a load racing a pre-detach drop could repopulate the SDK
content cache
+ // for a FileIO this reset already cleaned.
+ List<FileIO> retainedFileIos = collectManifestFileIos(catalogId);
super.invalidateCatalog(catalogId);
+ dropManifestFileIoCaches(retainedFileIos);
}
@Override
public void invalidateCatalogEntries(long catalogId) {
- dropManifestFileIoCacheForCatalog(catalogId);
+ List<FileIO> retainedFileIos = collectManifestFileIos(catalogId);
super.invalidateCatalogEntries(catalogId);
+ dropManifestFileIoCaches(retainedFileIos);
}
private IcebergTableCacheValue loadTableCacheValue(NameMapping
nameMapping) {
- CatalogIf catalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(nameMapping.getCtlId());
+ CatalogIf catalog = getCatalog(nameMapping.getCtlId());
if (catalog == null) {
throw new RuntimeException(String.format("Cannot find catalog %d
when loading table %s/%s.",
nameMapping.getCtlId(), nameMapping.getLocalDbName(),
nameMapping.getLocalTblName()));
}
IcebergMetadataOps ops = resolveMetadataOps(catalog);
- try {
- Table table = ((ExternalCatalog)
catalog).getExecutionAuthenticator()
- .execute(() ->
ops.loadTable(nameMapping.getRemoteDbName(), nameMapping.getRemoteTblName()));
- ExternalTable dorisTable = findExternalTable(nameMapping, ENGINE);
- return new IcebergTableCacheValue(table, () ->
loadSnapshotProjection(dorisTable, table));
- } catch (Exception e) {
- throw new RuntimeException(ExceptionUtils.getRootCauseMessage(e),
e);
- }
+ return executeAuthenticated(catalog, () -> {
+ Table table = ops.loadTable(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName());
+ IcebergTableCacheValue value = new IcebergTableCacheValue(table);
+ if (catalog instanceof ExternalCatalog) {
+ value.bindAuthenticator(((ExternalCatalog)
catalog).getExecutionAuthenticator());
Review Comment:
Fixed in 7fe5e769ccd: the miss loader now captures the execution
authenticator and metadata ops once, runs ops.loadTable() under that exact
context, binds the same authenticator instance, and re-validates the
acquisition (ops and authenticator identity) against the catalog before the
value is used or published; getWritableIcebergTable follows the same
acquire-execute-validate contract. A mid-flight reset turns into a clean
retryable failure instead of publishing a spliced handle. Latch regression
testAcquisitionsFailWhenCatalogResetsMidFlight covers both boundaries (miss
loader and writable handle) plus the coherent retry.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +114,265 @@ 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);
+ }
}
- public PaimonSnapshotCacheValue loadSnapshotProjection(ExternalTable
dorisTable, Table effectiveTable) {
- return
latestSnapshotProjectionLoader.load(dorisTable.getOrBuildNameMapping(),
effectiveTable);
+ 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());
+ }
+
+ /** 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 {
+ PaimonExternalCatalog catalog = tableLoader.catalog(nameMapping);
+ return new PaimonTableCacheValue(
+ tableLoader.load(nameMapping),
catalog.getExecutionAuthenticator());
Review Comment:
Fixed in 7fe5e769ccd: loadTableCacheValue now captures the catalog reference
and its execution authenticator before tableLoader.load(), and re-validates
both identities after the external load returns and before publication, so a
concurrent reset/reinitialization fails the miss (retryable) instead of
publishing a synthetic generation spliced with the next generation context.
Latch regression
testTableLoadSplicedByMidFlightCatalogResetFailsInsteadOfPublishing spans G1
load, mid-flight ALTER and the coherent retry.
--
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]
