github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3788969665
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java:
##########
@@ -446,6 +447,19 @@ protected void checkProperties(CatalogProperty property)
throws DdlException {
}
}
+ try {
+ Env currentEnv = Env.getCurrentEnv();
+ ExternalMetaCacheMgr extMetaCacheMgr = currentEnv == null ? null :
currentEnv.getExtMetaCacheMgr();
+ if (extMetaCacheMgr == null) {
+ // This fallback is only for isolated construction tests
before Env is initialized.
+
ExternalMetaCacheBudgetManager.fromConfig().validateCatalogMaxWeight(properties);
+ } else {
+ extMetaCacheMgr.validateCatalogCacheProperties(this,
properties);
Review Comment:
[P1] Retire cache groups when tentative validation rolls back
Legacy validators publish these candidate properties before this call, so a
concurrent first lookup can initialize the managed cache group from them. If a
later connector check rejects the ALTER, CatalogMgr restores CatalogProperty
but does not remove that group; subsequent lookups take the initialized fast
path and keep the rejected max-weight policy. Please make this validation
detached or retire the group under the same lifecycle fence after rollback, and
cover failed ALTER racing first initialization.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -216,27 +324,32 @@ private SchemaCacheValue
loadSchemaCacheValue(IcebergSchemaCacheKey key) {
key.getNameMapping().getLocalTblName(),
key.getSchemaId()));
}
- private IcebergSnapshotCacheValue loadSnapshotProjection(ExternalTable
dorisTable, Table icebergTable) {
+ private IcebergSnapshotCacheValue loadSnapshotProjection(
+ ExternalTable dorisTable, Table projectionTable, Table
retainedTable,
+ String retainedCurrentSnapshotJson, boolean isolateForQueries) {
if (!(dorisTable instanceof MTMVRelatedTableIf)) {
throw new RuntimeException(String.format("Table %s.%s is not a
valid MTMV related table.",
dorisTable.getDbName(), dorisTable.getName()));
}
try {
- // Freeze before deriving snapshot, partitions, and aliases;
BaseTable accessors share
- // refreshable operations and otherwise could mix two concurrent
metadata generations.
- Table retainedTable =
IcebergSnapshotCacheValue.retainTableGeneration(icebergTable);
MTMVRelatedTableIf table = (MTMVRelatedTableIf) dorisTable;
- IcebergSnapshot latestIcebergSnapshot =
IcebergUtils.getLatestIcebergSnapshot(retainedTable);
+ IcebergSnapshot latestIcebergSnapshot =
IcebergUtils.getLatestIcebergSnapshot(projectionTable);
IcebergPartitionInfo icebergPartitionInfo;
if (!table.isValidRelatedTable()) {
icebergPartitionInfo = IcebergPartitionInfo.empty();
} else {
- icebergPartitionInfo =
IcebergUtils.loadPartitionInfo(dorisTable, retainedTable,
+ icebergPartitionInfo =
IcebergUtils.loadPartitionInfo(dorisTable, projectionTable,
Review Comment:
[P1] Fence this schema dependency with the physical table generation
The new snapshot key correctly misses after a same-name recreation with a
different UUID, but loadPartitionInfo still resolves schema through
(NameMapping, schemaId). Recreated Iceberg tables restart schema IDs, and
refreshing the table entry does not retire schemaEntry, so the new snapshot can
be built with the old table's partition-column types and full schema. Please
include UUID/generation in the schema key or retire it with table-generation
replacement.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTableCacheValue.java:
##########
@@ -17,25 +17,90 @@
package org.apache.doris.datasource.iceberg;
-import com.google.common.base.Suppliers;
-import org.apache.iceberg.Table;
+import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
+import org.apache.doris.datasource.NameMapping;
+import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate;
+import org.apache.doris.datasource.metacache.MetaCacheSizeEstimator;
-import java.util.function.Supplier;
+import org.apache.iceberg.Table;
public class IcebergTableCacheValue {
- private final Table icebergTable;
- private final Supplier<IcebergSnapshotCacheValue> latestSnapshotCacheValue;
+ private Table icebergTable;
+ private String retainedCurrentSnapshotJson;
+ private volatile boolean queryIsolationPrepared;
+ private long retainedTablePayloadBytes;
+ private MetaCacheSizeEstimate sizeEstimate;
- public IcebergTableCacheValue(Table icebergTable,
Supplier<IcebergSnapshotCacheValue> latestSnapshotCacheValue) {
- this.icebergTable = icebergTable;
- this.latestSnapshotCacheValue =
Suppliers.memoize(latestSnapshotCacheValue::get);
+ public IcebergTableCacheValue(Table icebergTable) {
+ this.icebergTable =
IcebergSnapshotCacheValue.retainTableGeneration(icebergTable);
+ }
+
+ IcebergTableCacheValue(Table icebergTable, ExecutionAuthenticator
authenticator) {
+ this.icebergTable = IcebergSnapshotCacheValue.retainTableGeneration(
+ icebergTable, authenticator);
}
public Table getIcebergTable() {
+ return queryIsolationPrepared
Review Comment:
[P1] Route system tables through the stale-metadata retry fence
After weighted publication this accessor returns a lazy query view, and
IcebergSysExternalTable uses it directly. If metadata cleanup removes the
retained file, $history/$snapshots/$refs and ALL_* tables throw later from
their lazy accessors without invalidating the table entry, while
getQueryScopedIcebergTable() already has the required retry. Please use that
fenced accessor here (or wrap the lazy load) and cover a deleted pinned file
through a system-table query.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -119,8 +138,8 @@ public PaimonSchemaCacheValue
getPaimonSchemaCacheValue(NameMapping nameMapping,
private PaimonTableCacheValue loadTableCacheValue(NameMapping nameMapping)
{
Table paimonTable = tableLoader.load(nameMapping);
- return new PaimonTableCacheValue(paimonTable,
- () -> latestSnapshotProjectionLoader.load(nameMapping,
paimonTable));
+ PaimonSnapshotCacheValue fence =
latestSnapshotProjectionLoader.loadFence(nameMapping, paimonTable);
Review Comment:
[P1] Do not make table-only cache publication depend on unauthenticated
snapshot discovery
PaimonTableLoader returns after getPaimonTable's executionAuthenticator
scope closes, then this eager fence calls
copyWithLatestSchema/latestSnapshot/schemaManager.latest. That moves remote
work into every cold load/refresh, so even comment/property/isPartitionedTable
callers can now block or fail outside the catalog credential scope. Please
capture the fence within authentication while preserving a memoized lazy path
for table-only uses, and cover a table-only load with guarded snapshot metadata.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,7 +100,13 @@ public Table getPaimonTable(NameMapping nameMapping) {
public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable)
{
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ PaimonTableCacheValue tableValue =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ PaimonSnapshot fence =
tableValue.getLatestSnapshotFence().getSnapshot();
+ PaimonSnapshotEntryKey key = PaimonSnapshotEntryKey.of(
+ nameMapping, fence, tableValue.getGeneration());
+ MetaCacheEntry<PaimonSnapshotEntryKey, PaimonSnapshotCacheValue> entry
=
+ snapshotEntry.get(nameMapping.getCtlId());
+ return entry.get(key, ignored ->
latestSnapshotProjectionLoader.loadAtFence(nameMapping, fence));
Review Comment:
[P1] Carry the table generation into the schema lookup used by this miss
The snapshot key now separates reloaded tables, but loadAtFence immediately
looks up schema by only (NameMapping, schemaId). After same-name drop/recreate
with a restarted ID, that returns the old schema/partition columns and the
result is then cached under the new generation key. Please generation-fence
schemaEntry as well, or atomically retire schema and old snapshot keys when
tableEntry publishes a new generation.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java:
##########
@@ -113,12 +285,46 @@ static Table createWritableTable(Table retainedTable,
Table liveTable) {
throw new IllegalArgumentException(
"Iceberg commit table must provide writable table
operations");
}
- TableMetadata retainedMetadata = ((HasTableOperations)
retainedTable).operations().current();
- TableOperations liveOperations = ((HasTableOperations)
liveTable).operations();
+ TableOperations retainedOperations = ((HasTableOperations)
retainedTable).operations();
+ TableMetadata retainedMetadata = reloadRetainedMetadata
+ ? loadQueryMetadata(retainedOperations) :
retainedOperations.current();
+ TableOperations liveOperations = unwrapRetainedTableOperations(
+ ((HasTableOperations) liveTable).operations());
return tableWithOperations(retainedTable,
new WritableTableOperations(liveOperations, retainedMetadata));
}
+ static Table createWritableTable(Table retainedTable, Table liveTable) {
+ return createWritableTable(retainedTable, liveTable,
+ isNonGrowingGeneration(retainedTable));
+ }
+
+ private static boolean isNonGrowingGeneration(Table table) {
+ return isFrozenGeneration(table)
+ && ((FrozenTableOperations) ((HasTableOperations)
table).operations()).nonGrowing;
+ }
+
+ private static TableMetadata loadQueryMetadata(TableOperations
retainedOperations) {
+ TableMetadata retainedMetadata = retainedOperations.current();
+ TableOperations serviceOperations =
unwrapRetainedTableOperations(retainedOperations);
+ String metadataLocation = retainedMetadata.metadataFileLocation();
+ if (metadataLocation != null && !metadataLocation.isEmpty() &&
serviceOperations.io() != null) {
+ ExecutionAuthenticator authenticator = retainedOperations
instanceof FrozenTableOperations
+ ? ((FrozenTableOperations)
retainedOperations).authenticator : null;
+ try {
+ return authenticator == null
Review Comment:
[P1] Validate the parsed table generation before accepting this metadata file
A purged Hadoop table recreation can reuse the same
metadata/v1.metadata.json path; this read then succeeds even though its UUID
differs from the retained table. QueryScopedTable subsequently mixes the old
retained schema/current snapshot with replacement refs and history, so the new
snapshot-key UUID does not help. Please compare the parsed UUID/generation and
throw StaleMetadataException on mismatch so the existing invalidation/reload
path runs.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java:
##########
@@ -85,26 +130,153 @@ public Optional<Map<Integer, List<String>>>
getNameMapping() {
}
public Optional<Table> getIcebergTable() {
+ return queryIsolationPrepared
+ ? icebergTable.map(table -> createQueryScopedTable(
+ table, retainedCurrentSnapshotJson))
+ : icebergTable;
+ }
+
+ MetaCacheSizeEstimate prepareForCachePublication(IcebergSnapshotEntryKey
key) {
+ if (sizeEstimate == null) {
+ if (retainedCurrentSnapshotJson == null) {
+ retainedCurrentSnapshotJson = icebergTable
+
.map(IcebergSnapshotCacheValue::retainCurrentSnapshotJson).orElse(null);
+ }
+ retainedTablePayloadBytes = icebergTable
+
.map(IcebergCacheSizeEstimator::retainedTablePayloadBytes).orElse(0L);
+ sizeEstimate =
MetaCacheSizeEstimator.estimateSafely("iceberg_snapshot_preparation_failed",
+ () -> IcebergCacheSizeEstimator.estimateSnapshotEntry(key,
this));
+ if (sizeEstimate.isComplete()) {
+ icebergTable = icebergTable.map(
+ IcebergSnapshotCacheValue::retainNonGrowingGeneration);
+ queryIsolationPrepared = true;
+ }
+ }
+ return sizeEstimate;
+ }
+
+ public MetaCacheSizeEstimate getSizeEstimate() {
+ return sizeEstimate == null
+ ? MetaCacheSizeEstimate.incomplete("not_prepared") :
sizeEstimate;
+ }
+
+ long getRetainedNameMappingPayloadBytes() {
+ return retainedNameMappingPayloadBytes;
+ }
+
+ long getRetainedTablePayloadBytes() {
+ return retainedTablePayloadBytes;
+ }
+
+ long getRetainedCurrentSnapshotPayloadBytes() {
+ return retainedSnapshotJsonBytes(retainedCurrentSnapshotJson);
+ }
+
+ Optional<Table> getRetainedIcebergTable() {
return icebergTable;
}
static Table retainTableGeneration(Table table) {
+ return retainTableGeneration(table, null);
+ }
+
+ static Table retainTableGeneration(Table table, ExecutionAuthenticator
authenticator) {
if (!(table instanceof HasTableOperations) ||
isFrozenGeneration(table)) {
return table;
}
TableOperations operations = ((HasTableOperations) table).operations();
// Capture current() exactly once so every projection derived from the
returned table sees
// one metadata generation even when the shared catalog handle
refreshes concurrently.
- TableOperations frozenOperations = new
FrozenTableOperations(operations, operations.current());
+ TableOperations frozenOperations = new FrozenTableOperations(
+ operations, operations.current(), authenticator);
return tableWithOperations(table, frozenOperations);
}
+ static Table retainNonGrowingGeneration(Table table) {
+ if (!isFrozenGeneration(table)) {
+ return table;
+ }
+ TableOperations retainedOperations = ((HasTableOperations)
table).operations();
+ TableMetadata source = retainedOperations.current();
+ if (source.schemas().isEmpty() || source.specs().isEmpty()
+ || source.sortOrders().isEmpty()) {
+ return table;
+ }
+ TableMetadata.Builder builder =
TableMetadata.buildFromEmpty(source.formatVersion());
+ if (source.uuid() != null) {
+ builder.assignUUID(source.uuid());
+ }
+ for (Schema schema : source.schemas()) {
+ builder.addSchema(schema);
+ }
+ builder.setCurrentSchema(source.currentSchemaId());
+ for (PartitionSpec spec : source.specs()) {
+ builder.addPartitionSpec(spec);
+ }
+ builder.setDefaultPartitionSpec(source.defaultSpecId());
+ for (SortOrder sortOrder : source.sortOrders()) {
+ builder.addSortOrder(sortOrder);
+ }
+ builder.setDefaultSortOrder(source.defaultSortOrderId());
+ builder.setLocation(source.location());
+ builder.setProperties(source.properties());
+ if (source.currentSnapshot() != null) {
+ builder.setBranchSnapshot(
+ new NonGrowingSnapshot(source.currentSnapshot()),
SnapshotRef.MAIN_BRANCH);
+ }
+ TableMetadata retainedMetadata = builder.discardChanges()
+ .withMetadataLocation(source.metadataFileLocation()).build();
+ return tableWithOperations(table, new FrozenTableOperations(
Review Comment:
[P1] Restore previous metadata files for metadata_log_entries
This retained TableMetadata intentionally omits previousFiles(), but
QueryScopedTable still exposes these stripped operations. Iceberg 1.10.1's
[MetadataLogEntriesTable](https://github.com/apache/iceberg/blob/apache-iceberg-1.10.1/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java)
reads operations().current().previousFiles() directly, so enabling a
table/snapshot weight bound makes $metadata_log_entries return only the
synthetic current row. Please provide that system table an exact query-local
operations view and add a multi-generation weighted-cache regression.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -108,13 +126,95 @@ public Table getIcebergTable(ExternalTable dorisTable) {
return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
}
+ public Table getWritableIcebergTable(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ IcebergTableCacheValue tableValue =
+ tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ CatalogIf catalog = getCatalog(nameMapping.getCtlId());
+ if (catalog == null) {
+ throw new RuntimeException("Cannot find catalog " +
nameMapping.getCtlId()
+ + " when loading a writable Iceberg table");
+ }
+ IcebergMetadataOps ops = resolveMetadataOps(catalog);
+ Table liveTable = executeAuthenticated(catalog, () -> ops.loadTable(
+ nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName()));
+ try {
+ return tableValue.getWritableIcebergTable(liveTable);
+ } catch (IcebergSnapshotCacheValue.StaleMetadataException e) {
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> entry =
+ tableEntry.get(nameMapping.getCtlId());
+ entry.invalidateKeyIfSame(nameMapping, tableValue);
+ IcebergTableCacheValue refreshedValue = entry.get(nameMapping);
+ Table refreshedLiveTable = executeAuthenticated(catalog, () ->
ops.loadTable(
+ nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName()));
+ return refreshedValue.getWritableIcebergTable(refreshedLiveTable);
+ }
+ }
+
+ Table getQueryScopedIcebergTable(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> entry =
+ tableEntry.get(nameMapping.getCtlId());
+ IcebergTableCacheValue tableValue =
+ entry.get(nameMapping);
+ try {
+ return createQueryTable(nameMapping, tableValue);
+ } catch (IcebergSnapshotCacheValue.StaleMetadataException e) {
+ entry.invalidateKeyIfSame(nameMapping, tableValue);
+ return createQueryTable(nameMapping, entry.get(nameMapping));
+ }
+ }
+
+ private Table createQueryTable(
+ NameMapping nameMapping, IcebergTableCacheValue tableValue) {
+ boolean isolateForQueries = tableValue.isQueryIsolationPrepared()
+ || snapshotEntry.get(nameMapping.getCtlId()).isWeightBounded();
+ 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);
+ Table retainedTable = tableValue.getRetainedIcebergTable();
+ java.util.Optional<IcebergSnapshotEntryKey> optionalKey =
+ IcebergSnapshotEntryKey.tryCreate(nameMapping, retainedTable);
+ if (!optionalKey.isPresent()) {
+ boolean isolateForQueries = tableValue.isQueryIsolationPrepared();
+ return executeAuthenticated(nameMapping.getCtlId(),
+ () -> loadSnapshotProjection(
+ dorisTable,
+ isolateForQueries ?
tableValue.newQueryScopedTable()
+ : tableValue.getIcebergTable(),
+ tableValue.getRetainedIcebergTable(),
+ tableValue.getRetainedCurrentSnapshotJson(),
isolateForQueries));
+ }
+ IcebergSnapshotEntryKey key = optionalKey.get();
Review Comment:
[P2] Retire snapshot projections from the previous table generation
Each table auto-refresh that observes a newer Iceberg metadata file creates
a distinct key here, but snapshotEntry has no replacement hook to remove older
keys. Those partition projections are no longer addressable by later latest or
time-travel lookups, yet remain for the 24-hour TTL and consume
capacity/catalog-global reservations, displacing or rejecting current metadata.
Please couple table replacement to race-safe cleanup of prior keys and cover
repeated metadata advances.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,7 +100,13 @@ public Table getPaimonTable(NameMapping nameMapping) {
public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable)
{
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ PaimonTableCacheValue tableValue =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ PaimonSnapshot fence =
tableValue.getLatestSnapshotFence().getSnapshot();
+ PaimonSnapshotEntryKey key = PaimonSnapshotEntryKey.of(
Review Comment:
[P2] Retire the prior snapshot generation when tableEntry refreshes
Every successful ten-minute table reload allocates a fresh generation, so
this key correctly misses, but the contextual snapshot entry has no replacement
hook to remove older keys. A continuously queried table can retain roughly 144
unreachable partition projections for the 24-hour access TTL, consuming
quota/capacity and displacing live generations. Please couple table replacement
to race-safe invalidation of its prior snapshot keys and test repeated
refreshes.
--
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]
