Skip to content

fix(store): harden MultiListIterator close and next #3445

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vaidikgarg-bit wants to merge 2 commits into
base: master
Choose a base branch
from
+226 −6
Open

fix(store): harden MultiListIterator close and next #3445

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d8c2e42
fix(store): harden MultiListIterator close and next
vaidikgarg-bit Oct 31, 2026
4bfa917
fix(store): probe MultiListIterator children once per row
vaidikgarg-bit Nov 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 62 additions & 6 deletions ...store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.hugegraph.rocksdb.access.ScanIterator;
Expand All @@ -44,6 +45,17 @@ public class MultiListIterator implements ScanIterator {
*/
private ScanIterator innerIterator;

/**
* set once close() runs; hasNext()/isValid() stop probing the sources after that
*/
private boolean closed;

/**
* true when hasNext() found an element that next() has not returned yet, so the
* current sub-iterator is not probed again
*/
private boolean prepared;

public MultiListIterator() {
this.iterators = new CopyOnWriteArrayList<>();
}
Expand All @@ -69,6 +81,9 @@ public List<ScanIterator> getIterators() {
* Get inner iterator
*/
private void getInnerIterator() {
if (this.closed) {
return;
}
if (this.innerIterator != null && this.innerIterator.hasNext()) {
return;
}
Expand Down Expand Up @@ -97,14 +112,20 @@ private void getInnerIterator() {

@Override
public boolean hasNext() {
if (this.prepared) {
return true;
}
getInnerIterator();
return this.innerIterator != null;
this.prepared = this.innerIterator != null;
return this.prepared;
}

@Override
public boolean isValid() {
getInnerIterator();
if (this.innerIterator != null) {
if (this.closed) {
return false;
}
if (hasNext()) {
return this.innerIterator.isValid();
}
return true;
Expand All @@ -115,21 +136,55 @@ public boolean isValid() {
*/
@Override
public void close() {
if (this.innerIterator != null) {
this.innerIterator.close();
if (this.closed) {
return;
}
this.closed = true;
this.prepared = false;
ScanIterator current = this.innerIterator;
this.innerIterator = null;
if (this.innerListIterator == null) {
// never iterated, so every sub-iterator is still open
this.innerListIterator = this.iterators.iterator();
}

// keep closing the rest if one fails, then rethrow the first failure
Throwable failure = null;
if (current != null) {
failure = closeSource(current, failure);
}
while (this.innerListIterator.hasNext()) {
this.innerListIterator.next().close();
failure = closeSource(this.innerListIterator.next(), failure);
}
this.iterators.clear();
if (failure instanceof RuntimeException runtimeFailure) {
throw runtimeFailure;
}
if (failure != null) {
throw (Error) failure;
}
}

private static Throwable closeSource(ScanIterator iterator, Throwable failure) {
try {
iterator.close();
} catch (RuntimeException | Error e) {
if (failure == null) {
return e;
}
if (failure != e) {
failure.addSuppressed(e);
}
}
return failure;
}

@Override
public <T> T next() {
if (!hasNext()) {

@imbajin imbajin Nov 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

‼️ next() probes the current child again after the caller has already called hasNext(). In multi-branch scanIndex with DEDUP, indexIntersection supplies InAccurateUnionFilterIterator or InAccurateIntersectionIterator children, whose hasNext() consumes and replaces the current match. The second probe skips the first row; a single-row child can instead be exhausted and cause NoSuchElementException. Please cache the prepared state between hasNext() and next(), and add a regression test using a child whose hasNext() advances.

@vaidikgarg-bit vaidikgarg-bit Nov 1, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, done. hasNext() now records that it found an element, and next() and repeated hasNext()/isValid() calls use that record instead of probing the child again. testMultiListProbesAdvancingSourceOncePerRow uses a child whose hasNext() advances and replaces the pending row. It fails on the previous head, and on master too, since master's hasNext() probes the child again on every call. It passes now. This also removes the extra per-row probe I noted in the description.

On impact: the extra probe in next() came from my previous commit. Since #3424 the two InAccurate* iterators keep their pending element (if (current != null) return true;), and the other children I checked in BusinessHandlerImpl are also safe to probe twice, so I don't think it dropped rows with the current children. It did add a probe per row and relied on that idempotence, and now it doesn't.

throw new NoSuchElementException();
}
this.prepared = false;
return this.innerIterator.next();
}

Expand All @@ -154,6 +209,7 @@ public void seek(byte[] position) {
// range scan or prefix scan
if (this.innerIterator instanceof MultiPartitionIterator) {
this.innerIterator.seek(position);
this.prepared = false;
}
}
}
Expand Down
164 changes: 164 additions & 0 deletions .../src/main/java/org/apache/hugegraph/store/node/grpc/query/FilterIteratorContractTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,126 @@ public void testMultiListClosesEachSourceOnceAfterPartialIteration() {
assertEquals(1, third.closes);
}

@Test
public void testMultiListCloseFailureStillClosesRemainingSources() {
RowIterator first = singleRow(1);
RowIterator second = singleRow(2);
RowIterator third = singleRow(3);
IllegalStateException firstFailure = new IllegalStateException("first close failed");
IllegalStateException thirdFailure = new IllegalStateException("third close failed");
first.closeFailure = firstFailure;
third.closeFailure = thirdFailure;
MultiListIterator iterator = new MultiListIterator(List.of(first, second, third));

assertSame(firstFailure, assertThrows(IllegalStateException.class, iterator::close));

assertEquals(1, first.closes);
assertEquals(1, second.closes);
assertEquals(1, third.closes);
assertEquals(List.of(thirdFailure), Arrays.asList(firstFailure.getSuppressed()));
assertTrue(iterator.getIterators().isEmpty());
assertFalse(iterator.hasNext());
}

@Test
public void testMultiListCloseFailureDuringIterationStillClosesRemainingSources() {
RowIterator first = singleRow(1);
RowIterator second = singleRow(2);
IllegalStateException firstFailure = new IllegalStateException("first close failed");
IllegalStateException secondFailure = new IllegalStateException("second close failed");
first.closeFailure = firstFailure;
second.closeFailure = secondFailure;
MultiListIterator iterator = new MultiListIterator(List.of(first, second));
assertTrue(iterator.hasNext());

assertSame(firstFailure, assertThrows(IllegalStateException.class, iterator::close));

assertEquals(1, first.closes);
assertEquals(1, second.closes);
assertEquals(List.of(secondFailure), Arrays.asList(firstFailure.getSuppressed()));
assertTrue(iterator.getIterators().isEmpty());
}

@Test
public void testMultiListCloseErrorStillClosesRemainingSources() {
RowIterator first = singleRow(1);
RowIterator second = singleRow(2);
RowIterator third = singleRow(3);
LinkageError firstError = new LinkageError("first close failed");
IllegalStateException secondFailure = new IllegalStateException("second close failed");
first.closeError = firstError;
second.closeFailure = secondFailure;
MultiListIterator iterator = new MultiListIterator(List.of(first, second, third));

assertSame(firstError, assertThrows(LinkageError.class, iterator::close));

assertEquals(1, first.closes);
assertEquals(1, second.closes);
assertEquals(1, third.closes);
assertEquals(List.of(secondFailure), Arrays.asList(firstError.getSuppressed()));
assertTrue(iterator.getIterators().isEmpty());

iterator.close();
assertEquals(1, first.closes);
assertEquals(1, second.closes);
assertEquals(1, third.closes);
}

@Test
public void testMultiListNextWithoutHasNext() {
MultiListIterator iterator = new MultiListIterator(List.of(singleRow(1), singleRow(2)));

assertEquals(1, (int) iterator.<BackendColumn>next().name[0]);
assertEquals(2, (int) iterator.<BackendColumn>next().name[0]);
assertThrows(NoSuchElementException.class, iterator::next);

assertThrows(NoSuchElementException.class, new MultiListIterator()::next);
}

@Test
public void testMultiListProbesAdvancingSourceOncePerRow() {
MultiListIterator iterator = new MultiListIterator(List.of(
new AdvancingIterator(List.of(BackendColumn.of(new byte[]{1}, new byte[]{11}),
BackendColumn.of(new byte[]{2}, new byte[]{12}))),
new AdvancingIterator(List.of(BackendColumn.of(new byte[]{3}, new byte[]{13})))));

assertTrue(iterator.hasNext());
assertTrue(iterator.hasNext());
assertTrue(iterator.isValid());
assertEquals(1, (int) iterator.<BackendColumn>next().name[0]);
assertEquals(2, (int) iterator.<BackendColumn>next().name[0]);
assertTrue(iterator.hasNext());
assertEquals(3, (int) iterator.<BackendColumn>next().name[0]);
assertFalse(iterator.hasNext());
assertTrue(iterator.isValid());
assertThrows(NoSuchElementException.class, iterator::next);
}

@Test
public void testMultiListCloseDuringIterationStopsProbingSources() {
RowIterator first = singleRow(1);
RowIterator second = singleRow(2);
MultiListIterator iterator = new MultiListIterator(List.of(first, second));
assertTrue(iterator.hasNext());
int firstProbes = first.probes;
int secondProbes = second.probes;

iterator.close();

assertFalse(iterator.hasNext());
assertFalse(iterator.isValid());
assertEquals(firstProbes, first.probes);
assertEquals(secondProbes, second.probes);
assertEquals(1, first.closes);
assertEquals(1, second.closes);

assertThrows(NoSuchElementException.class, iterator::next);
iterator.close();
assertEquals(firstProbes, first.probes);
assertEquals(1, first.closes);
assertEquals(1, second.closes);
}

@Test(timeout = 5000)
public void testUnionUniqueStreamDoesNotLoseRecordsBetweenSingleRecordBatches() throws Exception {
assertSingleRecordBatches(FilterKind.UNION_UNIQUE);
Expand Down Expand Up @@ -375,6 +495,7 @@ private static class RowIterator implements ScanIterator {
private int closes;
private int probes;
private RuntimeException closeFailure;
private Error closeError;

private RowIterator(List<BackendColumn> rows) {
this.rows = rows.iterator();
Expand Down Expand Up @@ -407,6 +528,49 @@ public void close() {
if (this.closeFailure != null) {
throw this.closeFailure;
}
if (this.closeError != null) {
throw this.closeError;
}
}
}

/**
* models the dedup iterators before #3424: hasNext() pulls the next row and replaces
* the pending one, so a second hasNext() before next() skips a row
*/
private static class AdvancingIterator implements ScanIterator {

private final Iterator<BackendColumn> rows;
private BackendColumn current;

private AdvancingIterator(List<BackendColumn> rows) {
this.rows = rows.iterator();
}

@Override
public boolean hasNext() {
this.current = this.rows.hasNext() ? this.rows.next() : null;
return this.current != null;
}

@Override
public boolean isValid() {
return this.current != null;
}

@SuppressWarnings("unchecked")
@Override
public <T> T next() {
if (this.current == null) {
throw new NoSuchElementException();
}
BackendColumn row = this.current;
this.current = null;
return (T) row;
}

@Override
public void close() {
}
}
}
Loading