test(vectordata): mirror the reference's series and slab coverage

- a series whose shards disagree on shape is refused naming the odd
  one; a gap in the middle of a series is named, never skipped
- two shards sliced from one file share one storage and address their
  own windows of it; a local series reports local and complete
- the weakest access mode is the one true of every file, and a series
  with one unverified shard reports it over HTTP; a range-less server
  hands each shard over whole and the series still reads as one
- a mapped series window is prefetched without a fallback in play
- the whole-profile prebuffer visits slab facets whole, a sharded slab
  shard by shard, and an explicit entry that slices a slab plans its
  pages by file ordinal

nb-vectordata/src/test/java/io/nosqlbench/vectordata/PrefetchRemoteIntegrationTest.java

@@ -317,14 +317,20 @@ class PrefetchRemoteIntegrationTest {
     /// Publishes `big__0000..` shards of `records` each, `.mref`-published
     /// so chunk residency is real, and returns the server's base URL.
     private String publishSeries(int shards, int records, boolean mrefs) throws Exception {
+        return publishSeries(shards, records, mrefs ? shards : 0, true);
+    }
+
+    /// As above, with an `.mref` beside only the first `mrefShards`
+    /// shards, on a server that may refuse `Range`.
+    private String publishSeries(int shards, int records, int mrefShards, boolean acceptRanges) throws Exception {
         Path published = Files.createDirectories(temporary.resolve("pub"));
         for (int shard = 0; shard < shards; shard++) {
             byte[] payload = seriesShard(records, shard * records);
             Path file = published.resolve(String.format("big__%04d.fvec", shard));
             Files.write(file, payload);
-            if (mrefs) Files.write(published.resolve(file.getFileName() + ".mref"), FixtureSupport.mref(payload, CHUNK));
+            if (shard < mrefShards) Files.write(published.resolve(file.getFileName() + ".mref"), FixtureSupport.mref(payload, CHUNK));
         }
-        HttpServer server = serveDirectory(published, new AtomicInteger(), true);
+        HttpServer server = serveDirectory(published, new AtomicInteger(), acceptRanges);
         return "http://127.0.0.1:" + server.getAddress().getPort() + "/";
     }
 
@@ -507,4 +513,30 @@ class PrefetchRemoteIntegrationTest {
         assertTrue(small.prefetchPlan("metadata_content", DSWindow.parse("0..100")).isResident(), "the window's pages are resident");
         assertFalse(small.prefetchPlan("metadata_content", DSWindow.ALL).isResident(), "the rest of the slab is not");
     }
+
+    /// A server that ignores `Range` hands each shard over whole at its
+    /// first touch; the series still reads as one facet, and a window
+    /// still maps to fills, resident by then.
+    @Test void aSeriesReadsCorrectlyOnARangeLessServer() throws Exception {
+        String base = publishSeries(2, 25, 0, false);
+        TestDataView view = seriesView("rangeless", uniformYaml(base, 2, 25, ""), "default");
+        VectorReader<float[]> reader = view.baseVectors();
+        assertEquals(50, reader.count());
+        for (int i = 0; i < 50; i++) assertEquals(i * 100f, reader.get(i)[0], "record " + i);
+        PrefetchPlan plan = view.prefetchPlan("base_vectors", DSWindow.parse("20..30"));
+        assertFalse(plan.fills().isEmpty(), "the window maps to fills either way");
+        assertTrue(plan.isResident(), "and the files arrived whole at first touch");
+        assertEquals(AccessMode.FULL_TRANSFER, reader.cacheStats().accessMode());
+        assertEquals(2500f, reader.get(25)[0]);
+    }
+
+    /// A facet's access mode is a promise about every read it serves,
+    /// so a series with one unverified shard reports the weaker mode.
+    @Test void aSeriesReportsTheWeakestAccessModeOfItsShards() throws Exception {
+        String base = publishSeries(2, 25, 1, true);
+        VectorReader<float[]> reader = seriesView("weakest", uniformYaml(base, 2, 25, ""), "default").baseVectors();
+        assertEquals(AccessMode.MERKLE_CHUNKED, reader.cacheStats().accessMode(),
+            "one shard without an .mref makes the facet's promise chunked-but-unverified");
+        assertEquals(49 * 100f, reader.get(49)[0], "and both shards still read");
+    }
 }

nb-vectordata/src/test/java/io/nosqlbench/vectordata/ShardedFacetsTest.java

@@ -428,4 +428,71 @@ class ShardedFacetsTest {
         Path plain = dataset("generous", "format_version: 2\nname: g\nprofiles:\n  default:\n    base_vectors: base.fvec\n");
         assertNotNull(TestDataGroup.load(plain.toUri(), settings()), "a version higher than the content requires is merely generous");
     }
