From e03373ac9108f516cd4496b993dae2dade8026c1 Mon Sep 17 00:00:00 2001
From: Ekaterina Dimitrova <ekaterina.dimitrova@datastax.com>
Date: Wed, 15 Jan 2025 13:36:03 -0500
Subject: [PATCH 1/3] CNDB-12425: A preliminary patch to facilitate
 discussions. It does not include: - feature flag - checks we are on the new
 messaging version added for ANNOptions - we fall back to allow filtering only
 on Index Creation. Currently we also fall back to ALLOW FILTERING if we use
 nodetool to rebuild indexes

---
 .../restrictions/StatementRestrictions.java   |  13 +-
 .../cql3/statements/SelectStatement.java      |   2 +-
 .../org/apache/cassandra/db/ReadCommand.java  |  80 +++--
 .../cassandra/db/filter/ANNOptions.java       |   5 +-
 .../apache/cassandra/db/filter/RowFilter.java |  36 ++-
 .../org/apache/cassandra/index/Index.java     |  11 +
 .../apache/cassandra/index/IndexRegistry.java |   7 +
 .../index/SecondaryIndexManager.java          |  40 ++-
 .../cassandra/index/SingletonIndexGroup.java  |  21 +-
 .../index/sai/StorageAttachedIndexGroup.java  |  16 +
 .../cassandra/locator/ReplicaPlans.java       |  12 +-
 .../service/reads/AbstractReadExecutor.java   |   2 +-
 .../service/reads/range/RangeCommands.java    |   3 +-
 .../reads/range/ReplicaPlanIterator.java      |  10 +-
 .../test/sai/ANNOptionsDistributedTest.java   |   5 +-
 .../test/sai/IndexAvailabilityTest.java       | 226 +++++++++++++-
 .../entities/SecondaryIndexTest.java          |   2 +
 .../cassandra/db/filter/ANNOptionsTest.java   |   6 +
 .../cassandra/index/CustomIndexTest.java      |   6 +
 .../index/sai/cql/AllowFilteringTest.java     | 278 +++++++++++++++++-
 .../AssureSufficientLiveNodesTest.java        |  10 +-
 .../cassandra/locator/ReplicaPlansTest.java   |   4 +-
 ...pointGroupingRangeCommandIteratorTest.java |   3 +-
 .../reads/range/RangeCommandIteratorTest.java |   4 +-
 .../reads/range/RangeCommandsTest.java        |   7 +-
 .../reads/range/ReplicaPlanIteratorTest.java  |   2 +-
 .../reads/range/ReplicaPlanMergerTest.java    |   2 +-
 27 files changed, 708 insertions(+), 105 deletions(-)

diff --git a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
index b20d658867e8..228347fda5ed 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
@@ -772,6 +772,11 @@ public boolean hasIndxBasedOrdering()
         return nonPrimaryKeyRestrictions.restrictions().stream().anyMatch(SingleRestriction::isIndexBasedOrdering);
     }
 
+    public boolean hasIndxBasedBoundedAnn()
+    {
+        return nonPrimaryKeyRestrictions.restrictions().stream().anyMatch(SingleRestriction::isBoundedAnn);
+    }
+
     public void throwRequiresAllowFilteringError(TableMetadata table)
     {
         if (hasIndxBasedOrdering())
@@ -988,7 +993,7 @@ private boolean hasUnrestrictedClusteringColumns()
         return table.clusteringColumns().size() != clusteringColumnsRestrictions.size();
     }
 
