diff --git a/src/java/org/apache/cassandra/db/Clustering.java b/src/java/org/apache/cassandra/db/Clustering.java
index 577f3d53bdb..d062837a3e2 100644
--- a/src/java/org/apache/cassandra/db/Clustering.java
+++ b/src/java/org/apache/cassandra/db/Clustering.java
@@ -49,7 +49,7 @@ public default Clustering<?> clone(ByteBufferCloner cloner)
         ByteBuffer[] newValues = new ByteBuffer[size()];
         for (int i = 0; i < size(); i++)
         {
-            ByteBuffer val = accessor().toBuffer(get(i));
+            ByteBuffer val = bufferAt(i);
             newValues[i] = val == null ? null : cloner.clone(val);
         }
         return new BufferClustering(newValues);
@@ -84,7 +84,8 @@ public default String toCQLString(TableMetadata metadata)
         for (int i = 0; i < size(); i++)
         {
             ColumnMetadata c = metadata.clusteringColumns().get(i);
-            sb.append(i == 0 ? "" : ", ").append(c.type.getString(get(i), accessor()));
+            ByteBuffer value = bufferAt(i);
+            sb.append(i == 0 ? "" : ", ").append(c.type.toCQLString(value));
         }
         return sb.toString();
     }
diff --git a/src/java/org/apache/cassandra/db/ClusteringPrefix.java b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
index aa899d934ee..0d8d1abb745 100644
--- a/src/java/org/apache/cassandra/db/ClusteringPrefix.java
+++ b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
@@ -358,11 +358,11 @@ default int dataSize()
     default ByteBuffer serializeAsPartitionKey()
     {
         if (size() == 1)
-            return accessor().toBuffer(get(0));
+            return bufferAt(0);
 
         ByteBuffer[] values = new ByteBuffer[size()];
         for (int i = 0; i < size(); i++)
-            values[i] = accessor().toBuffer(get(i));
+            values[i] = bufferAt(i);
         return CompositeType.build(ByteBufferAccessor.instance, values);
     }
 
@@ -752,4 +752,4 @@ public static boolean equals(ClusteringPrefix<?> prefix, Object o)
         return equals(prefix, (ClusteringPrefix<?>) o);
     }
 
-}
\ No newline at end of file
+}
diff --git a/src/java/org/apache/cassandra/db/DataRange.java b/src/java/org/apache/cassandra/db/DataRange.java
index a394a92e5a9..71c1cef5492 100644
--- a/src/java/org/apache/cassandra/db/DataRange.java
+++ b/src/java/org/apache/cassandra/db/DataRange.java
@@ -19,7 +19,6 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
-import org.apache.cassandra.cql3.CqlBuilder;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.db.filter.*;
@@ -277,10 +276,6 @@ public String toCQLString(TableMetadata metadata)
         if (isUnrestricted())
             return "";
 
-        CqlBuilder builder = new CqlBuilder();
-
-        boolean needAnd = false;
-
         if (isSinglePartition())
         {
             /*
@@ -288,43 +283,40 @@ public String toCQLString(TableMetadata metadata)
              * key are the same. If that is the case, we want to print the query as an equality on the partition key
              * rather than a token range, as if it was a partition query, for better readability.
              */
-            builder.append(((DecoratedKey) startKey()).toCQLString(metadata));
-            needAnd = true;
+            return ((DecoratedKey) startKey()).toCQLString(metadata);
         }
         else
         {
+            StringBuilder builder = new StringBuilder();
             if (!startKey().isMinimum())
             {
-                appendClause(startKey(), builder, metadata, true, keyRange.isStartInclusive());
-                needAnd = true;
+                appendCQLClause(startKey(), builder, metadata, true, keyRange.isStartInclusive());
             }
             if (!stopKey().isMinimum())
             {
-                if (needAnd)
+                if (builder.length() > 0)
                     builder.append(" AND ");
-                appendClause(stopKey(), builder, metadata, false, keyRange.isEndInclusive());
-                needAnd = true;
+                appendCQLClause(stopKey(), builder, metadata, false, keyRange.isEndInclusive());
             }
+            return builder.toString();
         }
-
-        String filterString = clusteringIndexFilter.toCQLString(metadata);
-        if (!filterString.isEmpty())
-            builder.append(needAnd ? " AND " : "").append(filterString);
-
-        return builder.toString();
     }
 
