Minor improvements and hardening for IndexHints

- Enforce a more reasonable limit on the number of included/excluded indexes
- Serialize vints rather than shorts in IndexSetSerializer
- Return Iterable from notExcluded() to avoid set creation
- Avoid redundant iteration in MessagingService#endpointsWithConnectionsOnVersionBelow()

patch by Caleb Rackliffe; reviewed by Marcus Eriksson for CASSANDRA-20888

Author: maedhroz

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Minor improvements and hardening for IndexHints (CASSANDRA-20888)
  * Stop repair scheduler if two major versions are detected (CASSANDRA-20048)
  * Optimize audit logic for batch operations especially when audit is not enabled for DML (CASSANDRA-20885)
  * Implement nodetool history (CASSANDRA-20851)

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -5906,6 +5906,11 @@ public static int getSaiSSTableIndexesPerQueryFailThreshold()
         return conf.sai_sstable_indexes_per_query_fail_threshold;
     }
 
+    public static int getSecondaryIndexesPerTableFailThreshold()
+    {
+        return conf.secondary_indexes_per_table_fail_threshold;
+    }
+
     @VisibleForTesting
     public static void setTriggersPolicy(Config.TriggersPolicy policy)
     {

src/java/org/apache/cassandra/cql3/restrictions/MergedRestriction.java

@@ -282,10 +282,12 @@ public boolean needsFiltering(Index.Group indexGroup, IndexHints indexHints)
     {
         // multiple contains might require filtering on some indexes, since that is equivalent to a disjunction (or)
         boolean hasMultipleContains = containsCount > 1;
-        Set<Index> nonExcluded = indexHints.nonExcluded(indexGroup.getIndexes());
 
-        for (Index index : nonExcluded)
+        for (Index index : indexGroup.getIndexes())
         {
+            if (indexHints.excludes(index))
+                continue;
+
             if (isSupportedBy(index) && !(hasMultipleContains && index.filtersMultipleContains()))
                 return false;
         }

src/java/org/apache/cassandra/cql3/restrictions/SimpleRestriction.java

@@ -22,7 +22,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 import com.google.common.collect.RangeSet;
@@ -172,7 +171,7 @@ public void addFunctionsTo(List<Function> functions)
     @Override
     public boolean needsFiltering(Index.Group indexGroup, IndexHints indexHints)
     {
-        Set<Index> nonExcluded = indexHints.nonExcluded(indexGroup.getIndexes());
+        Iterable<Index> nonExcluded = indexHints.nonExcluded(indexGroup.getIndexes());
         for (ColumnMetadata column : columns())
         {
             if (!isSupportedBy(nonExcluded, column))

src/java/org/apache/cassandra/db/filter/IndexHints.java

@@ -32,6 +32,7 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.QualifiedName;
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.exceptions.InvalidRequestException;
@@ -44,6 +45,7 @@
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.vint.VIntCoding;
 
 import static java.lang.String.format;
 
@@ -57,7 +59,7 @@ public class IndexHints
     public static final String WRONG_KEYSPACE_ERROR = "Index %s is not in the same keyspace as the queried table.";
     public static final String MISSING_INDEX_ERROR = "Table %s doesn't have an index named %s";
     public static final String NON_INCLUDABLE_INDEXES_ERROR = "It's not possible to use all the specified included indexes with this query.";
-    public static final String TOO_MANY_INDEXES_ERROR = format("Cannot have more than %d included/excluded indexes, found ", Short.MAX_VALUE);
+    public static final String TOO_MANY_INDEXES_ERROR = "Cannot have more than 'secondary_indexes_per_table_fail_threshold' included/excluded indexes, found ";
 
     public static final IndexHints NONE = new IndexHints(Collections.emptySet(), Collections.emptySet())
     {
@@ -169,8 +171,11 @@ public boolean excludes(String indexName)
      * @param indexes a set of indexes
      * @return the indexes that are not excluded by these hints
      */
-    public <T extends Index> Set<T> nonExcluded(Iterable<T> indexes)
+    public <T extends Index> Iterable<T> nonExcluded(Iterable<T> indexes)
     {
+        if (excluded.isEmpty())
+            return indexes;
+
         Set<T> result = new HashSet<>();
         for (T index : indexes)
         {
@@ -370,10 +375,10 @@ public static IndexHints fromCQLNames(Set<QualifiedName> included,
                                           TableMetadata table,
                                           IndexRegistry indexRegistry)
     {
-        if (included != null && included.size() > Short.MAX_VALUE)
+        if (included != null && included.size() > maxIncludedOrExcludedIndexCount())
             throw new InvalidRequestException(TOO_MANY_INDEXES_ERROR + included.size());
 
-        if (excluded != null && excluded.size() > Short.MAX_VALUE)
+        if (excluded != null && excluded.size() > maxIncludedOrExcludedIndexCount())
             throw new InvalidRequestException(TOO_MANY_INDEXES_ERROR + excluded.size());
 
         IndexHints hints = IndexHints.create(fetchIndexes(included, table, indexRegistry),
@@ -399,6 +404,14 @@ public static IndexHints fromCQLNames(Set<QualifiedName> included,
         return hints;
     }
 
+    private static int maxIncludedOrExcludedIndexCount()
+    {
+        int guardrail = DatabaseDescriptor.getSecondaryIndexesPerTableFailThreshold();
+
+        // If no guardrail is configured, use a value that safely fits in a single byte for serialization:
+        return guardrail > 0 ? guardrail : 128;
+    }
+
     private static Set<IndexMetadata> fetchIndexes(Set<QualifiedName> indexNames, TableMetadata table, IndexRegistry indexRegistry)
     {
         if (indexNames == null || indexNames.isEmpty())
@@ -573,16 +586,16 @@ private void serialize(Set<IndexMetadata> indexes, DataOutputPlus out, int versi
                 return;
 
             int n = indexes.size();
-            assert n < Short.MAX_VALUE : TOO_MANY_INDEXES_ERROR + n;
+            assert n < maxIncludedOrExcludedIndexCount() : TOO_MANY_INDEXES_ERROR + n;
 
-            out.writeShort(n);
+            out.writeVInt32(n);
             for (IndexMetadata index : indexes)
                 IndexMetadata.serializer.serialize(index, out, version);
         }
 
         private Set<IndexMetadata> deserialize(DataInputPlus in, int version, TableMetadata table) throws IOException
         {
-            short n = in.readShort();
+            int n = in.readVInt32();
             Set<IndexMetadata> indexes = new HashSet<>(n);
             for (short i = 0; i < n; i++)
             {
@@ -597,7 +610,7 @@ private long serializedSize(Set<IndexMetadata> indexes, int version)
             if (indexes.isEmpty())
                 return 0;
 
-            long size = TypeSizes.SHORT_SIZE;
+            long size = VIntCoding.computeVIntSize(indexes.size());
             for (IndexMetadata index : indexes)
                 size += IndexMetadata.serializer.serializedSize(index, version);
             return size;

src/java/org/apache/cassandra/net/MessagingService.java

@@ -788,13 +788,26 @@ public Set<InetAddressAndPort> endpointsWithConnectionsOnVersionBelow(int versio
         Set<InetAddressAndPort> nodes = new HashSet<>();
         for (InetAddressAndPort node : ClusterMetadata.current().directory.allAddresses())
         {
-            ConnectionType.MESSAGING_TYPES.forEach(type -> {
-                OutboundConnections connections = getOutbound(node, false);
-                OutboundConnection connection = connections != null ? connections.connectionFor(type) : null;
-                if (connection != null && connection.messagingVersion() < version)
-                    nodes.add(node);
-            });
+            if (hasConnectionWithVersionBelow(node, version))
+                nodes.add(node);
         }
         return nodes;
     }
+
+    private boolean hasConnectionWithVersionBelow(InetAddressAndPort node, int version)
+    {
+        OutboundConnections connections = getOutbound(node, false);
+
+        if (connections == null)
+            return false;
+
+        for (ConnectionType type : ConnectionType.MESSAGING_TYPES)
+        {
+            OutboundConnection connection = connections.connectionFor(type);
+            if (connection != null && connection.messagingVersion() < version)
+                return true;
+        }
+
+        return false;
+    }
 }