-    public RowFilter getRowFilter(IndexRegistry indexManager, QueryOptions options, QueryState queryState, SelectOptions selectOptions)
+    public RowFilter getRowFilter(IndexRegistry indexManager, QueryOptions options, QueryState queryState, SelectOptions selectOptions, boolean allowsFiltering)
     {
         boolean hasAnnOptions = selectOptions.hasANNOptions();
 
@@ -1001,7 +1006,11 @@ public RowFilter getRowFilter(IndexRegistry indexManager, QueryOptions options,
         }
 
         ANNOptions annOptions = selectOptions.parseANNOptions();
-        RowFilter rowFilter = RowFilter.builder(indexManager)
+
+        if (allowsFiltering && (hasIndxBasedOrdering() || hasIndxBasedBoundedAnn()))
+            allowsFiltering = false;
+
+        RowFilter rowFilter = RowFilter.builder(indexManager, allowsFiltering)
                                        .buildFromRestrictions(this, table, options, queryState, annOptions);
 
         if (hasAnnOptions && !rowFilter.hasANN())
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 58b75aacab5a..b613a57361aa 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -1001,7 +1001,7 @@ private NavigableSet<Clustering<?>> getRequestedRows(QueryOptions options, Query
     public RowFilter getRowFilter(QueryOptions options, QueryState state) throws InvalidRequestException
     {
         IndexRegistry indexRegistry = IndexRegistry.obtain(table);
-        return restrictions.getRowFilter(indexRegistry, options, state, selectOptions);
+        return restrictions.getRowFilter(indexRegistry, options, state, selectOptions, parameters.allowFiltering);
     }
 
     private ResultSet process(PartitionIterator partitions,
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index b110a3ada2de..a2752fdf2a3f 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -18,12 +18,7 @@
 package org.apache.cassandra.db;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-import java.util.Set;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -65,6 +60,7 @@
 import org.apache.cassandra.guardrails.Threshold;
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.index.IndexRegistry;
+import org.apache.cassandra.index.SecondaryIndexManager;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.DataInputPlus;
@@ -83,19 +79,22 @@
 import org.apache.cassandra.sensors.Context;
 import org.apache.cassandra.sensors.read.TrackingRowIterator;
 import org.apache.cassandra.service.ActiveRepairService;
+import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.FBUtilities;
 
 import static com.google.common.collect.Iterables.any;
 import static com.google.common.collect.Iterables.filter;
+import static java.lang.String.format;
 import static org.apache.cassandra.db.partitions.UnfilteredPartitionIterators.MergeListener.NOOP;
+import static org.apache.cassandra.index.SecondaryIndexManager.getIndexStatus;
 import static org.apache.cassandra.utils.MonotonicClock.approxTime;
 
 /**
  * General interface for storage-engine read commands (common to both range and
  * single partition commands).
  * <p>
- * This contains all the informations needed to do a local read.
+ * This contains all the information needed to do a local read.
  */
 public abstract class ReadCommand extends AbstractReadQuery
 {
@@ -215,7 +214,7 @@ public int digestVersion()
      * this allows us to use the command as a carrier of the digest version even if we only call
      * setIsDigestQuery on some copy of it.
      *
-     * @param digestVersion the version for the digest is this command is used for digest query..
+     * @param digestVersion the version for the digest is this command is used for digest query.
      * @return this read command.
      */
     public ReadCommand setDigestVersion(int digestVersion)
@@ -408,10 +407,12 @@ public UnfilteredPartitionIterator executeLocally(ReadExecutionController execut
         Index.Searcher searcher = null;
         if (indexQueryPlan != null)
         {
-            cfs.indexManager.checkQueryability(indexQueryPlan);
-            searcher = indexSearcher();
-            Index index = indexQueryPlan.getFirst();
-            Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.keyspace, cfs.metadata.name, index.getIndexMetadata().name);
+            if (cfs.indexManager.isQueryableThroughIndex(indexQueryPlan, rowFilter().allowFiltering))
+            {
+                searcher = indexSearcher();
+                Index index = indexQueryPlan.getFirst();
+                Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.keyspace, cfs.metadata.name, index.getIndexMetadata().name);
+            }
         }
 
         Context context = Context.from(this);
@@ -531,9 +532,9 @@ public ReadExecutionController executionController(boolean trackRepairedStatus)
     }
 
     /**
-     * Allow to post-process the result of the query after it has been reconciled on the coordinator
+     * Allow to post-process the result of the query after it has been reconciled on the coordinator,
      * but before it is passed to the CQL layer to return the ResultSet.
-     *
+     * <p>
      * See CASSANDRA-8717 for why this exists.
      */
     public PartitionIterator postReconciliationProcessing(PartitionIterator result)
@@ -741,7 +742,7 @@ public Message<ReadCommand> createMessage(boolean trackRepairedData)
 
     // 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).
+    // are to some extent an artefact of compaction lagging behind and hence counting them is somewhat unintuitive).
     protected UnfilteredPartitionIterator withoutPurgeableTombstones(UnfilteredPartitionIterator iterator,
                                                                      ColumnFamilyStore cfs,
                                                                      ReadExecutionController controller)
@@ -769,7 +770,7 @@ protected LongPredicate getPurgeEvaluator()
      * Note that in general the returned string will not be exactly the original user string, first
      * because there isn't always a single syntax for a given query,  but also because we don't have
      * all the information needed (we know the non-PK columns queried but not the PK ones as internally
-     * we query them all). So this shouldn't be relied too strongly, but this should be good enough for
+     * we query them all). So this shouldn't be relied on too strongly, but this should be good enough for
      * debugging purpose which is what this is for.
      */
     public String toCQLString()
@@ -833,7 +834,7 @@ InputCollector<UnfilteredPartitionIterator> iteratorsForRange(ColumnFamilyStore.
      * input for a query. Separates them according to repaired status and of repaired
      * status is being tracked, handles the merge and wrapping in a digest generator of
      * the repaired iterators.
-     *
+     * <p>
      * Intentionally not AutoCloseable so we don't mistakenly use this in ARM blocks
      * as this prematurely closes the underlying iterators
      */
@@ -1041,7 +1042,7 @@ public ReadCommand deserialize(DataInputPlus in, int version) throws IOException
             int flags = in.readByte();
             boolean isDigest = isDigest(flags);
             boolean acceptsTransient = acceptsTransient(flags);
-            // Shouldn't happen or it's a user error (see comment above) but
+            // Shouldn't happen, or it's a user error (see comment above) but
             // better complain loudly than doing the wrong thing.
             if (isForThrift(flags))
                 throw new IllegalStateException("Received a command with the thrift flag set. "
@@ -1060,15 +1061,48 @@ public ReadCommand deserialize(DataInputPlus in, int version) throws IOException
             if (hasIndex)
             {
                 IndexMetadata index = deserializeIndexMetadata(in, version, metadata);
-                if (index != null)
+                indexQueryPlan = buildIndexQueryPlan(index, metadata, rowFilter);
+            }
+
+            return kind.selectionDeserializer.deserialize(in, version, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, indexQueryPlan);
+        }
+
+        private Index.QueryPlan buildIndexQueryPlan(IndexMetadata index, TableMetadata metadata, RowFilter rowFilter)
+        {
+            if (index == null)
+                return null;
+
+            Index.Group indexGroup = Keyspace.openAndGetStore(metadata).indexManager.getIndexGroup(index);
+            if (indexGroup == null)
+                return null;
+
+            Index.QueryPlan queryPlan = indexGroup.queryPlanFor(rowFilter);
+            if (queryPlan == null)
+                return null;
+
+            Set<Index> allIndexes = queryPlan.getIndexes();
+            Set<Index> availableIndexes = new HashSet<>();
+            for (Index plannedIndex : allIndexes)
+            {
+                String indexName = plannedIndex.getIndexMetadata().name;
+                Index.Status status = getIndexStatus(FBUtilities.getBroadcastAddressAndPort(),
+                                                     metadata.keyspace,
+                                                     indexName);
+                if (status != Index.Status.FULL_REBUILD_STARTED)
+                    availableIndexes.add(plannedIndex);
+                else
                 {
-                    Index.Group indexGroup = Keyspace.openAndGetStore(metadata).indexManager.getIndexGroup(index);
-                    if (indexGroup != null)
-                        indexQueryPlan = indexGroup.queryPlanFor(rowFilter);
+                    ClientWarn.instance.warn(format(SecondaryIndexManager.FELL_BACK_TO_ALLOW_FILTERING, indexName));
+                    logger.warn(format(SecondaryIndexManager.FELL_BACK_TO_ALLOW_FILTERING, indexName));
                 }
             }
 
-            return kind.selectionDeserializer.deserialize(in, version, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, indexQueryPlan);
+            if (availableIndexes.isEmpty())
+                return null;
+
+            return availableIndexes.size() < queryPlan.getIndexes().size()
+                   ? indexGroup.queryPlanForIndices(rowFilter, availableIndexes)
+                   : queryPlan;
         }
 
         private @Nullable IndexMetadata deserializeIndexMetadata(DataInputPlus in, int version, TableMetadata metadata) throws IOException
diff --git a/src/java/org/apache/cassandra/db/filter/ANNOptions.java b/src/java/org/apache/cassandra/db/filter/ANNOptions.java
index ea74ff6f1136..72ffb53318c4 100644
--- a/src/java/org/apache/cassandra/db/filter/ANNOptions.java
+++ b/src/java/org/apache/cassandra/db/filter/ANNOptions.java
@@ -35,6 +35,9 @@
  */
 public class ANNOptions
 {
+    public static final String REQUIRES_HIGHER_MESSAGING_VERSION =
+    "ANN options are not supported in clusters below DS 11.";
+
     public static final String RERANK_K_OPTION_NAME = "rerank_k";
 
     public static final ANNOptions NONE = new ANNOptions(null);
@@ -79,7 +82,7 @@ public static ANNOptions fromMap(Map<String, String> map)
         if (MessagingService.current_version < MessagingService.VERSION_DS_11)
             badNodes.add(FBUtilities.getBroadcastAddressAndPort());
         if (!badNodes.isEmpty())
-            throw new InvalidRequestException("ANN options are not supported in clusters below DS 11.");
+            throw new InvalidRequestException(REQUIRES_HIGHER_MESSAGING_VERSION);
 
         Integer rerankK = null;
 
diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java
index dad3bb212d8a..58da52082609 100644
--- a/src/java/org/apache/cassandra/db/filter/RowFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java
@@ -79,13 +79,16 @@ public class RowFilter
     private static final Logger logger = LoggerFactory.getLogger(RowFilter.class);
 
     public static final Serializer serializer = new Serializer();
-    public static final RowFilter NONE = new RowFilter(FilterElement.NONE);
+    public static final RowFilter NONE = new RowFilter(FilterElement.NONE, false);
 
     private final FilterElement root;
 
-    protected RowFilter(FilterElement root)
+    public final boolean allowFiltering;
+
+    protected RowFilter(FilterElement root, boolean allowFitlering)
     {
         this.root = root;
+        this.allowFiltering = allowFitlering;
     }
 
     public FilterElement root()
@@ -294,7 +297,7 @@ public RowFilter without(Expression expression)
         if (root.size() == 1)
             return RowFilter.NONE;
 
-        return new RowFilter(root.filter(e -> !e.equals(expression)));
+        return new RowFilter(root.filter(e -> !e.equals(expression)), allowFiltering);
     }
 
     public RowFilter withoutExpressions()
@@ -307,12 +310,12 @@ public RowFilter withoutExpressions()
      */
     public RowFilter withoutDisjunctions()
     {
-        return new RowFilter(root.withoutDisjunctions());
+        return new RowFilter(root.withoutDisjunctions(), allowFiltering);
     }
 
     public RowFilter restrict(Predicate<Expression> filter)
     {
-        return new RowFilter(root.filter(filter));
+        return new RowFilter(root.filter(filter), allowFiltering);
     }
 
     public boolean isEmpty()
@@ -328,12 +331,12 @@ public String toString()
 
     public static Builder builder()
     {
-        return new Builder(null);
+        return new Builder(null, false);
     }
 
-    public static Builder builder(IndexRegistry indexRegistry)
+    public static Builder builder(IndexRegistry indexRegistry, boolean allowFitlering)
     {
-        return new Builder(indexRegistry);
+        return new Builder(indexRegistry, allowFitlering);
     }
 
     public static class Builder
@@ -341,15 +344,17 @@ public static class Builder
         private FilterElement.Builder current = new FilterElement.Builder(false);
 
         private final IndexRegistry indexRegistry;
+        private final Boolean allowFiltering;
 
-        public Builder(IndexRegistry indexRegistry)
+        public Builder(IndexRegistry indexRegistry, boolean allowFiltering)
         {
             this.indexRegistry = indexRegistry;
+            this.allowFiltering = allowFiltering;
         }
 
         public RowFilter build()
         {
-            return new RowFilter(current.build());
+            return new RowFilter(current.build(), allowFiltering);
         }
 
         public RowFilter buildFromRestrictions(StatementRestrictions restrictions,
@@ -363,7 +368,7 @@ public RowFilter buildFromRestrictions(StatementRestrictions restrictions,
             if (Guardrails.queryFilters.enabled(queryState))
                 Guardrails.queryFilters.guard(root.numFilteredValues(), "Select query", false, queryState);
 
-            return new RowFilter(root);
+            return new RowFilter(root, allowFiltering);
         }
 
         private FilterElement doBuild(StatementRestrictions restrictions,
@@ -420,7 +425,7 @@ public void addAllAsConjunction(Consumer<Builder> addToRowFilterDelegate)
             {
                 // If we're in disjunction mode, we must not pass the current builder to addToRowFilter.
                 // We create a new conjunction sub-builder instead and add all expressions there.
-                var builder = new Builder(indexRegistry);
+                var builder = new Builder(indexRegistry, allowFiltering);
                 addToRowFilterDelegate.accept(builder);
 
                 if (builder.current.expressions.size() == 1 && builder.current.children.isEmpty())
@@ -1836,19 +1841,22 @@ public void serialize(RowFilter filter, DataOutputPlus out, int version) throws
         {
             out.writeBoolean(false); // Old "is for thrift" boolean
             FilterElement.serializer.serialize(filter.root, out, version);
+            out.writeBoolean(filter.allowFiltering);
         }
 
         public RowFilter deserialize(DataInputPlus in, int version, TableMetadata metadata) throws IOException
         {
             in.readBoolean(); // Unused
             FilterElement operation = FilterElement.serializer.deserialize(in, version, metadata);
-            return new RowFilter(operation);
+            boolean allowFiltering = in.readBoolean();
+            return new RowFilter(operation, allowFiltering);
         }
 
         public long serializedSize(RowFilter filter, int version)
         {
             return 1 // unused boolean
-                   + FilterElement.serializer.serializedSize(filter.root, version);
+                   + FilterElement.serializer.serializedSize(filter.root, version)
+                   + 1; // for allowFiltering
         }
     }
 }
diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java
index beb8280e674a..bd2d01051f02 100644
--- a/src/java/org/apache/cassandra/index/Index.java
+++ b/src/java/org/apache/cassandra/index/Index.java
@@ -807,6 +807,17 @@ Indexer indexerFor(Predicate<Index> indexSelector,
         @Nullable
         QueryPlan queryPlanFor(RowFilter rowFilter);
 
+        /**
+         * Returns a new {@link QueryPlan} for the specified {@link RowFilter} and set of {@link Index}, or {@code null}
+         * if none of the indexes in this group supports the expression in the row filter.
+         *
+         * @param rowFilter a row filter
+         * @param indexes a set of indexes
+         * @return a new query plan for the specified {@link RowFilter} and {@link Index}, {@code null} otherwise
+         */
+        @Nullable
+        QueryPlan queryPlanForIndices(RowFilter rowFilter, Set<Index> indexes);
+
         /**
          * Get flush observer to observe partition/cell events generated by flushing SSTable (memtable flush or compaction).
          *
diff --git a/src/java/org/apache/cassandra/index/IndexRegistry.java b/src/java/org/apache/cassandra/index/IndexRegistry.java
index 639d9aec350b..aab44179afd7 100644
--- a/src/java/org/apache/cassandra/index/IndexRegistry.java
+++ b/src/java/org/apache/cassandra/index/IndexRegistry.java
@@ -244,6 +244,13 @@ public Index.QueryPlan queryPlanFor(RowFilter rowFilter)
                 return null;
             }
 
+            @Nullable
+            @Override
+            public Index.QueryPlan queryPlanForIndices(RowFilter rowFilter, Set<Index> indexes)
+            {
+                return null;
+            }
+
             @Nullable
             @Override
             public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata, long keyCount)
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
index b1930e741804..638b75d762d1 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
@@ -125,6 +125,7 @@
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.Indexes;
+import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.service.pager.SinglePartitionPager;
 import org.apache.cassandra.tracing.Tracing;
@@ -134,6 +135,7 @@
 import org.apache.cassandra.utils.concurrent.Refs;
 import org.json.simple.JSONValue;
 
+import static java.lang.String.format;
 import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination;
 import static org.apache.cassandra.utils.ExecutorUtils.shutdown;
 
@@ -172,7 +174,7 @@
  * <br><br>
  * Finally, this class provides a clear and safe lifecycle to manage index builds, either full rebuilds via
  * {@link this#rebuildIndexesBlocking(Set)} or builds of new sstables
- * added via {@link org.apache.cassandra.notifications.SSTableAddedNotification}s, guaranteeing
+ * added via {@link SSTableAddedNotification}s, guaranteeing
  * the following:
  * <ul>
  * <li>The initialization task and any subsequent successful (re)build mark the index as built.</li>
@@ -185,6 +187,7 @@
 public class SecondaryIndexManager implements IndexRegistry, INotificationConsumer
 {
     private static final Logger logger = LoggerFactory.getLogger(SecondaryIndexManager.class);
+    public static final String FELL_BACK_TO_ALLOW_FILTERING = "Query fell back to ALLOW FILTERING because index %s is still building.";
 
     // default page size (in rows) when rebuilding the index for a whole partition
     public static final int DEFAULT_PAGE_SIZE = 10000;
@@ -362,13 +365,24 @@ public synchronized Future<?> addIndex(IndexMetadata indexDef, boolean isNewCF)
      * @param queryPlan a query plan
      * @throws IndexNotAvailableException if the query plan has any index that is not queryable
      */
-    public void checkQueryability(Index.QueryPlan queryPlan)
+    public boolean isQueryableThroughIndex(Index.QueryPlan queryPlan, boolean allowsFiltering)
     {
         for (Index index : queryPlan.getIndexes())
         {
+            if (isIndexBuilding(index))
+              if (allowsFiltering)
+              {
+                  ClientWarn.instance.warn(format(SecondaryIndexManager.FELL_BACK_TO_ALLOW_FILTERING, index.getIndexMetadata().name));
+                  logger.warn(format(SecondaryIndexManager.FELL_BACK_TO_ALLOW_FILTERING, index.getIndexMetadata().name));
+                  return false;
+              }
+
+            // We will reject the query here if the index is building and ALLOW FILTERING is not allowed
             if (!isIndexQueryable(index))
                 throw new IndexNotAvailableException(index);
         }
+
+        return true;
     }
 
     /**
@@ -382,6 +396,17 @@ public boolean isIndexQueryable(Index index)
         return isIndexQueryable(index.getIndexMetadata().name);
     }
 
+    /**
+     +     * Checks if the specified index is building.
+     +     *
+     +     * @param index the index
+     +     * @return <code>true</code> if the specified index is building, <code>false</code> otherwise
+     +     */
+    public boolean isIndexBuilding(Index index)
+    {
+            return isIndexBuilding(index.getIndexMetadata().name);
+    }
+
     /**
      * Checks if the specified index is queryable.
      *
@@ -616,7 +641,7 @@ private Future<?> buildIndexesAsync(Collection<SSTableReader> sstables, Set<Inde
                     sstables.stream().map(SSTableReader::toString).collect(Collectors.joining(",")));
 
         // Group all building tasks
-        Map<Index.IndexBuildingSupport, Set<Index>> byType = new HashMap<>();
+        Map<IndexBuildingSupport, Set<Index>> byType = new HashMap<>();
         for (Index index : indexes)
         {
             IndexBuildingSupport buildOrRecoveryTask = isFullRebuild
@@ -1168,7 +1193,7 @@ public void deletePartition(UnfilteredRowIterator partition, int nowInSec)
     /**
      * Called at query time to choose which (if any) of the registered index implementations to use for a given query.
      * <p>
-     * This is a two step processes, firstly compiling the set of searchable indexes then choosing the one which reduces
+     * This is a two-step processes, firstly compiling the set of searchable indexes then choosing the one which reduces
      * the search space the most.
      * <p>
      * In the first phase, if the command's RowFilter contains any custom index expressions, the indexes that they
@@ -1814,13 +1839,16 @@ public static void shutdownAndWait(long timeout, TimeUnit units) throws Interrup
      * @param indexQueryPlan index query plan used in the read command
      * @param level consistency level of read command
      */
-    public static <E extends Endpoints<E>> E filterForQuery(E liveEndpoints, Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel level)
+    public static <E extends Endpoints<E>> E filterForQuery(E liveEndpoints, Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel level, boolean allowsFiltering)
     {
         E queryableEndpoints = liveEndpoints.filter(replica -> {
-
             for (Index index : indexQueryPlan.getIndexes())
             {
                 Index.Status status = getIndexStatus(replica.endpoint(), keyspace.getName(), index.getIndexMetadata().name);
+                // if the status of the index is building and there is allow filtering - that is ok too
+                if (status == Index.Status.FULL_REBUILD_STARTED && allowsFiltering)
+                    continue;
+
                 if (!index.isQueryable(status))
                     return false;
             }
diff --git a/src/java/org/apache/cassandra/index/SingletonIndexGroup.java b/src/java/org/apache/cassandra/index/SingletonIndexGroup.java
index c6d5d735f1d8..d232b5d02fb4 100644
--- a/src/java/org/apache/cassandra/index/SingletonIndexGroup.java
+++ b/src/java/org/apache/cassandra/index/SingletonIndexGroup.java
@@ -72,7 +72,7 @@ public void addIndex(Index index)
         // This class does not work for SAI because the `componentsForNewBuid` method would be incorrect (so more
         // generally, it does not work for indexes that use dedicated sstable components, which is only SAI). See
         // comments on `componentsForNewBuid` for more details.
-        Preconditions.checkState(!(index instanceof StorageAttachedIndex), "This shoudl not be used with SAI");
+        Preconditions.checkState(!(index instanceof StorageAttachedIndex), "This should not be used with SAI");
         delegate = index;
         indexes.add(index);
     }
@@ -120,6 +120,25 @@ public Index.QueryPlan queryPlanFor(RowFilter rowFilter)
         return null;
     }
 
+    @Override
+    public Index.QueryPlan queryPlanForIndices(RowFilter rowFilter, Set<Index> indexes)
+    {
+        Preconditions.checkNotNull(delegate);
+
+        if (indexes.contains(delegate))
+            return null;
+
+        // Indexes using a singleton group don't support disjunctions,
+        // so we only consider the top-level AND expressions for index selection.
+        for (RowFilter.Expression e : rowFilter.withoutDisjunctions().expressions())
+        {
+            if (delegate.supportsExpression(e.column(), e.operator()))
+                return new SingletonIndexQueryPlan(delegate, delegate.getPostIndexQueryFilter(rowFilter));
+        }
+
+        return null;
+    }
+
     @Override
     public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata, long keyCount)
     {
diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
index d6cf0af7a78f..4e947afd45e6 100644
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
@@ -240,6 +240,22 @@ public StorageAttachedIndexQueryPlan queryPlanFor(RowFilter rowFilter)
         return StorageAttachedIndexQueryPlan.create(baseCfs, queryMetrics, indices, rowFilter);
     }
 
+    @Override
+    public StorageAttachedIndexQueryPlan queryPlanForIndices(RowFilter rowFilter, Set<Index> indexes)
+    {
+        Set<StorageAttachedIndex> saiIndexes = new HashSet<>();
+        for (Index index : indexes)
+        {
+            assert index instanceof StorageAttachedIndex : "Index is not a StorageAttachedIndex";
+            saiIndexes.add((StorageAttachedIndex) index);
+        }
+
+        // Verify that all provided indexes belong to this group
+        assert indexes.size() < indices.size() && indices.containsAll(saiIndexes);
+
+        return StorageAttachedIndexQueryPlan.create(baseCfs, queryMetrics, saiIndexes, rowFilter);
+    }
+
     @Override
     public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata, long keyCount)
     {
diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java
index 2c4186a89bd6..b70d90c0d65f 100644
--- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java
+++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java
@@ -659,11 +659,11 @@ public static ReplicaPlan.ForPaxosWrite forPaxos(Keyspace keyspace, DecoratedKey
     }
 
 
-    private static <E extends Endpoints<E>> E candidatesForRead(Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, E liveNaturalReplicas)
+    private static <E extends Endpoints<E>> E candidatesForRead(Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, E liveNaturalReplicas, boolean allowsFiltering)
     {
         E replicas = consistencyLevel.isDatacenterLocal() ? liveNaturalReplicas.filter(InOurDcTester.replicas()) : liveNaturalReplicas;
 
-        return indexQueryPlan != null ? SecondaryIndexManager.filterForQuery(replicas, keyspace, indexQueryPlan, consistencyLevel) : replicas;
+        return indexQueryPlan != null ? SecondaryIndexManager.filterForQuery(replicas, keyspace, indexQueryPlan, consistencyLevel, allowsFiltering) : replicas;
     }
 
     private static <E extends Endpoints<E>> E contactForEachQuorumRead(NetworkTopologyStrategy replicationStrategy, E candidates)
@@ -723,11 +723,11 @@ public static ReplicaPlan.ForRangeRead forSingleReplicaRead(Keyspace keyspace, A
      * The candidate collection can be used for speculation, although at present
      * it would break EACH_QUORUM to do so without further filtering
      */
-    public static ReplicaPlan.ForTokenRead forRead(Keyspace keyspace, Token token, Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, SpeculativeRetryPolicy retry)
+    public static ReplicaPlan.ForTokenRead forRead(Keyspace keyspace, Token token, Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, SpeculativeRetryPolicy retry, boolean allowsFiltering)
     {
         AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy();
         IEndpointSnitch endpointSnitch = DatabaseDescriptor.getEndpointSnitch();
-        EndpointsForToken candidates = candidatesForRead(keyspace, indexQueryPlan, consistencyLevel, ReplicaLayout.forTokenReadLiveSorted(replicationStrategy, token).natural())
+        EndpointsForToken candidates = candidatesForRead(keyspace, indexQueryPlan, consistencyLevel, ReplicaLayout.forTokenReadLiveSorted(replicationStrategy, token).natural(), allowsFiltering)
                                        .filter(endpointSnitch.filterByAffinity(keyspace.getName()));
         EndpointsForToken contacts = contactForRead(replicationStrategy, consistencyLevel, retry.equals(AlwaysSpeculativeRetryPolicy.INSTANCE), candidates)
                                      .filter(endpointSnitch.filterByAffinity(keyspace.getName()));
@@ -743,11 +743,11 @@ public static ReplicaPlan.ForTokenRead forRead(Keyspace keyspace, Token token, I
      *
      * There is no speculation for range read queries at present, so we never 'always speculate' here, and a failed response fails the query.
      */
-    public static ReplicaPlan.ForRangeRead forRangeRead(Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, AbstractBounds<PartitionPosition> range, int vnodeCount)
+    public static ReplicaPlan.ForRangeRead forRangeRead(Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, AbstractBounds<PartitionPosition> range, int vnodeCount, boolean allowsFiltering)
     {
         AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy();
         IEndpointSnitch endpointSnitch = DatabaseDescriptor.getEndpointSnitch();
-        EndpointsForRange candidates = candidatesForRead(keyspace, indexQueryPlan, consistencyLevel, ReplicaLayout.forRangeReadLiveSorted(replicationStrategy, range).natural())
+        EndpointsForRange candidates = candidatesForRead(keyspace, indexQueryPlan, consistencyLevel, ReplicaLayout.forRangeReadLiveSorted(replicationStrategy, range).natural(), allowsFiltering)
                                        .filter(endpointSnitch.filterByAffinity(keyspace.getName()));
         EndpointsForRange contacts = contactForRead(replicationStrategy, consistencyLevel, false, candidates)
                                      .filter(endpointSnitch.filterByAffinity(keyspace.getName()));
diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
index cb0018975e9e..acf0ddc9c83a 100644
--- a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
+++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
@@ -204,7 +204,7 @@ public static AbstractReadExecutor getReadExecutor(SinglePartitionReadCommand co
         ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(command.metadata().id);
         SpeculativeRetryPolicy retry = cfs.metadata().params.speculativeRetry;
 
-        ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forRead(keyspace, command.partitionKey().getToken(), command.indexQueryPlan(), consistencyLevel, retry);
+        ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forRead(keyspace, command.partitionKey().getToken(), command.indexQueryPlan(), consistencyLevel, retry, command.rowFilter().allowFiltering);
 
         if (replicaPlan.candidates().stream().noneMatch(replica -> replica.endpoint().equals(FBUtilities.getBroadcastAddressAndPort())))
         {
diff --git a/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java b/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java
index 63291c34a943..c592bbc6a435 100644
--- a/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java
+++ b/src/java/org/apache/cassandra/service/reads/range/RangeCommands.java
@@ -76,7 +76,8 @@ static RangeCommandIterator rangeCommandIterator(PartitionRangeReadCommand comma
         ReplicaPlanIterator replicaPlans = new ReplicaPlanIterator(command.dataRange().keyRange(),
                                                                    command.indexQueryPlan(),
                                                                    keyspace,
-                                                                   consistencyLevel);
+                                                                   consistencyLevel,
+                                                                   command.rowFilter().allowFiltering);
         if (command.isTopK())
             return new ScanAllRangesCommandIterator(keyspace, replicaPlans, command, replicaPlans.size(), queryStartNanoTime, readTracker);
 
diff --git a/src/java/org/apache/cassandra/service/reads/range/ReplicaPlanIterator.java b/src/java/org/apache/cassandra/service/reads/range/ReplicaPlanIterator.java
index 605afa4db632..405de0e3e6be 100644
--- a/src/java/org/apache/cassandra/service/reads/range/ReplicaPlanIterator.java
+++ b/src/java/org/apache/cassandra/service/reads/range/ReplicaPlanIterator.java
@@ -36,7 +36,6 @@
 import org.apache.cassandra.locator.ReplicaPlan;
 import org.apache.cassandra.locator.ReplicaPlans;
 import org.apache.cassandra.locator.TokenMetadata;
-import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.AbstractIterator;
 import org.apache.cassandra.utils.Pair;
 
@@ -45,6 +44,7 @@ class ReplicaPlanIterator extends AbstractIterator<ReplicaPlan.ForRangeRead>
     private final Keyspace keyspace;
     private final ConsistencyLevel consistency;
     private final Index.QueryPlan indexQueryPlan;
+    private final boolean allowsFiltering;
     @VisibleForTesting
     final Iterator<? extends AbstractBounds<PartitionPosition>> ranges;
     private final int rangeCount;
@@ -52,11 +52,13 @@ class ReplicaPlanIterator extends AbstractIterator<ReplicaPlan.ForRangeRead>
     ReplicaPlanIterator(AbstractBounds<PartitionPosition> keyRange,
                         Index.QueryPlan indexQueryPlan,
                         Keyspace keyspace,
-                        ConsistencyLevel consistency)
+                        ConsistencyLevel consistency,
+                        boolean allowsFiltering)
     {
         this.indexQueryPlan = indexQueryPlan;
         this.keyspace = keyspace;
         this.consistency = consistency;
+        this.allowsFiltering = allowsFiltering;
 
 
         List<? extends AbstractBounds<PartitionPosition>> l = keyspace.getReplicationStrategy() instanceof LocalStrategy
@@ -80,7 +82,7 @@ protected ReplicaPlan.ForRangeRead computeNext()
         if (!ranges.hasNext())
             return endOfData();
 
-        return ReplicaPlans.forRangeRead(keyspace, indexQueryPlan, consistency, ranges.next(), 1);
+        return ReplicaPlans.forRangeRead(keyspace, indexQueryPlan, consistency, ranges.next(), 1, allowsFiltering);
     }
 
     /**
@@ -102,7 +104,7 @@ private static List<AbstractBounds<PartitionPosition>> getRestrictedRanges(Token
         while (ringIter.hasNext())
         {
             /*
-             * remainder is a range/bounds of partition positions and we want to split it with a token. We want to split
+             * remainder is a range/bounds of partition positions, and we want to split it with a token. We want to split
              * using the key returned by token.maxKeyBound. For instance, if remainder is [DK(10, 'foo'), DK(20, 'bar')],
              * and we have 3 nodes with tokens 0, 15, 30, we want to split remainder to A=[DK(10, 'foo'), 15] and
              * B=(15, DK(20, 'bar')]. But since we can't mix tokens and keys at the same time in a range, we use
diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/ANNOptionsDistributedTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/ANNOptionsDistributedTest.java
index 26052606a5e6..592d06120bd2 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/sai/ANNOptionsDistributedTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/sai/ANNOptionsDistributedTest.java
@@ -22,6 +22,7 @@
 import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
 import net.bytebuddy.implementation.MethodDelegation;
 import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.db.filter.ANNOptions;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.api.ICoordinator;
@@ -70,7 +71,7 @@ public void testANNOptionsWithAllDS10() throws Throwable
                                            .withConfig(config -> config.with(GOSSIP).with(NETWORK))
                                            .start(), RF))
         {
-            test(cluster, "ANN options are not supported in clusters below DS 11.");
+            test(cluster, ANNOptions.REQUIRES_HIGHER_MESSAGING_VERSION);
         }
     }
 
@@ -85,7 +86,7 @@ public void testANNOptionsWithMixedDS10AndDS11() throws Throwable
                                            .withConfig(config -> config.with(GOSSIP).with(NETWORK).with(NATIVE_PROTOCOL))
                                            .start(), RF))
         {
-            test(cluster, "ANN options are not supported in clusters below DS 11.");
+            test(cluster, ANNOptions.REQUIRES_HIGHER_MESSAGING_VERSION);
         }
     }
 
diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
index 5f5a4918b95f..3268328028f9 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
@@ -21,8 +21,10 @@
 import java.net.InetAddress;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -32,6 +34,7 @@
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.distributed.Cluster;
@@ -50,6 +53,7 @@
 import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
 import static org.apache.cassandra.distributed.api.Feature.NETWORK;
 import static org.apache.cassandra.distributed.test.sai.SAIUtil.waitForIndexQueryable;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
 import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -61,7 +65,7 @@ public class IndexAvailabilityTest extends TestBaseImpl
                                                "WITH compaction = {'class' : 'SizeTieredCompactionStrategy', 'enabled' : false }";
     private static final String CREATE_INDEX = "CREATE CUSTOM INDEX %s ON %s.%s(%s) USING 'StorageAttachedIndex'";
     
-    private static Map<NodeIndex, Index.Status> expectedNodeIndexQueryability = new ConcurrentHashMap<>();
+    private static final Map<NodeIndex, Index.Status> expectedNodeIndexQueryability = new ConcurrentHashMap<>();
     private List<String> keyspaces;
     private List<String> indexesPerKs;
 
@@ -134,7 +138,7 @@ public void verifyIndexStatusPropagation() throws Exception
             assertIndexingStatus(cluster);
 
             // drop ks2 index2, there should be no ks2 index2 status on all node
-            cluster.schemaChange("DROP INDEX " + ks2 + "." + index2);
+            cluster.schemaChange("DROP INDEX " + ks2 + '.' + index2);
             expectedNodeIndexQueryability.keySet().forEach(k -> {
                 if (k.keyspace.equals(ks2) && k.index.equals(index2))
                     expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
@@ -142,7 +146,7 @@ public void verifyIndexStatusPropagation() throws Exception
             assertIndexingStatus(cluster);
 
             // drop ks3 cf1, there should be no ks3 index1/index2 status
-            cluster.schemaChange("DROP TABLE " + ks3 + "." + cf1);
+            cluster.schemaChange("DROP TABLE " + ks3 + '.' + cf1);
             expectedNodeIndexQueryability.keySet().forEach(k -> {
                 if (k.keyspace.equals(ks3))
                     expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
@@ -234,7 +238,196 @@ private void shouldSkipNonQueryableNode(int nodes, List<Integer>... nonQueryable
         }
     }
 
-    private void executeOnAllCoordinatorsAllConsistencies(Cluster cluster, String statement, int liveReplicas, int num) throws Exception
+    @Test
+    public void testAllowFilteringDuringIndexBuildsOn3NodeCluster() throws Exception
+    {
+        try (Cluster cluster = init(Cluster.build(3)
+                                           .withConfig(config -> config.with(GOSSIP)
+                                                                       .with(NETWORK))
+                                           .start()))
+        {
+            String ks2 = "ks2";
+            String table = "tbl";
+            String index1 = "tbl_idx1";
+            String index2 = "tbl_idx2";
+            String vectorIndex = "tbl_vec_idx";
+
+            cluster.schemaChange(String.format(CREATE_KEYSPACE, ks2, 2));
+            cluster.schemaChange("CREATE TABLE " + ks2 + '.' + table +
+                                 " (pk text, i int, j int, k int, v1 int, v2 int, vec vector<float, 2>, " +
+                                 "PRIMARY KEY((pk, i), j))");
+
+            // Insert test data
+            cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s(pk, i, j, k, v1, v2, vec) VALUES " +
+                                                         "('partition1', 1, 100, 200, 0, 0, [0.5, 1.5])",
+                                                         ks2, table),
+                                           ConsistencyLevel.QUORUM);
+            cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s(pk, i, j, k, v1, v2, vec) VALUES " +
+                                                         "('partition2', 2, 101, 201, 1, 1, [1.5, 2.5])",
+                                                         ks2, table),
+                                           ConsistencyLevel.QUORUM);
+            cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s(pk, i, j, k, v1, v2, vec) VALUES " +
+                                                         "('partition3', 1, 102, 202, 0, 1, [2.5, 3.5])",
+                                                         ks2, table),
+                                           ConsistencyLevel.QUORUM);
+            cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s(pk, i, j, k, v1, v2, vec) VALUES " +
+                                                         "('partition4', 2, 103, 203, 1, 0, [3.5, 4.5])",
+                                                         ks2, table),
+                                           ConsistencyLevel.QUORUM);
+            cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s(pk, i, j, k, v1, v2, vec) VALUES " +
+                                                         "('partition5', 1, 104, 204, 0, 0, [4.5, 5.5])",
+                                                         ks2, table),
+                                           ConsistencyLevel.QUORUM);
+
+            // Test queries before any index exists
+            executeOnAllCoordinators(cluster,
+                                     "SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
+                                     ConsistencyLevel.LOCAL_QUORUM,
+                                     2);
+            // Verify actual results using a direct query
+            Object[][] results = cluster.coordinator(1)
+                                        .execute("SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
+                                                 ConsistencyLevel.LOCAL_QUORUM);
+            assertResultContains(results, Arrays.asList("partition1", "partition5"));
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining(StatementRestrictions.REQUIRES_ALLOW_FILTERING_MESSAGE);
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT * FROM " + ks2 + '.' + table + " ORDER BY vec ANN OF [1, 1] ALLOW FILTERING",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"));
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT pk FROM " + ks2 + '.' + table + " WHERE GEO_DISTANCE(vec, [1, 1]) < 1000",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining(StatementRestrictions.GEO_DISTANCE_REQUIRES_INDEX_MESSAGE);
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT pk FROM " + ks2 + '.' + table + " WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining(StatementRestrictions.GEO_DISTANCE_REQUIRES_INDEX_MESSAGE);
+
+            // Create and verify indexes
+            cluster.schemaChange(String.format("CREATE CUSTOM INDEX %s ON %s.%s(vec) USING 'StorageAttachedIndex'",
+                                               vectorIndex, ks2, table));
+            cluster.schemaChange(String.format(CREATE_INDEX, index1, ks2, table, "v1"));
+
+            cluster.forEach(node -> expectedNodeIndexQueryability.put(NodeIndex.create(ks2, index1, node), Index.Status.BUILD_SUCCEEDED));
+            cluster.forEach(node -> expectedNodeIndexQueryability.put(NodeIndex.create(ks2, vectorIndex, node), Index.Status.BUILD_SUCCEEDED));
+            waitForIndexQueryable(cluster, ks2);
+
+            executeOnAllCoordinators(cluster,
+                                     "SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
+                                     ConsistencyLevel.LOCAL_QUORUM,
+                                     2);
+            // Verify actual results using a direct query
+            results = cluster.coordinator(1)
+                                        .execute("SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
+                                                 ConsistencyLevel.LOCAL_QUORUM);
+            assertResultContains(results, Arrays.asList("partition1", "partition5"));
+
+            executeOnAllCoordinators(cluster,
+                                     "SELECT * FROM " + ks2 + '.' + table + " ORDER BY vec ANN OF [1, 1] LIMIT 10 ALLOW FILTERING",
+                                     ConsistencyLevel.LOCAL_QUORUM,
+                                     5);
+
+            // Create one more index but mark it as building
+            cluster.schemaChange(String.format(CREATE_INDEX, index2, ks2, table, "v2"));
+            markIndexBuilding(cluster.get(1), ks2, table, index2);
+            waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(3), ks2, index2, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
+            markIndexBuilding(cluster.get(2), ks2, table, index2);
+            waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(3), ks2, index2, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT * FROM " + ks2 + '.' + table +
+                                                        " WHERE k > 0 ORDER BY vec ANN OF [1.0, 1.0] LIMIT 10 ALLOW FILTERING",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE);
+
+            // Verify that the query works with ALLOW FILTERING
+            executeOnAllCoordinators(cluster,
+                                     "SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
+                                     ConsistencyLevel.LOCAL_QUORUM,
+                                     2);
+            // Verify actual results using a direct query
+            results = cluster.coordinator(1)
+                                        .execute("SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
+                                                 ConsistencyLevel.LOCAL_QUORUM);
+            assertResultContains(results, Arrays.asList("partition1", "partition5"));
+
+            // mark the vector index as building, we should not be able to query it yet
+            markIndexBuilding(cluster.get(1), ks2, table, vectorIndex);
+            waitForIndexingStatus(cluster.get(2), ks2, vectorIndex, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(3), ks2, vectorIndex, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(1), ks2, vectorIndex, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
+            markIndexBuilding(cluster.get(2), ks2, table, vectorIndex);
+            waitForIndexingStatus(cluster.get(2), ks2, vectorIndex, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(3), ks2, vectorIndex, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(1), ks2, vectorIndex, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT * FROM " + ks2 + '.' + table + " ORDER BY vec ANN OF [1, 1] LIMIT 10 ALLOW FILTERING",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining("Operation failed - received 0 responses and 2 failures: INDEX_NOT_AVAILABLE");
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT pk FROM " + ks2 + '.' + table + " WHERE GEO_DISTANCE(vec, [1, 1]) < 1000",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining(StatementRestrictions.VECTOR_INDEX_PRESENT_NOT_SUPPORT_GEO_DISTANCE_MESSAGE);
+
+            assertThatThrownBy(() ->
+                               executeOnAllCoordinators(cluster,
+                                                        "SELECT pk FROM " + ks2 + '.' + table + " WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING",
+                                                        ConsistencyLevel.LOCAL_QUORUM,
+                                                        0))
+            .hasMessageContaining(StatementRestrictions.VECTOR_INDEX_PRESENT_NOT_SUPPORT_GEO_DISTANCE_MESSAGE);
+
+            // drop the index that does not support GEO DISTANCE
+            cluster.schemaChange("DROP INDEX IF EXISTS " + ks2 + '.' + vectorIndex);
+
+            // create a new index that supports GEO DISTANCE
+            cluster.schemaChange(String.format("CREATE CUSTOM INDEX %s ON %s.%s(vec) USING 'StorageAttachedIndex' WITH OPTIONS = {'similarity_function' : 'euclidean'}",
+                                               vectorIndex, ks2, table));
+            markIndexQueryable(cluster.get(1), ks2, table, index2);
+            markIndexQueryable(cluster.get(2), ks2, table, index2);
+            cluster.forEach(node -> expectedNodeIndexQueryability.put(NodeIndex.create(ks2, index2, node), Index.Status.BUILD_SUCCEEDED));
+            cluster.forEach(node -> expectedNodeIndexQueryability.put(NodeIndex.create(ks2, vectorIndex, node), Index.Status.BUILD_SUCCEEDED));
+            waitForIndexQueryable(cluster, ks2);
+
+            executeOnAllCoordinators(cluster,
+                                     "SELECT pk FROM " + ks2 + '.' + table + " WHERE GEO_DISTANCE(vec, [1, 1]) < 1000",
+                                     ConsistencyLevel.LOCAL_QUORUM,
+                                     0);
+
+            executeOnAllCoordinators(cluster,
+                                     "SELECT pk FROM " + ks2 + '.' + table + " WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING",
+                                     ConsistencyLevel.LOCAL_QUORUM,
+                                     0);
+        }
+    }
+
+    private void executeOnAllCoordinatorsAllConsistencies(Cluster cluster, String statement, int liveReplicas, int num)
     {
         int allReplicas = cluster.size();
 
@@ -245,7 +438,7 @@ private void executeOnAllCoordinatorsAllConsistencies(Cluster cluster, String st
         executeOnAllCoordinators(cluster, statement, ConsistencyLevel.ALL, liveReplicas >= allReplicas ? num : -1);
     }
 
-    private void executeOnAllCoordinators(Cluster cluster, String query, ConsistencyLevel level, int expected) throws Exception
+    private void executeOnAllCoordinators(Cluster cluster, String query, ConsistencyLevel level, int expected)
     {
         // test different coordinator
         for (int nodeId = 1; nodeId <= cluster.size(); nodeId++)
@@ -261,13 +454,13 @@ private void executeOnAllCoordinators(Cluster cluster, String query, Consistency
                 }
                 catch (Throwable e)
                 {
-                    assertTrue(e.getClass().getSimpleName().equals("ReadFailureException"));
+                    assertEquals("ReadFailureException", e.getClass().getSimpleName());
                 }
             }
         }
     }
 
-    private void markIndexNonQueryable(IInvokableInstance node, String keyspace, String table, String indexName) throws Exception
+    private void markIndexNonQueryable(IInvokableInstance node, String keyspace, String table, String indexName)
     {
         expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.BUILD_FAILED);
 
@@ -278,7 +471,7 @@ private void markIndexNonQueryable(IInvokableInstance node, String keyspace, Str
         });
     }
 
-    private void markIndexQueryable(IInvokableInstance node, String keyspace, String table, String indexName) throws Exception
+    private void markIndexQueryable(IInvokableInstance node, String keyspace, String table, String indexName)
     {
         expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.BUILD_SUCCEEDED);
 
@@ -289,7 +482,7 @@ private void markIndexQueryable(IInvokableInstance node, String keyspace, String
         });
     }
 
-    private void markIndexBuilding(IInvokableInstance node, String keyspace, String table, String indexName) throws Exception
+    private void markIndexBuilding(IInvokableInstance node, String keyspace, String table, String indexName)
     {
         expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.FULL_REBUILD_STARTED);
 
@@ -345,7 +538,7 @@ private static void waitForIndexingStatus(IInvokableInstance node, String keyspa
     {
         InetAddressAndPort replicaAddressAndPort = getFullAddress(replica);
         await().atMost(5, TimeUnit.SECONDS)
-               .until(() -> node.callOnInstance(() -> getIndexStatus(keyspace, index, replicaAddressAndPort) == status).booleanValue());
+               .until(() -> node.callOnInstance(() -> getIndexStatus(keyspace, index, replicaAddressAndPort) == status));
     }
 
     private static Index.Status getNodeIndexStatus(IInvokableInstance node, String keyspaceName, String indexName, InetAddressAndPort replica)
@@ -374,6 +567,19 @@ private static InetAddressAndPort getFullAddress(IInvokableInstance node)
         int port = node.callOnInstance(() -> FBUtilities.getBroadcastAddressAndPort().port);
         return InetAddressAndPort.getByAddressOverrideDefaults(address, port);
     }
+
+    private void assertResultContains(Object[][] results, List<String> expectedValues)
+    {
+        Set<String> actualValues = new HashSet<>();
+        for (Object[] row : results)
+            actualValues.add((String) row[0]); // Assuming first column contains the values we're checking
+
+        assertEquals("Result set size doesn't match expected values",
+                     expectedValues.size(), actualValues.size());
+        assertTrue("Results don't contain all expected values: " +
+                   "Expected=" + expectedValues + ", Actual=" + actualValues,
+                   actualValues.containsAll(expectedValues));
+    }
     
     private static class NodeIndex
     {
diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
index 63bf2c8c6b22..4dfd8596948f 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
@@ -1156,6 +1156,8 @@ public void testIndexQueriesWithIndexNotReady()
         {
             execute("DROP index " + KEYSPACE + ".testIndex");
         }
+
+        execute("SELECT value FROM %s WHERE value = 2 ALLOW FILTERING");
     }
 
     @Test // A Bad init could leave an index only accepting reads
diff --git a/test/unit/org/apache/cassandra/db/filter/ANNOptionsTest.java b/test/unit/org/apache/cassandra/db/filter/ANNOptionsTest.java
index 791068fab39c..e14ad5363f80 100644
--- a/test/unit/org/apache/cassandra/db/filter/ANNOptionsTest.java
+++ b/test/unit/org/apache/cassandra/db/filter/ANNOptionsTest.java
@@ -20,6 +20,11 @@
 
 import javax.annotation.Nullable;
 
+import org.apache.cassandra.index.IndexNotAvailableException;
+import org.apache.cassandra.index.sai.SAITester;
+import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.inject.Injections;
+import org.apache.cassandra.inject.InvokePointBuilder;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -47,6 +52,7 @@
 import org.assertj.core.api.Assertions;
 import org.quicktheories.QuickTheory;
 
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
 import static org.quicktheories.generators.SourceDSL.integers;
 
 /**
diff --git a/test/unit/org/apache/cassandra/index/CustomIndexTest.java b/test/unit/org/apache/cassandra/index/CustomIndexTest.java
index f1155a82512a..7057533a5b2f 100644
--- a/test/unit/org/apache/cassandra/index/CustomIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/CustomIndexTest.java
@@ -1620,6 +1620,12 @@ public QueryPlan queryPlanFor(RowFilter rowFilter)
                 throw new UnsupportedOperationException();
             }
 
+            @Override
+            public QueryPlan queryPlanForIndices(RowFilter rowFilter, Set<Index> indexes)
+            {
+                throw new UnsupportedOperationException();
+            }
+
             @Override
             public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata, long keyCount)
             {
diff --git a/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java b/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java
index 181f74de95f4..4d4ec73696a7 100644
--- a/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java
@@ -17,12 +17,26 @@
  */
 package org.apache.cassandra.index.sai.cql;
 
+import java.util.Arrays;
+
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.db.filter.ANNOptions;
+import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
+import org.apache.cassandra.index.IndexNotAvailableException;
+import org.apache.cassandra.index.SecondaryIndexManager;
 import org.apache.cassandra.index.sai.SAITester;
 import org.apache.cassandra.index.sai.StorageAttachedIndex;
-
+import org.apache.cassandra.inject.Injections;
+import org.apache.cassandra.inject.InvokePointBuilder;
+import org.apache.cassandra.service.ClientWarn;
+import org.assertj.core.api.Assertions;
+import org.assertj.core.api.ListAssert;
+
+import static java.lang.String.format;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertNotNull;
 
 /**
@@ -30,6 +44,140 @@
  */
 public class AllowFilteringTest extends SAITester
 {
+    private final Injections.Barrier blockIndexBuild = Injections.newBarrier("block_index_build", 2, false)
+                                                                 .add(InvokePointBuilder.newInvokePoint().onClass(StorageAttachedIndex.class)
+                                                                                  .onMethod("startInitialBuild"))
+                                                                 .build();
+
+    @Test
+    public void testAllowFilteringDuringIndexBuild() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk text, i int, j int, k int, vec vector<float, 2>, PRIMARY KEY((pk, i), j))");
+        execute("INSERT INTO %s (pk, i, j, k, vec) VALUES ('partition1', 1, 100, 200, [0.5, 1.5])");
+        Injections.inject(blockIndexBuild);
+
+        // 1. Queries before index creation
+
+        assertInvalidThrowMessage(StatementRestrictions.REQUIRES_ALLOW_FILTERING_MESSAGE,
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE k=0");
+        execute("SELECT * FROM %s WHERE k=0 ALLOW FILTERING");
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1, 1] ALLOW FILTERING WITH ann_options = {}");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1, 1] ALLOW FILTERING");
+
+        String idx = createIndexAsync(String.format("CREATE CUSTOM INDEX ON %%s(k) USING '%s'", StorageAttachedIndex.class.getName()));
+        String idx2 = createIndexAsync(String.format("CREATE CUSTOM INDEX ON %%s(vec) USING '%s'", StorageAttachedIndex.class.getName()));
+
+        // 2. Queries during index build - as there is already index and AF is not an option,
+        // it just complains the index is not available from now on.
+
+        assertInvalidThrowMessage("The secondary index '" + idx + "' is not yet available",
+                                  IndexNotAvailableException.class,
+                                  "SELECT * FROM %s WHERE k=0");
+
+        assertInvalidThrowMessage(ANNOptions.REQUIRES_HIGHER_MESSAGING_VERSION,
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1, 1] ALLOW FILTERING WITH ann_options = {}");
+
+        assertInvalidThrowMessage("The secondary index '" + idx2 + "' is not yet available",
+                                  IndexNotAvailableException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1, 1] LIMIT 10 ALLOW FILTERING");
+
+        execute("SELECT * FROM %s WHERE k=0 ALLOW FILTERING");
+
+        // A warning was emitted because the index is not yet available
+        assertRows(row("partition1", 1, 100, 200, Arrays.asList(0.5f, 1.5f)))
+        .hasSize(1)
+        .contains(format(SecondaryIndexManager.FELL_BACK_TO_ALLOW_FILTERING, idx));
+
+        blockIndexBuild.countDown();
+        blockIndexBuild.disable();
+        waitForIndexQueryable(idx);
+        waitForIndexQueryable(idx2);
+
+        // 3. Queries after index creation
+
+        execute("SELECT * FROM %s WHERE k=0");
+        execute("SELECT * FROM %s WHERE k=0 ALLOW FILTERING");
+        assertInvalidThrowMessage(ANNOptions.REQUIRES_HIGHER_MESSAGING_VERSION,
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1, 1] ALLOW FILTERING WITH ann_options = {}");
+        execute("SELECT * FROM %s ORDER BY vec ANN OF [1, 1] LIMIT 10 ALLOW FILTERING");
+    }
+
+    private ListAssert<String> assertRows(Object[]... rows)
+    {
+        ClientWarn.instance.captureWarnings();
+        CQLTester.disablePreparedReuseForTest();
+        assertRows(execute("SELECT * FROM %s WHERE k=200 ALLOW FILTERING"), rows);
+        ListAssert<String> assertion = Assertions.assertThat(ClientWarn.instance.getWarnings());
+        ClientWarn.instance.resetWarnings();
+        return assertion;
+    }
+
+    @Test
+    public void testAllowFilteringGeoDistance() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk text, i int, j int, k int, vec vector<float, 2>, PRIMARY KEY((pk, i), j))");
+        Injections.inject(blockIndexBuild);
+
+        // 1. Queries before index creation
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000"))
+        .hasMessage(StatementRestrictions.GEO_DISTANCE_REQUIRES_INDEX_MESSAGE)
+        .isInstanceOf(InvalidRequestException.class);
+
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING"))
+        .hasMessage(StatementRestrictions.GEO_DISTANCE_REQUIRES_INDEX_MESSAGE)
+        .isInstanceOf(InvalidRequestException.class);
+
+        String idx = createIndexAsync(String.format("CREATE CUSTOM INDEX ON %%s(vec) USING '%s'", StorageAttachedIndex.class.getName()));
+
+        // 2. Queries during not applicable index build.
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING"))
+        .hasMessage(StatementRestrictions.VECTOR_INDEX_PRESENT_NOT_SUPPORT_GEO_DISTANCE_MESSAGE)
+        .isInstanceOf(InvalidRequestException.class);
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING"))
+        .hasMessage(StatementRestrictions.VECTOR_INDEX_PRESENT_NOT_SUPPORT_GEO_DISTANCE_MESSAGE)
+        .isInstanceOf(InvalidRequestException.class);
+
+        execute(String.format("DROP INDEX IF EXISTS %s.%s", KEYSPACE, idx));
+
+        // 3. Queries during valid index build - as there is already index and AF is not an option,
+        // it just complains the index is not available from now on.
+        idx = createIndexAsync("CREATE CUSTOM INDEX ON %s(vec) USING 'StorageAttachedIndex' WITH OPTIONS = {'similarity_function' : 'euclidean'}");
+
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000"))
+        .hasMessage(String.format("The secondary index '" + idx + "' is not yet available"))
+        .isInstanceOf(IndexNotAvailableException.class);
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000"))
+        .hasMessage(String.format("The secondary index '" + idx + "' is not yet available"))
+        .isInstanceOf(IndexNotAvailableException.class);
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING"))
+        .hasMessage(String.format("The secondary index '" + idx + "' is not yet available"))
+        .isInstanceOf(IndexNotAvailableException.class);
+        assertThatThrownBy(() -> execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING"))
+        .hasMessage(String.format("The secondary index '" + idx + "' is not yet available"))
+        .isInstanceOf(IndexNotAvailableException.class);
+
+        // 4. Queries after index creation
+
+        blockIndexBuild.countDown();
+        blockIndexBuild.disable();
+        waitForIndexQueryable(idx);
+
+        execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000");
+        execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000");
+
+        // ALLOW FILTERING is here, but we use the index
+        execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING");
+        execute( "SELECT pk FROM %s WHERE GEO_DISTANCE(vec, [1, 1]) < 1000 ALLOW FILTERING");
+    }
+
     @Test
     public void testAllowFilteringOnFirstClusteringKeyColumn()
     {
@@ -366,17 +514,17 @@ public void testUnsupportedIndexRestrictions()
         execute("INSERT INTO %s (a, b, c, d) VALUES ('Test3', 'Test3', 'Test3', 'Test3')");
 
         // Single restriction
-        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, "b"), "SELECT * FROM %s WHERE b > 'Test'");
-        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, "c"), "SELECT * FROM %s WHERE c > 'Test'");
-        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, "d"), "SELECT * FROM %s WHERE d > 'Test'");
+        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, 'b'), "SELECT * FROM %s WHERE b > 'Test'");
+        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, 'c'), "SELECT * FROM %s WHERE c > 'Test'");
+        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, 'd'), "SELECT * FROM %s WHERE d > 'Test'");
 
         // Supported and unsupported restriction
