CNDB-15608 port CASSANDRA-18673 to reduce disk usage of row-aware indexes in 5.0

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

@@ -971,7 +971,10 @@ public Pair<Set<SSTableIndex>, Set<SSTableContext>> getBuiltIndexes(Collection<S
                 else
                 {
                     long count = context.primaryKeyMapFactory().count();
-                    logger.debug(logMessage("Successfully loaded index for SSTable {} with {} rows."), context.descriptor(), count);
+                    if (count > 0)
+                        logger.debug(logMessage("Successfully loaded index for SSTable {} with {} rows."), context.descriptor(), count);
+                    else
+                        logger.debug(logMessage("Skipped loading index for SSTable {} as it is empty."), context.descriptor());
                 }
 
                 // Try to add new index to the set, if set already has such index, we'll simply release and move on.

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

@@ -146,7 +146,8 @@ public PrimaryKeyMap.Factory primaryKeyMapFactory()
      */
     public int openFilesPerSSTable()
     {
-        return perSSTableComponents.onDiskFormat().openFilesPerSSTable();
+        return perSSTableComponents.onDiskFormat()
+                                   .openFilesPerSSTable(sstable.metadata().hasClustering());
     }
 
     @Override

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

@@ -28,6 +28,7 @@
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
+
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -268,7 +269,7 @@ public StorageAttachedIndexQueryPlan queryPlanFor(RowFilter rowFilter)
     @Override
     public SSTableFlushObserver getFlushObserver(Descriptor descriptor, LifecycleNewTracker tracker, TableMetadata tableMetadata, long keyCount)
     {
-        IndexDescriptor indexDescriptor = IndexDescriptor.empty(descriptor);
+        IndexDescriptor indexDescriptor = IndexDescriptor.empty(descriptor, tableMetadata.comparator);
         try
         {
             return new StorageAttachedIndexWriter(indexDescriptor, tableMetadata, indices, tracker, keyCount, baseCfs.metric);
@@ -293,7 +294,7 @@ public boolean handles(IndexTransaction.Type type)
     @Override
     public Set<Component> componentsForNewSSTable()
     {
-        return IndexDescriptor.componentsForNewlyFlushedSSTable(indices, version);
+        return IndexDescriptor.componentsForNewlyFlushedSSTable(indices, version, baseCfs.metadata().hasClustering());
     }
 
     @Override

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

@@ -21,6 +21,7 @@
 
 import com.google.common.base.Stopwatch;
 
+import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.index.sai.utils.PrimaryKey;
 
 /**
@@ -30,7 +31,13 @@ public interface PerSSTableWriter
 {
     public static final PerSSTableWriter NONE = (key) -> {};
 
-    default void startPartition(long position) throws IOException
+    /**
+     * Allows implementations to perform any necessary setup for a new partition.
+     *
+     * @param decoratedKey The key being appended to SSTable.
+     * @param position     The position of the key in the component preferred for reading keys
+     */
+    default void startPartition(DecoratedKey decoratedKey, long position) throws IOException
     {}
 
     void nextRow(PrimaryKey primaryKey) throws IOException;

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

@@ -67,7 +67,7 @@ public class PostingListKeyRangeIterator extends KeyRangeIterator
     private final AtomicBoolean isClosed = new AtomicBoolean(false);
 
     private boolean needsSkipping = false;
-    private PrimaryKey skipToToken = null;
+    private PrimaryKey skipToKey = null;
     private long lastSegmentRowId = -1;
 
     /**
@@ -100,13 +100,13 @@ protected void performSkipTo(PrimaryKey nextKey)
         if (indexContext.getDefinition().isStatic())
             nextKey = nextKey.forStaticRow();
 
-        // If skipToToken is equal to nextKey, we take the nextKey because in practice, it is greater than or equal
-        // to the skipToToken. This is because token only PKs are considered equal to all PKs with the same token,
+        // If skipToKey is equal to nextKey, we take the nextKey because in practice, it is greater than or equal
+        // to the skipToKey. This is because token only PKs are considered equal to all PKs with the same token,
         // and for a range query, we first skip on the token-only PK.
-        if (skipToToken != null && skipToToken.compareTo(nextKey) > 0)
+        if (skipToKey != null && skipToKey.compareTo(nextKey) > 0)
             return;
 
-        skipToToken = nextKey;
+        skipToKey = nextKey;
         needsSkipping = true;
     }
 
@@ -150,16 +150,16 @@ public void close() throws IOException
 
             FileUtils.closeQuietly(postingList, primaryKeyMap);
         }
-        else {
+        else
+        {
             logger.warn("PostingListKeyRangeIterator is already closed",
                         new IllegalStateException("PostingListKeyRangeIterator is already closed"));
         }
-
     }
 
     private boolean exhausted()
     {
-        return needsSkipping && skipToToken.compareTo(getMaximum()) > 0;
+        return needsSkipping && skipToKey.compareTo(getMaximum()) > 0;
     }
 
     /**
@@ -170,8 +170,8 @@ private long getNextRowId() throws IOException
         long segmentRowId;
         if (needsSkipping)
         {
-            long targetSstableRowId = primaryKeyMap.ceiling(skipToToken);
-            // skipToToken is larger than max token in token file
+            long targetSstableRowId = primaryKeyMap.ceiling(skipToKey);
+            // skipToKey is larger than max token in token file
             if (targetSstableRowId < 0)
             {
                 return PostingList.END_OF_STREAM;

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

@@ -131,7 +131,7 @@ public void startPartition(DecoratedKey key, long position, long keyPositionForS
 
         try
         {
-            perSSTableWriter.startPartition(position);
+            perSSTableWriter.startPartition(key, position);
         }
         catch (Throwable t)
         {

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

@@ -34,7 +34,7 @@ public enum IndexComponentType
      */
     META("Meta"),
     /**
-     * KDTree written by {@code BKDWriter} indexes mappings of term to one ore more segment row IDs
+     * KDTree written by {@code BKDWriter} indexes mappings of term to one or more segment row IDs
      * (segment row ID = SSTable row ID - segment row ID offset).
      *
      * V1
@@ -116,7 +116,53 @@ public enum IndexComponentType
     /**
      * Stores document length information for BM25 scoring
      */
-    DOC_LENGTHS("DocLengths");
+    DOC_LENGTHS("DocLengths"),
+
+    /**
+     * An on-disk block packed index mapping rowIds to token values.
+     * <p>
+     * V9
+     */
+    ROW_TO_TOKEN("RowToToken"),
+
+    /**
+     * An on-disk block packed index mapping rowIds to partitionIds.
+     * <p>
+     * V9
+     */
+    ROW_TO_PARTITION("RowToPartition"),
+
+    /**
+     * An on-disk block packed index mapping partitionIds to the number of rows for the partition.
+     * <p>
+     * V9
+     */
+    PARTITION_TO_SIZE("PartitionToSize"),
+
+    /**
+     * Prefix-compressed blocks of partition keys used for rowId to partition key lookups
+     * <p>
+     * V9
+     */
+    PARTITION_KEY_BLOCKS("PKBlocks"),
+    /**
+     * Encoded sequence of offsets to partition key blocks
+     * <p>
+     * V9
+     */
+    PARTITION_KEY_BLOCK_OFFSETS("PKBlockOffsets"),
+    /**
+     * Prefix-compressed blocks of clustering keys used for rowId to clustering key lookups
+     * <p>
+     * V9
+     */
+    CLUSTERING_KEY_BLOCKS("CKBlocks"),
+    /**
+     * Encoded sequence of offsets to clustering key blocks
+     * <p>
+     * V9
+     */
+    CLUSTERING_KEY_BLOCK_OFFSETS("CKBlockOffsets");
 
     public final String representation;
 

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

@@ -33,6 +33,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.db.ClusteringComparator;
 import org.apache.cassandra.db.lifecycle.Tracker;
 import org.apache.cassandra.index.sai.IndexContext;
 import org.apache.cassandra.io.sstable.Component;
@@ -186,9 +187,13 @@ default Set<IndexComponentType> expectedComponentsForVersion()
     {
         return isPerIndexGroup()
                ? onDiskFormat().perIndexComponentTypes(context())
-               : onDiskFormat().perSSTableComponentTypes();
+               : onDiskFormat().perSSTableComponentTypes(hasClustering());
     }
 
+    boolean hasClustering();
+
+    ClusteringComparator comparator();
+
     default ByteComparable.Version byteComparableVersionFor(IndexComponentType component)
     {
         return version().byteComparableVersionFor(component, descriptor().version);

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

@@ -35,11 +35,14 @@
 
 import javax.annotation.Nullable;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.db.ClusteringComparator;
 import org.apache.cassandra.db.lifecycle.Tracker;
 import org.apache.cassandra.index.sai.IndexContext;
 import org.apache.cassandra.index.sai.StorageAttachedIndex;
@@ -55,6 +58,7 @@
 import org.apache.cassandra.io.storage.StorageProvider;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileHandle;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.lucene.store.BufferedChecksumIndexInput;
 import org.apache.lucene.store.ChecksumIndexInput;
@@ -89,22 +93,32 @@ public class IndexDescriptor
     */
 
     public final Descriptor descriptor;
+    public final boolean hasClustering;
     private final ComponentsBuildId emptyGroupMarker;
+    private final ClusteringComparator comparator;
 
     // The per-sstable components for this descriptor. This is never `null` in practice, but 1) it's a bit easier to
     // initialize it outsides of the ctor, and 2) it can actually change upon calls to `reload`.
     private IndexComponentsImpl perSSTable;
     private final Map<IndexContext, IndexComponentsImpl> perIndexes = new ConcurrentHashMap<>();
 
-    private IndexDescriptor(Descriptor descriptor)
+    private IndexDescriptor(Descriptor descriptor, ClusteringComparator comparator)
     {
         this.descriptor = descriptor;
         this.emptyGroupMarker = ComponentsBuildId.of(Version.current(descriptor.ksname), -1);
+        this.comparator = comparator;
+        this.hasClustering = comparator.size() > 0;
     }
 
+    @VisibleForTesting
     public static IndexDescriptor empty(Descriptor descriptor)
     {
-        IndexDescriptor created = new IndexDescriptor(descriptor);
+        return empty(descriptor, new ClusteringComparator());
+    }
+
+    public static IndexDescriptor empty(Descriptor descriptor, ClusteringComparator comparator)
+    {
+        IndexDescriptor created = new IndexDescriptor(descriptor, comparator);
         // Some code assumes that you can always at least call `perSSTableComponents()` and not get `null`, so we
         // set it to an empty group here.
         created.perSSTable = created.createEmptyGroup(null);
@@ -114,7 +128,8 @@ public static IndexDescriptor empty(Descriptor descriptor)
     public static IndexDescriptor load(SSTableReader sstable, Set<IndexContext> indices)
     {
         SSTableIndexComponentsState discovered = IndexComponentDiscovery.instance().discoverComponents(sstable);
-        IndexDescriptor descriptor = new IndexDescriptor(sstable.descriptor);
+        IndexDescriptor descriptor = new IndexDescriptor(sstable.descriptor,
+                                                         sstable.metadata().comparator);
         descriptor.initialize(indices, discovered);
         return descriptor;
     }
@@ -134,7 +149,7 @@ private void initializeIndexes(Set<IndexContext> indices, SSTableIndexComponents
     private Set<IndexComponentType> expectedComponentsForVersion(Version version, @Nullable IndexContext context)
     {
         return context == null
-               ? version.onDiskFormat().perSSTableComponentTypes()
+               ? version.onDiskFormat().perSSTableComponentTypes(hasClustering)
                : version.onDiskFormat().perIndexComponentTypes(context);
     }
 
@@ -182,11 +197,11 @@ private IndexComponentsImpl createEmptyGroup(@Nullable IndexContext context)
      * Please note that the final sstable may not contain all of these components, as some may be empty or not written
      * due to the specific of the flush, but this should be a superset of the components written.
      */
-    public static Set<Component> componentsForNewlyFlushedSSTable(Collection<StorageAttachedIndex> indices, Version version)
+    public static Set<Component> componentsForNewlyFlushedSSTable(Collection<StorageAttachedIndex> indices, Version version, boolean hasClustering)
     {
         ComponentsBuildId buildId = ComponentsBuildId.forNewSSTable(version);
         Set<Component> components = new HashSet<>();
-        for (IndexComponentType component : buildId.version().onDiskFormat().perSSTableComponentTypes())
+        for (IndexComponentType component : buildId.version().onDiskFormat().perSSTableComponentTypes(hasClustering))
             components.add(customComponentFor(buildId, component, null));
 
         for (StorageAttachedIndex index : indices)
@@ -197,7 +212,7 @@ public static Set<Component> componentsForNewlyFlushedSSTable(Collection<Storage
     /**
      * The set of per-index components _expected_ to be written for a newly flushed sstable for the provided index.
      * <p>
-     * This is a subset of {@link #componentsForNewlyFlushedSSTable(Collection, Version)} and has the same caveats.
+     * This is a subset of {@link #componentsForNewlyFlushedSSTable(Collection, Version, boolean)} and has the same caveats.
      */
     public static Set<Component> perIndexComponentsForNewlyFlushedSSTable(IndexContext context)
     {
@@ -430,6 +445,18 @@ public boolean isEmpty()
             return isComplete() && components.size() == 1;
         }
 
+        @Override
+        public boolean hasClustering()
+        {
+            return hasClustering;
+        }
+
+        @Override
+        public ClusteringComparator comparator()
+        {
+            return comparator;
+        }
+
         @Override
         public Collection<IndexComponent.ForRead> all()
         {
@@ -656,8 +683,13 @@ public File file()
             @Override
             public FileHandle createFileHandle()
             {
-                var builder = StorageProvider.instance.fileHandleBuilderFor(this);
-                var b = builder.order(byteOrder());
+                FileHandle.Builder builder = StorageProvider.instance.fileHandleBuilderFor(this);
+                if (logger.isTraceEnabled())
+                {
+                    logger.trace(this.parent().logMessage("Opening file handle for {} ({})"),
+                                 file, FBUtilities.prettyPrintMemory(file.length()));
+                }
+                FileHandle.Builder b = builder.order(byteOrder());
                 return b.complete();
             }
 

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

@@ -150,9 +150,10 @@ public PerIndexWriter newPerIndexWriter(StorageAttachedIndex index,
      * This is a complete set of componentstypes that could exist on-disk. It does not imply that the
      * components currently exist on-disk.
      *
+     * @param hasClustering true if the SSTable forms part of a table using clustering columns
      * @return The set of {@link IndexComponentType} for the per-SSTable index
      */
-    public Set<IndexComponentType> perSSTableComponentTypes();
+    Set<IndexComponentType> perSSTableComponentTypes(boolean hasClustering);
 
     /**
      * Returns the set of {@link IndexComponentType} for the per-index part of an index.
@@ -174,9 +175,10 @@ default public Set<IndexComponentType> perIndexComponentTypes(IndexContext index
      * This is a static indication of the files that can be held open by an index
      * for queries. It is not a dynamic calculation.
      *
+     * @param hasClustering true if the SSTable forms part of a table using clustering columns
      * @return The number of open per-SSTable files
      */
-    public int openFilesPerSSTable();
+    int openFilesPerSSTable(boolean hasClustering);
 
     /**
      * Return the number of open per-index files that can be open during a query.
@@ -215,5 +217,4 @@ default public Set<IndexComponentType> perIndexComponentTypes(IndexContext index
      * @return the JVector file format version that this on-disk format uses.
      */
     int jvectorFileFormatVersion();
-
 }