github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3824387362
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +111,169 @@ 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.getPaimonTable()).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 executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration()));
+ }
+ // 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 -> executeAuthenticated(nameMapping, () -> {
+ loaded.set(true);
+ return latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration());
+ }));
+ 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;
+ }
}
public PaimonSnapshotCacheValue loadSnapshotProjection(ExternalTable
dorisTable, Table effectiveTable) {
- return
latestSnapshotProjectionLoader.load(dorisTable.getOrBuildNameMapping(),
effectiveTable);
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.load(nameMapping,
effectiveTable));
}
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.getPaimonTable());
}
public PaimonSnapshotCacheValue loadSnapshotAtFence(
ExternalTable dorisTable, PaimonSnapshot fence) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return latestSnapshotProjectionLoader.loadAtFence(nameMapping, fence);
+ return executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(nameMapping,
fence));
}
public PaimonSnapshotCacheValue loadSnapshotAtFence(
ExternalTable dorisTable, Table effectiveTable, PaimonSnapshot
fence) {
- return latestSnapshotProjectionLoader.loadEffectiveAtFence(
- dorisTable.getOrBuildNameMapping(), effectiveTable, fence);
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.loadEffectiveAtFence(
+ nameMapping, effectiveTable, fence));
}
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());
+ }
+
+ PaimonSchemaCacheValue getPaimonSchemaCacheValue(
+ NameMapping nameMapping, long schemaId, long tableGeneration,
Table retainedTable) {
+ 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) executeAuthenticated(nameMapping,
+ () -> loadSchemaCacheValue(key, retainedTable));
+ }
+ MetaCacheEntry<PaimonSchemaCacheKey, SchemaCacheValue> entry =
schemaEntry.get(nameMapping.getCtlId());
+ SchemaCacheValue schemaCacheValue = entry.get(key,
Review Comment:
[P1] Authenticate automatic schema refreshes
This lambda authenticates a schema miss, but `schemaEntry` is still an
auto-refreshing `MetaCacheEntryDef.of(...)`. On a hit after
`external_cache_refresh_time_minutes`, Caffeine invokes the registered default
loader `this::loadSchemaCacheValue` instead of this contextual loader; that
path reaches `DataTable.schemaManager().schema(schemaId)` after the table-load
authenticator scope has ended. Credential-scoped catalogs can therefore fail
the timed refresh and keep serving the old schema (the new test covers only a
miss). Please make this generation-owned schema entry
contextual/non-refreshing, or authenticate and generation-fence its default
refresh loader, with a refresh-path authenticator 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]
