diff --git a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizer.java b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizer.java
index 4a190f9808..f4d89e2b81 100644
--- a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizer.java
+++ b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizer.java
@@ -37,7 +37,9 @@
 import com.apple.foundationdb.relational.recordlayer.metadata.DataTypeUtils;
 import com.apple.foundationdb.relational.recordlayer.metadata.RecordLayerInvokedRoutine;
 import com.apple.foundationdb.relational.recordlayer.metadata.RecordLayerSchemaTemplate;
+import com.apple.foundationdb.relational.recordlayer.query.cache.PlanCacheSchemaKey;
 import com.apple.foundationdb.relational.recordlayer.query.cache.QueryCacheKey;
+import com.google.common.collect.ImmutableSortedMap;
 import com.apple.foundationdb.relational.recordlayer.util.ExceptionUtil;
 import com.apple.foundationdb.relational.util.Assert;
 import com.google.common.annotations.VisibleForTesting;
@@ -630,10 +632,10 @@ public static NormalizationResult normalizeAst(@Nonnull final SchemaTemplate sch
             }
         }
         return new NormalizationResult(
-                recordLayerSchemaTemplate.getName(),
+                PlanCacheSchemaKey.of(recordLayerSchemaTemplate.getName()),
                 QueryCacheKey.of(astNormalizer.getCanonicalSqlString(), getQuerySpecificPlannerConfig(plannerConfiguration, astNormalizer.getQueryOptions()),
                         recordLayerSchemaTemplate.getTransactionBoundMetadataAsString(),
-                        recordLayerSchemaTemplate.getVersion(), userVersion),
+                        ImmutableSortedMap.of(recordLayerSchemaTemplate.getName(), recordLayerSchemaTemplate.getVersion()), userVersion),
                 astNormalizer.getQueryExecutionParameters(),
                 parseTreeInfo.getRootContext(),
                 astNormalizer.getQueryCachingFlags(),
@@ -677,7 +679,7 @@ public enum QueryCachingFlags {
         }
 
         @Nonnull
-        private final String schemaTemplateName;
+        private final PlanCacheSchemaKey schemaKey;
 
         @Nonnull
         private final QueryCacheKey queryCacheKey;