+
+    // -- Files that are not one facet, and what a series promises --
+
+    @Test void aSeriesWithADisagreeingShardIsRefused() throws Exception {
+        Path dir = Files.createDirectories(temporary.resolve("disagree"));
+        fvec(dir, "base__0000.fvec", 10, 0);
+        FixtureSupport.fvec(dir, "base__0001.fvec", new float[10][8]);
+        Files.writeString(dir.resolve("dataset.yaml"), """
+            name: disagree
+            profiles:
+              default:
+                base_vectors:
+                  source: base__NNNN.fvec
+                  shard_stride: 10
+                  shard_count: 2
+                  record_count: 20
+            """);
+        VectorDataException refused = assertThrows(VectorDataException.class, () -> view(dir).baseVectors());
+        assertTrue(refused.getMessage().contains("share one shape"), "shards of different dimension are not one facet: " + refused.getMessage());
+        assertTrue(refused.getMessage().contains("base__0001.fvec"), "and the message names the odd one out: " + refused.getMessage());
+    }
+
+    @Test void aGapInTheMiddleOfASeriesIsReportedByName() throws Exception {
+        Path dir = Files.createDirectories(temporary.resolve("midgap"));
+        fvec(dir, "base__0000.fvec", 10, 0);
+        fvec(dir, "base__0001.fvec", 10, 10);
+        fvec(dir, "base__0003.fvec", 5, 30);
+        Files.writeString(dir.resolve("dataset.yaml"), """
+            name: midgap
+            profiles:
+              default:
+                base_vectors:
+                  source: base__NNNN.fvec
+                  shard_stride: 10
+                  shard_count: 4
+                  record_count: 35
+            """);
+        VectorDataException missing = assertThrows(VectorDataException.class, () -> view(dir).baseVectors());
+        assertTrue(missing.getMessage().contains("base__0002.fvec"), "a hole in the middle is named, never skipped: " + missing.getMessage());
+    }
+
+    @Test void aLocalSeriesReportsLocalThroughItsReader() throws Exception {
+        VectorReader<float[]> base = view(uniformSeries("")).baseVectors();
+        CacheStats stats = base.cacheStats();
+        assertEquals(AccessMode.LOCAL, stats.accessMode(), "every file is local, so the facet's promise is local");
+        assertEquals(240 * BPR, stats.totalBytes(), "bytes are summed over every shard");
+        assertTrue(stats.complete());
+        assertTrue(base.isComplete());
+    }
+
+    @Test void theWeakestAccessModeIsTheOneTrueOfEveryFile() {
+        assertEquals(AccessMode.MERKLE_HASHED, AccessMode.weakest(List.of(AccessMode.LOCAL, AccessMode.MERKLE_HASHED)));
+        assertEquals(AccessMode.MERKLE_CHUNKED, AccessMode.weakest(List.of(AccessMode.MERKLE_HASHED, AccessMode.MERKLE_CHUNKED)),
+            "a trusted-bytes shard is weaker than a verified one");
+        assertEquals(AccessMode.FULL_TRANSFER, AccessMode.weakest(List.of(AccessMode.LOCAL, AccessMode.MERKLE_HASHED, AccessMode.FULL_TRANSFER)),
+            "one shard that must download whole makes the facet's promise that");
+        assertEquals(AccessMode.LOCAL, AccessMode.weakest(List.of(AccessMode.LOCAL, AccessMode.LOCAL)));
+        assertNull(AccessMode.weakest(List.of()), "a facet with no files makes no promise, which is not the weakest promise");
+    }
+
+    @Test void prefetchingAMappedSeriesWindowNeedsNoConsent() throws Exception {
+        TestDataView view = view(uniformSeries(""));
+        PrefetchReport report = view.prefetch("base_vectors", DSWindow.parse("80..220"), WholeFacetFallback.REFUSE);
+        assertEquals(3, report.rangesFetched(), "one range per shard the window spans, fetched without a fallback in play");
+        assertFalse(report.planned().degradesToFullDownload());
+        view.prefetchInBackground("base_vectors", DSWindow.parse("0..240"), WholeFacetFallback.REFUSE).join();
+    }
 }

nb-vectordata/src/test/java/io/nosqlbench/vectordata/SlabWindowsTest.java

@@ -157,4 +157,63 @@ class SlabWindowsTest {
         assertEquals(0, plan.byteRanges().get(1).start(), "records 0..10 of the second shard are its first page");
     }
 
