Avoid availability gap between UP and queryability marking for already built SAI indexes on bounce

patch by Caleb Rackliffe; reviewed by David Capwell and Dmitry Konstantinov for CASSANDRA-20732

Author: maedhroz

CHANGES.txt

@@ -1,4 +1,5 @@
 5.0.5
+ * Avoid availability gap between UP and queryability marking for already built SAI indexes on bounce (CASSANDRA-20732)
  * Make Commitlog flush data safely in Direct IO mode (CASSANDRA-20692)
  * Get SAI MemtableIndex refs before SSTableIndex refs at query time (CASSANDRA-20709)
  * Fix MAX_SEGMENT_SIZE < chunkSize in MmappedRegions::updateState (CASSANDRA-20636)

src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java

@@ -331,9 +331,30 @@ public IndexMetadata getIndexMetadata()
     public Callable<?> getInitializationTask()
     {
         // New storage-attached indexes will be available for queries after on disk index data are built.
-        // Memtable data will be indexed via flushing triggered by schema change
-        // We only want to validate the index files if we are starting up
-        IndexValidation validation = StorageService.instance.isStarting() ? IndexValidation.HEADER_FOOTER : IndexValidation.NONE;
+        // Memtable data will be indexed via flushing triggered by schema change.
+        // We only want to validate the index files if we are starting up.
+        boolean isStarting = StorageService.instance.isStarting();
+        IndexValidation validation = isStarting ? IndexValidation.HEADER_FOOTER : IndexValidation.NONE;
+
+        // Only attempt to make the index queryable if we are starting up. Otherwise, if we create a new index on top
+        // of nothing but existing Memtable data (i.e. no SSTables), that data will temporarily be lost until flush.
+        if (isStarting)
+        {
+            StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
+            assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
+
+            Collection<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, validation);
+
+            if (nonIndexed.isEmpty())
+            {
+                // If the index is complete, mark it queryable and avoid an initial build:
+                baseCfs.indexManager.makeIndexQueryable(this, Status.BUILD_SUCCEEDED);
+                logger.debug(indexIdentifier.logMessage("Skipping initial build, as index is already queryable..."));
+                initBuildStarted = true;
+                return () -> ImmediateFuture.success(null);
+            }
+        }
+
         return () -> startInitialBuild(baseCfs, validation).get();
     }
 
@@ -843,15 +864,12 @@ private Future<?> startInitialBuild(ColumnFamilyStore baseCfs, IndexValidation v
         // Force another flush to make sure on disk index is generated for memtable data before marking it queryable.
         // In the case of offline scrub, there are no live memtables.
         if (!baseCfs.getTracker().getView().liveMemtables.isEmpty())
-        {
             baseCfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.INDEX_BUILD_STARTED);
-        }
 
         // It is now safe to flush indexes directly from flushing Memtables.
         initBuildStarted = true;
 
         StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
-
         assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
 
         List<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, validation);
@@ -888,8 +906,7 @@ private Future<?> startPreJoinTask()
             }
 
             StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
-
-            assert indexGroup != null : "Index group does not exist for table";
+            assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
 
             Collection<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, IndexValidation.HEADER_FOOTER);
 

src/java/org/apache/cassandra/service/StorageService.java

@@ -2029,6 +2029,12 @@ public void setNormalModeUnsafe()
         setMode(Mode.NORMAL, true);
     }
 
+    @VisibleForTesting
+    public void setStartingModeUnsafe()
+    {
+        setMode(Mode.STARTING, true);
+    }
+
     private void setMode(Mode m, boolean log)
     {
         setMode(m, null, log);

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

@@ -18,13 +18,21 @@
 
 package org.apache.cassandra.index.sai.cql;
 
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.index.sai.SAITester;
 
 public class EmptyStringLifecycleTest extends SAITester
 {
+    @BeforeClass
+    public static void setup()
+    {
+        setUpClass();
+        requireNetwork(); // Ensure the node has advanced out of STARTING mode
+    }
+
     @Test
     public void testBeforeAndAfterFlush()
     {

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

@@ -76,6 +76,7 @@
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.Throwables;
 import org.assertj.core.api.Assertions;
 import org.mockito.Mockito;
@@ -1311,6 +1312,40 @@ public void nodetoolStopInitialIndexBuild() throws Throwable
         assertTrue(verifyChecksum(numericIndexTermType, numericIndexIdentifier));
     }
 
+    @Test
+    public void shouldMarkQueryableInInitializationTask() throws Throwable
+    {
+        createTable(CREATE_TABLE_TEMPLATE);
+        disableCompaction(KEYSPACE);
+        IndexIdentifier idxIdentifier = createIndexIdentifier(createIndexAsync(String.format(CREATE_INDEX_TEMPLATE, "v1")));
+
+        // create 10 SSTables
+        for (int i = 0; i < 10; i++)
+        {
+            execute("INSERT INTO %s (id1, v1, v2) VALUES (?, ?, ?)", String.valueOf(i), i, String.valueOf(i));
+            flush();
+        }
+
+        ResultSet rows = executeNet("SELECT id1 FROM %s WHERE v1 >= 5");
+        assertEquals(5, rows.all().size());
+
+        // Make the index artificially non-queryable:
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+        Index index = cfs.indexManager.getIndexByName(idxIdentifier.indexName);
+        cfs.indexManager.makeIndexNonQueryable(index, Index.Status.BUILD_FAILED);
+
+        // Query should fail with the index in an artificially non-queryable state:
+        assertThatThrownBy(() -> executeNet("SELECT id1 FROM %s WHERE v1 >= 5")).isInstanceOf(ReadFailureException.class);
+
+        // Node must be in STARTING mode for it to be necessary for the initialization task to pre-emptively validate:
+        StorageService.instance.setStartingModeUnsafe();
+        // Simply getting the initialization task (and not running it) will validate and mark the index queryable again:
+        cfs.indexManager.buildIndex(index);
+        StorageService.instance.setNormalModeUnsafe();
+        rows = executeNet("SELECT id1 FROM %s WHERE v1 >= 5");
+        assertEquals(5, rows.all().size());
+    }
+
     @Test
     public void shouldRejectQueriesWithCustomExpressions()
     {

test/unit/org/apache/cassandra/index/sai/functional/FailureTest.java

@@ -20,6 +20,7 @@
  */
 package org.apache.cassandra.index.sai.functional;
 
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.db.marshal.Int32Type;
@@ -36,6 +37,13 @@
 
 public class FailureTest extends SAITester
 {
+    @BeforeClass
+    public static void setup()
+    {
+        setUpClass();
+        requireNetwork(); // Ensure the node has advanced out of STARTING mode
+    }
+
     @Test
     public void shouldMakeIndexNonQueryableOnSSTableContextFailureDuringFlush() throws Throwable
     {

test/unit/org/apache/cassandra/index/sai/virtual/IndexesSystemViewTest.java

@@ -21,7 +21,6 @@
 import org.junit.BeforeClass;
 import org.junit.Test;
 
-import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.virtual.VirtualKeyspace;
 import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry;
@@ -60,7 +59,8 @@ public static void setup()
     {
         VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(SchemaConstants.VIRTUAL_VIEWS, ImmutableList.of(new ColumnIndexesSystemView(SchemaConstants.VIRTUAL_VIEWS))));
 
-        CQLTester.setUpClass();
+        setUpClass();
+        requireNetwork(); // Ensure the node has advanced out of STARTING mode
     }
 
     @Test