You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by al...@apache.org on 2026/09/21 20:11:24 UTC
(asterixdb) branch master updated: [ASTERIXDB-3817][TEST] Pin VTree cursor agreement
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new bf9a2bf539 [ASTERIXDB-3817][TEST] Pin VTree cursor agreement
bf9a2bf539 is described below
commit bf9a2bf539bc6f3a3542926e0e2f7e440b281b91
Author: Ali Alsuliman <al...@gmail.com>
AuthorDate: Wed Sep 16 11:22:54 2026 -0700
[ASTERIXDB-3817][TEST] Pin VTree cursor agreement
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
The two search cursors decide different things -- the top-K one what a
query returns, the streaming one what a merge writes into durable bytes
-- and they reconcile delete markers by different routes. Nothing
checked that they agree, so a record could be visible to queries and
dropped at merge, or the reverse, which resurrects a deleted row.
LSMVTreeCursorAgreementTest runs one query through both and requires the
same visible set, then forces a full merge and queries after it.
LSMVTreeMultiClusterAgreementTest does the same where components hold
different clusters and the tie is built by DML, which is where a page
ordered on distance alone used to separate a marker from its twin.
Two fixture defects surfaced while writing them. The record generator
numbers records decimally within a distance ring while the index
compares their keys as strings, so record 10 sorted before record 9, and
floating point left a ring's distances slightly out of order as well.
Either way a cluster of eleven or more loaded a chain not ordered on the
key the merge cursor reconciles with. It now emits each cluster in the
order the job feeding the loader sorts it into, keys unchanged.
LSMVTreeChainOrderTest asserted "after the bulk load" on a tree holding
only its static structure, so it never saw any of this. It now loads the
records first, and enough of them to reach the point where the two
orders disagree.
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: I48d06f5e106b3209c9f909f1cd4621ce7770d697
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21810
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Ali Alsuliman <al...@gmail.com>
Reviewed-by: Murtadha Hubail <mh...@apache.org>
Reviewed-by: Ali Alsuliman <al...@gmail.com>
Reviewed-by: Hongyu Shi <le...@gmail.com>
---
.../am/lsm/vector/LSMVTreeChainOrderTest.java | 11 +-
.../am/lsm/vector/LSMVTreeCursorAgreementTest.java | 344 +++++++++++++++++++++
.../vector/LSMVTreeMultiClusterAgreementTest.java | 268 ++++++++++++++++
.../LSMVTreeAntimatterCollisionQuantizedTest.java | 5 +-
.../am/lsm/vector/util/VectorTestStructure.java | 38 ++-
5 files changed, 656 insertions(+), 10 deletions(-)
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeChainOrderTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeChainOrderTest.java
index 7dfe6aad2b..34be165615 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeChainOrderTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeChainOrderTest.java
@@ -61,8 +61,12 @@ public class LSMVTreeChainOrderTest {
private final LSMVTreeTestHarness harness = new LSMVTreeTestHarness();
private final VectorTreeTestUtils testUtils = new VectorTreeTestUtils();
- /** Records per cluster in the bulk-loaded component; the ties are built by insert, not by load. */
- private static final int BULK_RECORDS_PER_CLUSTER = 8;
+ /**
+ * Records per cluster in the bulk-loaded component; the ties are built by insert, not by load.
+ * Past ten, so the generated keys cross the point where their string order stops agreeing with the
+ * order they are numbered in — which is where a load that is not sorted on the full key shows up.
+ */
+ private static final int BULK_RECORDS_PER_CLUSTER = 14;
/** Every tied record carries this vector, so all of them sit at one distance from their centroid. */
private static final double[] TIE_VECTOR = { 20.2, 30.0, 20.0 };
@@ -168,6 +172,9 @@ public class LSMVTreeChainOrderTest {
@Test
public void everyDiskWriterLeavesTheChainSorted() throws Exception {
withIndex(newContext(false), (ctx, tree) -> {
+ // The records have to actually be loaded for this first assertion to say anything: without
+ // it the tree holds only the static structure and there is no chain to be out of order.
+ testUtils.bulkLoadRecords(ctx);
assertChainSorted("after the bulk load", tree);
testUtils.insertRecordsIntoMemoryComponent(ctx, List.of(tiedRecords()));
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeCursorAgreementTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeCursorAgreementTest.java
new file mode 100644
index 0000000000..8050ec9ded
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeCursorAgreementTest.java
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hyracks.storage.am.lsm.vector;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.util.HyracksConstants;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.dataflow.common.data.marshalling.DoubleArraySerializerDeserializer;
+import org.apache.hyracks.storage.am.common.TestOperationCallback;
+import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
+import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationStatus;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.vector.impls.LSMVTree;
+import org.apache.hyracks.storage.am.lsm.vector.impls.LSMVTreeTopKSearchCursor;
+import org.apache.hyracks.storage.am.lsm.vector.util.LSMVTreeTestContext;
+import org.apache.hyracks.storage.am.lsm.vector.util.LSMVTreeTestHarness;
+import org.apache.hyracks.storage.am.lsm.vector.util.QuantizedSearchTestDriver;
+import org.apache.hyracks.storage.am.lsm.vector.util.VectorTestStructure;
+import org.apache.hyracks.storage.am.lsm.vector.util.VectorTestStructure.BulkLoadRecordFormat;
+import org.apache.hyracks.storage.am.vector.AbstractVectorTreeTestContext;
+import org.apache.hyracks.storage.am.vector.TestDoubleArrayVectorAccessor;
+import org.apache.hyracks.storage.am.vector.VectorTreeTestUtils;
+import org.apache.hyracks.storage.am.vector.api.IVTreeBinaryAccessorFactory;
+import org.apache.hyracks.storage.am.vector.api.IVTreeQuantizer;
+import org.apache.hyracks.storage.am.vector.impls.VTreeSearchPredicate;
+import org.apache.hyracks.storage.am.vector.utils.NoOpVectorQuantizer;
+import org.apache.hyracks.storage.common.IIndexAccessor;
+import org.apache.hyracks.storage.common.IIndexCursor;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Asserts that the two LSM VTree search cursors agree on what a record's history makes visible.
+ * <p>
+ * The index has two: {@code LSMVTreeTopKSearchCursor} answers production ANN queries, and
+ * {@code LSMVTreeSearchCursor} streams components during merges — so the first decides what a user sees
+ * and the second decides what survives into durable bytes. They reconcile delete markers by different
+ * routes: the streaming cursor drains every older same-key version against a held reference, the top-K
+ * cursor groups the equal-distance run and takes the newest entry per key. Both rest on the same ordering
+ * key (the index's {@code comparatorFields}), but nothing before this test checked that they reach the
+ * same answer.
+ * <p>
+ * If they ever diverge, a record could be visible to queries and dropped at merge, or the reverse — the
+ * latter resurrecting a deleted row. That is what makes agreement, rather than either cursor's own
+ * output, the property worth pinning.
+ * <p>
+ * K is set well above the record count on purpose: the question is which records are <em>visible</em>,
+ * and a K that truncates would make the two sets differ for reasons that have nothing to do with
+ * reconciliation.
+ */
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED)
+public class LSMVTreeCursorAgreementTest {
+
+ private static final VectorTestStructure STRUCT_3D = VectorTestStructure.threeDim1Centroid();
+ /** Above the record count, so neither cursor truncates and the comparison is about visibility. */
+ private static final int K = 10;
+
+ private final LSMVTreeTestHarness harness = new LSMVTreeTestHarness();
+ private final VectorTreeTestUtils testUtils = new VectorTreeTestUtils();
+
+ @Before
+ public void setUp() throws HyracksDataException {
+ harness.setUp();
+ }
+
+ @After
+ public void tearDown() throws HyracksDataException {
+ harness.tearDown();
+ }
+
+ /**
+ * A delete whose distance-to-centroid ties with an unrelated record in the same component, so that
+ * record lands between the delete marker and its live twin in the merged sequence.
+ * <p>
+ * The streaming cursor holds one antimatter element and cancels it against the next queue head, so it
+ * rests on a marker and its twin being adjacent in the merged stream. They are, because every
+ * component's chain is ordered on the same {@code <distance, PK>} key the cursor reconciles with.
+ * Before the pages carried the primary key, an intervening tie released the hold and the deleted
+ * record was emitted — by the cursor that merges components, i.e. into durable bytes.
+ */
+ @Test
+ public void bothCursorsAgreeAfterADeleteAtACollidingDistance() throws Exception {
+ AbstractVectorTreeTestContext ctx = newContext(collidingDistanceCluster());
+ try {
+ ctx.getIndex().create();
+ ctx.getIndex().activate();
+ testUtils.buildStaticStructure(ctx);
+ testUtils.bulkLoadRecords(ctx);
+
+ // Delete A, whose distance ties with B. The marker lands in the memory component, so this is
+ // cross-component reconciliation.
+ testUtils.deleteRecordsFromIndex(ctx,
+ Arrays.asList(VectorTestStructure.createInsertTuple(new double[] { 5, 0, 0 }, "pk_collide_A")));
+
+ assertCursorsAgree(ctx, new double[] { 5, 0, 0 }, "pk_collide_A");
+ } finally {
+ ctx.getIndex().deactivate();
+ }
+ }
+
+ /**
+ * Insert, delete, then re-insert the same primary key. Three versions of one key exist across
+ * components, which is where reconciliation that only cancels a single pair leaks the third.
+ */
+ @Test
+ public void bothCursorsAgreeAfterDeleteThenReinsert() throws Exception {
+ AbstractVectorTreeTestContext ctx = newContext(collidingDistanceCluster());
+ try {
+ ctx.getIndex().create();
+ ctx.getIndex().activate();
+ testUtils.buildStaticStructure(ctx);
+ testUtils.bulkLoadRecords(ctx);
+
+ ITupleReference recordA = VectorTestStructure.createInsertTuple(new double[] { 5, 0, 0 }, "pk_collide_A");
+ testUtils.deleteRecordsFromIndex(ctx, Arrays.asList(recordA));
+ // Re-insert the same key: the newest version is matter again, so it must be visible to BOTH.
+ testUtils.insertRecordsIntoMemoryComponent(ctx, Arrays.asList(Arrays.asList(recordA)));
+
+ assertCursorsAgree(ctx, new double[] { 5, 0, 0 });
+ } finally {
+ ctx.getIndex().deactivate();
+ }
+ }
+
+ /** With no deletes at all the two cursors must still see the same records. */
+ @Test
+ public void bothCursorsAgreeWithNoDeletes() throws Exception {
+ AbstractVectorTreeTestContext ctx = newContext(collidingDistanceCluster());
+ try {
+ ctx.getIndex().create();
+ ctx.getIndex().activate();
+ testUtils.buildStaticStructure(ctx);
+ testUtils.bulkLoadRecords(ctx);
+
+ assertCursorsAgree(ctx, new double[] { 5, 0, 0 });
+ } finally {
+ ctx.getIndex().deactivate();
+ }
+ }
+
+ /**
+ * The durable half: force the streaming cursor to actually write the merged bytes, then query.
+ * <p>
+ * The other tests read the streaming cursor as a query cursor, which production never does — queries
+ * always take the top-K path. This one puts the cursor where it really runs. A full merge over every
+ * disk component drops delete markers rather than preserving them, so a marker that fails to cancel
+ * its twin leaves the twin alone in the merged component with nothing left to suppress it: the record
+ * is resurrected in durable bytes, and the query afterwards sees it.
+ */
+ @Test
+ public void aFullMergeDoesNotResurrectADeletedRecordAtACollidingDistance() throws Exception {
+ AbstractVectorTreeTestContext ctx = newContext(collidingDistanceCluster());
+ try {
+ ctx.getIndex().create();
+ ctx.getIndex().activate();
+ testUtils.buildStaticStructure(ctx);
+ testUtils.bulkLoadRecords(ctx);
+
+ LSMVTree lsmVTree = (LSMVTree) ctx.getIndex();
+ Assert.assertEquals("one disk component after bulk load", 1, lsmVTree.getDiskComponents().size());
+
+ // A's matter is on disk, so the delete leaves a marker in the memory component; flush it down
+ // so both live in disk components and the merge has to reconcile them.
+ testUtils.deleteRecordsFromIndex(ctx,
+ Arrays.asList(VectorTestStructure.createInsertTuple(new double[] { 5, 0, 0 }, "pk_collide_A")));
+ flush(ctx);
+ Assert.assertEquals("marker flushed to its own component", 2, lsmVTree.getDiskComponents().size());
+
+ Set<String> beforeMerge = visiblePrimaryKeys(ctx, new double[] { 5, 0, 0 }, true);
+ Assert.assertFalse("precondition: the delete is already invisible to queries",
+ beforeMerge.contains("pk_collide_A"));
+
+ // Full merge: every disk component, so markers are dropped instead of carried forward.
+ List<ILSMDiskComponent> allDiskComponents = new ArrayList<>(lsmVTree.getDiskComponents());
+ ILSMIndexAccessor lsmAccessor =
+ (ILSMIndexAccessor) ctx.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ ILSMIOOperation mergeOp = lsmAccessor.scheduleMerge(allDiskComponents);
+ mergeOp.sync();
+ if (mergeOp.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(mergeOp.getFailure());
+ }
+ Assert.assertEquals("one component after a full merge", 1, lsmVTree.getDiskComponents().size());
+
+ Set<String> afterMerge = visiblePrimaryKeys(ctx, new double[] { 5, 0, 0 }, true);
+ Assert.assertFalse("the deleted record was written back into the merged component",
+ afterMerge.contains("pk_collide_A"));
+ // The merge must not have taken the live records with it.
+ Assert.assertEquals("merge changed what is visible beyond the deleted record", beforeMerge, afterMerge);
+ Assert.assertTrue("the same-distance sibling must survive", afterMerge.contains("pk_collide_B"));
+ } finally {
+ ctx.getIndex().deactivate();
+ }
+ }
+
+ private void flush(AbstractVectorTreeTestContext ctx) throws HyracksDataException, InterruptedException {
+ ILSMIndexAccessor accessor =
+ (ILSMIndexAccessor) ctx.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ ILSMIOOperation flushOp = accessor.scheduleFlush();
+ flushOp.sync();
+ if (flushOp.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(flushOp.getFailure());
+ }
+ }
+
+ // ---- the agreement assertion --------------------------------------------------------------
+
+ /**
+ * Runs one query through both cursors and requires the same visible set, then requires every named
+ * key to be absent from it. The absence check is separate because two cursors that both wrongly
+ * showed a deleted record would agree with each other.
+ */
+ private void assertCursorsAgree(AbstractVectorTreeTestContext ctx, double[] queryVector, String... mustBeAbsent)
+ throws Exception {
+ Set<String> viaTopK = visiblePrimaryKeys(ctx, queryVector, true);
+ Set<String> viaStreaming = visiblePrimaryKeys(ctx, queryVector, false);
+
+ Assert.assertEquals("the two cursors disagree on what is visible", viaStreaming, viaTopK);
+ for (String absent : mustBeAbsent) {
+ Assert.assertFalse("deleted record " + absent + " is visible to the top-K cursor",
+ viaTopK.contains(absent));
+ Assert.assertFalse("deleted record " + absent + " is visible to the streaming cursor",
+ viaStreaming.contains(absent));
+ }
+ Assert.assertFalse("the query returned nothing, so agreement is vacuous", viaTopK.isEmpty());
+ }
+
+ /** @param useTopK selects the cursor exactly the way the production search operator does */
+ private Set<String> visiblePrimaryKeys(AbstractVectorTreeTestContext ctx, double[] queryVector, boolean useTopK)
+ throws Exception {
+ ArrayTupleBuilder queryTupleBuilder = new ArrayTupleBuilder(1);
+ queryTupleBuilder.addField(DoubleArraySerializerDeserializer.INSTANCE, queryVector);
+ ArrayTupleReference queryTuple = new ArrayTupleReference();
+ queryTuple.reset(queryTupleBuilder.getFieldEndOffsets(), queryTupleBuilder.getByteArray());
+
+ VTreeSearchPredicate predicate = new VTreeSearchPredicate();
+ predicate.setQueryTuple(queryTuple);
+ predicate.setQueryFieldIndex(0);
+ predicate.setK(K);
+
+ IndexAccessParameters iap =
+ new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE);
+ iap.getParameters().put(IVTreeBinaryAccessorFactory.IAP_KEY, TestDoubleArrayVectorAccessor.Factory.INSTANCE);
+ iap.getParameters().put(IVTreeQuantizer.IAP_KEY, NoOpVectorQuantizer.INSTANCE);
+ iap.getParameters().put(HyracksConstants.HYRACKS_TASK_CONTEXT, ctx.getHyracksTaskContext());
+ if (useTopK) {
+ iap.getParameters().put(LSMVTreeTopKSearchCursor.IAP_KEY, Boolean.TRUE);
+ }
+
+ IIndexAccessor accessor = ctx.getIndex().createAccessor(iap);
+ IIndexCursor cursor = accessor.createSearchCursor(false);
+ Set<String> primaryKeys = new LinkedHashSet<>();
+ try {
+ accessor.search(cursor, predicate);
+ try {
+ while (cursor.hasNext()) {
+ cursor.next();
+ // Extracted at emission time: getTuple() hands back a reference the cursor mutates.
+ primaryKeys.add(testUtils.extractPrimaryKeyFromOptimizedTuple(cursor.getTuple()));
+ }
+ } finally {
+ cursor.close();
+ }
+ } finally {
+ cursor.destroy();
+ }
+ return primaryKeys;
+ }
+
+ // ---- fixture -----------------------------------------------------------------------------
+
+ /**
+ * One leaf cluster whose records are distance-ordered, with two records tied at distance 5, so an
+ * unrelated record shares the delete marker's distance.
+ * <p>
+ * The tie is listed in primary-key order because the bulk loader takes its input as the stored order
+ * and rejects anything else. Mis-ordering is covered where it can still arise: at the loader's own
+ * boundary by {@code LSMVTreeBulkLoadOrderTest}, and through DML by
+ * {@code LSMVTreeMultiClusterAgreementTest}.
+ */
+ private static List<List<ITupleReference>> collidingDistanceCluster() throws Exception {
+ List<ITupleReference> cluster = new ArrayList<>();
+ cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(3.0, 0, new double[] { 3, 0, 0 },
+ "pk_filler_3"));
+ cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(4.0, 0, new double[] { 4, 0, 0 },
+ "pk_filler_4"));
+ cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(5.0, 0, new double[] { 5, 0, 0 },
+ "pk_collide_A"));
+ cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(5.0, 0, new double[] { 0, 5, 0 },
+ "pk_collide_B"));
+ cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(6.0, 0, new double[] { 6, 0, 0 },
+ "pk_filler_6"));
+ List<List<ITupleReference>> leafRecords = new ArrayList<>();
+ leafRecords.add(cluster);
+ return leafRecords;
+ }
+
+ private AbstractVectorTreeTestContext newContext(List<List<ITupleReference>> leafRecords) throws Exception {
+ ISerializerDeserializer[] dataRecordSerdes = STRUCT_3D.getDataRecordSerdes(BulkLoadRecordFormat.QUANTIZED);
+ AbstractVectorTreeTestContext ctx = LSMVTreeTestContext.create(harness.getNcConfig(), harness.getIOManager(),
+ harness.getVirtualBufferCaches(), harness.getFileReference(), harness.getDiskBufferCache(),
+ dataRecordSerdes, STRUCT_3D.getVectorDimension(), harness.getMergePolicy(),
+ harness.getOperationTracker(), harness.getIOScheduler(), harness.getIOOperationCallbackFactory(),
+ harness.getPageWriteCallbackFactory(), harness.getMetadataPageManagerFactory(),
+ harness.getDataTupleBuilderFactory());
+ ctx.setHyracksTaskContext(harness.getHyracksTastContext());
+ ctx.setStaticStructureCentroids(STRUCT_3D.buildCentroidTuples());
+ ctx.setNumClustersPerLevel(STRUCT_3D.getNumClustersPerLevel());
+ ctx.setNumCentroidsPerLevel(STRUCT_3D.getCentroidsPerCluster());
+ ctx.setDataRecords(leafRecords);
+ return ctx;
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeMultiClusterAgreementTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeMultiClusterAgreementTest.java
new file mode 100644
index 0000000000..6e0a3dfab1
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeMultiClusterAgreementTest.java
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hyracks.storage.am.lsm.vector;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.util.HyracksConstants;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+import org.apache.hyracks.dataflow.common.data.marshalling.DoubleArraySerializerDeserializer;
+import org.apache.hyracks.storage.am.common.TestOperationCallback;
+import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
+import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationStatus;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.vector.impls.LSMVTree;
+import org.apache.hyracks.storage.am.lsm.vector.impls.LSMVTreeTopKSearchCursor;
+import org.apache.hyracks.storage.am.lsm.vector.util.LSMVTreeTestContext;
+import org.apache.hyracks.storage.am.lsm.vector.util.LSMVTreeTestHarness;
+import org.apache.hyracks.storage.am.lsm.vector.util.VectorTestStructure;
+import org.apache.hyracks.storage.am.lsm.vector.util.VectorTestStructure.BulkLoadRecordFormat;
+import org.apache.hyracks.storage.am.vector.AbstractVectorTreeTestContext;
+import org.apache.hyracks.storage.am.vector.TestDoubleArrayVectorAccessor;
+import org.apache.hyracks.storage.am.vector.VectorTreeTestUtils;
+import org.apache.hyracks.storage.am.vector.api.IVTreeBinaryAccessorFactory;
+import org.apache.hyracks.storage.am.vector.api.IVTreeQuantizer;
+import org.apache.hyracks.storage.am.vector.impls.VTreeSearchPredicate;
+import org.apache.hyracks.storage.am.vector.utils.NoOpVectorQuantizer;
+import org.apache.hyracks.storage.common.IIndexAccessor;
+import org.apache.hyracks.storage.common.IIndexCursor;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * The multi-cluster half of {@link LSMVTreeCursorAgreementTest}, over a tie built by DML rather than by
+ * bulk load. The streaming cursor cancels a delete marker against the next queue head only, so it rests
+ * on the marker and its twin being adjacent in the merged stream. Two things have to hold for that:
+ * every component's chain is ordered on the ordering key, and the components are all probing the same
+ * cluster at the same time.
+ * <p>
+ * The tie is inserted in the order that used to defeat this. A data page positioned a tuple by distance
+ * alone, so two records tied on distance were stored in arrival order; arriving after its tie partner
+ * put the record that sorts <em>first</em> physically <em>second</em>, and the marker's twin was then no
+ * longer the next head. Reverse the two inserts and the merged stream is accidentally correct, so the
+ * order here is the point of the fixture rather than an accident of it.
+ * <p>
+ * Components are given unequal cluster occupancy on purpose — the memory component holds only the
+ * marker's cluster, one disk component only cluster A, another only cluster B, the bulk-loaded one every
+ * cluster — and cluster B's record sits closer to its centroid than the deleted one does to cluster A's.
+ * If per-component cluster iteration were not positionally aligned, the merged stream would interleave
+ * the two clusters and release the hold before the twin arrived.
+ */
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED)
+public class LSMVTreeMultiClusterAgreementTest {
+
+ private static final VectorTestStructure STRUCT = VectorTestStructure.threeDim3Level();
+
+ /** Above the record count, so neither cursor truncates and the comparison is about visibility. */
+ private static final int K = 200;
+
+ private static final int BULK_RECORDS_PER_CLUSTER = 4;
+
+ /** Cluster A is leaf centroid c10 at {20,30,20}; both vectors sit at distance 3 from it. */
+ private static final double[] TIE_FIRST_WRITTEN = { 23.0, 30.0, 20.0 };
+ private static final double[] TIE_SECOND_WRITTEN = { 20.0, 33.0, 20.0 };
+
+ /** The deleted key sorts before its tie partner, but is written after it. */
+ private static final String PK_TIE_KEPT = "pk_tieA_b";
+ private static final String PK_TIE_DELETED = "pk_tieA_a";
+
+ /** Cluster B is leaf centroid c13 at {-20,30,20}, holding a record closer to it than the tie. */
+ private static final double[] NEAR_CLUSTER_B = { -19.0, 30.0, 20.0 };
+ private static final String PK_CLUSTER_B = "pk_clusterB";
+
+ private final LSMVTreeTestHarness harness = new LSMVTreeTestHarness();
+ private final VectorTreeTestUtils testUtils = new VectorTreeTestUtils();
+
+ @Before
+ public void setUp() throws HyracksDataException {
+ harness.setUp();
+ }
+
+ @After
+ public void tearDown() throws HyracksDataException {
+ harness.tearDown();
+ }
+
+ @Test
+ public void bothCursorsAgreeAcrossClustersAfterADelete() throws Exception {
+ AbstractVectorTreeTestContext ctx = newContext();
+ try {
+ buildFixture(ctx);
+ assertCursorsAgree(ctx, TIE_FIRST_WRITTEN, PK_TIE_DELETED);
+ } finally {
+ ctx.getIndex().deactivate();
+ }
+ }
+
+ /**
+ * The durable half: a full merge drops delete markers rather than carrying them forward, so a marker
+ * that fails to cancel leaves its twin alone in the merged component with nothing left to suppress
+ * it.
+ */
+ @Test
+ public void aFullMergeAcrossClustersDoesNotResurrectTheDeletedRecord() throws Exception {
+ AbstractVectorTreeTestContext ctx = newContext();
+ try {
+ buildFixture(ctx);
+ LSMVTree tree = (LSMVTree) ctx.getIndex();
+
+ Set<String> beforeMerge = visiblePrimaryKeys(ctx, TIE_FIRST_WRITTEN, true);
+ Assert.assertFalse("precondition: the delete is already invisible to queries",
+ beforeMerge.contains(PK_TIE_DELETED));
+
+ List<ILSMDiskComponent> allDiskComponents = new ArrayList<>(tree.getDiskComponents());
+ Assert.assertTrue("the fixture must leave several disk components to merge", allDiskComponents.size() >= 3);
+ ILSMIndexAccessor lsmAccessor =
+ (ILSMIndexAccessor) ctx.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ ILSMIOOperation mergeOp = lsmAccessor.scheduleMerge(allDiskComponents);
+ mergeOp.sync();
+ if (mergeOp.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(mergeOp.getFailure());
+ }
+
+ Set<String> afterMerge = visiblePrimaryKeys(ctx, TIE_FIRST_WRITTEN, true);
+ Assert.assertFalse("the deleted record was written back into the merged component",
+ afterMerge.contains(PK_TIE_DELETED));
+ Assert.assertTrue("its tie partner must survive the merge", afterMerge.contains(PK_TIE_KEPT));
+ } finally {
+ ctx.getIndex().deactivate();
+ }
+ }
+
+ /**
+ * Bulk-load every cluster, then split the later writes across two clusters and two components so no
+ * two components hold the same set, and leave the delete marker alone in the memory component.
+ */
+ private void buildFixture(AbstractVectorTreeTestContext ctx) throws Exception {
+ ctx.getIndex().create();
+ ctx.getIndex().activate();
+ testUtils.buildStaticStructure(ctx);
+ testUtils.bulkLoadRecords(ctx);
+
+ testUtils.insertRecordsIntoMemoryComponent(ctx,
+ List.of(List.of(VectorTestStructure.createInsertTuple(TIE_FIRST_WRITTEN, PK_TIE_KEPT),
+ VectorTestStructure.createInsertTuple(TIE_SECOND_WRITTEN, PK_TIE_DELETED))));
+ flush(ctx);
+
+ testUtils.insertRecordsIntoMemoryComponent(ctx,
+ List.of(List.of(VectorTestStructure.createInsertTuple(NEAR_CLUSTER_B, PK_CLUSTER_B))));
+ flush(ctx);
+
+ // The matter is on disk by now, so the delete finds nothing local to remove and leaves a marker.
+ testUtils.deleteRecordsFromIndex(ctx,
+ Arrays.asList(VectorTestStructure.createInsertTuple(TIE_SECOND_WRITTEN, PK_TIE_DELETED)));
+ }
+
+ private void assertCursorsAgree(AbstractVectorTreeTestContext ctx, double[] queryVector, String... mustBeAbsent)
+ throws Exception {
+ Set<String> viaTopK = visiblePrimaryKeys(ctx, queryVector, true);
+ Set<String> viaStreaming = visiblePrimaryKeys(ctx, queryVector, false);
+
+ Assert.assertEquals("the two cursors disagree on what is visible", viaStreaming, viaTopK);
+ for (String absent : mustBeAbsent) {
+ Assert.assertFalse("deleted record " + absent + " is visible to the top-K cursor",
+ viaTopK.contains(absent));
+ Assert.assertFalse("deleted record " + absent + " is visible to the streaming cursor",
+ viaStreaming.contains(absent));
+ }
+ Assert.assertTrue("the tie partner must stay visible", viaTopK.contains(PK_TIE_KEPT));
+ Assert.assertTrue("the other cluster's record must stay visible", viaTopK.contains(PK_CLUSTER_B));
+ }
+
+ /** @param useTopK selects the cursor exactly the way the production search operator does */
+ private Set<String> visiblePrimaryKeys(AbstractVectorTreeTestContext ctx, double[] queryVector, boolean useTopK)
+ throws Exception {
+ ArrayTupleBuilder queryTupleBuilder = new ArrayTupleBuilder(1);
+ queryTupleBuilder.addField(DoubleArraySerializerDeserializer.INSTANCE, queryVector);
+ ArrayTupleReference queryTuple = new ArrayTupleReference();
+ queryTuple.reset(queryTupleBuilder.getFieldEndOffsets(), queryTupleBuilder.getByteArray());
+
+ VTreeSearchPredicate predicate = new VTreeSearchPredicate();
+ predicate.setQueryTuple(queryTuple);
+ predicate.setQueryFieldIndex(0);
+ predicate.setK(K);
+
+ IndexAccessParameters iap =
+ new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE);
+ iap.getParameters().put(IVTreeBinaryAccessorFactory.IAP_KEY, TestDoubleArrayVectorAccessor.Factory.INSTANCE);
+ iap.getParameters().put(IVTreeQuantizer.IAP_KEY, NoOpVectorQuantizer.INSTANCE);
+ iap.getParameters().put(HyracksConstants.HYRACKS_TASK_CONTEXT, ctx.getHyracksTaskContext());
+ if (useTopK) {
+ iap.getParameters().put(LSMVTreeTopKSearchCursor.IAP_KEY, Boolean.TRUE);
+ }
+
+ IIndexAccessor accessor = ctx.getIndex().createAccessor(iap);
+ IIndexCursor cursor = accessor.createSearchCursor(false);
+ Set<String> primaryKeys = new LinkedHashSet<>();
+ try {
+ accessor.search(cursor, predicate);
+ try {
+ while (cursor.hasNext()) {
+ cursor.next();
+ // Extracted at emission time: getTuple() hands back a reference the cursor mutates.
+ primaryKeys.add(testUtils.extractPrimaryKeyFromOptimizedTuple(cursor.getTuple()));
+ }
+ } finally {
+ cursor.close();
+ }
+ } finally {
+ cursor.destroy();
+ }
+ return primaryKeys;
+ }
+
+ private void flush(AbstractVectorTreeTestContext ctx) throws HyracksDataException, InterruptedException {
+ ILSMIndexAccessor accessor =
+ (ILSMIndexAccessor) ctx.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ ILSMIOOperation flushOp = accessor.scheduleFlush();
+ flushOp.sync();
+ if (flushOp.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(flushOp.getFailure());
+ }
+ }
+
+ private AbstractVectorTreeTestContext newContext() throws Exception {
+ ISerializerDeserializer[] dataRecordSerdes = STRUCT.getDataRecordSerdes(BulkLoadRecordFormat.QUANTIZED);
+ AbstractVectorTreeTestContext ctx = LSMVTreeTestContext.create(harness.getNcConfig(), harness.getIOManager(),
+ harness.getVirtualBufferCaches(), harness.getFileReference(), harness.getDiskBufferCache(),
+ dataRecordSerdes, STRUCT.getVectorDimension(), harness.getMergePolicy(), harness.getOperationTracker(),
+ harness.getIOScheduler(), harness.getIOOperationCallbackFactory(),
+ harness.getPageWriteCallbackFactory(), harness.getMetadataPageManagerFactory(),
+ harness.getDataTupleBuilderFactory());
+ ctx.setHyracksTaskContext(harness.getHyracksTastContext());
+ ctx.setStaticStructureCentroids(STRUCT.buildCentroidTuples());
+ ctx.setNumClustersPerLevel(STRUCT.getNumClustersPerLevel());
+ ctx.setNumCentroidsPerLevel(STRUCT.getCentroidsPerCluster());
+ ctx.setDataRecords(STRUCT.generateBulkLoadRecords(BulkLoadRecordFormat.QUANTIZED, BULK_RECORDS_PER_CLUSTER));
+ return ctx;
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/quantized/LSMVTreeAntimatterCollisionQuantizedTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/quantized/LSMVTreeAntimatterCollisionQuantizedTest.java
index ec606db2fd..c285492f31 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/quantized/LSMVTreeAntimatterCollisionQuantizedTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/quantized/LSMVTreeAntimatterCollisionQuantizedTest.java
@@ -84,10 +84,11 @@ public class LSMVTreeAntimatterCollisionQuantizedTest {
"pk_filler_3"));
cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(4.0, 0, new double[] { 4, 0, 0 },
"pk_filler_4"));
- cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(5.0, 0, new double[] { 0, 5, 0 },
- "pk_collide_B"));
+ // The tie is listed in primary-key order: the bulk loader takes its input as the stored order.
cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(5.0, 0, new double[] { 5, 0, 0 },
"pk_collide_A"));
+ cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(5.0, 0, new double[] { 0, 5, 0 },
+ "pk_collide_B"));
cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(6.0, 0, new double[] { 6, 0, 0 },
"pk_filler_6"));
cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(7.0, 0, new double[] { 7, 0, 0 },
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/util/VectorTestStructure.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/util/VectorTestStructure.java
index 90c4d2823f..5ca445529e 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/util/VectorTestStructure.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/util/VectorTestStructure.java
@@ -21,6 +21,7 @@ package org.apache.hyracks.storage.am.lsm.vector.util;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Random;
@@ -380,7 +381,11 @@ public class VectorTestStructure {
List<List<ITupleReference>> allRecords = new ArrayList<>();
for (int centroidIndex = 0; centroidIndex < leafCentroids.length; centroidIndex++) {
- List<ITupleReference> clusterRecords = new ArrayList<>();
+ // Generated, then sorted before it is handed over: the bulk loader takes its input as the
+ // stored order, and in production the job feeding it sorts on <centroidId, distance, PK>.
+ // Generation order is neither — a ring puts records at one distance, and their keys are
+ // compared as strings, so record 10 sorts before record 9.
+ List<GeneratedRecord> generated = new ArrayList<>();
int centroidId = firstLeafCentroidId + centroidIndex;
double[] centroid = leafCentroids[centroidIndex];
@@ -403,9 +408,10 @@ public class VectorTestStructure {
double[] vector = addVectors(centroid, offset);
double actualDistance = computeEuclideanDistance(vector, centroid);
+ ITupleReference record;
switch (format) {
case NAIVE:
- clusterRecords.add(createNaiveBulkLoadRecord(actualDistance, centroidId, primaryKey));
+ record = createNaiveBulkLoadRecord(actualDistance, centroidId, primaryKey);
break;
case NAIVE_WITH_INCLUDES:
if (includeFieldSerdes == null || includeFieldValueGenerator == null) {
@@ -413,28 +419,48 @@ public class VectorTestStructure {
"NAIVE_WITH_INCLUDES requires include fields configured via withIncludeFields()");
}
Object[] includeValues = includeFieldValueGenerator.generate(centroidId, recordCount);
- clusterRecords.add(createNaiveWithIncludesBulkLoadRecord(actualDistance, centroidId,
- primaryKey, includeFieldSerdes, includeValues));
+ record = createNaiveWithIncludesBulkLoadRecord(actualDistance, centroidId, primaryKey,
+ includeFieldSerdes, includeValues);
break;
case QUANTIZED:
- clusterRecords
- .add(createQuantizedBulkLoadRecord(actualDistance, centroidId, vector, primaryKey));
+ record = createQuantizedBulkLoadRecord(actualDistance, centroidId, vector, primaryKey);
break;
default:
throw new UnsupportedOperationException("Format not yet supported: " + format);
}
+ generated.add(new GeneratedRecord(actualDistance, primaryKey, record));
recordCount++;
}
baseDistance += 0.2;
}
+ // Primary keys are ASCII here, so string order is their comparator order.
+ generated.sort(
+ Comparator.<GeneratedRecord> comparingDouble(r -> r.distance).thenComparing(r -> r.primaryKey));
+ List<ITupleReference> clusterRecords = new ArrayList<>(generated.size());
+ for (GeneratedRecord r : generated) {
+ clusterRecords.add(r.record);
+ }
allRecords.add(clusterRecords);
}
return allRecords;
}
+ /** A generated record with the two values its stored order is decided by. */
+ private static final class GeneratedRecord {
+ private final double distance;
+ private final String primaryKey;
+ private final ITupleReference record;
+
+ private GeneratedRecord(double distance, String primaryKey, ITupleReference record) {
+ this.distance = distance;
+ this.primaryKey = primaryKey;
+ this.record = record;
+ }
+ }
+
// ===== Insert Record Generation =====
/**