CNDB-16252: Add IndexComponentsImpl#tmpFileFor to allow vector CompactionGraph to index multiple columns (#2177)

### What is the issue
Fixes riptano/cndb#16252

### What does this PR fix and why was it fixed
Add a new method to the `IndexComponents.ForWrite` interface named
`tmpFileFor`. The new method is expected to create temporary files that
are namespaced to the index build. This fixes an issue in the
`CompactionGraph` where we incorrectly created temp files with the same
name in such a way that they collided. Now the files have the column
name and the build id, so they will be properly namespaced while
maintaining some form of meaningful name.

Author: michaeljmarshall

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

@@ -35,6 +35,7 @@
 import org.apache.cassandra.io.sstable.Component;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTable;
+import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable;
 
 /**
@@ -346,5 +347,14 @@ interface ForWrite extends ForRead
          * should be added to this writer after this call).
          */
         void markComplete() throws IOException;
+
+        /**
+         * Create a temporary {@link File} namespaced within the per index components. Repeated calls with the same
+         * componentName will produce the same file.
+         * @param componentName - unique name within the per index components
+         * @return a temprory file for use during index construction
+         * @throws IOException
+         */
+        File tmpFileFor(String componentName) throws IOException;
     }
 }

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

@@ -510,6 +510,14 @@ public IndexComponent.ForWrite getForWrite(IndexComponentType component)
             return info;
         }
 
+        @Override
+        public File tmpFileFor(String componentName) throws IOException
+        {
+            String name = context != null ? String.format("%s_%s_%s", buildId, context.getColumnName(), componentName)
+                                          :  String.format("%s_%s", buildId, componentName);
+            return descriptor.tmpFileFor(new Component(Component.Type.CUSTOM, name));
+        }
+
         @Override
         public void forceDeleteAllComponents()
         {

src/java/org/apache/cassandra/index/sai/disk/vector/CompactionGraph.java

@@ -184,8 +184,7 @@ public CompactionGraph(IndexComponents.ForWrite perIndexComponents, VectorCompre
         this.useSyntheticOrdinals = !V5OnDiskFormat.writeV5VectorPostings(context.version()) || !allRowsHaveVectors;
 
         // the extension here is important to signal to CFS.scrubDataDirectories that it should be removed if present at restart
-        Component tmpComponent = new Component(Component.Type.CUSTOM, "chronicle" + Descriptor.TMP_EXT);
-        postingsFile = dd.fileFor(tmpComponent);
+        postingsFile = perIndexComponents.tmpFileFor("postings_chonicle_map");
         postingsMap = ChronicleMapBuilder.of((Class<VectorFloat<?>>) (Class) VectorFloat.class, (Class<CompactionVectorPostings>) (Class) CompactionVectorPostings.class)
                                          .averageKeySize(dimension * Float.BYTES)
                                          .averageValueSize(VectorPostings.emptyBytesUsed() + RamUsageEstimator.NUM_BYTES_OBJECT_REF + 2 * Integer.BYTES)
@@ -195,8 +194,7 @@ public CompactionGraph(IndexComponents.ForWrite perIndexComponents, VectorCompre
                                          .createPersistedTo(postingsFile.toJavaIOFile());
 
         // Formatted so that the full resolution vector is written at the ordinal * vector dimension offset
-        Component vectorsByOrdinalComponent = new Component(Component.Type.CUSTOM, "vectors_by_ordinal");
-        vectorsByOrdinalTmpFile = dd.tmpFileFor(vectorsByOrdinalComponent);
+        vectorsByOrdinalTmpFile = perIndexComponents.tmpFileFor("vectors_by_ordinal");
         vectorsByOrdinalBufferedWriter = new BufferedRandomAccessWriter(vectorsByOrdinalTmpFile.toPath());
 
         // VSTODO add LVQ

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

@@ -1060,6 +1060,23 @@ public void testUpdateRowScoreToWorsePositionButIncludeInBatch()
         assertRows(execute("SELECT c FROM %s ORDER BY r ANN OF [0.1, 0.1] LIMIT 10"), row(2), row(1));
     }
 
+    @Test
+    public void testIndexingMultipleVectorColumns() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, val1 vector<float, 128>, val2 vector<float, 128>, PRIMARY KEY(pk))");
+        createIndex("CREATE CUSTOM INDEX ON %s(val1) USING 'StorageAttachedIndex'");
+        createIndex("CREATE CUSTOM INDEX ON %s(val2) USING 'StorageAttachedIndex'");
+
+        for (int i = 0; i < 2 * CassandraOnHeapGraph.MIN_PQ_ROWS; i++)
+            execute("INSERT INTO %s (pk, val1, val2) VALUES (?, ?, ?)", i, randomVectorBoxed(128), randomVectorBoxed(128));
+
+        runThenFlushThenCompact(() -> {
+            // Run a search on each as a sanity check
+            assertRowCount(execute("SELECT pk FROM %s ORDER BY val1 ANN OF ? LIMIT 10", randomVectorBoxed(128)), 10);
+            assertRowCount(execute("SELECT pk FROM %s ORDER BY val2 ANN OF ? LIMIT 10", randomVectorBoxed(128)), 10);
+        });
+    }
+
     @Test
     public void testRowIdIteratorClosedOnHasNextFailure() throws Throwable
     {