diff --git a/CHANGES.txt b/CHANGES.txt
index 201b3d9664..114520abfc 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 5.0.7
+ * Dynamically skip sharding L0 when SAI Vector index present (CASSANDRA-19661)
  * Optionally force IndexStatusManager to use the optimized index status format (CASSANDRA-21132) 
  * No need to evict already prepared statements, as it creates a race condition between multiple threads (CASSANDRA-17401)
  * Upgrade logback version to 1.5.18 and slf4j dependencies to 2.0.17 (CASSANDRA-21137)
diff --git a/src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java
index 8ff0565064..64df6ee8f6 100644
--- a/src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java
@@ -270,7 +270,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 108a85ff74..acc468f1a4 100644
--- a/src/java/org/apache/cassandra/index/Index.java
+++ b/src/java/org/apache/cassandra/index/Index.java
@@ -890,6 +890,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 13c3d52ed7..10c2a93cc9 100644
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
@@ -703,6 +703,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 02e7971814..fbdea4a42c 100644
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java
@@ -382,6 +382,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 26b64c57c3..dafa248b44 100644
--- a/test/unit/org/apache/cassandra/index/sai/cql/VectorLocalTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/cql/VectorLocalTest.java
@@ -511,6 +511,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);