CNDB-16043: Add more tests for mixed version SAI queries (#2325)

Extends IndexQuerySupport framework with new test scenarios mixing
row-aware and non-row-aware index versions.

In the added scenarios, rows are first inserted and flushed into AA
indexes, then the node(s) are upgraded to a row-aware index SAI version
and finally more rows are inserted and flushed. Then existing workloads
are executed against those mixed version indexes. Compaction of mixed
version indexes is also tested.

This commit does not add new test queries nor new datamodels, but only
reuses the existing tests we had that were executed against a single
version at a time.

Co-authored-by: Piotr Kołaczkowski <pkolaczk@datastax.com>

Author: driftx

test/distributed/org/apache/cassandra/distributed/test/sai/datamodels/MultiNodeExecutor.java

@@ -19,15 +19,20 @@
 package org.apache.cassandra.distributed.test.sai.datamodels;
 
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.test.sai.SAIUtil;
 import org.apache.cassandra.distributed.util.ColumnTypeUtil;
+import org.apache.cassandra.index.sai.SAITester;
 import org.apache.cassandra.index.sai.cql.datamodels.DataModel;
+import org.apache.cassandra.index.sai.disk.format.Version;
 
 public class MultiNodeExecutor implements DataModel.Executor
 {
@@ -56,6 +61,15 @@ public void compact(String keyspace, String table)
         cluster.forEach(node -> node.forceCompact(keyspace, table));
     }
 
+    @Override
+    public void setCurrentVersion(Version version)
+    {
+        // need to pass version as String, because Version is not serializable and cannot be easily made to be so
+        String versionName = version.toString();
+        cluster.forEach(node ->
+                        node.runOnInstance(() -> org.apache.cassandra.index.sai.SAIUtil.setCurrentVersion(Version.parse(versionName))));
+    }
+
     @Override
     public void disableCompaction(String keyspace, String table)
     {
@@ -98,4 +112,22 @@ public long getCounter()
     {
         return MultiNodeQueryTester.Counter.get();
     }
+
+    @Override
+    public Set<Version> getSSTableIndexVersions(String keyspace, String indexedTable)
+    {
+        // Need to pass version as String, because Version is not serializable
+        var versions = new HashSet<String>();
+        cluster.forEach(node ->
+                        versions.addAll(
+                            node.callsOnInstance(() ->
+                                SAITester.getSSTableIndexVersions(keyspace, indexedTable)
+                                         .stream()
+                                         .map(Version::toString)
+                                         .collect(Collectors.toSet())
+                            ).call()
+            )
+        );
+        return versions.stream().map(Version::parse).collect(Collectors.toSet());
+    }
 }

test/distributed/org/apache/cassandra/distributed/test/sai/datamodels/QueryCellDeletionsUpgradeTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.sai.datamodels;
+
+import org.apache.cassandra.index.sai.cql.datamodels.IndexQuerySupport;
+import org.junit.Test;
+
+public class QueryCellDeletionsUpgradeTest extends MultiNodeQueryTester
+{
+    @Test
+    public void testCellDeletionsAfterUpgrade() throws Throwable
+    {
+        IndexQuerySupport.cellDeletionsAfterUpgrade(executor, dataModel.get(), sets);
+    }
+}

test/distributed/org/apache/cassandra/distributed/test/sai/datamodels/QueryRowDeletionsUpgradeTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.sai.datamodels;
+
+import org.apache.cassandra.index.sai.cql.datamodels.IndexQuerySupport;
+import org.junit.Test;
+
+public class QueryRowDeletionsUpgradeTest extends MultiNodeQueryTester
+{
+    @Test
+    public void testRowDeletionsAfterUpgrade() throws Throwable
+    {
+        IndexQuerySupport.rowDeletionsAfterUpgrade(executor, dataModel.get(), sets);
+    }
+}

test/distributed/org/apache/cassandra/distributed/test/sai/datamodels/QueryTimeToLiveUpgradeTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.sai.datamodels;
+
+import org.apache.cassandra.index.sai.cql.datamodels.IndexQuerySupport;
+import org.junit.Test;
+
+public class QueryTimeToLiveUpgradeTest extends MultiNodeQueryTester
+{
+    @Test
+    public void testTimeToLiveAfterUpgrade() throws Throwable
+    {
+        IndexQuerySupport.timeToLiveAfterUpgrade(executor, dataModel.get(), sets);
+    }
+}

test/distributed/org/apache/cassandra/distributed/test/sai/datamodels/QueryWriteLifecycleUpgradeTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.sai.datamodels;
+
+import org.apache.cassandra.index.sai.cql.datamodels.IndexQuerySupport;
+import org.junit.Test;
+
+public class QueryWriteLifecycleUpgradeTest extends MultiNodeQueryTester
+{
+    @Test
+    public void testWriteLifecycleAfterUpgrade() throws Throwable
+    {
+        IndexQuerySupport.writeLifecycleAfterUpgrade(executor, dataModel.get(), sets);
+    }
+}

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

@@ -454,6 +454,36 @@ protected void verifySAIVersionInUse(Version expectedVersion, String keyspace, S
         }
     }
 
