This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 914257a1fd0 branch-4.2: [improvement](datasource) Version Iceberg
partition views and SQL cache metadata (#69299) (#69983)
914257a1fd0 is described below
commit 914257a1fd054265225bb19637546f5ab659ad4a
Author: 924060929 <[email protected]>
AuthorDate: Tue Nov 3 11:48:14 2026 +0800
branch-4.2: [improvement](datasource) Version Iceberg partition views and
SQL cache metadata (#69299) (#69983)
cherry pick from #69299
---
.../doris/common/cache/NereidsSqlCacheManager.java | 45 +-
.../doris/datasource/hive/HMSExternalTable.java | 3 +
.../doris/datasource/hive/IcebergDlaTable.java | 17 +-
.../iceberg/IcebergCacheSizeEstimator.java | 6 +-
.../iceberg/IcebergExternalMetaCache.java | 95 +++-
.../datasource/iceberg/IcebergExternalTable.java | 55 ++-
.../datasource/iceberg/IcebergPartitionInfo.java | 13 +-
.../iceberg/IcebergSnapshotCacheValue.java | 26 +-
.../doris/datasource/iceberg/IcebergUtils.java | 82 +++-
.../datasource/iceberg/source/IcebergScanNode.java | 9 +
.../doris/datasource/mvcc/MvccReadSelector.java | 116 +++++
.../org/apache/doris/mtmv/MTMVRefreshContext.java | 3 +-
.../org/apache/doris/nereids/SqlCacheContext.java | 64 ++-
.../doris/nereids/rules/analysis/BindRelation.java | 15 +-
.../org/apache/doris/qe/cache/CacheAnalyzer.java | 24 +-
.../org/apache/doris/qe/cache/SqlCacheVersion.java | 44 ++
.../doris/qe/cache/SqlCacheVersionProvider.java | 31 ++
.../iceberg/IcebergExternalMetaCacheTest.java | 511 ++++++++++++++++++++-
.../iceberg/IcebergPartitionInfoTest.java | 3 +-
.../iceberg/IcebergSqlCacheVersionTest.java | 113 +++++
20 files changed, 1182 insertions(+), 93 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/cache/NereidsSqlCacheManager.java
b/fe/fe-core/src/main/java/org/apache/doris/common/cache/NereidsSqlCacheManager.java
index 1f77f1fc29b..31dd8bb1705 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/common/cache/NereidsSqlCacheManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/cache/NereidsSqlCacheManager.java
@@ -34,6 +34,8 @@ import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.CatalogMgr;
import org.apache.doris.datasource.hive.HMSExternalTable;
+import org.apache.doris.datasource.mvcc.MvccReadSelector;
+import org.apache.doris.datasource.mvcc.MvccSnapshot;
import org.apache.doris.mysql.privilege.DataMaskPolicy;
import org.apache.doris.mysql.privilege.RowFilterPolicy;
import org.apache.doris.nereids.CascadesContext;
@@ -42,6 +44,7 @@ import org.apache.doris.nereids.SqlCacheContext.CacheKeyType;
import org.apache.doris.nereids.SqlCacheContext.FullColumnName;
import org.apache.doris.nereids.SqlCacheContext.FullTableName;
import org.apache.doris.nereids.SqlCacheContext.ScanTable;
+import org.apache.doris.nereids.SqlCacheContext.TableReference;
import org.apache.doris.nereids.SqlCacheContext.TableVersion;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundVariable;
@@ -67,6 +70,8 @@ import org.apache.doris.qe.SessionVariable;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.qe.cache.CacheAnalyzer;
import org.apache.doris.qe.cache.SqlCache;
+import org.apache.doris.qe.cache.SqlCacheVersion;
+import org.apache.doris.qe.cache.SqlCacheVersionProvider;
import org.apache.doris.rpc.RpcException;
import com.github.benmanes.caffeine.cache.Cache;
@@ -126,8 +131,8 @@ public class NereidsSqlCacheManager {
for (Entry<String, SqlCacheContext> kv : sqlCaches.asMap().entrySet())
{
String key = kv.getKey();
SqlCacheContext context = kv.getValue();
- for (Entry<FullTableName, TableVersion> nameToVersion :
context.getUsedTables().entrySet()) {
- FullTableName tableName = nameToVersion.getKey();
+ for (Entry<TableReference, TableVersion> nameToVersion :
context.getUsedTables().entrySet()) {
+ FullTableName tableName = nameToVersion.getKey().tableName;
TableVersion tableVersion = nameToVersion.getValue();
if (tableVersion.id == tableIf.getId()) {
invalidateKeys.add(key);
@@ -341,7 +346,7 @@ public class NereidsSqlCacheManager {
if (privilegeChanged(connectContext, env, sqlCacheContext)) {
return invalidateCache(key);
}
- IsChanged tableIsChanged = tablesOrDataChanged(env,
sqlCacheContext);
+ IsChanged tableIsChanged = tablesOrDataChanged(env,
sqlCacheContext, connectContext.getStatementContext());
if (tableIsChanged.changed) {
if (tableIsChanged.invalidCache) {
return invalidateCache(key);
@@ -427,7 +432,7 @@ public class NereidsSqlCacheManager {
}
}
- private IsChanged tablesOrDataChanged(Env env, SqlCacheContext
sqlCacheContext) {
+ private IsChanged tablesOrDataChanged(Env env, SqlCacheContext
sqlCacheContext, StatementContext statement) {
if (sqlCacheContext.hasUnsupportedTables()) {
return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
}
@@ -435,13 +440,14 @@ public class NereidsSqlCacheManager {
// the query maybe scan empty partition of the table, we should check
these table version too,
// but the table not exists in sqlCacheContext.getScanTables(), so we
need check here.
// check table type and version
- for (Entry<FullTableName, TableVersion> scanTable :
sqlCacheContext.getUsedTables().entrySet()) {
+ for (Entry<TableReference, TableVersion> scanTable :
sqlCacheContext.getUsedTables().entrySet()) {
TableVersion tableVersion = scanTable.getValue();
if (tableVersion.type != TableType.OLAP && tableVersion.type !=
TableType.MATERIALIZED_VIEW
- && tableVersion.type != TableType.HMS_EXTERNAL_TABLE) {
+ && tableVersion.type != TableType.HMS_EXTERNAL_TABLE
+ && tableVersion.type != TableType.ICEBERG_EXTERNAL_TABLE) {
return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
}
- TableIf tableIf = findTableIf(env, scanTable.getKey());
+ TableIf tableIf = findTableIf(env, scanTable.getKey().tableName);
if (tableIf == null) {
return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
} else if (tableIf.isTemporary()) {
@@ -471,9 +477,24 @@ public class NereidsSqlCacheManager {
return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
}
}
+ } else if (tableVersion.sqlCacheVersion != null) {
+ MvccReadSelector selector = scanTable.getKey().selector;
+ Optional<MvccSnapshot> pin = Optional.empty();
+ if (!selector.isLatest()) {
+ // Reuse the analyzer's resolver: snapshot existence,
timestamp semantics, ref kind,
+ // branch/tag movement and schema selection all remain
owned by the existing MVCC path.
+ statement.loadSnapshots(tableIf,
selector.getTableSnapshot(), selector.getScanParams());
+ pin = statement.getSnapshot(tableIf,
selector.getTableSnapshot(), selector.getScanParams());
+ }
+ Optional<SqlCacheVersion> current = ((SqlCacheVersionProvider)
tableIf)
+ .getSqlCacheVersion(pin);
+ if (!current.isPresent() ||
!tableVersion.sqlCacheVersion.getVersion()
+ .equals(current.get().getVersion())) {
+ return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
+ }
} else if (tableIf instanceof HMSExternalTable) {
- HMSExternalTable hiveTable = (HMSExternalTable) tableIf;
- if (tableVersion.version != hiveTable.getUpdateTime()) {
+ // Preserve the existing Hive update-time validation when no
version provider is available.
+ if (tableVersion.version != ((HMSExternalTable)
tableIf).getUpdateTime()) {
return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
}
} else {
@@ -494,7 +515,8 @@ public class NereidsSqlCacheManager {
return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
}
}
- } else if (!(tableIf instanceof HMSExternalTable)) {
+ } else if (!(tableIf instanceof HMSExternalTable)
+ && !(tableIf instanceof SqlCacheVersionProvider)) {
return IsChanged.CHANGED_AND_INVALIDATE_CACHE;
}
}
@@ -559,7 +581,8 @@ public class NereidsSqlCacheManager {
*/
private boolean tryLockTables(ConnectContext connectContext, Env env,
SqlCacheContext sqlCacheContext) {
StatementContext currentStatementContext =
connectContext.getStatementContext();
- for (FullTableName fullTableName :
sqlCacheContext.getUsedTables().keySet()) {
+ for (TableReference reference :
sqlCacheContext.getUsedTables().keySet()) {
+ FullTableName fullTableName = reference.tableName;
TableIf tableIf = findTableIf(env, fullTableName);
if (tableIf == null) {
return false;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
index a02b5a6662e..60d437f246a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
@@ -1102,6 +1102,9 @@ public class HMSExternalTable extends ExternalTable
implements MTMVRelatedTableI
if (dlaTable instanceof HiveDlaTable) {
return ((HiveDlaTable)
dlaTable).getPartitionSnapshots(partitionNames, snapshot);
}
+ if (dlaTable instanceof IcebergDlaTable) {
+ return IcebergUtils.getPartitionSnapshots(this, partitionNames,
context, snapshot);
+ }
return MTMVRelatedTableIf.super.getPartitionSnapshots(partitionNames,
context, snapshot);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/IcebergDlaTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/IcebergDlaTable.java
index 260fa7e55f4..886eeea112f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/IcebergDlaTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/IcebergDlaTable.java
@@ -68,21 +68,8 @@ public class IcebergDlaTable extends HMSDlaTable {
@Override
public MTMVSnapshotIf getPartitionSnapshot(String partitionName,
MTMVRefreshContext context,
Optional<MvccSnapshot>
snapshot) throws AnalysisException {
- IcebergSnapshotCacheValue snapshotValue =
IcebergUtils.getSnapshotCacheValue(snapshot, hmsTable);
- long latestSnapshotId =
snapshotValue.getPartitionInfo().getLatestSnapshotId(partitionName);
- // If partition snapshot ID is unavailable (<= 0), fallback to table
snapshot ID
- // This can happen when last_updated_snapshot_id is null in Iceberg
metadata
- if (latestSnapshotId <= 0) {
- long tableSnapshotId = snapshotValue.getSnapshot().getSnapshotId();
- // If table snapshot ID is also invalid, it means empty table
- if (tableSnapshotId <= 0) {
- throw new AnalysisException("can not find partition: " +
partitionName
- + ", and table snapshot ID is also invalid");
- }
- // Use table snapshot ID as fallback when partition snapshot ID is
unavailable
- return new MTMVSnapshotIdSnapshot(tableSnapshotId);
- }
- return new MTMVSnapshotIdSnapshot(latestSnapshotId);
+ return IcebergUtils.getPartitionSnapshot(partitionName,
+ IcebergUtils.getSnapshotCacheValue(snapshot, hmsTable));
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeEstimator.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeEstimator.java
index 62922f2a6f9..651398fa83d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeEstimator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeEstimator.java
@@ -96,6 +96,8 @@ final class IcebergCacheSizeEstimator {
private static final long METADATA_ENTRY_WEIGHT = 128L;
private static final long SNAPSHOT_WEIGHT = 512L;
private static final long CURRENT_SNAPSHOT_WEIGHT = 1024L;
+ // Reserve the projection's lazy maximum partition-update timestamp and
its loaded flag.
+ private static final long SNAPSHOT_UPDATE_TIME_STATE_WEIGHT = 32L;
private static final long SNAPSHOT_LOG_WEIGHT = 64L;
private static final long METADATA_LOG_WEIGHT = 160L;
private static final long SNAPSHOT_REF_WEIGHT = 256L;
@@ -161,7 +163,7 @@ final class IcebergCacheSizeEstimator {
static MetaCacheSizeEstimate estimateSnapshotEntry(
IcebergSnapshotEntryKey key, IcebergSnapshotCacheValue value) {
- long bytes = KEY_BASE_WEIGHT;
+ long bytes = KEY_BASE_WEIGHT + SNAPSHOT_UPDATE_TIME_STATE_WEIGHT;
bytes = MetaCacheWeightUtils.saturatedAdd(bytes,
MetaCacheWeightUtils.estimatedNameMappingBytes(key.getNameMapping()));
bytes = MetaCacheWeightUtils.saturatedAdd(bytes,
@@ -169,7 +171,7 @@ final class IcebergCacheSizeEstimator {
bytes = MetaCacheWeightUtils.saturatedAdd(bytes,
MetaCacheWeightUtils.estimatedStringBytes(key.getMetadataFileLocation()));
- IcebergPartitionInfo partitionInfo = value.getPartitionInfo();
+ IcebergPartitionInfo partitionInfo = value.getRetainedPartitionInfo();
bytes = addCount(bytes, partitionInfo.getNameToPartitionItem().size(),
PARTITION_WEIGHT);
bytes = addCount(bytes,
partitionInfo.getNameToIcebergPartition().size(), PARTITION_WEIGHT);
bytes = addCount(bytes,
partitionInfo.getNameToIcebergPartitionNames().size(),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
index 4c31f3b2689..23bc3bf17fa 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
@@ -68,6 +68,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
+import java.util.function.Supplier;
import javax.annotation.Nullable;
/**
@@ -99,6 +100,7 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
public static final String ENGINE = "iceberg";
public static final String ENTRY_TABLE = "table";
public static final String ENTRY_SNAPSHOT = "snapshot";
+ public static final String ENTRY_PARTITION = "partition";
public static final String ENTRY_VIEW = "view";
public static final String ENTRY_MANIFEST = "manifest";
public static final String ENTRY_SCHEMA = "schema";
@@ -107,6 +109,7 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
private final EntryHandle<NameMapping, IcebergTableCacheValue> tableEntry;
private final EntryHandle<IcebergSnapshotEntryKey,
IcebergSnapshotCacheValue> snapshotEntry;
private final EntryHandle<NameMapping, View> viewEntry;
+ private final EntryHandle<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitionEntry;
private final EntryHandle<IcebergManifestEntryKey, ManifestCacheValue>
manifestEntry;
private final EntryHandle<IcebergSchemaCacheKey, SchemaCacheValue>
schemaEntry;
@@ -131,6 +134,12 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
IcebergSnapshotEntryKey.class,
IcebergSnapshotCacheValue.class, defaultEntryCacheSpec(),
MetaCacheEntryInvalidation.forNameMapping(IcebergSnapshotEntryKey::getNameMapping))
.withSizeEstimator((key, value) ->
value.prepareForCachePublication(key)));
+ partitionEntry =
registerEntry(MetaCacheEntryDef.contextualOnly(ENTRY_PARTITION,
+ IcebergSnapshotEntryKey.class, IcebergPartitionInfo.class,
defaultEntryCacheSpec(),
+
MetaCacheEntryInvalidation.forNameMapping(IcebergSnapshotEntryKey::getNameMapping))
+ .withSizeEstimator((key, value) ->
IcebergCacheSizeEstimator.estimateSnapshotEntry(key,
+ new IcebergSnapshotCacheValue(value,
+ new IcebergSnapshot(key.getSnapshotId(),
0)))));
viewEntry = registerEntry(MetaCacheEntryDef.of(ENTRY_VIEW,
NameMapping.class, View.class, this::loadView,
defaultEntryCacheSpec(),
MetaCacheEntryInvalidation.forNameMapping(nameMapping -> nameMapping)));
manifestEntry =
registerEntry(MetaCacheEntryDef.contextualOnly(ENTRY_MANIFEST,
IcebergManifestEntryKey.class,
@@ -384,7 +393,7 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
private IcebergSnapshotCacheValue getSnapshotCache(
ExternalTable dorisTable, NameMapping nameMapping,
IcebergTableCacheValue tableValue) {
Table retainedTable = tableValue.getRetainedIcebergTable();
- java.util.Optional<IcebergSnapshotEntryKey> optionalKey =
+ Optional<IcebergSnapshotEntryKey> optionalKey =
IcebergSnapshotEntryKey.tryCreate(nameMapping, retainedTable);
if (IcebergUtils.useSessionCatalog(dorisTable) ||
!optionalKey.isPresent()) {
boolean isolateForQueries = tableValue.isQueryIsolationPrepared();
@@ -966,6 +975,14 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
snapshots.invalidateIf((key, value) ->
key.getNameMapping().equals(nameMapping)
&& (!key.belongsTo(currentValue) ||
!sharesOperationalResources(currentValue, value)));
}
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions =
+ partitionEntry.getIfInitialized(nameMapping.getCtlId());
+ if (partitions != null) {
+ // Partition maps retain no FileIO; preserve reuse across
handles of the same
+ // physical generation, but retire maps no latest reader can
reach anymore.
+ partitions.invalidateIf(key ->
key.getNameMapping().equals(nameMapping)
+ && !key.belongsTo(currentValue));
+ }
Optional<String> currentUuid = currentValue.getTableUuid();
MetaCacheEntry<IcebergSchemaCacheKey, SchemaCacheValue> schemas =
schemaEntry.getIfInitialized(nameMapping.getCtlId());
@@ -990,6 +1007,11 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
if (snapshots != null) {
snapshots.invalidateIf(key ->
key.getNameMapping().equals(nameMapping));
}
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions =
+ partitionEntry.getIfInitialized(nameMapping.getCtlId());
+ if (partitions != null) {
+ partitions.invalidateIf(key ->
key.getNameMapping().equals(nameMapping));
+ }
MetaCacheEntry<IcebergSchemaCacheKey, SchemaCacheValue> schemas =
schemaEntry.getIfInitialized(nameMapping.getCtlId());
if (schemas != null) {
@@ -1167,27 +1189,44 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
}
try {
IcebergSnapshot latestIcebergSnapshot =
IcebergUtils.getLatestIcebergSnapshot(projectionTable);
- IcebergPartitionInfo icebergPartitionInfo;
- // The projection and its partition eligibility must be classified
from the same
- // retained table generation. The no-argument method performs
another cache lookup,
- // which can cross a concurrent refresh/reset boundary.
+ // Ordinary scans and SQL-cache versions need only this frozen
metadata generation.
+ // Partition consumers load the separate, generation-keyed and
budgeted view on demand.
+ IcebergPartitionInfo icebergPartitionInfo =
IcebergPartitionInfo.empty();
boolean validRelatedTable =
IcebergUtils.isValidRelatedTable(projectionTable);
- if (!validRelatedTable) {
- icebergPartitionInfo = IcebergPartitionInfo.empty();
- } else {
- icebergPartitionInfo =
IcebergUtils.loadPartitionInfo(dorisTable, projectionTable,
- latestIcebergSnapshot.getSnapshotId(),
latestIcebergSnapshot.getSchemaId(), authenticator,
- enableMappingVarbinary, enableMappingTimestampTz);
- }
Optional<Map<Integer, List<String>>> nameMapping =
IcebergUtils.getNameMapping(projectionTable);
- return isolateForQueries
+ IcebergSnapshotCacheValue value = isolateForQueries
? new IcebergSnapshotCacheValue(
icebergPartitionInfo, latestIcebergSnapshot,
nameMapping,
retainedTable, retainedCurrentSnapshotJson)
: new IcebergSnapshotCacheValue(
icebergPartitionInfo, latestIcebergSnapshot,
nameMapping,
retainedTable);
+ if (validRelatedTable) {
+ NameMapping tableIdentity = dorisTable.getOrBuildNameMapping();
+ String displayName = dorisTable.getCatalog().getName() + "." +
dorisTable.getDbName()
+ + "." + dorisTable.getName();
+ Optional<IcebergSnapshotEntryKey> key =
IcebergSnapshotEntryKey.tryCreate(tableIdentity, retainedTable);
+ boolean useSharedCache = key.isPresent() &&
!IcebergUtils.useSessionCatalog(dorisTable);
+ // Retained work can start after reset retired the group.
Capture an existing entry
+ // when applicable; direct loads need only the retained
metadata and authenticator.
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions = useSharedCache
+ ?
partitionEntry.getIfInitialized(key.get().getNameMapping().getCtlId()) : null;
+ Supplier<IcebergPartitionInfo> loader = () -> {
+ try {
+ Table table =
IcebergSnapshotCacheValue.createQueryScopedTable(
+ retainedTable, retainedCurrentSnapshotJson);
+ return IcebergUtils.loadPartitionInfo(displayName,
table,
+ latestIcebergSnapshot.getSnapshotId(),
latestIcebergSnapshot.getSchemaId(),
+ authenticator, enableMappingVarbinary,
enableMappingTimestampTz);
+ } catch (AnalysisException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
+ };
+ value.withPartitionInfoLoader(partitions != null
+ ? () -> getPartitionInfo(key.get(), partitions,
loader) : loader);
+ }
+ return value;
} catch (AnalysisException e) {
throw new RuntimeException(ExceptionUtils.getRootCauseMessage(e),
e);
} catch (UserException e) {
@@ -1196,6 +1235,36 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
}
}
+ private IcebergPartitionInfo getPartitionInfo(IcebergSnapshotEntryKey key,
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions,
+ Supplier<IcebergPartitionInfo> loader) {
+ if (!canCachePartitionInfo(key)) {
+ // A pinned old projection can still read its immutable generation
after retirement.
+ return loader.get();
+ }
+ IcebergPartitionInfo info = partitions.get(key, ignored ->
loader.get());
+ if (!canCachePartitionInfo(key)) {
+ // Replacement can race the load, including a count-mode
contextual miss. Do not leave
+ // the superseded map published after returning it to the query
that still owns it.
+ partitions.invalidateKeyIfSame(key, info);
+ }
+ return info;
+ }
+
+ private boolean canCachePartitionInfo(IcebergSnapshotEntryKey key) {
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> tables =
+ tableEntry.getIfInitialized(key.getNameMapping().getCtlId());
+ if (tables == null) {
+ return false;
+ }
+ if (!tables.isEffectivelyEnabled()) {
+ // A disabled table entry never publishes handles; physical keys
still permit reuse.
+ return true;
+ }
+ IcebergTableCacheValue current =
tables.peekIfPresent(key.getNameMapping());
+ return current != null && key.belongsTo(current);
+ }
+
private IcebergMetadataOps resolveMetadataOps(CatalogIf catalog) {
if (catalog instanceof HMSExternalCatalog) {
return ((HMSExternalCatalog) catalog).getIcebergMetadataOps();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java
index 687f49a4328..8a897b4b606 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java
@@ -44,6 +44,8 @@ import org.apache.doris.mtmv.MTMVSnapshotIdSnapshot;
import org.apache.doris.mtmv.MTMVSnapshotIf;
import org.apache.doris.nereids.trees.plans.commands.info.SortFieldInfo;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.cache.SqlCacheVersion;
+import org.apache.doris.qe.cache.SqlCacheVersionProvider;
import org.apache.doris.statistics.AnalysisInfo;
import org.apache.doris.statistics.BaseAnalysisTask;
import org.apache.doris.statistics.ExternalAnalysisTask;
@@ -56,9 +58,11 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.view.SQLViewRepresentation;
import org.apache.iceberg.view.View;
import org.apache.iceberg.view.ViewVersion;
@@ -71,7 +75,8 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
-public class IcebergExternalTable extends ExternalTable implements
MTMVRelatedTableIf, MTMVBaseTableIf, MvccTable {
+public class IcebergExternalTable extends ExternalTable
+ implements MTMVRelatedTableIf, MTMVBaseTableIf, MvccTable,
SqlCacheVersionProvider {
private boolean isValidRelatedTableCached = false;
private boolean isValidRelatedTable = false;
@@ -84,6 +89,27 @@ public class IcebergExternalTable extends ExternalTable
implements MTMVRelatedTa
super(id, name, remoteName, catalog, db,
TableType.ICEBERG_EXTERNAL_TABLE);
}
+ @Override
+ public Optional<SqlCacheVersion> getSqlCacheVersion(Optional<MvccSnapshot>
snapshot) {
+ return IcebergUtils.withSnapshotCacheValue(snapshot, this,
IcebergExternalTable::sqlCacheVersionOf);
+ }
+
+ static Optional<SqlCacheVersion>
sqlCacheVersionOf(IcebergSnapshotCacheValue value) {
+ Optional<Table> table = value.getIcebergTable();
+ if (!table.isPresent() || !(table.get() instanceof
HasTableOperations)) {
+ return Optional.empty();
+ }
+ TableMetadata metadata = ((HasTableOperations) table.get())
+ .operations().current();
+ if (metadata.uuid() == null || metadata.metadataFileLocation() ==
null) {
+ return Optional.empty();
+ }
+ IcebergSnapshot selected = value.getSnapshot();
+ String identity = metadata.uuid() + "|" +
metadata.metadataFileLocation()
+ + "|" + selected.getSnapshotId() + "|" +
selected.getSchemaId();
+ return Optional.of(new SqlCacheVersion(identity,
metadata.lastUpdatedMillis()));
+ }
+
@Override
public String getMetaCacheEngine() {
return IcebergExternalMetaCache.ENGINE;
@@ -195,21 +221,13 @@ public class IcebergExternalTable extends ExternalTable
implements MTMVRelatedTa
@Override
public MTMVSnapshotIf getPartitionSnapshot(String partitionName,
MTMVRefreshContext context,
Optional<MvccSnapshot>
snapshot) throws AnalysisException {
- IcebergSnapshotCacheValue snapshotValue =
IcebergUtils.getSnapshotCacheValue(snapshot, this);
- long latestSnapshotId =
snapshotValue.getPartitionInfo().getLatestSnapshotId(partitionName);
- // If partition snapshot ID is unavailable (<= 0), fallback to table
snapshot ID
- // This can happen when last_updated_snapshot_id is null in Iceberg
metadata
- if (latestSnapshotId <= 0) {
- long tableSnapshotId = snapshotValue.getSnapshot().getSnapshotId();
- // If table snapshot ID is also invalid, it means empty table
- if (tableSnapshotId <= 0) {
- throw new AnalysisException("can not find partition: " +
partitionName
- + ", and table snapshot ID is also invalid");
- }
- // Use table snapshot ID as fallback when partition snapshot ID is
unavailable
- return new MTMVSnapshotIdSnapshot(tableSnapshotId);
- }
- return new MTMVSnapshotIdSnapshot(latestSnapshotId);
+ return IcebergUtils.getPartitionSnapshot(partitionName,
IcebergUtils.getSnapshotCacheValue(snapshot, this));
+ }
+
+ @Override
+ public Map<String, MTMVSnapshotIf> getPartitionSnapshots(Set<String>
partitionNames,
+ MTMVRefreshContext context, Optional<MvccSnapshot> snapshot)
throws AnalysisException {
+ return IcebergUtils.getPartitionSnapshots(this, partitionNames,
context, snapshot);
}
@Override
@@ -227,9 +245,8 @@ public class IcebergExternalTable extends ExternalTable
implements MTMVRelatedTa
@Override
public long getNewestUpdateVersionOrTime() {
- return IcebergUtils.getLatestSnapshotCacheValue(this)
-
.getPartitionInfo().getNameToIcebergPartition().values().stream()
-
.mapToLong(IcebergPartition::getLastUpdateTime).max().orElse(0);
+ return IcebergUtils.withLatestSnapshotCacheValue(this,
+ IcebergSnapshotCacheValue::getNewestPartitionUpdateTime);
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfo.java
index b618b4b51db..69659428162 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfo.java
@@ -20,6 +20,8 @@ package org.apache.doris.datasource.iceberg;
import org.apache.doris.catalog.PartitionItem;
import org.apache.doris.datasource.metacache.MetaCacheWeightUtils;
+import com.google.common.collect.ImmutableMap;
+
import java.util.Collections;
import java.util.Map;
import java.util.Set;
@@ -40,7 +42,6 @@ public class IcebergPartitionInfo {
private final Map<String, IcebergPartition> nameToIcebergPartition;
private final Map<String, Set<String>> nameToIcebergPartitionNames;
private final long retainedPayloadBytes;
-
private static final IcebergPartitionInfo EMPTY = new
IcebergPartitionInfo();
private IcebergPartitionInfo() {
@@ -63,9 +64,13 @@ public class IcebergPartitionInfo {
Map<String, IcebergPartition>
nameToIcebergPartition,
Map<String, Set<String>>
nameToIcebergPartitionNames,
long retainedPayloadBytes) {
- this.nameToPartitionItem = nameToPartitionItem;
- this.nameToIcebergPartition = nameToIcebergPartition;
- this.nameToIcebergPartitionNames = nameToIcebergPartitionNames;
+ this.nameToPartitionItem = ImmutableMap.copyOf(nameToPartitionItem);
+ this.nameToIcebergPartition =
ImmutableMap.copyOf(nameToIcebergPartition);
+ ImmutableMap.Builder<String, Set<String>> aliases =
+ ImmutableMap.builder();
+ nameToIcebergPartitionNames.forEach((name, names) -> aliases.put(name,
+ com.google.common.collect.ImmutableSet.copyOf(names)));
+ this.nameToIcebergPartitionNames = aliases.build();
this.retainedPayloadBytes = retainedPayloadBytes;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java
index e9728aaf72d..443437d76a2 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java
@@ -44,11 +44,15 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.function.Supplier;
import javax.annotation.Nullable;
public class IcebergSnapshotCacheValue {
private final IcebergPartitionInfo partitionInfo;
+ private Supplier<IcebergPartitionInfo> partitionInfoLoader;
+ private long newestPartitionUpdateTime;
+ private boolean newestPartitionUpdateTimeLoaded;
private final IcebergSnapshot snapshot;
private final Optional<Map<Integer, List<String>>> nameMapping;
private Optional<Table> icebergTable;
@@ -137,9 +141,29 @@ public class IcebergSnapshotCacheValue {
}
public IcebergPartitionInfo getPartitionInfo() {
+ return partitionInfoLoader == null ? partitionInfo :
partitionInfoLoader.get();
+ }
+
+ synchronized long getNewestPartitionUpdateTime() {
+ if (!newestPartitionUpdateTimeLoaded) {
+ // A dictionary check needs only this scalar. Do not retain an
unadmitted full view
+ // in the shared projection, and keep zero as a valid result for
an empty view.
+ newestPartitionUpdateTime =
getPartitionInfo().getNameToIcebergPartition().values().stream()
+
.mapToLong(IcebergPartition::getLastUpdateTime).max().orElse(0);
+ newestPartitionUpdateTimeLoaded = true;
+ }
+ return newestPartitionUpdateTime;
+ }
+
+ IcebergPartitionInfo getRetainedPartitionInfo() {
return partitionInfo;
}
+ IcebergSnapshotCacheValue
withPartitionInfoLoader(Supplier<IcebergPartitionInfo> loader) {
+ this.partitionInfoLoader = loader;
+ return this;
+ }
+
public IcebergSnapshot getSnapshot() {
return snapshot;
}
@@ -222,7 +246,7 @@ public class IcebergSnapshotCacheValue {
}
IcebergSnapshotCacheValue withoutRetainedTable() {
- return new IcebergSnapshotCacheValue(partitionInfo, snapshot,
nameMapping);
+ return new IcebergSnapshotCacheValue(getPartitionInfo(), snapshot,
nameMapping);
}
public Optional<Table> getIcebergTable() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
index e8f499b9cab..3471d9d9726 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
@@ -65,6 +65,10 @@ import
org.apache.doris.datasource.property.metastore.HMSBaseProperties;
import
org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties;
import org.apache.doris.datasource.property.storage.S3Properties;
import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.mtmv.MTMVRefreshContext;
+import org.apache.doris.mtmv.MTMVRelatedTableIf;
+import org.apache.doris.mtmv.MTMVSnapshotIdSnapshot;
+import org.apache.doris.mtmv.MTMVSnapshotIf;
import org.apache.doris.nereids.exceptions.NotSupportedException;
import org.apache.doris.nereids.trees.expressions.literal.Result;
import org.apache.doris.nereids.types.VarBinaryType;
@@ -158,6 +162,7 @@ import java.util.ArrayList;
import java.util.Comparator;
import java.util.EnumMap;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
@@ -1746,11 +1751,7 @@ public class IcebergUtils {
}
} else {
Table table = icebergTable != null ? icebergTable :
getIcebergTable(dorisTable);
- if (schemaId == NEWEST_SCHEMA_ID ||
table.currentSnapshot() == null) {
- schema = table.schema();
- } else {
- schema = table.schemas().get((int) schemaId);
- }
+ schema = schemaForTable(table, schemaId);
}
String type = isView ? "view" : "table";
Preconditions.checkNotNull(schema,
@@ -1764,6 +1765,11 @@ public class IcebergUtils {
}
+ private static Schema schemaForTable(Table table, long schemaId) {
+ return schemaId == NEWEST_SCHEMA_ID || table.currentSnapshot() == null
+ ? table.schema() : table.schemas().get((int) schemaId);
+ }
+
/**
* Parse iceberg schema to doris schema
*/
@@ -2320,15 +2326,35 @@ public class IcebergUtils {
static IcebergPartitionInfo loadPartitionInfo(ExternalTable dorisTable,
Table table, long snapshotId,
long schemaId, ExecutionAuthenticator authenticator,
boolean enableMappingVarbinary, boolean enableMappingTimestampTz)
throws AnalysisException {
+ String displayName = dorisTable.getCatalog().getName() + "." +
dorisTable.getDbName()
+ + "." + dorisTable.getName();
+ return loadPartitionInfo(displayName, table, snapshotId, schemaId,
authenticator,
+ enableMappingVarbinary, enableMappingTimestampTz);
+ }
+
+ static IcebergPartitionInfo loadPartitionInfo(String displayName, Table
table,
+ long snapshotId, long schemaId, ExecutionAuthenticator
authenticator,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz)
throws AnalysisException {
if (snapshotId == IcebergUtils.UNKNOWN_SNAPSHOT_ID) {
return IcebergPartitionInfo.empty();
}
List<IcebergPartition> icebergPartitions;
+ List<Column> partitionColumns;
try {
icebergPartitions = authenticator.execute(() ->
loadIcebergPartition(table, snapshotId));
+ // A retained projection may outlive its catalog cache group. Its
columns and ranges
+ // must use the same frozen table/schema, without looking up a new
catalog generation.
+ partitionColumns = authenticator.execute(() -> {
+ List<Column> schema =
parseSchema(Preconditions.checkNotNull(schemaForTable(table, schemaId),
+ "Schema for Iceberg table %s is null", displayName),
+ enableMappingVarbinary, enableMappingTimestampTz);
+ // This retained-table path bypasses the schema-cache loader's
validation boundary.
+ new SchemaCacheValue(schema).validateSchema();
+ return buildTableSchemaCacheValue(table,
schema).getPartitionColumns();
+ });
} catch (Exception e) {
- String errorMsg = String.format("Failed to get iceberg partition
info, table: %s.%s.%s, snapshotId: %s",
- dorisTable.getCatalog().getName(), dorisTable.getDbName(),
dorisTable.getName(), snapshotId);
+ String errorMsg = String.format("Failed to get iceberg partition
info, table: %s, snapshotId: %s",
+ displayName, snapshotId);
LOG.warn(errorMsg, e);
throw new AnalysisException(errorMsg, e);
}
@@ -2336,9 +2362,6 @@ public class IcebergUtils {
Map<String, PartitionItem> nameToPartitionItem = Maps.newHashMap();
long retainedPayloadBytes = 0L;
- List<Column> partitionColumns =
icebergExternalMetaCache(dorisTable).getIcebergSchemaCacheValue(
- dorisTable.getOrBuildNameMapping(), schemaId, table,
authenticator,
- enableMappingVarbinary,
enableMappingTimestampTz).getPartitionColumns();
for (IcebergPartition partition : icebergPartitions) {
nameToPartition.put(partition.getPartitionName(), partition);
retainedPayloadBytes = MetaCacheWeightUtils.saturatedAdd(
@@ -2626,6 +2649,45 @@ public class IcebergUtils {
return withLatestSnapshotCacheValue(dorisTable, action);
}
+ /** Derive all MV partition versions from one view without retaining an
unbudgeted map. */
+ public static Map<String, MTMVSnapshotIf>
getPartitionSnapshots(ExternalTable dorisTable,
+ Set<String> partitionNames, MTMVRefreshContext context,
Optional<MvccSnapshot> snapshot)
+ throws AnalysisException {
+ Map<String, MTMVSnapshotIf> versions = new LinkedHashMap<>();
+ if (partitionNames.isEmpty()) {
+ return versions;
+ }
+ IcebergSnapshotCacheValue value = getSnapshotCacheValue(snapshot,
dorisTable);
+ IcebergPartitionInfo info = value.getPartitionInfo();
+ for (String name : partitionNames) {
+ try {
+ versions.put(name, partitionSnapshotOf(name, info,
value.getSnapshot().getSnapshotId()));
+ } catch (AnalysisException e) {
+ context.recordPartitionSnapshotFailure((MTMVRelatedTableIf)
dorisTable, name, e);
+ }
+ }
+ return versions;
+ }
+
+ public static MTMVSnapshotIf getPartitionSnapshot(String partitionName,
IcebergSnapshotCacheValue value)
+ throws AnalysisException {
+ return partitionSnapshotOf(partitionName, value.getPartitionInfo(),
value.getSnapshot().getSnapshotId());
+ }
+
+ private static MTMVSnapshotIf partitionSnapshotOf(String partitionName,
+ IcebergPartitionInfo info, long tableSnapshotId) throws
AnalysisException {
+ long latestSnapshotId = info.getLatestSnapshotId(partitionName);
+ // last_updated_snapshot_id can be absent; retain the table-snapshot
fallback for MV refresh.
+ if (latestSnapshotId <= 0) {
+ if (tableSnapshotId <= 0) {
+ throw new AnalysisException("can not find partition: " +
partitionName
+ + ", and table snapshot ID is also invalid");
+ }
+ return new MTMVSnapshotIdSnapshot(tableSnapshotId);
+ }
+ return new MTMVSnapshotIdSnapshot(latestSnapshotId);
+ }
+
public static IcebergSnapshotCacheValue
getSnapshotCacheValue(Optional<MvccSnapshot> snapshot,
ExternalTable dorisTable) {
if (snapshot.isPresent() && snapshot.get() instanceof
IcebergMvccSnapshot) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
index a9cf412aa18..743736f649a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
@@ -74,6 +74,8 @@ import org.apache.doris.planner.PlanNodeId;
import org.apache.doris.planner.ScanContext;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.qe.cache.SqlCacheVersion;
+import org.apache.doris.qe.cache.SqlCacheVersionProvider;
import org.apache.doris.spi.Split;
import org.apache.doris.statistics.StatisticalType;
import org.apache.doris.system.Backend;
@@ -3211,6 +3213,13 @@ public class IcebergScanNode extends FileQueryScanNode {
manifestCacheHits, manifestCacheMisses,
manifestCacheFailures));
}
+ public Optional<SqlCacheVersion> getSqlCacheVersion() {
+ TableIf table = getTargetTable();
+ return table instanceof SqlCacheVersionProvider
+ ? ((SqlCacheVersionProvider) table)
+ .getSqlCacheVersion(getPinnedRelationSnapshot()) :
Optional.empty();
+ }
+
@Override
public TableIf getTargetTable() {
return source.getTargetTable();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/MvccReadSelector.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/MvccReadSelector.java
new file mode 100644
index 00000000000..7610d1b7d18
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/MvccReadSelector.java
@@ -0,0 +1,116 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.mvcc;
+
+import org.apache.doris.analysis.TableScanParams;
+import org.apache.doris.analysis.TableSnapshot;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
+
+/**
+ * Immutable read selector shared by statement snapshot lookup and SQL
result-cache dependencies.
+ * Stores the original selector, never a prior execution's resolved dynamic
parameters. Every resolution
+ * receives fresh copies, so rebinding or PREPARE/EXECUTE cannot mutate a
cached dependency.
+ */
+public final class MvccReadSelector {
+ public static final MvccReadSelector LATEST = new
MvccReadSelector(Optional.empty(), Optional.empty());
+
+ private final Optional<TableSnapshot> tableSnapshot;
+ private final Optional<TableScanParams> scanParams;
+ private final String key;
+
+ private MvccReadSelector(Optional<TableSnapshot> tableSnapshot,
Optional<TableScanParams> scanParams) {
+ this.tableSnapshot = tableSnapshot.map(TableSnapshot::new);
+ this.scanParams = scanParams.map(MvccReadSelector::copyScanParams);
+ this.key = versionKeyOf(this.tableSnapshot, this.scanParams);
+ }
+
+ public static MvccReadSelector of(Optional<TableSnapshot> tableSnapshot,
Optional<TableScanParams> scanParams) {
+ return !tableSnapshot.isPresent() && !scanParams.isPresent()
+ ? LATEST : new MvccReadSelector(tableSnapshot, scanParams);
+ }
+
+ public Optional<TableSnapshot> getTableSnapshot() {
+ return tableSnapshot.map(TableSnapshot::new);
+ }
+
+ public Optional<TableScanParams> getScanParams() {
+ return scanParams.map(MvccReadSelector::copyScanParams);
+ }
+
+ public boolean isLatest() {
+ return key.isEmpty();
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ private static TableScanParams copyScanParams(TableScanParams params) {
+ return new TableScanParams(params.getParamType(),
params.getMapParams(), params.getListParams());
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ return other instanceof MvccReadSelector &&
key.equals(((MvccReadSelector) other).key);
+ }
+
+ @Override
+ public int hashCode() {
+ return key.hashCode();
+ }
+
+ private static String versionKeyOf(Optional<TableSnapshot> tableSnapshot,
Optional<TableScanParams> scanParams) {
+ // Concatenate both selectors (rather than returning on the first) so
the key stays injective even
+ // if a reference ever carried BOTH a snapshot and scan params. Today
the two are mutually exclusive
+ // (e.g. IcebergUtils.getQuerySpecSnapshot rejects FOR-TIME together
with @branch/@tag), so in every
+ // reachable case exactly one part is non-empty and the key is
"v:...", "p:...", or "".
+ StringBuilder key = new StringBuilder();
+ if (tableSnapshot.isPresent()) {
+ TableSnapshot ts = tableSnapshot.get();
+ key.append('v');
+ appendVersionKeyPart(key, ts.getType());
+ appendVersionKeyPart(key, ts.getValue());
+ }
+ if (scanParams.isPresent()) {
+ TableScanParams sp = scanParams.get();
+ key.append('p');
+ appendVersionKeyPart(key, sp.getParamType());
+ Map<String, String> sortedParams = new
TreeMap<>(sp.getMapParams());
+ key.append('m').append(sortedParams.size()).append(':');
+ for (Map.Entry<String, String> entry : sortedParams.entrySet()) {
+ // Length prefixes preserve entry boundaries even when user
values contain Map.toString()
+ // delimiters; sorting separately keeps semantically identical
maps order-independent.
+ appendVersionKeyPart(key, entry.getKey());
+ appendVersionKeyPart(key, entry.getValue());
+ }
+ key.append('l').append(sp.getListParams().size()).append(':');
+ for (String value : sp.getListParams()) {
+ appendVersionKeyPart(key, value);
+ }
+ }
+ return key.toString();
+ }
+
+ private static void appendVersionKeyPart(StringBuilder key, Object value) {
+ String text = String.valueOf(value);
+ key.append(text.length()).append(':').append(text);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
index 77fe3312119..bb9df121ca2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
@@ -135,7 +135,8 @@ public class MTMVRefreshContext {
knownMissing.addAll(stillMissing);
}
- void recordPartitionSnapshotFailure(
+ /** Preserve a per-partition failure reported by a related table's bulk
loader. */
+ public void recordPartitionSnapshotFailure(
MTMVRelatedTableIf table, String partitionName, AnalysisException
failure) {
partitionSnapshotFailureCache.computeIfAbsent(table, ignored -> new
LinkedHashMap<>())
.put(partitionName, failure);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/SqlCacheContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/SqlCacheContext.java
index e532ce611fe..11e454b9442 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/SqlCacheContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/SqlCacheContext.java
@@ -27,6 +27,8 @@ import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.common.Pair;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.hive.HMSExternalTable;
+import org.apache.doris.datasource.mvcc.MvccReadSelector;
+import org.apache.doris.datasource.mvcc.MvccSnapshot;
import org.apache.doris.mysql.FieldInfo;
import org.apache.doris.mysql.privilege.Auth;
import org.apache.doris.mysql.privilege.DataMaskPolicy;
@@ -40,6 +42,8 @@ import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.ResultSet;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.qe.cache.CacheProxy;
+import org.apache.doris.qe.cache.SqlCacheVersion;
+import org.apache.doris.qe.cache.SqlCacheVersionProvider;
import org.apache.doris.thrift.TUniqueId;
import com.google.common.collect.ImmutableList;
@@ -74,7 +78,7 @@ public class SqlCacheContext {
private volatile long latestPartitionVersion = -1;
private volatile long sumOfPartitionNum = -1;
// value: version of table
- private final Map<FullTableName, TableVersion> usedTables =
Maps.newLinkedHashMap();
+ private final Map<TableReference, TableVersion> usedTables =
Maps.newLinkedHashMap();
// value: ddl sql
private final Map<FullTableName, String> usedViews =
Maps.newLinkedHashMap();
// value: usedColumns
@@ -172,6 +176,17 @@ public class SqlCacheContext {
/** addUsedTable */
public synchronized void addUsedTable(TableIf tableIf) {
+ addUsedTable(tableIf, Optional.empty());
+ }
+
+ /** Record the exact MVCC generation consumed by this relation. */
+ public synchronized void addUsedTable(TableIf tableIf,
Optional<MvccSnapshot> snapshot) {
+ addUsedTable(tableIf, snapshot, MvccReadSelector.LATEST);
+ }
+
+ /** Record one table reference independently of other versions of the same
table. */
+ public synchronized void addUsedTable(TableIf tableIf,
Optional<MvccSnapshot> snapshot,
+ MvccReadSelector selector) {
if (tableIf == null) {
return;
}
@@ -190,10 +205,24 @@ public class SqlCacheContext {
}
long version = 0;
+ SqlCacheVersion sqlCacheVersion = null;
try {
+ if (tableIf instanceof SqlCacheVersionProvider) {
+ sqlCacheVersion = ((SqlCacheVersionProvider)
tableIf).getSqlCacheVersion(snapshot).orElse(null);
+ }
+ if (tableIf instanceof SqlCacheVersionProvider && sqlCacheVersion
== null
+ && !(tableIf instanceof HMSExternalTable)) {
+ setHasUnsupportedTables(true);
+ return;
+ }
+ if (!selector.isLatest() && sqlCacheVersion == null) {
+ // Legacy freshness is table-wide, so it cannot validate a
historical/branch dependency.
+ setHasUnsupportedTables(true);
+ return;
+ }
if (tableIf instanceof OlapTable) {
version = ((OlapTable) tableIf).getVisibleVersion();
- } else if (tableIf instanceof HMSExternalTable) {
+ } else if (sqlCacheVersion == null && tableIf instanceof
HMSExternalTable) {
HMSExternalTable hmsExternalTable = (HMSExternalTable) tableIf;
version = hmsExternalTable.getUpdateTime();
}
@@ -204,11 +233,12 @@ public class SqlCacheContext {
}
usedTables.put(
- new FullTableName(database.getCatalog().getName(),
database.getFullName(), tableIf.getName()),
+ new TableReference(new
FullTableName(database.getCatalog().getName(),
+ database.getFullName(), tableIf.getName()), selector),
new TableVersion(
tableIf.getId(),
version,
- tableIf.getType()
+ tableIf.getType(), sqlCacheVersion
)
);
}
@@ -358,7 +388,7 @@ public class SqlCacheContext {
this.cacheProxy = cacheProxy;
}
- public Map<FullTableName, TableVersion> getUsedTables() {
+ public Map<TableReference, TableVersion> getUsedTables() {
return Collections.unmodifiableMap(usedTables);
}
@@ -442,15 +472,18 @@ public class SqlCacheContext {
public synchronized PUniqueId doComputeCacheKeyMd5(
Set<Variable> usedVariables, SessionVariable sessionVariable) {
StringBuilder cacheKey = new
StringBuilder(NereidsParser.removeCommentAndTrimBlank(originSql.trim()));
- for (Entry<FullTableName, TableVersion> entry : usedTables.entrySet())
{
+ for (Entry<TableReference, TableVersion> entry :
usedTables.entrySet()) {
TableVersion tableVersion = entry.getValue();
- FullTableName tableName = entry.getKey();
+ TableReference reference = entry.getKey();
+ String selectorKey = reference.selector.getKey();
cacheKey.append("|")
- .append(tableName)
+ .append(reference.tableName)
+
.append("#").append(selectorKey.length()).append(":").append(selectorKey)
.append("=")
.append(tableVersion.id)
.append("@")
- .append(tableVersion.version);
+ .append(tableVersion.sqlCacheVersion == null ?
Long.toString(tableVersion.version)
+ : "sql-cache:" +
tableVersion.sqlCacheVersion.getVersion());
}
for (Entry<FullTableName, String> entry : usedViews.entrySet()) {
cacheKey.append("|")
@@ -586,6 +619,14 @@ public class SqlCacheContext {
}
}
+ /** A table dependency identified by the original read selector, not by an
alias or resolved snapshot. */
+ @lombok.Data
+ @lombok.AllArgsConstructor
+ public static class TableReference {
+ public final FullTableName tableName;
+ public final MvccReadSelector selector;
+ }
+
/** TableVersion */
@lombok.Data
@lombok.AllArgsConstructor
@@ -593,6 +634,11 @@ public class SqlCacheContext {
public final long id;
public final long version;
public final TableType type;
+ public final SqlCacheVersion sqlCacheVersion;
+
+ public TableVersion(long id, long version, TableType type) {
+ this(id, version, type, null);
+ }
}
/** CacheKeyType */
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
index 19bc96b89f6..5aa6f8cb6ea 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
@@ -44,6 +44,7 @@ import org.apache.doris.datasource.hive.HMSExternalTable;
import org.apache.doris.datasource.hive.HMSExternalTable.DLAType;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergSysExternalTable;
+import org.apache.doris.datasource.mvcc.MvccReadSelector;
import org.apache.doris.datasource.mvcc.MvccSnapshot;
import org.apache.doris.datasource.paimon.PaimonExternalTable;
import org.apache.doris.datasource.paimon.PaimonScanParams;
@@ -107,6 +108,7 @@ import org.apache.doris.nereids.util.RelationUtil;
import org.apache.doris.nereids.util.Utils;
import org.apache.doris.qe.AutoCloseSessionVariable;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.cache.SqlCacheVersionProvider;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -635,9 +637,18 @@ public class BindRelation extends OneAnalysisRuleFactory {
sqlCacheContext.setHasUnsupportedTables(true);
} else if (table instanceof OlapTable) {
sqlCacheContext.addUsedTable(table);
- } else if (table instanceof HMSExternalTable
+ } else if ((table instanceof HMSExternalTable || table
instanceof SqlCacheVersionProvider)
&&
cascadesContext.getConnectContext().getSessionVariable().enableHiveSqlCache) {
- sqlCacheContext.addUsedTable(table);
+ TableScanParams params =
unboundRelation.getScanParams();
+ if (params != null && !params.isBranch() &&
!params.isTag()) {
+ sqlCacheContext.setHasUnsupportedTables(true);
+ } else {
+ MvccReadSelector selector =
MvccReadSelector.of(unboundRelation.getTableSnapshot(),
+ Optional.ofNullable(params));
+ sqlCacheContext.addUsedTable(table,
cascadesContext.getStatementContext()
+ .getSnapshot(table,
selector.getTableSnapshot(), selector.getScanParams()),
+ selector);
+ }
} else {
sqlCacheContext.setHasUnsupportedTables(true);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/cache/CacheAnalyzer.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/CacheAnalyzer.java
index e68ac3c25a9..6c918e5bbd2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/cache/CacheAnalyzer.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/CacheAnalyzer.java
@@ -38,6 +38,7 @@ import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.hive.HMSExternalTable;
import org.apache.doris.datasource.hive.source.HiveScanNode;
+import org.apache.doris.datasource.iceberg.source.IcebergScanNode;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.SqlCacheContext;
@@ -313,9 +314,12 @@ public class CacheAnalyzer {
MetricRepo.COUNTER_QUERY_TABLE.increase(1L);
long olapScanNodeSize = 0;
long hiveScanNodeSize = 0;
+ long versionedScanNodeSize = 0;
for (ScanNode scanNode : scanNodes) {
if (scanNode instanceof OlapScanNode) {
olapScanNodeSize++;
+ } else if (scanNode instanceof IcebergScanNode) {
+ versionedScanNodeSize++;
} else if (scanNode instanceof HiveScanNode) {
hiveScanNodeSize++;
}
@@ -327,7 +331,8 @@ public class CacheAnalyzer {
MetricRepo.COUNTER_QUERY_HIVE_TABLE.increase(1L);
}
- if (!(olapScanNodeSize == scanNodes.size() || hiveScanNodeSize ==
scanNodes.size())) {
+ if (!(olapScanNodeSize == scanNodes.size()
+ || hiveScanNodeSize + versionedScanNodeSize ==
scanNodes.size())) {
if (LOG.isDebugEnabled()) {
LOG.debug("only support olap/hive table with non-federated
query, "
+ "other types are not supported now, queryId {}",
DebugUtil.printId(queryId));
@@ -340,7 +345,9 @@ public class CacheAnalyzer {
ScanNode node = scanNodes.get(i);
CacheTable cTable = node instanceof OlapScanNode
? buildCacheTableForOlapScanNode((OlapScanNode) node)
- : buildCacheTableForHiveScanNode((HiveScanNode) node);
+ : node instanceof IcebergScanNode
+ ?
buildCacheTableForIcebergScanNode((IcebergScanNode) node)
+ :
buildCacheTableForHiveScanNode((HiveScanNode) node);
tblTimeList.add(cTable);
}
Collections.sort(tblTimeList);
@@ -493,6 +500,19 @@ public class CacheAnalyzer {
return cacheTable;
}
+ private CacheTable buildCacheTableForIcebergScanNode(IcebergScanNode node)
{
+ SqlCacheVersion version = node.getSqlCacheVersion()
+ .orElseThrow(() -> new IllegalStateException("No Iceberg SQL
cache version"));
+ CacheTable cacheTable = new CacheTable();
+ cacheTable.table = node.getTargetTable();
+ cacheTable.partitionNum = 1;
+ cacheTable.latestPartitionTime = version.getUpdateTimeMillis();
+ DatabaseIf database = cacheTable.table.getDatabase();
+ scanTables.add(Pair.of(new ScanTable(new
FullTableName(database.getCatalog().getName(),
+ database.getFullName(), cacheTable.table.getName())),
cacheTable.table));
+ return cacheTable;
+ }
+
private CacheTable buildCacheTableForHiveScanNode(HiveScanNode node) {
CacheTable cacheTable = new CacheTable();
cacheTable.table = node.getTargetTable();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/cache/SqlCacheVersion.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/SqlCacheVersion.java
new file mode 100644
index 00000000000..e336290cdd0
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/SqlCacheVersion.java
@@ -0,0 +1,44 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.qe.cache;
+
+import java.util.Objects;
+
+/**
+ * Immutable whole-table SQL result-cache identity and its wall-clock update
time.
+ * The identity is compared for equality, never ordered. It must cover data,
schema and table recreation,
+ * and describe the same pinned generation as the scan. Reading it must not
enumerate files or partitions.
+ * The timestamp is only for the cache's quiet-window eligibility check, never
for identity comparisons.
+ */
+public final class SqlCacheVersion {
+ private final String version;
+ private final long updateTimeMillis;
+
+ public SqlCacheVersion(String version, long updateTimeMillis) {
+ this.version = Objects.requireNonNull(version, "version");
+ this.updateTimeMillis = updateTimeMillis;
+ }
+
+ public String getVersion() {
+ return version;
+ }
+
+ public long getUpdateTimeMillis() {
+ return updateTimeMillis;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/cache/SqlCacheVersionProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/SqlCacheVersionProvider.java
new file mode 100644
index 00000000000..32fcbc4b887
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/SqlCacheVersionProvider.java
@@ -0,0 +1,31 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.qe.cache;
+
+import org.apache.doris.datasource.mvcc.MvccSnapshot;
+
+import java.util.Optional;
+
+/**
+ * Supplies a whole-table SQL result-cache version independently of
MV/dictionary freshness semantics.
+ * A supplied snapshot is the query's read generation; empty probes the
current generation for cache lookup.
+ * Empty results preserve the caller's legacy freshness path. Failures disable
caching for the query.
+ */
+public interface SqlCacheVersionProvider {
+ Optional<SqlCacheVersion> getSqlCacheVersion(Optional<MvccSnapshot>
snapshot);
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCacheTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCacheTest.java
index 5dc212da922..96670c69197 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCacheTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCacheTest.java
@@ -27,6 +27,7 @@ import org.apache.doris.datasource.SchemaCacheValue;
import org.apache.doris.datasource.SessionContext;
import org.apache.doris.datasource.hive.HMSExternalCatalog;
import org.apache.doris.datasource.hive.HMSExternalTable;
+import org.apache.doris.datasource.hive.IcebergDlaTable;
import org.apache.doris.datasource.iceberg.cache.ManifestCacheValue;
import org.apache.doris.datasource.iceberg.source.IcebergTableQueryInfo;
import org.apache.doris.datasource.metacache.EstimatorCalibrationAssertions;
@@ -34,6 +35,8 @@ import org.apache.doris.datasource.metacache.MetaCacheEntry;
import org.apache.doris.datasource.metacache.MetaCacheEntryStats;
import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate;
import org.apache.doris.datasource.metacache.MetaCacheWeightUtils;
+import org.apache.doris.mtmv.MTMVRefreshContext;
+import org.apache.doris.mtmv.MTMVSnapshotIf;
import org.apache.doris.qe.ConnectContext;
import org.apache.hadoop.conf.Configuration;
@@ -789,6 +792,13 @@ public class IcebergExternalMetaCacheTest {
ExecutorService executor = Executors.newSingleThreadExecutor();
IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
IcebergExternalTable dorisTable =
Mockito.mock(IcebergExternalTable.class);
+ NameMapping tableIdentity = NameMapping.createForTest(1L, "db", "tbl");
+ IcebergExternalCatalog catalog =
Mockito.mock(IcebergExternalCatalog.class);
+ Mockito.when(dorisTable.getCatalog()).thenReturn(catalog);
+ Mockito.when(catalog.getName()).thenReturn("ice");
+ Mockito.when(dorisTable.getDbName()).thenReturn("db");
+ Mockito.when(dorisTable.getName()).thenReturn("tbl");
+
Mockito.when(dorisTable.getOrBuildNameMapping()).thenReturn(tableIdentity);
Table projectionTable = Mockito.mock(Table.class);
Snapshot snapshot = Mockito.mock(Snapshot.class);
Schema schema = Mockito.mock(Schema.class);
@@ -806,7 +816,7 @@ public class IcebergExternalMetaCacheTest {
try (MockedStatic<IcebergUtils> icebergUtils = Mockito.mockStatic(
IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
icebergUtils.when(() -> IcebergUtils.loadPartitionInfo(
- dorisTable, projectionTable, 11L, 3L,
capturedAuthenticator, true, false))
+ "ice.db.tbl", projectionTable, 11L, 3L,
capturedAuthenticator, true, false))
.thenReturn(IcebergPartitionInfo.empty());
icebergUtils.when(() ->
IcebergUtils.getNameMapping(projectionTable))
.thenReturn(Optional.empty());
@@ -814,11 +824,15 @@ public class IcebergExternalMetaCacheTest {
.thenReturn(true);
icebergUtils.clearInvocations();
- loader.invoke(cache, dorisTable, projectionTable, projectionTable,
+ IcebergSnapshotCacheValue value = (IcebergSnapshotCacheValue)
loader.invoke(
+ cache, dorisTable, projectionTable, projectionTable,
null, false, capturedAuthenticator, true, false);
icebergUtils.verify(() -> IcebergUtils.loadPartitionInfo(
- dorisTable, projectionTable, 11L, 3L,
capturedAuthenticator, true, false));
+ "ice.db.tbl", projectionTable, 11L, 3L,
capturedAuthenticator, true, false), Mockito.never());
+ value.getPartitionInfo();
+ icebergUtils.verify(() -> IcebergUtils.loadPartitionInfo(
+ "ice.db.tbl", projectionTable, 11L, 3L,
capturedAuthenticator, true, false));
icebergUtils.verify(() ->
IcebergUtils.isValidRelatedTable(projectionTable));
Mockito.verify(dorisTable, Mockito.never()).isValidRelatedTable();
} finally {
@@ -827,6 +841,497 @@ public class IcebergExternalMetaCacheTest {
}
}
+ @Test
+ public void testMvBulkPartitionSnapshotsReuseDisabledPartitionView()
throws Exception {
+ assertMvBulkPartitionSnapshotsReuseUnadmittedView(
+
Collections.singletonMap("meta.cache.iceberg.partition.enable", "false"));
+ }
+
+ @Test
+ public void testMvBulkPartitionSnapshotsReuseWeightRejectedPartitionView()
throws Exception {
+ assertMvBulkPartitionSnapshotsReuseUnadmittedView(
+
Collections.singletonMap("meta.cache.iceberg.partition.max-weight", "1B"));
+ }
+
+ private void assertMvBulkPartitionSnapshotsReuseUnadmittedView(Map<String,
String> properties) throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
+ cache.initCatalog(1L, properties);
+ NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
+ IcebergExternalTable dorisTable = partitionConsumerForTest(identity);
+ Table table =
tableWithMetadataLocation("/metadata/unadmitted-partitions.json");
+ cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class)
+ .put(identity, new IcebergTableCacheValue(table));
+ IcebergPartitionInfo info = realPartitionInfo(3);
+ AtomicInteger loads = new AtomicInteger();
+ ExecutionAuthenticator authenticator = new ExecutionAuthenticator() {};
+ try (MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ utils.when(() ->
IcebergUtils.isValidRelatedTable(Mockito.any(Table.class))).thenReturn(true);
+ utils.when(() ->
IcebergUtils.loadPartitionInfo(Mockito.eq("ice.db.tbl"),
Mockito.any(Table.class),
+ Mockito.anyLong(), Mockito.anyLong(),
Mockito.same(authenticator), Mockito.eq(false), Mockito.eq(false)))
+ .thenAnswer(call -> {
+ loads.incrementAndGet();
+ return info;
+ });
+ IcebergSnapshotCacheValue value =
partitionProjectionForTest(cache, dorisTable, table, authenticator);
+ Optional<org.apache.doris.datasource.mvcc.MvccSnapshot> snapshot =
+ Optional.of(new IcebergMvccSnapshot(value));
+ MTMVRefreshContext context = new MTMVRefreshContext(null);
+ Mockito.when(dorisTable.getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context),
+ Mockito.eq(snapshot))).thenCallRealMethod();
+ Mockito.when(dorisTable.getPartitionSnapshot(Mockito.anyString(),
Mockito.same(context),
+ Mockito.eq(snapshot))).thenCallRealMethod();
+ Map<String, MTMVSnapshotIf> versions =
dorisTable.getPartitionSnapshots(
+ new java.util.LinkedHashSet<>(Arrays.asList("part=0",
"part=1", "part=2")), context, snapshot);
+ Assert.assertEquals(3, versions.size());
+ versions.values().forEach(version -> Assert.assertEquals(1L,
version.getSnapshotVersion()));
+ Assert.assertEquals("one full view load for the complete refresh
batch", 1, loads.get());
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions = cache.entry(
+ 1L, IcebergExternalMetaCache.ENTRY_PARTITION,
+ IcebergSnapshotEntryKey.class, IcebergPartitionInfo.class);
+
Assert.assertNull(partitions.peekIfPresent(IcebergSnapshotEntryKey.tryCreate(identity,
table).get()));
+ Assert.assertSame(IcebergPartitionInfo.empty(),
value.getRetainedPartitionInfo());
+ dorisTable.getPartitionSnapshot("part=0", context, snapshot);
+ Assert.assertEquals("the shared snapshot must not retain an
unbudgeted view", 2, loads.get());
+ } finally {
+ cache.close();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testDictionaryChecksReuseVersionWhenPartitionCacheDisabled()
throws Exception {
+
assertDictionaryChecksReuseScalar(Collections.singletonMap("meta.cache.iceberg.partition.enable",
"false"),
+ false);
+ }
+
+ @Test
+ public void testDictionaryChecksReuseVersionWhenPartitionViewRejected()
throws Exception {
+
assertDictionaryChecksReuseScalar(Collections.singletonMap("meta.cache.iceberg.partition.max-weight",
"1B"),
+ false);
+ }
+
+ @Test
+ public void
testDictionaryChecksReuseVersionAfterPartitionEntryRetirement() throws
Exception {
+ assertDictionaryChecksReuseScalar(Collections.emptyMap(), true);
+ }
+
+ private void assertDictionaryChecksReuseScalar(Map<String, String>
properties, boolean retired) throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
+ cache.initCatalog(1L, properties);
+ NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
+ IcebergExternalTable table = partitionConsumerForTest(identity);
+
Mockito.when(table.getNewestUpdateVersionOrTime()).thenCallRealMethod();
+ Table first =
tableWithMetadataLocation("/metadata/dictionary-v1.json");
+ cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class)
+ .put(identity, new IcebergTableCacheValue(first));
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions = cache.entry(
+ 1L, IcebergExternalMetaCache.ENTRY_PARTITION,
+ IcebergSnapshotEntryKey.class, IcebergPartitionInfo.class);
+ AtomicInteger loads = new AtomicInteger();
+ AtomicReference<IcebergSnapshotCacheValue> projection = new
AtomicReference<>();
+ AtomicReference<IcebergPartitionInfo> info = new
AtomicReference<>(partitionUpdateTimes(3L, 7L, 5L));
+ ExecutionAuthenticator authenticator = new ExecutionAuthenticator() {};
+ try (MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ utils.when(() ->
IcebergUtils.isValidRelatedTable(Mockito.any(Table.class))).thenReturn(true);
+ utils.when(() ->
IcebergUtils.loadPartitionInfo(Mockito.eq("ice.db.tbl"),
Mockito.any(Table.class),
+ Mockito.anyLong(), Mockito.anyLong(),
Mockito.same(authenticator), Mockito.eq(false),
+ Mockito.eq(false))).thenAnswer(call -> {
+ loads.incrementAndGet();
+ return info.get();
+ });
+ utils.when(() -> IcebergUtils.getLatestSnapshotCacheValue(table))
+ .thenAnswer(call ->
projection.get().withoutRetainedTable());
+ utils.when(() ->
IcebergUtils.withLatestSnapshotCacheValue(Mockito.same(table),
+
Mockito.<java.util.function.Function<IcebergSnapshotCacheValue, Long>>any()))
+ .thenAnswer(call -> {
+ java.util.function.Function<IcebergSnapshotCacheValue,
Long> action = call.getArgument(1);
+ return action.apply(projection.get());
+ });
+ projection.set(partitionProjectionForTest(cache, table, first,
authenticator));
+ if (retired) {
+ cache.invalidateCatalog(1L);
+ }
+ for (int check = 0; check < 3; check++) {
+ Assert.assertEquals(7L, table.getNewestUpdateVersionOrTime());
+ }
+ Assert.assertEquals("repeated dictionary checks must not reload
the full view", 1, loads.get());
+ Assert.assertSame(IcebergPartitionInfo.empty(),
projection.get().getRetainedPartitionInfo());
+
Assert.assertNull(partitions.peekIfPresent(IcebergSnapshotEntryKey.tryCreate(identity,
first).get()));
+
+ cache.initCatalog(1L, properties);
+ Table second =
tableWithMetadataLocation("/metadata/dictionary-v2.json");
+ cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class)
+ .put(identity, new IcebergTableCacheValue(second));
+ info.set(partitionUpdateTimes(11L, 9L));
+ projection.set(partitionProjectionForTest(cache, table, second,
authenticator));
+ Assert.assertEquals(11L, table.getNewestUpdateVersionOrTime());
+ Assert.assertEquals(11L, table.getNewestUpdateVersionOrTime());
+ Assert.assertEquals("a new metadata generation must compute its
own version", 2, loads.get());
+ Assert.assertSame(IcebergPartitionInfo.empty(),
projection.get().getRetainedPartitionInfo());
+ Table empty =
tableWithMetadataLocation("/metadata/dictionary-empty.json");
+ cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class)
+ .put(identity, new IcebergTableCacheValue(empty));
+ info.set(IcebergPartitionInfo.empty());
+ projection.set(partitionProjectionForTest(cache, table, empty,
authenticator));
+ Assert.assertEquals(0L, table.getNewestUpdateVersionOrTime());
+ Assert.assertEquals(0L, table.getNewestUpdateVersionOrTime());
+ Assert.assertEquals("an empty view's zero version is memoized
too", 3, loads.get());
+ } finally {
+ cache.close();
+ executor.shutdownNow();
+ }
+ }
+
+ private IcebergPartitionInfo partitionUpdateTimes(long... updateTimes) {
+ Map<String, IcebergPartition> partitions = new HashMap<>();
+ for (int i = 0; i < updateTimes.length; i++) {
+ String name = "part=" + i;
+ partitions.put(name, new IcebergPartition(name, 0, 1L, 1L, 1L,
updateTimes[i], 1L,
+ Collections.singletonList(Integer.toString(i)),
Collections.singletonList("day")));
+ }
+ return new IcebergPartitionInfo(Collections.emptyMap(), partitions,
Collections.emptyMap());
+ }
+
+ @Test
+ public void testTableReplacementRetiresOnlyObsoletePartitionViews() throws
Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
+ try {
+ cache.initCatalog(1L, Collections.emptyMap());
+ NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
+ Table first =
tableWithMetadataLocation("/metadata/partition-replace-v1.json");
+ Table second =
tableWithMetadataLocation("/metadata/partition-replace-v2.json");
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> tables =
cache.entry(
+ 1L, IcebergExternalMetaCache.ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class);
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions = cache.entry(
+ 1L, IcebergExternalMetaCache.ENTRY_PARTITION,
+ IcebergSnapshotEntryKey.class, IcebergPartitionInfo.class);
+ IcebergSnapshotEntryKey oldKey =
IcebergSnapshotEntryKey.tryCreate(identity, first).get();
+ IcebergSnapshotEntryKey newKey =
IcebergSnapshotEntryKey.tryCreate(identity, second).get();
+ IcebergSnapshotEntryKey unrelatedKey =
IcebergSnapshotEntryKey.tryCreate(
+ NameMapping.createForTest(1L, "db", "other"), first).get();
+ IcebergPartitionInfo info = realPartitionInfo(1);
+ tables.put(identity, new IcebergTableCacheValue(first));
+ partitions.put(oldKey, info);
+ partitions.put(unrelatedKey, info);
+ tables.put(identity, new IcebergTableCacheValue(first));
+ Assert.assertSame("same metadata generation stays reusable", info,
partitions.peekIfPresent(oldKey));
+ partitions.put(newKey, info);
+ tables.put(identity, new IcebergTableCacheValue(second));
+ Assert.assertNull(partitions.peekIfPresent(oldKey));
+ Assert.assertSame(info, partitions.peekIfPresent(newKey));
+ Assert.assertSame(info, partitions.peekIfPresent(unrelatedKey));
+ } finally {
+ cache.close();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testOldPartitionLoadsCannotRepublishAfterTableReplacement()
throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
+ cache.initCatalog(1L, Collections.emptyMap());
+ NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
+ IcebergExternalTable dorisTable = partitionConsumerForTest(identity);
+ Table oldTable =
tableWithMetadataLocation("/metadata/partition-load-v1.json");
+ Table newTable =
tableWithMetadataLocation("/metadata/partition-load-v2.json");
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> tables =
cache.entry(
+ 1L, IcebergExternalMetaCache.ENTRY_TABLE, NameMapping.class,
IcebergTableCacheValue.class);
+ tables.put(identity, new IcebergTableCacheValue(oldTable));
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo>
partitions = cache.entry(
+ 1L, IcebergExternalMetaCache.ENTRY_PARTITION,
+ IcebergSnapshotEntryKey.class, IcebergPartitionInfo.class);
+ IcebergSnapshotEntryKey oldKey =
IcebergSnapshotEntryKey.tryCreate(identity, oldTable).get();
+ IcebergPartitionInfo info = realPartitionInfo(1);
+ AtomicInteger loads = new AtomicInteger();
+ java.util.concurrent.atomic.AtomicBoolean replaceDuringLoad = new
java.util.concurrent.atomic.AtomicBoolean(true);
+ ExecutionAuthenticator authenticator = new ExecutionAuthenticator() {};
+ try (MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ utils.when(() ->
IcebergUtils.isValidRelatedTable(Mockito.any(Table.class))).thenReturn(true);
+ utils.when(() ->
IcebergUtils.loadPartitionInfo(Mockito.eq("ice.db.tbl"),
Mockito.any(Table.class),
+ Mockito.anyLong(), Mockito.anyLong(),
Mockito.same(authenticator), Mockito.eq(false), Mockito.eq(false)))
+ .thenAnswer(call -> {
+ loads.incrementAndGet();
+ if (replaceDuringLoad.getAndSet(false)) {
+ tables.put(identity, new
IcebergTableCacheValue(newTable));
+ }
+ return info;
+ });
+ IcebergSnapshotCacheValue old = partitionProjectionForTest(cache,
dorisTable, oldTable, authenticator);
+ Assert.assertSame(info, old.getPartitionInfo());
+ Assert.assertNull("replacement during loading must fence
publication", partitions.peekIfPresent(oldKey));
+ Assert.assertSame(info, old.getPartitionInfo());
+ Assert.assertNull("late access through a pinned old projection
stays uncached",
+ partitions.peekIfPresent(oldKey));
+ IcebergSnapshotCacheValue current =
partitionProjectionForTest(cache, dorisTable, newTable, authenticator);
+ Assert.assertSame(current.getPartitionInfo(),
current.getPartitionInfo());
+ Assert.assertEquals(3, loads.get());
+ } finally {
+ cache.close();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testHmsIcebergBulkSnapshotsReuseOnePartitionView() throws
Exception {
+ HMSExternalCatalog catalog = Mockito.mock(HMSExternalCatalog.class);
+ org.apache.doris.datasource.hive.HMSExternalDatabase database =
+
Mockito.mock(org.apache.doris.datasource.hive.HMSExternalDatabase.class);
+ Mockito.when(database.getFullName()).thenReturn("db");
+ Mockito.when(database.getRemoteName()).thenReturn("db");
+ HMSExternalTable table = new HMSExternalTable(1L, "tbl", "tbl",
catalog, database) {
+ @Override
+ protected void makeSureInitialized() {
+ // This test installs the Iceberg DLA and pinned snapshot
directly.
+ }
+ };
+ Field dla = HMSExternalTable.class.getDeclaredField("dlaTable");
+ dla.setAccessible(true);
+ dla.set(table, new IcebergDlaTable(table));
+ AtomicInteger loads = new AtomicInteger();
+ IcebergPartitionInfo info = realPartitionInfo(2);
+ IcebergSnapshotCacheValue value = new IcebergSnapshotCacheValue(
+ IcebergPartitionInfo.empty(), new IcebergSnapshot(11L, 0L))
+ .withPartitionInfoLoader(() -> {
+ loads.incrementAndGet();
+ return info;
+ });
+ Map<String, MTMVSnapshotIf> versions = table.getPartitionSnapshots(
+ new java.util.LinkedHashSet<>(Arrays.asList("part=0",
"part=1")),
+ new MTMVRefreshContext(null), Optional.of(new
IcebergMvccSnapshot(value)));
+ Assert.assertEquals(2, versions.size());
+ Assert.assertEquals(1, loads.get());
+ }
+
+ @Test
+ public void testBulkSnapshotsPreserveTableFallbackAndPerPartitionFailure()
throws Exception {
+ IcebergExternalTable table =
partitionConsumerForTest(NameMapping.createForTest(1L, "db", "tbl"));
+ MTMVRefreshContext context = Mockito.mock(MTMVRefreshContext.class);
+ Mockito.when(table.getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context), Mockito.any()))
+ .thenCallRealMethod();
+ IcebergPartitionInfo info = Mockito.mock(IcebergPartitionInfo.class);
+
Mockito.when(info.getLatestSnapshotId("unknown_version")).thenReturn(0L);
+
Mockito.when(info.getLatestSnapshotId("known_version")).thenReturn(42L);
+ Set<String> names = new
java.util.LinkedHashSet<>(Arrays.asList("unknown_version", "known_version"));
+ IcebergSnapshotCacheValue valid = new IcebergSnapshotCacheValue(info,
new IcebergSnapshot(77L, 0L));
+ Map<String, MTMVSnapshotIf> versions =
table.getPartitionSnapshots(names, context,
+ Optional.of(new IcebergMvccSnapshot(valid)));
+ Assert.assertEquals(77L,
versions.get("unknown_version").getSnapshotVersion());
+ Assert.assertEquals(42L,
versions.get("known_version").getSnapshotVersion());
+ IcebergSnapshotCacheValue empty = new IcebergSnapshotCacheValue(info,
new IcebergSnapshot(-1L, 0L));
+ Map<String, MTMVSnapshotIf> partial =
table.getPartitionSnapshots(names, context,
+ Optional.of(new IcebergMvccSnapshot(empty)));
+ Assert.assertEquals(Collections.singleton("known_version"),
partial.keySet());
+
Mockito.verify(context).recordPartitionSnapshotFailure(Mockito.same(table),
Mockito.eq("unknown_version"),
+ Mockito.argThat(error -> error.getMessage().contains("table
snapshot ID is also invalid")));
+ Assert.assertTrue(table.getPartitionSnapshots(Collections.emptySet(),
context, Optional.empty()).isEmpty());
+ }
+
+ private IcebergExternalTable partitionConsumerForTest(NameMapping
identity) {
+ IcebergExternalTable table = Mockito.mock(IcebergExternalTable.class);
+ IcebergExternalCatalog catalog =
Mockito.mock(IcebergExternalCatalog.class);
+ Mockito.when(table.getCatalog()).thenReturn(catalog);
+ Mockito.when(catalog.getName()).thenReturn("ice");
+ Mockito.when(table.getDbName()).thenReturn("db");
+ Mockito.when(table.getName()).thenReturn("tbl");
+ Mockito.when(table.getOrBuildNameMapping()).thenReturn(identity);
+ return table;
+ }
+
+ private IcebergSnapshotCacheValue
partitionProjectionForTest(IcebergExternalMetaCache cache,
+ IcebergExternalTable table, Table retainedTable,
ExecutionAuthenticator authenticator) throws Exception {
+ Method loader = IcebergExternalMetaCache.class.getDeclaredMethod(
+ "loadSnapshotProjection", ExternalTable.class, Table.class,
Table.class,
+ String.class, boolean.class, ExecutionAuthenticator.class,
boolean.class, boolean.class);
+ loader.setAccessible(true);
+ return (IcebergSnapshotCacheValue) loader.invoke(cache, table,
+ retainedTable, retainedTable, null, false, authenticator,
false, false);
+ }
+
+ @Test
+ public void
testLazyRetainedPartitionRejectsCaseInsensitiveDuplicateColumns() throws
Exception {
+ assertDuplicateSchemaRejectedByRetainedPartitionLoad(false);
+ }
+
+ @Test
+ public void
testRetiredPartitionProjectionRejectsCaseInsensitiveDuplicateColumns() throws
Exception {
+ assertDuplicateSchemaRejectedByRetainedPartitionLoad(true);
+ }
+
+ private void assertDuplicateSchemaRejectedByRetainedPartitionLoad(boolean
resetBeforeLoad) throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
+ NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
+ IcebergExternalTable dorisTable = partitionConsumerForTest(identity);
+ Schema schema = new Schema(Types.NestedField.optional(1, "dt",
Types.DateType.get()),
+ Types.NestedField.optional(2, "DT", Types.IntegerType.get()));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).day("dt").build();
+ String location =
temporaryFolder.newFolder("retained-partition-duplicate-schema").toURI().toString();
+ Table table = new HadoopTables(new Configuration()).create(schema,
spec, location);
+ PartitionData partition = new PartitionData(spec.rawPartitionType());
+ partition.set(0, 19724);
+ table.newFastAppend().appendFile(DataFiles.builder(spec)
+ .withPath(location +
"/data/old.parquet").withPartition(partition)
+ .withFileSizeInBytes(10).withRecordCount(1).build()).commit();
+ IcebergTableCacheValue generation = new IcebergTableCacheValue(table);
+ try (IcebergTableCacheValue.Lease retained = generation.tryAcquire();
+ MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ Assert.assertNotNull(retained);
+ utils.when(() ->
IcebergUtils.useSessionCatalog(dorisTable)).thenReturn(false);
+ cache.initCatalog(1L, Collections.emptyMap());
+ cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE,
+ NameMapping.class,
IcebergTableCacheValue.class).put(identity, generation);
+ IcebergSnapshotCacheValue projection = partitionProjectionForTest(
+ cache, dorisTable, retained.getIcebergTable(), new
ExecutionAuthenticator() { });
+ if (resetBeforeLoad) {
+ cache.invalidateCatalog(1L);
+ }
+ RuntimeException error =
Assert.assertThrows(RuntimeException.class, projection::getPartitionInfo);
+
Assert.assertTrue(org.apache.commons.lang3.exception.ExceptionUtils.getRootCauseMessage(error)
+ .contains("Duplicate column name found: DT"));
+
Assert.assertTrue(org.apache.commons.lang3.exception.ExceptionUtils.getThrowableList(error).stream()
+ .anyMatch(cause -> cause instanceof
org.apache.doris.common.AnalysisException
+ && cause.getMessage().contains("ice.db.tbl")
+ &&
cause.getMessage().contains(Long.toString(table.currentSnapshot().snapshotId()))));
+ Mockito.verify(dorisTable.getCatalog(),
Mockito.never()).getExecutionAuthenticator();
+ } finally {
+ cache.close();
+ generation.retire();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testLazyPartitionLoadSurvivesCatalogReset() throws Exception {
+ assertRetainedPartitionLoadAfterReset(false, false);
+ }
+
+ @Test
+ public void testPartitionProjectionCreatedAfterCatalogReset() throws
Exception {
+ assertRetainedPartitionLoadAfterReset(true, false);
+ }
+
+ @Test
+ public void testSessionPartitionProjectionNeedsNoCatalogCacheGroup()
throws Exception {
+ assertRetainedPartitionLoadAfterReset(true, true);
+ }
+
+ private void assertRetainedPartitionLoadAfterReset(boolean
resetBeforeProjection, boolean sessionCatalog)
+ throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
+ NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
+ IcebergExternalTable dorisTable = partitionConsumerForTest(identity);
+ Schema schema = new Schema(Types.NestedField.optional(1, "dt",
Types.DateType.get()));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).day("dt").build();
+ String location =
temporaryFolder.newFolder("retained-partition-reset").toURI().toString();
+ Table table = new HadoopTables(new Configuration()).create(schema,
spec, location);
+ PartitionData partition = new PartitionData(spec.rawPartitionType());
+ partition.set(0, 19724);
+ table.newFastAppend().appendFile(DataFiles.builder(spec)
+ .withPath(location +
"/data/old.parquet").withPartition(partition)
+ .withFileSizeInBytes(10).withRecordCount(1).build()).commit();
+ AtomicInteger authenticatedCalls = new AtomicInteger();
+ ExecutionAuthenticator captured = new ExecutionAuthenticator() {
+ @Override
+ public <T> T execute(Callable<T> action) throws Exception {
+ authenticatedCalls.incrementAndGet();
+ return action.call();
+ }
+ };
+ IcebergTableCacheValue generation = new IcebergTableCacheValue(table);
+ try (IcebergTableCacheValue.Lease retained = generation.tryAcquire();
+ MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ Assert.assertNotNull(retained);
+ utils.when(() ->
IcebergUtils.useSessionCatalog(dorisTable)).thenReturn(sessionCatalog);
+ if (!sessionCatalog) {
+ cache.initCatalog(1L, Collections.emptyMap());
+ cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE,
+ NameMapping.class, IcebergTableCacheValue.class)
+ .put(identity, generation);
+ if (resetBeforeProjection) {
+ cache.invalidateCatalog(1L);
+ }
+ }
+ IcebergSnapshotCacheValue projection = partitionProjectionForTest(
+ cache, dorisTable, retained.getIcebergTable(), captured);
+ Assert.assertEquals("partition enumeration remains lazy", 0,
authenticatedCalls.get());
+ if (!resetBeforeProjection) {
+ cache.invalidateCatalog(1L);
+ }
+ IcebergPartitionInfo info = projection.getPartitionInfo();
+ Assert.assertEquals(1, info.getNameToPartitionItem().size());
+ Assert.assertEquals(1, info.getNameToIcebergPartition().size());
+ Assert.assertEquals(table.currentSnapshot().snapshotId(),
+
info.getNameToIcebergPartition().values().iterator().next().getLastSnapshotId());
+ Assert.assertTrue(authenticatedCalls.get() > 0);
+ Mockito.verify(dorisTable.getCatalog(),
Mockito.never()).getExecutionAuthenticator();
+ // Retired work must not populate the next catalog generation's
budgeted entries.
+ cache.initCatalog(1L, Collections.emptyMap());
+ Assert.assertNull(cache.entry(1L,
IcebergExternalMetaCache.ENTRY_PARTITION,
+ IcebergSnapshotEntryKey.class, IcebergPartitionInfo.class)
+ .peekIfPresent(IcebergSnapshotEntryKey.tryCreate(identity,
table).get()));
+ } finally {
+ cache.close();
+ generation.retire();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void
testPartitionViewLoadsOncePerPhysicalGenerationAndRefreshInvalidatesIt() throws
Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(executor);
+ cache.initCatalog(1L, Collections.emptyMap());
+ NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
+ IcebergExternalTable dorisTable =
Mockito.mock(IcebergExternalTable.class);
+ IcebergExternalCatalog catalog =
Mockito.mock(IcebergExternalCatalog.class);
+ Mockito.when(dorisTable.getCatalog()).thenReturn(catalog);
+ Mockito.when(catalog.getName()).thenReturn("ice");
+ Mockito.when(dorisTable.getDbName()).thenReturn("db");
+ Mockito.when(dorisTable.getName()).thenReturn("tbl");
+ Mockito.when(dorisTable.getOrBuildNameMapping()).thenReturn(identity);
+ Table first = tableWithMetadataLocation("/metadata/partition-v1.json");
+ Table second =
tableWithMetadataLocation("/metadata/partition-v2.json");
+ cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class)
+ .put(identity, new IcebergTableCacheValue(first));
+ ExecutionAuthenticator authenticator = new ExecutionAuthenticator() {
};
+ java.util.concurrent.atomic.AtomicInteger loads = new
java.util.concurrent.atomic.AtomicInteger();
+ Method loader = IcebergExternalMetaCache.class.getDeclaredMethod(
+ "loadSnapshotProjection", ExternalTable.class, Table.class,
Table.class,
+ String.class, boolean.class, ExecutionAuthenticator.class,
boolean.class, boolean.class);
+ loader.setAccessible(true);
+ try (MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ utils.when(() ->
IcebergUtils.isValidRelatedTable(Mockito.any(Table.class))).thenReturn(true);
+ utils.when(() ->
IcebergUtils.loadPartitionInfo(Mockito.eq("ice.db.tbl"),
Mockito.any(Table.class),
+ Mockito.anyLong(), Mockito.anyLong(),
Mockito.same(authenticator), Mockito.eq(false), Mockito.eq(false)))
+ .thenAnswer(call -> {
+ loads.incrementAndGet();
+ return IcebergPartitionInfo.empty();
+ });
+ IcebergSnapshotCacheValue one = (IcebergSnapshotCacheValue)
loader.invoke(cache, dorisTable,
+ first, first, null, false, authenticator, false, false);
+ IcebergSnapshotCacheValue same = (IcebergSnapshotCacheValue)
loader.invoke(cache, dorisTable,
+ first, first, null, false, authenticator, false, false);
+ Assert.assertEquals(0, loads.get());
+ Assert.assertSame(one.getPartitionInfo(), same.getPartitionInfo());
+ Assert.assertEquals(1, loads.get());
+ cache.invalidateTable(1L, "db", "tbl");
+ one.getPartitionInfo();
+ Assert.assertEquals(2, loads.get());
+ IcebergSnapshotCacheValue newer = (IcebergSnapshotCacheValue)
loader.invoke(cache, dorisTable,
+ second, second, null, false, authenticator, false, false);
+ newer.getPartitionInfo();
+ Assert.assertEquals(3, loads.get());
+ } finally {
+ cache.close();
+ executor.shutdownNow();
+ }
+ }
+
@Test
public void
testHmsSnapshotProjectionIgnoresRelatedTableCacheFromAnotherGeneration() throws
Exception {
ExecutorService executor = Executors.newSingleThreadExecutor();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfoTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfoTest.java
index c184debd3d7..a4b50f30921 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfoTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergPartitionInfoTest.java
@@ -65,7 +65,8 @@ public class IcebergPartitionInfoTest {
names.add("p2");
nameToIcebergPartitionNames.put("p1", names);
- IcebergPartitionInfo info = new IcebergPartitionInfo(null,
nameToIcebergPartition, nameToIcebergPartitionNames);
+ IcebergPartitionInfo info = new IcebergPartitionInfo(
+ Collections.emptyMap(), nameToIcebergPartition,
nameToIcebergPartitionNames);
long snapshot1 = info.getLatestSnapshotId("p1");
long snapshot2 = info.getLatestSnapshotId("p2");
long snapshot3 = info.getLatestSnapshotId("p3");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergSqlCacheVersionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergSqlCacheVersionTest.java
new file mode 100644
index 00000000000..4ccd879410d
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergSqlCacheVersionTest.java
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.iceberg;
+
+import org.apache.doris.analysis.TableScanParams;
+import org.apache.doris.analysis.TableSnapshot;
+import org.apache.doris.datasource.mvcc.MvccReadSelector;
+import org.apache.doris.nereids.SqlCacheContext;
+import org.apache.doris.qe.cache.SqlCacheVersion;
+
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.Optional;
+
+class IcebergSqlCacheVersionTest {
+ private SqlCacheVersion version(String uuid, String metadataFile, long
snapshotId, long schemaId) {
+ IcebergSnapshotCacheValue value =
Mockito.mock(IcebergSnapshotCacheValue.class);
+ Table table = Mockito.mock(Table.class,
Mockito.withSettings().extraInterfaces(HasTableOperations.class));
+ TableOperations operations = Mockito.mock(TableOperations.class);
+ TableMetadata metadata = Mockito.mock(TableMetadata.class);
+ Mockito.when(((HasTableOperations)
table).operations()).thenReturn(operations);
+ Mockito.when(operations.current()).thenReturn(metadata);
+ Mockito.when(metadata.uuid()).thenReturn(uuid);
+ Mockito.when(metadata.metadataFileLocation()).thenReturn(metadataFile);
+ Mockito.when(metadata.lastUpdatedMillis()).thenReturn(1234L);
+ Mockito.when(value.getIcebergTable()).thenReturn(Optional.of(table));
+ Mockito.when(value.getSnapshot()).thenReturn(new
IcebergSnapshot(snapshotId, schemaId));
+ Optional<SqlCacheVersion> result =
IcebergExternalTable.sqlCacheVersionOf(value);
+ Mockito.verify(value, Mockito.never()).getPartitionInfo();
+ return result.orElse(null);
+ }
+
+ @Test
+ void
versionCoversDataSchemaRecreationAndMetadataChangesWithoutPartitionReads() {
+ SqlCacheVersion original = version("uuid", "metadata-1", 11, 3);
+ Assertions.assertEquals(1234L, original.getUpdateTimeMillis());
+ Assertions.assertEquals(original.getVersion(), version("uuid",
"metadata-1", 11, 3).getVersion());
+ Assertions.assertNotEquals(original.getVersion(), version("uuid",
"metadata-1", 12, 3).getVersion());
+ Assertions.assertNotEquals(original.getVersion(), version("uuid",
"metadata-1", 11, 4).getVersion());
+ Assertions.assertNotEquals(original.getVersion(), version("recreated",
"metadata-1", 11, 3).getVersion());
+ Assertions.assertNotEquals(original.getVersion(), version("uuid",
"metadata-2", 11, 3).getVersion());
+ }
+
+ @Test
+ void missingPhysicalIdentityDisablesCaching() {
+ Assertions.assertNull(version(null, "metadata-1", 11, 3));
+ Assertions.assertNull(version("uuid", null, 11, 3));
+ }
+
+ @Test
+ void cachedSelectorsKeepOriginalValuesAndDoNotShareResolvedCopies() {
+ TableSnapshot input = TableSnapshot.versionOf("11");
+ MvccReadSelector version = MvccReadSelector.of(Optional.of(input),
Optional.empty());
+ input.setValue("12");
+ version.getTableSnapshot().get().setValue("13");
+ Assertions.assertEquals("11",
version.getTableSnapshot().get().getValue());
+ TableScanParams branch = new TableScanParams(TableScanParams.BRANCH,
Collections.emptyMap(),
+ Collections.singletonList("daily"));
+ MvccReadSelector reference = MvccReadSelector.of(Optional.empty(),
Optional.of(branch));
+
reference.getScanParams().get().reuseResolvedMapParams(Collections.singletonMap("snapshot",
"11"));
+
Assertions.assertFalse(reference.getScanParams().get().getResolvedMapParams().isPresent());
+ Assertions.assertNotEquals(version, reference);
+ Assertions.assertNotEquals(MvccReadSelector.LATEST, reference);
+ }
+
+ @Test
+ void oneTableRetainsLatestSnapshotAndBranchDependenciesIndependently() {
+ IcebergExternalTable table = Mockito.mock(IcebergExternalTable.class);
+ IcebergExternalDatabase database =
Mockito.mock(IcebergExternalDatabase.class);
+ IcebergExternalCatalog catalog =
Mockito.mock(IcebergExternalCatalog.class);
+ Mockito.when(table.getDatabase()).thenReturn(database);
+ Mockito.when(database.getCatalog()).thenReturn(catalog);
+ Mockito.when(database.getFullName()).thenReturn("db");
+ Mockito.when(catalog.getName()).thenReturn("ice");
+
Mockito.when(catalog.getProperties()).thenReturn(Collections.emptyMap());
+ Mockito.when(table.getName()).thenReturn("t");
+ Mockito.when(table.getSqlCacheVersion(Mockito.any())).thenReturn(
+ Optional.of(new SqlCacheVersion("latest", 1)),
+ Optional.of(new SqlCacheVersion("historical", 2)),
+ Optional.of(new SqlCacheVersion("branch", 3)));
+ SqlCacheContext context = new SqlCacheContext(null);
+ context.addUsedTable(table, Optional.empty(), MvccReadSelector.LATEST);
+ context.addUsedTable(table, Optional.empty(),
+
MvccReadSelector.of(Optional.of(TableSnapshot.versionOf("11")),
Optional.empty()));
+ context.addUsedTable(table, Optional.empty(),
MvccReadSelector.of(Optional.empty(), Optional.of(
+ new TableScanParams(TableScanParams.BRANCH,
Collections.emptyMap(),
+ Collections.singletonList("daily")))));
+ Assertions.assertEquals(3, context.getUsedTables().size());
+ Assertions.assertFalse(context.hasUnsupportedTables());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