-    private void appendClause(PartitionPosition pos, CqlBuilder builder, TableMetadata metadata, boolean isStart, boolean isInclusive)
+    private void appendCQLClause(PartitionPosition pos,
+                                 StringBuilder builder,
+                                 TableMetadata metadata,
+                                 boolean isStart,
+                                 boolean isInclusive)
     {
         builder.append("token(");
         builder.append(ColumnMetadata.toCQLString(metadata.partitionKeyColumns()));
         builder.append(") ");
         if (pos instanceof DecoratedKey)
         {
-            builder.append(getOperator(isStart, isInclusive)).append(" ");
+            builder.append(getOperator(isStart, isInclusive)).append(' ');
             builder.append("token(");
             appendKeyString(builder, metadata.partitionKeyType, ((DecoratedKey)pos).getKey());
-            builder.append(")");
+            builder.append(')');
         }
         else
         {
@@ -343,18 +335,18 @@ private static String getOperator(boolean isStart, boolean isInclusive)
 
     // TODO: this is reused in SinglePartitionReadCommand but this should not really be here. Ideally
     // we need a more "native" handling of composite partition keys.
-    public static void appendKeyString(CqlBuilder builder, AbstractType<?> type, ByteBuffer key)
+    public static void appendKeyString(StringBuilder builder, AbstractType<?> type, ByteBuffer key)
     {
         if (type instanceof CompositeType)
         {
             CompositeType ct = (CompositeType)type;
             ByteBuffer[] values = ct.split(key);
             for (int i = 0; i < ct.subTypes().size(); i++)
-                builder.append(i == 0 ? "" : ", ").append(ct.subTypes().get(i).getString(values[i]));
+                builder.append(i == 0 ? "" : ", ").append(ct.subTypes().get(i).toCQLString(values[i]));
         }
         else
         {
-            builder.append(type.getString(key));
+            builder.append(type.toCQLString(key));
         }
     }
 
diff --git a/src/java/org/apache/cassandra/db/DecoratedKey.java b/src/java/org/apache/cassandra/db/DecoratedKey.java
index e71257fdad3..a47cf0165c6 100644
--- a/src/java/org/apache/cassandra/db/DecoratedKey.java
+++ b/src/java/org/apache/cassandra/db/DecoratedKey.java
@@ -170,6 +170,8 @@ public String toString()
      * Generate CQL representation of this partition key for the given table.
      * For single-column keys: "k = 0"
      * For multi-column keys: "k1 = 1 AND k2 = 2"
+     *
+     * @param metadata the table metadata
      */
     public String toCQLString(TableMetadata metadata)
     {
@@ -189,7 +191,7 @@ public String toCQLString(TableMetadata metadata)
 
     private static String toCQLString(ColumnMetadata metadata, ByteBuffer key)
     {
-        return String.format("%s = %s", metadata.name.toCQLString(), metadata.type.getString(key));
+        return String.format("%s = %s", metadata.name.toCQLString(), metadata.type.toCQLString(key));
     }
 
     public Token getToken()
diff --git a/src/java/org/apache/cassandra/db/MultiPartitionReadQuery.java b/src/java/org/apache/cassandra/db/MultiPartitionReadQuery.java
new file mode 100644
index 00000000000..ff4d33c03bd
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/MultiPartitionReadQuery.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 org.apache.cassandra.db;
+
+import java.util.List;
+
+import org.apache.cassandra.cql3.CqlBuilder;
+import org.apache.cassandra.schema.TableMetadata;
+
+/**
+ * A {@code ReadQuery} for multiple partitions, restricted by one of more data ranges.
+ */
+public interface MultiPartitionReadQuery extends ReadQuery
+{
+    List<DataRange> ranges();
+
+    default void appendCQLWhereClause(CqlBuilder builder)
+    {
+        // Append the data ranges.
+        TableMetadata metadata = metadata();
+        boolean hasRanges = appendRanges(builder);
+
+        // Append the clustering index filter and the row filter.
+        String filter = ranges().get(0).clusteringIndexFilter.toCQLString(metadata, rowFilter());
+        if (!filter.isEmpty())
+        {
+            if (filter.startsWith("ORDER BY"))
+                builder.append(" ");
+            else if (hasRanges)
+                builder.append(" AND ");
+            else
+                builder.append(" WHERE ");
+            builder.append(filter);
+        }
+    }
+
+    private boolean appendRanges(CqlBuilder builder)
+    {
+        List<DataRange> ranges = ranges();
+        if (ranges.size() == 1)
+        {
+            DataRange range = ranges.get(0);
+            if (range.isUnrestricted())
+                return false;
+
+            String rangeString = range.toCQLString(metadata());
+            if (!rangeString.isEmpty())
+            {
+                builder.append(" WHERE ").append(rangeString);
+                return true;
+            }
+        }
+        else
+        {
+            builder.append(" WHERE ").append('(');
+            for (int i = 0; i < ranges.size(); i++)
+            {
+                if (i > 0)
+                    builder.append(" OR ");
+                builder.append(ranges.get(i).toCQLString(metadata()));
+            }
+            builder.append(')');
+            return true;
+        }
+        return false;
+    }
+}
diff --git a/src/java/org/apache/cassandra/db/MultiRangeReadCommand.java b/src/java/org/apache/cassandra/db/MultiRangeReadCommand.java
index 864cbd7bea1..0ef61e12cb3 100644
--- a/src/java/org/apache/cassandra/db/MultiRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/MultiRangeReadCommand.java
@@ -56,7 +56,7 @@
  * Note: digest is not supported because each replica is responsible for different token ranges, there is no point on
  * sending digest.
  */
-public class MultiRangeReadCommand extends ReadCommand
+public class MultiRangeReadCommand extends ReadCommand implements MultiPartitionReadQuery
 {
     protected static final SelectionDeserializer selectionDeserializer = new Deserializer();
 
@@ -142,6 +142,7 @@ public boolean isSinglePartition()
     /**
      * @return all token ranges to be queried
      */
+    @Override
     public List<DataRange> ranges()
     {
         return dataRanges;
@@ -320,31 +321,9 @@ public Verb verb()
     }
 
     @Override
-    protected void appendCQLWhereClause(CqlBuilder builder)
+    public void appendCQLWhereClause(CqlBuilder builder)
     {
-        if (ranges().size() == 1 && ranges().get(0).isUnrestricted() && rowFilter().isEmpty())
-            return;
-
-        builder.append(" WHERE ");
-        // We put the row filter first because the data range can end by "ORDER BY"
-        if (!rowFilter().isEmpty())
-        {
-            builder.append(rowFilter());
-            builder.append(" AND ");
-        }
-
-        boolean isFirst = true;
-        for (int i = 0; i < ranges().size(); i++)
-        {
-            DataRange dataRange = ranges().get(i);
-            if (!dataRange.isUnrestricted())
-            {
-                if (!isFirst)
-                    builder.append(" AND ");
-                isFirst = false;
-                builder.append(dataRange.toCQLString(metadata()));
-            }
-        }
+        MultiPartitionReadQuery.super.appendCQLWhereClause(builder);
     }
 
     @Override
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index 071f90bd452..4dbb43cb785 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -394,23 +394,9 @@ public Verb verb()
     }
 
     @Override
-    protected void appendCQLWhereClause(CqlBuilder builder)
+    public void appendCQLWhereClause(CqlBuilder builder)
     {
-        if (dataRange.isUnrestricted() && rowFilter().isEmpty())
-            return;
-
-        builder.append(" WHERE ");
-        // We put the row filter first because the data range can end by "ORDER BY"
-        if (!rowFilter().isEmpty())
-        {
-            builder.append(rowFilter());
-            if (!dataRange.isUnrestricted())
-                builder.append(" AND ");
-        }
-        if (!dataRange.isUnrestricted())
-        {
-            builder.append(dataRange.toCQLString(metadata()));
-        }
+        PartitionRangeReadQuery.super.appendCQLWhereClause(builder);
     }
 
     @Override
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadQuery.java b/src/java/org/apache/cassandra/db/PartitionRangeReadQuery.java
index 12624e70b2e..d5658782699 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadQuery.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadQuery.java
@@ -17,6 +17,8 @@
  */
 package org.apache.cassandra.db;
 
+import java.util.List;
+
 import org.apache.cassandra.db.filter.ColumnFilter;
 import org.apache.cassandra.db.filter.DataLimits;
 import org.apache.cassandra.db.filter.RowFilter;
@@ -29,7 +31,7 @@
 /**
  *  A {@code ReadQuery} for a range of partitions.
  */
-public interface PartitionRangeReadQuery extends ReadQuery
+public interface PartitionRangeReadQuery extends MultiPartitionReadQuery
 {
     static ReadQuery create(TableMetadata table,
                             int nowInSec,
@@ -46,6 +48,12 @@ static ReadQuery create(TableMetadata table,
 
     DataRange dataRange();
 
+    @Override
+    default List<DataRange> ranges()
+    {
+        return List.of(dataRange());
+    }
+
     /**
      * Creates a new {@code PartitionRangeReadQuery} with the updated limits.
      *
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index 8f8f4feb9e0..1d42f0ae505 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -37,7 +37,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.cql3.CqlBuilder;
 import org.apache.cassandra.db.filter.ClusteringIndexFilter;
 import org.apache.cassandra.db.filter.ColumnFilter;
 import org.apache.cassandra.db.filter.DataLimits;
@@ -762,8 +761,6 @@ public Message<ReadCommand> createMessage(boolean trackRepairedData)
 
     public abstract Verb verb();
 
-    protected abstract void appendCQLWhereClause(CqlBuilder builder);
-
     // Skip purgeable tombstones. We do this because it's safe to do (post-merge of the memtable and sstable at least), it
     // can save us some bandwith, and avoid making us throw a TombstoneOverwhelmingException for purgeable tombstones (which
     // are to some extend an artefact of compaction lagging behind and hence counting them is somewhat unintuitive).
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index 557f9b48454..ddcb912dc63 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@ -41,7 +41,6 @@
 import org.apache.cassandra.db.transform.Transformation;
 import org.apache.cassandra.exceptions.RequestExecutionException;
 import org.apache.cassandra.index.Index;
-import org.apache.cassandra.index.sai.utils.RowWithSourceTable;
 import org.apache.cassandra.io.sstable.format.RowIndexEntry;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.SSTableReadsListener;
@@ -1153,19 +1152,9 @@ public Verb verb()
     }
 
     @Override
-    protected void appendCQLWhereClause(CqlBuilder builder)
+    public void appendCQLWhereClause(CqlBuilder builder)
     {
-        builder.append(" WHERE ");
-
-        builder.append(partitionKey().toCQLString(metadata()));
-
-        // We put the row filter first because the clustering index filter can end by "ORDER BY"
-        if (!rowFilter().isEmpty())
-            builder.append(" AND ").append(rowFilter());
-
-        String filterString = clusteringIndexFilter().toCQLString(metadata());
-        if (!filterString.isEmpty())
-            builder.append(" AND ").append(filterString);
+        SinglePartitionReadQuery.super.appendCQLWhereClause(builder);
     }
 
     protected void serializeSelection(DataOutputPlus out, int version) throws IOException
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadQuery.java b/src/java/org/apache/cassandra/db/SinglePartitionReadQuery.java
index 755d55227e6..1f55368b4db 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadQuery.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadQuery.java
@@ -26,6 +26,7 @@
 
 import org.apache.commons.lang3.tuple.Pair;
 
+import org.apache.cassandra.cql3.CqlBuilder;
 import org.apache.cassandra.db.filter.ClusteringIndexFilter;
 import org.apache.cassandra.db.filter.ColumnFilter;
 import org.apache.cassandra.db.filter.DataLimits;
@@ -158,6 +159,23 @@ default boolean selectsClustering(DecoratedKey key, Clustering<?> clustering)
         return rowFilter().clusteringKeyRestrictionsAreSatisfiedBy(clustering);
     }
 
+    default void appendCQLWhereClause(CqlBuilder builder)
+    {
+        builder.append(" WHERE ");
+
+        // Append the partition key restrictions.
+        TableMetadata metadata = metadata();
+        builder.append(partitionKey().toCQLString(metadata));
+
+        // Append the clustering index filter and the row filter.
+        String filter = clusteringIndexFilter().toCQLString(metadata(), rowFilter());
+        if (!filter.isEmpty())
+        {
+            builder.append(filter.startsWith("ORDER BY") ? " " : " AND ")
+                   .append(filter);
+        }
+    }
+
     /**
      * Groups multiple single partition read queries.
      */
diff --git a/src/java/org/apache/cassandra/db/Slices.java b/src/java/org/apache/cassandra/db/Slices.java
index 04e4eb13fd4..e23a806e3bb 100644
--- a/src/java/org/apache/cassandra/db/Slices.java
+++ b/src/java/org/apache/cassandra/db/Slices.java
@@ -24,6 +24,8 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Iterators;
 
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.db.filter.RowFilter;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -146,7 +148,7 @@ public ClusteringBound<?> end()
      */
     public abstract boolean intersects(Slice slice);
 
-    public abstract String toCQLString(TableMetadata metadata);
+    public abstract String toCQLString(TableMetadata metadata, RowFilter rowFilter);
 
     /**
      * Checks if this <code>Slices</code> is empty.
@@ -555,7 +557,8 @@ public String toString()
             return sb.append("}").toString();
         }
 
-        public String toCQLString(TableMetadata metadata)
+        @Override
+        public String toCQLString(TableMetadata metadata, RowFilter rowFilter)
         {
             StringBuilder sb = new StringBuilder();
 
@@ -599,7 +602,7 @@ public String toCQLString(TableMetadata metadata)
                         sb.append(" AND ");
                     needAnd = true;
 
-                    sb.append(column.name);
+                    sb.append(column.name.toCQLString());
 
                     Set<ByteBuffer> values = new LinkedHashSet<>();
                     for (int j = 0; j < componentInfo.size(); j++)
@@ -607,20 +610,25 @@ public String toCQLString(TableMetadata metadata)
 
                     if (values.size() == 1)
                     {
-                        sb.append(" = ").append(column.type.getString(first.startValue));
+                        sb.append(" = ").append(column.type.toCQLString(first.startValue));
+                        rowFilter = rowFilter.withoutFirstLevelExpression(column, Operator.EQ, first.startValue);
                     }
                     else
                     {
                         sb.append(" IN (");
                         int j = 0;
                         for (ByteBuffer value : values)
-                            sb.append(j++ == 0 ? "" : ", ").append(column.type.getString(value));
-                        sb.append(")");
+                        {
+                            sb.append(j++ == 0 ? "" : ", ").append(column.type.toCQLString(value));
+                            rowFilter = rowFilter.withoutFirstLevelExpression(column, Operator.EQ, value);
+                        }
+                        sb.append(')');
                     }
                 }
                 else
                 {
                     boolean isReversed = column.isReversedType();
+                    Operator operator;
 
                     // As said above, we assume (without checking) that this means all ComponentOfSlice for this column
                     // are the same, so we only bother about the first.
@@ -629,27 +637,41 @@ public String toCQLString(TableMetadata metadata)
                         if (needAnd)
                             sb.append(" AND ");
                         needAnd = true;
-                        sb.append(column.name);
+                        sb.append(column.name.toCQLString());
                         if (isReversed)
-                            sb.append(first.startInclusive ? " <= " : " < ");
+                            operator = first.startInclusive ? Operator.LTE : Operator.LT;
                         else
-                            sb.append(first.startInclusive ? " >= " : " > ");
-                        sb.append(column.type.getString(first.startValue));
+                            operator = first.startInclusive ? Operator.GTE : Operator.GT;
+                        sb.append(' ').append(operator).append(' ')
+                          .append(column.type.toCQLString(first.startValue));
+                        rowFilter = rowFilter.withoutFirstLevelExpression(column, operator, first.startValue);
                     }
                     if (first.endValue != null)
                     {
                         if (needAnd)
                             sb.append(" AND ");
                         needAnd = true;
-                        sb.append(column.name);
+                        sb.append(column.name.toCQLString());
                         if (isReversed)
-                            sb.append(first.endInclusive ? " >= " : " > ");
+                            operator = first.endInclusive ? Operator.GTE : Operator.GT;
                         else
-                            sb.append(first.endInclusive ? " <= " : " < ");
-                        sb.append(column.type.getString(first.endValue));
+                            operator = first.endInclusive ? Operator.LTE : Operator.LT;
+                        sb.append(' ').append(operator).append(' ')
+                          .append(column.type.toCQLString(first.endValue));
+                        rowFilter = rowFilter.withoutFirstLevelExpression(column, operator, first.endValue);
                     }
                 }
             }
+            rowFilter = rowFilter.withoutFirstLevelExpression(metadata.clusteringColumns().get(0), Operator.IN);
+
+            // Append the row filter.
+            if (!rowFilter.isEmpty())
+            {
+                String filter = rowFilter.toCQLString();
+                sb.append(filter.startsWith("ORDER BY") ? " " : " AND ");
+                sb.append(filter);
+            }
+
             return sb.toString();
         }
 
@@ -771,9 +793,10 @@ public String toString()
             return "ALL";
         }
 
-        public String toCQLString(TableMetadata metadata)
+        @Override
+        public String toCQLString(TableMetadata metadata, RowFilter rowFilter)
         {
-            return "";
+            return rowFilter.toCQLString();
         }
     }
 
@@ -852,7 +875,8 @@ public String toString()
             return "NONE";
         }
 
-        public String toCQLString(TableMetadata metadata)
+        @Override
+        public String toCQLString(TableMetadata metadata, RowFilter rowFilter)
         {
             return "";
         }
diff --git a/src/java/org/apache/cassandra/db/VirtualTablePartitionRangeReadQuery.java b/src/java/org/apache/cassandra/db/VirtualTablePartitionRangeReadQuery.java
index fe864ffc4fd..9e73bfb0907 100644
--- a/src/java/org/apache/cassandra/db/VirtualTablePartitionRangeReadQuery.java
+++ b/src/java/org/apache/cassandra/db/VirtualTablePartitionRangeReadQuery.java
@@ -95,20 +95,8 @@ protected UnfilteredPartitionIterator queryVirtualTable()
     }
 
     @Override
-    protected void appendCQLWhereClause(CqlBuilder builder)
+    public void appendCQLWhereClause(CqlBuilder builder)
     {
-        if (dataRange.isUnrestricted() && rowFilter().isEmpty())
-            return;
-
-        builder.append(" WHERE ");
-        // We put the row filter first because the data range can end by "ORDER BY"
-        if (!rowFilter().isEmpty())
-        {
-            builder.append(rowFilter());
-            if (!dataRange.isUnrestricted())
-                builder.append(" AND ");
-        }
-        if (!dataRange.isUnrestricted())
-            builder.append(dataRange.toCQLString(metadata()));
+        PartitionRangeReadQuery.super.appendCQLWhereClause(builder);
     }
 }
diff --git a/src/java/org/apache/cassandra/db/VirtualTableSinglePartitionReadQuery.java b/src/java/org/apache/cassandra/db/VirtualTableSinglePartitionReadQuery.java
index bc51bb4bc73..acea03e1032 100644
--- a/src/java/org/apache/cassandra/db/VirtualTableSinglePartitionReadQuery.java
+++ b/src/java/org/apache/cassandra/db/VirtualTableSinglePartitionReadQuery.java
@@ -76,20 +76,9 @@ private VirtualTableSinglePartitionReadQuery(TableMetadata metadata,
     }
 
     @Override
-    protected void appendCQLWhereClause(CqlBuilder builder)
+    public void appendCQLWhereClause(CqlBuilder builder)
     {
-        builder.append(" WHERE ");
-
-        builder.append(ColumnMetadata.toCQLString(metadata().partitionKeyColumns())).append(" = ");
-        DataRange.appendKeyString(builder, metadata().partitionKeyType, partitionKey().getKey());
-
-        // We put the row filter first because the clustering index filter can end by "ORDER BY"
-        if (!rowFilter().isEmpty())
-            builder.append(" AND ").append(rowFilter());
-
-        String filterString = clusteringIndexFilter().toCQLString(metadata());
-        if (!filterString.isEmpty())
-            builder.append(" AND ").append(filterString);
+        SinglePartitionReadQuery.super.appendCQLWhereClause(builder);
     }
 
     @Override
diff --git a/src/java/org/apache/cassandra/db/filter/AbstractClusteringIndexFilter.java b/src/java/org/apache/cassandra/db/filter/AbstractClusteringIndexFilter.java
index 63c2783e0af..71f0afd95ca 100644
--- a/src/java/org/apache/cassandra/db/filter/AbstractClusteringIndexFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/AbstractClusteringIndexFilter.java
@@ -47,21 +47,26 @@ public boolean isEmpty(ClusteringComparator comparator)
         return false;
     }
 
-    protected abstract void serializeInternal(DataOutputPlus out, int version) throws IOException;
-    protected abstract long serializedSizeInternal(int version);
-
     protected void appendOrderByToCQLString(TableMetadata metadata, StringBuilder sb)
     {
         if (reversed)
         {
-            sb.append(" ORDER BY (");
+            if (sb.length() > 0)
+                sb.append(' ');
+            sb.append("ORDER BY ");
             int i = 0;
             for (ColumnMetadata column : metadata.clusteringColumns())
-                sb.append(i++ == 0 ? "" : ", ").append(column.name).append(column.type instanceof ReversedType ? " ASC" : " DESC");
-            sb.append(')');
+            {
+                sb.append(i++ == 0 ? "" : ", ")
+                  .append(column.name.toCQLString())
+                  .append(column.type instanceof ReversedType ? " ASC" : " DESC");
+            }
         }
     }
 
+    protected abstract void serializeInternal(DataOutputPlus out, int version) throws IOException;
+    protected abstract long serializedSizeInternal(int version);
+
     private static class FilterSerializer implements Serializer
     {
         public void serialize(ClusteringIndexFilter pfilter, DataOutputPlus out, int version) throws IOException
diff --git a/src/java/org/apache/cassandra/db/filter/ClusteringIndexFilter.java b/src/java/org/apache/cassandra/db/filter/ClusteringIndexFilter.java
index 8d3b350988f..edc3cf07aa5 100644
--- a/src/java/org/apache/cassandra/db/filter/ClusteringIndexFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ClusteringIndexFilter.java
@@ -153,7 +153,7 @@ static interface InternalDeserializer
     public Kind kind();
 
     public String toString(TableMetadata metadata);
-    public String toCQLString(TableMetadata metadata);
+    String toCQLString(TableMetadata metadata, RowFilter rowFilter);
 
     public interface Serializer
     {
diff --git a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
index 45ee6d22647..207a54254c7 100644
--- a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.util.*;
 
+import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.db.rows.*;
@@ -161,34 +162,41 @@ public String toString(TableMetadata metadata)
         return sb.append(')').toString();
     }
 
-    public String toCQLString(TableMetadata metadata)
+    @Override
+    public String toCQLString(TableMetadata metadata, RowFilter rowFilter)
     {
         if (metadata.clusteringColumns().isEmpty() || clusterings.isEmpty())
-            return "";
+            return rowFilter.toCQLString();
 
-        StringBuilder sb = new StringBuilder();
+        boolean isSingleColumn = metadata.clusteringColumns().size() == 1;
+        boolean isSingleClustering = clusterings.size() == 1;
 
-        boolean multipleColumns = metadata.clusteringColumns().size() > 1;
-        boolean multipleClusterings = clusterings.size() > 1;
+        StringBuilder sb = new StringBuilder();
+        sb.append(isSingleColumn ? "" : '(')
+          .append(ColumnMetadata.toCQLString(metadata.clusteringColumns()))
+          .append(isSingleColumn ? "" : ')');
 
-        if (multipleColumns)
-            sb.append('(');
-        sb.append(ColumnMetadata.toCQLString(metadata.clusteringColumns()));
-        if (multipleColumns)
-            sb.append(')');
-        sb.append(multipleClusterings ? " IN (" : " = ");
+        sb.append(isSingleClustering ? " = " : " IN (");
         int i = 0;
         for (Clustering<?> clustering : clusterings)
         {
-            sb.append(i++ == 0 ? "" : ", ");
-            if (multipleColumns)
-                sb.append('(');
-            sb.append(clustering.toCQLString(metadata));
-            if (multipleColumns)
-                sb.append(')');
+            sb.append(i++ == 0 ? "" : ", ")
+              .append(isSingleColumn ? "" : '(')
+              .append(clustering.toCQLString(metadata))
+              .append(isSingleColumn ? "" : ')');
+
+            for (int j = 0; j < clustering.size(); j++)
+                rowFilter = rowFilter.withoutFirstLevelExpression(metadata.clusteringColumns().get(j), Operator.EQ, clustering.bufferAt(j));
+        }
+        sb.append(isSingleClustering ? "" : ")");
+        rowFilter = rowFilter.withoutFirstLevelExpression(metadata.clusteringColumns().get(0), Operator.IN);
+
+        if (!rowFilter.isEmpty())
+        {
+            String filter = rowFilter.toCQLString();
+            sb.append(filter.startsWith("ORDER BY") ? " " : " AND ");
+            sb.append(filter);
         }
-        if (multipleClusterings)
-            sb.append(')');
 
         appendOrderByToCQLString(metadata, sb);
         return sb.toString();
diff --git a/src/java/org/apache/cassandra/db/filter/ClusteringIndexSliceFilter.java b/src/java/org/apache/cassandra/db/filter/ClusteringIndexSliceFilter.java
index 5b72a6f91bb..85820771a6d 100644
--- a/src/java/org/apache/cassandra/db/filter/ClusteringIndexSliceFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ClusteringIndexSliceFilter.java
@@ -133,15 +133,12 @@ public String toString(TableMetadata metadata)
         return String.format("slice(slices=%s, reversed=%b)", slices, reversed);
     }
 
-    public String toCQLString(TableMetadata metadata)
+    @Override
+    public String toCQLString(TableMetadata metadata, RowFilter rowFilter)
     {
         StringBuilder sb = new StringBuilder();
-
-        if (!selectsAllPartition())
-            sb.append(slices.toCQLString(metadata));
-
+        sb.append(slices.toCQLString(metadata, rowFilter));
         appendOrderByToCQLString(metadata, sb);
-
         return sb.toString();
     }
 
diff --git a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
index d95a27dd1c7..dbb11ad2625 100644
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@ -952,7 +952,7 @@ private String toString(Iterator<ColumnMetadata> columns, boolean cql)
                 if (s.isEmpty())
                     joiner.add(columnName);
                 else
-                    s.forEach(subSel -> joiner.add(String.format("%s%s", columnName, subSel)));
+                    s.forEach(subSel -> joiner.add(String.format("%s%s", columnName, subSel.toString(cql))));
             }
             return joiner.toString();
         }
diff --git a/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java b/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java
index 0358bd6f5f8..1438663a917 100644
--- a/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java
@@ -88,6 +88,14 @@ public int compareTo(ColumnSubselection other)
      */
     public abstract int compareInclusionOf(CellPath path);
 
+    @Override
+    public String toString()
+    {
+        return toString(false);
+    }
+
+    protected abstract String toString(boolean cql);
+
     private static class Slice extends ColumnSubselection
     {
         private final CellPath from;
@@ -122,11 +130,13 @@ else if (cmp.compare(to, path) < 0)
         }
 
         @Override
-        public String toString()
+        protected String toString(boolean cql)
         {
-            // This assert we're dealing with a collection since that's the only thing it's used for so far.
+            // This asserts we're dealing with a collection since that's the only thing it's used for so far.
             AbstractType<?> type = ((CollectionType<?>)column().type).nameComparator();
-            return String.format("[%s:%s]", from == CellPath.BOTTOM ? "" : type.getString(from.get(0)), to == CellPath.TOP ? "" : type.getString(to.get(0)));
+            return String.format("[%s:%s]",
+                                 from == CellPath.BOTTOM ? "" : (cql ? type.toCQLString(from.get(0)) : type.getString(from.get(0))),
+                                 to == CellPath.TOP ? "" : (cql ? type.toCQLString(to.get(0)) : type.getString(to.get(0))));
         }
     }
 
@@ -156,11 +166,11 @@ public int compareInclusionOf(CellPath path)
         }
 
         @Override
-        public String toString()
+        protected String toString(boolean cql)
         {
             // This assert we're dealing with a collection since that's the only thing it's used for so far.
             AbstractType<?> type = ((CollectionType<?>)column().type).nameComparator();
-            return String.format("[%s]", type.getString(element.get(0)));
+            return String.format("[%s]", cql ? type.toCQLString(element.get(0)) : type.getString(element.get(0)));
         }
     }
 
diff --git a/src/java/org/apache/cassandra/db/filter/DataLimits.java b/src/java/org/apache/cassandra/db/filter/DataLimits.java
index 869a2983f44..8785e8e3dae 100644
--- a/src/java/org/apache/cassandra/db/filter/DataLimits.java
+++ b/src/java/org/apache/cassandra/db/filter/DataLimits.java
@@ -709,7 +709,7 @@ public String toString()
             if (bytesLimit != NO_LIMIT)
                 limits.add("BYTES LIMIT " + bytesLimit);
             if (rowLimit != NO_LIMIT)
-                limits.add("ROWS LIMIT " + rowLimit);
+                limits.add("LIMIT " + rowLimit);
             if (perPartitionLimit != NO_LIMIT)
                 limits.add("PER PARTITION LIMIT " + perPartitionLimit);
 
diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java
index 62558c5b7c6..d939ace2279 100644
--- a/src/java/org/apache/cassandra/db/filter/RowFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java
@@ -285,8 +285,8 @@ public boolean clusteringKeyRestrictionsAreSatisfiedBy(Clustering<?> clustering)
     }
 
     /**
-     * Returns this filter but without the provided expression. This method
-     * *assumes* that the filter contains the provided expression.
+     * Returns this filter but without the provided expression. This method *assumes* that the filter contains the
+     * provided expression, and it looks in all the levels of the filter tree.
      */
     public RowFilter without(Expression expression)
     {
@@ -297,6 +297,24 @@ public RowFilter without(Expression expression)
         return new RowFilter(root.filter(e -> !e.equals(expression)), indexHints);
     }
 
+    /**
+     * Returns a copy of this filter but without first level expressions with the provided column, operator and value.
+     * If this filter doesn't contain the specified expressions this method will just return an identical copy of this filter.
+     */
+    public RowFilter withoutFirstLevelExpression(ColumnMetadata column, Operator op, ByteBuffer value)
+    {
+        return restrictFirstLevel(e -> !(e.column.equals(column) && e.operator == op && e.value.equals(value)));
+    }
+
+    /**
+     * Returns a copy of this filter but without first level expressions with the provided column and operator.
+     * If this filter doesn't contain the specified expressions this method will just return an identical copy of this filter.
+     */
+    public RowFilter withoutFirstLevelExpression(ColumnMetadata column, Operator op)
+    {
+        return restrictFirstLevel(e -> !(e.column.equals(column) && e.operator == op));
+    }
+
     public RowFilter withoutExpressions()
     {
         return NONE;
@@ -315,6 +333,11 @@ public RowFilter restrict(Predicate<Expression> filter)
         return new RowFilter(root.filter(filter), indexHints);
     }
 
+    public RowFilter restrictFirstLevel(Predicate<Expression> filter)
+    {
+        return new RowFilter(root.filterFirstLevel(filter), indexHints);
+    }
+
     public boolean isEmpty()
     {
         return root.isEmpty();
@@ -326,6 +349,11 @@ public String toString()
         return root.toString();
     }
 
+    public String toCQLString()
+    {
+        return root.toCQLString();
+    }
+
     public static Builder builder()
     {
         return new Builder(null, IndexHints.NONE);
@@ -409,7 +437,7 @@ private FilterElement doBuild(StatementRestrictions restrictions,
          * <p>
          *
          * This wrapper method makes sure we pass a {@code RowFilter.Builder} that is always in conjunction mode to the
-         * respective {@code addToRowFilterDelegate} method. If multiple expressions are added to the row filter, this 
+         * respective {@code addToRowFilterDelegate} method. If multiple expressions are added to the row filter, this
          * method makes sure they are joined with AND in their own {@link FilterElement}.
          *
          * @param addToRowFilterDelegate a function that adds expressions / child filter elements
@@ -625,6 +653,14 @@ public FilterElement filter(Predicate<Expression> filter)
             return builder.build();
         }
 
+        public FilterElement filterFirstLevel(Predicate<Expression> filter)
+        {
+            FilterElement.Builder builder = new Builder(isDisjunction);
+            expressions.stream().filter(filter).forEach(builder.expressions::add);
+            builder.children.addAll(children);
+            return builder.build();
+        }
+
         public List<FilterElement> children()
         {
             return children;
@@ -716,22 +752,37 @@ private int numFilteredValues()
 
         @Override
         public String toString()
+        {
+            return toCQLString();
+        }
+
+        public String toCQLString()
         {
             StringBuilder sb = new StringBuilder();
-            for (int i = 0; i < expressions.size(); i++)
+            for (Expression expression : expressions)
             {
+                if (expression.isOrderingExpression())
+                    continue;
                 if (sb.length() > 0)
                     sb.append(isDisjunction ? " OR " : " AND ");
-                sb.append(expressions.get(i));
+                sb.append(expression.toCQLString());
             }
-            for (int i = 0; i < children.size(); i++)
+            for (FilterElement child : children)
             {
                 if (sb.length() > 0)
                     sb.append(isDisjunction ? " OR " : " AND ");
                 sb.append('(');
-                sb.append(children.get(i));
+                sb.append(child.toCQLString());
                 sb.append(')');
             }
+            for (Expression expression : expressions)
+            {
+                if (!expression.isOrderingExpression())
+                    continue;
+                if (sb.length() > 0)
+                    sb.append(' ');
+                sb.append(expression.toCQLString());
+            }
             return sb.toString();
         }
 
@@ -877,6 +928,11 @@ public Index.Analyzer analyzer()
             return null;
         }
 
+        public boolean isOrderingExpression()
+        {
+            return operator == Operator.ANN || operator == Operator.BM25 || operator == Operator.ORDER_BY_ASC || operator == Operator.ORDER_BY_DESC;
+        }
+
         protected boolean isSatisfiedBy(AbstractType<?> type, ByteBuffer foundValue)
         {
             if (foundValue == null)
@@ -1008,6 +1064,17 @@ public int hashCode()
             return Objects.hashCode(column.name, operator, value);
         }
 
+        @Override
+        public String toString()
+        {
+            return toCQLString();
+        }
+
+        public String toCQLString()
+        {
+            return "";
+        }
+
         public static class Serializer
         {
             public void serialize(Expression expression, DataOutputPlus out, int version) throws IOException
@@ -1313,7 +1380,7 @@ private boolean containsKey(TableMetadata metadata, DecoratedKey partitionKey, R
         }
 
         @Override
-        public String toString()
+        public String toCQLString()
         {
             AbstractType<?> type = column.type;
             switch (operator)
@@ -1336,14 +1403,21 @@ public String toString()
                 case ORDER_BY_ASC:
                 case ORDER_BY_DESC:
                     // These don't have a value, so we return here to prevent an error calling type.getString(value)
-                    return String.format("%s %s", column.name, operator);
+                    return String.format("ORDER BY %s %s", column.name.toCQLString(), operator);
+                case ANN:
+                    return String.format("ORDER BY %s ANN OF %s", column.name.toCQLString(), valueAsCQLString(type, value));
                 default:
                     break;
             }
-            var valueString = type.getString(value);
+            return String.format("%s %s %s", column.name.toCQLString(), operator, valueAsCQLString(type, value));
+        }
+
+        private static String valueAsCQLString(AbstractType<?> type, ByteBuffer value)
+        {
+            var valueString = type.toCQLString(value);
             if (valueString.length() > 9)
                 valueString = valueString.substring(0, 6) + "...";
-            return String.format("%s %s %s", column.name, operator, valueString);
+            return valueString;
         }
 
         @Override
@@ -1455,10 +1529,14 @@ private boolean isSatisfiedByEq(TableMetadata metadata, DecoratedKey partitionKe
         }
 
         @Override
-        public String toString()
+        public String toCQLString()
         {
-            MapType<?, ?> mt = (MapType<?, ?>)column.type;
-            return String.format("%s[%s] %s %s", column.name, mt.nameComparator().getString(key), operator, mt.valueComparator().getString(value));
+            MapType<?, ?> mt = (MapType<?, ?>) column.type;
+            return String.format("%s[%s] %s %s",
+                                 column.name.toCQLString(),
+                                 mt.nameComparator().toCQLString(key),
+                                 operator,
+                                 mt.valueComparator().toCQLString(value));
         }
 
         @Override
@@ -1625,10 +1703,13 @@ public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey,
         }
 
         @Override
-        public String toString()
+        public String toCQLString()
         {
-            return String.format("GEO_DISTANCE(%s, %s) %s %s", column.name, column.type.getString(value),
-                                 distanceOperator, FloatType.instance.getString(distance));
+            return String.format("GEO_DISTANCE(%s, %s) %s %s",
+                                 column.name.toCQLString(),
+                                 column.type.toCQLString(value),
+                                 distanceOperator,
+                                 FloatType.instance.toCQLString(distance));
         }
 
         @Override
@@ -1702,7 +1783,7 @@ public ByteBuffer getValue()
         }
 
         @Override
-        public String toString()
+        public String toCQLString()
         {
             return String.format("expr(%s, %s)",
                                  targetIndex.name,
diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractType.java b/src/java/org/apache/cassandra/db/marshal/AbstractType.java
index e24f766fd28..2faebaf1d7e 100644
--- a/src/java/org/apache/cassandra/db/marshal/AbstractType.java
+++ b/src/java/org/apache/cassandra/db/marshal/AbstractType.java
@@ -33,7 +33,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.cql3.AssignmentTestable;
 import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.cql3.ColumnSpecification;
@@ -183,6 +182,16 @@ public ByteBuffer decompose(T value)
         return getSerializer().serialize(value);
     }
 
+    /**
+     * Generates a CQL literal representing the specified binary value.
+     *
+     * @param bytes the value to convert to a CQL literal.
+     */
+    public String toCQLString(ByteBuffer bytes)
+    {
+        return bytes == null ? "null" : asCQL3Type().toCQLLiteral(bytes, ProtocolVersion.CURRENT);
+    }
+
     /** get a string representation of the bytes used for various identifier (NOT just for log messages) */
     public <V> String getString(V value, ValueAccessor<V> accessor)
     {
diff --git a/src/java/org/apache/cassandra/schema/ColumnMetadata.java b/src/java/org/apache/cassandra/schema/ColumnMetadata.java
index 5a109750343..46ab86c865f 100644
--- a/src/java/org/apache/cassandra/schema/ColumnMetadata.java
+++ b/src/java/org/apache/cassandra/schema/ColumnMetadata.java
@@ -592,9 +592,9 @@ public static String toCQLString(Iterator<ColumnMetadata> defs)
             return "";
 
         StringBuilder sb = new StringBuilder();
-        sb.append(defs.next().name);
+        sb.append(defs.next().name.toCQLString());
         while (defs.hasNext())
-            sb.append(", ").append(defs.next().name);
+            sb.append(", ").append(defs.next().name.toCQLString());
         return sb.toString();
     }
 
@@ -611,7 +611,7 @@ public void appendNameAndOrderTo(CqlBuilder builder)
      *
      * This is the same than the column type, except for non-frozen collections where it's the 'valueComparator'
      * of the collection.
-     * 
+     *
      * This method should not be used to get value type of non-frozon UDT.
      */
     public AbstractType<?> cellValueType()
diff --git a/test/unit/org/apache/cassandra/db/MultiRangeReadCommandCQLTest.java b/test/unit/org/apache/cassandra/db/MultiRangeReadCommandCQLTest.java
new file mode 100644
index 00000000000..5c5ef39bd4f
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/MultiRangeReadCommandCQLTest.java
@@ -0,0 +1,130 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * 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 org.apache.cassandra.db;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.utils.Pair;
+import org.assertj.core.api.Assertions;
+
+public class MultiRangeReadCommandCQLTest extends ReadCommandCQLTester<MultiRangeReadCommand>
+{
+    private static final IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
+
+    @Test
+    public void testToCQLString()
+    {
+        createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c))");
+
+        // prepare a token to query and the midpoints of the left and right ranges
+        Token token = partitioner.getToken(Int32Type.instance.decompose(0));
+        AbstractBounds<PartitionPosition> boundsRight = new Range<>(token.minKeyBound(), partitioner.getMaximumToken().minKeyBound());
+        Token right = partitioner.midpoint(boundsRight.left.getToken(), boundsRight.right.getToken());
+        AbstractBounds<PartitionPosition> boundsLeft = new Range<>(partitioner.getMinimumToken().minKeyBound(), token.maxKeyBound());
+        Token left = partitioner.midpoint(boundsLeft.left.getToken(), boundsLeft.right.getToken());
+
+        // error message for multi-range queries, which are sent to the replicas after splitting a command but are not
+        // directly supported by CQL on the coordinators
+        String multiRangeError = "Restriction on partition key column k must not be nested under OR operator";
+
+        // test with a secondary index
+        createIndex("CREATE INDEX ON %s(v)");
+        assertToCQLString("SELECT * FROM %s WHERE v = 0",
+                          "SELECT * FROM %s WHERE (token(k) <= -1 OR token(k) > -1) AND v = 0 ALLOW FILTERING",
+                          multiRangeError);
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 0 ",
+                          "SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE token(k) > token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE (token(k) > " + token + " AND token(k) <= " + right + " OR token(k) > " + right + ") AND v = 0 ALLOW FILTERING",
+                          multiRangeError);
+        assertToCQLString("SELECT * FROM %s WHERE token(k) >= token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE (token(k) >= " + token + " AND token(k) <= " + right + " OR token(k) > " + right + ") AND v = 0 ALLOW FILTERING",
+                          multiRangeError);
+        assertToCQLString("SELECT * FROM %s WHERE token(k) < token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE (token(k) <= " + left + " OR token(k) > " + left + " AND token(k) < " + token + ") AND v = 0 ALLOW FILTERING",
+                          multiRangeError);
+        assertToCQLString("SELECT * FROM %s WHERE token(k) <= token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE (token(k) <= " + left + " OR token(k) > " + left + " AND token(k) <= " + token + ") AND v = 0 ALLOW FILTERING",
+                          multiRangeError);
+        assertToCQLString("SELECT * FROM %s WHERE token(k) >= token(0) AND token(k) <= token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE token(k) >= " + token + " AND token(k) <= " + token + " AND v = 0 ALLOW FILTERING");
+
+        // test generic index-based ORDER BY
+        createTable("CREATE TABLE %s (k int, c int, n int, PRIMARY KEY (k, c))");
+        createIndex("CREATE CUSTOM INDEX ON %s(n) USING 'StorageAttachedIndex'");
+        assertToCQLString("SELECT * FROM %s ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s WHERE (token(k) <= -1 OR token(k) > -1) ORDER BY n ASC LIMIT 10 ALLOW FILTERING",
+                          multiRangeError);
+        assertToCQLString("SELECT * FROM %s ORDER BY n DESC LIMIT 10",
+                          "SELECT * FROM %s WHERE (token(k) <= -1 OR token(k) > -1) ORDER BY n DESC LIMIT 10 ALLOW FILTERING",
+                          multiRangeError);
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s WHERE k = 0 ORDER BY n ASC LIMIT 10 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE n = 0 ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s WHERE (token(k) <= -1 OR token(k) > -1) AND n = 0 ORDER BY n ASC LIMIT 10 ALLOW FILTERING",
+                          multiRangeError);
+
+        // test ANN index-based ORDER BY
+        createTable("CREATE TABLE %s (k int, c int, n int, v vector<float, 2>, PRIMARY KEY (k, c))");
+        createIndex("CREATE CUSTOM INDEX ON %s(v) USING 'StorageAttachedIndex'");
+        createIndex("CREATE CUSTOM INDEX ON %s(n) USING 'StorageAttachedIndex'");
+        String truncationError = "no viable alternative at input '..'";
+        assertToCQLString("SELECT * FROM %s ORDER BY v ANN OF [1, 2] LIMIT 10",
+                          "SELECT * FROM %s WHERE (token(k) <= -1 OR token(k) > -1) ORDER BY v ANN OF [1.0, ... LIMIT 10 ALLOW FILTERING",
+                          truncationError);
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY v ANN OF [1, 2] LIMIT 10",
+                          "SELECT * FROM %s WHERE k = 0 ORDER BY v ANN OF [1.0, ... LIMIT 10 ALLOW FILTERING",
+                          truncationError);
+        assertToCQLString("SELECT * FROM %s WHERE n = 0 ORDER BY v ANN OF [1, 2] LIMIT 10",
+                          "SELECT * FROM %s WHERE (token(k) <= -1 OR token(k) > -1) AND n = 0 ORDER BY v ANN OF [1.0, ... LIMIT 10 ALLOW FILTERING",
+                          truncationError);
+    }
+
+    @Override
+    protected MultiRangeReadCommand parseCommand(String query)
+    {
+        ReadCommand command = parseReadCommand(query);
+        Assertions.assertThat(command).isInstanceOf(PartitionRangeReadCommand.class);
+        PartitionRangeReadCommand rangeCommand = (PartitionRangeReadCommand) command;
+
+        // split the range into two consecutive ranges if possible, so the command is actually multi-range
+        AbstractBounds<PartitionPosition> bounds = rangeCommand.dataRange().keyRange();
+        Token token = partitioner.midpoint(bounds.left.getToken(), bounds.right.getToken());
+        List<AbstractBounds<PartitionPosition>> ranges;
+        if (bounds.contains(token.maxKeyBound()))
+        {
+            Pair<AbstractBounds<PartitionPosition>, AbstractBounds<PartitionPosition>> split = bounds.split(token.maxKeyBound());
+            ranges = Arrays.asList(split.left, split.right);
+        }
+        else
+        {
+            ranges = List.of(bounds);
+        }
+
+        return MultiRangeReadCommand.create(rangeCommand, ranges, true);
+    }
+}
diff --git a/test/unit/org/apache/cassandra/db/PartitionRangeReadCommandCQLTest.java b/test/unit/org/apache/cassandra/db/PartitionRangeReadCommandCQLTest.java
index 511adb1492b..29c4b12a3c7 100644
--- a/test/unit/org/apache/cassandra/db/PartitionRangeReadCommandCQLTest.java
+++ b/test/unit/org/apache/cassandra/db/PartitionRangeReadCommandCQLTest.java
@@ -29,21 +29,34 @@ public void testToCQLString()
     {
         createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c))");
 
