diff --git a/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-against-ctas-materialized.json b/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-against-ctas-materialized.json
index a7f370f4630a..7892d8f98935 100644
--- a/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-against-ctas-materialized.json
+++ b/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-against-ctas-materialized.json
@@ -794,6 +794,351 @@
         ]}
       ]
     },
+    {
+      "name": "windowed - select star with HAVING filter - AVRO two keys",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, V0 STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', format='AVRO');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, V0, COUNT(1) AS COUNT FROM INPUT WINDOW TUMBLING(SIZE 1 SECOND) GROUP BY ID, V0 HAVING COUNT(1) > 1;",
+        "SELECT * FROM AGGREGATE WHERE ID='10' and V0='v10a';"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12346, "key": {"ID": "11", "v0": "v10b"}, "value": {"val": 1}},
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"val": 2}},
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"val": 5}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `V0` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", "v10a", 12000, 13000, 2]}}
+        ]}
+      ]
+    },
+    {
+      "name": "windowed - select star with HAVING filter - AVRO",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='AVRO');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW TUMBLING(SIZE 1 SECOND) GROUP BY ID HAVING COUNT(1) > 1;",
+        "SELECT * FROM AGGREGATE;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12346, "key": "11", "value": {"val": 1}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {"val": 2}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {"val": 5}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 2]}}
+        ]}
+      ]
+    },
+    {
+      "name": "tumbling windowed single key lookup with window start range - AVRO",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='AVRO');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW TUMBLING(SIZE 1 SECOND) WHERE ID='10' GROUP BY ID;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 100 <= WindowStart AND WindowStart < 200000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 <= WindowStart AND WindowStart < 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 < WindowStart AND WindowStart <= 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE WindowStart > 17000 AND 11234756356 > WindowStart;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12001, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 12211, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 14253, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 15364, "key": "10", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}},
+          {"row":{"columns":["10", 14000, 15000, 14253, 1]}},
+          {"row":{"columns":["10", 15000, 16000, 15364, 1]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 14000, 15000, 14253, 1]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]}
+      ]
+    },
+    {
+      "name": "tumbling windowed single key lookup with window start range HAVING CLAUSE - AVRO",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='AVRO');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW TUMBLING(SIZE 1 SECOND) WHERE ID='10' GROUP BY ID HAVING COUNT(1)>1;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 100 <= WindowStart AND WindowStart < 200000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 <= WindowStart AND WindowStart < 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 < WindowStart AND WindowStart <= 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE WindowStart > 17000 AND 11234756356 > WindowStart;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12001, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 12211, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 14253, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 15364, "key": "10", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]}
+      ]
+    },
+    {
+      "name": "hopping windowed single key lookup with window start-end range HAVING CLAUSE - AVRO",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='AVRO');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW HOPPING(SIZE 5 SECOND, ADVANCE BY 1 SECONDS) WHERE ID IN ('9', '10') GROUP BY ID HAVING COUNT(1)>2;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 7000 <= WindowStart AND WindowEnd < 16000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 7000 < WindowStart AND WindowEnd <= 16000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 13001 <= WindowStart AND WindowEnd < 11234756356;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 10021, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 10345, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 13251, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 13251, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 10345, "key": "9", "value": {}},
+        {"topic": "test_topic", "timestamp": 13251, "key": "9", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 9000, 14000, 13251, 3]}},
+          {"row":{"columns":["10", 10000, 15000, 13251, 3]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 9000, 14000, 13251, 3]}},
+          {"row":{"columns":["10", 10000, 15000, 13251, 3]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]}
+      ]
+    },
+    {
+      "name": "session windowed single key lookup with window start range - AVRO",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='AVRO');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW SESSION(5 SECOND) WHERE ID = '10' GROUP BY ID HAVING COUNT(1)<2;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE ID = '10' AND 10 <= WindowStart AND WindowStart < 200000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE ID = '10' AND 8001 <= WindowStart AND WindowStart < 19444;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE ID = '10' AND 8001 < WindowStart AND WindowStart <= 19444;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 8001, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 10456, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 19444, "key": "10", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 19444, 19444, 19444, 1]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 19444, 19444, 19444, 1]}}
+        ]}
+      ]
+    },
+    {
+      "name": "windowed - select star with HAVING filter - PROTOBUF",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='PROTOBUF');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW TUMBLING(SIZE 1 SECOND) GROUP BY ID HAVING COUNT(1) > 1;",
+        "SELECT * FROM AGGREGATE;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12346, "key": "11", "value": {"val": 1}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {"val": 2}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {"val": 5}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 2]}}
+        ]}
+      ]
+    },
+    {
+      "name": "tumbling windowed single key lookup with window start range - PROTOBUF",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='PROTOBUF');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW TUMBLING(SIZE 1 SECOND) WHERE ID='10' GROUP BY ID;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 100 <= WindowStart AND WindowStart < 200000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 <= WindowStart AND WindowStart < 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 < WindowStart AND WindowStart <= 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE WindowStart > 17000 AND 11234756356 > WindowStart;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12001, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 12211, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 14253, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 15364, "key": "10", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}},
+          {"row":{"columns":["10", 14000, 15000, 14253, 1]}},
+          {"row":{"columns":["10", 15000, 16000, 15364, 1]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 14000, 15000, 14253, 1]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]}
+      ]
+    },
+    {
+      "name": "tumbling windowed single key lookup with window start range HAVING CLAUSE - PROTOBUF",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='PROTOBUF');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW TUMBLING(SIZE 1 SECOND) WHERE ID='10' GROUP BY ID HAVING COUNT(1)>1;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 100 <= WindowStart AND WindowStart < 200000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 <= WindowStart AND WindowStart < 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 12000 < WindowStart AND WindowStart <= 14000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE WindowStart > 17000 AND 11234756356 > WindowStart;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12001, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 12345, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 12211, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 14253, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 15364, "key": "10", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 12000, 13000, 12345, 2]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]}
+      ]
+    },
+    {
+      "name": "hopping windowed single key lookup with window start-end range HAVING CLAUSE - PROTOBUF",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='PROTOBUF');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW HOPPING(SIZE 5 SECOND, ADVANCE BY 1 SECONDS) WHERE ID IN ('9', '10') GROUP BY ID HAVING COUNT(1)>2;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 7000 <= WindowStart AND WindowEnd < 16000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 7000 < WindowStart AND WindowEnd <= 16000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE 13001 <= WindowStart AND WindowEnd < 11234756356;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 10021, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 10345, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 13251, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 13251, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 10345, "key": "9", "value": {}},
+        {"topic": "test_topic", "timestamp": 13251, "key": "9", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 9000, 14000, 13251, 3]}},
+          {"row":{"columns":["10", 10000, 15000, 13251, 3]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 9000, 14000, 13251, 3]}},
+          {"row":{"columns":["10", 10000, 15000, 13251, 3]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]}
+      ]
+    },
+    {
+      "name": "session windowed single key lookup with window start range - PROTOBUF",
+      "statements": [
+        "CREATE STREAM INPUT (ID STRING KEY, IGNORED INT) WITH (kafka_topic='test_topic', value_format='PROTOBUF');",
+        "CREATE TABLE AGGREGATE AS SELECT ID, COUNT(1) AS COUNT FROM INPUT WINDOW SESSION(5 SECOND) WHERE ID = '10' GROUP BY ID HAVING COUNT(1)<2;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE ID = '10' AND 10 <= WindowStart AND WindowStart < 200000;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE ID = '10' AND 8001 <= WindowStart AND WindowStart < 19444;",
+        "SELECT ID, WINDOWSTART, WINDOWEND, ROWTIME, COUNT FROM AGGREGATE WHERE ID = '10' AND 8001 < WindowStart AND WindowStart <= 19444;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": "11", "value": {}},
+        {"topic": "test_topic", "timestamp": 8001, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 10456, "key": "10", "value": {}},
+        {"topic": "test_topic", "timestamp": 19444, "key": "10", "value": {}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 19444, 19444, 19444, 1]}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}}
+        ]},
+        {"query": [
+          {"header":{"schema":"`ID` STRING KEY, `WINDOWSTART` BIGINT KEY, `WINDOWEND` BIGINT KEY, `ROWTIME` BIGINT, `COUNT` BIGINT"}},
+          {"row":{"columns":["10", 19444, 19444, 19444, 1]}}
+        ]}
+      ]
+    },
     {
       "name": "should not allow pull query with disallowed pseudocolumn in SELECT clause",
       "statements": [
diff --git a/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-with-predicates.json b/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-with-predicates.json
index 42d0681b1f9c..52494359d902 100644
--- a/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-with-predicates.json
+++ b/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-with-predicates.json
@@ -224,6 +224,226 @@
           {"row":{"columns":["11", "v11a", "v11b"]}}
         ]}
       ]
