diff --git a/CHANGES.txt b/CHANGES.txt index 5b08d8b08c..12c457895a 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java index f6a3fbaeb9..f548e98a6a 100644 --- a/src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java +++ b/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, diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java index 06106f8683..61127407bb 100644 --- a/src/java/org/apache/cassandra/index/Index.java +++ b/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; + } } /** diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java index 33ae11e3af..69217a6825 100644 --- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java +++ b/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; diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java index 046af69618..fd555960f7 100644 --- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java +++ b/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. diff --git a/test/unit/org/apache/cassandra/index/sai/cql/VectorLocalTest.java b/test/unit/org/apache/cassandra/index/sai/cql/VectorLocalTest.java index 1f9705e11b..b1208f8678 100644 --- a/test/unit/org/apache/cassandra/index/sai/cql/VectorLocalTest.java +++ b/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) 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 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);