[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.

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();

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);
         }

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;
+    }
+}

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)
+                    .append(tableVersion.getCacheKey());
-                            : "sql-cache:" + tableVersion.sqlCacheVersion.getVersion());
         }
         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 */

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());

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(

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;

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));
+        }
+    }
+}

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
+

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
+

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
+

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

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}')"
+        }
+    }
+}

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}')"
+        }
+    }
+}

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}')"
+        }
+    }
+}