Extend the exclusion of replica filtering protection to other indices instead of just SASI

patch by Stefan Miklosovic; reviewed by Andrés de la Peña and Zhao Yang for CASSANDRA-16311#

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0.24:
+ * Extend the exclusion of replica filtering protection to other indices instead of just SASI (CASSANDRA-16311)
  * Synchronize transaction logs for JBOD (CASSANDRA-16225)
  * Fix the counting of cells per partition (CASSANDRA-16259)
  * Fix serial read/non-applying CAS linearizability (CASSANDRA-12126)

src/java/org/apache/cassandra/index/Index.java

@@ -432,6 +432,23 @@ public interface Index
     {
     }
 
+    /**
+     * Tells whether this index supports replica fitering protection or not.
+     *
+     * Replica filtering protection might need to run the query row filter in the coordinator to detect stale results.
+     * An index implementation will be compatible with this protection mechanism if it returns the same results for the
+     * row filter as CQL will return with {@code ALLOW FILTERING} and without using the index. This means that index
+     * implementations using custom query syntax or applying transformations to the indexed data won't support it.
+     * See CASSANDRA-8272 for further details.
+     *
+     * @param rowFilter rowFilter of query to decide if it supports replica filtering protection or not
+     * @return true if this index supports replica filtering protection, false otherwise
+     */
+    default boolean supportsReplicaFilteringProtection(RowFilter rowFilter)
+    {
+        return true;
+    }
+
     /**
      * Return a function which performs post processing on the results of a partition range read command.
      * In future, this may be used as a generalized mechanism for transforming results on the coordinator prior

src/java/org/apache/cassandra/service/DataResolver.java

@@ -39,7 +39,9 @@ import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.ExcludingBounds;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.exceptions.ReadTimeoutException;
+import org.apache.cassandra.index.Index;
 import org.apache.cassandra.net.*;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.FBUtilities;
 
@@ -94,7 +96,25 @@ public class DataResolver extends ResponseResolver
 
     private boolean needsReplicaFilteringProtection()
     {
-        return !command.rowFilter().isEmpty();
+        if (command.rowFilter().isEmpty())
+            return false;
+
+        IndexMetadata indexMetadata = command.indexMetadata();
+
+        if (indexMetadata == null || !indexMetadata.isCustom())
+        {
+            return true;
+        }
+
+        ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(command.metadata().cfId);
+
+        assert cfs != null;
+
+        Index index = command.getIndex(cfs);
+
+        assert index != null;
+
+        return index.supportsReplicaFilteringProtection(command.rowFilter());
     }
 
     private class ResolveContext