Repros 744

docs/batched-vector-access.md

@@ -87,12 +87,14 @@ useDatasetPrefetching: true
 ```
 
 Omitted or `false` preserves the resident path. The existing VM option
-`-Djvector.dataset_loader=buffered` is an explicit override; `preload` overrides the
-flag in the other direction. Neither runner changes a user's YAML files.
+`-Djvector.dataset_loader=buffered` explicitly overrides the YAML flag; `preload`
+overrides the flag in the other direction. Neither runner changes a user's YAML files.
 
 The harness uses file input for bulk consumption in quantized construction and
 existing disk storage. Uncompressed construction, or a search configuration that
 requires raw input rather than stored vectors/codes, still uses resident input.
+The VM option does not bypass this resident-input routing. If an explicit
+`buffered` request requires resident input, the harness logs the reason for using preload.
 A mixed configuration requiring resident input remains resident; split those cases
 into separate configurations to measure dataset prefetching independently. This is
 source routing, not a restriction on the vector-access API.
@@ -267,6 +269,8 @@ reads and payload allocation, not physical device I/O.
 - Close cursors with try-with-resources, stop consumers, then close the root source.
 - Copies share the root's storage and I/O executor. Closing a copy affects that view
   only; closing the root invalidates all copies and releases the shared resources.
+- If point reads are still running when the root closes, close waits for each read
+  in progress to finish; later point reads fail with `IllegalStateException`.
 - Direct payload buffers become eligible for JVM reclamation after references are
   released; closure does not promise immediate native-memory reclamation.
 

docs/release notes/4.0.2/744.feature.md

@@ -6,7 +6,9 @@ of the original base dataset; existing vector sources remain supported.
 BenchYAML and AutoBenchYAML support `useDatasetPrefetching: true` in their index
 configuration. The flag defaults to false, preserving resident loading and scoring.
 Comparison-heavy uncompressed configurations retain resident input. The existing
-`jvector.dataset_loader` VM option remains an explicit override.
+`jvector.dataset_loader` VM option overrides the flag, while configurations
+requiring resident raw vectors still preload; an explicit buffered request logs
+the reason for that fallback.
 
 See [batched vector access](../../batched-vector-access.md) for ownership,
 resource limits, examples and the benchmark opt-in. Encoded vectors, training

jvector-base/src/main/java/io/github/jbellis/jvector/disk/FvecFileVectorValues.java

@@ -55,7 +55,9 @@
  * has its own borrowed vector.
  *
  * <p>Close all cursors and stop consumers before closing the source. Closing the root
- * releases all resources, including resources used by copies. No memory mapping or
+ * releases all resources, including resources used by copies. A point read already in
+ * progress finishes before its channel is closed; later point reads, including ones
+ * waiting for a reader, fail with {@link IllegalStateException}. No memory mapping or
  * OS cache eviction is performed. Retained training vectors are caller-owned memory.
  */
 public final class FvecFileVectorValues implements BatchedVectorValues, AutoCloseable {
@@ -463,6 +465,8 @@ void readPointInto(int ordinal, VectorFloat<?> destination, int offset) {
             PointReader reader = null;
             try {
                 reader = borrowPointReader();
+                // Close waits for each lease, so a closed source can't reopen this reader's channel.
+                if (closed) throw new IllegalStateException("Vector source is closed");
                 pointHint.set(reader);
                 var input = pointChannel(reader);
                 var buffer = reader.buffer;
@@ -626,10 +630,12 @@ void close() throws IOException {
             }
             IOException failure = null;
             for (PointReader reader : pointReaders) {
+                // Wait for an in-flight point read to finish; readers that lease afterwards see closed.
+                reader.permit.acquireUninterruptibly();
                 try { if (reader.channel != null) reader.channel.close(); }
                 catch (IOException error) {
                     if (failure == null) failure = error; else failure.addSuppressed(error);
-                } finally { reader.channel = null; reader.buffer = null; }
+                } finally { reader.channel = null; reader.buffer = null; reader.permit.release(); }
             }
             if (pointHint != null) pointHint.remove();
             try { channel.close(); }

jvector-base/src/main/java/io/github/jbellis/jvector/graph/similarity/BuildScoreProvider.java

@@ -110,11 +110,23 @@ static BuildScoreProvider randomAccessScoreProvider(RandomAccessVectorValues rav
         var vectorsCopy = ravv.threadLocalSupplier();
 
         return new BuildScoreProvider() {
+            /// Exact scores against `node1`.
+            ///
+            /// For a value-shared source the anchor is read on the first comparison and kept as
+            /// a private copy, so creating a score function reads nothing, later reads through the
+            /// same views can't overwrite the anchor, and each comparison reads only `node2`.
             private ScoreFunction.ExactScoreFunction scoreFunctionFor(int node1) {
                 if (ravv.isValueShared()) {
-                    // Resolve both ordinals using separate borrowed views; no anchor escapes a read.
-                    return node2 -> similarityFunction.compare(vectors.get().getVector(node1),
-                            vectorsCopy.get().getVector(node2));
+                    return new ScoreFunction.ExactScoreFunction() {
+                        private VectorFloat<?> anchor;
+
+                        @Override
+                        public float similarityTo(int node2) {
+                            var a = anchor;
+                            if (a == null) anchor = a = vectors.get().getVector(node1).copy();
+                            return similarityFunction.compare(a, vectorsCopy.get().getVector(node2));
+                        }
+                    };
                 }
                 var vector = vectors.get().getVector(node1);
                 var vc = vectorsCopy.get();

jvector-examples/pom.xml

@@ -15,6 +15,7 @@
         <awssdk.version>2.21.10</awssdk.version>
         <!-- Empty uses public-catalog e5-small-v2-100k in the batched-access examples. -->
         <vectorFile></vectorFile>
+        <argLine></argLine>
     </properties>
     <build>
         <plugins>
@@ -23,6 +24,7 @@
                 <artifactId>maven-surefire-plugin</artifactId>
                 <configuration>
                     <skip>false</skip>
+                    <argLine>${argLine}</argLine>
                     <workingDirectory>${project.parent.basedir}</workingDirectory>
                 </configuration>
             </plugin>
@@ -275,6 +277,9 @@
         </profile>
         <profile>
             <id>jdk20</id>
+            <properties>
+                <argLine>--add-modules jdk.incubator.vector</argLine>
+            </properties>
             <dependencies>
                 <dependency>
                     <groupId>io.github.jbellis</groupId>
@@ -395,6 +400,9 @@
         </profile>
         <profile>
             <id>jdk22</id>
+            <properties>
+                <argLine>--add-modules jdk.incubator.vector --enable-native-access=ALL-UNNAMED</argLine>
+            </properties>
             <activation>
                 <activeByDefault>true</activeByDefault>
             </activation>

jvector-examples/src/main/java/io/github/jbellis/jvector/example/benchmarks/datasets/BaseVectorLoading.java

@@ -17,6 +17,7 @@
 package io.github.jbellis.jvector.example.benchmarks.datasets;
 
 import java.util.Locale;
+import java.util.logging.Logger;
 import io.github.jbellis.jvector.example.yaml.MultiConfig;
 import io.github.jbellis.jvector.graph.disk.feature.FeatureId;
 
@@ -25,6 +26,8 @@ public enum BaseVectorLoading {
     BUFFERED,
     PRELOAD;
 
+    private static final Logger LOG = Logger.getLogger(BaseVectorLoading.class.getName());
+
     /** Existing VM option is an explicit override of the YAML flag. */
     public static BaseVectorLoading forBenchmark(boolean useDatasetPrefetching) {
         String value = System.getProperty("jvector.dataset_loader",
@@ -37,16 +40,25 @@ public static BaseVectorLoading forBenchmark(boolean useDatasetPrefetching) {
 
     public static BaseVectorLoading forBenchmark() { return forBenchmark(false); }
 
-    /** Bulk input mode applies only where the harness consumes quantized or stored vectors. */
+    /**
+     * The VM option overrides the YAML flag. Configurations requiring resident raw
+     * vectors retain preload so construction and query scoring stay unchanged.
+     */
     public static BaseVectorLoading forBenchmark(MultiConfig config) {
         var loading = forBenchmark(config.useDatasetPrefetching);
         if (loading == PRELOAD) return loading;
-        // Uncompressed construction still uses main's resident raw-vector source.
         if (config.construction.compression.stream().anyMatch(c -> "None".equals(c.type)))
-            return PRELOAD;
+            return residentInput("uncompressed construction");
         boolean exactSearch = config.search.compression.stream().anyMatch(c -> "None".equals(c.type));
         boolean needsRawInput = exactSearch && config.construction.getFeatureSets().stream().anyMatch(features ->
                 !features.contains(FeatureId.FUSED_PQ));
-        return needsRawInput ? PRELOAD : BUFFERED;
+        return needsRawInput ? residentInput("raw-input exact search") : BUFFERED;
+    }
+
+    private static BaseVectorLoading residentInput(String reason) {
+        if ("buffered".equalsIgnoreCase(System.getProperty("jvector.dataset_loader")))
+            LOG.warning("Using preload instead of jvector.dataset_loader=buffered for " + reason
+                    + "; retaining resident input to preserve scoring behavior.");
+        return PRELOAD;
     }
 }

jvector-tests/src/test/java/io/github/jbellis/jvector/disk/TestFvecCloseDuringPointReads.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.github.jbellis.jvector.disk;
+
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.channels.ClosedChannelException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.SplittableRandom;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+
+
+/// Closing the root while point reads are in flight must fail those reads cleanly, as the rest
+/// of the source does ("Vector source is closed" or a closed-channel I/O error), and must leave
+/// no file descriptors open on the input file.
+///
+/// The race is timing-dependent, so the test repeats close-under-load until it sees a
+/// violation or the round limit is reached. A pass is not proof of absence.
+public class TestFvecCloseDuringPointReads {
+    private static final int ROWS = 10_000, DIMENSIONS = 16, READERS = 12, ROUNDS = 300;
+
+    private static Path writeFvecs() throws Exception {
+        var file = Files.createTempFile("close-during-point-reads", ".fvecs");
+        var bytes = ByteBuffer.allocate(ROWS * (DIMENSIONS + 1) * 4).order(ByteOrder.LITTLE_ENDIAN);
+        for (int row = 0; row < ROWS; row++) {
+            bytes.putInt(DIMENSIONS);
+            for (int d = 0; d < DIMENSIONS; d++) bytes.putFloat(row);
+        }
+        Files.write(file, bytes.array());
+        return file;
+    }
+
+    /// Expected failures after close: the source reports closure, or the leased channel was
+    /// closed underneath the read.
+    private static boolean isCleanClosureFailure(Throwable t) {
+        if (t instanceof IllegalStateException) return true;
+        return t instanceof UncheckedIOException && t.getCause() instanceof ClosedChannelException;
+    }
+
+    private static long openDescriptorsOn(Path file) throws Exception {
+        var real = file.toRealPath();
+        long count = 0;
+        try (var fds = Files.list(Path.of("/proc/self/fd"))) {
+            for (var fd : (Iterable<Path>) fds::iterator) {
+                try { if (Files.readSymbolicLink(fd).equals(real)) count++; }
+                catch (java.io.IOException ignored) { } // descriptor closed while listing
+            }
+        }
+        return count;
+    }
+
+    @Test(timeout = 120_000)
+    public void closeUnderPointLoadFailsCleanlyAndLeaksNoDescriptors() throws Exception {
+        Assume.assumeTrue("needs /proc/self/fd", Files.isDirectory(Path.of("/proc/self/fd")));
+        var file = writeFvecs();
+        try {
+            for (int round = 0; round < ROUNDS; round++) {
+                var unexpected = new ConcurrentLinkedQueue<Throwable>();
+                var root = FvecFileVectorValues.open(file);
+                var started = new CountDownLatch(READERS);
+                List<Thread> readers = new ArrayList<>();
+                for (int r = 0; r < READERS; r++) {
+                    var view = root.copy();
+                    long seed = round * 1000L + r;
+                    var thread = new Thread(() -> {
+                        var random = new SplittableRandom(seed);
+                        started.countDown();
+                        try {
+                            while (true) view.getVector(random.nextInt(ROWS));
+                        } catch (Throwable t) {
+                            if (!isCleanClosureFailure(t)) unexpected.add(t);
+                        }
+                    }, "point-reader-" + r);
+                    readers.add(thread);
+                    thread.start();
+                }
+                started.await();
+                Thread.sleep(1);
+                root.close();
+                for (var thread : readers) thread.join();
+
+                long leaked = openDescriptorsOn(file);
+                if (!unexpected.isEmpty() || leaked != 0) {
+                    var first = unexpected.peek();
+                    var error = new AssertionError("round " + round + ": " + unexpected.size()
+                            + " point read(s) failed with an unexpected exception after close"
+                            + (first == null ? "" : " (first: " + first + ")")
+                            + "; " + leaked + " file descriptor(s) still open on the input");
+                    if (first != null) error.initCause(first);
+                    throw error;
+                }
+            }
+        } finally {
+            Files.delete(file);
+        }
+    }
+}

jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetLoaderOverride.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.github.jbellis.jvector.example;
+
+import io.github.jbellis.jvector.example.benchmarks.datasets.BaseVectorLoading;
+import io.github.jbellis.jvector.example.yaml.ConstructionParameters;
+import io.github.jbellis.jvector.example.yaml.MultiConfig;
+import io.github.jbellis.jvector.example.yaml.SearchParameters;
+import io.github.jbellis.jvector.example.yaml.Compression;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.logging.Handler;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/** The VM override selects loading without changing resident construction/query routing. */
+public class TestDatasetLoaderOverride {
+    private static final String PROPERTY = "jvector.dataset_loader";
+    private String previous;
+    private final List<String> warnings = new ArrayList<>();
+    private final Logger logger = Logger.getLogger(BaseVectorLoading.class.getName());
+    private final Handler handler = new Handler() {
+        @Override public void publish(LogRecord record) { warnings.add(record.getMessage()); }
+        @Override public void flush() { }
+        @Override public void close() { }
+    };
+
+    @Before public void saveProperty() {
+        previous = System.getProperty(PROPERTY);
+        logger.addHandler(handler);
+    }
+
+    @After public void restoreProperty() {
+        logger.removeHandler(handler);
+        if (previous == null) System.clearProperty(PROPERTY);
+        else System.setProperty(PROPERTY, previous);
+    }
+
+    private static MultiConfig config(String build, String search) {
+        var config = new MultiConfig();
+        config.construction = new ConstructionParameters();
+        config.search = new SearchParameters();
+        var bc = new Compression(); bc.type = build;
+        var sc = new Compression(); sc.type = search;
+        config.construction.compression = List.of(bc);
+        config.construction.fusedGraph = List.of(false);
+        config.construction.reranking = List.of("NVQ");
+        config.search.compression = List.of(sc);
+        return config;
+    }
+
+    @Test public void explicitBufferedPropertyRetainsExactSearchRoutingAndWarns() {
+        System.setProperty(PROPERTY, "buffered");
+        assertEquals(BaseVectorLoading.PRELOAD, BaseVectorLoading.forBenchmark(config("PQ", "None")));
+        assertTrue(warnings.stream().anyMatch(message -> message.contains("raw-input exact search")));
+    }
+
+    @Test public void explicitBufferedPropertyRetainsUncompressedConstructionAndWarns() {
+        System.setProperty(PROPERTY, "buffered");
+        assertEquals(BaseVectorLoading.PRELOAD, BaseVectorLoading.forBenchmark(config("None", "PQ")));
+        assertTrue(warnings.stream().anyMatch(message -> message.contains("uncompressed construction")));
+    }
+
+    @Test public void explicitBufferedPropertyOverridesDisabledFlagForQuantizedWorkloads() {
+        System.setProperty(PROPERTY, "buffered");
+        assertEquals(BaseVectorLoading.BUFFERED, BaseVectorLoading.forBenchmark(config("PQ", "PQ")));
+        assertTrue(warnings.isEmpty());
+    }
+
+    @Test public void explicitPreloadPropertyOverridesEnabledFlag() {
+        System.setProperty(PROPERTY, "preload");
+        var config = config("PQ", "PQ");
+        config.useDatasetPrefetching = true;
+        assertEquals(BaseVectorLoading.PRELOAD, BaseVectorLoading.forBenchmark(config));
+        assertTrue(warnings.isEmpty());
+    }
+}