-        assertToCQLString("SELECT * FROM %s", "SELECT * FROM %s ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s",
+                          "SELECT * FROM %s ALLOW FILTERING");
 
-        assertToCQLString("SELECT * FROM %s WHERE c = 0 ALLOW FILTERING", "SELECT * FROM %s WHERE c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE (c) = (0) ALLOW FILTERING", "SELECT * FROM %s WHERE c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c > 0 ALLOW FILTERING", "SELECT * FROM %s WHERE c > 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c < 0 ALLOW FILTERING", "SELECT * FROM %s WHERE c < 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c >= 0 ALLOW FILTERING", "SELECT * FROM %s WHERE c >= 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c <= 0 ALLOW FILTERING", "SELECT * FROM %s WHERE c <= 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c = 0 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE (c) = (0) ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c > 0 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c > 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c < 0 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c < 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c >= 0 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c >= 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c <= 0 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c <= 0 ALLOW FILTERING");
 
-        assertToCQLString("SELECT * FROM %s WHERE v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE v = 1 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c = 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE v = 1 AND c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c > 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE v = 1 AND c > 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c < 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE v = 1 AND c < 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c >= 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE v = 1 AND c >= 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE c <= 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE v = 1 AND c <= 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c = 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c = 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c > 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c > 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c < 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c < 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c >= 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c >= 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE c <= 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE c <= 0 AND v = 1 ALLOW FILTERING");
 
         // test with token restrictions
         IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