-        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, "b"), "SELECT * FROM %s WHERE b > 'Test' AND c = 'Test1'");
-        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, "c"), "SELECT * FROM %s WHERE c > 'Test' AND d = 'Test1'");
-        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, "d"), "SELECT * FROM %s WHERE d > 'Test' AND b = 'Test1'");
+        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, 'b'), "SELECT * FROM %s WHERE b > 'Test' AND c = 'Test1'");
+        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, 'c'), "SELECT * FROM %s WHERE c > 'Test' AND d = 'Test1'");
+        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, 'd'), "SELECT * FROM %s WHERE d > 'Test' AND b = 'Test1'");
 
         // Two unsupported restrictions
-        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, "b"), "SELECT * FROM %s WHERE b > 'Test' AND b < 'Test3'");
+        assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_SINGLE, 'b'), "SELECT * FROM %s WHERE b > 'Test' AND b < 'Test3'");
         assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_MULTI, "[b, c]"), "SELECT * FROM %s WHERE c > 'Test' AND b < 'Test3'");
         assertInvalidMessage(String.format(StatementRestrictions.HAS_UNSUPPORTED_INDEX_RESTRICTION_MESSAGE_MULTI, "[b, d]"), "SELECT * FROM %s WHERE d > 'Test' AND b < 'Test3'");
 
