CNDB-13534: Add RequestFailureReason.FEATURE_NEEDS_INDEX_REBUILD

The new request failure reason should be used when the on-disk format of an
index in a replica doesn't support a requested feature.

Also, rename org.apache.cassandra.index.sai.disk.format.Version.latest() to Version.current(),
and add separate Version.CURRENT and Version.LATEST constants.

Author: adelapena

src/java/org/apache/cassandra/config/CassandraRelevantProperties.java

@@ -678,6 +678,8 @@ public enum CassandraRelevantProperties
 
     /** Class used to discover/load the proper SAI index components file for a given sstable. */
     SAI_ANN_USE_SYNTHETIC_SCORE("cassandra.sai.ann_use_synthetic_score", "false"),
+    /** The current version of the SAI on-disk index format. */
+    SAI_CURRENT_VERSION("cassandra.sai.latest.version", "dc"),
     SAI_CUSTOM_COMPONENTS_DISCOVERY_CLASS("cassandra.sai.custom_components_discovery_class"),
     SAI_ENABLE_EDGES_CACHE("cassandra.sai.enable_edges_cache", "false"),
     SAI_ENABLE_GENERAL_ORDER_BY("cassandra.sai.general_order_by", "true"),

src/java/org/apache/cassandra/exceptions/RequestFailureReason.java

@@ -24,6 +24,7 @@
 import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
 import org.apache.cassandra.index.IndexBuildInProgressException;
 import org.apache.cassandra.index.IndexNotAvailableException;
+import org.apache.cassandra.index.FeatureNeedsIndexRebuildException;
 import org.apache.cassandra.index.sai.utils.AbortedOperationException;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
@@ -47,7 +48,8 @@ public enum RequestFailureReason
     UNKNOWN_COLUMN           (500),
     UNKNOWN_TABLE            (501),
     REMOTE_STORAGE_FAILURE   (502),
-    INDEX_BUILD_IN_PROGRESS  (503);
+    INDEX_BUILD_IN_PROGRESS  (503),
+    FEATURE_NEEDS_INDEX_REBUILD(504); // The index uses an old version that doesn't support the requested feature
 
     public static final Serializer serializer = new Serializer();
 
@@ -85,6 +87,7 @@ public int codeForNativeProtocol()
         exceptionToReasonMap.put(UnknownColumnException.class, UNKNOWN_COLUMN);
         exceptionToReasonMap.put(UnknownTableException.class, UNKNOWN_TABLE);
         exceptionToReasonMap.put(IndexBuildInProgressException.class, INDEX_BUILD_IN_PROGRESS);
+        exceptionToReasonMap.put(FeatureNeedsIndexRebuildException.class, FEATURE_NEEDS_INDEX_REBUILD);
 
         if (exceptionToReasonMap.size() != reasons.length-5)
             throw new RuntimeException("A new RequestFailureReasons was probably added and you may need to update the exceptionToReasonMap");

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

@@ -0,0 +1,44 @@
+/*
+ * 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.index;
+
+import org.apache.cassandra.exceptions.InternalRequestExecutionException;
+import org.apache.cassandra.exceptions.RequestFailureReason;
+
+/**
+ * Thrown if a secondary index is still in an old version that doesn't support the requested feature.
+ * </p>
+ * Users hitting this exception should probably rebuild their index to a newer version.
+ */
+public final class FeatureNeedsIndexRebuildException extends RuntimeException implements InternalRequestExecutionException
+{
+    /**
+     * Creates a new {@link FeatureNeedsIndexRebuildException} for the specified message.
+     *
+     * @param message the explanation of the error
+     */
+    public FeatureNeedsIndexRebuildException(String message)
+    {
+        super(message);
+    }
+
+    @Override
+    public RequestFailureReason getReason()
+    {
+        return RequestFailureReason.FEATURE_NEEDS_INDEX_REBUILD;
+    }
+}

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

@@ -910,7 +910,7 @@ default void unload() { }
          * Returns the set of sstable-attached components that this group will create for a newly flushed sstable.
          *
          * Note that the result of this method is only valid for newly flushed/written sstables as the components