@@ -57,42 +70,129 @@ public void testToCQLString()
 
         // test with a secondary index (indexed queries are always mapped to range commands)
         createIndex("CREATE INDEX ON %s(v)");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0", "SELECT * FROM %s WHERE v = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0", "SELECT * FROM %s WHERE v = 0 AND k = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c = 0", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c > 0", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c > 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c < 0", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c < 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c >= 0", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c >= 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c <= 0", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c <= 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c IN (0)", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c IN (0, 1)", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c IN (0, 1) ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND token(k) > token(0)",
-                          "SELECT * FROM %s WHERE v = 0 AND token(k) > " + token + " ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND token(k) >= token(0)",
-                          "SELECT * FROM %s WHERE v = 0 AND token(k) >= " + token + " ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND token(k) >= token(0) AND token(k) <= token(0)",
-                          "SELECT * FROM %s WHERE v = 0 AND token(k) >= " + token + " AND token(k) <= " + token + " ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE v = 0",
+                          "SELECT * FROM %s WHERE v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND  v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 0 ",
+                          "SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c > 0 AND v = 0 ",
+                          "SELECT * FROM %s WHERE k = 0 AND c > 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c < 0 AND v = 0 ",
+                          "SELECT * FROM %s WHERE k = 0 AND c < 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c >= 0 AND v = 0 ",
+                          "SELECT * FROM %s WHERE k = 0 AND c >= 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c <= 0 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c <= 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c IN (0) AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c IN (0, 1) AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c IN (0, 1) AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE token(k) > token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE token(k) > " + token + " AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE token(k) >= token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE token(k) >= " + token + " AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE token(k) >= token(0) AND token(k) <= token(0) AND v = 0",
+                          "SELECT * FROM %s WHERE token(k) >= " + token + " AND token(k) <= " + token + " AND v = 0 ALLOW FILTERING");
+
+        // test with multi-column partition key and index
+        createTable("CREATE TABLE %s (k1 int, k2 int, c int, v int, PRIMARY KEY ((k1, k2), c))");
+        createIndex("CREATE INDEX ON %s(v)");
+        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND v = 1",
+                          "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3 AND v = 1",
+                          "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3 AND v = 1 ALLOW FILTERING");
 
         // test with index and multi-column clustering
         createTable("CREATE TABLE %s (k int, c1 int, c2 int,v int, PRIMARY KEY (k, c1, c2))");
         createIndex("CREATE INDEX ON %s(v)");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0", "SELECT * FROM %s WHERE v = 0 AND k = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 > 1", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 > 1 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 < 1", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 < 1 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 >= 1", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 >= 1 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 <= 1", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 <= 1 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 = 2", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND (c1, c2) = (1, 2) ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 > 2", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 > 2 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 < 2", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 < 2 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 >= 2", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 >= 2 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 <= 2", "SELECT * FROM %s WHERE v = 0 AND k = 0 AND c1 = 1 AND c2 <= 2 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 > 1 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 > 1 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 < 1 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 < 1 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 >= 1 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 >= 1 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 <= 1 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 <= 1 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 = 2 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND (c1, c2) = (1, 2) AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 > 2 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 > 2 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 < 2 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 < 2 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 >= 2 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 >= 2 AND v = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 <= 2 AND v = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 = 1 AND c2 <= 2 AND v = 0 ALLOW FILTERING");
 
