CalvinKirs commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3836926607
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -108,13 +136,141 @@ public Table getIcebergTable(ExternalTable dorisTable) {
return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
}
+ public Table getWritableIcebergTable(ExternalTable dorisTable) {
+ return getWritableIcebergTable(dorisTable, null);
+ }
+
+ /**
+ * Acquire a writable table for a caller that retains the {@code
IcebergMetadataOps} of the
+ * generation its operation was dispatched on (a DDL method of that ops
instance, a DML
+ * transaction). The acquisition fails when the catalog has moved to a
different generation,
+ * so the caller's later updates - executed through its retained ops and
authenticator -
+ * can never operate a newer generation's handle.
+ */
+ public Table getWritableIcebergTable(ExternalTable dorisTable, @Nullable
IcebergMetadataOps expectedOps) {
+ 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);
+ IcebergMetadataOps ops = resolveMetadataOps(catalog);
+ if (expectedOps != null && ops != expectedOps) {
+ throw catalogGenerationMoved(nameMapping);
+ }
+ Table table = execute(authenticator, () -> ops.loadTable(
+ nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName()));
+ ensureCatalogGenerationStable(catalog, ops, authenticator,
nameMapping);
+ return table;
+ }
+
+ Table getQueryScopedIcebergTable(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> entry =
+ tableEntry.get(nameMapping.getCtlId());
+ IcebergTableCacheValue tableValue =
+ entry.get(nameMapping);
+ return createQueryTable(nameMapping, tableValue);
+ }
+
+ /** Resolve the current table generation, exposing the handle and its
captured context together. */
+ IcebergTableCacheValue getTableCacheValue(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ }
+
+ /** Query-scoped view of an already-resolved generation; see {@link
#getTableCacheValue}. */
+ Table createQueryScopedTable(ExternalTable dorisTable,
IcebergTableCacheValue tableValue) {
+ return createQueryTable(dorisTable.getOrBuildNameMapping(),
tableValue);
+ }
+
+ private Table createQueryTable(
+ NameMapping nameMapping, IcebergTableCacheValue tableValue) {
+ boolean isolateForQueries = tableValue.isQueryIsolationPrepared()
+ ||
snapshotEntry.get(nameMapping.getCtlId()).isWeightAccounting();
+ if (!isolateForQueries) {
+ return tableValue.getIcebergTable();
+ }
+ Table queryTable = tableValue.newQueryScopedTable();
+ IcebergSnapshotCacheValue.loadQueryMetadataForStatement(queryTable);
+ return queryTable;
+ }
+
public IcebergSnapshotCacheValue getSnapshotCache(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ IcebergTableCacheValue tableValue =
+ tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
Review Comment:
Fixed in bfca15ad5d4: CatalogMgr now calls
ExternalMetaCacheMgr.onCatalogOperationalContextChanged after the committed
ALTER (fresh and replayed), which retires the routed engines cached entries
(groups and policies stay) - Iceberg base tables, snapshot/schema projections
and Paimon memoized latest state included - so an immediate post-ALTER
statement loads a generation bound to the new context instead of retrying
against the unplannable A1 base. Failed validations retire nothing. Regression
testCommittedAlterRetiresTheOperationalContextButFailedAlterDoesNot covers both
paths.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java:
##########
@@ -227,16 +390,84 @@ public void invalidateCatalogByEngine(long catalogId,
String engine) {
() -> cache.invalidateCatalogEntries(catalogId)));
}
+ /**
+ * Run an action under the same per-catalog lifecycle fence that guards
lazy initialization
+ * and group retirement. The stripe lock is reentrant, so fenced actions
may call back into
+ * removeCatalog/rollbackCatalogProperties for the same catalog.
+ */
+ public <T> T withCatalogLifecycleLock(long catalogId,
java.util.function.Supplier<T> action) {
+ Lock lifecycleLock = catalogLifecycleLocks.get(catalogId);
+ lifecycleLock.lock();
+ try {
+ return action.get();
+ } finally {
+ lifecycleLock.unlock();
+ }
+ }
+
public void removeCatalog(long catalogId) {
- routeCatalogEngines(catalogId, cache -> safeInvalidate(
- cache, catalogId, "removeCatalog",
- () -> cache.invalidateCatalog(catalogId)));
+ Lock lifecycleLock = catalogLifecycleLocks.get(catalogId);
+ lifecycleLock.lock();
+ try {
+ routeCatalogEngines(catalogId, cache -> safeInvalidate(
+ cache, catalogId, "removeCatalog",
+ () -> cache.invalidateCatalog(catalogId)));
+ } finally {
+ lifecycleLock.unlock();
+ }
+ }
+
+ /**
+ * DROP CATALOG (or its replay): the id is never reused. Retires the cache
groups like
+ * {@link #removeCatalog} and additionally releases engine side state that
must survive
+ * same-id policy rebuilds; the hook reaches every engine even when its
entry group is
+ * already retired.
+ */
+ public void removeCatalogPermanently(long catalogId) {
+ Lock lifecycleLock = catalogLifecycleLocks.get(catalogId);
+ lifecycleLock.lock();
+ try {
+ routeCatalogEngines(catalogId, cache -> safeInvalidate(
+ cache, catalogId, "removeCatalogPermanently",
+ () -> cache.invalidateCatalog(catalogId)));
+ for (ExternalMetaCache cache : cacheRegistry.allCaches()) {
Review Comment:
Fixed in bfca15ad5d4 together with the membership concern: the terminal
state is no longer published by the engine hook at all - the lookup probes the
catalog manager, which removes the catalog before removeCatalogPermanently
detaches or closes any engine group, so a retained lookup during a blocked
close already observes the drop and fails immediately. Post-close engine
cleanup stays where it was, and rename keeps the catalog registered so its
transient absence still gets the bounded retry.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/AbstractExternalMetaCache.java:
##########
@@ -222,16 +337,59 @@ protected final ExternalTable
findExternalTable(NameMapping nameMapping, String
nameMapping.getLocalTblName(), engineNameForError));
}
+ // A contended cache-policy handoff resolves within this window; see
requireCatalogEntryGroup.
+ private static final long PREPARE_RETRY_WINDOW_NANOS = 2_000_000_000L;
+ private static final long PREPARE_RETRY_SLEEP_MS = 50L;
+
private CatalogEntryGroup requireCatalogEntryGroup(long catalogId) {
CatalogEntryGroup group = catalogEntries.get(catalogId);
+ if (group == null && catalogPreparer != null) {
+ // The caller prepared the catalog before capturing this engine,
but a cache-policy
+ // ALTER retired the group in between. Re-prepare under the
lifecycle fence so the
+ // lookup observes the new policy instead of failing a valid
catalog. The preparer
+ // never blocks on the fence (a nested default loader may hold a
Caffeine bin lock
+ // that retirement itself needs), so a contended handoff is
absorbed with a bounded
+ // sleep-and-retry: the ALTER finishes within the window, or the
lookup fails as
+ // before without any deadlock.
+ long deadlineNanos = System.nanoTime() +
PREPARE_RETRY_WINDOW_NANOS;
+ while (!permanentlyRemovedCatalogs.contains(catalogId)) {
+ catalogPreparer.accept(catalogId);
+ group = catalogEntries.get(catalogId);
+ if (group != null || System.nanoTime() >= deadlineNanos
+ || permanentlyRemovedCatalogs.contains(catalogId)) {
+ break;
+ }
+ try {
+ Thread.sleep(PREPARE_RETRY_SLEEP_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
+ }
if (group == null) {
+ if (permanentlyRemovedCatalogs.contains(catalogId)) {
+ throw new IllegalStateException(String.format(
+ "Catalog %d was dropped; engine '%s' serves no
metadata for it.",
+ catalogId, engine));
+ }
throw new IllegalStateException(String.format(
"Catalog %d is not initialized for engine '%s'.",
catalogId, engine));
}
return group;
}
+ @Override
+ public void bindCatalogPreparer(LongConsumer catalogPreparer) {
+ this.catalogPreparer = catalogPreparer;
+ }
+
+ @Override
+ public void onCatalogPermanentlyRemoved(long catalogId) {
+ permanentlyRemovedCatalogs.add(catalogId);
Review Comment:
Fixed in bfca15ad5d4: the per-engine dropped-id membership is gone.
requireCatalogEntryGroup now derives the terminal outcome from the catalog
manager (ids are never reused and the manager removes the catalog before any
engine detachment), so catalog churn retains no per-engine state at all; a test
probe hook replaces the manager lookup in isolated tests. The drop/rename
regression asserts immediate terminal failure and the preserved
transient-absence 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]
