CNDB-14614-re-merge: abort compaction task or index build if index is unloaded after tenant unassignment (#1766)

### What is the issue
When schema is unloaded after tenant unassignment, compaction task might
finishes without corresponding index files, making index non-queryable.

### What does this PR fix and why was it fixed
Replace `isValid` with `isDropped` and `isUnloaded`. If index is
dropped, compaction task or index build can proceed without the index,
same behavior as before. If index is unloaded, compaction task or index
build will be aborted to avoid completing without index files.

---

#1754 was reverted because[
CNDB PR](riptano/cndb#14179) compiled failed
with wrong hash. Re-merge it again.

Author: jasonstack

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

@@ -234,8 +234,9 @@ public List<SecondaryIndexBuilder> getParallelIndexBuildTasks(ColumnFamilyStore
     // Tracks whether or not we've started the index build on initialization.
     private volatile boolean canFlushFromMemtableIndex = false;
 
-    // Tracks whether the index has been invalidated due to removal, a table drop, etc.
-    private volatile boolean valid = true;
+    // Tracks whether the index has been dropped due to removal, a table drop, etc or index schema is unloaded after schema unassignment
+    private volatile boolean dropped = false;
+    private volatile boolean unloaded = false;
 
     /**
      * Called via reflection from SecondaryIndexManager
@@ -554,7 +555,7 @@ public Callable<?> getInvalidateTask()
         return () ->
         {
             // mark index as invalid, in-progress SSTableIndexWriters will abort
-            valid = false;
+            dropped = true;
 
             // in case of dropping table, SSTable indexes should already been removed by SSTableListChangedNotification.
             for (SSTableIndex sstableIndex : indexContext.getView().getIndexes())
@@ -574,7 +575,7 @@ public Callable<?> getUnloadTask()
         return () ->
         {
             // mark index as invalid, in-progress SSTableIndexWriters will abort
-            valid = false;
+            unloaded = true;
 
             indexContext.invalidate(false);
             return null;
@@ -600,9 +601,14 @@ public boolean canFlushFromMemtableIndex()
         return canFlushFromMemtableIndex;
     }
 
-    public BooleanSupplier isIndexValid()
+    public BooleanSupplier isDropped()
     {
-        return () -> valid;
+        return () -> dropped;
+    }
+
+    public BooleanSupplier isUnloaded()
+    {
+        return () -> unloaded;
     }
 
     private Future<?> startPreJoinTask()

src/java/org/apache/cassandra/index/sai/disk/v1/SSTableIndexWriter.java

@@ -68,7 +68,8 @@ public class SSTableIndexWriter implements PerIndexWriter
     private final IndexContext indexContext;
     private final int nowInSec = FBUtilities.nowInSeconds();
     private final NamedMemoryLimiter limiter;
-    private final BooleanSupplier isIndexValid;
+    private final BooleanSupplier isIndexDropped;
+    private final BooleanSupplier isIndexUnloaded;
     private final long keyCount;
 
     private boolean aborted = false;
@@ -77,13 +78,15 @@ public class SSTableIndexWriter implements PerIndexWriter
     private SegmentBuilder currentBuilder;
     private final List<SegmentMetadata> segments = new ArrayList<>();
 
-    public SSTableIndexWriter(IndexComponents.ForWrite perIndexComponents, NamedMemoryLimiter limiter, BooleanSupplier isIndexValid, long keyCount)
+    public SSTableIndexWriter(IndexComponents.ForWrite perIndexComponents, NamedMemoryLimiter limiter,
+                              BooleanSupplier isIndexDropped, BooleanSupplier isIndexUnloaded, long keyCount)
     {
         this.perIndexComponents = perIndexComponents;
         this.indexContext = perIndexComponents.context();
         Preconditions.checkNotNull(indexContext, "Provided components %s are the per-sstable ones, expected per-index ones", perIndexComponents);
         this.limiter = limiter;
-        this.isIndexValid = isIndexValid;
+        this.isIndexDropped = isIndexDropped;
+        this.isIndexUnloaded = isIndexUnloaded;
         this.keyCount = keyCount;
     }
 
@@ -222,11 +225,23 @@ private boolean maybeAbort()
         if (aborted)
             return true;
 
-        if (isIndexValid.getAsBoolean())
+        boolean dropped = isIndexDropped.getAsBoolean();
+        boolean unloaded = isIndexUnloaded.getAsBoolean();
+        if (!dropped && !unloaded)
             return false;
 
-        abort(new RuntimeException(String.format("index %s is dropped", indexContext.getIndexName())));
-        return true;
+        String message = String.format("index %s is %s", indexContext.getIndexName(), dropped ? "dropped" : "unloaded");
+        RuntimeException runtimeException = new RuntimeException(message);
+
+        // abort index build for remove on disk index file
+        abort(runtimeException);
+
+        // if index is dropped, we can continue compaction task or index build without current index
+        if (dropped)
+            return true;
+
+        // if index is unloaded after unassigning tenant, fail the compaction task or index build to avoid incomplete index files
+        throw runtimeException;
     }
 
     private boolean addTerm(ByteBuffer term, PrimaryKey key, long sstableRowId, AbstractType<?> type) throws IOException

src/java/org/apache/cassandra/index/sai/disk/v1/V1OnDiskFormat.java

@@ -207,7 +207,7 @@ public PerIndexWriter newPerIndexWriter(StorageAttachedIndex index,
             logger.debug(index.getIndexContext().logMessage("Starting a compaction index build. Global segment memory usage: {}"),
                          prettyPrintMemory(limiter.currentBytesUsed()));
 
-            return new SSTableIndexWriter(perIndexComponents, limiter, index.isIndexValid(), keyCount);
+            return new SSTableIndexWriter(perIndexComponents, limiter, index.isDropped(), index.isUnloaded(), keyCount);
         }
 
         return new MemtableIndexWriter(context.getPendingMemtableIndex(tracker),

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

@@ -62,6 +62,7 @@
 import org.apache.cassandra.index.sai.StorageAttachedIndex;
 import org.apache.cassandra.index.sai.StorageAttachedIndexBuilder;
 import org.apache.cassandra.index.sai.StorageAttachedIndexGroup;
+import org.apache.cassandra.index.sai.disk.StorageAttachedIndexWriter;
 import org.apache.cassandra.index.sai.disk.format.IndexComponentType;
 import org.apache.cassandra.index.sai.disk.format.Version;
 import org.apache.cassandra.index.sai.disk.v1.SegmentBuilder;
@@ -125,6 +126,11 @@ public class NativeIndexDDLTest extends SAITester
                                                                                                 .onMethod("<init>"))
                                                                          .add(ActionBuilder.newActionBuilder().actions().doThrow(RuntimeException.class, Expression.quote("Injected failure!")))
                                                                          .build();
+
+    private static final Injection pauseSAIWriterComplete = Injections.newPause("pause_sai_writer_complete", 10_000)
+                                                                      .add(InvokePointBuilder.newInvokePoint().onClass(StorageAttachedIndexWriter.class).onMethod("complete"))
+                                                                      .build();
+
     @BeforeClass
     public static void init()
     {
@@ -960,6 +966,124 @@ public void testIndexCompactionAborted() throws Throwable
         assertEquals(4, rows.all().size());
     }
 
+    @Test
+    public void testIndexCompactionWhenIndexIsDropped() throws Throwable
+    {
+        // prepare schema and data
+        createTable(CREATE_TABLE_TEMPLATE);
+        String indexName1 = createIndex(String.format(CREATE_INDEX_TEMPLATE, "v1"));
+        createIndex(String.format(CREATE_INDEX_TEMPLATE, "v2"));
+
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('0', 0, '0');");
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('1', 1, '0');");
+        flush();
+
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('2', 0, '0');");
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('3', 1, '0');");
+        flush();
+
+        ResultSet rows = executeNet("SELECT id1 FROM %s WHERE v1>=0");
+        assertEquals(4, rows.all().size());
+
+        // pause index writer completion
+        Injections.inject(pauseSAIWriterComplete);
+        pauseSAIWriterComplete.enable();
+
+        try
+        {
+
+            ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+            assertThat(cfs.getLiveSSTables()).hasSize(2);
+
+            new TestWithConcurrentVerification(
+            () -> cfs.forceMajorCompaction(),
+            () -> {
+                try
+                {
+                    // wait for compaction paused
+                    FBUtilities.sleepQuietly(5000);
+
+                    dropIndex("DROP INDEX %s." + indexName1);
+                }
+                catch (Exception e)
+                {
+                    throw new RuntimeException(e);
+                }
+            }, -1 // run verification task once
+            ).start();
+
+            // compaction completed
+            assertThat(cfs.getLiveSSTables()).hasSize(1);
+
+            StorageAttachedIndexGroup saiGroup = StorageAttachedIndexGroup.getIndexGroup(cfs);
+            assertThat(saiGroup.sstableContextManager().size()).isEqualTo(1);
+        }
+        finally
+        {
+            pauseSAIWriterComplete.disable();
+        }
+    }
+
+    @Test
+    public void testIndexCompactionWhenIndexIsUnloaded() throws Throwable
+    {
+        // prepare schema and data
+        createTable(CREATE_TABLE_TEMPLATE);
+        createIndex(String.format(CREATE_INDEX_TEMPLATE, "v1"));
+        createIndex(String.format(CREATE_INDEX_TEMPLATE, "v2"));
+
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('0', 0, '0');");
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('1', 1, '0');");
+        flush();
+
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('2', 0, '0');");
+        execute("INSERT INTO %s (id1, v1, v2) VALUES ('3', 1, '0');");
+        flush();
+
+        ResultSet rows = executeNet("SELECT id1 FROM %s WHERE v1>=0");
+        assertEquals(4, rows.all().size());
+
+        // pause index writer completion
+        Injections.inject(pauseSAIWriterComplete);
+        pauseSAIWriterComplete.enable();
+
+        try
+        {
+            ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+            assertThat(cfs.getLiveSSTables()).hasSize(2);
+            TestWithConcurrentVerification compactionTask = new TestWithConcurrentVerification(
+            () -> cfs.forceMajorCompaction(),
+            () -> {
+                try
+                {
+                    // wait for compaction paused
+                    FBUtilities.sleepQuietly(5000);
+
+                    SecondaryIndexManager sim = cfs.getIndexManager();
+                    sim.unloadAllIndexes();
+                }
+                catch (Exception e)
+                {
+                    throw new RuntimeException(e);
+                }
+            }, -1 // run verification task once
+            );
+
+            assertThatThrownBy(compactionTask::start).hasMessageContaining("is unloaded");
+
+            // compaction is aborted
+            assertThat(cfs.getLiveSSTables()).hasSize(2);
+
+            // indexes are unloaded
+            StorageAttachedIndexGroup saiGroup = StorageAttachedIndexGroup.getIndexGroup(cfs);
+            assertThat(saiGroup.sstableContextManager().size()).isEqualTo(0);
+        }
+        finally
+        {
+            pauseSAIWriterComplete.disable();
+        }
+    }
+
     @Test
     public void verifyRebuildCorruptedFiles() throws Throwable
     {

test/unit/org/apache/cassandra/index/sai/disk/v1/SegmentFlushTest.java

@@ -155,7 +155,7 @@ private void testFlushBetweenRowIds(long sstableRowId1, long sstableRowId2, int
                                                      MockSchema.newCFS("ks"));
 
         IndexComponents.ForWrite components = indexDescriptor.newPerIndexComponentsForWrite(indexContext);
-        SSTableIndexWriter writer = new SSTableIndexWriter(components, V1OnDiskFormat.SEGMENT_BUILD_MEMORY_LIMITER, () -> true, 2);
+        SSTableIndexWriter writer = new SSTableIndexWriter(components, V1OnDiskFormat.SEGMENT_BUILD_MEMORY_LIMITER, () -> false, () -> false, 2);
 
         List<DecoratedKey> keys = Arrays.asList(dk("1"), dk("2"));
         Collections.sort(keys);