You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/09/04 00:04:14 UTC
[L] Change in asterixdb[master]: [ASTERIXDB-3817][STO] Fix delete cancellation in the merge cursor
From Ali Alsuliman <al...@gmail.com>:
Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21681?usp=email )
Change subject: [ASTERIXDB-3817][STO] Fix delete cancellation in the merge cursor
......................................................................
[ASTERIXDB-3817][STO] Fix delete cancellation in the merge cursor
LSMVTreeSearchCursor merges components, so it decides what survives into
durable bytes. It cancelled a delete marker by holding one element and
comparing it against the next queue head only. When an unrelated record
at the same distance-to-centroid arrived first, the hold was released
and the marker's live twin came through. A full merge drops markers
rather than carrying them forward, so the twin then sat alone in the
merged component with nothing left to suppress it: the deleted record
was resurrected in durable bytes and visible to queries afterwards.
The ordering key does include the primary key, which is why this looked
safe. It is not sufficient: the queue holds one element per component,
so the key only orders across components. It cannot make a component's
own stream PK-ordered inside a distance tie, and a data page stores
same-distance records in load order.
Markers now record their <distance, PK> for the cluster being probed
instead of being held, so a cancellation survives any number of
intervening records. Two scoping details matter and both came from test
failures rather than from reading the code:
Cluster-scoped, not distance-run-scoped. Probing a new cluster can bring
in records whose distance-to-centroid is smaller than the one just
emitted, so "the distance advanced" does not mean "no more versions of
this key". A key's marker and its matter always share a cluster, because
a delete replicates into the same clusters the insert did.
Suppression applies only to strictly older components. Across components
a lower cursor index is newer, but within one component later in the
stream is newer -- and a memory component can hold both a marker and a
re-inserted matter for one key, since the delete finds no local matter
to remove and leaves a marker behind. Suppressing on an equal index
dropped that re-insert, which is still live.
VTreeMergeKey extracts the ordering key the two cursors each computed
from their own copy of the same loop. Both reconciliation algorithms
rest on it, and nothing stopped one copy from being changed alone.
LSMVTreeCursorAgreementTest runs one query through both cursors and
requires the same visible set, and separately forces a full merge and
queries after it. Against the code before this change, two of its four
cases fail: the cursors disagree, and the merged component carries the
deleted record. Neither the existing collision test nor the existing
merge test could reach this -- the first checks only the top-K cursor,
the second uses no colliding distances.
Ext-ref: MB-73194
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: I13c62501af6abc87b5530260716fc60a09e182b4
---
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeSearchCursor.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeTopKSearchCursor.java
A hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/utils/VTreeMergeKey.java
A hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-vtree-test/src/test/java/org/apache/hyracks/storage/am/lsm/vector/LSMVTreeCursorAgreementTest.java
4 files changed, 557 insertions(+), 91 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/81/21681/1
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeSearchCursor.java
index 023b53b..01fdfaa 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeSearchCursor.java
@@ -19,13 +19,16 @@
package org.apache.hyracks.storage.am.lsm.vector.impls;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
import java.util.PriorityQueue;
import java.util.Set;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.util.CleanupUtils;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.dataflow.common.utils.TupleUtils;
import org.apache.hyracks.storage.am.common.api.ITupleFilter;
import org.apache.hyracks.storage.am.common.tuples.ReferenceFrameTupleReference;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
@@ -34,6 +37,7 @@
import org.apache.hyracks.storage.am.lsm.common.api.ILSMTreeTupleReference;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMIndexSearchCursor;
import org.apache.hyracks.storage.am.lsm.vector.utils.LSMVTreeUtils;
+import org.apache.hyracks.storage.am.lsm.vector.utils.VTreeMergeKey;
import org.apache.hyracks.storage.am.vector.impls.ClusterSearchResult;
import org.apache.hyracks.storage.am.vector.impls.VTree;
import org.apache.hyracks.storage.am.vector.impls.VTree.VTreeAccessor;
@@ -141,6 +145,47 @@
// excluded, since they may legitimately differ between matter/antimatter twins.
private int numPrimaryKeyFields;
+ /**
+ * Keys ({@code <distance, PK>}) whose newest version in the cluster being probed is a delete marker.
+ * Every older version of such a key must be dropped, however far away it lands in the merged sequence.
+ * <p>
+ * A single held element cannot do this. The queue carries one element per component, so the ordering
+ * key only sorts across components — it cannot make a component's own stream PK-ordered inside a
+ * distance tie, and a data page stores same-distance records in load order. A marker and its live twin
+ * can therefore be separated by an unrelated record at the same distance, which released the hold and
+ * let the twin through (LSMVTreeCursorAgreementTest).
+ * <p>
+ * Scoped to the cluster, not to the distance run: probing a new cluster can bring in records whose
+ * distance-to-centroid is *smaller* than the one just emitted, so "the distance changed" does not mean
+ * "no more versions of this key". A key's matter and its marker always share a cluster — a delete
+ * replicates into the same clusters the insert did — so clearing on cluster advance is both sound and
+ * bounded by the number of deleted keys in one cluster.
+ * <p>
+ * Each entry also carries the component the marker came from, because "newest" is not simply
+ * "popped first". Across components a lower cursor index is newer, so a marker cancels matters from
+ * strictly higher indices. Within one component, later in the stream is newer — and a memory component
+ * can hold both a marker and a re-inserted matter for the same key (delete, then insert again: the
+ * delete finds no matter to remove locally and leaves a marker). Suppressing on equal index would drop
+ * that re-insert, which is still live.
+ * <p>
+ * Entries are copies: the tuples they came from point into cursor frames that move as cursors advance.
+ * <p>
+ * <b>Cost.</b> Memory is O(cancelled keys in the cluster being probed) — one copied data tuple per
+ * delete marker, released when {@link #advanceAllComponentsToNextCluster()} clears the list, so the
+ * footprint never accumulates across a scan. The copy is of the whole tuple (as the top-K cursor's
+ * group buffer also does) rather than just {@code <distance, PK>}, which is the simpler thing to keep
+ * correct; narrowing it would cut the per-entry size but not the bound. Lookup is a linear scan per
+ * popped element, so the added comparison work is O(popped elements x cancelled keys in the cluster) —
+ * the same shape the top-K cursor already pays over an equal-distance group.
+ * <p>
+ * Both are bounded by how many deletes one cluster holds, which is the number of deleted records
+ * sharing a centroid, not the size of the index. A cluster whose records have almost all been deleted
+ * is the worst case; if that ever needs bounding, the fix is to switch this list for a hash structure
+ * keyed on the primary key, not to shorten its lifetime — a marker has to outlive every intervening
+ * record in its cluster for the cancellation to be correct at all.
+ */
+ private final List<CancelledKey> suppressedKeys = new ArrayList<>();
+
public LSMVTreeSearchCursor(ILSMIndexOperationContext opCtx) {
this(opCtx, false, false, NoOpIndexCursorStats.INSTANCE);
}
@@ -513,12 +558,21 @@
* @return true if this is a valid output tuple, false if processing should continue
*/
private boolean processTopElement(PriorityQueueElement checkElement) throws HyracksDataException {
- // Check if top element is antimatter
+ // Antimatter: the newest version of this key in this cluster is a delete marker, so record the key
+ // and drop every older version of it as they surface. Recording the key rather than holding the
+ // element is what lets the cancellation survive an unrelated same-distance record arriving first.
if (isDeleted(checkElement) && !returnDeletedTuples) {
- // Antimatter tuple - hold for cancellation check with next tuple
- outputElement = outputPriorityQueue.poll();
- needPushElementIntoQueue = true;
antimatterTuplesDetected++;
+ suppressedKeys.add(
+ new CancelledKey(TupleUtils.copyTuple(checkElement.getTuple()), checkElement.getCursorIndex()));
+ pushIntoQueueAndAdvanceClusterIfNeeded(outputPriorityQueue.poll());
+ return false; // Continue processing
+ }
+
+ // An older version of a key whose newest version was a delete marker.
+ if (isSuppressed(checkElement)) {
+ cancellationsMade++;
+ pushIntoQueueAndAdvanceClusterIfNeeded(outputPriorityQueue.poll());
return false; // Continue processing
}
@@ -535,6 +589,35 @@
}
/**
+ * Whether this element is an older version of a key already cancelled by a delete marker in this
+ * cluster. Linear in the number of deleted keys in the cluster, which is the same shape of scan the
+ * top-K cursor does over an equal-distance group.
+ */
+ private boolean isSuppressed(PriorityQueueElement element) throws HyracksDataException {
+ for (int i = 0; i < suppressedKeys.size(); i++) {
+ CancelledKey cancelled = suppressedKeys.get(i);
+ // Strictly older component only: an equal index means the same component, where a matter that
+ // arrives after the marker is a re-insert and outlives it.
+ if (element.getCursorIndex() > cancelled.cursorIndex
+ && compare(cmp, element.getTuple(), cancelled.key) == 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** A {@code <distance, PK>} cancelled by a delete marker, and the component the marker came from. */
+ private static final class CancelledKey {
+ private final ITupleReference key;
+ private final int cursorIndex;
+
+ private CancelledKey(ITupleReference key, int cursorIndex) {
+ this.key = key;
+ this.cursorIndex = cursorIndex;
+ }
+ }
+
+ /**
* Check if tuple passes the INCLUDE field filter.
* Filter is applied AFTER antimatter reconciliation because:
* 1. Antimatter from newer components comes first in queue (lower cursorIndex)
@@ -568,15 +651,14 @@
*/
private void processWithPendingElement(PriorityQueueElement checkElement) throws HyracksDataException {
if (compare(cmp, outputElement.getTuple(), checkElement.getTuple()) == 0) {
- // Same <distance, PK> as the reference (a held antimatter, or the just-output matter). This
- // head is an older, superseded version of that key. Discard ONLY this head and pull the next
- // tuple from ITS cursor, keeping outputElement as the running reference and letting
- // checkPriorityQueue() keep looping so EVERY older same-key version drains — not just one.
+ // Same <distance, PK> as the just-output matter, so this head is an older, superseded version
+ // of a key already emitted. Discard ONLY this head and pull the next tuple from ITS cursor,
+ // keeping outputElement as the running reference and letting checkPriorityQueue() keep looping
+ // so EVERY older same-key version drains — not just one, which would duplicate the key when
+ // three or more components carry it.
//
- // This mirrors LSMIndexSearchCursor's reconciliation. Draining only a single matter/antimatter
- // pair (and resetting the reference) would leak the 3rd+ version of the same key: e.g.
- // insert -> delete -> re-insert of the same vector across >=3 components would resurface the
- // deleted row and duplicate the key.
+ // Delete markers no longer arrive here: processTopElement() records them in suppressedKeys
+ // instead of holding them, so outputElement is only ever a matter this cursor has emitted.
//
// Safe w.r.t. cluster advancing: the queue holds at most one element per component and the
// reference's own component is held out (not advanced) during the drain, so its
@@ -779,47 +861,15 @@
}
/**
- * The <distance, PK> key shared by antimatter cancellation
+ * The {@code <distance, PK>} key shared by antimatter cancellation
* ({@link #compare(MultiComparator, ITupleReference, ITupleReference)}) and merge ordering
- * ({@link VectorPriorityQueueComparator#compare}). Both must define the exact same key, otherwise a
- * delete marker and its live twin could sort apart and fail to cancel.
- *
- * <p>Tuple format (non-quantized): {@code [distance, centroidId, PKs..., includes...]};
- * quantized: {@code [distance, centroidId, quantized_distance, quantized_embedding, PKs..., includes...]}.
- * The key is exactly {@code <distance (field 0), PK fields>}: secondary fields (centroidId, and for
- * quantized layouts quantized_distance/quantized_embedding) and trailing INCLUDE fields are excluded —
- * they may differ between matter/antimatter twins.
+ * ({@link VectorPriorityQueueComparator#compare}) — and, via {@link VTreeMergeKey}, with
+ * {@code LSMVTreeTopKSearchCursor}. All of them must define the exact same key, otherwise a delete
+ * marker and its live twin could sort apart and fail to cancel; see that class for why.
*/
private int compareKey(MultiComparator cmp, ITupleReference tupleA, ITupleReference tupleB)
throws HyracksDataException {
- // Compare field 0 (distance) using ADM-aware comparator 0
- int result = cmp.getComparators()[0].compare(tupleA.getFieldData(0), tupleA.getFieldStart(0),
- tupleA.getFieldLength(0), tupleB.getFieldData(0), tupleB.getFieldStart(0), tupleB.getFieldLength(0));
-
- if (result != 0) {
- return result;
- }
-
- // Compare the PK fields starting at pkStartField
- int numRemainingFields = Math.min(cmp.getComparators().length - pkStartField, numPrimaryKeyFields);
- for (int i = 0; i < numRemainingFields; i++) {
- int fieldIdx = pkStartField + i;
- int cmpIdx = pkStartField + i;
-
- // Check if field exists in tuple before comparing
- if (fieldIdx >= tupleA.getFieldCount() || fieldIdx >= tupleB.getFieldCount()) {
- break;
- }
-
- result = cmp.getComparators()[cmpIdx].compare(tupleA.getFieldData(fieldIdx), tupleA.getFieldStart(fieldIdx),
- tupleA.getFieldLength(fieldIdx), tupleB.getFieldData(fieldIdx), tupleB.getFieldStart(fieldIdx),
- tupleB.getFieldLength(fieldIdx));
- if (result != 0) {
- return result;
- }
- }
-
- return 0;
+ return VTreeMergeKey.compare(cmp, tupleA, tupleB, pkStartField, numPrimaryKeyFields);
}
@Override
@@ -857,18 +907,7 @@
* This ensures synchronized nprobe checking across all LSM components.
*/
private int getMinClustersProbed() {
- int minProbed = Integer.MAX_VALUE;
-
- for (int i = 0; i < rangeCursors.length; i++) {
- VTreeSearchCursor vcCursor = (VTreeSearchCursor) rangeCursors[i];
- int probed = vcCursor.getClustersProbed();
- if (probed < minProbed) {
- minProbed = probed;
- }
- }
-
- // If no VTreeSearchCursors found, return 0
- return minProbed == Integer.MAX_VALUE ? 0 : minProbed;
+ return VTreeMergeKey.minClustersProbed(rangeCursors);
}
/**
@@ -962,6 +1001,9 @@
* blowing the stack.
*/
private void advanceAllComponentsToNextCluster() throws HyracksDataException {
+ // A new cluster brings a fresh set of keys: cancellations recorded for the previous cluster
+ // cannot apply to it, and a key's marker and matter never span clusters.
+ suppressedKeys.clear();
while (true) {
// Reset exhaustion flags for new cluster
Arrays.fill(clusterExhausted, false);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeTopKSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeTopKSearchCursor.java
index 2ad7b52..c9b442b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeTopKSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/impls/LSMVTreeTopKSearchCursor.java
@@ -39,6 +39,7 @@
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMTreeTupleReference;
import org.apache.hyracks.storage.am.lsm.vector.utils.LSMVTreeUtils;
+import org.apache.hyracks.storage.am.lsm.vector.utils.VTreeMergeKey;
import org.apache.hyracks.storage.am.vector.api.IVTreeBinaryAccessor;
import org.apache.hyracks.storage.am.vector.api.IVTreeBinaryAccessorFactory;
import org.apache.hyracks.storage.am.vector.api.IVTreeDistanceFunction;
@@ -453,6 +454,12 @@
// queue element — cancels the twin wherever it lands in the run, so a deleted record cannot
// resurface (see LSMVTreeAntimatterCollisionQuantizedTest).
//
+ // The comparator's key does include the PK (see VTreeMergeKey), but that only orders the
+ // elements the queue holds — one per component. It cannot make a component's OWN stream
+ // PK-ordered within a distance tie, which is what the separation above comes from. This is
+ // the difference between this cursor and LSMVTreeSearchCursor's held-reference reconciliation;
+ // see LSMVTreeCursorAgreementTest.
+ //
// Each element's tuple points into its component cursor's frame and is invalidated when that
// cursor advances, so capture a stable COPY before pushIntoQueueAndAdvanceClusterIfNeeded.
PriorityQueueElement first = outputPriorityQueue.poll();
@@ -634,16 +641,7 @@
* Get the minimum number of clusters probed across all VTreeSearchCursors.
*/
private int getMinClustersProbed() {
- int minProbed = Integer.MAX_VALUE;
- for (int i = 0; i < rangeCursors.length; i++) {
- if (rangeCursors[i] instanceof VTreeSearchCursor) {
- int probed = ((VTreeSearchCursor) rangeCursors[i]).getClustersProbed();
- if (probed < minProbed) {
- minProbed = probed;
- }
- }
- }
- return minProbed == Integer.MAX_VALUE ? 0 : minProbed;
+ return VTreeMergeKey.minClustersProbed(rangeCursors);
}
/**
@@ -801,29 +799,12 @@
ITupleReference tupleB = b.tuple;
try {
- int result = cmp.getComparators()[0].compare(tupleA.getFieldData(0), tupleA.getFieldStart(0),
- tupleA.getFieldLength(0), tupleB.getFieldData(0), tupleB.getFieldStart(0),
- tupleB.getFieldLength(0));
+ // Same <distance, PK> key as LSMVTreeSearchCursor: both cursors' reconciliation depends on
+ // every version of a record being contiguous in this order.
+ int result = VTreeMergeKey.compare(cmp, tupleA, tupleB, pkStartField, numPrimaryKeyFields);
if (result != 0) {
return result;
}
-
- // Cap at the PK fields; trailing INCLUDE fields are excluded from the ordering key so a
- // delete marker and its live twin (which may differ in INCLUDE values) order together.
- int numRemainingFields = Math.min(cmp.getComparators().length - pkStartField, numPrimaryKeyFields);
- for (int i = 0; i < numRemainingFields; i++) {
- int fieldIdx = pkStartField + i;
- if (fieldIdx >= tupleA.getFieldCount() || fieldIdx >= tupleB.getFieldCount()) {
- break;
- }
- result = cmp.getComparators()[pkStartField + i].compare(tupleA.getFieldData(fieldIdx),
- tupleA.getFieldStart(fieldIdx), tupleA.getFieldLength(fieldIdx),
- tupleB.getFieldData(fieldIdx), tupleB.getFieldStart(fieldIdx),
- tupleB.getFieldLength(fieldIdx));
- if (result != 0) {
- return result;
- }
- }
} catch (Throwable e) {
// Matches the LSMIndexSearchCursor / LSMRTree / LSMBTree comparator idiom: Comparator.compare
// cannot throw a checked exception. The cause is a coded HyracksDataException and
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/utils/VTreeMergeKey.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/utils/VTreeMergeKey.java
new file mode 100644
index 0000000..63e093b8
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/utils/VTreeMergeKey.java
@@ -0,0 +1,106 @@
+/*
+ * 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.utils;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.vector.impls.VTreeSearchCursor;
+import org.apache.hyracks.storage.common.IIndexCursor;
+import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.annotations.AiProvenance;
+
+/**
+ * The {@code <distance, primary key>} ordering key shared by both LSM VTree search cursors, and the one
+ * thing their antimatter reconciliation depends on.
+ * <p>
+ * Each cursor merges one stream per component and must decide, for a record that was inserted and later
+ * deleted, that the delete wins. Including the primary key here is necessary for that: without it two
+ * versions of one record would not even sort together among the elements the queue holds.
+ * <p>
+ * It is <em>not</em> sufficient, and this is the subtle part. The queue holds one element per component,
+ * so this key only orders across components — it cannot make a single component's own stream PK-ordered
+ * within a distance tie, and a data page stores same-distance records in load order. A delete marker and
+ * its live twin can therefore still be separated in the merged sequence by an unrelated record at the
+ * same distance. {@code LSMVTreeTopKSearchCursor} handles that by grouping the whole equal-distance run
+ * and taking the newest entry per key; {@code LSMVTreeSearchCursor}'s held-reference reconciliation does
+ * not, which {@code LSMVTreeCursorAgreementTest} demonstrates.
+ * <p>
+ * This lives in one place because the two cursors previously computed the same key from two independent
+ * copies of the loop, so nothing stopped one from being changed alone.
+ * <p>
+ * Excluded from the key on purpose: {@code centroidId} and, in quantized layouts,
+ * {@code quantized_distance} / {@code quantized_embedding}, plus all trailing INCLUDE fields. A delete
+ * marker and its live twin may legitimately differ in those, and including them would sort the pair apart.
+ */
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Extracted from both LSM VTree cursors; behaviour unchanged")
+public final class VTreeMergeKey {
+
+ private VTreeMergeKey() {
+ }
+
+ /**
+ * Compares two data tuples by {@code <distance (field 0), PK fields>}.
+ *
+ * @param pkStartField first primary-key field, which depends on the quantized layout
+ * @param numPrimaryKeyFields how many PK fields to take, capped by what the comparator array holds
+ */
+ public static int compare(MultiComparator cmp, ITupleReference tupleA, ITupleReference tupleB, int pkStartField,
+ int numPrimaryKeyFields) throws HyracksDataException {
+ int result = cmp.getComparators()[0].compare(tupleA.getFieldData(0), tupleA.getFieldStart(0),
+ tupleA.getFieldLength(0), tupleB.getFieldData(0), tupleB.getFieldStart(0), tupleB.getFieldLength(0));
+ if (result != 0) {
+ return result;
+ }
+
+ int numRemainingFields = Math.min(cmp.getComparators().length - pkStartField, numPrimaryKeyFields);
+ for (int i = 0; i < numRemainingFields; i++) {
+ int fieldIdx = pkStartField + i;
+ // A tuple shorter than the key (a rebuilt tuple on the full-scan merge path) ends the key here
+ // rather than reading past its end.
+ if (fieldIdx >= tupleA.getFieldCount() || fieldIdx >= tupleB.getFieldCount()) {
+ break;
+ }
+ result = cmp.getComparators()[fieldIdx].compare(tupleA.getFieldData(fieldIdx),
+ tupleA.getFieldStart(fieldIdx), tupleA.getFieldLength(fieldIdx), tupleB.getFieldData(fieldIdx),
+ tupleB.getFieldStart(fieldIdx), tupleB.getFieldLength(fieldIdx));
+ if (result != 0) {
+ return result;
+ }
+ }
+ return 0;
+ }
+
+ /**
+ * The fewest clusters any component cursor has probed, which is what both cursors compare against
+ * {@code nprobe} to decide whether to keep probing. Non-VTree cursors are skipped rather than cast,
+ * so a mixed array reports on the ones that can answer instead of throwing.
+ *
+ * @return the minimum, or 0 when no cursor could report one
+ */
+ public static int minClustersProbed(IIndexCursor[] rangeCursors) {
+ int minProbed = Integer.MAX_VALUE;
+ for (IIndexCursor rangeCursor : rangeCursors) {
+ if (rangeCursor instanceof VTreeSearchCursor) {
+ minProbed = Math.min(minProbed, ((VTreeSearchCursor) rangeCursor).getClustersProbed());
+ }
+ }
+ return minProbed == Integer.MAX_VALUE ? 0 : minProbed;
+ }
+}
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 0000000..571674d
--- /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,337 @@
+/*
+ * 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 ({@code VTreeMergeKey}), 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>
+ * This case used to fail: the streaming cursor held one antimatter element and cancelled it only
+ * against the next queue head, so the intervening tie released the hold and the deleted record was
+ * emitted — by the cursor that merges components, i.e. into durable bytes. It now records cancelled
+ * keys for the cluster instead of holding one element.
+ */
+ @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 that an
+ * unrelated same-distance record sits between a delete marker and its twin.
+ */
+ 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[] { 0, 5, 0 },
+ "pk_collide_B"));
+ cluster.add(QuantizedSearchTestDriver.createOptimizedSearchRecordTuple(5.0, 0, new double[] { 5, 0, 0 },
+ "pk_collide_A"));
+ 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;
+ }
+}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21681?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I13c62501af6abc87b5530260716fc60a09e182b4
Gerrit-Change-Number: 21681
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>