This is an automated email from the ASF dual-hosted git repository.
CalvinKirs pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 0976b2cd502 [opt](sql-cache) Preserve SQL cache across unrelated
partition writes (#69883)
0976b2cd502 is described below
commit 0976b2cd5020607205969bf7986dcb551b6a7325
Author: Calvin Kirs <[email protected]>
AuthorDate: Mon Nov 2 18:08:28 2026 +0800
[opt](sql-cache) Preserve SQL cache across unrelated partition writes
(#69883)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
SQL cache currently invalidates an OLAP query whenever the table visible
version changes. For example, a cached query reading historical
partition p1 is discarded after an insert into unrelated partition p2,
even though its result is unchanged.
Record the partition versions actually assigned to the scan, including
selected empty partitions removed during planning. When the table
version advances, reuse the cache only if those dependencies and the
schema/partition topology still match. Use the same stable dependency
identity for the BE cache key. Keep table-wide validation for metadata
aggregates, variable-dependent keys and materialized-view rewrites whose
physical scans can hide source dependencies.
The change is FE-only and preserves the existing BE cache protocol.
Regression cases use only 3–5 source rows and cover unrelated/relevant
writes, empty partitions and branches, self joins, partition DDL,
partial MV rewrites and eliminated empty MVs.
---
.../doris/common/cache/NereidsSqlCacheManager.java | 22 +-
.../org/apache/doris/nereids/NereidsPlanner.java | 14 +
.../doris/nereids/OlapTableSqlCacheVersion.java | 191 ++++++++++++++
.../org/apache/doris/nereids/SqlCacheContext.java | 79 +++++-
.../nereids/rules/rewrite/PruneEmptyPartition.java | 9 +
.../rewrite/RewriteSimpleAggToConstantRule.java | 3 +
.../org/apache/doris/planner/OlapScanNode.java | 8 +
.../nereids/OlapTableSqlCacheVersionTest.java | 289 +++++++++++++++++++++
.../data/query_p0/cache/sql_cache_empty_mtmv.out | 12 +
.../data/query_p0/cache/sql_cache_partial_mtmv.out | 18 ++
.../query_p0/cache/sql_cache_partition_version.out | 36 +++
.../suites/query_p0/cache/mv_with_sql_cache.groovy | 36 +++
.../query_p0/cache/sql_cache_empty_mtmv.groovy | 86 ++++++
.../query_p0/cache/sql_cache_partial_mtmv.groovy | 89 +++++++
.../cache/sql_cache_partition_version.groovy | 105 ++++++++
15 files changed, 984 insertions(+), 13 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 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<Long, Pair<Integer, Integer>> schemas;
+ private final Map<Long, String> partitionNames = new TreeMap<>();
+ // Include empty partitions: pruning can remove their scan nodes before
they receive their first rows.
+ private final Map<Long, Pair<Long, Long>> 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<Long, Pair<Integer, Integer>> schemaVersions(OlapTable
table) {
+ Map<Long, Pair<Integer, Integer>> versions = new TreeMap<>();
+ for (Map.Entry<Long, MaterializedIndexMeta> entry :
table.getIndexIdToMeta().entrySet()) {
+ MaterializedIndexMeta meta = entry.getValue();
+ versions.put(entry.getKey(), Pair.of(meta.getSchemaVersion(),
meta.getSchemaHash()));
+ }
+ return versions;
+ }
+
+ private static Pair<Long, Long> 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<Long> 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<Long, Long> 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<Long, Long> 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<Long, Long> scanned = version(partition,
entry.getValue());
+ Pair<Long, Long> 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<Long> getPartitionVersions(List<Partition> partitions)
throws RpcException {
+ if (!Config.isCloudMode()) {
+ return Partition.getVisibleVersions(partitions);
+ }
+ List<CloudPartition> 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<Partition> partitions = new ArrayList<>(partitionVersions.size());
+ for (Long id : partitionVersions.keySet()) {
+ partitions.add(table.getPartition(id));
+ }
+ List<Long> 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<Long, Long> 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<Variable> 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<Long, Long>
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<Long> 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<FullTableName, String> 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<FullTableName, String> 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<Long> 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<Long> selectedPartitionIds = Lists.newArrayList();
+
+ private final Map<Long, Long> sqlCachePartitionVersions =
Maps.newHashMap();
private Map<Long, String> 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<Tablet> 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<Long, Long> 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> config = Mockito.mockStatic(Config.class);
+ MockedStatic<CloudPartition> 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> config = Mockito.mockStatic(Config.class);
+ MockedStatic<CloudPartition> 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}')"
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