@@ -416,9 +564,9 @@ public void testIndexedColumnDoesNotSupportLikeRestriction()
         createIndex(String.format("CREATE CUSTOM INDEX ON %%s(d) USING '%s'", StorageAttachedIndex.class.getName()));
 
         // LIKE restriction
-        assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_LIKE_MESSAGE, "b"), "SELECT * FROM %s WHERE b LIKE 'Test'");
-        assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_LIKE_MESSAGE, "c"), "SELECT * FROM %s WHERE c LIKE 'Test'");
-        assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_LIKE_MESSAGE, "d"), "SELECT * FROM %s WHERE d LIKE 'Test'");
+        assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_LIKE_MESSAGE, 'b'), "SELECT * FROM %s WHERE b LIKE 'Test'");
+        assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_LIKE_MESSAGE, 'c'), "SELECT * FROM %s WHERE c LIKE 'Test'");
+        assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_LIKE_MESSAGE, 'd'), "SELECT * FROM %s WHERE d LIKE 'Test'");
     }
 
     @Test
@@ -430,26 +578,124 @@ public void testIndexedColumnDoesNotSupportAnalyzerRestriction()
         createIndex(String.format("CREATE CUSTOM INDEX ON %%s(d) USING '%s'", StorageAttachedIndex.class.getName()));
 
         // Analyzer restriction