-        // test with multi-column partition key and index
-        createTable("CREATE TABLE %s (k1 int, k2 int, c int, v int, PRIMARY KEY ((k1, k2), c))");
-        createIndex("CREATE INDEX ON %s(v)");
-        assertToCQLString("SELECT * FROM %s WHERE v = 1 AND k1 = 1 AND k2 = 2", "SELECT * FROM %s WHERE v = 1 AND k1 = 1 AND k2 = 2 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE v = 1 AND k1 = 1 AND k2 = 2 AND c = 3", "SELECT * FROM %s WHERE v = 1 AND k1 = 1 AND k2 = 2 AND c = 3 ALLOW FILTERING");
+        // test generic index-based ORDER BY
+        createTable("CREATE TABLE %s (k int, c int, n int, PRIMARY KEY (k, c))");
+        createIndex("CREATE CUSTOM INDEX ON %s(n) USING 'StorageAttachedIndex'");
+        createIndex("CREATE CUSTOM INDEX ON %s(c) USING 'StorageAttachedIndex'");
+        assertToCQLString("SELECT * FROM %s ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s ORDER BY n ASC LIMIT 10 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s ORDER BY n DESC LIMIT 10",
+                          "SELECT * FROM %s ORDER BY n DESC LIMIT 10 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s WHERE k = 0 ORDER BY n ASC LIMIT 10 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c = 0 ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s WHERE k = 0 AND c = 0 ORDER BY n ASC LIMIT 10 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c IN (0, 1) ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s WHERE k = 0 AND c IN (0, 1) ORDER BY n ASC LIMIT 10 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE n = 0 ORDER BY n LIMIT 10",
+                          "SELECT * FROM %s WHERE n = 0 ORDER BY n ASC LIMIT 10 ALLOW FILTERING");
+
+        // test ANN index-based ORDER BY
+        createTable("CREATE TABLE %s (k int, c int, n int, v vector<float, 2>, PRIMARY KEY (k, c))");
+        createIndex("CREATE CUSTOM INDEX ON %s(v) USING 'StorageAttachedIndex'");
+        createIndex("CREATE CUSTOM INDEX ON %s(n) USING 'StorageAttachedIndex'");
+        String truncationError = "no viable alternative at input '..'";
+        assertToCQLString("SELECT * FROM %s ORDER BY v ANN OF [1, 2] LIMIT 10",
+                          "SELECT * FROM %s ORDER BY v ANN OF [1.0, ... LIMIT 10 ALLOW FILTERING",
+                          truncationError);
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY v ANN OF [1, 2] LIMIT 10",
+                          "SELECT * FROM %s WHERE k = 0 ORDER BY v ANN OF [1.0, ... LIMIT 10 ALLOW FILTERING",
+                          truncationError);
+        assertToCQLString("SELECT * FROM %s WHERE n = 0 ORDER BY v ANN OF [1, 2] LIMIT 10",
+                          "SELECT * FROM %s WHERE n = 0 ORDER BY v ANN OF [1.0, ... LIMIT 10 ALLOW FILTERING",
+                          truncationError);
+
+        // test literals
+        createTable("CREATE TABLE %s (k text, c text, m map<text, text>, PRIMARY KEY (k, c))");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE m['key'] = 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE m['key'] = 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE m['key'] < 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE m['key'] < 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE m['key'] > 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE m['key'] > 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE m['key'] <= 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE m['key'] <= 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE m['key'] >= 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE m['key'] >= 'value' ALLOW FILTERING");
+
+        // test with quoted identifiers
+        createKeyspace("CREATE KEYSPACE \"K\" WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");
+        createTable("CREATE TABLE \"K\".\"T\" (\"K\" int, \"C\" int, \"S\" int static, \"V\" int, PRIMARY KEY(\"K\", \"C\"))");
+        assertToCQLString("SELECT * FROM \"K\".\"T\"",
+                          "SELECT * FROM \"K\".\"T\" ALLOW FILTERING");
+        assertToCQLString("SELECT \"K\" FROM \"K\".\"T\"",
+                          "SELECT * FROM \"K\".\"T\" ALLOW FILTERING");
+        assertToCQLString("SELECT \"S\" FROM \"K\".\"T\"",
+                          "SELECT \"S\" FROM \"K\".\"T\" ALLOW FILTERING");
+        assertToCQLString("SELECT \"V\" FROM \"K\".\"T\"",
+                          "SELECT \"V\" FROM \"K\".\"T\" ALLOW FILTERING");
+        assertToCQLString("SELECT \"K\", \"C\", \"S\", \"V\" FROM \"K\".\"T\"",
+                          "SELECT \"S\", \"V\" FROM \"K\".\"T\" ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"V\" = 0 ALLOW FILTERING",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"V\" = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"S\" = 0 ALLOW FILTERING",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"S\" = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"C\" = 0 ALLOW FILTERING",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"C\" = 0 ALLOW FILTERING");
     }
 
     @Override
