Ensure correct filters with SASI clustering index

Patch by Sam Tunnicliffe; reviewed by Jordan West for CASSANDRA-11397

CHANGES.txt

@@ -1,4 +1,5 @@
 3.5
+ * Fix clustering and row filters for LIKE queries on clustering columns (CASSANDRA-11397)
 Merged from 3.0:
  * Allocate merkletrees with the correct size (CASSANDRA-11390)
  * Support streaming pre-3.0 sstables (CASSANDRA-10990)

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

@@ -54,6 +54,11 @@ final class PrimaryKeyRestrictionSet extends AbstractPrimaryKeyRestrictions
      */
     private boolean in;
 
+    /**
+     * <code>true</code> if the restrictions are corresponding to a LIKE, <code>false</code> otherwise.
+     */
+    private boolean like;
+
     /**
      * <code>true</code> if the restrictions are corresponding to a Slice, <code>false</code> otherwise.
      */
@@ -106,6 +111,8 @@ else if (restriction.isContains() || primaryKeyRestrictions.isContains())
             this.contains = true;
         else if (restriction.isIN() || primaryKeyRestrictions.isIN())
             this.in = true;
+        else if (restriction.isLIKE() || primaryKeyRestrictions.isLIKE())
+            this.like = true;
         else
             this.eq = true;
     }
@@ -137,6 +144,12 @@ public boolean isIN()
         return in;
     }
 
