Merge branch 'cassandra-5.0' into trunk

* cassandra-5.0:
  Dynamically skip sharding L0 when SAI Vector index present

CHANGES.txt

@@ -296,6 +296,7 @@
  * Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
  * Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
 Merged from 5.0:
+ * Dynamically skip sharding L0 when SAI Vector index present (CASSANDRA-19661)
  * Optionally force IndexStatusManager to use the optimized index status format (CASSANDRA-21132)
  * Automatically disable zero-copy streaming for legacy sstables with old bloom filter format (CASSANDRA-21092)
  * Fix CQLSSTableWriter serialization of vector of date and time (CASSANDRA-20979)

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

@@ -308,7 +308,9 @@ public class UnifiedCompactionStrategy extends AbstractCompactionStrategy
     {
         ShardManager shardManager = getShardManager();
         double flushDensity = cfs.metric.flushSizeOnDisk.get() * shardManager.shardSetCoverage() / shardManager.localSpaceCoverage();
-        ShardTracker boundaries = shardManager.boundaries(controller.getNumShards(flushDensity));
+        boolean supportsSharding = sstableLevel > 0 || indexGroups.stream().allMatch(Index.Group::supportsL0Shards);
+        int numShards = supportsSharding ? controller.getNumShards(flushDensity) : 1;
+        ShardTracker boundaries = shardManager.boundaries(numShards);
         return new ShardedMultiWriter(cfs,
                                       descriptor,
                                       keyCount,

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

@@ -896,6 +896,15 @@ public interface Index
         {
             return true;
         }
+
+        /**
+         * Whether this index group supports sharding when flushing memtables, e.g. level 0 of UCS.
+         * @return true iff all indexes in the group support L0 sharding.
+         */
+        default boolean supportsL0Shards()
+        {
+            return true;
+        }
     }
 
     /**

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

@@ -708,6 +708,15 @@ public class StorageAttachedIndex implements Index
         return () -> valid;
     }
 
+    /**
+     * Vector indexes do not supporrt L0 shards due to the cost associated with resharding at flush time.
+     * @return true iff the index supports sharding at L0.
+     */
+    public boolean supportsL0Shards()
+    {
+        return !indexTermType.isVector();
+    }
+
     public boolean hasClustering()
     {
         return baseCfs.getComparator().size() > 0;

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

@@ -384,6 +384,17 @@ public class StorageAttachedIndexGroup implements Index.Group, INotificationCons
         return complete;
     }
 
+    @Override
+    public boolean supportsL0Shards()
+    {
+        for (StorageAttachedIndex index : indexes)
+            if (!index.supportsL0Shards())
+                return false;
+
+        // All indexes must support L0 sharding for the flush to shard at L0
+        return true;
+    }
+
     /**
      * open index files by checking number of {@link SSTableContext} and {@link SSTableIndex},
      * so transient open files during validation and files that are still open for in-flight requests will not be tracked.

test/unit/org/apache/cassandra/index/sai/cql/VectorLocalTest.java

@@ -513,6 +513,29 @@ public class VectorLocalTest extends VectorTester
         }
     }
 
+    @Test
+    public void flushSuccessfullyVectorIndexToShardedSSTable()
+    {
+        // UCS is configured to use 2 static shards
+        createTable(String.format("CREATE TABLE %%s (k int PRIMARY KEY, v vector<float, %d>) WITH compaction = {\n" +
+                                  "    'class': 'UnifiedCompactionStrategy',\n" +
+                                  "    'base_shard_count': '2',\n" +
+                                  "    'min_sstable_size' : '0MiB', \n" +
+                                  "    'sstable_growth' : '1'\n" +
+                                  "}", word2vec.dimension()));
+        createIndex("CREATE CUSTOM INDEX ON %s(v) USING 'StorageAttachedIndex'");
+
+
+        int vectorCount = 100;
+        List<float[]> vectors = IntStream.range(0, vectorCount).mapToObj(s -> randomVector()).collect(Collectors.toList());
+
+        int pk = 0;
+        for (float[] vector : vectors)
+            execute("INSERT INTO %s (k, v) VALUES (?," + vectorString(vector) + ")", pk++);
+
+        flush();
+    }
+
     private UntypedResultSet search(String stringValue, float[] queryVector, int limit)
     {
         UntypedResultSet result = execute("SELECT * FROM %s WHERE str_val = '" + stringValue + "' ORDER BY val ann of " + Arrays.toString(queryVector) + " LIMIT " + limit);