mirror of
https://github.com/nosqlbench/nosqlbench.git
synced 2026-10-12 09:24:52 -05:00
fix(vectordata): map ivec windows at the uniform header stride
In this catalog format, ivec/ivecs records are length-qualified but fixed throughout — ground-truth neighbor files — so a window maps at the header stride like any other xvec, and only the *vvec extensions carry variable-length records. Classifying ivec as variable sent neighbor-facet windows through the offset-index path, and a remote ground-truth file publishes no sidecar, so the window degraded and the consent gate refused an eager warm-up of three records. ElementType now recognizes the *vvec extension families, and the variable-length test fixtures use them instead of overloading .ivecs. Verified against the real remote dataset: a neighbor window resolves to exactly its records' bytes and the reader decodes them. The Rust source's is_vvec_ext still classifies ivec as variable; noted in RUST_COMPATIBILITY.md for upstream reconciliation.
This commit is contained in:
1 parent
683c2d5849
commit
ad58eb804c
7 files changed
+91
-38
No files matched your search
@@ -90,6 +90,13 @@ source directly as of the baseline commit above.
|
||||
|
||||
Remaining representation differences:
|
||||
|
||||
- **`ivec`/`ivecs` facets are uniform-stride.** The format requires
|
||||
length-qualified records that are fixed throughout — ground-truth
|
||||
neighbor files — so windows map at the header stride, and only the
|
||||
`*vvec` extensions carry variable-length records. (The Rust source's
|
||||
`is_vvec_ext` still classifies `ivec` as variable; to be reconciled
|
||||
upstream.)
|
||||
|
||||
- **A non-range-capable remote facet degrades a window** rather than
|
||||
planning a partial fetch it cannot perform. Rust reaches the same end
|
||||
state by downloading such a facet whole at open; Java opens lazily, so
|
||||
|
||||
@@ -49,17 +49,17 @@ public enum ElementType {
|
||||
if (dot < 0) throw new VectorDataException("Vector source has no extension: " + source);
|
||||
String extension = value.substring(dot + 1);
|
||||
return switch (extension) {
|
||||
case "bvec", "bvecs", "u8", "u8vec", "u8vecs" -> U8;
|
||||
case "i8", "i8vec", "i8vecs" -> I8;
|
||||
case "u16", "u16vec", "u16vecs" -> U16;
|
||||
case "i16", "i16vec", "i16vecs" -> I16;
|
||||
case "u32", "u32vec", "u32vecs" -> U32;
|
||||
case "ivec", "ivecs", "i32", "i32vec", "i32vecs" -> I32;
|
||||
case "u64", "u64vec", "u64vecs" -> U64;
|
||||
case "i64", "i64vec", "i64vecs" -> I64;
|
||||
case "mvec", "mvecs", "f16", "f16vec", "f16vecs" -> F16;
|
||||
case "fvec", "fvecs", "f32", "f32vec", "f32vecs" -> F32;
|
||||
case "dvec", "dvecs", "f64", "f64vec", "f64vecs" -> F64;
|
||||
case "bvec", "bvecs", "u8", "u8vec", "u8vecs", "bvvec", "bvvecs", "u8vvec", "u8vvecs" -> U8;
|
||||
case "i8", "i8vec", "i8vecs", "i8vvec", "i8vvecs" -> I8;
|
||||
case "u16", "u16vec", "u16vecs", "u16vvec", "u16vvecs" -> U16;
|
||||
case "i16", "i16vec", "i16vecs", "svvec", "svvecs", "i16vvec", "i16vvecs" -> I16;
|
||||
case "u32", "u32vec", "u32vecs", "u32vvec", "u32vvecs" -> U32;
|
||||
case "ivec", "ivecs", "i32", "i32vec", "i32vecs", "ivvec", "ivvecs", "i32vvec", "i32vvecs" -> I32;
|
||||
case "u64", "u64vec", "u64vecs", "u64vvec", "u64vvecs" -> U64;
|
||||
case "i64", "i64vec", "i64vecs", "i64vvec", "i64vvecs" -> I64;
|
||||
case "mvec", "mvecs", "f16", "f16vec", "f16vecs", "mvvec", "mvvecs", "f16vvec", "f16vvecs" -> F16;
|
||||
case "fvec", "fvecs", "f32", "f32vec", "f32vecs", "fvvec", "fvvecs", "f32vvec", "f32vvecs" -> F32;
|
||||
case "dvec", "dvecs", "f64", "f64vec", "f64vecs", "dvvec", "dvvecs", "f64vvec", "f64vvecs" -> F64;
|
||||
default -> throw new VectorDataException("Unsupported vector extension '." + extension + "' in " + source);
|
||||
};
|
||||
}
|
||||
|
||||
@@ -320,24 +320,34 @@ public final class Prefetcher {
|
||||
if (winEnd <= winStart) return null;
|
||||
String ext = extensionOf(handle.facet.source().toString());
|
||||
if (isVvecExt(ext)) return vvecRangeToBytes(handle, winStart, winEnd);
|
||||
if (isScalarExt(ext)) {
|
||||
int width;
|
||||
try { width = ElementType.forExtension(handle.facet.source().toString()).width(); }
|
||||
catch (VectorDataException unsupported) { return null; }
|
||||
long total = handle.data().size();
|
||||
long byteStart = Math.min(saturatedMultiply(winStart, width), total);
|
||||
long byteEnd = Math.min(saturatedMultiply(winEnd, width), total);
|
||||
return byteStart >= byteEnd ? null : new MappedRange(byteStart, byteEnd, 0);
|
||||
}
|
||||
return uniformRangeToBytes(handle, winStart, winEnd);
|
||||
}
|
||||
|
||||
/// The uniform-stride mapping: `4 + dim × elemSize`, with `dim`
|
||||
/// read from the header at byte 0 — the same first-chunk read any
|
||||
/// reader does on first access, so nothing to report as a
|
||||
/// prerequisite.
|
||||
private static MappedRange uniformRangeToBytes(FacetHandle handle, long winStart, long winEnd) {
|
||||
int width;
|
||||
try { width = ElementType.forExtension(handle.facet.source().toString()).width(); }
|
||||
catch (VectorDataException unsupported) { return null; }
|
||||
ByteStorage storage = handle.data();
|
||||
long total = storage.size();
|
||||
if (isScalarExt(ext)) {
|
||||
long byteStart = Math.min(saturatedMultiply(winStart, width), total);
|
||||
long byteEnd = Math.min(saturatedMultiply(winEnd, width), total);
|
||||
return byteStart >= byteEnd ? null : new MappedRange(byteStart, byteEnd, 0);
|
||||
}
|
||||
if (total < 4) return null;
|
||||
int dim = storage.read(0, 4).order(ByteOrder.LITTLE_ENDIAN).getInt();
|
||||
if (dim <= 0 || dim > 1_000_000) return null; // sanity vs corrupt header
|
||||
long bytesPerRecord = 4 + (long) dim * width;
|
||||
long byteStart = Math.min(saturatedMultiply(winStart, bytesPerRecord), total);
|
||||
long byteEnd = Math.min(saturatedMultiply(winEnd, bytesPerRecord), total);
|
||||
// A uniform-stride mapping costs a 4-byte header read, which
|
||||
// every reader pays on first access anyway — nothing to report.
|
||||
return byteStart >= byteEnd ? null : new MappedRange(byteStart, byteEnd, 0);
|
||||
}
|
||||
|
||||
@@ -372,7 +382,11 @@ public final class Prefetcher {
|
||||
return dot < 0 ? "" : value.substring(dot + 1);
|
||||
}
|
||||
|
||||
private static boolean isVvecExt(String ext) { return ext.contains("vvec") || ext.equals("ivec") || ext.equals("ivecs"); }
|
||||
/// Only the `*vvec` extensions carry variable-length records. In
|
||||
/// this catalog format, `ivec`/`ivecs` records are length-qualified
|
||||
/// but fixed throughout — ground-truth neighbor files — and map at
|
||||
/// the uniform header stride like every other xvec.
|
||||
private static boolean isVvecExt(String ext) { return ext.contains("vvec"); }
|
||||
|
||||
private static boolean isScalarExt(String ext) { return ext.matches("u8|i8|u16|i16|u32|i32|u64|i64"); }
|
||||
}
|
||||
+40
-8
@@ -26,6 +26,7 @@ import java.nio.ByteOrder;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
@@ -89,11 +90,11 @@ class PrefetchRemoteIntegrationTest {
|
||||
Files.write(published.resolve("base.fvec"), fvec);
|
||||
int[][] ragged = new int[400][];
|
||||
for (int record = 0; record < 400; record++) { ragged[record] = new int[1 + record % 17]; }
|
||||
FixtureSupport.vvec(published, "meta.ivecs", ragged);
|
||||
FixtureSupport.vvec(published, "meta.ivvec", ragged);
|
||||
byte[] blob = new byte[40_000]; Arrays.fill(blob, (byte) 7);
|
||||
Files.write(published.resolve("blob.parquet"), blob);
|
||||
if (publishMref) {
|
||||
for (String file : new String[] {"base.fvec", "meta.ivecs", "blob.parquet"}) {
|
||||
for (String file : new String[] {"base.fvec", "meta.ivvec", "blob.parquet"}) {
|
||||
byte[] payload = Files.readAllBytes(published.resolve(file));
|
||||
Files.write(published.resolve(file + ".mref"), FixtureSupport.mref(payload, CHUNK));
|
||||
}
|
||||
@@ -106,7 +107,7 @@ class PrefetchRemoteIntegrationTest {
|
||||
profiles:
|
||||
default:
|
||||
base_vectors: %sbase.fvec
|
||||
metadata_content: %smeta.ivecs
|
||||
metadata_content: %smeta.ivvec
|
||||
metadata_predicates: %sblob.parquet
|
||||
""".formatted(name, base, base, base));
|
||||
VectorDataSettings settings = VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
|
||||
@@ -227,17 +228,17 @@ class PrefetchRemoteIntegrationTest {
|
||||
Path published = Files.createDirectories(temporary.resolve("pub"));
|
||||
int[][] ragged = new int[40][];
|
||||
for (int record = 0; record < 40; record++) ragged[record] = new int[1 + record % 7];
|
||||
FixtureSupport.vvec(published, "meta.ivecs", ragged);
|
||||
Files.delete(published.resolve("IDXFOR__meta.ivecs.i32"));
|
||||
byte[] payload = Files.readAllBytes(published.resolve("meta.ivecs"));
|
||||
Files.write(published.resolve("meta.ivecs.mref"), FixtureSupport.mref(payload, CHUNK));
|
||||
FixtureSupport.vvec(published, "meta.ivvec", ragged);
|
||||
Files.delete(published.resolve("IDXFOR__meta.ivvec.i32"));
|
||||
byte[] payload = Files.readAllBytes(published.resolve("meta.ivvec"));
|
||||
Files.write(published.resolve("meta.ivvec.mref"), FixtureSupport.mref(payload, CHUNK));
|
||||
var server = serveDirectory(published, rangeRequests, true);
|
||||
Path dataset = Files.createDirectories(temporary.resolve("no-sidecar"));
|
||||
Files.writeString(dataset.resolve("dataset.yaml"), """
|
||||
name: no-sidecar
|
||||
profiles:
|
||||
default:
|
||||
metadata_content: http://127.0.0.1:%d/meta.ivecs
|
||||
metadata_content: http://127.0.0.1:%d/meta.ivvec
|
||||
""".formatted(server.getAddress().getPort()));
|
||||
VectorDataSettings isolated = VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
|
||||
TestDataView view = TestDataGroup.load(dataset.resolve("dataset.yaml").toUri(), isolated).profile("default");
|
||||
@@ -254,6 +255,37 @@ class PrefetchRemoteIntegrationTest {
|
||||
"consenting to the degrade fetches the whole facet");
|
||||
}
|
||||
|
||||
/// In this format, `ivec` records are length-qualified but fixed
|
||||
/// throughout — ground-truth neighbor files — so a window maps at
|
||||
/// the uniform header stride with no offset index involved. Only
|
||||
/// the `*vvec` extensions carry variable-length records.
|
||||
@Test void aRemoteIvecWindowMapsAtTheHeaderStride() throws Exception {
|
||||
Path published = Files.createDirectories(temporary.resolve("pub"));
|
||||
int dim = 100, records = 50, recordBytes = 4 + dim * 4;
|
||||
ByteBuffer gt = ByteBuffer.allocate(records * recordBytes).order(ByteOrder.LITTLE_ENDIAN);
|
||||
for (int record = 0; record < records; record++) { gt.putInt(dim); for (int at = 0; at < dim; at++) gt.putInt(record + at); }
|
||||
Files.write(published.resolve("gt.ivecs"), gt.array());
|
||||
Files.write(published.resolve("gt.ivecs.mref"), FixtureSupport.mref(gt.array(), CHUNK));
|
||||
var server = serveDirectory(published, new AtomicInteger(), true);
|
||||
Path dataset = Files.createDirectories(temporary.resolve("uniform-ivec"));
|
||||
Files.writeString(dataset.resolve("dataset.yaml"), """
|
||||
name: uniform-ivec
|
||||
profiles:
|
||||
default:
|
||||
neighbor_indices: http://127.0.0.1:%d/gt.ivecs
|
||||
""".formatted(server.getAddress().getPort()));
|
||||
VectorDataSettings isolated = VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
|
||||
TestDataView view = TestDataGroup.load(dataset.resolve("dataset.yaml").toUri(), isolated).profile("default");
|
||||
|
||||
PrefetchPlan plan = view.prefetchPlan("neighbor_indices", DSWindow.parse("[0..3)"));
|
||||
assertFalse(plan.degradesToFullDownload(), "a uniform ivec is windowable by its header stride");
|
||||
assertEquals(List.of(new ByteRange(0, 3L * recordBytes)), plan.requestedRanges());
|
||||
|
||||
view.prefetch("neighbor_indices", DSWindow.parse("[0..3)"), WholeFacetFallback.REFUSE);
|
||||
assertArrayEquals(new int[] {2, 3}, java.util.Arrays.copyOfRange((int[]) view.neighborIndices().get(2), 0, 2),
|
||||
"the reader decodes the records the prefetch warmed");
|
||||
}
|
||||
|
||||
@Test void allowingTheFallbackFetchesAWholeRemoteFacet() throws Exception {
|
||||
TestDataView view = remoteView("allow-remote", true, true, new AtomicInteger());
|
||||
DSWindow window = DSWindow.parse("2..4");
|
||||
|
||||
@@ -133,13 +133,13 @@ class PrefetchWindowsTest {
|
||||
offsets[record] = at;
|
||||
at += 4 + dims[record] * 4L;
|
||||
}
|
||||
Path data = FixtureSupport.vvec(dataset, "meta.ivecs", records);
|
||||
Path data = FixtureSupport.vvec(dataset, "meta.ivvec", records);
|
||||
Files.writeString(dataset.resolve("dataset.yaml"), """
|
||||
name: prefetch-vvec
|
||||
profiles:
|
||||
default:
|
||||
base_vectors: base_vectors.fvec
|
||||
metadata_content: meta.ivecs
|
||||
metadata_content: meta.ivvec
|
||||
""");
|
||||
return new VvecFixture(TestDataGroup.load(dataset.toUri(), settings()).profile("default"), offsets, Files.size(data));
|
||||
}
|
||||
@@ -179,13 +179,13 @@ class PrefetchWindowsTest {
|
||||
for (int record = 0; record < dims.length; record++) { offsets[record] = size; size += 4 + dims[record] * 4; }
|
||||
ByteBuffer ragged = ByteBuffer.allocate(size).order(ByteOrder.LITTLE_ENDIAN);
|
||||
for (int dim : dims) { ragged.putInt(dim); for (int element = 0; element < dim; element++) ragged.putInt(element); }
|
||||
Files.write(dataset.resolve("ragged.ivecs"), ragged.array());
|
||||
Files.write(dataset.resolve("ragged.ivvec"), ragged.array());
|
||||
Files.writeString(dataset.resolve("dataset.yaml"), """
|
||||
name: prefetch-walk
|
||||
profiles:
|
||||
default:
|
||||
base_vectors: base_vectors.fvec
|
||||
metadata_content: ragged.ivecs
|
||||
metadata_content: ragged.ivvec
|
||||
""");
|
||||
TestDataView view = TestDataGroup.load(dataset.toUri(), settings()).profile("default");
|
||||
PrefetchPlan plan = view.prefetchPlan("metadata_content", DSWindow.parse("1..4"));
|
||||
@@ -194,7 +194,7 @@ class PrefetchWindowsTest {
|
||||
// The walk persists its result beside the data — starts only,
|
||||
// in the width the payload size calls for — so it is paid once
|
||||
// rather than per view.
|
||||
Path persisted = dataset.resolve("IDXFOR__ragged.ivecs.i32");
|
||||
Path persisted = dataset.resolve("IDXFOR__ragged.ivvec.i32");
|
||||
assertTrue(Files.isRegularFile(persisted), "the rebuilt index is persisted beside the data");
|
||||
assertEquals(dims.length * 4L, Files.size(persisted), "record starts only, no sentinel");
|
||||
PrefetchPlan again = TestDataGroup.load(dataset.toUri(), settings()).profile("default")
|
||||
|
||||
@@ -43,10 +43,10 @@ class VectorReadersTest {
|
||||
assertEquals(-1, scalars.get(0)); assertEquals(7, scalars.get(1));
|
||||
TypedReader typed = TypedReaders.open(scalar.toUri(), settings(), "test");
|
||||
assertEquals(-1, typed.scalar(0)); assertEquals(7, typed.scalar(1));
|
||||
Path variable = FixtureSupport.vvec(temporary, "metadata.ivecs", new int[][] {{1}, {2, 3}});
|
||||
Path variable = FixtureSupport.vvec(temporary, "metadata.ivvec", new int[][] {{1}, {2, 3}});
|
||||
@SuppressWarnings("unchecked") VvecReader<int[]> vectors = (VvecReader<int[]>) VectorReaders.openVvec(variable.toUri(), settings(), "test");
|
||||
assertEquals(2, vectors.count()); assertEquals(2, vectors.dimensionAt(1)); assertArrayEquals(new int[] {2, 3}, vectors.get(1));
|
||||
Path variable64 = FixtureSupport.vvec64(temporary, "metadata64.ivecs", new int[][] {{4}, {5, 6}});
|
||||
Path variable64 = FixtureSupport.vvec64(temporary, "metadata64.ivvec", new int[][] {{4}, {5, 6}});
|
||||
@SuppressWarnings("unchecked") VvecReader<int[]> vectors64 = (VvecReader<int[]>) VectorReaders.openVvec(variable64.toUri(), settings(), "test");
|
||||
assertArrayEquals(new int[] {5, 6}, vectors64.get(1));
|
||||
}
|
||||
@@ -55,10 +55,10 @@ class VectorReadersTest {
|
||||
// and no end-of-data sentinel; both layouts must read
|
||||
// identically, or a dataset Rust has opened locally becomes
|
||||
// unreadable here.
|
||||
Path data = FixtureSupport.vvec(temporary, "starts.ivecs", new int[][] {{1}, {2, 3}, {4, 5, 6}});
|
||||
Path data = FixtureSupport.vvec(temporary, "starts.ivvec", new int[][] {{1}, {2, 3}, {4, 5, 6}});
|
||||
java.nio.ByteBuffer startsOnly = java.nio.ByteBuffer.allocate(3 * 4).order(java.nio.ByteOrder.LITTLE_ENDIAN);
|
||||
startsOnly.putInt(0).putInt(8).putInt(20);
|
||||
java.nio.file.Files.write(temporary.resolve("IDXFOR__starts.ivecs.i32"), startsOnly.array());
|
||||
java.nio.file.Files.write(temporary.resolve("IDXFOR__starts.ivvec.i32"), startsOnly.array());
|
||||
@SuppressWarnings("unchecked") VvecReader<int[]> vectors = (VvecReader<int[]>) VectorReaders.openVvec(data.toUri(), settings(), "test");
|
||||
assertEquals(3, vectors.count());
|
||||
assertArrayEquals(new int[] {2, 3}, vectors.get(1));
|
||||
|
||||
+2
-2
@@ -59,7 +59,7 @@ public class VectorDataBindingsTest {
|
||||
ivec(dataset.resolve("gt.ivecs"), new int[][] {{3, 4}});
|
||||
ivec(dataset.resolve("results.ivecs"), new int[][] {{1, 0}, {0, 1}});
|
||||
scalarI32(dataset.resolve("labels.i32"), 7, 1, 2, 3);
|
||||
vvec(dataset, "meta.ivecs", new int[][] {{5}, {6, 7, 8}, {9, 10}, {11, 12, 13, 14}});
|
||||
vvec(dataset, "meta.ivvec", new int[][] {{5}, {6, 7, 8}, {9, 10}, {11, 12, 13, 14}});
|
||||
Files.writeString(dataset.resolve("dataset.yaml"), """
|
||||
name: example
|
||||
profiles:
|
||||
@@ -69,7 +69,7 @@ public class VectorDataBindingsTest {
|
||||
gt: gt.ivecs
|
||||
distances: dist.fvec
|
||||
metadata_results: results.ivecs
|
||||
metadata_content: meta.ivecs
|
||||
metadata_content: meta.ivvec
|
||||
labels: labels.i32
|
||||
custom_floats: extra.fvec
|
||||
windowed:
|
||||
|
||||
Reference in new issue
Block a user