diff --git a/test/unit/org/apache/cassandra/db/ReadCommandCQLTester.java b/test/unit/org/apache/cassandra/db/ReadCommandCQLTester.java
index dda7af9f356..aa1a017b29b 100644
--- a/test/unit/org/apache/cassandra/db/ReadCommandCQLTester.java
+++ b/test/unit/org/apache/cassandra/db/ReadCommandCQLTester.java
@@ -15,6 +15,8 @@
  */
 package org.apache.cassandra.db;
 
+import javax.annotation.Nullable;
+
 import org.apache.cassandra.cql3.CQLTester;
 import org.assertj.core.api.Assertions;
 
@@ -23,10 +25,20 @@ public abstract class ReadCommandCQLTester<T extends ReadCommand> extends CQLTes
     protected abstract T parseCommand(String query);
 
     protected void assertToCQLString(String query, String expected)
+    {
+        assertToCQLString(query, expected, null);
+    }
+
+    protected void assertToCQLString(String query, String expected, @Nullable String expectedErrorMessage)
     {
         T command = parseCommand(query);
-        Assertions.assertThat(command.toCQLString())
-                  .isEqualTo(formatQuery(expected));
+        String actual = command.toCQLString();
+
+        Assertions.assertThat(actual).isEqualTo(formatQuery(expected));
+
+        if (expectedErrorMessage == null)
+            execute(actual);
+        else
+            Assertions.assertThatThrownBy(() -> execute(actual)).hasMessageContaining(expectedErrorMessage);
     }
 }