-        assertInvalidMessage(String.format(": restriction is only supported on properly indexed columns. a : 'Test' is not valid."), "SELECT * FROM %s WHERE a : 'Test'");
+        assertInvalidMessage(": restriction is only supported on properly indexed columns. a : 'Test' is not valid.", "SELECT * FROM %s WHERE a : 'Test'");
         assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_ANALYZER_MATCHES_MESSAGE, 'b'), "SELECT * FROM %s WHERE b : 'Test'");
         assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_ANALYZER_MATCHES_MESSAGE, 'c'), "SELECT * FROM %s WHERE c : 'Test'");
         assertInvalidMessage(String.format(StatementRestrictions.INDEX_DOES_NOT_SUPPORT_ANALYZER_MATCHES_MESSAGE, 'd'), "SELECT * FROM %s WHERE d : 'Test'");
     }
 
     @Test
-    public void testQueryRequiresFilteringButHasANNRestriction()
+    public void testQueryRequiresFilteringButHasANNRestriction() throws Throwable
     {
         createTable("CREATE TABLE %s (pk text, i int, j int, k int, vec vector<float, 3>, PRIMARY KEY((pk, i), j))");
-        createIndex("CREATE CUSTOM INDEX ON %s(vec) USING 'StorageAttachedIndex'");
+        Injections.inject(blockIndexBuild);
+
+        // 1. Queries before index creation
+
+        // Ordering on non-clustering column vec requires the column to be indexed.
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1,1,1] LIMIT 10;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1,1,1] LIMIT 10 ALLOW FILTERING;");
+
+        // Below changes after we add index to different requirement, not sure if we want/can easily improve that
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                             "SELECT * FROM %s WHERE k > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE k > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE k > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE j > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE j > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' AND i > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        assertInvalidThrowMessage(String.format(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, "vec"),
+                                  InvalidRequestException.class,
+                                  "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' AND i > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        // 2. Queries during index build - as there is already index and AF is not an option,
+        // it just complains the index is not available from now on.
+
+        String idx = createIndexAsync("CREATE CUSTOM INDEX ON %s(vec) USING 'StorageAttachedIndex'");
+
+        assertInvalidThrowMessage(String.format("The secondary index '" + idx + "' is not yet available"), IndexNotAvailableException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1,1,1] LIMIT 10;");
+
+        // Should not fail because allow filtering is set but not required
+        assertInvalidThrowMessage(String.format("The secondary index '" + idx + "' is not yet available"), IndexNotAvailableException.class,
+                                  "SELECT * FROM %s ORDER BY vec ANN OF [1,1,1] LIMIT 10 ALLOW FILTERING;");
+
+        // Do not recommend ALLOW FILTERING for non-primary key, non clustering column restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE k > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        // Do not let ALLOW FILTERING to lead to query execution for non-primary key, non clustering column restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE k > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        // Do not recommend ALLOW FILTERING for clustering column restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE j > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        // Do not let ALLOW FILTERING lead to query execution for clustering column restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE j > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        // Do not recommend ALLOW FILTERING for partial partition key restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        // Do not let ALLOW FILTERING lead to query execution for partial partition key restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        // Do not recommend ALLOW FILTERING for complete partition key restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' AND i > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
+
+        // Do not let ALLOW FILTERING lead to query execution for complete partition key restrictions
+        assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
+                             "SELECT * FROM %s WHERE pk > 'A' AND pk < 'C' AND i > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
+
+        blockIndexBuild.countDown();
+        blockIndexBuild.disable();
+        waitForIndexQueryable(idx);
+
+        // 3. Queries after index creation
 
         // Should not fail because allow filtering is set but not required
         assertRows(execute("SELECT * FROM %s ORDER BY vec ANN OF [1,1,1] LIMIT 10 ALLOW FILTERING;"));
 
