CNDB-13353: Queries with index-based ordering should reject aggregation queries (#1638)

Queries with index-based ordering (ANN, BM25, generic `ORDER BY`) don't
support CQL's aggregation queries, such as `GROUP BY` or `SELECT
COUNT(*)`. The reason is that the coordinator internally uses paging to
fetch more results from the replicas until the aggregated limit is
satisfied. Since top-k index-based ordering doesn't support paging,
these queries will fail.

We should reject these queries until we have a way to support them.

Author: adelapena

src/java/org/apache/cassandra/cql3/statements/SelectStatement.java

@@ -114,6 +114,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
 
     private static final Logger logger = LoggerFactory.getLogger(SelectStatement.class);
     private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(SelectStatement.logger, 1, TimeUnit.MINUTES);
+    public static final String TOPK_AGGREGATION_ERROR = "Top-K queries can not be run with aggregation";
     public static final String TOPK_CONSISTENCY_LEVEL_ERROR = "Top-K queries can only be run with consistency level ONE/LOCAL_ONE. Consistency level %s was used.";
     public static final String TOPK_CONSISTENCY_LEVEL_WARNING = "Top-K queries can only be run with consistency level ONE " +
                                                                 "/ LOCAL_ONE / NODE_LOCAL. Consistency level %s was requested. " +
@@ -431,10 +432,16 @@ public ReadQuery getQuery(QueryState queryState,
 
             // We don't support offset for top-k queries.
             checkFalse(userOffset != NO_OFFSET, String.format(TOPK_OFFSET_ERROR, userOffset));
+
+            // We don't support aggregation for top-k queries because we don't support paging.
+            checkFalse(aggregationSpec != null, TOPK_AGGREGATION_ERROR);
         }
 
         selectOptions.validate(queryState, table.keyspace, userLimit);
 
+        // If there's a secondary index that the command can use, have it validate the request parameters.
+        query.maybeValidateIndexes();
+
         return query;
     }
 
@@ -783,13 +790,7 @@ private ReadQuery getRangeCommand(QueryOptions options, ColumnFilter columnFilte
         if (keyBounds == null)
             return ReadQuery.empty(table);
 
-        ReadQuery command =
-            PartitionRangeReadQuery.create(table, nowInSec, columnFilter, rowFilter, limit, new DataRange(keyBounds, clusteringIndexFilter));
-
-        // If there's a secondary index that the command can use, have it validate the request parameters.
-        command.maybeValidateIndexes();
-
-        return command;
+        return PartitionRangeReadQuery.create(table, nowInSec, columnFilter, rowFilter, limit, new DataRange(keyBounds, clusteringIndexFilter));
     }
 
     private ClusteringIndexFilter makeClusteringIndexFilter(QueryOptions options, ColumnFilter columnFilter, QueryState queryState)

test/unit/org/apache/cassandra/index/sai/cql/BM25Test.java

@@ -27,13 +27,16 @@
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.cql3.statements.SelectStatement;
+import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.index.sai.SAITester;
 import org.apache.cassandra.index.sai.SAIUtil;
 import org.apache.cassandra.index.sai.disk.format.Version;
 import org.apache.cassandra.index.sai.disk.v1.SegmentBuilder;
 import org.apache.cassandra.index.sai.plan.QueryController;
 
 import static org.apache.cassandra.index.sai.analyzer.AnalyzerEqOperatorSupport.EQ_AMBIGUOUS_ERROR;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
 import static org.junit.Assert.assertEquals;
 
@@ -597,8 +600,32 @@ public void testWildcardSelection()
 
         var result = execute("SELECT * FROM %s ORDER BY v BM25 OF 'apple' LIMIT 3");
         assertThat(result).hasSize(1);
+    }
+
+    @Test
+    public void cannotHaveAggregationOnBM25Query()
+    {
+        createSimpleTable();
+
+        execute("INSERT INTO %s (k, v) VALUES (1, '4')");
+        execute("INSERT INTO %s (k, v) VALUES (2, '3')");
+        execute("INSERT INTO %s (k, v) VALUES (3, '2')");
+        execute("INSERT INTO %s (k, v) VALUES (4, '1')");
+
+        assertThatThrownBy(() -> execute("SELECT max(v) FROM %s ORDER BY v BM25 OF 'apple' LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT max(v) FROM %s WHERE k = 1 ORDER BY v BM25 OF 'apple' LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT * FROM %s GROUP BY k ORDER BY v BM25 OF 'apple' LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
 
-        var result2 = execute("SELECT * FROM %s GROUP BY k, c ORDER BY v BM25 OF 'apple' LIMIT 3");
-        assertThat(result2).hasSize(1);
+        assertThatThrownBy(() -> execute("SELECT count(*) FROM %s ORDER BY v BM25 OF 'apple' LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
     }
 }

test/unit/org/apache/cassandra/index/sai/cql/GenericOrderByTest.java

@@ -23,9 +23,13 @@
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.statements.SelectStatement;
+import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.index.sai.SAITester;
 import org.apache.cassandra.index.sai.plan.QueryController;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
 public class GenericOrderByTest extends SAITester
 {
     @Test
@@ -226,4 +230,31 @@ public void testSelectionAndOrderByOnTheSameColumnWithLargeRowCountDesc() throws
         testSelectionAndOrderByOnTheSameColumnWithLargeRowCount(false);
     }
 
+    @Test
+    public void cannotHaveAggregationOnOrderByQuery()
+    {
+        createTable("CREATE TABLE %s (k int PRIMARY KEY, v int)");
+        createIndex("CREATE CUSTOM INDEX ON %s(v) USING 'StorageAttachedIndex'");
+
+        execute("INSERT INTO %s (k, v) VALUES (1, 4)");
+        execute("INSERT INTO %s (k, v) VALUES (2, 3)");
+        execute("INSERT INTO %s (k, v) VALUES (3, 2)");
+        execute("INSERT INTO %s (k, v) VALUES (4, 1)");
+
+        assertThatThrownBy(() -> execute("SELECT sum(v) FROM %s ORDER BY v LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT sum(v) FROM %s WHERE k = 1 ORDER BY v LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT * FROM %s GROUP BY k ORDER BY v LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT count(*) FROM %s ORDER BY v LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+    }
 }

test/unit/org/apache/cassandra/index/sai/cql/VectorInvalidQueryTest.java

@@ -240,6 +240,34 @@ public void disallowClusteringColumnPredicateWithoutSupportingIndex()
         assertRows(execute("SELECT num FROM %s WHERE pk=3 AND num > 3 ORDER BY v ANN OF [1,1] LIMIT 1"), row(4));
     }
 
+    @Test
+    public void cannotHaveAggregationOnANNQuery()
+    {
+        createTable("CREATE TABLE %s (k int PRIMARY KEY, v vector<float, 1>, c int)");
+        createIndex("CREATE CUSTOM INDEX ON %s(v) USING 'StorageAttachedIndex' WITH OPTIONS = {'similarity_function' : 'euclidean'}");
+
+        execute("INSERT INTO %s (k, v, c) VALUES (1, [4], 1)");
+        execute("INSERT INTO %s (k, v, c) VALUES (2, [3], 10)");
+        execute("INSERT INTO %s (k, v, c) VALUES (3, [2], 100)");
+        execute("INSERT INTO %s (k, v, c) VALUES (4, [1], 1000)");
+
+        assertThatThrownBy(() -> execute("SELECT sum(c) FROM %s ORDER BY v ANN OF [0] LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT sum(c) FROM %s WHERE k = 1 ORDER BY v ANN OF [0] LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT * FROM %s GROUP BY k ORDER BY v ANN OF [0] LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+
+        assertThatThrownBy(() -> execute("SELECT count(*) FROM %s ORDER BY v ANN OF [0] LIMIT 4"))
+                .isInstanceOf(InvalidRequestException.class)
+                .hasMessage(SelectStatement.TOPK_AGGREGATION_ERROR);
+    }
+
     @Test
     public void canOnlyExecuteWithCorrectConsistencyLevel()
     {