CNDB-14477: Validate SAI Components in Verifier (#1816)

- **CNDB-14477: Validate SAI Components in Verifier**
- **Add follow up comment to vector corruption check**

### What is the issue
Fixes: riptano/cndb#14477

### What does this PR fix and why was it fixed
This PR leverages the existing SAI verification framework and wires it
up to the `Verifier` logic in the code compaction code.

Author: michaeljmarshall

src/java/org/apache/cassandra/db/compaction/Verifier.java

@@ -64,6 +64,7 @@
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.RandomAccessReader;
 import org.apache.cassandra.io.util.ReadPattern;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.service.StorageService;
@@ -237,6 +238,9 @@ public void verify()
             }
         }
 
+        // Validate the secondary indexes. Only validate checksums if quick is disabled.
+        validateSecondaryIndexes(!options.quick);
+
         if (options.quick)
             return;
 
@@ -382,6 +386,34 @@ public void verify()
         outputHandler.output("Verify of " + sstable + " succeeded. All " + goodRows + " rows read successfully");
     }
 
+    private void validateSecondaryIndexes(boolean validateChecksum)
+    {
+        if (realm == null)
+        {
+            outputHandler.output("Skipping secondary index component validation for " + sstable +
+                                 " because the compaction realm is not available");
+            return;
+        }
+
+        outputHandler.output(String.format("Checking secondary index components for %s, validateChecksum=%s", sstable, validateChecksum));
+        try
+        {
+            var indexManager = realm.getIndexManager();
+            for (IndexMetadata indexMetadata : sstable.metadata().indexes)
+            {
+                var index = indexManager.getIndexGroup(indexMetadata);
+                if (index == null)
+                    throw new IllegalStateException("Cannot verify index components for " + sstable + " because the index " + indexMetadata.name + " is not registered");
+                index.validateComponents(sstable, validateChecksum);
+            }
+        }
+        catch (Throwable t)
+        {
+            outputHandler.warn(t);
+            markAndThrow(t);
+        }
+    }
+
     /**
      * Use the fact that check(..) is called with sorted tokens - we keep a pointer in to the normalized ranges
      * and only bump the pointer if the key given is out of range. This is done to avoid calling .contains(..) many

src/java/org/apache/cassandra/index/Index.java

@@ -57,6 +57,7 @@
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.lucene.index.CorruptIndexException;
 
 
 /**
@@ -859,6 +860,19 @@ default void unload() { }
          */
         Set<Component> activeComponents(SSTableReader sstable);
 
+        /**
+         * Validate the sstable-attached components belonging to the group that are currently "active" for the
+         * provided sstable. Method is side effect free.
+         * <p>
+         * The "active" components are those returned by {@link #activeComponents}.
+         *
+         * @param sstable          the sstable to validate components for.
+         * @param validateChecksum if {@code true}, the checksum of the components will be validated. Otherwise, only
+         *                         basic checks on the header and footers will be performed.
+         * @throws CorruptIndexException if the validation fails.
+         */
+        void validateComponents(SSTableReader sstable, boolean validateChecksum) throws CorruptIndexException;
+
         /**
          * @return true if this index group is capable of supporting multiple contains restrictions, false otherwise
          */

src/java/org/apache/cassandra/index/IndexRegistry.java

@@ -256,6 +256,11 @@ public Set<Component> activeComponents(SSTableReader sstable)
             {
                 return null;
             }
+
+            @Override
+            public void validateComponents(SSTableReader sstable, boolean validateChecksum)
+            {
+            }
         };
 
         public void registerIndex(Index index, Index.Group.Key groupKey, Supplier<Index.Group> groupSupplier)

src/java/org/apache/cassandra/index/SingletonIndexGroup.java

@@ -147,4 +147,10 @@ public Set<Component> activeComponents(SSTableReader sstable)
         // Same rermarks as for `componentsForNewBuid`.
         return Collections.emptySet();
     }
+
+    @Override
+    public void validateComponents(SSTableReader sstable, boolean validateChecksum)
+    {
+        // Same remarks as for `componentsForNewBuid`.
+    }
 }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.index.sai;
 
+import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -59,6 +60,7 @@
 import org.apache.cassandra.index.sai.plan.StorageAttachedIndexQueryPlan;
 import org.apache.cassandra.index.transactions.IndexTransaction;
 import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTableWatcher;
 import org.apache.cassandra.io.sstable.format.SSTableFlushObserver;
@@ -71,6 +73,7 @@
 import org.apache.cassandra.notifications.SSTableListChangedNotification;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.Throwables;
+import org.apache.lucene.index.CorruptIndexException;
 
 /**
  * Orchestrates building of storage-attached indices, and manages lifecycle of resources shared between them.
@@ -307,6 +310,21 @@ public Set<Component> activeComponents(SSTableReader sstable)
         return components;
     }
 
+    @Override
+    public void validateComponents(SSTableReader sstable, boolean validateChecksum) throws CorruptIndexException
+    {
+        IndexDescriptor indexDescriptor = descriptorFor(sstable);
+        if (!indexDescriptor.perSSTableComponents().isValid(validateChecksum))
+            throw new CorruptIndexException("Failed validation of per-SSTable components", sstable.getFilename());
+
+        for (StorageAttachedIndex index : indices)
+        {
+            IndexContext ctx = index.getIndexContext();
+            if (!indexDescriptor.perIndexComponents(ctx).isValid(validateChecksum))
+                throw new CorruptIndexException("Failed validation of per-column components for " + ctx, sstable.getFilename());
+        }
+    }
+
     @Override
     public void handleNotification(INotification notification, Object sender)
     {

src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java

@@ -414,7 +414,7 @@ public boolean isValid(boolean validateChecksum, Consumer<IndexComponent> onInva
             for (IndexComponentType expected : expectedComponentsForVersion())
             {
                 var component = components.get(expected);
-                if (component == null)
+                if (component == null || !component.file().exists())
                 {
                     logger.warn(logMessage("Missing index component {} from SSTable {}"), expected, descriptor);
                     isValid = false;

test/unit/org/apache/cassandra/index/CustomIndexTest.java

@@ -1687,6 +1687,11 @@ public Set<Component> activeComponents(SSTableReader sstable)
             {
                 return Collections.emptySet();
             }
+
+            @Override
+            public void validateComponents(SSTableReader sstable, boolean validateChecksum)
+            {
+            }
         }
     }
 }