-        // Do not recommend ALLOW FILTERING for non primary key, non clustering column restrictions
+        // Do not recommend ALLOW FILTERING for non-primary key, non clustering column restrictions
         assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
                              "SELECT * FROM %s WHERE k > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10;");
 
-        // Do not let ALLOW FILTERING to lead to query execution for non primary key, non clustering column restrictions
+        // Do not let ALLOW FILTERING to lead to query execution for non-primary key, non clustering column restrictions
         assertInvalidMessage(StatementRestrictions.NON_CLUSTER_ORDERING_REQUIRES_ALL_RESTRICTED_NON_PARTITION_KEY_COLUMNS_INDEXED_MESSAGE,
                              "SELECT * FROM %s WHERE k > 0 ORDER BY vec ANN OF [2.5, 3.5, 4.5] LIMIT 10 ALLOW FILTERING;");
 
diff --git a/test/unit/org/apache/cassandra/locator/AssureSufficientLiveNodesTest.java b/test/unit/org/apache/cassandra/locator/AssureSufficientLiveNodesTest.java
index 2b84a5f3cf31..3cfa97f1face 100644
--- a/test/unit/org/apache/cassandra/locator/AssureSufficientLiveNodesTest.java
+++ b/test/unit/org/apache/cassandra/locator/AssureSufficientLiveNodesTest.java
@@ -150,7 +150,7 @@ public void addDatacenterShouldNotCausesUnavailableWithEachQuorumTest() throws T
             // alter to
             KeyspaceParams.nts(DC1, 3, DC2, 3),
             // test