+    },
+    {
+      "name": "schema registry table-test one partition avro",
+      "statements": [
+        "CREATE TABLE INPUT (ID INT PRIMARY KEY, V0 STRING PRIMARY KEY, V1 STRING) WITH (kafka_topic='test_topic', format='AVRO', PARTITIONS=1);",
+        "CREATE TABLE MATVIEW AS SELECT * FROM INPUT;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `V1` STRING"}},
+          {"row":{"columns":[10, "v10a", "v10b"]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry table-test two partitions json",
+      "statements": [
+        "CREATE TABLE INPUT (ID INT PRIMARY KEY, V0 STRING PRIMARY KEY, V1 STRING) WITH (kafka_topic='test_topic', format='JSON', PARTITIONS=2);",
+        "CREATE TABLE MATVIEW AS SELECT * FROM INPUT;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `V1` STRING"}},
+          {"row":{"columns":[10, "v10a", "v10b"]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry table-test two partitions",
+      "statements": [
+        "CREATE TABLE INPUT (ID INT PRIMARY KEY, V0 STRING PRIMARY KEY, V1 STRING) WITH (kafka_topic='test_topic', format='AVRO', PARTITIONS=2);",
+        "CREATE TABLE MATVIEW AS SELECT * FROM INPUT;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `V1` STRING"}},
+          {"row":{"columns":[10, "v10a", "v10b"]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry stream-test one partition",
+      "statements": [
+        "CREATE STREAM INPUT (ID INT KEY, V0 STRING KEY, V1 STRING) WITH (kafka_topic='test_topic', format='AVRO', PARTITIONS=1);",
+        "CREATE TABLE MATVIEW AS SELECT ID, V0, COUNT(1) AS CNT FROM INPUT GROUP BY ID, V0;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `CNT` BIGINT"}},
+          {"row":{"columns":[10, "v10a", 1]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry stream-test two partitions",
+      "statements": [
+        "CREATE STREAM INPUT (ID INT KEY, V0 STRING KEY, V1 STRING) WITH (kafka_topic='test_topic', format='AVRO', PARTITIONS=2);",
+        "CREATE TABLE MATVIEW AS SELECT ID, V0, COUNT(1) AS CNT FROM INPUT GROUP BY ID, V0;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `CNT` BIGINT"}},
+          {"row":{"columns":[10, "v10a", 1]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry stream-test one partition protobuf",
+      "statements": [
+        "CREATE STREAM INPUT (ID INT KEY, V0 STRING KEY, V1 STRING) WITH (kafka_topic='test_topic', format='PROTOBUF', PARTITIONS=1);",
+        "CREATE TABLE MATVIEW AS SELECT ID, V0, COUNT(1) AS CNT FROM INPUT GROUP BY ID, V0;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `CNT` BIGINT"}},
+          {"row":{"columns":[10, "v10a", 1]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry stream-test two partitions protobuf",
+      "statements": [
+        "CREATE STREAM INPUT (ID INT KEY, V0 STRING KEY, V1 STRING) WITH (kafka_topic='test_topic', format='PROTOBUF', PARTITIONS=2);",
+        "CREATE TABLE MATVIEW AS SELECT ID, V0, COUNT(1) AS CNT FROM INPUT GROUP BY ID, V0;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `CNT` BIGINT"}},
+          {"row":{"columns":[10, "v10a", 1]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry table-test one partition protobuf",
+      "statements": [
+        "CREATE TABLE INPUT (ID INT PRIMARY KEY, V0 STRING PRIMARY KEY, V1 STRING) WITH (kafka_topic='test_topic', format='PROTOBUF', PARTITIONS=1);",
+        "CREATE TABLE MATVIEW AS SELECT * FROM INPUT;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `V1` STRING"}},
+          {"row":{"columns":[10, "v10a", "v10b"]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry table-test two partitions protobuf",
+      "statements": [
+        "CREATE TABLE INPUT (ID INT PRIMARY KEY, V0 STRING PRIMARY KEY, V1 STRING) WITH (kafka_topic='test_topic', format='PROTOBUF', PARTITIONS=2);",
+        "CREATE TABLE MATVIEW AS SELECT * FROM INPUT;",
+        "SELECT * FROM MATVIEW WHERE ID=10 and V0='v10a';"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": {"ID": "10", "v0": "v10a"}, "value": {"v1": "v10b"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `V0` STRING KEY, `V1` STRING"}},
+          {"row":{"columns":[10, "v10a", "v10b"]}}
+        ]}
+      ]
+    },
+    {
+      "name": "schema registry table-test kafka-avro",
+      "statements": [
+        "CREATE TABLE INPUT (ID INT PRIMARY KEY, V0 STRING) WITH (kafka_topic='test_topic', key_format='KAFKA', value_format='AVRO', partitions=4);",
+        "CREATE TABLE MATVIEW WITH (format='AVRO') as SELECT id, id + 1, COUNT(*) from INPUT group by id, id + 1;",
+        "SELECT * FROM MATVIEW WHERE ID=1 and KSQL_COL_0=2;"
+      ],
+      "properties": {
+        "ksql.query.pull.table.scan.enabled":  true
+      },
+      "inputs": [
+        {"topic": "test_topic", "timestamp": 12345, "key": 1, "value": {"v0": "v10a"}}
+      ],
+      "responses": [
+        {"admin": {"@type": "currentStatus"}},
+        {"admin": {"@type": "currentStatus"}},
+        {"query": [
+          {"header":{"schema":"`ID` INTEGER KEY, `KSQL_COL_0` INTEGER KEY, `KSQL_COL_1` BIGINT"}},
+          {"row":{"columns":[1, 2, 1]}}
+        ]}
+      ]
     }
   ]
 }
\ No newline at end of file
diff --git a/ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocator.java b/ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocator.java
index fe7bfc9fcaf8..416e199ac604 100644
--- a/ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocator.java
+++ b/ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocator.java
@@ -119,9 +119,15 @@ public List<KsqlPartitionLocation> locate(
     // Depending on whether this is a key-based lookup, determine which metadata method to use.
     // If we don't have keys, find the metadata for all partitions since we'll run the query for
     // all partitions of the state store rather than a particular one.
-    final List<PartitionMetadata> metadata = keys.isEmpty() || isRangeScan
-        ? getMetadataForAllPartitions(filterPartitions, keySet)
-        : getMetadataForKeys(keys, filterPartitions);
+    //For issue #7174. Temporarily turn off metadata finding for a partition with keys
+    //if there are more than one key.
+    final List<PartitionMetadata> metadata;
+    if (keys.size() == 1 && keys.get(0).getKey().size() == 1 && !isRangeScan) {
+      metadata = getMetadataForKeys(keys, filterPartitions);
+    } else {
+      metadata = getMetadataForAllPartitions(filterPartitions, keySet);
+    }
+
     // Go through the metadata and group them by partition.
     for (PartitionMetadata partitionMetadata : metadata) {
       LOG.debug("Handling pull query for partition {} of state store {}.",
diff --git a/ksqldb-streams/src/test/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocatorTest.java b/ksqldb-streams/src/test/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocatorTest.java
index 355e2eda9c45..625cf3a33168 100644
--- a/ksqldb-streams/src/test/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocatorTest.java
+++ b/ksqldb-streams/src/test/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocatorTest.java
@@ -61,6 +61,7 @@
 import org.apache.kafka.streams.processor.internals.namedtopology.KafkaStreamsNamedTopologyWrapper;
 import org.apache.kafka.streams.state.HostInfo;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
@@ -461,7 +462,10 @@ public void shouldReturnOneStandByWhenActiveAndOtherStandByDown() {
     assertThat(nodeList.stream().findFirst().get(), is(standByNode2));
   }
 
+  @Ignore
   @Test
+  //For issue #7174. Temporarily ignore this test. It will call getMetadataForAllPartitions().
+  //Formerly it called getMetadataForKeys().
   public void shouldGroupKeysByLocation() {
     // Given:
     getActiveStandbyMetadata(SOME_KEY, 0, ACTIVE_HOST_INFO, STANDBY_HOST_INFO1);