@@ -697,14 +699,14 @@ public enum QueryCachingFlags {
         @Nonnull
         private final String query;
 
-        public NormalizationResult(@Nonnull final String schemaTemplateName,
+        public NormalizationResult(@Nonnull final PlanCacheSchemaKey schemaKey,
                                    @Nonnull final QueryCacheKey queryCacheKey,
                                    @Nonnull final QueryExecutionContext queryExecutionContext,
                                    @Nonnull final ParseTree parseTree,
                                    @Nonnull final Set<QueryCachingFlags> queryCachingFlags,
                                    @Nonnull final Options queryOptions,
                                    @Nonnull final String query) {
-            this.schemaTemplateName = schemaTemplateName;
+            this.schemaKey = schemaKey;
             this.queryCacheKey = queryCacheKey;
             this.queryExecutionContext = queryExecutionContext;
             this.parseTree = parseTree;
@@ -714,8 +716,8 @@ public NormalizationResult(@Nonnull final String schemaTemplateName,
         }
 
         @Nonnull
-        public String getSchemaTemplateName() {
-            return schemaTemplateName;
+        public PlanCacheSchemaKey getSchemaKey() {
+            return schemaKey;
         }
 
         @Nonnull
diff --git a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/PlanGenerator.java b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/PlanGenerator.java
index 6066385e16..e8019cecc3 100644
--- a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/PlanGenerator.java
+++ b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/PlanGenerator.java
@@ -173,7 +173,7 @@ private Plan<?> getPlanInternal(@Nonnull String query, @Nonnull KeyValueLogMessa
             final var planEquivalence = PhysicalPlanEquivalence.of(astHashResult.getQueryExecutionContext().getEvaluationContext());
             return planContext.getMetricsCollector().clock(RelationalMetric.RelationalEvent.CACHE_LOOKUP, () ->
                     cache.get().reduce(
-                            astHashResult.getSchemaTemplateName(),
+                            astHashResult.getSchemaKey(),
                             astHashResult.getQueryCacheKey(),
                             planEquivalence,
                             () -> {
diff --git a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/PlanCacheSchemaKey.java b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/PlanCacheSchemaKey.java
new file mode 100644
index 0000000000..0fcfe6cfd3
--- /dev/null
+++ b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/PlanCacheSchemaKey.java
@@ -0,0 +1,84 @@
+/*
+ * PlanCacheSchemaKey.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2021-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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 com.apple.foundationdb.relational.recordlayer.query.cache;
+
+import com.apple.foundationdb.annotation.API;
+import com.google.common.collect.ImmutableSortedSet;
+
+import javax.annotation.Nonnull;
+import java.util.Collection;
+
+/**
+ * Primary key for {@link RelationalPlanCache}. Represents the set of schema template names that a
+ * cached plan was compiled against. Single-schema queries produce a one-element set; cross-schema
+ * queries produce a multi-element set.
+ * <p>
+ * The set is always sorted so that {@code {"a","b"}} and {@code {"b","a"}} map to the same bucket.
+ */
+@API(API.Status.EXPERIMENTAL)
+public final class PlanCacheSchemaKey {
+
+    @Nonnull
+    private final ImmutableSortedSet<String> schemaNames;
+
+    private final int memoizedHashCode;
+
+    private PlanCacheSchemaKey(@Nonnull final ImmutableSortedSet<String> schemaNames) {
+        this.schemaNames = schemaNames;
+        this.memoizedHashCode = schemaNames.hashCode();
+    }
+
+    @Nonnull
+    public ImmutableSortedSet<String> getSchemaNames() {
+        return schemaNames;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+        return schemaNames.equals(((PlanCacheSchemaKey) other).schemaNames);
+    }
+
+    @Override
+    public int hashCode() {
+        return memoizedHashCode;
+    }
+
+    @Override
+    public String toString() {
+        return String.join("|", schemaNames);
+    }
+
+    @Nonnull
+    public static PlanCacheSchemaKey of(@Nonnull final String schemaName) {
+        return new PlanCacheSchemaKey(ImmutableSortedSet.of(schemaName));
+    }
+
+    @Nonnull
+    public static PlanCacheSchemaKey of(@Nonnull final Collection<String> schemaNames) {
+        return new PlanCacheSchemaKey(ImmutableSortedSet.copyOf(schemaNames));
+    }
+}
diff --git a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/QueryCacheKey.java b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/QueryCacheKey.java
index fb2a8a9112..2f5cf6cbcb 100644
--- a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/QueryCacheKey.java
+++ b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/QueryCacheKey.java
@@ -24,6 +24,7 @@
 
 import com.apple.foundationdb.relational.recordlayer.query.AstNormalizer;
 import com.apple.foundationdb.relational.recordlayer.query.PlannerConfiguration;
+import com.google.common.collect.ImmutableSortedMap;
 
 import javax.annotation.Nonnull;
 import java.util.Objects;
@@ -32,53 +33,13 @@
  * This is used to look up a plan in the primary cache (see {@link MultiStageCache} for more information).
  * It comprises the following fields:
  * <ul>
- *     <li>The schema template name to which the query is bound. It is necessary to use it so we can segregate
- *     the otherwise identical plans but coming from different schemas (see Example 1 below)</li>
- *     <li>The schema template version, user version, and a bit-set of all readable indexes</li>
- *     <li>The canonical query string where all literals are removed and white spaces are normalised (see example 2)</li>
+ *     <li>The schema template versions to which the query is bound, keyed by schema template name. A plan is
+ *     invalidated when <em>any</em> participating schema bumps its version.</li>
+ *     <li>The user version and a bit-set of all readable indexes.</li>
+ *     <li>The canonical query string where all literals are removed and white spaces are normalised (see example below).</li>
  *     <li>The hash of the query, see {@link AstNormalizer} for more information on how is this generated.</li>
  * </ul>
- * <b>Example1</b>
- * <br>
- * Let us assume we have two schema templates ({@code s1} and {@code s2}) defined as the following:
- * <pre>
- * {@code
- * create schema template s1
- * create table t1(id bigint, col1 bigint, primary key(id))
- *
- * create schema template s2
- * create table t1(id bigint, col1 bigint, col3 bigint, primary key(id))
- * }
- * </pre>
- * If we run a query like this:
- * <pre>
- * {@code
- * create schema /FRL/YOUSSEF/s1s with s1
- * connect: "jdbc:embed:/FRL/YOUSSEF?schema=S1S"
- * select * from t1 where col1 > 42;
- * }
- * </pre>
- * we would compile the query and return a result set comprising two columns {@code id, col1}.
- * we will also cache this query using a {@link QueryCacheKey} key of something like {@code "s1", "select * from t1 where col1 > ? ", 123456789}
- * <br>
- * if we run the <i>same</i> query, however after connecting to a schema that uses a {@code s2} instead:
- * <pre>
- * {@code
- * create schema /FRL/YOUSSEF/s2s with s2
- * connect: "jdbc:embed:/FRL/YOUSSEF?schema=S2S"
- * select * from t1 where col1 > 53;
- * }
- * </pre>
- * we would correctly compile this query and return a result set comprising three columns {@code id, col1, col2}.
- * we will also cache this query using {@link QueryCacheKey} key of something like {@code "s2", "select * from t1 where col1 > ? ", 123456789}
- * <br>
- * Note that without having the schema template name as part of the {@link QueryCacheKey} both keys would be identical.
- * Therefore, we might incorrectly choose the compiled plan of the first query to execute the second query (because we find
- * a match in the cache) causing an error since the result set structure is different because table {@code T1} is defined
- * differently in {@code S1} and {@code S2}.
- * <br>
- * <br>
- * <b>Example 2</b>
+ * <b>Example</b>
  * <br>
  * Although these queries appear different, their canonical representation is the same, and will end up using the same
  * compiled plan.
@@ -108,7 +69,12 @@ public final class QueryCacheKey {
     @Nonnull
     private final String auxiliaryMetadata;
 
-    private final int schemaTemplateVersion;
+    /**
+     * Maps each participating schema template name to its version. A cache miss is triggered when
+     * any entry in this map changes, which is the correct behaviour for cross-schema queries.
+     */
+    @Nonnull
+    private final ImmutableSortedMap<String, Integer> schemaVersions;
 
     private final int userVersion;
 
@@ -117,18 +83,15 @@ public final class QueryCacheKey {
     private QueryCacheKey(@Nonnull final String canonicalQueryString,
                           @Nonnull final PlannerConfiguration plannerConfiguration,
                           @Nonnull final String auxiliaryMetadata,
-                          int schemaTemplateVersion,
+                          @Nonnull final ImmutableSortedMap<String, Integer> schemaVersions,
                           int userVersion) {
         this.canonicalQueryString = canonicalQueryString;
-        this.schemaTemplateVersion = schemaTemplateVersion;
+        this.schemaVersions = schemaVersions;
         this.userVersion = userVersion;
         this.auxiliaryMetadata = auxiliaryMetadata;
         this.plannerConfiguration = plannerConfiguration;
 
-        // Memoize the hash code. Because this object is used as a key in a hash map, it is important that
-        // hashCode() be quick. Note that this includes information about the query (canonicalQueryString is like the query hash),
-        // the schema template version, and the schema (like the set of readable indexes)
-        this.memoizedHashCode = Objects.hash(canonicalQueryString, schemaTemplateVersion, plannerConfiguration, userVersion, auxiliaryMetadata);
+        this.memoizedHashCode = Objects.hash(canonicalQueryString, schemaVersions, plannerConfiguration, userVersion, auxiliaryMetadata);
     }
 
     @Override
@@ -140,11 +103,11 @@ public boolean equals(Object other) {
             return false;
         }
         final var that = (QueryCacheKey) other;
-        return schemaTemplateVersion == that.schemaTemplateVersion &&
-                userVersion == that.userVersion &&
+        return userVersion == that.userVersion &&
                 Objects.equals(canonicalQueryString, that.canonicalQueryString) &&
                 Objects.equals(auxiliaryMetadata, that.auxiliaryMetadata) &&
-                Objects.equals(plannerConfiguration, that.plannerConfiguration);
+                Objects.equals(plannerConfiguration, that.plannerConfiguration) &&
+                Objects.equals(schemaVersions, that.schemaVersions);
     }
 
     @Override
@@ -157,8 +120,9 @@ public String getCanonicalQueryString() {
         return canonicalQueryString;
     }
 
-    public int getSchemaTemplateVersion() {
-        return schemaTemplateVersion;
+    @Nonnull
+    public ImmutableSortedMap<String, Integer> getSchemaVersions() {
+        return schemaVersions;
     }
 
     @Nonnull
@@ -177,16 +141,15 @@ public String getAuxiliaryMetadata() {
 
     @Override
     public String toString() {
-        return "(" + schemaTemplateVersion + " || " + auxiliaryMetadata + ")" + "||" + canonicalQueryString + "||" + memoizedHashCode;
+        return "(" + schemaVersions + " || " + auxiliaryMetadata + ")" + "||" + canonicalQueryString + "||" + memoizedHashCode;
     }
 
     @Nonnull
     public static QueryCacheKey of(@Nonnull final String query,
                                    @Nonnull final PlannerConfiguration plannerConfiguration,
                                    @Nonnull final String auxiliaryMetadata,
-                                   int schemaTemplateVersion,
+                                   @Nonnull final ImmutableSortedMap<String, Integer> schemaVersions,
                                    int userVersion) {
-        return new QueryCacheKey(query, plannerConfiguration, auxiliaryMetadata, schemaTemplateVersion,
-                userVersion);
+        return new QueryCacheKey(query, plannerConfiguration, auxiliaryMetadata, schemaVersions, userVersion);
     }
 }
diff --git a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCache.java b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCache.java
index 13e8192d8a..d93fc8e43e 100644
--- a/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCache.java
+++ b/fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCache.java
@@ -34,7 +34,7 @@
  * This is just a specialization of {@link MultiStageCache} with concrete types specific to plan caching.
  */
 @API(API.Status.EXPERIMENTAL)
-public final class RelationalPlanCache extends MultiStageCache<String, QueryCacheKey, PhysicalPlanEquivalence, Plan<?>> {
+public final class RelationalPlanCache extends MultiStageCache<PlanCacheSchemaKey, QueryCacheKey, PhysicalPlanEquivalence, Plan<?>> {
 
     @Nonnull
     private static final TimeUnit DEFAULT_TTL_TIME_UNIT = TimeUnit.MILLISECONDS;
@@ -61,7 +61,7 @@ private RelationalPlanCache(int size,
         super(size, secondarySize, tertiarySize, ttl, ttlTimeUnit, secondaryTtl, secondaryTtlTimeUnit, tertiaryTtl, tertiaryTtlTimeUnit, executor, secondaryExecutor, tertiaryExecutor, ticker);
     }
 
-    public static final class RelationalCacheBuilder extends MultiStageCache.Builder<String, QueryCacheKey, PhysicalPlanEquivalence, Plan<?>, RelationalCacheBuilder> {
+    public static final class RelationalCacheBuilder extends MultiStageCache.Builder<PlanCacheSchemaKey, QueryCacheKey, PhysicalPlanEquivalence, Plan<?>, RelationalCacheBuilder> {
 
         public RelationalCacheBuilder() {
             size = (Integer) (Options.defaultOptions().get(Options.Name.PLAN_CACHE_PRIMARY_MAX_ENTRIES));
diff --git a/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizerTests.java b/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizerTests.java
index ba95364cf3..18496cf7b9 100644
--- a/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizerTests.java
+++ b/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/AstNormalizerTests.java
@@ -39,6 +39,7 @@
 import com.apple.foundationdb.relational.recordlayer.metadata.RecordLayerSchemaTemplate;
 import com.apple.foundationdb.relational.recordlayer.metadata.RecordLayerTable;
 import com.apple.foundationdb.relational.recordlayer.query.cache.QueryCacheKey;
+import com.google.common.collect.ImmutableSortedMap;
 import com.apple.foundationdb.relational.recordlayer.query.functions.CompiledSqlFunction;
 import com.apple.foundationdb.relational.recordlayer.util.Hex;
 import com.apple.foundationdb.relational.util.Assert;
@@ -1496,8 +1497,9 @@ void visitFullDescribeStatementThrows() throws ReflectiveOperationException {
 
     @Test
     void queryCacheKeyToString() {
-        final var key = QueryCacheKey.of("select ? from testTable", plannerConfiguration, "someAuxiliaryMetadata", 3, 7);
-        final var expected = "(3 || someAuxiliaryMetadata)||select ? from testTable||" + key.hashCode();
+        final var key = QueryCacheKey.of("select ? from testTable", plannerConfiguration, "someAuxiliaryMetadata",
+                ImmutableSortedMap.of("testSchema", 3), 7);
+        final var expected = "({testSchema=3} || someAuxiliaryMetadata)||select ? from testTable||" + key.hashCode();
         Assertions.assertThat(key).hasToString(expected);
     }
 }
diff --git a/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/cache/PlanCacheSchemaKeyTest.java b/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/cache/PlanCacheSchemaKeyTest.java
new file mode 100644
index 0000000000..66debd3a4d
--- /dev/null
+++ b/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/cache/PlanCacheSchemaKeyTest.java
@@ -0,0 +1,71 @@
+/*
+ * PlanCacheSchemaKeyTest.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2021-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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 com.apple.foundationdb.relational.recordlayer.query.cache;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+class PlanCacheSchemaKeyTest {
+
+    @Test
+    void ofCollectionProducesSortedKey() {
+        final List<String> names = Arrays.asList("schema_b", "schema_a", "schema_c");
+        final PlanCacheSchemaKey key = PlanCacheSchemaKey.of(names);
+        assertEquals(List.of("schema_a", "schema_b", "schema_c"), List.copyOf(key.getSchemaNames()));
+    }
+
+    @Test
+    void ofCollectionEqualsSingletonOfForSingleElement() {
+        final PlanCacheSchemaKey fromCollection = PlanCacheSchemaKey.of(List.of("my_schema"));
+        final PlanCacheSchemaKey fromSingle = PlanCacheSchemaKey.of("my_schema");
+        assertEquals(fromSingle, fromCollection);
+        assertEquals(fromSingle.hashCode(), fromCollection.hashCode());
+    }
+
+    @Test
+    void ofCollectionOrderIndependent() {
+        final PlanCacheSchemaKey k1 = PlanCacheSchemaKey.of(Arrays.asList("alpha", "beta"));
+        final PlanCacheSchemaKey k2 = PlanCacheSchemaKey.of(Arrays.asList("beta", "alpha"));
+        assertEquals(k1, k2);
+        assertEquals(k1.hashCode(), k2.hashCode());
+    }
+
+    @Test
+    void toStringSingleSchema() {
+        assertEquals("my_schema", PlanCacheSchemaKey.of("my_schema").toString());
+    }
+
+    @Test
+    void toStringMultiSchemaJoinedWithPipe() {
+        final PlanCacheSchemaKey key = PlanCacheSchemaKey.of(Arrays.asList("schema_b", "schema_a"));
+        assertEquals("schema_a|schema_b", key.toString());
+    }
+
+    @Test
+    void differentSchemasDontMatch() {
+        assertNotEquals(PlanCacheSchemaKey.of("schema_x"), PlanCacheSchemaKey.of("schema_y"));
+    }
+}
diff --git a/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCacheTests.java b/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCacheTests.java
index 027f3d1dbe..91bd650ce9 100644
--- a/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCacheTests.java
+++ b/fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/query/cache/RelationalPlanCacheTests.java
@@ -54,6 +54,7 @@
 import com.apple.foundationdb.relational.utils.TestSchemas;
 import com.apple.test.BooleanSource;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.testing.FakeTicker;
 import org.apache.commons.lang3.tuple.Pair;
 import org.assertj.core.groups.Tuple;
@@ -407,12 +408,12 @@ private static QueryPlanConstraint cons(@Nonnull final QueryPlanConstraint... co
 
     private static void shouldBe(@Nonnull final RelationalPlanCache cache, @Nonnull final Map<Tuple, Map<PhysicalPlanEquivalence, String>> expectedLayout) {
         Map<Tuple, Map<PhysicalPlanEquivalence, String>> result = new HashMap<>();
-        for (String key : cache.getStats().getAllKeys()) {
+        for (PlanCacheSchemaKey key : cache.getStats().getAllKeys()) {
             for (QueryCacheKey secondaryKey : cache.getStats().getAllSecondaryKeys(key)) {
                 final var resMap = result.computeIfAbsent(
                         new Tuple(secondaryKey.getCanonicalQueryString(),
                                     key,
-                                    secondaryKey.getSchemaTemplateVersion(),
+                                    secondaryKey.getSchemaVersions(),
                                     secondaryKey.getUserVersion(),
                                     secondaryKey.getPlannerConfiguration(),
                                     secondaryKey.getAuxiliaryMetadata()),
@@ -515,13 +516,13 @@ void testCachingDifferentQueries() throws Exception {
 
         // adding an entry to the cache to warm it up.
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // let's add an identical query with different boundaries, so we can use a different index (i1980).
         // however, we end up with in the same primary cache bucket because the primary key is identical.
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1980 AND YEAR < 1985", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1980);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(
                         ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(cons(c1980Cp0(7), c1980Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1980)));
@@ -529,11 +530,11 @@ void testCachingDifferentQueries() throws Exception {
         // let's try another query, now the query itself is different -> must be a different entry in the _primary_ cache.
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1980 OR YEAR < 1985", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), Scan);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(
                         ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(cons(c1980Cp0(7), c1980Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1980),
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? OR \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? OR \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 // tautology is expected since a primary scan accepts everything.
                 Map.of(ppe(tautology, cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), Scan)
         ));
@@ -546,16 +547,16 @@ void testCachingDifferentSchemaTemplateNames() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues exactly the same query, the environment is identical to first query, but the schema template is different ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_2", 10, 100, Set.of(i1970, i1980), i1970);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970),
                 // so it is added as a separate entry in the main cache
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_2", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_2"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_2", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
     }
 
@@ -566,16 +567,16 @@ void testCachingDifferentSchemaTemplateVersions() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues exactly the same query, the environment is identical to first query, but the schema template version is different ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 11, 100, Set.of(i1970, i1980), i1970);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970),
                 // so it is added as a separate entry in the main cache
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 11, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 11), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
     }
 
@@ -586,16 +587,16 @@ void testCachingDifferentUserVersion() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues exactly the same query, the environment is identical to first query, but the user version is different ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 101, Set.of(i1970, i1980), i1970);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970),
                 // so it is added as a separate entry in the main cache
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 101, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 101, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
     }
 
@@ -606,7 +607,7 @@ void testCachingDifferentReadableIndexes() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues exactly the same query, the environment is identical to first query, but the readable indexes set is different ...
@@ -614,10 +615,10 @@ void testCachingDifferentReadableIndexes() throws Exception {
         // in that case, we want to make sure that we re-plan the query giving the optimizer a chance to consider the newly created index for producing a better plan.
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1970);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970),
                 // so it is added as a separate entry in the main cache
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
     }
 
@@ -628,12 +629,12 @@ void testCachingDifferentConstraints() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues exactly the same query, the environment is identical to first query, but which predicates that fall outside the ranges of the chosen index of the first query
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1980 AND YEAR < 1983", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1980);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(
                         ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(cons(c1980Cp0(7), c1980Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1980)));
@@ -646,7 +647,7 @@ void testEvictionFromPrimaryCache() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // 10 MS TTL for primary cache, if we pass 11 -> item must be evicted from primary cache.
@@ -661,24 +662,24 @@ void testEvictionFromPrimaryCacheWithLru() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues a different query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 OR YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), Scan);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970),
                 // ... which is added as a separate entry in the main cache
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? OR \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? OR \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(tautology, cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), Scan)));
 
         // customer 3 issues yet a different query, cache size is two, evicts an item ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), Scan);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(tautology, cons(ofTypeIntCp0(7), isNotNullInt(7))), Scan),
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
     }
 
@@ -689,7 +690,7 @@ void testEvictionFromSecondaryCacheRemovesPrimaryKeyWhenEmpty() throws Exception
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // 10 MS TTL for primary cache, 5 MS TTL for secondary cache, if we pass 7 -> item must be evicted from secondary cache
@@ -710,7 +711,7 @@ void testEvictionFromTertiaryCache() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // pass some time, so we have some jitter between secondary cache items necessary for producing deterministic test results.
@@ -718,7 +719,7 @@ void testEvictionFromTertiaryCache() throws Exception {
 
         // customer 2 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1980 AND YEAR < 1985", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1980);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(
                         ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(cons(c1980Cp0(7), c1980Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1980)));
@@ -726,7 +727,7 @@ void testEvictionFromTertiaryCache() throws Exception {
         // 10 MS TTL for primary cache, 5 MS TTL for secondary cache, we already passed 2, now if pass 3 more we'll
         // evict the first cached item in the secondary cache, but _not_ the more recent one.
         ticker.advance(Duration.of(3, ChronoUnit.MILLIS));
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1980Cp0(7), c1980Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1980)));
     }
 
@@ -737,12 +738,12 @@ void testEvictionFromSecondaryCacheWithLru() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1980 AND YEAR < 1985", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1980);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), ""),
                 Map.of(
                         ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(cons(c1980Cp0(7), c1980Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1980)));
@@ -750,7 +751,7 @@ void testEvictionFromSecondaryCacheWithLru() throws Exception {
         // secondary cache has capacity of two, attempting to add a third item causes an eviction (LRU).
         // customer 3 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1990 AND YEAR < 1992", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1990);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), ""),
                 Map.of(
                         ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(cons(c1990Cp0(7), c1990Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1990)));
@@ -763,7 +764,7 @@ void testEvictionFromTertiaryCacheRemovesSecondaryKeyWhenEmpty() throws Exceptio
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11),
                         isNotNullInt(7), isNotNullInt(11))), i1970)));
 
@@ -778,7 +779,7 @@ void testEvictionFromTertiaryCacheRemovesSecondaryKeyWhenEmpty() throws Exceptio
         cache.cleanUp();
 
         // Only one secondary key, the first one was removed
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(tautology, cons(ofTypeIntCp0(7), isNotNullInt(7))), Scan)));
     }
 
@@ -789,14 +790,14 @@ void testPlanReductionViaCosting() throws Exception {
 
         // customer 1 issues a query ...
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11),
                         isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues a query that forces a "bad" plan within the same secondary cache entry of the plan above
         // because of scan boundaries that fall outside the range of the filtered index
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 2005 AND YEAR < 2010", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), Scan);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11),
                         isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(tautology, cons(ofTypeIntCp0(7), ofTypeIntCp1(11),
@@ -804,7 +805,7 @@ void testPlanReductionViaCosting() throws Exception {
 
         // we should still get back the "good" plan (scanning i1970)
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11),
                                 isNotNullInt(7), isNotNullInt(11))), i1970,
                         ppe(tautology, cons(ofTypeIntCp0(7), ofTypeIntCp1(11),
@@ -821,7 +822,7 @@ void testConstraintsWithTemporaryFunction() throws Exception {
                 "CREATE TEMPORARY FUNCTION SCI_FI_BOOKS_OF_80S() ON COMMIT DROP FUNCTION AS SELECT * FROM SCI_FI_BOOKS() WHERE YEAR > 1980 AND YEAR < 1989"),
                 "SELECT * FROM SCI_FI_BOOKS_OF_80S()", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1980);
 
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"SCI_FI_BOOKS_OF_80S\" ( ) ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), "CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS\" ( ) " +
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"SCI_FI_BOOKS_OF_80S\" ( ) ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), "CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"BOOKS\" WHERE \"TITLE\" LIKE ? ||CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS_OF_80S\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"SCI_FI_BOOKS\" ( ) WHERE \"YEAR\" > ? AND \"YEAR\" < ? "),
                 Map.of(ppe(cons(
@@ -848,7 +849,7 @@ void testConstraintsWithTemporaryFunctionsIncludingUnusedOnes() throws Exception
                         "CREATE TEMPORARY FUNCTION SCI_FI_BOOKS_OF_80S() ON COMMIT DROP FUNCTION AS SELECT * FROM SCI_FI_BOOKS() WHERE YEAR > 1980 AND YEAR < 1989"),
                 "SELECT * FROM SCI_FI_BOOKS_OF_80S()", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1980);
 
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"SCI_FI_BOOKS_OF_80S\" ( ) ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), "CREATE TEMPORARY FUNCTION \"OTHER_BOOKS\" ( ) " +
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"SCI_FI_BOOKS_OF_80S\" ( ) ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), "CREATE TEMPORARY FUNCTION \"OTHER_BOOKS\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"BOOKS\" WHERE \"TITLE\" LIKE ? ||CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"BOOKS\" WHERE \"TITLE\" LIKE ? ||CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS_OF_80S\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"SCI_FI_BOOKS\" ( ) WHERE \"YEAR\" > ? AND \"YEAR\" < ? "),
@@ -873,7 +874,7 @@ void testConstraintsWithTemporaryFunctionsMultipleReferences() throws Exception
                         "CREATE TEMPORARY FUNCTION SCI_FI_BOOKS_OF_80S() ON COMMIT DROP FUNCTION AS SELECT * FROM SCI_FI_BOOKS() WHERE YEAR > 1980 AND YEAR < 1989"),
                 "SELECT * FROM SCI_FI_BOOKS_OF_80S(), SCI_FI_BOOKS_OF_80S()", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1980);
 
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"SCI_FI_BOOKS_OF_80S\" ( ) , \"SCI_FI_BOOKS_OF_80S\" ( ) ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"SCI_FI_BOOKS_OF_80S\" ( ) , \"SCI_FI_BOOKS_OF_80S\" ( ) ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)),
                         "CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"BOOKS\" WHERE \"TITLE\" LIKE ? ||CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS_OF_80S\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"SCI_FI_BOOKS\" ( ) WHERE \"YEAR\" > ? AND \"YEAR\" < ? "),
@@ -900,7 +901,7 @@ void testConstraintsWithTemporaryFunctionsMultipleLiterals() throws Exception {
                 "SELECT * FROM SCI_FI_BOOKS_OF_80S() AS A, OTHER_BOOKS() AS B WHERE A.YEAR > 1985 AND A.TITLE = 'OTHER'", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980, i1990), i1980);
 
         shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"SCI_FI_BOOKS_OF_80S\" ( ) AS \"A\" , \"OTHER_BOOKS\" ( ) AS \"B\" WHERE \"A\" . \"YEAR\" > ? AND \"A\" . \"TITLE\" = ? ",
-                        "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980, i1990)), "CREATE TEMPORARY FUNCTION \"OTHER_BOOKS\" ( ) " +
+                        PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980, i1990)), "CREATE TEMPORARY FUNCTION \"OTHER_BOOKS\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"BOOKS\" WHERE \"TITLE\" LIKE ? ||CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"BOOKS\" WHERE \"TITLE\" LIKE ? ||CREATE TEMPORARY FUNCTION \"SCI_FI_BOOKS_OF_80S\" ( ) " +
                         "ON COMMIT DROP FUNCTION AS SELECT * FROM \"SCI_FI_BOOKS\" ( ) WHERE \"YEAR\" > ? AND \"YEAR\" < ? "),
@@ -929,7 +930,7 @@ void testPlanningQueryWithAndWithoutDisabledPlannerRewriteRules() throws Excepti
         // customer 1 issues a query with disabled planner rules
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), disabledRulesOption, i1970);
         shouldBe(cache, Map.of(
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980), disabledRulesOption), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980), disabledRulesOption), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // customer 2 issues the same query, with enabled planner rules.
@@ -938,10 +939,10 @@ void testPlanningQueryWithAndWithoutDisabledPlannerRewriteRules() throws Excepti
         // cache should contain two entries
         shouldBe(cache, Map.of(
                 // ... one with disabled planner rules.
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980), disabledRulesOption), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980), disabledRulesOption), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970),
                 // ... another one with enabled planner rules.
-                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // this is needed because expiration is done passively for better performance.
@@ -966,15 +967,15 @@ void testPlanningQueryWithAndWithoutPlanRightDeepOption(boolean specifyInQuery)
 
         planQuery(cache, queryWithOption, "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), optionsWithOption, i1970);
         shouldBe(cache, Map.of(
-                new Tuple(expectedCononicalString, "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980), rightDeepOption), ""),
+                new Tuple(expectedCononicalString, PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980), rightDeepOption), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         // now run the plain query (no option either way)
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), Options.none(), i1970);
         shouldBe(cache, Map.of(
-                new Tuple(expectedCononicalString, "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980), rightDeepOption), ""),
+                new Tuple(expectedCononicalString, PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980), rightDeepOption), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970),
-                new Tuple(expectedCononicalString, "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+                new Tuple(expectedCononicalString, PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         cache.cleanUp();
@@ -986,15 +987,15 @@ void testExplainReusesCachedPlan() throws Exception {
         final var cache = getCache(ticker);
 
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         planQuery(cache, "SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
 
         planQuery(cache, "EXPLAIN SELECT * FROM BOOKS WHERE YEAR > 1970 AND YEAR < 1979", "SCHEMA_TEMPLATE_1", 10, 100, Set.of(i1970, i1980), i1970);
-        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", "SCHEMA_TEMPLATE_1", 10, 100, configOf(Set.of(i1970, i1980)), ""),
+        shouldBe(cache, Map.of(new Tuple("SELECT * FROM \"BOOKS\" WHERE \"YEAR\" > ? AND \"YEAR\" < ? ", PlanCacheSchemaKey.of("SCHEMA_TEMPLATE_1"), ImmutableSortedMap.of("SCHEMA_TEMPLATE_1", 10), 100, configOf(Set.of(i1970, i1980)), ""),
                 Map.of(ppe(cons(c1970Cp0(7), c1970Cp1(11)), cons(ofTypeIntCp0(7), ofTypeIntCp1(11), isNotNullInt(7), isNotNullInt(11))), i1970)));
     }
 }