CNDB-13925: Prefer not analyzed indexes for contains queries (#1718)

Prefer not-analyzed indexes over analyzed indexes for contains queries,
so they have a deterministic behaviour.
Also, emit a client warning when a not-analyzed index is selected over
an analyzed index.

Otherwise, different points in the codebase will make different,
pseudo-random decisions about what index should be used for a certain
contains expression, leading to erratic behaviour.

Author: adelapena

src/java/org/apache/cassandra/cql3/Operator.java

@@ -575,6 +575,16 @@ public boolean isSlice()
         return this == LT || this == LTE || this == GT || this == GTE;
     }
 
+    /**
+     * Checks if this operator is any of the variations of contains ({@code [NOT] CONTAINS [KEY]}).
+     *
+     * @return {@code true} if this operator is any kind of contains operator, {@code false} otherwise.
+     */
+    public boolean isContains()
+    {
+        return this == CONTAINS || this == CONTAINS_KEY || this == NOT_CONTAINS || this == NOT_CONTAINS_KEY;
+    }
+
     @Override
     public String toString()
     {

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

@@ -38,6 +38,7 @@
 import org.apache.cassandra.index.IndexRegistry;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.serializers.ListSerializer;
+import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.transport.ProtocolVersion;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.Pair;
@@ -564,6 +565,8 @@ public String toString()
     // This holds CONTAINS, CONTAINS_KEY, NOT CONTAINS, NOT CONTAINS KEY and map[key] = value restrictions because we might want to have any combination of them.
     public static final class ContainsRestriction extends SingleColumnRestriction
     {
+        public static final String MULTIPLE_INDEXES_WARNING = "Multiple indexes found for CONTAINS restriction on %s. Using not-analyzed index %s";
+
         private final List<Term> values = new ArrayList<>(); // for CONTAINS
         private final List<Term> negativeValues = new ArrayList<>(); // for NOT_CONTAINS
         private final List<Term> keys = new ArrayList<>(); // for CONTAINS_KEY
@@ -732,6 +735,38 @@ public int numberOfNegativeEntries()
             return negativeEntryKeys.size();
         }
 
+        @Override
+        public Index findSupportingIndex(IndexRegistry indexRegistry)
+        {
+            // if there are multiple supporting indexes, we prefer those without an analyzer (see CNDB-13925)
+            Index notAnalyzedIndex = null;
+            Index analyzedIndex = null;
+            for (Index index : indexRegistry.listIndexes())
+            {
+                if (isSupportedBy(index))
+                {
+                    if (index.getAnalyzer(null).isPresent())
+                        analyzedIndex = index;
+                    else
+                        notAnalyzedIndex = index;
+                }
+            }
+
+            if (notAnalyzedIndex != null)
+            {
+                // We prefer the not analyzed index, but if there was also an analyzed index, we warn the user.
+                // We use a client warning key so the warning is emitted just once per query.
+                if (analyzedIndex != null)
+                {
+                    String msg = String.format(MULTIPLE_INDEXES_WARNING, columnDef.name, notAnalyzedIndex.getIndexMetadata().name);
+                    ClientWarn.instance.warn(msg, "multiple_indexes_for_contains_on_" + columnDef.name);
+                }
+                return notAnalyzedIndex;
+            }
+
+            return analyzedIndex;
+        }
+
         @Override
         public void addFunctionsTo(List<Function> functions)
         {

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

@@ -1033,4 +1033,41 @@ enum Status
         BUILD_SUCCEEDED,
         DROPPED
     }
+
+    /**
+     * Returns the best index for a given column and operator among the specified collection of indexes.
+     * </p>
+     * The best index is simply the first one that supports the column and operator, because we shouldn't have multiple
+     * indexes for the same column and operator, so we can just return the first one. The only exception to this is when
+     * the operator is {@code [NOT] CONTAINS [KEY]}, in which case the best index be the first one that doesn't use an
+     * analyzer.
+     *
+     * @param indexes a collection on indexes
+     * @param column a column
+     * @param operator an operator
+     * @return the best index for the column and operator among the specified indexes
+     */
+    static <T extends Index> Optional<T> getBestIndexFor(Collection<T> indexes, ColumnMetadata column, Operator operator)
+    {
+        // we simply return the first index that supports the expression, unless it's a contains operator
+        if (!operator.isContains())
+            return indexes.stream().filter((i) -> i.supportsExpression(column, operator)).findFirst();
+
+        // if we have a contains operator, we prefer indexes without an analyzer (see CNDB-13925)
+        T firstAnalyzedIndex = null;
+        for (T index : indexes)
+        {
+            if (index.supportsExpression(column, operator))
+            {
+                // we prefer indexes without an analyzer
+                if (index.getAnalyzer(null).isEmpty())
+                    return Optional.of(index);
+
+                if (firstAnalyzedIndex == null)
+                    firstAnalyzedIndex = index;
+            }
+        }
+
+        return Optional.ofNullable(firstAnalyzedIndex);
+    }
 }

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