-         * returned will assume a version of {@link Version#latest()} and a generation of 0. SSTables for which some
+         * returned will assume a version of {@link Version#current()} and a generation of 0. SSTables for which some
          * index have been rebuild may have index components that do not match what this method return in particular.
          */
         Set<Component> componentsForNewSSTable();

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

@@ -181,7 +181,7 @@ public IndexContext(@Nonnull String keyspace,
         this.viewManager = new IndexViewManager(this);
         this.validator = TypeUtil.cellValueType(column, indexType);
         this.cfs = cfs;
-        this.primaryKeyFactory = Version.latest().onDiskFormat().newPrimaryKeyFactory(clusteringComparator);
+        this.primaryKeyFactory = Version.current().onDiskFormat().newPrimaryKeyFactory(clusteringComparator);
 
         String columnName = column.name.toString();
 
@@ -662,7 +662,7 @@ public View getView()
      */
     public int openPerIndexFiles()
     {
-        return viewManager.getView().size() * Version.latest().onDiskFormat().openFilesPerIndex(this);
+        return viewManager.getView().size() * Version.current().onDiskFormat().openFilesPerIndex(this);
     }
 
     public void prepareSSTablesForRebuild(Collection<SSTableReader> sstablesToRebuild)

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

@@ -76,6 +76,7 @@
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.index.IndexBuildDecider;
 import org.apache.cassandra.index.IndexRegistry;
+import org.apache.cassandra.index.FeatureNeedsIndexRebuildException;
 import org.apache.cassandra.index.SecondaryIndexBuilder;
 import org.apache.cassandra.index.SecondaryIndexManager;
 import org.apache.cassandra.index.TargetParser;
@@ -85,6 +86,7 @@
 import org.apache.cassandra.index.sai.analyzer.NonTokenizingOptions;
 import org.apache.cassandra.index.sai.disk.StorageAttachedIndexWriter;
 import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
+import org.apache.cassandra.index.sai.disk.format.Version;
 import org.apache.cassandra.index.sai.disk.v1.IndexWriterConfig;
 import org.apache.cassandra.index.sai.utils.TypeUtil;
 import org.apache.cassandra.index.sai.view.View;
@@ -447,7 +449,17 @@ private Future<?> startInitialBuild(ColumnFamilyStore baseCfs, boolean validate,
             return ImmediateFuture.success(null);
         }
 
-        // stop in-progress compaction tasks to prevent compacted sstable not being index.
+        if (indexContext.isVector() && Version.current().compareTo(Version.JVECTOR_EARLIEST) < 0)
+        {
+            throw new FeatureNeedsIndexRebuildException(String.format("The current configured on-disk format version %s does not support vector indexes. " +
+                                                                      "The minimum version that supports vectors is %s. " +
+                                                                      "The on-disk format version can be set via the -D%s system property.",
+                                                                      Version.current(),
+                                                                      Version.JVECTOR_EARLIEST,
+                                                                      CassandraRelevantProperties.SAI_CURRENT_VERSION.name()));
+        }
+
+        // stop in-progress compaction tasks to prevent compacted sstables not being indexed.
         logger.debug(indexContext.logMessage("Stopping active compactions to make sure all sstables are indexed after initial build."));
         CompactionManager.instance.interruptCompactionFor(Collections.singleton(baseCfs.metadata()),
                                                           OperationType.REWRITES_SSTABLES,

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

@@ -290,10 +290,10 @@ private static void prepareForRebuild(IndexComponents.ForRead components, Set<Co
         // The current components are "replaced" (by "other" components) if the build create different components than
         // the existing ones. This will happen in the following cases:
         // 1. if we use immutable components, that's the point of immutable components.
-        // 2. when we do not use immutable components, the rebuild components will always be for the latest version and
+        // 2. when we do not use immutable components, the rebuild components will always be for the current version and
         // for generation 0, so if the current components are not for that specific built, then we won't be rebuilding
         // the exact same components, and we're "replacing", not "overwriting" ()
-        //   a) the old components are from an older version: a new build will alawys be for `Version.latest()` and
+        //   a) the old components are from an older version: a new build will alawys be for `Version.current()` and
         //     so will create new files in that case (Note that "normally" we should not have non-0 generation in the
         //     first place if immutable components are not used, but we handle this case to better support "downgrades"
         //     where immutable components was enabled, but then disabled for some reason. If that happens, we still

src/java/org/apache/cassandra/index/sai/disk/StorageAttachedIndexWriter.java

@@ -90,9 +90,9 @@ public StorageAttachedIndexWriter(IndexDescriptor indexDescriptor,
                                       TableMetrics tableMetrics) throws IOException
     {
         // We always write at the latest version (through what that version is can be configured for specific cases)
-        var onDiskFormat = Version.latest().onDiskFormat();
+        var onDiskFormat = Version.current().onDiskFormat();
         this.indexDescriptor = indexDescriptor;
-        // Note: I think there is a silent assumption here. That is, the PK factory we use here must be for the latest
+        // Note: I think there is a silent assumption here. That is, the PK factory we use here must be for the current
         // format version, because that is what `IndexContext.keyFactory` always uses (see ctor)
         this.primaryKeyFactory = onDiskFormat.newPrimaryKeyFactory(tableMetadata.comparator);
         this.indices = indices;

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

@@ -31,7 +31,7 @@
  */
 public class ComponentsBuildId implements Comparable<ComponentsBuildId>
 {
-    private static final ComponentsBuildId FOR_NEW_SSTABLE = ComponentsBuildId.latest(0);
+    private static final ComponentsBuildId FOR_NEW_SSTABLE = ComponentsBuildId.current(0);
 
     private final Version version;
     private final int generation;
@@ -47,9 +47,9 @@ public static ComponentsBuildId of(Version version, int generation)
         return new ComponentsBuildId(version, generation);
     }
 
-    public static ComponentsBuildId latest(int generation)
+    public static ComponentsBuildId current(int generation)
     {
-        return of(Version.latest(), generation);
+        return of(Version.current(), generation);
     }
 
     public static ComponentsBuildId forNewSSTable()
@@ -59,7 +59,7 @@ public static ComponentsBuildId forNewSSTable()
 
     public static ComponentsBuildId forNewBuild(@Nullable ComponentsBuildId previousBuild, Predicate<ComponentsBuildId> newBuildIsUsablePredicate)
     {
-        Version version = Version.latest();
+        Version version = Version.current();
         // If we're not using immutable components, we always use generation 0, and we're fine if that overrides existing files
         if (!version.useImmutableComponentFiles())
             return new ComponentsBuildId(version, 0);

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

@@ -76,7 +76,7 @@
  * of the completion marker, but not of the other component of the group). The bumping of generations takes incomplete
  * groups into account, and so incomplete groups are not overridden either. Essentially, the generation used by a new
  * build is always one more than the highest generation of any component found on disk (for the group in question, and
- * the version we writting, usually {@link Version#latest()}).
+ * the version we writting, usually {@link Version#current()}).
  */
 public interface IndexComponents
 {

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

@@ -79,7 +79,7 @@ public class IndexDescriptor
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
     private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES);
 
-    private static final ComponentsBuildId EMPTY_GROUP_MARKER = ComponentsBuildId.latest(-1);
+    private static final ComponentsBuildId EMPTY_GROUP_MARKER = ComponentsBuildId.current(-1);
 
     // 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

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

@@ -52,9 +52,9 @@ public interface IndexFeatureSet
      * multiple sources. This will include all the SSTables included in a query and all the indexes
      * attached to those SSTables, added using {@link Accumulator#accumulate}.
      * <p>
-     * The feature set of the latest version denoted by {@link Version#latest()}
+     * The feature set of the current version denoted by {@link Version#current()}
      * is implicitly added, so the result feature set will include only the features supported by the
-     * latest version.
+     * current version.
      * <p>
      * The {@code Accumulator} creates an {@code IndexFeatureSet} this contains the features from
      * all the associated feature sets where {@code false} is the highest priority. This means if any
@@ -70,7 +70,7 @@ class Accumulator
 
         public Accumulator()
         {
-            accumulate(Version.latest().onDiskFormat().indexFeatureSet());
+            accumulate(Version.current().onDiskFormat().indexFeatureSet());
         }
 
         /**

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

@@ -61,19 +61,22 @@ 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
+    // term frequencies index component (support for BM25)
     public static final Version EC = new Version("ec", V7OnDiskFormat.instance, (c, i, g) -> stargazerFileNameFormat(c, i, g, "ec"));
 
     // These are in reverse-chronological order so that the latest version is first. Version matching tests
-    // are more likely to match the latest version so we want to test that one first.
+    // are more likely to match the latest version, so we want to test that one first.
     public static final List<Version> ALL = Lists.newArrayList(EC, EB, DC, DB, CA, BA, AA);
 
     public static final Version EARLIEST = AA;
     public static final Version VECTOR_EARLIEST = BA;
-    // The latest version can be configured to be an earlier version to support partial upgrades that don't
+    public static final Version JVECTOR_EARLIEST = CA;
+    public static final Version BM25_EARLIEST = EC;
+    public static final Version LATEST = EC;
+    // The current version can be configured to be an earlier version to support partial upgrades that don't
     // write newer versions of the on-disk formats. This is volatile rather than final so that tests may
     // use reflection to change it and safely publish across threads.
-    private static volatile Version LATEST = parse(CassandraRelevantProperties.SAI_LATEST_VERSION.getString());
+    public static volatile Version CURRENT = parse(currentVersionProperty());
 
     private static final Pattern GENERATION_PATTERN = Pattern.compile("\\d+");
 
@@ -88,6 +91,11 @@ private Version(String version, OnDiskFormat onDiskFormat, FileNameFormatter fil
         this.fileNameFormatter = fileNameFormatter;
     }
 
+    private static String currentVersionProperty()
+    {
+        return CassandraRelevantProperties.SAI_CURRENT_VERSION.getString();
+    }
+
     public static Version parse(String input)
     {
         checkArgument(input != null);
@@ -99,9 +107,9 @@ public static Version parse(String input)
         throw new IllegalArgumentException("Unrecognized SAI version string " + input);
     }
 
-    public static Version latest()
+    public static Version current()
     {
-        return LATEST;
+        return CURRENT;
     }
 
     /**
@@ -125,7 +133,7 @@ public static int calculateIndexNameAllowedLength()
     private static int getAddedLengthFromDescriptorAndVersion()
     {
         // Prefixes and suffixes constructed by Version.stargazerFileNameFormat
-        int versionNameLength = latest().toString().length();
+        int versionNameLength = current().toString().length();
         // room for up to 999 generations
         int generationLength = 3 + SAI_SEPARATOR.length();
         int addedLength = SAI_DESCRIPTOR.length()

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

@@ -30,6 +30,7 @@
 import java.util.stream.Collectors;
 
 import com.google.common.base.MoreObjects;
+import org.apache.cassandra.index.sai.plan.QueryController;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,6 +43,7 @@
 import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.index.FeatureNeedsIndexRebuildException;
 import org.apache.cassandra.index.sai.IndexContext;
 import org.apache.cassandra.index.sai.QueryContext;
 import org.apache.cassandra.index.sai.SSTableContext;
@@ -187,7 +189,10 @@ public CloseableIterator<PrimaryKeyWithSortKey> orderBy(Orderer orderer, Express
             return toMetaSortedIterator(iter, queryContext);
         }
         if (docLengthsMeta == null)
-            throw new InvalidRequestException(indexContext.getIndexName() + " does not support BM25 scoring until it is rebuilt");
+        {
+            throw new FeatureNeedsIndexRebuildException(String.format(QueryController.INDEX_VERSION_DOES_NOT_SUPPORT_BM25,
+                                                                      indexContext.getIndexName()));
+        }
 
         // find documents that match each term
         var queryTerms = orderer.getQueryTerms();
@@ -264,8 +269,12 @@ public CloseableIterator<PrimaryKeyWithSortKey> orderResultsBy(SSTableReader rea
     {
         if (!orderer.isBM25())
             return super.orderResultsBy(reader, queryContext, keys, orderer, limit);
+        
         if (docLengthsMeta == null)
-            throw new InvalidRequestException(indexContext.getIndexName() + " does not support BM25 scoring until it is rebuilt");
+        {
+            throw new InvalidRequestException(String.format(QueryController.INDEX_VERSION_DOES_NOT_SUPPORT_BM25,
+                                                            indexContext.getIndexName()));
+        }
 
         var queryTerms = orderer.getQueryTerms();
         // compute documentFrequencies from either histogram or an index search

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

@@ -212,7 +212,7 @@ private long flush(DecoratedKey minKey, DecoratedKey maxKey, AbstractType<?> ter
 
     private boolean writeFrequencies()
     {
-        return indexContext().isAnalyzed() && Version.latest().onOrAfter(Version.EC);
+        return indexContext().isAnalyzed() && Version.current().onOrAfter(Version.BM25_EARLIEST);
     }
 
     private void flushVectorIndex(DecoratedKey minKey, DecoratedKey maxKey, long startTime, Stopwatch stopwatch) throws IOException

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

@@ -202,7 +202,7 @@ public static class RAMStringSegmentBuilder extends SegmentBuilder
 
         private boolean writeFrequencies()
         {
-            return !(analyzer instanceof NoOpAnalyzer) && Version.latest().onOrAfter(Version.EC);
+            return !(analyzer instanceof NoOpAnalyzer) && Version.current().onOrAfter(Version.BM25_EARLIEST);
         }
 
         public boolean isEmpty()
@@ -499,8 +499,8 @@ private long add(List<ByteBuffer> terms, PrimaryKey key, long sstableRowId)
         // Update term boundaries for all terms in this row
         for (ByteBuffer term : terms)
         {
-            minTerm = TypeUtil.min(term, minTerm, termComparator, Version.latest());
-            maxTerm = TypeUtil.max(term, maxTerm, termComparator, Version.latest());
+            minTerm = TypeUtil.min(term, minTerm, termComparator, Version.current());
+            maxTerm = TypeUtil.max(term, maxTerm, termComparator, Version.current());
         }
 
         // segmentRowIdOffset should encode sstableRowId into Integer