+    /**
+     * Returns the set containing all SAI index versions found among all sstable indexes of the given table.
+     * Memtable indexes are not included.
+     */
+    public static Set<Version> getSSTableIndexVersions(String keyspace, String table)
+    {
+        ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
+        var group = StorageAttachedIndexGroup.getIndexGroup(cfs);
+        if (group == null)  // no indexes present on the table
+            return Set.of();
+
+        Set<IndexContext> indexContexts = group.getIndexes()
+                                               .stream()
+                                               .map(StorageAttachedIndex::getIndexContext)
+                                               .collect(Collectors.toSet());
+
+        var versions = new HashSet<Version>();
+        for (var sstable : cfs.getLiveSSTables())
+        {
+            var indexDescriptor = group.descriptorFor(sstable);
+            if (indexDescriptor != null)
+            {
+                versions.add(indexDescriptor.perSSTableComponents().version());
+                for (IndexContext indexContext : indexContexts)
+                    versions.add(indexDescriptor.perIndexComponents(indexContext).version());
+            }
+        }
+        return versions;
+    }
+
     protected static void assertFailureReason(ReadFailureException e, RequestFailureReason reason)
     {
         int expected = reason.codeForNativeProtocol();

test/unit/org/apache/cassandra/index/sai/cql/datamodels/DataModel.java

@@ -31,6 +31,7 @@
 import com.google.common.collect.Sets;
 
 import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.index.sai.disk.format.Version;
 import org.apache.cassandra.utils.Pair;
 
 public interface DataModel
@@ -155,6 +156,10 @@ public interface DataModel
 
     void insertRows(Executor tester) throws Throwable;
 
+    void insertRowsPartA(Executor tester) throws Throwable;
+
+    void insertRowsPartB(Executor tester) throws Throwable;
+
     void insertRowsWithTTL(Executor tester) throws Throwable;
 
     void updateCells(Executor tester) throws Throwable;
@@ -167,6 +172,9 @@ public interface DataModel
 
     List<Object> executeNonIndexed(Executor tester, String query, int fetchSize, Object... values) throws Throwable;
 
+    /** Returns the set of all SAI index versions in use by the model for the on-disk indexes */
+    Set<Version> getSSTableIndexVersions(Executor tester) throws Throwable;
+
     public class BaseDataModel implements DataModel
     {
         final List<Pair<String, String>> columns;
@@ -272,6 +280,27 @@ public void insertRows(Executor tester) throws Throwable
             }
         }
 
+        @Override
+        public void insertRowsPartA(Executor tester) throws Throwable
+        {
+            insertRowsRange(tester, 0, keys.size() / 2);
+        }
+
+        @Override
+        public void insertRowsPartB(Executor tester) throws Throwable
+        {
+            insertRowsRange(tester, keys.size() / 2, keys.size());
+        }
+
+        private void insertRowsRange(Executor tester, int startIndex, int endIndex) throws Throwable
+        {
+            String template = "INSERT INTO %%s (%s, %s) VALUES (%s, %s)";
+            for (int i = startIndex; i < endIndex; i++)
+            {
+                executeLocal(tester, String.format(template, primaryKey, columnNames, keys.get(i), rows.get(i)));
+            }
+        }
+
         public void insertRowsWithTTL(Executor tester) throws Throwable
         {
             String template = "INSERT INTO %%s (%s, %s) VALUES (%s, %s)%s";
@@ -341,6 +370,12 @@ public List<Object> executeNonIndexed(Executor tester, String query, int fetchSi
             return tester.executeRemote(formatNonIndexedQuery(query), fetchSize, values);
         }
 
+        @Override
+        public Set<Version> getSSTableIndexVersions(Executor tester) throws Throwable
+        {
+            return tester.getSSTableIndexVersions(KEYSPACE, indexedTable);
+        }
+
         protected Set<Integer> deletable()
         {
             return Sets.newHashSet(3, 7, 9, 12);
@@ -614,6 +649,8 @@ public static interface Executor
 
         void compact(String keyspace, String table);
 
+        void setCurrentVersion(Version version);
+
         void disableCompaction(String keyspace, String table);
 
         void waitForTableIndexesQueryable(String keyspace, String table);
@@ -625,5 +662,11 @@ public static interface Executor
         void counterReset();
 
         long getCounter();
+
+        /**
+         * Returns the set of all SAI index versions in use by the specified table for the on-disk indexes.
+         * Memtable indexes are not included.
+         */
+        Set<Version> getSSTableIndexVersions(String keyspace, String indexedTable);
     }
 }