@@ -26,7 +26,6 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.Callable;
-import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
@@ -95,7 +94,7 @@ public Index getIndex(IndexMetadata indexMetadata)
         }
 
         @Override
-        public Optional<Index> getBestIndexFor(RowFilter.Expression expression)
+        public Optional<Index> getBestIndexFor(ColumnMetadata column, Operator operator)
         {
             return Optional.empty();
         }
@@ -283,7 +282,8 @@ public Collection<Index.Group> listIndexGroups()
             return Collections.singletonList(group);
         }
 
-        public Optional<Index> getBestIndexFor(RowFilter.Expression expression)
+        @Override
+        public Optional<Index> getBestIndexFor(ColumnMetadata column, Operator operator)
         {
             return Optional.empty();
         }
@@ -306,19 +306,15 @@ default void registerIndex(Index index)
 
     default Optional<Index.Analyzer> getAnalyzerFor(ColumnMetadata column, Operator operator, ByteBuffer value)
     {
-        for (Index index : listIndexes())
-        {
-            if (index.supportsExpression(column, operator))
-            {
-                Optional<Index.Analyzer> analyzer = index.getAnalyzer(value);
-                if (analyzer.isPresent())
-                    return analyzer;
-            }
-        }
-        return Optional.empty();
+        return getBestIndexFor(column, operator).flatMap(i -> i.getAnalyzer(value));
+    }
+
+    default Optional<Index> getBestIndexFor(RowFilter.Expression expression)
+    {
+        return getBestIndexFor(expression.column(), expression.operator());
     }
 
