github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3830155254
##########
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:
[P1] Capture the Paimon table and authenticator atomically
`tableLoader.load()` resolves the mutable catalog again and loads T1 through
G1/A1; only after it returns does this expression read the first catalog
reference's current authenticator. A concurrent storage/credential ALTER can
reset and reinitialize that object in between, so the miss either throws after
the external load or publishes synthetic generation T1/A2 (ordinary credential
ALTER does not retire the Paimon group). All later fence/schema work then
deliberately trusts A2, and the generation check cannot detect the splice. This
is earlier than the existing hydration threads, which start from a coherent
captured generation. Please return the table plus its exact execution
context/resource epoch as one atomic result, or validate the lifecycle epoch
before use/publication, with a latch test spanning G1 load,
ALTER/reinitialization, and publication.
##########
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:
[P1] Acquire Iceberg handles and auth from one catalog generation
`executeAuthenticated()` captures A1 before this lambda, but after
`ops.loadTable()` returns this line rereads the mutable catalog. A concurrent
storage/credential ALTER can reset and reinitialize it in between, so this miss
either throws after loading T1 or publishes T1 bound to A2 (those ALTERs do not
retire the Iceberg cache group). The new `getWritableIcebergTable()` has the
same split in reverse: it resolves G1 `ops` before entering the current
authenticator, so reset can close/null that ops object before line 148 uses it.
The existing projection thread starts with a coherent table generation, so it
cannot cover these acquisition races. Please acquire the SDK ops/table, exact
authenticator, and lifecycle epoch as one generation-owned bundle (or validate
a lifecycle lease before use/publication), and latch-test both boundaries
across ALTER.
--
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]