-
diff --git a/test/unit/org/apache/cassandra/db/SinglePartitionReadCommandCQLTest.java b/test/unit/org/apache/cassandra/db/SinglePartitionReadCommandCQLTest.java
index 977fe7cb541..20adfd2072d 100644
--- a/test/unit/org/apache/cassandra/db/SinglePartitionReadCommandCQLTest.java
+++ b/test/unit/org/apache/cassandra/db/SinglePartitionReadCommandCQLTest.java
@@ -44,26 +44,108 @@ public void testToCQLString()
     {
         createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c))");
 
-        assertToCQLString("SELECT * FROM %s WHERE k = 0", "SELECT * FROM %s WHERE k = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0",
+                          "SELECT * FROM %s WHERE k = 0 ALLOW FILTERING");
 
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c = 0", "SELECT * FROM %s WHERE k = 0 AND c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND (c) = (0)", "SELECT * FROM %s WHERE k = 0 AND c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c > 0", "SELECT * FROM %s WHERE k = 0 AND c > 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c < 0", "SELECT * FROM %s WHERE k = 0 AND c < 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c >= 0", "SELECT * FROM %s WHERE k = 0 AND c >= 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c <= 0", "SELECT * FROM %s WHERE k = 0 AND c <= 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c = 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND (c) = (0)",
+                          "SELECT * FROM %s WHERE k = 0 AND c = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c > 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c > 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c < 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c < 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c >= 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c >= 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c <= 0",
+                          "SELECT * FROM %s WHERE k = 0 AND c <= 0 ALLOW FILTERING");
 
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE k = 0 AND v = 1 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE k = 0 AND v = 1 AND c = 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c > 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE k = 0 AND v = 1 AND c > 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c < 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE k = 0 AND v = 1 AND c < 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c >= 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE k = 0 AND v = 1 AND c >= 0 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c <= 0 AND v = 1 ALLOW FILTERING", "SELECT * FROM %s WHERE k = 0 AND v = 1 AND c <= 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE k = 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE k = 0 AND c = 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c > 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE k = 0 AND c > 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c < 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE k = 0 AND c < 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c >= 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE k = 0 AND c >= 0 AND v = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c <= 0 AND v = 1 ALLOW FILTERING",
+                          "SELECT * FROM %s WHERE k = 0 AND c <= 0 AND v = 1 ALLOW FILTERING");
+
+        // test clustering-based ORDER BY
+        createTable("CREATE TABLE %s (k int, c1 int, c2 int, PRIMARY KEY (k, c1, c2))");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY c1",
+                          "SELECT * FROM %s WHERE k = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY c1 ASC",
+                          "SELECT * FROM %s WHERE k = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY c1 ASC, c2 ASC",
+                          "SELECT * FROM %s WHERE k = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY c1 DESC",
+                          "SELECT * FROM %s WHERE k = 0 ORDER BY c1 DESC, c2 DESC ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 ORDER BY c1 DESC, c2 DESC",
+                          "SELECT * FROM %s WHERE k = 0 ORDER BY c1 DESC, c2 DESC ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 > 0 ORDER BY c1",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 > 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 > 0 ORDER BY c1 ASC",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 > 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 > 0 ORDER BY c1 ASC, c2 ASC",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 > 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 > 0 ORDER BY c1 DESC",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 > 0 ORDER BY c1 DESC, c2 DESC ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k = 0 AND c1 > 0 ORDER BY c1 DESC, c2 DESC",
+                          "SELECT * FROM %s WHERE k = 0 AND c1 > 0 ORDER BY c1 DESC, c2 DESC ALLOW FILTERING");
+
+        // test with multi-column partition key (without index to test SinglePartitionReadCommand directly)
+        createTable("CREATE TABLE %s (k1 int, k2 int, c int, v int, PRIMARY KEY ((k1, k2), c))");
+        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2",
+                          "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3",
+                          "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3 ALLOW FILTERING");
 
         // test with multi-column partition key (without index to test SinglePartitionReadCommand directly)
         createTable("CREATE TABLE %s (k1 int, k2 int, c int, v int, PRIMARY KEY ((k1, k2), c))");
