CNDB-14501: Enable checksum for jvector index files starting at version EC (#1841)

michaeljmarshall · driftx · commit 37372a737dd4 · 2025-07-23T06:05:06.000-05:00
Fixes: riptano/cndb#14501 CNDB test pr: riptano/cndb#14660 This PR introduces vector index checksum validation starting with files in version `EC` at above. For non terms files, we compute the checksum correctly. For terms files, we skip validation of the checksum because it is not computed correctly. There is a new issue to track fixing the checksum logic on those TERMS files riptano/cndb#14656. I implemented the version-dependent validation by overriding `V7OnDiskFormat#validateIndexComponent`. This allows us to validate new files while skipping the old ones. In doing so, I removed the misleading `IndexFeatureSet` method for checksums. The modified comment describes why that feature didn't make sense as an index-level feature.
diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
@@ -34,6 +34,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -411,9 +412,18 @@ public boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables
 
         for (SSTableReader sstable : sstables)
         {
-            IndexDescriptor indexDescriptor = contextManager.getOrLoadIndexDescriptor(sstable, indices);
+            logger.debug("Validating {} SAI indices for {}", indices.size(), sstable.getFilename());
+            if (indices.isEmpty())
+            {
+                logger.debug("No SAI indices to validate for {}", sstable.getFilename());
+                return true;
+            }
+            
+            // For validation, we need to load a fresh descriptor to ensure we see the current state of files
+            IndexDescriptor indexDescriptor = IndexDescriptor.load(sstable, contexts());
             IndexComponents.ForRead perSSTableComponents = indexDescriptor.perSSTableComponents();
 
+            logger.debug("Per-SSTable components complete: {}", indexDescriptor.perSSTableComponents().isComplete());
             if (indexDescriptor.perSSTableComponents().isComplete())
             {
                 perSSTableComponents.validateComponents(sstable, baseCfs.getTracker(), validateChecksum, true);
@@ -422,6 +432,7 @@ public boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables
                 {
                     IndexComponents.ForRead perIndexComponents = indexDescriptor.perIndexComponents(index.getIndexContext());
 
+                    logger.debug("Per-index components complete for {}: {}", index.getIndexContext().getIndexName(), perIndexComponents.isComplete());
                     if (perIndexComponents.isComplete())
                         perIndexComponents.validateComponents(sstable, baseCfs.getTracker(), validateChecksum, true);
                     else if (throwOnIncomplete)
@@ -555,4 +566,12 @@ public void reset()
         indices.forEach(index -> index.makeIndexNonQueryable());
         onSSTableChanged(baseCfs.getLiveSSTables(), Collections.emptySet(), indices, false);
     }
+
+    private Set<IndexContext> contexts()
+    {
+        Set<IndexContext> contexts = Sets.newHashSetWithExpectedSize(indices.size());
+        for (StorageAttachedIndex index : indices)
+            contexts.add(index.getIndexContext());
+        return contexts;
+    }
 }
diff --git a/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java b/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java
@@ -85,7 +85,7 @@ public class IndexDescriptor
     // TODO Because indexes can be added at any time to existing data, the Version of a column index
     // may not match the Version of the base sstable.  OnDiskFormat + IndexFeatureSet + IndexDescriptor
     // was not designed with this in mind, leading to some awkwardness, notably in IFS where some features
-    // are per-sstable (`isRowAware`) and some are per-column (`hasVectorIndexChecksum`).
+    // are per-sstable (`isRowAware`) and some are per-column (`hasTermsHistogram`).
 
     public final Descriptor descriptor;
 
diff --git a/src/java/org/apache/cassandra/index/sai/disk/format/IndexFeatureSet.java b/src/java/org/apache/cassandra/index/sai/disk/format/IndexFeatureSet.java
@@ -37,11 +37,6 @@ public interface IndexFeatureSet
      */
     boolean isRowAware();
 
-    /**
-     * @return true if vector index files include a checksum at the end
-     */
-    boolean hasVectorIndexChecksum();
-
     /**
      * @return true if index metadata contains term histograms for fast cardinality estimation
      */
@@ -64,7 +59,6 @@ public interface IndexFeatureSet
     class Accumulator
     {
         boolean isRowAware = true;
-        boolean hasVectorIndexChecksum = true;
         boolean hasTermsHistogram = true;
         boolean complete = false;
 
@@ -83,8 +77,6 @@ public void accumulate(IndexFeatureSet indexFeatureSet)
             assert !complete : "Cannot accumulate after complete has been called";
             if (!indexFeatureSet.isRowAware())
                 isRowAware = false;
-            if (!indexFeatureSet.hasVectorIndexChecksum())
-                hasVectorIndexChecksum = false;
             if (!indexFeatureSet.hasTermsHistogram())
                 hasTermsHistogram = false;
         }
@@ -106,12 +98,6 @@ public boolean isRowAware()
                     return isRowAware;
                 }
 
-                @Override
-                public boolean hasVectorIndexChecksum()
-                {
-                    return hasVectorIndexChecksum;
-                }
-
                 @Override
                 public boolean hasTermsHistogram()
                 {
diff --git a/src/java/org/apache/cassandra/index/sai/disk/format/Version.java b/src/java/org/apache/cassandra/index/sai/disk/format/Version.java
@@ -52,7 +52,9 @@ public class Version implements Comparable<Version>
     public static final Version AA = new Version("aa", V1OnDiskFormat.instance, Version::aaFileNameFormat);
     // Stargazer
     public static final Version BA = new Version("ba", V2OnDiskFormat.instance, (c, i, g) -> stargazerFileNameFormat(c, i, g, "ba"));
-    // Converged Cassandra with JVector
+    // Converged Cassandra with JVector with file format version 2
+    // Note: vector index checksums for TERMS files were computed in two different ways for this version. As such,
+    // we do not validate checksums for this version or any subsequent version until EC.
     public static final Version CA = new Version("ca", V3OnDiskFormat.instance, (c, i, g) -> stargazerFileNameFormat(c, i, g, "ca"));
     // NOTE: use DB to prevent collisions with upstream file formats
     // Encode trie entries using their AbstractType to ensure trie entries are sorted for range queries and are prefix free.
@@ -61,7 +63,9 @@ public class Version implements Comparable<Version>
     public static final Version DC = new Version("dc", V5OnDiskFormat.instance, (c, i, g) -> stargazerFileNameFormat(c, i, g, "dc"));
     // histograms in index metadata
     public static final Version EB = new Version("eb", V6OnDiskFormat.instance, (c, i, g) -> stargazerFileNameFormat(c, i, g, "eb"));
-    // term frequencies index component (support for BM25)
+    // term frequencies index component (support for BM25); bump jvector file format version to 4
+    // Start validating vector index component checksums, except for the TERMS_FILE because it's checksum is non-standard
+    // and isn't easily validated when an sstable index has multiple segments within the TERMS_FILE.
     public static final Version EC = new Version("ec", V7OnDiskFormat.instance, (c, i, g) -> stargazerFileNameFormat(c, i, g, "ec"));
     // total terms count serialization in index metadata
     public static final Version ED = new Version("ed", V7OnDiskFormat.instance, (c, i, g) -> stargazerFileNameFormat(c, i, g, "ed"));
diff --git a/src/java/org/apache/cassandra/index/sai/disk/v1/V1OnDiskFormat.java b/src/java/org/apache/cassandra/index/sai/disk/v1/V1OnDiskFormat.java
@@ -133,12 +133,6 @@ public boolean isRowAware()
             return false;
         }
 
-        @Override
-        public boolean hasVectorIndexChecksum()
-        {
-            return false;
-        }
-
         @Override
         public boolean hasTermsHistogram()
         {
@@ -236,13 +230,10 @@ public void validateIndexComponent(IndexComponent.ForRead component, boolean che
         if (component.isCompletionMarker())
             return;
 
-        // starting with v3, vector components include proper headers and checksum; skip for earlier versions
+        // We do not validate vector components until V7, so we skip for earlier versions
         IndexContext context = component.parent().context();
-        if (isVectorDataComponent(context, component.componentType())
-            && !component.parent().onDiskFormat().indexFeatureSet().hasVectorIndexChecksum())
-        {
+        if (isVectorDataComponent(context, component.componentType()))
             return;
-        }
 
         Version earliest = getExpectedEarliestVersion(context, component.componentType());
         try (IndexInput input = component.openInput())
diff --git a/src/java/org/apache/cassandra/index/sai/disk/v2/V2OnDiskFormat.java b/src/java/org/apache/cassandra/index/sai/disk/v2/V2OnDiskFormat.java
@@ -74,12 +74,6 @@ public boolean isRowAware()
             return true;
         }
 
-        @Override
-        public boolean hasVectorIndexChecksum()
-        {
-            return false;
-        }
-
         @Override
         public boolean hasTermsHistogram()
         {
diff --git a/src/java/org/apache/cassandra/index/sai/disk/v3/V3OnDiskFormat.java b/src/java/org/apache/cassandra/index/sai/disk/v3/V3OnDiskFormat.java
@@ -82,12 +82,6 @@ public boolean isRowAware()
             return true;
         }
 
-        @Override
-        public boolean hasVectorIndexChecksum()
-        {
-            return false;
-        }
-
         @Override
         public boolean hasTermsHistogram()
         {
diff --git a/src/java/org/apache/cassandra/index/sai/disk/v6/V6OnDiskFormat.java b/src/java/org/apache/cassandra/index/sai/disk/v6/V6OnDiskFormat.java
@@ -33,12 +33,6 @@ public boolean isRowAware()
             return true;
         }
 
-        @Override
-        public boolean hasVectorIndexChecksum()
-        {
-            return false;
-        }
-
         @Override
         public boolean hasTermsHistogram()
         {
diff --git a/src/java/org/apache/cassandra/index/sai/disk/v7/V7OnDiskFormat.java b/src/java/org/apache/cassandra/index/sai/disk/v7/V7OnDiskFormat.java
@@ -18,16 +18,30 @@
 
 package org.apache.cassandra.index.sai.disk.v7;
 
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.lang.invoke.MethodHandles;
 import java.util.EnumSet;
 import java.util.Set;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.index.sai.IndexContext;
+import org.apache.cassandra.index.sai.disk.format.IndexComponent;
 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.v6.V6OnDiskFormat;
+import org.apache.cassandra.index.sai.utils.SAICodecUtils;
 import org.apache.cassandra.index.sai.utils.TypeUtil;
+import org.apache.cassandra.utils.Throwables;
+import org.apache.lucene.store.IndexInput;
 
 public class V7OnDiskFormat extends V6OnDiskFormat
 {
+    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
     public static final V7OnDiskFormat instance = new V7OnDiskFormat();
 
     private static final Set<IndexComponentType> LITERAL_COMPONENTS = EnumSet.of(IndexComponentType.COLUMN_COMPLETION_MARKER,
@@ -56,4 +70,45 @@ public int jvectorFileFormatVersion()
         // compatibility.
         return 4;
     }
+
+    @Override
+    public void validateIndexComponent(IndexComponent.ForRead component, boolean checksum)
+    {
+        if (component.isCompletionMarker())
+            return;
+
+        IndexContext context = component.parent().context();
+        if (context != null && context.isVector() && component.parent().version().onOrAfter(Version.EC))
+        {
+            try (IndexInput input = component.openInput())
+            {
+                // We can't validate TERMS_DATA with checksum because the checksum was computed incorrectly through
+                // V7. See https://github.com/riptano/cndb/issues/14656. We can still call the basic validate method
+                // which does not check the checksum. (The issue is in the way the checksum was computed. It didn't
+                // include the header/footer bytes, and for multi-segment builds, it didn't include the bytes from
+                // all previous segments, which is the design for all index components to date.)
+                if (!checksum || component.componentType() == IndexComponentType.TERMS_DATA)
+                    SAICodecUtils.validate(input, getExpectedEarliestVersion(context, component.componentType()));
+                else
+                    SAICodecUtils.validateChecksum(input, getExpectedEarliestVersion(context, component.componentType()));
+            }
+            catch (Throwable e)
+            {
+                logger.warn(component.parent().logMessage("{} failed for index component {} on SSTable {}"),
+                            (checksum ? "Checksum validation" : "Validation"),
+                            component,
+                            component.parent().descriptor(),
+                            e);
+                if (e instanceof IOException)
+                    throw new UncheckedIOException((IOException) e);
+                if (e.getCause() instanceof IOException)
+                    throw new UncheckedIOException((IOException) e.getCause());
+                throw Throwables.unchecked(e);
+            }
+        }
+        else
+        {
+            super.validateIndexComponent(component, checksum);
+        }
+    }
 }
diff --git a/src/java/org/apache/cassandra/index/sai/utils/SAICodecUtils.java b/src/java/org/apache/cassandra/index/sai/utils/SAICodecUtils.java
@@ -75,6 +75,8 @@ public static void writeFooter(IndexOutput out) throws IOException
         writeChecksum(out);
     }
 
+    // Warning: this method produces an incomplete checksum when using other Lucene tooling because it computes
+    // the checksum without including the FOOTER_MAGIC and 0. See https://github.com/riptano/cndb/issues/14501.
     public static void writeFooter(RandomAccessWriter braw, long checksum) throws IOException
     {
         var out = toLuceneOutput(braw);
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SortedTableVerifier.java b/src/java/org/apache/cassandra/io/sstable/format/SortedTableVerifier.java
@@ -19,6 +19,7 @@
 package org.apache.cassandra.io.sstable.format;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.Collections;
@@ -48,6 +49,7 @@
 import org.apache.cassandra.dht.LocalPartitioner;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.index.SecondaryIndexManager;
 import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.io.sstable.IVerifier;
 import org.apache.cassandra.io.sstable.KeyIterator;
@@ -156,6 +158,8 @@ public void verify()
 
         verifyBloomFilter();
 
+        verifyStorageAttachedIndexes();
+
         if (options.checkOwnsTokens && !isOffline && !(sstable.getPartitioner() instanceof LocalPartitioner))
         {
             if (verifyOwnedRanges() == 0)
@@ -187,6 +191,52 @@ protected void verifyBloomFilter()
         }
     }
 
+    protected void verifyStorageAttachedIndexes()
+    {
+        if (realm == null || isOffline)
+            return;
+
+        try
+        {
+            SecondaryIndexManager indexManager = realm.getIndexManager();
+            if (indexManager != null)
+            {
+                // Quick verification skips checksum validation
+                boolean validateChecksum = !options.quick;
+                outputHandler.debug("Validating storage attached indexes for %s", sstable);
+                logger.debug("SortedTableVerifier: Validating SAI for {}", sstable.getFilename());
+                // Pass true for throwOnIncomplete to ensure incomplete indexes throw exceptions
+                boolean complete = indexManager.validateSSTableAttachedIndexes(Collections.singleton(sstable), true, validateChecksum);
+                if (!complete)
+                {
+                    // This shouldn't happen since we're throwing on incomplete, but just in case
+                    throw new IOException("SAI validation failed for " + sstable.getFilename());
+                }
+            }
+            else
+            {
+                logger.debug("SortedTableVerifier: No index manager for {}", sstable.getFilename());
+            }
+        }
+        catch (IllegalStateException e)
+        {
+            // Convert IllegalStateException from incomplete indexes to CorruptSSTableException
+            outputHandler.warn(e);
+            markAndThrow(new CorruptSSTableException(e, sstable.getFilename()));
+        }
+        catch (UncheckedIOException e)
+        {
+            // Convert UncheckedIOException from SAI validation to CorruptSSTableException
+            outputHandler.warn(e);
+            markAndThrow(new CorruptSSTableException(e.getCause(), sstable.getFilename()));
+        }
+        catch (Throwable t)
+        {
+            outputHandler.warn(t);
+            markAndThrow(t);
+        }
+    }
+
     protected void verifySSTableMetadata()
     {
         outputHandler.output("Deserializing sstable metadata for %s ", sstable);
diff --git a/test/unit/org/apache/cassandra/index/sai/SAITester.java b/test/unit/org/apache/cassandra/index/sai/SAITester.java
@@ -391,16 +391,9 @@ protected boolean verifyChecksum(IndexContext context)
                 if (indexDescriptor.isIndexEmpty(context))
                     continue;
                 
-                // For vector indexes, only validate checksums if the version supports it
-                boolean validateChecksum = true;
-                if (context.isVector() && !indexDescriptor.perSSTableComponents().onDiskFormat().indexFeatureSet().hasVectorIndexChecksum())
-                {
-                    // Vector components don't have checksums in versions prior to v7, so we skip checksum validation
-                    validateChecksum = false;
-                }
-                
-                if (!indexDescriptor.perSSTableComponents().validateComponents(sstable, cfs.getTracker(), validateChecksum, false)
-                    || !indexDescriptor.perIndexComponents(context).validateComponents(sstable, cfs.getTracker(), validateChecksum, false))
+                // Checksum validation is now handled internally by the OnDiskFormat's validateIndexComponent method
+                if (!indexDescriptor.perSSTableComponents().validateComponents(sstable, cfs.getTracker(), true, false)
+                    || !indexDescriptor.perIndexComponents(context).validateComponents(sstable, cfs.getTracker(), true, false))
                     return false;
             }
         }
diff --git a/test/unit/org/apache/cassandra/index/sai/disk/v7/V7OnDiskFormatIntegrationTest.java b/test/unit/org/apache/cassandra/index/sai/disk/v7/V7OnDiskFormatIntegrationTest.java
diff --git a/test/unit/org/apache/cassandra/index/sai/disk/v7/V7OnDiskFormatTest.java b/test/unit/org/apache/cassandra/index/sai/disk/v7/V7OnDiskFormatTest.java
diff --git a/test/unit/org/apache/cassandra/index/sai/functional/VerifyTest.java b/test/unit/org/apache/cassandra/index/sai/functional/VerifyTest.java
diff --git a/test/unit/org/apache/cassandra/io/sstable/format/SortedTableVerifierTest.java b/test/unit/org/apache/cassandra/io/sstable/format/SortedTableVerifierTest.java

Original file line number	Diff line number	Diff line change
`@@ -74,12 +74,6 @@ public boolean isRowAware()`
`74`	`74`	`return true;`
`75`	`75`	`}`
`76`	`76`
`77`		`- @Override`
`78`		`- public boolean hasVectorIndexChecksum()`
`79`		`- {`
`80`		`- return false;`
`81`		`- }`
`82`		`-`
`83`	`77`	`@Override`
`84`	`78`	`public boolean hasTermsHistogram()`
`85`	`79`	`{`
Original file line number	Diff line number	Diff line change
`@@ -82,12 +82,6 @@ public boolean isRowAware()`
`82`	`82`	`return true;`
`83`	`83`	`}`
`84`	`84`
`85`		`- @Override`
`86`		`- public boolean hasVectorIndexChecksum()`
`87`		`- {`
`88`		`- return false;`
`89`		`- }`
`90`		`-`
`91`	`85`	`@Override`
`92`	`86`	`public boolean hasTermsHistogram()`
`93`	`87`	`{`
Original file line number	Diff line number	Diff line change
`@@ -33,12 +33,6 @@ public boolean isRowAware()`
`33`	`33`	`return true;`
`34`	`34`	`}`
`35`	`35`
`36`		`- @Override`
`37`		`- public boolean hasVectorIndexChecksum()`
`38`		`- {`
`39`		`- return false;`
`40`		`- }`
`41`		`-`
`42`	`36`	`@Override`
`43`	`37`	`public boolean hasTermsHistogram()`
`44`	`38`	`{`
Original file line number	Diff line number	Diff line change
`@@ -75,6 +75,8 @@ public static void writeFooter(IndexOutput out) throws IOException`
`75`	`75`	`writeChecksum(out);`
`76`	`76`	`}`
`77`	`77`
	`78`	`+ // Warning: this method produces an incomplete checksum when using other Lucene tooling because it computes`
	`79`	`+ // the checksum without including the FOOTER_MAGIC and 0. See https://github.com/riptano/cndb/issues/14501.`
`78`	`80`	`public static void writeFooter(RandomAccessWriter braw, long checksum) throws IOException`
`79`	`81`	`{`
`80`	`82`	`var out = toLuceneOutput(braw);`