+    @Override
+    public boolean isLIKE()
+    {
+        return like;
+    }
+
     @Override
     public boolean isContains()
     {
@@ -220,7 +233,7 @@ public NavigableSet<Slice.Bound> boundsAsClustering(Bound bound, QueryOptions op
         {
             ColumnDefinition def = r.getFirstColumn();
 
-            if (keyPosition != def.position() || r.isContains())
+            if (keyPosition != def.position() || r.isContains() || r.isLIKE())
                 break;
 
             if (r.isSlice())
@@ -296,7 +309,7 @@ public void addRowFilterTo(RowFilter filter,
             ColumnDefinition columnDef = restriction.getFirstColumn();
 
             // We ignore all the clustering columns that can be handled by slices.
-            if (!isPartitionKey && !restriction.isContains()&& position == columnDef.position())
+            if (!isPartitionKey && !(restriction.isContains() || restriction.isLIKE()) && position == columnDef.position())
             {
                 position = restriction.getLastColumn().position() + 1;
                 if (!restriction.hasSupportingIndex(indexManager))

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

@@ -704,10 +704,10 @@ public void addRowFilterTo(RowFilter filter,
         @Override
         public MultiCBuilder appendTo(MultiCBuilder builder, QueryOptions options)
         {
-            // LIKE could be used with clustering columns as soon as they are indexed,
-            // but we have to hide such expression from clustering filter since it
-            // can only filter based on the complete values.
-            return builder;
+            // LIKE can be used with clustering columns, but as it doesn't
+            // represent an actual clustering value, it can't be used in a
+            // clustering filter.
+            throw new UnsupportedOperationException();
         }
 
         @Override

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

@@ -737,7 +737,7 @@ public boolean isColumnRange()
         // this would mean a 'SELECT *' on a static compact table would query whole partitions, even though we'll only return
         // the static part as far as CQL is concerned. This is thus mostly an optimization to use the query-by-name path).
         int numberOfClusteringColumns = cfm.isStaticCompactTable() ? 0 : cfm.clusteringColumns().size();
-        // it is a range query if it has at least one the column alias for which no relation is defined or is not EQ.
+        // it is a range query if it has at least one the column alias for which no relation is defined or is not EQ or IN.
         return clusteringColumnsRestrictions.size() < numberOfClusteringColumns
             || (!clusteringColumnsRestrictions.isEQ() && !clusteringColumnsRestrictions.isIN());
     }

test/unit/org/apache/cassandra/SchemaLoader.java

@@ -577,6 +577,11 @@ public static CFMetaData sasiCFMD(String ksName, String cfName)
     }
 
     public static CFMetaData clusteringSASICFMD(String ksName, String cfName)
+    {
+        return clusteringSASICFMD(ksName, cfName, "location", "age", "height", "score");
+    }
+
+    public static CFMetaData clusteringSASICFMD(String ksName, String cfName, String...indexedColumns)
     {
         CFMetaData cfm = CFMetaData.Builder.create(ksName, cfName)
                                            .addPartitionKey("name", UTF8Type.instance)
@@ -586,32 +591,17 @@ public static CFMetaData clusteringSASICFMD(String ksName, String cfName)
                                            .addRegularColumn("score", DoubleType.instance)
                                            .build();
 
-        cfm.indexes(cfm.getIndexes()
-                       .with(IndexMetadata.fromSchemaMetadata("location", IndexMetadata.Kind.CUSTOM, new HashMap<String, String>()
-                       {{
-                           put(IndexTarget.CUSTOM_INDEX_OPTION_NAME, SASIIndex.class.getName());
-                           put(IndexTarget.TARGET_OPTION_NAME, "location");
-                           put("mode", OnDiskIndexBuilder.Mode.PREFIX.toString());
-                       }}))
-                       .with(IndexMetadata.fromSchemaMetadata("age", IndexMetadata.Kind.CUSTOM, new HashMap<String, String>()
-                       {{
-                           put(IndexTarget.CUSTOM_INDEX_OPTION_NAME, SASIIndex.class.getName());
-                           put(IndexTarget.TARGET_OPTION_NAME, "age");
-                           put("mode", OnDiskIndexBuilder.Mode.PREFIX.toString());
-                       }}))
-                       .with(IndexMetadata.fromSchemaMetadata("height", IndexMetadata.Kind.CUSTOM, new HashMap<String, String>()
-                       {{
-                           put(IndexTarget.CUSTOM_INDEX_OPTION_NAME, SASIIndex.class.getName());
-                           put(IndexTarget.TARGET_OPTION_NAME, "height");
-                           put("mode", OnDiskIndexBuilder.Mode.PREFIX.toString());
-                       }}))
-                       .with(IndexMetadata.fromSchemaMetadata("score", IndexMetadata.Kind.CUSTOM, new HashMap<String, String>()
-                       {{
-                           put(IndexTarget.CUSTOM_INDEX_OPTION_NAME, SASIIndex.class.getName());
-                           put(IndexTarget.TARGET_OPTION_NAME, "score");
-                           put("mode", OnDiskIndexBuilder.Mode.PREFIX.toString());
-                       }})));
-
+        Indexes indexes = cfm.getIndexes();
+        for (String indexedColumn : indexedColumns)
+        {
+            indexes = indexes.with(IndexMetadata.fromSchemaMetadata(indexedColumn, IndexMetadata.Kind.CUSTOM, new HashMap<String, String>()
+            {{
+                put(IndexTarget.CUSTOM_INDEX_OPTION_NAME, SASIIndex.class.getName());
+                put(IndexTarget.TARGET_OPTION_NAME, indexedColumn);
+                put("mode", OnDiskIndexBuilder.Mode.PREFIX.toString());
+            }}));
+        }
+        cfm.indexes(indexes);
         return cfm;
     }
 

test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java

@@ -80,7 +80,8 @@ public class SASIIndexTest
 
     private static final String KS_NAME = "sasi";
     private static final String CF_NAME = "test_cf";
-    private static final String CLUSTERING_CF_NAME = "clustering_test_cf";
+    private static final String CLUSTERING_CF_NAME_1 = "clustering_test_cf_1";
+    private static final String CLUSTERING_CF_NAME_2 = "clustering_test_cf_2";
 
     @BeforeClass
     public static void loadSchema() throws ConfigurationException
@@ -90,7 +91,8 @@ public static void loadSchema() throws ConfigurationException
         MigrationManager.announceNewKeyspace(KeyspaceMetadata.create(KS_NAME,
                                                                      KeyspaceParams.simpleTransient(1),
                                                                      Tables.of(SchemaLoader.sasiCFMD(KS_NAME, CF_NAME),
-                                                                               SchemaLoader.clusteringSASICFMD(KS_NAME, CLUSTERING_CF_NAME))));
+                                                                               SchemaLoader.clusteringSASICFMD(KS_NAME, CLUSTERING_CF_NAME_1),
+                                                                               SchemaLoader.clusteringSASICFMD(KS_NAME, CLUSTERING_CF_NAME_2, "location"))));
     }
 
     @After