-        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2", "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 ALLOW FILTERING");
-        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3", "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2",
+                          "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3",
+                          "SELECT * FROM %s WHERE k1 = 1 AND k2 = 2 AND c = 3 ALLOW FILTERING");
+
+        // test literals
+        createTable("CREATE TABLE %s (k text, c text, m map<text, text>, PRIMARY KEY (k, c))");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE k = 'k' AND m['key'] = 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE k = 'k' AND m['key'] = 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE k = 'k' AND c = 'c' AND m['key'] = 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE k = 'k' AND c = 'c' AND m['key'] = 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE k = 'k' AND c > 'c' AND m['key'] > 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE k = 'k' AND c > 'c' AND m['key'] > 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE k = 'k' AND c < 'c' AND m['key'] < 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE k = 'k' AND c < 'c' AND m['key'] < 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE k = 'k' AND c >= 'c' AND m['key'] >= 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE k = 'k' AND c >= 'c' AND m['key'] >= 'value' ALLOW FILTERING");
+        assertToCQLString("SELECT m['key'] FROM %s WHERE k = 'k' AND c <= 'c' AND m['key'] <= 'value' ALLOW FILTERING",
+                          "SELECT m FROM %s WHERE k = 'k' AND c <= 'c' AND m['key'] <= 'value' ALLOW FILTERING");
+
+        // test with quoted identifiers
+        createKeyspace("CREATE KEYSPACE \"K\" WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");
+        createTable("CREATE TABLE \"K\".\"T\" (\"K\" int, \"C\" int, \"S\" int static, \"V\" int, PRIMARY KEY(\"K\", \"C\"))");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT \"K\" FROM \"K\".\"T\" WHERE \"K\" = 0",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT \"S\" FROM \"K\".\"T\" WHERE \"K\" = 0",
+                          "SELECT \"S\" FROM \"K\".\"T\" WHERE \"K\" = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT \"V\" FROM \"K\".\"T\" WHERE \"K\" = 0",
+                          "SELECT \"V\" FROM \"K\".\"T\" WHERE \"K\" = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT \"K\", \"C\", \"S\", \"V\" FROM \"K\".\"T\" WHERE \"K\" = 0",
+                          "SELECT \"S\", \"V\" FROM \"K\".\"T\" WHERE \"K\" = 0 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 AND \"C\" = 1",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 AND \"C\" = 1 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 AND \"C\" > 1 AND \"C\" <= 2",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 AND \"C\" > 1 AND \"C\" <= 2 ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 ORDER BY \"C\" DESC",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 ORDER BY \"C\" DESC ALLOW FILTERING");
+        assertToCQLString("SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 AND \"C\" = 1 ORDER BY \"C\" DESC",
+                          "SELECT * FROM \"K\".\"T\" WHERE \"K\" = 0 AND \"C\" = 1 ORDER BY \"C\" DESC ALLOW FILTERING");
     }
 
     @Override
diff --git a/test/unit/org/apache/cassandra/db/filter/DataLimitsTest.java b/test/unit/org/apache/cassandra/db/filter/DataLimitsTest.java
index e88cba45f92..099f977c262 100644
--- a/test/unit/org/apache/cassandra/db/filter/DataLimitsTest.java
+++ b/test/unit/org/apache/cassandra/db/filter/DataLimitsTest.java
@@ -85,17 +85,17 @@ public void toStringTest()
         String lastRetKeyStr = String.format("lastReturnedKey=%s", ByteBufferUtil.bytesToHex(lastReturnedKey));
         String lastRetKeyRemainingStr = "lastReturnedKeyRemaining=5";
 
-        assertThat(cqlLimits.toString()).contains("ROWS LIMIT 19").contains("PER PARTITION LIMIT 17").doesNotContain("BYTES LIMIT");
-        assertThat(cqlLimitsForPagingInRows.toString()).contains("ROWS LIMIT 13").contains("PER PARTITION LIMIT 17").doesNotContain("BYTES LIMIT");
-        assertThat(cqlLimitsForPagingInBytes.toString()).contains("BYTES LIMIT 13").contains("ROWS LIMIT 19").contains("PER PARTITION LIMIT 17");
-        assertThat(cqlLimitsForPagingInRowsWithLastRow.toString()).contains("ROWS LIMIT 13").contains("PER PARTITION LIMIT 17").doesNotContain("BYTES LIMIT").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
-        assertThat(cqlLimitsForPagingInBytesWithLastRow.toString()).contains("BYTES LIMIT 13").contains("ROWS LIMIT 19").contains("PER PARTITION LIMIT 17").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
-
-        assertThat(groupByLimits.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").doesNotContain("ROWS LIMIT").doesNotContain("BYTES LIMIT");
-        assertThat(groupByLimitsForPagingInRows.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").contains("ROWS LIMIT 13").doesNotContain("BYTES LIMIT");
-        assertThat(groupByLimitsForPagingInBytes.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").doesNotContain("ROWS LIMIT").contains("BYTES LIMIT 13");
-        assertThat(groupByLimitsForPagingInRowsWithLastRow.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").contains("ROWS LIMIT 13").doesNotContain("BYTES LIMIT").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
-        assertThat(groupByLimitsForPagingInBytesWithLastRow.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").doesNotContain("ROWS LIMIT").contains("BYTES LIMIT 13").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
+        assertThat(cqlLimits.toString()).contains("LIMIT 19").contains("PER PARTITION LIMIT 17").doesNotContain("BYTES LIMIT");
+        assertThat(cqlLimitsForPagingInRows.toString()).contains("LIMIT 13").contains("PER PARTITION LIMIT 17").doesNotContain("BYTES LIMIT");
+        assertThat(cqlLimitsForPagingInBytes.toString()).contains("BYTES LIMIT 13").contains("LIMIT 19").contains("PER PARTITION LIMIT 17");
+        assertThat(cqlLimitsForPagingInRowsWithLastRow.toString()).contains("LIMIT 13").contains("PER PARTITION LIMIT 17").doesNotContain("BYTES LIMIT").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
+        assertThat(cqlLimitsForPagingInBytesWithLastRow.toString()).contains("BYTES LIMIT 13").contains("LIMIT 19").contains("PER PARTITION LIMIT 17").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
+
+        assertThat(groupByLimits.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").doesNotContain("BYTES LIMIT");
+        assertThat(groupByLimitsForPagingInRows.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").contains("LIMIT 13").doesNotContain("BYTES LIMIT");
+        assertThat(groupByLimitsForPagingInBytes.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").contains("BYTES LIMIT 13");
+        assertThat(groupByLimitsForPagingInRowsWithLastRow.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").contains("LIMIT 13").doesNotContain("BYTES LIMIT").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
+        assertThat(groupByLimitsForPagingInBytesWithLastRow.toString()).contains("GROUP LIMIT 19").contains("GROUP PER PARTITION LIMIT 17").contains("BYTES LIMIT 13").contains(lastRetKeyStr).contains(lastRetKeyRemainingStr);
     }
 
     private void checkSerialization(MessagingService.Version version, DataLimits limits, String name)
@@ -136,4 +136,4 @@ private void checkSerialization(MessagingService.Version version, DataLimits lim
             fail(msg);
         }
     }
-}
\ No newline at end of file
+}
diff --git a/test/unit/org/apache/cassandra/index/sai/plan/PlanTest.java b/test/unit/org/apache/cassandra/index/sai/plan/PlanTest.java
index 590326e4c83..43a2f89c6ce 100644
--- a/test/unit/org/apache/cassandra/index/sai/plan/PlanTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/plan/PlanTest.java
@@ -107,7 +107,7 @@ private static Expression saiPred(String column, Expression.Op operation, boolea
     private static RowFilter.Expression filerPred(String column, Operator operation)
     {
         RowFilter.Expression pred = Mockito.mock(RowFilter.Expression.class);
-        Mockito.when(pred.toString()).thenReturn(column + ' ' + operation + " X");
+        Mockito.when(pred.toCQLString()).thenReturn(column + ' ' + operation + " X");
         Mockito.when(pred.operator()).thenReturn(operation);
         return pred;
     }