working on new ID based join iterator

Author: hmottestad

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/IdAccessor.java

@@ -13,6 +13,8 @@
 import java.util.Set;
 
 import org.eclipse.rdf4j.common.annotation.InternalUseOnly;
+import org.eclipse.rdf4j.query.MutableBindingSet;
+import org.eclipse.rdf4j.query.QueryEvaluationException;
 
 /**
  * Read-only accessor for variable ID lookups against a record long[] produced by a RecordIterator.
@@ -28,4 +30,7 @@ public interface IdAccessor {
 	 * should return {@code -1} when the variable is not part of the record.
 	 */
 	int getRecordIndex(String varName);
+
+	boolean applyRecord(long[] record, MutableBindingSet target, ValueStore valueStore)
+			throws QueryEvaluationException;
 }

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/IdBindingInfo.java

@@ -101,6 +101,7 @@ public long getId(long[] record, String varName) {
 		return record[i];
 	}
 
+	@Override
 	public boolean applyRecord(long[] record, MutableBindingSet target, ValueStore valueStore)
 			throws QueryEvaluationException {
 		// Fast path: pre-resolved getters/setters from the evaluation context

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/join/LmdbIdJoinIterator.java

@@ -131,7 +131,8 @@ public long getId(long[] record, String varName) {
 			return record[indices[0]];
 		}
 
-		boolean applyRecord(long[] record, MutableBindingSet target, ValueStore valueStore)
+		@Override
+		public boolean applyRecord(long[] record, MutableBindingSet target, ValueStore valueStore)
 				throws QueryEvaluationException {
 			for (Map.Entry<String, int[]> entry : indexByVar.entrySet()) {
 				String name = entry.getKey();
@@ -220,8 +221,8 @@ private Value resolveValue(long id, int position, ValueStore valueStore) throws
 
 	private final RecordIterator leftIterator;
 	private final RecordIteratorFactory rightFactory;
-	private final PatternInfo leftInfo;
-	private final PatternInfo rightInfo;
+	private final IdAccessor leftInfo;
+	private final IdAccessor rightInfo;
 	private final Set<String> sharedVariables;
 	private final QueryEvaluationContext context;
 	private final BindingSet initialBindings;
@@ -231,8 +232,8 @@ private Value resolveValue(long id, int position, ValueStore valueStore) throws
 	private long[] currentLeftRecord;
 	private BindingSet currentLeftBinding;
 
-	LmdbIdJoinIterator(RecordIterator leftIterator, RecordIteratorFactory rightFactory, PatternInfo leftInfo,
-			PatternInfo rightInfo, Set<String> sharedVariables, QueryEvaluationContext context,
+	LmdbIdJoinIterator(RecordIterator leftIterator, RecordIteratorFactory rightFactory, IdAccessor leftInfo,
+			IdAccessor rightInfo, Set<String> sharedVariables, QueryEvaluationContext context,
 			BindingSet initialBindings, ValueStore valueStore) {
 		this.leftIterator = leftIterator;
 		this.rightFactory = rightFactory;

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/join/LmdbIdJoinQueryEvaluationStep.java

@@ -210,8 +210,8 @@ public CloseableIteration<BindingSet> evaluate(BindingSet bindings) {
 					return dataset.getRecordIterator(snapshot, subjIdx, predIdx, objIdx, ctxIdx, patternIds);
 				};
 
-				return new LmdbIdJoinIterator(leftIterator, rightFactory, leftInfo, rightInfo, sharedVariables, context,
-						bindings, valueStore);
+				return new LmdbIdJoinIterator(leftIterator, rightFactory, leftInfo, bindingInfo, sharedVariables,
+						context, bindings, valueStore);
 			}
 
 			// Default: materialize right side via BindingSet and use LmdbIdJoinIterator

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/join/LmdbIdMergeJoinIterator.java

@@ -33,7 +33,6 @@ public class LmdbIdMergeJoinIterator implements RecordIterator {
 	private final String mergeVariable;
 	private final IdBindingInfo bindingInfo;
 	private final Set<String> sharedVariables;
-	private final int mergeIndex;
 	private final int bindingSize;
 	private static final int COLUMN_COUNT = TripleStore.CONTEXT_IDX + 1;
 
@@ -48,6 +47,8 @@ public class LmdbIdMergeJoinIterator implements RecordIterator {
 	private boolean closed;
 	private final int[] leftColumnBindingIndex;
 	private final int[] rightColumnBindingIndex;
+	private final int leftMergeColumn;
+	private final int rightMergeColumn;
 	private static final boolean DEBUG = Boolean.getBoolean("rdf4j.lmdb.mergeJoinDebug");
 	private final AtomicInteger debugCounter = DEBUG ? new AtomicInteger() : null;
 
@@ -65,6 +66,8 @@ public LmdbIdMergeJoinIterator(RecordIterator leftIterator, RecordIterator right
 					+ " must be present in both join operands for LMDB merge join");
 		}
 
+		this.leftMergeColumn = leftInfo.getRecordIndex(mergeVariable);
+		this.rightMergeColumn = rightInfo.getRecordIndex(mergeVariable);
 		this.leftIterator = new PeekMarkRecordIterator(leftIterator);
 		this.rightIterator = new PeekMarkRecordIterator(rightIterator);
 		this.mergeVariable = mergeVariable;
@@ -75,7 +78,6 @@ public LmdbIdMergeJoinIterator(RecordIterator leftIterator, RecordIterator right
 			throw new IllegalArgumentException(
 					"Merge variable " + mergeVariable + " is not tracked in the binding info for LMDB merge join");
 		}
-		this.mergeIndex = idx;
 		this.bindingSize = bindingInfo.size();
 
 		Set<String> shared = new HashSet<>(leftInfo.getVariableNames());
@@ -131,7 +133,7 @@ private long[] computeNextElement() throws QueryEvaluationException {
 				return null;
 			}
 
-			int compare = compare(currentLeftKey, key(rightPeek));
+			int compare = compare(currentLeftKey, key(rightPeek, rightMergeColumn));
 			if (compare == 0) {
 				long[] result = equal();
 				if (result != null) {
@@ -173,7 +175,7 @@ private boolean advanceLeft() {
 						+ Arrays.toString(candidate));
 			}
 			currentLeftRecord = candidate;
-			currentLeftKey = key(candidate);
+			currentLeftKey = key(candidate, leftMergeColumn);
 			hasCurrentLeftKey = true;
 			return true;
 		}
@@ -223,7 +225,7 @@ private void doLeftPeek() {
 		if (leftPeekRecord == null) {
 			leftPeekRecord = leftIterator.peek();
 			if (leftPeekRecord != null) {
-				leftPeekKey = key(leftPeekRecord);
+				leftPeekKey = key(leftPeekRecord, leftMergeColumn);
 				hasLeftPeekKey = true;
 			} else {
 				hasLeftPeekKey = false;
@@ -257,7 +259,8 @@ private long[] joinWithCurrentLeft(long[] rightRecord) {
 		return combined;
 	}
 
-	private static int[] computeColumnBindingMap(LmdbIdJoinIterator.PatternInfo patternInfo, IdBindingInfo bindingInfo) {
+	private static int[] computeColumnBindingMap(LmdbIdJoinIterator.PatternInfo patternInfo,
+			IdBindingInfo bindingInfo) {
 		int[] map = new int[COLUMN_COUNT];
 		Arrays.fill(map, -1);
 		for (String var : patternInfo.getVariableNames()) {
@@ -304,8 +307,8 @@ private int compare(long left, long right) {
 		return Long.compare(left, right);
 	}
 
-	private long key(long[] record) {
-		long id = valueAt(record, mergeIndex);
+	private long key(long[] record, int columnIndex) {
+		long id = record[columnIndex];
 		if (id == LmdbValue.UNKNOWN_ID) {
 			throw new QueryEvaluationException(
 					"Merge variable " + mergeVariable + " is unbound in the current record; cannot perform merge join");
@@ -331,10 +334,4 @@ public void close() {
 		}
 	}
 
-	private static long valueAt(long[] record, int index) {
-		if (index < 0 || index >= record.length) {
-			return LmdbValue.UNKNOWN_ID;
-		}
-		return record[index];
-	}
 }

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/join/LmdbIdOrderVerifier.java

@@ -0,0 +1,64 @@
+/*******************************************************************************
+ * Copyright (c) 2025 Eclipse RDF4J contributors.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Distribution License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ ******************************************************************************/
+package org.eclipse.rdf4j.sail.lmdb.join;
+
+import java.util.Arrays;
+
+import org.eclipse.rdf4j.sail.lmdb.RecordIterator;
+
+public final class LmdbIdOrderVerifier {
+
+	private LmdbIdOrderVerifier() {
+	}
+
+	public static RecordIterator wrap(String label, int mergeIndex, RecordIterator delegate) {
+		if (label == null || label.isEmpty()) {
+			throw new IllegalArgumentException("Label must be provided");
+		}
+		if (mergeIndex < 0) {
+			throw new IllegalArgumentException("Merge index must be non-negative");
+		}
+		return new RecordIterator() {
+			private long position = 0;
+			private long previous = Long.MIN_VALUE;
+			private boolean first = true;
+
+			@Override
+			public long[] next() {
+				long[] record = delegate.next();
+				if (record == null) {
+					return null;
+				}
+				if (mergeIndex >= record.length) {
+					throw new AssertionError(String.format(
+							"%s iterator produced record #%d shorter than merge index %d: %s",
+							label, position, mergeIndex, Arrays.toString(record)));
+				}
+				long current = record[mergeIndex];
+				if (!first && current < previous) {
+					throw new AssertionError(String.format(
+							"%s iterator out of order at position %d: previous=%d current=%d (record=%s)",
+							label, position, previous, current, Arrays.toString(record)));
+				}
+				first = false;
+				previous = current;
+				long[] copy = Arrays.copyOf(record, record.length);
+				position++;
+				return copy;
+			}
+
+			@Override
+			public void close() {
+				delegate.close();
+			}
+		};
+	}
+}

core/sail/lmdb/src/test/java/org/eclipse/rdf4j/sail/lmdb/LmdbIdJoinEvaluationTest.java

@@ -26,6 +26,8 @@
 import org.eclipse.rdf4j.model.IRI;
 import org.eclipse.rdf4j.model.ValueFactory;
 import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
+import org.eclipse.rdf4j.model.vocabulary.RDF;
+import org.eclipse.rdf4j.model.vocabulary.RDFS;
 import org.eclipse.rdf4j.query.BindingSet;
 import org.eclipse.rdf4j.query.QueryEvaluationException;
 import org.eclipse.rdf4j.query.QueryLanguage;
@@ -42,6 +44,7 @@
 import org.eclipse.rdf4j.query.algebra.evaluation.impl.QueryEvaluationContext;
 import org.eclipse.rdf4j.query.explanation.Explanation;
 import org.eclipse.rdf4j.query.impl.EmptyBindingSet;
+import org.eclipse.rdf4j.query.impl.MapBindingSet;
 import org.eclipse.rdf4j.query.parser.ParsedTupleQuery;
 import org.eclipse.rdf4j.query.parser.QueryParserUtil;
 import org.eclipse.rdf4j.repository.RepositoryConnection;
@@ -247,6 +250,100 @@ public void mergeJoinUsesArrayDatasetApi(@TempDir Path tempDir) throws Exception
 		}
 	}
 
+	@Test
+	public void mergeJoinRespectsQueryBindings(@TempDir Path tempDir) throws Exception {
+		LmdbStore store = new LmdbStore(tempDir.toFile());
+		SailRepository repository = new SailRepository(store);
+		repository.init();
+
+		ValueFactory vf = SimpleValueFactory.getInstance();
+		IRI painter = vf.createIRI(NS, "Painter");
+		IRI painting = vf.createIRI(NS, "Painting");
+		IRI paints = vf.createIRI(NS, "paints");
+		IRI picasso = vf.createIRI(NS, "picasso");
+		IRI guernica = vf.createIRI(NS, "guernica");
+
+		try (RepositoryConnection conn = repository.getConnection()) {
+			conn.add(painter, RDF.TYPE, RDFS.CLASS);
+			conn.add(painting, RDF.TYPE, RDFS.CLASS);
+			conn.add(picasso, RDF.TYPE, painter);
+			conn.add(guernica, RDF.TYPE, painting);
+			conn.add(picasso, paints, guernica);
+		}
+
+		String query = "SELECT ?X WHERE {\n" +
+				"  ?X a ?Y .\n" +
+				"  ?Y a <http://www.w3.org/2000/01/rdf-schema#Class> .\n" +
+				"}";
+
+		ParsedTupleQuery parsed = QueryParserUtil.parseTupleQuery(QueryLanguage.SPARQL, query, null);
+		TupleExpr tupleExpr = parsed.getTupleExpr();
+		TupleExpr joinExpr = unwrap(tupleExpr);
+		assertThat(joinExpr).isInstanceOf(Join.class);
+		Join join = (Join) joinExpr;
+
+		StatementPattern left = (StatementPattern) join.getLeftArg();
+		StatementPattern right = (StatementPattern) join.getRightArg();
+		Var joinVar = left.getObjectVar();
+		join.setOrder(joinVar);
+		left.setOrder(joinVar);
+		right.setOrder(right.getSubjectVar());
+		join.setMergeJoin(true);
+
+		SailSource branch = store.getBackingStore().getExplicitSailSource();
+		SailDataset dataset = branch.dataset(IsolationLevels.SNAPSHOT_READ);
+
+		try {
+			SailDatasetTripleSource tripleSource = new SailDatasetTripleSource(repository.getValueFactory(), dataset);
+			EvaluationStrategyFactory factory = store.getEvaluationStrategyFactory();
+			EvaluationStrategy strategy = factory.createEvaluationStrategy(null, tripleSource,
+					store.getBackingStore().getEvaluationStatistics());
+			LmdbEvaluationDataset lmdbDataset = (LmdbEvaluationDataset) dataset;
+			RecordingDataset recordingDataset = new RecordingDataset(lmdbDataset);
+			QueryEvaluationContext context = new LmdbQueryEvaluationContext(null,
+					tripleSource.getValueFactory(), tripleSource.getComparator(), recordingDataset,
+					lmdbDataset.getValueStore());
+
+			QueryEvaluationStep step = strategy.precompile(join, context);
+
+			MapBindingSet bindings = new MapBindingSet(2);
+
+			try (CloseableIteration<BindingSet> iter = step.evaluate(bindings)) {
+				List<? extends BindingSet> results = Iterations.asList(iter);
+				assertThat(results).hasSize(2);
+				assertThat(results).extracting(bs -> bs.getValue("X")).containsExactlyInAnyOrder(picasso, guernica);
+			}
+
+			bindings.addBinding("Y", painter);
+			try (CloseableIteration<BindingSet> iter = step.evaluate(bindings)) {
+				List<? extends BindingSet> results = Iterations.asList(iter);
+				assertThat(results).hasSize(1);
+				assertThat(results.get(0).getValue("X")).isEqualTo(picasso);
+			}
+
+			bindings.addBinding("Z", painting);
+			try (CloseableIteration<BindingSet> iter = step.evaluate(bindings)) {
+				List<? extends BindingSet> results = Iterations.asList(iter);
+				assertThat(results).hasSize(1);
+				assertThat(results.get(0).getValue("X")).isEqualTo(picasso);
+			}
+
+			bindings.removeBinding("Y");
+			try (CloseableIteration<BindingSet> iter = step.evaluate(bindings)) {
+				List<? extends BindingSet> results = Iterations.asList(iter);
+				assertThat(results).hasSize(2);
+				assertThat(results).extracting(bs -> bs.getValue("X")).containsExactlyInAnyOrder(picasso, guernica);
+			}
+
+			assertThat(recordingDataset.wasArrayOrderedApiUsed()).isTrue();
+			assertThat(join.getAlgorithmName()).isEqualTo("LmdbIdMergeJoinIterator");
+		} finally {
+			dataset.close();
+			branch.close();
+			repository.shutDown();
+		}
+	}
+
 	@Test
 	public void mergeJoinFallsBackWhenOrderUnsupported(@TempDir Path tempDir) throws Exception {
 		LmdbStore store = new LmdbStore(tempDir.toFile());

core/sail/lmdb/src/test/java/org/eclipse/rdf4j/sail/lmdb/join/LmdbIdMergeJoinIteratorTest.java

@@ -11,6 +11,7 @@
 package org.eclipse.rdf4j.sail.lmdb.join;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.Arrays;
 
@@ -91,6 +92,23 @@ void nextReturnsNullWhenNoMatches() throws Exception {
 		}
 	}
 
+	@Test
+	void orderVerifierDetectsDisorder() {
+		RecordIterator unordered = new ArrayRecordIterator(
+				new long[] { 1L, LmdbValue.UNKNOWN_ID, LmdbValue.UNKNOWN_ID, 0L },
+				new long[] { 0L, LmdbValue.UNKNOWN_ID, LmdbValue.UNKNOWN_ID, 0L });
+
+		RecordIterator verifying = LmdbIdOrderVerifier.wrap("left", 0, unordered);
+
+		assertThatThrownBy(() -> {
+			while (verifying.next() != null) {
+				// consume iterator
+			}
+		})
+				.isInstanceOf(AssertionError.class)
+				.hasMessageContaining("left");
+	}
+
 	private static final class ArrayRecordIterator implements RecordIterator {
 		private final long[][] data;
 		private int index;