-    Optional<Index> getBestIndexFor(RowFilter.Expression expression);
+    Optional<Index> getBestIndexFor(ColumnMetadata column, Operator operator);
 
     /**
      * Called at write time to ensure that values present in the update

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

@@ -71,6 +71,7 @@
 import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.cql3.PageSize;
 import org.apache.cassandra.cql3.statements.schema.IndexTarget;
 import org.apache.cassandra.db.Clustering;
@@ -1261,19 +1262,20 @@ private static String commaSeparated(Collection<Index> indexes)
         return indexes.stream().map(i -> i.getIndexMetadata().name).collect(Collectors.joining(","));
     }
 
-
-    public Optional<Index> getBestIndexFor(RowFilter.Expression expression)
+    @Override
+    public Optional<Index> getBestIndexFor(ColumnMetadata column, Operator operator)
     {
-        return indexes.values().stream().filter((i) -> i.supportsExpression(expression.column(), expression.operator())).findFirst();
+        return Index.getBestIndexFor(indexes.values(), column, operator);
     }
 
     public <T extends Index> Optional<T> getBestIndexFor(RowFilter.Expression expression, Class<T> indexType)
     {
-        return indexes.values()
-                      .stream()
-                      .filter(i -> indexType.isInstance(i) && i.supportsExpression(expression.column(), expression.operator()))
-                      .map(indexType::cast)
-                      .findFirst();
+        Set<T> candidates = indexes.values()
+                                   .stream()
+                                   .filter(indexType::isInstance)
+                                   .map(indexType::cast)
+                                   .collect(Collectors.toSet());
+        return Index.getBestIndexFor(candidates, expression.column(), expression.operator());
     }
 
     /**

src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexQueryPlan.java

@@ -18,6 +18,7 @@
 package org.apache.cassandra.index.sai.plan;
 
 import java.util.HashSet;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -162,16 +163,14 @@ private static boolean selectedIndexes(RowFilter.Expression expression,
         if (expression.isUserDefined())
             return false;
 
-        boolean hasIndex = false;
-        for (StorageAttachedIndex index : allIndexes)
+        // collect the best index from those that support the specified expression
+        Optional<StorageAttachedIndex> index = Index.getBestIndexFor(allIndexes, expression.column(), expression.operator());
+        if (index.isPresent())
         {
-            if (index.supportsExpression(expression.column(), expression.operator()))
-            {
-                selectedIndexes.add(index);
-                hasIndex = true;
-            }
+            selectedIndexes.add(index.get());
+            return true;
         }
-        return hasIndex;
+        return false;
     }
 
     @Override
@@ -196,7 +195,7 @@ public boolean shouldEstimateInitialConcurrency()
     }
 
     @Override
-    public Index.Searcher searcherFor(ReadCommand command)
+    public StorageAttachedIndexSearcher searcherFor(ReadCommand command)
     {
         return new StorageAttachedIndexSearcher(cfs,
                                                 queryMetrics,

src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java

@@ -28,11 +28,13 @@
 import java.util.Map;
 import java.util.PriorityQueue;
 import java.util.Queue;
+import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -99,6 +101,24 @@ public ReadCommand command()
         return command;
     }
 
+    @VisibleForTesting
+    public final Set<String> plannedIndexes()
+    {
+        try
+        {
+            Plan plan = controller.buildPlan().optimize();
+            Set<String> indexes = new HashSet<>();
+            plan.nodesOfType(Plan.IndexScan.class)
+                .forEach(s -> indexes.add(s.getIndexName()));
+            return indexes;
+        }
+        finally
+        {
+            // we need to call this to clean up the resources opened by the plan
+            controller.abort();
+        }
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     public UnfilteredPartitionIterator search(ReadExecutionController executionController) throws RequestTimeoutException

test/unit/org/apache/cassandra/index/sai/SAITester.java

@@ -62,6 +62,7 @@
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.ReadCommand;
 import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.db.lifecycle.SSTableSet;
 import org.apache.cassandra.db.lifecycle.View;
@@ -73,6 +74,8 @@
 import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
 import org.apache.cassandra.index.sai.disk.format.Version;
 import org.apache.cassandra.index.sai.plan.QueryController;
+import org.apache.cassandra.index.sai.plan.StorageAttachedIndexQueryPlan;
+import org.apache.cassandra.index.sai.plan.StorageAttachedIndexSearcher;
 import org.apache.cassandra.index.sai.utils.PrimaryKey;
 import org.apache.cassandra.index.sai.utils.ResourceLeakDetector;
 import org.apache.cassandra.inject.ActionBuilder;
@@ -88,10 +91,12 @@
 import org.apache.cassandra.schema.MockSchema;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.lucene.codecs.CodecUtil;
+import org.assertj.core.api.Assertions;
 
 import static org.apache.cassandra.inject.ActionBuilder.newActionBuilder;
 import static org.apache.cassandra.inject.Expression.expr;
@@ -959,4 +964,57 @@ public void start()
             }
         }
     }
+
+    protected PlanSelectionAssertion assertThatPlanFor(String query, Object[]... expectedRows)
+    {
+        // First execute the query and check the number of rows returned
+        ClientWarn.instance.captureWarnings();
+        assertRowsIgnoringOrder(execute(query), expectedRows);
+        List<String> warnings = ClientWarn.instance.getWarnings();
+        ClientWarn.instance.resetWarnings();
+
+        ReadCommand command = parseReadCommand(query);
+        Index.QueryPlan queryPlan = command.indexQueryPlan();
+        if (queryPlan == null)
+            return new PlanSelectionAssertion(null, warnings);
+
+        StorageAttachedIndexQueryPlan saiQueryPlan = (StorageAttachedIndexQueryPlan) queryPlan;
+        Assertions.assertThat(saiQueryPlan).isNotNull();
+        StorageAttachedIndexSearcher searcher = saiQueryPlan.searcherFor(command);
+        Set<String> selectedIndexes = searcher.plannedIndexes();
+        return new PlanSelectionAssertion(selectedIndexes, warnings);
+    }
+
+    protected static class PlanSelectionAssertion
+    {
+        private final List<String> warnings;
+        private final Set<String> selectedIndexes;
+
+        public PlanSelectionAssertion(@Nullable Set<String> selectedIndexes, @Nullable List<String> warnings)
+        {
+            this.warnings = warnings;
+            this.selectedIndexes = selectedIndexes;
+        }
+
+        public PlanSelectionAssertion uses(String... indexes)
+        {
+            Assertions.assertThat(selectedIndexes)
+                      .isNotNull()
+                      .as("Expected to select only %s, but got: %s", indexes, selectedIndexes)
+                      .isEqualTo(Set.of(indexes));
+            return this;
+        }
+
+        public void doesntWarn()
+        {
+            Assert.assertNull(warnings);
+        }
+
+        public void warns(String expectedWarning)
+        {
+            Assert.assertNotNull(warnings);
+            Assert.assertEquals(1, warnings.size());
+            Assert.assertEquals(expectedWarning, warnings.get(0));
+        }
+    }
 }