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 5c9fb5fb33e..dbdac30e7f7 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 @@ -451,26 +451,28 @@ public class NereidsSqlCacheManager { TableIf tableIf = findTableIf(env, scanTable.getKey().tableName); if (tableIf == null) { return IsChanged.CHANGED_AND_INVALIDATE_CACHE; - } else if (tableIf.isTemporary()) { + } + if (tableIf.isTemporary()) { return IsChanged.CHANGED_BUT_NOT_INVALIDATE_CACHE; - } else if (tableVersion.id != tableIf.getId()) { // should after isTemporary condition + } + if (tableVersion.id != tableIf.getId()) { // should after isTemporary condition return IsChanged.CHANGED_AND_INVALIDATE_CACHE; } if (tableIf instanceof OlapTable) { OlapTable olapTable = (OlapTable) tableIf; - long currentTableVersion = 0L; try { - currentTableVersion = olapTable.getVisibleVersion(); + long currentTableVersion = olapTable.getVisibleVersion(); + boolean versionMatches = currentTableVersion == tableVersion.version; + if (tableVersion.olapVersion != null && tableVersion.olapVersion.isUsable()) { + versionMatches = tableVersion.olapVersion.matches(olapTable, currentTableVersion); + } + if (!versionMatches) { + return IsChanged.CHANGED_AND_INVALIDATE_CACHE; + } } catch (RpcException e) { LOG.warn("table {}, in cloud getVisibleVersion exception", olapTable.getName(), e); return IsChanged.CHANGED_AND_INVALIDATE_CACHE; } - long cacheTableVersion = tableVersion.version; - // some partitions have been dropped, or delete or updated or replaced, - // or insert rows into new partition? - if (currentTableVersion != cacheTableVersion) { - return IsChanged.CHANGED_AND_INVALIDATE_CACHE; - } if (tableIf instanceof MTMV) { // mtmv maybe access old data when grace_period > 0, we should disable cache at this case long gracePeriod = ((MTMV) tableIf).getGracePeriod(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java index de117e3a0e9..ea4807e0497 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java @@ -22,6 +22,7 @@ import org.apache.doris.analysis.ExplainOptions; import org.apache.doris.analysis.StatementBase; import org.apache.doris.catalog.Column; import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.OlapTableWrapper; import org.apache.doris.catalog.TableIf; import org.apache.doris.common.Config; import org.apache.doris.common.FeConstants; @@ -74,6 +75,7 @@ import org.apache.doris.nereids.trees.plans.physical.PhysicalRelation; import org.apache.doris.nereids.trees.plans.physical.PhysicalSqlCache; import org.apache.doris.nereids.trees.plans.physical.TopnFilter; import org.apache.doris.planner.AddLocalExchange; +import org.apache.doris.planner.OlapScanNode; import org.apache.doris.planner.PlanFragment; import org.apache.doris.planner.PlanNodeId; import org.apache.doris.planner.Planner; @@ -740,6 +742,18 @@ public class NereidsPlanner extends Planner { // update scan nodes visible version at the end of plan phase for cloud mode. try { ScanNode.setVisibleVersionForOlapScanNodes(getScanNodes()); + statementContext.getSqlCacheContext().ifPresent(cache -> { + for (ScanNode scan : getScanNodes()) { + if (!(scan instanceof OlapScanNode)) { + continue; + } + OlapScanNode olapScan = (OlapScanNode) scan; + if (olapScan.getOlapTable() instanceof OlapTableWrapper) { + continue; + } + cache.addOlapScan(olapScan.getOlapTable(), olapScan.getSqlCachePartitionVersions()); + } + }); } catch (UserException ue) { throw new NereidsException(ue.getMessage(), ue); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/OlapTableSqlCacheVersion.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/OlapTableSqlCacheVersion.java new file mode 100644 index 00000000000..bf3875fadd6 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/OlapTableSqlCacheVersion.java @@ -0,0 +1,191 @@ +// 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.nereids; + +import org.apache.doris.catalog.MaterializedIndexMeta; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; +import org.apache.doris.cloud.catalog.CloudPartition; +import org.apache.doris.common.Config; +import org.apache.doris.common.Pair; +import org.apache.doris.rpc.RpcException; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; + +/** Metadata and partition dependencies of a partitioned OLAP SQL cache entry. */ +public class OlapTableSqlCacheVersion { + private final long tableVersion; + private final long baseIndexId; + private final long topologyVersion; + private final Map> schemas; + private final Map partitionNames = new TreeMap<>(); + // Include empty partitions: pruning can remove their scan nodes before they receive their first rows. + private final Map> partitionVersions = new TreeMap<>(); + private boolean hasScan; + private boolean usable = true; + private volatile long validatedTableVersion; + + /** Capture table metadata while the planner holds the table read lock. */ + public OlapTableSqlCacheVersion(OlapTable table, long tableVersion) { + this.tableVersion = tableVersion; + this.validatedTableVersion = tableVersion; + this.baseIndexId = table.getBaseIndexId(); + this.topologyVersion = table.getPartitionTopologyVersion(); + this.schemas = schemaVersions(table); + } + + private static Map> schemaVersions(OlapTable table) { + Map> versions = new TreeMap<>(); + for (Map.Entry entry : table.getIndexIdToMeta().entrySet()) { + MaterializedIndexMeta meta = entry.getValue(); + versions.put(entry.getKey(), Pair.of(meta.getSchemaVersion(), meta.getSchemaHash())); + } + return versions; + } + + private static Pair version(Partition partition, long version) { + // MS has no update timestamp for an empty partition and FE may synthesize one on every refresh. + return Pair.of(version, version == Partition.PARTITION_INIT_VERSION ? 0L : partition.getVisibleVersionTime()); + } + + /** Track negative dependencies removed by PruneEmptyPartition, not unrelated empty partitions. */ + public void addEmptyPartitions(OlapTable table, Set partitionIds) { + for (Long id : partitionIds) { + Partition partition = table.getPartition(id); + if (table.isTemporaryPartition(id) + || partition.getCachedVisibleVersion() != Partition.PARTITION_INIT_VERSION) { + usable = false; + return; + } + partitionNames.put(id, partition.getName()); + partitionVersions.put(id, version(partition, Partition.PARTITION_INIT_VERSION)); + } + } + + private boolean metadataMatches(OlapTable table) { + return table.getBaseIndexId() == baseIndexId + && table.getPartitionTopologyVersion() == topologyVersion + && schemas.equals(schemaVersions(table)); + } + + /** Record the versions actually assigned to the BE scan, before executing the query. */ + public void addScan(OlapTable table, Map scanVersions) throws RpcException { + if (!usable) { + return; + } + long currentVersion = table.getVisibleVersion(); + // Native versions are protected by the statement's table read lock; the extra lock is cloud-only. + if (Config.isCloudMode()) { + table.versionReadLock(); + } + try { + // Do not attach a newer snapshot to a plan analyzed before a concurrent commit or DDL. + if (currentVersion != tableVersion || !metadataMatches(table)) { + usable = false; + return; + } + for (Map.Entry entry : scanVersions.entrySet()) { + Partition partition = table.getPartition(entry.getKey()); + if (table.isTemporaryPartition(entry.getKey()) + || partition.getCachedVisibleVersion() != entry.getValue()) { + usable = false; + return; + } + partitionNames.put(entry.getKey(), partition.getName()); + // Singleton cloud refreshes can advance the cache despite the table's version read lock. + // Keep the version assigned to the scan, rather than reading the mutable cache again. + Pair scanned = version(partition, entry.getValue()); + Pair previous = partitionVersions.putIfAbsent(entry.getKey(), scanned); + if (previous != null && !previous.equals(scanned)) { + usable = false; + return; + } + hasScan = true; + } + } finally { + if (Config.isCloudMode()) { + table.versionReadUnlock(); + } + } + } + + /** Keep table-wide invalidation when some dependencies have no ordinary scan ranges. */ + public void useTableVersion() { + usable = false; + } + + public boolean isUsable() { + return usable && hasScan; + } + + private static List getPartitionVersions(List partitions) throws RpcException { + if (!Config.isCloudMode()) { + return Partition.getVisibleVersions(partitions); + } + List cloudPartitions = new ArrayList<>(partitions.size()); + for (Partition partition : partitions) { + cloudPartitions.add((CloudPartition) partition); + } + // A fresh table version must not be validated against stale FE partition-version caches. + return CloudPartition.getSnapshotVisibleVersionFromMs(cloudPartitions); + } + + /** Check metadata first, and refresh partition dependencies when the table's data version changes. */ + public boolean matches(OlapTable table, long currentVersion) throws RpcException { + if (!metadataMatches(table)) { + return false; + } + if (currentVersion == validatedTableVersion) { + return true; + } + List partitions = new ArrayList<>(partitionVersions.size()); + for (Long id : partitionVersions.keySet()) { + partitions.add(table.getPartition(id)); + } + List versions = getPartitionVersions(partitions); + // Native versions are protected by the statement's table read lock; the extra lock is cloud-only. + if (Config.isCloudMode()) { + table.versionReadLock(); + } + try { + for (int i = 0; i < partitions.size(); i++) { + Pair cached = partitionVersions.get(partitions.get(i).getId()); + if (!cached.first.equals(versions.get(i)) + || (cached.first != Partition.PARTITION_INIT_VERSION + && cached.second != partitions.get(i).getVisibleVersionTime())) { + return false; + } + } + validatedTableVersion = currentVersion; + return true; + } finally { + if (Config.isCloudMode()) { + table.versionReadUnlock(); + } + } + } + + /** A stable BE key for identical metadata and partition dependencies, independent of other writes. */ + public String getCacheKey() { + return "partitions-v1:" + baseIndexId + ":" + schemas + ":" + partitionNames + ":" + partitionVersions; + } +} 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 d1679bcf304..fbfce360293 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 @@ -24,6 +24,7 @@ import org.apache.doris.authorization.RowFilterSpec; import org.apache.doris.catalog.DatabaseIf; import org.apache.doris.catalog.MTMV; import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.OlapTableWrapper; import org.apache.doris.catalog.TableIf; import org.apache.doris.catalog.TableIf.TableType; import org.apache.doris.common.Pair; @@ -44,6 +45,7 @@ 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.SqlCacheVersionProvider; +import org.apache.doris.rpc.RpcException; import org.apache.doris.thrift.TUniqueId; import com.google.common.collect.ImmutableList; @@ -244,7 +246,11 @@ public class SqlCacheContext { new TableVersion( tableIf.getId(), version, - tableIf.getType(), sqlCacheVersion + tableIf.getType(), sqlCacheVersion, + usedVariables.isEmpty() && tableIf.getType() == TableType.OLAP && tableIf instanceof OlapTable + && !(tableIf instanceof OlapTableWrapper) + && ((OlapTable) tableIf).isPartitionedTable() + ? new OlapTableSqlCacheVersion((OlapTable) tableIf, version) : null ) ); } @@ -318,8 +324,12 @@ public class SqlCacheContext { return ImmutableMap.copyOf(dataMaskPolicies); } + /** Record a variable and retain partition-independent lookup keys for its cached value variants. */ public synchronized void addUsedVariable(Variable value) { usedVariables.add(value); + // Variable-aware FE lookup derives another value's key before planning its partition pruning. + // Keep the existing table-wide key for these queries, including variables resolved after table binding. + useSourceTableVersions(); } public synchronized List getUsedVariables() { @@ -394,6 +404,57 @@ public class SqlCacheContext { return Collections.unmodifiableMap(usedTables); } + /** Record native scan dependencies before the first SQL cache key is computed. */ + public synchronized void addOlapScan(OlapTable table, Map partitionVersions) { + if (table instanceof MTMV) { + // MV substitution can hide source partitions from the final physical scans. Preserve the + // logical source tables' versions before the BE key is computed, including partial UNION rewrites. + useSourceTableVersions(); + return; + } + for (TableVersion version : usedTables.values()) { + if (version.id == table.getId() && version.olapVersion != null) { + try { + version.olapVersion.addScan(table, partitionVersions); + } catch (RpcException e) { + setHasUnsupportedTables(true); + LOG.warn("Cannot record SQL cache partitions for {}", table.getName(), e); + } + } + } + } + + /** Preserve table-wide validation for metadata aggregates. */ + public synchronized void useOlapTableVersion(OlapTable table) { + for (TableVersion version : usedTables.values()) { + if (version.id == table.getId() && version.olapVersion != null) { + version.olapVersion.useTableVersion(); + } + } + } + + /** Retain empty partitions that were eligible for this query before empty-partition pruning. */ + public synchronized void addPrunedOlapPartitions(OlapTable table, Set partitionIds) { + if (table instanceof MTMV) { + // An empty MV can disappear before physical scan capture, hiding its source dependencies. + useSourceTableVersions(); + return; + } + for (TableVersion version : usedTables.values()) { + if (version.id == table.getId() && version.olapVersion != null) { + version.olapVersion.addEmptyPartitions(table, partitionIds); + } + } + } + + private void useSourceTableVersions() { + for (TableVersion version : usedTables.values()) { + if (version.olapVersion != null) { + version.olapVersion.useTableVersion(); + } + } + } + public Map getUsedViews() { return ImmutableMap.copyOf(usedViews); } @@ -484,8 +545,7 @@ public class SqlCacheContext { .append("=") .append(tableVersion.id) .append("@") - .append(tableVersion.sqlCacheVersion == null ? Long.toString(tableVersion.version) - : "sql-cache:" + tableVersion.sqlCacheVersion.getVersion()); + .append(tableVersion.getCacheKey()); } for (Entry entry : usedViews.entrySet()) { cacheKey.append("|") @@ -637,10 +697,23 @@ public class SqlCacheContext { public final long version; public final TableType type; public final ConnectorSqlCacheVersion sqlCacheVersion; + public final OlapTableSqlCacheVersion olapVersion; + + public TableVersion(long id, long version, TableType type, ConnectorSqlCacheVersion sqlCacheVersion) { + this(id, version, type, sqlCacheVersion, null); + } public TableVersion(long id, long version, TableType type) { this(id, version, type, null); } + + /** Data-version component shared by FE validation and BE cache lookup. */ + public String getCacheKey() { + if (olapVersion != null && olapVersion.isUsable()) { + return olapVersion.getCacheKey(); + } + return sqlCacheVersion == null ? Long.toString(version) : "sql-cache:" + sqlCacheVersion.getVersion(); + } } /** CacheKeyType */ diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java index aa0fa7fed52..01e32857705 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java @@ -29,7 +29,9 @@ import org.apache.doris.transaction.TransactionEntry; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import java.util.HashSet; import java.util.List; +import java.util.Set; /** * Used to prune empty partition. @@ -55,6 +57,13 @@ public class PruneEmptyPartition extends OneRewriteRuleFactory { selectNonEmptyPartitionIdsForTxnLoad(ctx.connectContext.getTxnEntry(), table, scan.getSelectedIndexId(), partitionIdsToPrune, ids); } + if (partitionIdsToPrune.size() != ids.size()) { + ctx.statementContext.getSqlCacheContext().ifPresent(cache -> { + Set emptyPartitions = new HashSet<>(partitionIdsToPrune); + emptyPartitions.removeAll(ids); + cache.addPrunedOlapPartitions(table, emptyPartitions); + }); + } if (ids.isEmpty()) { return new LogicalEmptyRelation(ConnectContext.get().getStatementContext().getNextRelationId(), scan.getOutput()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/RewriteSimpleAggToConstantRule.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/RewriteSimpleAggToConstantRule.java index 9347b7edb72..36b001b2e52 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/RewriteSimpleAggToConstantRule.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/RewriteSimpleAggToConstantRule.java @@ -156,6 +156,9 @@ public class RewriteSimpleAggToConstantRule implements RewriteRuleFactory { return null; } + // Metadata-derived values still depend on the whole table, even if another branch scans fewer partitions. + statementContext.getSqlCacheContext().ifPresent(cache -> cache.useOlapTableVersion(table)); + // Build: LogicalProject(constants) -> LogicalOneRowRelation(dummy) // The OneRowRelation provides a single-row source; all real values come from the project. LogicalOneRowRelation oneRowRelation = new LogicalOneRowRelation( diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java index a975f8f2bf3..bbe09c49a95 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java @@ -189,6 +189,8 @@ public class OlapScanNode extends ScanNode { private long totalTabletsNum = 0; private long selectedIndexId = -1; private Collection selectedPartitionIds = Lists.newArrayList(); + + private final Map sqlCachePartitionVersions = Maps.newHashMap(); private Map selectedPartitionNames = Collections.emptyMap(); // Partition boundaries must come from the same planning snapshot as runtime-filter // partition monotonicity. Thrift serialization can happen after query queueing, when @@ -479,6 +481,7 @@ public class OlapScanNode extends ScanNode { final List tablets = selectedTable.getTablets(); Long visibleVersion = visibleVersionMap.get(partitionId); assert visibleVersion != null : "the acquried version is not exists in the visible version map"; + sqlCachePartitionVersions.put(partitionId, visibleVersion); String visibleVersionStr = String.valueOf(visibleVersion); for (Tablet tablet : tablets) { TScanRangeLocations locations = locationsMap.get(tablet.getId()); @@ -496,6 +499,10 @@ public class OlapScanNode extends ScanNode { return tabletBytes.get(tabletId); } + public Map getSqlCachePartitionVersions() { + return sqlCachePartitionVersions; + } + public long getMaxVersion() { return maxVersion; } @@ -515,6 +522,7 @@ public class OlapScanNode extends ScanNode { } // for non-cloud mode. for cloud mode see `updateScanRangeVersions` maxVersion = Math.max(maxVersion, visibleVersion); + sqlCachePartitionVersions.put(partition.getId(), visibleVersion); int useFixReplica = -1; boolean skipMissingVersion = false; diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/OlapTableSqlCacheVersionTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/OlapTableSqlCacheVersionTest.java new file mode 100644 index 00000000000..5fa4d41315a --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/OlapTableSqlCacheVersionTest.java @@ -0,0 +1,289 @@ +// 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.nereids; + +import org.apache.doris.analysis.UserIdentity; +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.MTMV; +import org.apache.doris.catalog.MaterializedIndex; +import org.apache.doris.catalog.MaterializedIndex.IndexState; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; +import org.apache.doris.cloud.catalog.CloudPartition; +import org.apache.doris.common.Config; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.nereids.analyzer.UnboundVariable.VariableType; +import org.apache.doris.nereids.trees.expressions.Variable; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; +import org.apache.doris.nereids.util.PlanConstructor; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +class OlapTableSqlCacheVersionTest { + private OlapTable table() { + OlapTable table = PlanConstructor.newOlapTable(1, "sql_cache_partition_versions", 0); + table.addPartition(partition(table, 10, 2)); + table.addPartition(partition(table, 20, 1)); + table.updateVisibleVersionAndTime(2, 100); + return table; + } + + private Partition partition(OlapTable table, long id, long version) { + Partition partition = new Partition(id, "p" + id, + new MaterializedIndex(-1, IndexState.NORMAL), table.getDefaultDistributionInfo()); + partition.updateVisibleVersionAndTime(version, 100); + return partition; + } + + private OlapTableSqlCacheVersion snapshot(OlapTable table) throws Exception { + OlapTableSqlCacheVersion snapshot = new OlapTableSqlCacheVersion(table, table.getVisibleVersion()); + snapshot.addScan(table, ImmutableMap.of(10L, 2L)); + Assertions.assertTrue(snapshot.isUsable()); + return snapshot; + } + + @Test + void testUnrelatedPartitionWritePreservesKeyAndCache() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = snapshot(table); + table.getPartition(20).updateVisibleVersionAndTime(2, 200); + table.updateVisibleVersionAndTime(3, 200); + Assertions.assertTrue(cached.matches(table, 3)); + Assertions.assertEquals(cached.getCacheKey(), snapshot(table).getCacheKey()); + } + + @Test + void testScannedPartitionWriteChangesKey() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = snapshot(table); + table.getPartition(10).updateVisibleVersionAndTime(3, 200); + table.updateVisibleVersionAndTime(3, 200); + Assertions.assertFalse(cached.matches(table, 3)); + OlapTableSqlCacheVersion current = new OlapTableSqlCacheVersion(table, 3); + current.addScan(table, ImmutableMap.of(10L, 3L)); + Assertions.assertNotEquals(cached.getCacheKey(), current.getCacheKey()); + } + + @Test + void testPrunedEmptyPartitionFirstWriteInvalidates() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = snapshot(table); + cached.addEmptyPartitions(table, ImmutableSet.of(20L)); + table.getPartition(20).updateVisibleVersionAndTime(2, 200); + table.updateVisibleVersionAndTime(3, 200); + Assertions.assertFalse(cached.matches(table, 3)); + } + + @Test + void testMetadataChangesInvalidate() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = snapshot(table); + table.addPartition(partition(table, 30, 1)); + Assertions.assertFalse(cached.matches(table, 2)); + cached = snapshot(table); + table.getIndexMetaByIndexId(table.getBaseIndexId()).setSchemaVersion(1); + Assertions.assertFalse(cached.matches(table, 2)); + } + + @Test + void testConcurrentCommitDuringPlanningKeepsTableVersionFallback() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = new OlapTableSqlCacheVersion(table, 2); + table.updateVisibleVersionAndTime(3, 200); + cached.addScan(table, ImmutableMap.of(10L, 2L)); + Assertions.assertFalse(cached.isUsable()); + Assertions.assertFalse(new OlapTableSqlCacheVersion(table, 3).isUsable()); + } + + @Test + void testAllSelfJoinScansContributeDependencies() throws Exception { + OlapTable table = table(); + table.getPartition(20).updateVisibleVersionAndTime(2, 100); + OlapTableSqlCacheVersion cached = snapshot(table); + cached.addScan(table, ImmutableMap.of(20L, 2L)); + table.getPartition(20).updateVisibleVersionAndTime(3, 200); + table.updateVisibleVersionAndTime(3, 200); + Assertions.assertFalse(cached.matches(table, 3)); + } + + @Test + void testMetadataAggregateRetainsWholeTableDependency() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = snapshot(table); + cached.useTableVersion(); + Assertions.assertFalse(cached.isUsable()); + } + + @Test + void testEmptyPartitionRefreshDoesNotChangeKey() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = snapshot(table); + cached.addEmptyPartitions(table, ImmutableSet.of(20L)); + table.getPartition(20).updateVisibleVersionAndTime(1, 200); + table.updateVisibleVersionAndTime(3, 200); + Assertions.assertTrue(cached.matches(table, 3)); + OlapTableSqlCacheVersion refreshed = snapshot(table); + refreshed.addEmptyPartitions(table, ImmutableSet.of(20L)); + Assertions.assertEquals(cached.getCacheKey(), refreshed.getCacheKey()); + } + + private OlapTable cacheableTable() { + OlapTable table = Mockito.spy(table()); + DatabaseIf database = Mockito.mock(DatabaseIf.class); + CatalogIf catalog = Mockito.mock(CatalogIf.class); + Mockito.doReturn(database).when(table).getDatabase(); + Mockito.doReturn(true).when(table).isPartitionedTable(); + Mockito.when(database.getCatalog()).thenReturn(catalog); + Mockito.when(database.getFullName()).thenReturn("cache_db"); + Mockito.when(catalog.getName()).thenReturn("internal"); + Mockito.when(catalog.isInternalCatalog()).thenReturn(true); + return table; + } + + @Test + void testVariableAwareCacheKeysRetainTableVersionInBothBindingOrders() throws Exception { + OlapTable table = cacheableTable(); + Variable variable = new Variable("partition_id", VariableType.USER, new IntegerLiteral(1)); + + SqlCacheContext tableFirst = new SqlCacheContext(UserIdentity.ROOT); + tableFirst.addUsedTable(table); + tableFirst.addOlapScan(table, ImmutableMap.of(10L, 2L)); + SqlCacheContext.TableVersion version = tableFirst.getUsedTables().values().iterator().next(); + Assertions.assertTrue(version.olapVersion.isUsable()); + tableFirst.addUsedVariable(variable); + Assertions.assertFalse(version.olapVersion.isUsable()); + Assertions.assertEquals("2", version.getCacheKey()); + + SqlCacheContext variableFirst = new SqlCacheContext(UserIdentity.ROOT); + variableFirst.addUsedVariable(variable); + variableFirst.addUsedTable(table); + variableFirst.addOlapScan(table, ImmutableMap.of(10L, 2L)); + Assertions.assertEquals("2", variableFirst.getUsedTables().values().iterator().next().getCacheKey()); + } + + @Test + void testMaterializedViewScanRetainsSourceVersionInBothScanOrders() { + OlapTable table = cacheableTable(); + MTMV mv = Mockito.mock(MTMV.class); + for (boolean mvFirst : new boolean[] {true, false}) { + SqlCacheContext context = new SqlCacheContext(UserIdentity.ROOT); + context.addUsedTable(table); + if (mvFirst) { + context.addOlapScan(mv, ImmutableMap.of()); + } + context.addOlapScan(table, ImmutableMap.of(10L, 2L)); + if (!mvFirst) { + context.addOlapScan(mv, ImmutableMap.of()); + } + SqlCacheContext.TableVersion version = context.getUsedTables().values().iterator().next(); + Assertions.assertFalse(version.olapVersion.isUsable()); + Assertions.assertEquals("2", version.getCacheKey()); + } + } + + @Test + void testConcurrentPartitionRefreshCannotUpgradeScanDependency() throws Exception { + try (MockedStatic config = Mockito.mockStatic(Config.class); + MockedStatic cloud = Mockito.mockStatic(CloudPartition.class)) { + config.when(Config::isCloudMode).thenReturn(true); + OlapTable table = Mockito.spy(table()); + Mockito.doReturn(2L).when(table).getVisibleVersion(); + CloudPartition partition = Mockito.spy(new CloudPartition(10, "p10", + new MaterializedIndex(-1, IndexState.NORMAL), table.getDefaultDistributionInfo(), + 1, table.getId())); + partition.setCachedVisibleVersion(2, 100); + table.addPartition(partition); + // A singleton MS refresh can update the cache despite the table's version read lock. + Mockito.doAnswer(invocation -> { + long version = (long) invocation.callRealMethod(); + partition.setCachedVisibleVersion(3, 200); + return version; + }).when(partition).getCachedVisibleVersion(); + + OlapTableSqlCacheVersion cached = new OlapTableSqlCacheVersion(table, 2); + cached.addScan(table, ImmutableMap.of(10L, 2L)); + cloud.when(() -> CloudPartition.getSnapshotVisibleVersionFromMs(ImmutableList.of(partition))) + .thenReturn(ImmutableList.of(3L)); + Assertions.assertFalse(cached.isUsable() && cached.matches(table, 3)); + } + } + + @Test + void testConcurrentFirstWriteCannotUpgradeEmptyDependency() throws Exception { + OlapTable table = table(); + OlapTableSqlCacheVersion cached = snapshot(table); + Partition partition = Mockito.spy(table.getPartition(20)); + table.addPartition(partition); + Mockito.doAnswer(invocation -> { + long version = (long) invocation.callRealMethod(); + partition.updateVisibleVersionAndTime(2, 200); + return version; + }).when(partition).getCachedVisibleVersion(); + + cached.addEmptyPartitions(table, ImmutableSet.of(20L)); + table.updateVisibleVersionAndTime(3, 200); + Assertions.assertFalse(cached.isUsable() && cached.matches(table, 3)); + } + + @Test + void testPrunedMaterializedViewRetainsSourceVersionInBothScanOrders() { + OlapTable table = cacheableTable(); + MTMV mv = Mockito.mock(MTMV.class); + for (boolean mvFirst : new boolean[] {true, false}) { + SqlCacheContext context = new SqlCacheContext(UserIdentity.ROOT); + context.addUsedTable(table); + if (mvFirst) { + context.addPrunedOlapPartitions(mv, ImmutableSet.of(30L)); + } + context.addOlapScan(table, ImmutableMap.of(10L, 2L)); + if (!mvFirst) { + context.addPrunedOlapPartitions(mv, ImmutableSet.of(30L)); + } + SqlCacheContext.TableVersion version = context.getUsedTables().values().iterator().next(); + Assertions.assertFalse(version.olapVersion.isUsable()); + Assertions.assertEquals("2", version.getCacheKey()); + } + } + + @Test + void testCloudRevalidationBypassesStalePartitionVersionCache() throws Exception { + try (MockedStatic config = Mockito.mockStatic(Config.class); + MockedStatic cloud = Mockito.mockStatic(CloudPartition.class)) { + config.when(Config::isCloudMode).thenReturn(true); + OlapTable table = Mockito.spy(table()); + Mockito.doReturn(2L).when(table).getVisibleVersion(); + CloudPartition partition = new CloudPartition(10, "p10", + new MaterializedIndex(-1, IndexState.NORMAL), table.getDefaultDistributionInfo(), 1, table.getId()); + partition.setCachedVisibleVersion(2, 100); + table.addPartition(partition); + OlapTableSqlCacheVersion cached = snapshot(table); + cloud.when(() -> CloudPartition.getSnapshotVisibleVersionFromMs(ImmutableList.of(partition))) + .thenReturn(ImmutableList.of(2L), ImmutableList.of(3L)); + Assertions.assertTrue(cached.matches(table, 3)); + Assertions.assertFalse(cached.matches(table, 4)); + cloud.verify(() -> CloudPartition.getSnapshotVisibleVersionFromMs(ImmutableList.of(partition)), + Mockito.times(2)); + } + } +} diff --git a/regression-test/data/query_p0/cache/sql_cache_empty_mtmv.out b/regression-test/data/query_p0/cache/sql_cache_empty_mtmv.out new file mode 100644 index 00000000000..290bb96fa06 --- /dev/null +++ b/regression-test/data/query_p0/cache/sql_cache_empty_mtmv.out @@ -0,0 +1,12 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !before_write -- +2026-10-27 10 + +-- !after_write -- +2026-10-27 10 +2026-10-28 20 + +-- !without_cache -- +2026-10-27 10 +2026-10-28 20 + diff --git a/regression-test/data/query_p0/cache/sql_cache_partial_mtmv.out b/regression-test/data/query_p0/cache/sql_cache_partial_mtmv.out new file mode 100644 index 00000000000..2dfc8c6143f --- /dev/null +++ b/regression-test/data/query_p0/cache/sql_cache_partial_mtmv.out @@ -0,0 +1,18 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !before_write -- +2026-10-27 10 +2026-10-27 11 +2026-10-28 20 + +-- !after_write -- +2026-10-27 10 +2026-10-27 11 +2026-10-28 20 +2026-10-28 21 + +-- !without_cache -- +2026-10-27 10 +2026-10-27 11 +2026-10-28 20 +2026-10-28 21 + diff --git a/regression-test/data/query_p0/cache/sql_cache_partition_version.out b/regression-test/data/query_p0/cache/sql_cache_partition_version.out new file mode 100644 index 00000000000..6b7909edf1e --- /dev/null +++ b/regression-test/data/query_p0/cache/sql_cache_partition_version.out @@ -0,0 +1,36 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !history -- +10 + +-- !unrelated_write -- +10 + +-- !relevant_write -- +40 + +-- !including_empty -- +60 + +-- !empty_branch -- +\N +40 + +-- !empty_branch_write -- +40 +40 + +-- !empty_first_write -- +100 + +-- !self_join -- +30 + +-- !self_join_write -- +170 + +-- !add_partition -- +40 + +-- !truncate_partition -- +\N + diff --git a/regression-test/suites/query_p0/cache/mv_with_sql_cache.groovy b/regression-test/suites/query_p0/cache/mv_with_sql_cache.groovy index 51a1df0a841..bef70364289 100644 --- a/regression-test/suites/query_p0/cache/mv_with_sql_cache.groovy +++ b/regression-test/suites/query_p0/cache/mv_with_sql_cache.groovy @@ -116,6 +116,15 @@ suite("mv_with_sql_cache") { create_sync_mv(dbName, tb_name1, mv_name2, mv_sql2) sleep(10 * 1000) + // Adding rollup schemas invalidates existing partition-dependent cache entries. + assertNoCache "select * from ${tb_name1}" + assertNoCache mv_sql1 + assertNoCache mv_sql2 + + sql "select * from ${tb_name1}" + sql mv_sql1 + sql mv_sql2 + assertHasCache "select * from ${tb_name1}" assertHasCache mv_sql1 assertHasCache mv_sql2 @@ -179,6 +188,15 @@ suite("mv_with_sql_cache") { create_sync_mv(dbName, tb_name1, mv_name2, mv_sql2) sleep(10 * 1000) + // Adding rollup schemas invalidates existing partition-dependent cache entries. + assertNoCache "select * from ${tb_name1}" + assertNoCache mv_sql1 + assertNoCache mv_sql2 + + sql "select * from ${tb_name1}" + sql mv_sql1 + sql mv_sql2 + assertHasCache "select * from ${tb_name1}" assertHasCache mv_sql1 assertHasCache mv_sql2 @@ -242,6 +260,15 @@ suite("mv_with_sql_cache") { create_sync_mv(dbName, tb_name1, mv_name2, mv_sql2) sleep(10 * 1000) + // Adding rollup schemas invalidates existing partition-dependent cache entries. + assertNoCache "select * from ${tb_name1}" + assertNoCache mv_sql1 + assertNoCache mv_sql2 + + sql "select * from ${tb_name1}" + sql mv_sql1 + sql mv_sql2 + assertHasCache "select * from ${tb_name1}" assertHasCache mv_sql1 assertHasCache mv_sql2 @@ -254,6 +281,15 @@ suite("mv_with_sql_cache") { sql """drop materialized view ${mv_name1} on ${tb_name1};""" sleep(10 * 1000) + // Dropping a rollup also changes the schema dependencies of cached queries. + assertNoCache "select * from ${tb_name1}" + assertNoCache mv_sql1 + assertNoCache mv_sql2 + + sql "select * from ${tb_name1}" + sql mv_sql1 + sql mv_sql2 + assertHasCache "select * from ${tb_name1}" assertHasCache mv_sql1 assertHasCache mv_sql2 diff --git a/regression-test/suites/query_p0/cache/sql_cache_empty_mtmv.groovy b/regression-test/suites/query_p0/cache/sql_cache_empty_mtmv.groovy new file mode 100644 index 00000000000..bf8dc40353c --- /dev/null +++ b/regression-test/suites/query_p0/cache/sql_cache_empty_mtmv.groovy @@ -0,0 +1,86 @@ +// 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. + +suite("sql_cache_empty_mtmv") { + withGlobalLock("cache_last_version_interval_second") { + def originalInterval = sql("ADMIN SHOW FRONTEND CONFIG LIKE 'cache_last_version_interval_second'")[0][1] + sql "ADMIN SET ALL FRONTENDS CONFIG ('cache_last_version_interval_second' = '0')" + try { + sql "set enable_sql_cache=false" + sql "set enable_strong_consistency_read=true" + sql "set cloud_table_version_cache_ttl_ms=0" + sql "set cloud_partition_version_cache_ttl_ms=0" + sql "set enable_materialized_view_rewrite=true" + sql "drop materialized view if exists sql_cache_empty_mtmv_view" + sql "drop table if exists sql_cache_empty_mtmv_base" + sql """ + create table sql_cache_empty_mtmv_base (dt date, v int) + duplicate key(dt) + partition by range(dt) ( + partition p1 values [('2026-10-27'), ('2026-10-28')), + partition p2 values [('2026-10-28'), ('2026-10-29')) + ) distributed by hash(dt) buckets 1 + properties ('replication_num'='1') + """ + // The source p2 is nonempty, but its filtered MV is empty. + sql """insert into sql_cache_empty_mtmv_base values + ('2026-10-27', 10), ('2026-10-28', -1)""" + sql """ + create materialized view sql_cache_empty_mtmv_view + build immediate refresh auto on manual + distributed by random buckets 1 + properties ('replication_num'='1', 'grace_period'='0') + as select dt, v from sql_cache_empty_mtmv_base + where dt = '2026-10-28' and v > 0 + """ + def db = sql("select database()")[0][0].toString() + waitingMTMVTaskFinished(getJobName(db, "sql_cache_empty_mtmv_view")) + def query = """select /*+ use_mv(sql_cache_empty_mtmv_view) */ dt, v + from sql_cache_empty_mtmv_base where dt = '2026-10-28' and v > 0 + union all select dt, v from sql_cache_empty_mtmv_base where dt = '2026-10-27' + order by dt, v""" + explain { + sql "physical plan ${query}" + // The empty MV branch must disappear; only the ordinary p1 scan remains. + check { plan -> + assertEquals(1, plan.count("PhysicalOlapScan")) + assertFalse(plan.contains("PhysicalOlapScan[sql_cache_empty_mtmv_view")) + } + } + sql "set enable_sql_cache=true" + order_qt_before_write query + explain { + sql "physical plan ${query}" + contains "PhysicalSqlCache" + } + sql "insert into sql_cache_empty_mtmv_base values ('2026-10-28', 20)" + explain { + sql "physical plan ${query}" + notContains "PhysicalSqlCache" + } + order_qt_after_write query + explain { + sql "physical plan ${query}" + contains "PhysicalSqlCache" + } + sql "set enable_sql_cache=false" + order_qt_without_cache query + } finally { + sql "ADMIN SET ALL FRONTENDS CONFIG ('cache_last_version_interval_second' = '${originalInterval}')" + } + } +} diff --git a/regression-test/suites/query_p0/cache/sql_cache_partial_mtmv.groovy b/regression-test/suites/query_p0/cache/sql_cache_partial_mtmv.groovy new file mode 100644 index 00000000000..308b00a35d7 --- /dev/null +++ b/regression-test/suites/query_p0/cache/sql_cache_partial_mtmv.groovy @@ -0,0 +1,89 @@ +// 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. + +suite("sql_cache_partial_mtmv") { + withGlobalLock("cache_last_version_interval_second") { + def originalInterval = sql("ADMIN SHOW FRONTEND CONFIG LIKE 'cache_last_version_interval_second'")[0][1] + sql "ADMIN SET ALL FRONTENDS CONFIG ('cache_last_version_interval_second' = '0')" + try { + sql "set enable_sql_cache=false" + sql "set enable_strong_consistency_read=true" + sql "set cloud_table_version_cache_ttl_ms=0" + sql "set cloud_partition_version_cache_ttl_ms=0" + sql "set enable_materialized_view_rewrite=true" + sql "set enable_materialized_view_union_rewrite=true" + sql "drop materialized view if exists sql_cache_partial_mtmv_view" + sql "drop table if exists sql_cache_partial_mtmv_base" + sql """ + create table sql_cache_partial_mtmv_base (dt date, v int) + duplicate key(dt) + partition by range(dt) ( + partition p1 values [('2026-10-27'), ('2026-10-28')), + partition p2 values [('2026-10-28'), ('2026-10-29')), + partition p3 values [('2026-10-29'), ('2026-10-30')) + ) distributed by hash(dt) buckets 1 + properties ('replication_num'='1') + """ + sql """insert into sql_cache_partial_mtmv_base values + ('2026-10-27', 10), ('2026-10-28', 20), ('2026-10-29', 30)""" + sql """ + create materialized view sql_cache_partial_mtmv_view + build immediate refresh auto on manual + partition by(dt) + distributed by random buckets 1 + properties ('replication_num'='1', 'grace_period'='0') + as select dt, v from sql_cache_partial_mtmv_base + """ + def db = sql("select database()")[0][0].toString() + waitingMTMVTaskFinished(getJobName(db, "sql_cache_partial_mtmv_view")) + // p1 becomes stale in the MV; p2 must still be served by the MV. + sql "insert into sql_cache_partial_mtmv_base values ('2026-10-27', 11)" + // Select the MV plan even when this tiny table makes a direct scan cheaper. + def query = """select /*+ use_mv(sql_cache_partial_mtmv_view) */ dt, v + from sql_cache_partial_mtmv_base + where dt >= '2026-10-27' and dt < '2026-10-29' order by dt, v""" + explain { + sql "physical plan ${query}" + contains "PhysicalUnion" + contains "sql_cache_partial_mtmv_view" + contains "sql_cache_partial_mtmv_base" + } + sql "set enable_sql_cache=true" + order_qt_before_write query + explain { + sql "physical plan ${query}" + contains "PhysicalSqlCache" + } + + // The physical base scan covered only p1, but the query also depends on base p2. + sql "insert into sql_cache_partial_mtmv_base values ('2026-10-28', 21)" + explain { + sql "physical plan ${query}" + notContains "PhysicalSqlCache" + } + order_qt_after_write query + explain { + sql "physical plan ${query}" + contains "PhysicalSqlCache" + } + sql "set enable_sql_cache=false" + order_qt_without_cache query + } finally { + sql "ADMIN SET ALL FRONTENDS CONFIG ('cache_last_version_interval_second' = '${originalInterval}')" + } + } +} diff --git a/regression-test/suites/query_p0/cache/sql_cache_partition_version.groovy b/regression-test/suites/query_p0/cache/sql_cache_partition_version.groovy new file mode 100644 index 00000000000..4d6dd7f137b --- /dev/null +++ b/regression-test/suites/query_p0/cache/sql_cache_partition_version.groovy @@ -0,0 +1,105 @@ +// 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. + +suite("sql_cache_partition_version") { + withGlobalLock("cache_last_version_interval_second") { + def originalInterval = sql("ADMIN SHOW FRONTEND CONFIG LIKE 'cache_last_version_interval_second'")[0][1] + sql "ADMIN SET ALL FRONTENDS CONFIG ('cache_last_version_interval_second' = '0')" + try { + sql "set enable_sql_cache=true" + sql "set enable_strong_consistency_read=true" + sql "set cloud_table_version_cache_ttl_ms=0" + sql "set cloud_partition_version_cache_ttl_ms=0" + sql "drop table if exists sql_cache_partition_version" + sql """ + create table sql_cache_partition_version (dt date, v int) + duplicate key(dt) + partition by range(dt) ( + partition p1 values [('2026-10-27'), ('2026-10-28')), + partition p2 values [('2026-10-28'), ('2026-10-29')), + partition p3 values [('2026-10-29'), ('2026-10-30')) + ) distributed by hash(dt) buckets 1 + properties ('replication_num'='1') + """ + sql "insert into sql_cache_partition_version values ('2026-10-27', 10)" + def history = "select sum(v) from sql_cache_partition_version where dt = '2026-10-27'" + def assertCached = { String query -> + explain { + sql "physical plan ${query}" + contains "PhysicalSqlCache" + } + } + def assertNotCached = { String query -> + explain { + sql "physical plan ${query}" + notContains "PhysicalSqlCache" + } + } + + order_qt_history history + assertCached(history) + // The first write to an unrelated empty partition must preserve the cached query. + sql "insert into sql_cache_partition_version values ('2026-10-28', 20)" + assertCached(history) + order_qt_unrelated_write history + + sql "insert into sql_cache_partition_version values ('2026-10-27', 30)" + assertNotCached(history) + order_qt_relevant_write history + assertCached(history) + + // p3 is selected by the predicate but removed by empty-partition pruning. + def includingEmpty = "select sum(v) from sql_cache_partition_version where dt >= '2026-10-27'" + order_qt_including_empty includingEmpty + assertCached(includingEmpty) + def emptyBranch = """select sum(v) from sql_cache_partition_version where dt = '2026-10-27' + union all select sum(v) from sql_cache_partition_version where dt = '2026-10-29'""" + order_qt_empty_branch emptyBranch + assertCached(emptyBranch) + sql "insert into sql_cache_partition_version values ('2026-10-29', 40)" + assertNotCached(includingEmpty) + assertNotCached(emptyBranch) + order_qt_empty_branch_write emptyBranch + order_qt_empty_first_write includingEmpty + assertCached(includingEmpty) + assertCached(history) + + // A second occurrence of the same table must contribute its partitions as well. + def selfJoin = """select sum(a.v + b.v) from sql_cache_partition_version a + join sql_cache_partition_version b on a.v < b.v + where a.dt = '2026-10-27' and b.dt = '2026-10-28'""" + order_qt_self_join selfJoin + assertCached(selfJoin) + sql "insert into sql_cache_partition_version values ('2026-10-28', 50)" + assertNotCached(selfJoin) + order_qt_self_join_write selfJoin + assertCached(selfJoin) + assertCached(history) + + // Partition topology changes invalidate even if all scanned versions stay the same. + sql "alter table sql_cache_partition_version add partition p4 values [('2026-10-30'), ('2026-10-31'))" + assertNotCached(history) + order_qt_add_partition history + assertCached(history) + sql "truncate table sql_cache_partition_version partition(p1)" + assertNotCached(history) + order_qt_truncate_partition history + } finally { + sql "ADMIN SET ALL FRONTENDS CONFIG ('cache_last_version_interval_second' = '${originalInterval}')" + } + } +}