-            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, EACH_QUORUM, NeverSpeculativeRetryPolicy.INSTANCE)
+            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, EACH_QUORUM, NeverSpeculativeRetryPolicy.INSTANCE, false)
         );
     }
 
@@ -183,7 +183,7 @@ public void addDatacenterShouldNotCausesUnavailableWithQuorumTest() throws Throw
             // alter to
             KeyspaceParams.nts(DC1, 3, DC2, 3),
             // test
-            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, QUORUM, NeverSpeculativeRetryPolicy.INSTANCE)
+            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, QUORUM, NeverSpeculativeRetryPolicy.INSTANCE, false)
         );
         raceOfReplicationStrategyTest(
             // init. The # of live endpoints is 3 = 2 + 1
@@ -191,7 +191,7 @@ public void addDatacenterShouldNotCausesUnavailableWithQuorumTest() throws Throw
             // alter to. (3 + 3) / 2 + 1 > 3
             KeyspaceParams.nts(DC1, 2, DC2, 1, DC3, 3),
             // test
-            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, QUORUM, NeverSpeculativeRetryPolicy.INSTANCE)
+            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, QUORUM, NeverSpeculativeRetryPolicy.INSTANCE, false)
         );
     }
 
@@ -215,7 +215,7 @@ public void raceOnRemoveDatacenterNotCausesUnavailable() throws Throwable
             // alter to
             KeyspaceParams.nts(DC1, 3),
             // test
-            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, EACH_QUORUM, NeverSpeculativeRetryPolicy.INSTANCE)
+            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, EACH_QUORUM, NeverSpeculativeRetryPolicy.INSTANCE, false)
         );
     }
 
@@ -239,7 +239,7 @@ public void increaseReplicationFactorShouldNotCausesUnavailableTest() throws Thr
             // alter to
             KeyspaceParams.nts(DC1, 3),
             // test
-            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, LOCAL_QUORUM, NeverSpeculativeRetryPolicy.INSTANCE)
+            keyspace -> ReplicaPlans.forRead(keyspace, tk, null, LOCAL_QUORUM, NeverSpeculativeRetryPolicy.INSTANCE, false)
         );
     }
 
diff --git a/test/unit/org/apache/cassandra/locator/ReplicaPlansTest.java b/test/unit/org/apache/cassandra/locator/ReplicaPlansTest.java
index 486db68b5073..a0bedf122646 100644
--- a/test/unit/org/apache/cassandra/locator/ReplicaPlansTest.java
+++ b/test/unit/org/apache/cassandra/locator/ReplicaPlansTest.java
@@ -122,11 +122,11 @@ public void testFilterByAffinity() throws UnknownHostException
         Token token = new Murmur3Partitioner.LongToken(0);
 
         Keyspace keyspace1 = keyspaceWithSnitch(KEYSPACE1, filterByKeyspaceAffinitySnitch);
-        ReplicaPlan.ForTokenRead plan1 = ReplicaPlans.forRead(keyspace1, token, null, ANY, NeverSpeculativeRetryPolicy.INSTANCE);
+        ReplicaPlan.ForTokenRead plan1 = ReplicaPlans.forRead(keyspace1, token, null, ANY, NeverSpeculativeRetryPolicy.INSTANCE, false);
         assertEndpointsMatchKeyspaceAffinity(KEYSPACE1, plan1.contacts());
 
         Keyspace keyspace2 = keyspaceWithSnitch(KEYSPACE2, filterByKeyspaceAffinitySnitch);
-        ReplicaPlan.ForTokenRead plan2 = ReplicaPlans.forRead(keyspace2, token, null, ANY, NeverSpeculativeRetryPolicy.INSTANCE);
+        ReplicaPlan.ForTokenRead plan2 = ReplicaPlans.forRead(keyspace2, token, null, ANY, NeverSpeculativeRetryPolicy.INSTANCE, false);
         assertEndpointsMatchKeyspaceAffinity(KEYSPACE2, plan2.contacts());
     }
 
diff --git a/test/unit/org/apache/cassandra/service/reads/range/EndpointGroupingRangeCommandIteratorTest.java b/test/unit/org/apache/cassandra/service/reads/range/EndpointGroupingRangeCommandIteratorTest.java
index 0ffcc79038cd..1ffad426a947 100644
--- a/test/unit/org/apache/cassandra/service/reads/range/EndpointGroupingRangeCommandIteratorTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/range/EndpointGroupingRangeCommandIteratorTest.java
@@ -41,7 +41,6 @@
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.locator.ReplicaPlan;
 import org.apache.cassandra.schema.KeyspaceParams;
-import org.apache.cassandra.service.QueryInfoTracker;
 import org.apache.cassandra.utils.CloseableIterator;
 
 import static org.apache.cassandra.service.QueryInfoTracker.*;
@@ -155,7 +154,7 @@ private static NonGroupingRangeCommandIterator tokenOrderIterator(PartitionRange
     private static CloseableIterator<ReplicaPlan.ForRangeRead> replicaPlanIterator(PartitionRangeReadCommand command)
     {
         AbstractBounds<PartitionPosition> keyRange = command.dataRange().keyRange();
-        CloseableIterator<ReplicaPlan.ForRangeRead> replicaPlans = new ReplicaPlanIterator(keyRange, null, keyspace, ConsistencyLevel.ONE);
+        CloseableIterator<ReplicaPlan.ForRangeRead> replicaPlans = new ReplicaPlanIterator(keyRange, null, keyspace, ConsistencyLevel.ONE, command.rowFilter().allowFiltering);
         return new ReplicaPlanMerger(replicaPlans, keyspace, ConsistencyLevel.ONE);
     }
 }
diff --git a/test/unit/org/apache/cassandra/service/reads/range/RangeCommandIteratorTest.java b/test/unit/org/apache/cassandra/service/reads/range/RangeCommandIteratorTest.java
index 6b313d41e1fd..a93d5a74db7a 100644
--- a/test/unit/org/apache/cassandra/service/reads/range/RangeCommandIteratorTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/range/RangeCommandIteratorTest.java
@@ -71,7 +71,7 @@ public void testRangeCountWithRangeMerge()
         for (int i = 0; i + 1 < tokens.size(); i++)
         {
             Range<PartitionPosition> range = Range.makeRowRange(tokens.get(i), tokens.get(i + 1));
-            ranges.add(ReplicaPlans.forRangeRead(keyspace, null, ConsistencyLevel.ONE, range, 1));
+            ranges.add(ReplicaPlans.forRangeRead(keyspace, null, ConsistencyLevel.ONE, range, 1, false));
             vnodeCount++;
         }
 