@@ -1621,64 +1623,64 @@ public void testClusteringIndexes() throws Exception
 
     public void testClusteringIndexes(boolean forceFlush) throws Exception
     {
-        ColumnFamilyStore store = Keyspace.open(KS_NAME).getColumnFamilyStore(CLUSTERING_CF_NAME);
+        ColumnFamilyStore store = Keyspace.open(KS_NAME).getColumnFamilyStore(CLUSTERING_CF_NAME_1);
 
-        executeCQL("INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Pavel", "US", 27, 183, 1.0);
-        executeCQL("INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Pavel", "BY", 28, 182, 2.0);
-        executeCQL("INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Jordan", "US", 27, 182, 1.0);
+        executeCQL(CLUSTERING_CF_NAME_1, "INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Pavel", "US", 27, 183, 1.0);
+        executeCQL(CLUSTERING_CF_NAME_1, "INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Pavel", "BY", 28, 182, 2.0);
+        executeCQL(CLUSTERING_CF_NAME_1 ,"INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Jordan", "US", 27, 182, 1.0);
 
         if (forceFlush)
             store.forceBlockingFlush();
 
         UntypedResultSet results;
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE location = ? ALLOW FILTERING", "US");
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location = ? ALLOW FILTERING", "US");
         Assert.assertNotNull(results);
         Assert.assertEquals(2, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE age >= ? AND height = ? ALLOW FILTERING", 27, 182);
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE age >= ? AND height = ? ALLOW FILTERING", 27, 182);
         Assert.assertNotNull(results);
         Assert.assertEquals(2, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE age = ? AND height = ? ALLOW FILTERING", 28, 182);
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE age = ? AND height = ? ALLOW FILTERING", 28, 182);
         Assert.assertNotNull(results);
         Assert.assertEquals(1, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE age >= ? AND height = ? AND score >= ? ALLOW FILTERING", 27, 182, 1.0);
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE age >= ? AND height = ? AND score >= ? ALLOW FILTERING", 27, 182, 1.0);
         Assert.assertNotNull(results);
         Assert.assertEquals(2, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE age >= ? AND height = ? AND score = ? ALLOW FILTERING", 27, 182, 1.0);
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE age >= ? AND height = ? AND score = ? ALLOW FILTERING", 27, 182, 1.0);
         Assert.assertNotNull(results);
         Assert.assertEquals(1, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE location = ? AND age >= ? ALLOW FILTERING", "US", 27);
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location = ? AND age >= ? ALLOW FILTERING", "US", 27);
         Assert.assertNotNull(results);
         Assert.assertEquals(2, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE location = ? ALLOW FILTERING", "BY");
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location = ? ALLOW FILTERING", "BY");
         Assert.assertNotNull(results);
         Assert.assertEquals(1, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE location LIKE 'U%%' ALLOW FILTERING");
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location LIKE 'U%%' ALLOW FILTERING");
         Assert.assertNotNull(results);
         Assert.assertEquals(2, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE location LIKE 'U%%' AND height >= 183 ALLOW FILTERING");
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location LIKE 'U%%' AND height >= 183 ALLOW FILTERING");
         Assert.assertNotNull(results);
         Assert.assertEquals(1, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE location LIKE 'US%%' ALLOW FILTERING");
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location LIKE 'US%%' ALLOW FILTERING");
         Assert.assertNotNull(results);
         Assert.assertEquals(2, results.size());
 
-        results = executeCQL("SELECT * FROM %s.%s WHERE location LIKE 'US' ALLOW FILTERING");
+        results = executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location LIKE 'US' ALLOW FILTERING");
         Assert.assertNotNull(results);
         Assert.assertEquals(2, results.size());
 
         try
         {
-            executeCQL("SELECT * FROM %s.%s WHERE location LIKE '%%U' ALLOW FILTERING");
+            executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location LIKE '%%U' ALLOW FILTERING");
             Assert.fail();
         }
         catch (InvalidRequestException e)
@@ -1689,7 +1691,7 @@ public void testClusteringIndexes(boolean forceFlush) throws Exception
 
         try
         {
-            executeCQL("SELECT * FROM %s.%s WHERE location LIKE '%%' ALLOW FILTERING");
+            executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location LIKE '%%' ALLOW FILTERING");
             Assert.fail();
         }
         catch (SyntaxException e)
@@ -1700,14 +1702,27 @@ public void testClusteringIndexes(boolean forceFlush) throws Exception
 
         try
         {
-            executeCQL("SELECT * FROM %s.%s WHERE location LIKE '%%%%' ALLOW FILTERING");
+            executeCQL(CLUSTERING_CF_NAME_1 ,"SELECT * FROM %s.%s WHERE location LIKE '%%%%' ALLOW FILTERING");
             Assert.fail();
         }
         catch (SyntaxException e)
         {
             Assert.assertTrue(e.getMessage().contains("empty"));
             // expected
         }
+
+        // check restrictions on non-indexed clustering columns when preceding columns are indexed
+        store = Keyspace.open(KS_NAME).getColumnFamilyStore(CLUSTERING_CF_NAME_2);
+        executeCQL(CLUSTERING_CF_NAME_2 ,"INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Tony", "US", 43, 184, 2.0);
+        executeCQL(CLUSTERING_CF_NAME_2 ,"INSERT INTO %s.%s (name, location, age, height, score) VALUES (?, ?, ?, ?, ?)", "Christopher", "US", 27, 180, 1.0);
+
+        if (forceFlush)
+            store.forceBlockingFlush();
+
+        results = executeCQL(CLUSTERING_CF_NAME_2 ,"SELECT * FROM %s.%s WHERE location LIKE 'US' AND age = 43 ALLOW FILTERING");
+        Assert.assertNotNull(results);
+        Assert.assertEquals(1, results.size());
+        Assert.assertEquals("Tony", results.one().getString("name"));
     }
 
     @Test
@@ -2037,7 +2052,7 @@ private void cleanupData()
     {
         Keyspace ks = Keyspace.open(KS_NAME);
         ks.getColumnFamilyStore(CF_NAME).truncateBlocking();
-        ks.getColumnFamilyStore(CLUSTERING_CF_NAME).truncateBlocking();
+        ks.getColumnFamilyStore(CLUSTERING_CF_NAME_1).truncateBlocking();
     }
 
     private static Set<String> getIndexed(ColumnFamilyStore store, int maxResults, Expression... expressions)
@@ -2150,9 +2165,9 @@ private static List<String> convert(final Set<DecoratedKey> keys)
         }};
     }
 
-    private UntypedResultSet executeCQL(String query, Object... values)
+    private UntypedResultSet executeCQL(String cfName, String query, Object... values)
     {
-        return QueryProcessor.executeOnceInternal(String.format(query, KS_NAME, CLUSTERING_CF_NAME), values);
+        return QueryProcessor.executeOnceInternal(String.format(query, KS_NAME, cfName), values);
     }
 
     private Set<String> executeCQLWithKeys(String rawStatement) throws Exception