github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3835758230
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java:
##########
@@ -293,13 +293,13 @@ public void beginDelete(ExternalTable dorisTable, Table
targetTable) throws User
}
private Table createTransactionTable(ExternalTable dorisTable, Table
retainedTable) {
- if (!IcebergSnapshotCacheValue.isFrozenGeneration(retainedTable)) {
+ if (!IcebergSnapshotCacheValue.isRetainedGeneration(retainedTable)) {
return retainedTable;
}
// Reads stay on the retained generation; commit refreshes may follow
data-only snapshots,
// while writer-contract changes still invalidate files produced for
the retained metadata.
return IcebergSnapshotCacheValue.createWritableTable(
- retainedTable, IcebergUtils.getIcebergTable(dorisTable));
+ retainedTable,
IcebergUtils.getWritableIcebergTable(dorisTable));
Review Comment:
[P1] Keep the writable transaction on one catalog generation
`BaseExternalTableInsertExecutor` retains the transaction manager (and
therefore G1 `IcebergMetadataOps`/authenticator). If a credential/storage ALTER
resets the catalog after transaction creation and another access reinitializes
G2 before `beginInsert`/delete/merge, this changed call validly returns a G2
table through O2/A2, but the surrounding transaction creation and finish/update
methods still run its operations through O1/A1; `getThreadPoolWithPreAuth()`
can simultaneously resolve the replacement G2 pool. The acquisition itself is
coherent, so the existing mid-acquisition thread does not catch this
caller-side generation split. Please carry the writable table, authenticator,
and pre-auth executor as one generation-owned binding (or fail/restart when the
retained manager epoch differs), with a reset/reinitialize-before-begin
regression.
##########
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:
[P1] Execute writable DDL on the acquired catalog generation
These DDL methods belong to an `IcebergMetadataOps` instance whose
authenticator was captured with G1. If a credential/storage ALTER resets the
catalog after this DDL is dispatched and another access initializes G2 before
this line, `getWritableIcebergTable()` validly returns a G2 table through
O2/A2, but this method later invokes `manageSnapshots`/`commit` through
retained A1; the changed branch/tag, schema, reorder, and partition-evolution
siblings do the same. Before this patch `getIcebergTable()` hit the
still-published G1 table entry for ordinary credential updates, so caller and
table stayed aligned. This is the DDL sibling of the retained-transaction
split, not the existing mid-acquisition thread. Please return an authenticated
writable binding and execute each update through that binding, or
reject/restart when the invoking ops epoch differs.
##########
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:
[P1] Initialize Iceberg before capturing its authenticator
After a credential/storage ALTER calls `resetToUninitialized()`, an
already-bound query, action, or DML transaction can reach this new writable
accessor while `executionAuthenticator` is null. It now calls
`requireExecutionAuthenticator()` before `resolveMetadataOps()`, even though
the latter's `getMetadataOps()` is what would run `makeSureInitialized()`; the
request therefore throws before G2 acquisition starts. The ordinary table-miss
loader has the same new order at lines 339-340 (and previously resolved ops
first), so retained reads are affected too. This is earlier than the existing
mid-acquisition reset thread, whose test always supplies a non-null
authenticator. Please expose one atomic initialized ops/authenticator
acquisition and cover both reset-before-writable and reset-before-table-miss.
##########
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:
[P1] Initialize Paimon before capturing its authenticator
A query can retain this external table across a credential/storage ALTER
that calls `resetToUninitialized()` without necessarily retiring the Paimon
cache group. That reset clears `executionAuthenticator`, and the next
table-entry miss now calls `getExecutionAuthenticator()` before
`tableLoader.load()`; the latter is the call whose `getPaimonTable()` would run
`makeSureInitialized()`. The miss therefore throws `ExecutionAuthenticator is
null` before the catalog can reinitialize. This is earlier than the existing
mid-load generation-splice thread. Please expose an atomic initialized
catalog/table/authenticator acquisition (or otherwise initialize under a
generation fence) and cover reset-completes-before-miss through a retained
`PaimonExternalTable`.
--
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]