+    /// The whole-profile prebuffer visits slab facets like any other:
+    /// an unwindowed one is planned whole, a sharded one shard by shard.
+    @Test void prebufferingAProfileVisitsItsSlabFacetsWhole() throws Exception {
+        TestDataView view = view(pagedFacet("visit", 40, ""), "default");
+        assertDoesNotThrow(() -> view.prebuffer(WholeFacetFallback.REFUSE, PrebufferProgress.NONE));
+        PrefetchPlan whole = view.prefetchPlan("metadata_content", DSWindow.ALL);
+        assertEquals(1, whole.byteRanges().size());
+        assertEquals(whole.facetBytes(), bytes(whole), "no window, so the whole file");
+
+        Path dir = Files.createDirectories(temporary.resolve("visit-sharded"));
+        FixtureSupport.slab(dir, "m__0000.slab", 20, 10, 64);
+        FixtureSupport.slab(dir, "m__0001.slab", 20, 10, 64);
+        FixtureSupport.fvec(dir, "base.fvec", new float[][] {{0f, 1f, 2f, 3f}});
+        Files.writeString(dir.resolve("dataset.yaml"), """
+            name: visit-sharded
+            profiles:
+              default:
+                base_vectors: base.fvec
+                metadata_content:
+                  source: m__NNNN.slab
+                  shard_stride: 20
+                  shard_count: 2
+                  record_count: 40
+            """);
+        TestDataView sharded = view(dir, "default");
+        assertDoesNotThrow(() -> sharded.prebuffer(WholeFacetFallback.REFUSE, PrebufferProgress.NONE));
+        assertEquals(2, sharded.prefetchPlan("metadata_content", DSWindow.ALL).byteRanges().size(), "one range per shard");
+    }
+
+    /// An explicit entry that slices a slab carries its window in that
+    /// file's ordinals, so its shard's pages are found by file ordinal —
+    /// a plan over the sliced facet reaches into the middle of the file.
+    @Test void aSlicedSlabEntryPlansItsOwnPages() throws Exception {
+        Path dir = Files.createDirectories(temporary.resolve("sliced-slab"));
+        FixtureSupport.slab(dir, "m.slab", 200, 25, 64);
+        FixtureSupport.fvec(dir, "base.fvec", new float[][] {{0f, 1f, 2f, 3f}});
+        Files.writeString(dir.resolve("dataset.yaml"), """
+            name: sliced-slab
+            profiles:
+              default:
+                base_vectors: base.fvec
+                metadata_content:
+                  source:
+                    - m.slab[0..50]=50
+                    - m.slab[150..200]=50
+                  record_count: 100
+            """);
+        TestDataView view = view(dir, "default");
+        long fileBytes = Files.size(dir.resolve("m.slab"));
+        PrefetchPlan both = view.prefetchPlan("metadata_content", DSWindow.parse("0..100"));
+        assertEquals(2, both.byteRanges().size(), "the window spans both slices: " + both.byteRanges());
+        assertEquals(0, both.byteRanges().get(0).start(), "the first slice starts at the file's first page");
+        assertTrue(both.byteRanges().get(1).start() > both.byteRanges().get(0).end(), "the second slice's pages lie further into the file");
+        assertTrue(bytes(both) < fileBytes * 3 / 5 && bytes(both) > fileBytes * 2 / 5, "half the records cost about half the file: " + bytes(both) + " of " + fileBytes);
+        PrefetchPlan inner = view.prefetchPlan("metadata_content", DSWindow.parse("60..70"));
+        assertEquals(1, inner.byteRanges().size());
+        assertEquals(1, inner.byteRanges().get(0).shard(), "facet ordinals 60..70 are the second slice's 10..20");
+        assertEquals(both.byteRanges().get(1).start(), inner.byteRanges().get(0).start(), "which lives in that slice's first page");
+    }
 }

nb-vectordata/src/test/java/io/nosqlbench/vectordata/internal/ShardsTest.java

@@ -145,6 +145,25 @@ class ShardsTest {
         assertTrue(short_.getMessage().contains("record_count"), "a total that leaves the last shard empty disagrees with itself: " + short_.getMessage());
     }
 
+    @Test void twoShardsOfOneFileShareOneStorage() throws Exception {
+        Path dir = Files.createDirectories(temporary.resolve("shared"));
+        ByteBuffer values = ByteBuffer.allocate(10_000 * 4).order(ByteOrder.LITTLE_ENDIAN);
+        for (int i = 0; i < 10_000; i++) values.putInt(i);
+        Files.write(dir.resolve("corpus.u32"), values.array());
+        String file = dir.resolve("corpus.u32").toUri().toString();
+        VectorDataSettings settings = VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
+        Shards shards = Shards.realize("layout",
+            new Shards.Declaration(List.of(file + "[0..10]=10", file + "[9990..10000]=10"), true, null, null, 20L), s -> 0);
+        FacetSeries series = new FacetSeries(shards, settings, "shared");
+        assertEquals(2, shards.shardCount(), "two shards");
+        assertEquals(1, series.fileCount(), "one file");
+        assertEquals(series.fileIndexOfShard(0), series.fileIndexOfShard(1));
+        assertSame(series.file(0), series.file(series.fileIndexOfShard(1)), "and one storage between them");
+        assertArrayEquals(new long[] {0, 40}, series.shardByteExtent(0), "each shard addresses its own window of the file");
+        assertArrayEquals(new long[] {9990 * 4, 10_000 * 4}, series.shardByteExtent(1));
+        assertEquals(10_000 * 4, series.tryTotalSize(), "the file is counted once");
+    }
+
     @Test void theOpenFileCapIsDerivedAndLeavesHeadroom() {
         assertEquals(256, FacetSeries.resolveOpenFileCap(null, 1024L), "a quarter of the soft limit");
         assertEquals(256, FacetSeries.resolveOpenFileCap(null, null), "1024 when the limit is unknown");