@@ -165,7 +165,7 @@ private static CloseableIterator<ReplicaPlan.ForRangeRead> replicaPlanIterator(A
                                                                                    Keyspace keyspace,
                                                                                    boolean withRangeMerger)
     {
-        CloseableIterator<ReplicaPlan.ForRangeRead> replicaPlans = new ReplicaPlanIterator(keyRange, null, keyspace, ConsistencyLevel.ONE);
+        CloseableIterator<ReplicaPlan.ForRangeRead> replicaPlans = new ReplicaPlanIterator(keyRange, null, keyspace, ConsistencyLevel.ONE, false);
         if (withRangeMerger)
             replicaPlans = new ReplicaPlanMerger(replicaPlans, keyspace, ConsistencyLevel.ONE);
 
diff --git a/test/unit/org/apache/cassandra/service/reads/range/RangeCommandsTest.java b/test/unit/org/apache/cassandra/service/reads/range/RangeCommandsTest.java
index f64e446e32b8..ff683276f37e 100644
--- a/test/unit/org/apache/cassandra/service/reads/range/RangeCommandsTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/range/RangeCommandsTest.java
@@ -38,7 +38,6 @@
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.index.StubIndex;
 import org.apache.cassandra.schema.IndexMetadata;
-import org.apache.cassandra.service.QueryInfoTracker;
 
 import static org.apache.cassandra.db.ConsistencyLevel.ONE;
 import static org.apache.cassandra.service.QueryInfoTracker.*;
@@ -77,7 +76,7 @@ public void tesConcurrencyFactor()
         // verify that a low concurrency factor is not capped by the max concurrency factor
         PartitionRangeReadCommand command = command(cfs, 50, 50);
         try (RangeCommandIterator partitions = RangeCommands.rangeCommandIterator(command, ONE, System.nanoTime(), ReadTracker.NOOP);
-             ReplicaPlanIterator ranges = new ReplicaPlanIterator(command.dataRange().keyRange(), command.indexQueryPlan(), keyspace, ONE))
+             ReplicaPlanIterator ranges = new ReplicaPlanIterator(command.dataRange().keyRange(), command.indexQueryPlan(), keyspace, ONE, false))
         {
             assertEquals(2, partitions.concurrencyFactor());
             assertEquals(MAX_CONCURRENCY_FACTOR, partitions.maxConcurrencyFactor());
@@ -87,7 +86,7 @@ public void tesConcurrencyFactor()
         // verify that a high concurrency factor is capped by the max concurrency factor
         command = command(cfs, 1000, 50);
         try (RangeCommandIterator partitions = RangeCommands.rangeCommandIterator(command, ONE, System.nanoTime(), ReadTracker.NOOP);
-             ReplicaPlanIterator ranges = new ReplicaPlanIterator(command.dataRange().keyRange(), command.indexQueryPlan(), keyspace, ONE))
+             ReplicaPlanIterator ranges = new ReplicaPlanIterator(command.dataRange().keyRange(), command.indexQueryPlan(), keyspace, ONE, command.rowFilter().allowFiltering))
         {
             assertEquals(MAX_CONCURRENCY_FACTOR, partitions.concurrencyFactor());
             assertEquals(MAX_CONCURRENCY_FACTOR, partitions.maxConcurrencyFactor());
@@ -97,7 +96,7 @@ public void tesConcurrencyFactor()
         // with 0 estimated results per range the concurrency factor should be 1
         command = command(cfs, 1000, 0);
         try (RangeCommandIterator partitions = RangeCommands.rangeCommandIterator(command, ONE, System.nanoTime(), ReadTracker.NOOP);
-             ReplicaPlanIterator ranges = new ReplicaPlanIterator(command.dataRange().keyRange(), command.indexQueryPlan(), keyspace, ONE))
+             ReplicaPlanIterator ranges = new ReplicaPlanIterator(command.dataRange().keyRange(), command.indexQueryPlan(), keyspace, ONE, command.rowFilter().allowFiltering))
         {
             assertEquals(1, partitions.concurrencyFactor());
             assertEquals(MAX_CONCURRENCY_FACTOR, partitions.maxConcurrencyFactor());
diff --git a/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanIteratorTest.java b/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanIteratorTest.java
index 84f3a5e2e750..ea278328c19a 100644
--- a/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanIteratorTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanIteratorTest.java
@@ -163,7 +163,7 @@ private final void testRanges(AbstractBounds<PartitionPosition> queryRange, Abst
     @SafeVarargs
     private final void testRanges(Keyspace keyspace, AbstractBounds<PartitionPosition> queryRange, AbstractBounds<PartitionPosition>... expected)
     {
-        try (ReplicaPlanIterator iterator = new ReplicaPlanIterator(queryRange, null, keyspace, ConsistencyLevel.ANY))
+        try (ReplicaPlanIterator iterator = new ReplicaPlanIterator(queryRange, null, keyspace, ConsistencyLevel.ANY, false))
         {
             List<AbstractBounds<PartitionPosition>> restrictedRanges = new ArrayList<>(expected.length);
             while (iterator.hasNext())
diff --git a/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanMergerTest.java b/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanMergerTest.java
index b577678c3ba0..fce5b7d0a850 100644
--- a/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanMergerTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/range/ReplicaPlanMergerTest.java
@@ -392,7 +392,7 @@ private final void testRanges(ConsistencyLevel consistencyLevel,
                                   AbstractBounds<PartitionPosition> queryRange,
                                   AbstractBounds<PartitionPosition>... expected)
     {
-        try (ReplicaPlanIterator originals = new ReplicaPlanIterator(queryRange, null, keyspace, ANY); // ANY avoids endpoint erros
+        try (ReplicaPlanIterator originals = new ReplicaPlanIterator(queryRange, null, keyspace, ANY, false); // ANY avoids endpoint erros
              ReplicaPlanMerger merger = new ReplicaPlanMerger(originals, keyspace, consistencyLevel))
         {
             // collect the merged ranges

From 9ed9b6158553806ab126b72291cef2ee5906a3f0 Mon Sep 17 00:00:00 2001
From: Ekaterina Dimitrova <ekaterina.dimitrova@datastax.com>
Date: Sun, 9 Feb 2025 12:55:40 -0500
Subject: [PATCH 2/3] Empty-Commit


From 71383786913af8dcde381dd8605470dd49ca026b Mon Sep 17 00:00:00 2001
From: Ekaterina Dimitrova <ekaterina.dimitrova@datastax.com>
Date: Sun, 9 Feb 2025 16:15:58 -0500
Subject: [PATCH 3/3] Make testAllowFilteringDuringIndexBuildsOn3NodeCluster
 deterministic

---
 .../test/sai/IndexAvailabilityTest.java       | 30 ++++++++++++++++---
 1 file changed, 26 insertions(+), 4 deletions(-)

diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
index 3268328028f9..4af27f60a0df 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
@@ -251,6 +251,8 @@ public void testAllowFilteringDuringIndexBuildsOn3NodeCluster() throws Exception
             String index1 = "tbl_idx1";
             String index2 = "tbl_idx2";
             String vectorIndex = "tbl_vec_idx";
+            keyspaces = List.of(ks2);
+            indexesPerKs = Arrays.asList(index1, index2, vectorIndex);
 
             cluster.schemaChange(String.format(CREATE_KEYSPACE, ks2, 2));
             cluster.schemaChange("CREATE TABLE " + ks2 + '.' + table +
@@ -331,6 +333,7 @@ public void testAllowFilteringDuringIndexBuildsOn3NodeCluster() throws Exception
                                      "SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
                                      ConsistencyLevel.LOCAL_QUORUM,
                                      2);
+
             // Verify actual results using a direct query
             results = cluster.coordinator(1)
                                         .execute("SELECT pk FROM " + ks2 + '.' + table + " WHERE v1=0 AND v2=0 ALLOW FILTERING",
@@ -352,6 +355,11 @@ public void testAllowFilteringDuringIndexBuildsOn3NodeCluster() throws Exception
             waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
             waitForIndexingStatus(cluster.get(3), ks2, index2, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
             waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+            markIndexBuilding(cluster.get(3), ks2, table, index2);
+            waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(3), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(3), ks2, index2, cluster.get(3), Index.Status.FULL_REBUILD_STARTED);
+            waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(3), Index.Status.FULL_REBUILD_STARTED);
+            assertIndexingStatus(cluster);
 
             assertThatThrownBy(() ->
                                executeOnAllCoordinators(cluster,
@@ -381,6 +389,7 @@ public void testAllowFilteringDuringIndexBuildsOn3NodeCluster() throws Exception
             waitForIndexingStatus(cluster.get(2), ks2, vectorIndex, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
             waitForIndexingStatus(cluster.get(3), ks2, vectorIndex, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
             waitForIndexingStatus(cluster.get(1), ks2, vectorIndex, cluster.get(2), Index.Status.FULL_REBUILD_STARTED);
+            assertIndexingStatus(cluster);
 
             assertThatThrownBy(() ->
                                executeOnAllCoordinators(cluster,
@@ -405,15 +414,28 @@ public void testAllowFilteringDuringIndexBuildsOn3NodeCluster() throws Exception
 
             // drop the index that does not support GEO DISTANCE
             cluster.schemaChange("DROP INDEX IF EXISTS " + ks2 + '.' + vectorIndex);
+            expectedNodeIndexQueryability.keySet().forEach(k -> {
+                if (k.keyspace.equals(ks2) && k.index.equals(vectorIndex))
+                    expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
+            });
+            assertIndexingStatus(cluster);
 
             // create a new index that supports GEO DISTANCE
             cluster.schemaChange(String.format("CREATE CUSTOM INDEX %s ON %s.%s(vec) USING 'StorageAttachedIndex' WITH OPTIONS = {'similarity_function' : 'euclidean'}",
                                                vectorIndex, ks2, table));
-            markIndexQueryable(cluster.get(1), ks2, table, index2);
-            markIndexQueryable(cluster.get(2), ks2, table, index2);
-            cluster.forEach(node -> expectedNodeIndexQueryability.put(NodeIndex.create(ks2, index2, node), Index.Status.BUILD_SUCCEEDED));
             cluster.forEach(node -> expectedNodeIndexQueryability.put(NodeIndex.create(ks2, vectorIndex, node), Index.Status.BUILD_SUCCEEDED));
-            waitForIndexQueryable(cluster, ks2);
+            waitForIndexingStatus(cluster.get(2), ks2, vectorIndex, cluster.get(1), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(3), ks2, vectorIndex, cluster.get(1), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(1), ks2, vectorIndex, cluster.get(1), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(2), ks2, vectorIndex, cluster.get(2), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(3), ks2, vectorIndex, cluster.get(2), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(1), ks2, vectorIndex, cluster.get(2), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(2), ks2, vectorIndex, cluster.get(3), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(3), ks2, vectorIndex, cluster.get(3), Index.Status.BUILD_SUCCEEDED);
+            waitForIndexingStatus(cluster.get(1), ks2, vectorIndex, cluster.get(3), Index.Status.BUILD_SUCCEEDED);
+
+            // other indexes or keyspaces should not be affected
+            assertIndexingStatus(cluster);
 
             executeOnAllCoordinators(cluster,
                                      "SELECT pk FROM " + ks2 + '.' + table + " WHERE GEO_DISTANCE(vec, [1, 1]) < 1000",