CNDB-15135: use EmptyFactory to avoid loading SAI per-sstable files if SAI_INDEX_READS_DISABLED is true (#1958)

### What is the issue
CNDB-15135

### What does this PR fix and why was it fixed
Avoid reading SAI per-sstable files if SAI_INDEX_READS_DISABLED is true

Author: jasonstack

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

@@ -38,6 +38,7 @@
 import org.slf4j.LoggerFactory;
 
 import io.github.jbellis.jvector.vector.VectorSimilarityFunction;
+import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.cql3.statements.schema.IndexTarget;
 import org.apache.cassandra.db.ClusteringComparator;
@@ -91,6 +92,7 @@
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
+import static org.apache.cassandra.config.CassandraRelevantProperties.SAI_INDEX_READS_DISABLED;
 import static org.apache.cassandra.config.CassandraRelevantProperties.VALIDATE_MAX_TERM_SIZE_AT_COORDINATOR;
 
 /**
@@ -934,8 +936,15 @@ public Pair<Set<SSTableIndex>, Set<SSTableContext>> getBuiltIndexes(Collection<S
                 }
 
                 SSTableIndex index = new SSTableIndex(context, perIndexComponents);
-                long count = context.primaryKeyMapFactory().count();
-                logger.debug(logMessage("Successfully loaded index for SSTable {} with {} rows."), context.descriptor(), count);
+                if (SAI_INDEX_READS_DISABLED.getBoolean())
+                {
+                    logger.debug(logMessage("Skipped loading index for SSTable {} as it's disabled by {}"), context.descriptor(), SAI_INDEX_READS_DISABLED.getKey());
+                }
+                else
+                {
+                    long count = context.primaryKeyMapFactory().count();
+                    logger.debug(logMessage("Successfully loaded index for SSTable {} with {} rows."), context.descriptor(), count);
+                }
 
                 // Try to add new index to the set, if set already has such index, we'll simply release and move on.
                 // This covers situation when SSTable collection has the same SSTable multiple

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

@@ -19,6 +19,7 @@
 
 import com.google.common.base.Objects;
 
+import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.index.sai.disk.PrimaryKeyMap;
 import org.apache.cassandra.index.sai.disk.format.IndexComponents;
 import org.apache.cassandra.index.sai.utils.PrimaryKey;
@@ -68,6 +69,7 @@ private SSTableContext(SSTableContext copy)
     public static SSTableContext create(SSTableReader sstable, IndexComponents.ForRead perSSTableComponents)
     {
         var onDiskFormat = perSSTableComponents.onDiskFormat();
+        // no disk access thus no need to use EmptyFactory
         PrimaryKey.Factory primaryKeyFactory = onDiskFormat.newPrimaryKeyFactory(sstable.metadata().comparator);
 
         Ref<? extends SSTableReader> sstableRef = null;
@@ -82,7 +84,10 @@ public static SSTableContext create(SSTableReader sstable, IndexComponents.ForRe
                 throw new IllegalStateException("Couldn't acquire reference to the sstable: " + sstable);
             }
 
-            primaryKeyMapFactory = onDiskFormat.newPrimaryKeyMapFactory(perSSTableComponents, primaryKeyFactory, sstable);
+            // avoid opening SAI metadata if reads are disabled
+            primaryKeyMapFactory = CassandraRelevantProperties.SAI_INDEX_READS_DISABLED.getBoolean()
+                                   ? new PrimaryKeyMap.DummyThrowingFactory()
+                                   : onDiskFormat.newPrimaryKeyMapFactory(perSSTableComponents, primaryKeyFactory, sstable);
 
             Cleanup cleanup = new Cleanup(primaryKeyMapFactory, sstableRef);
 

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

@@ -124,4 +124,28 @@ default void close() throws IOException
     default void close() throws IOException
     {
     }
+
+    /**
+     * When SAI_INDEX_READS_DISABLED is true, this is used to avoid loading SAI files to reduce disk access and memory usage
+     */
+    class DummyThrowingFactory implements Factory
+    {
+        @Override
+        public PrimaryKeyMap newPerSSTablePrimaryKeyMap()
+        {
+            throw new UnsupportedOperationException("EmptyFactory doesn't support newPerSSTablePrimaryKeyMap()");
+        }
+
+        @Override
+        public long count()
+        {
+            throw new UnsupportedOperationException("EmptyFactory doesn't support count()");
+        }
+
+        @Override
+        public void close() throws IOException
+        {
+            // no-op
+        }
+    }
 }