diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java index 85665c2bb8..47c766d9c3 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java @@ -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; @@ -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<>(); } @@ -69,6 +81,9 @@ public List getIterators() { * Get inner iterator */ private void getInnerIterator() { + if (this.closed) { + return; + } if (this.innerIterator != null && this.innerIterator.hasNext()) { return; } @@ -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; @@ -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 next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + this.prepared = false; return this.innerIterator.next(); } @@ -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; } } } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/node/grpc/query/FilterIteratorContractTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/node/grpc/query/FilterIteratorContractTest.java index 5e7b615aab..a60250a4ee 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/node/grpc/query/FilterIteratorContractTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/node/grpc/query/FilterIteratorContractTest.java @@ -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.next().name[0]); + assertEquals(2, (int) iterator.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.next().name[0]); + assertEquals(2, (int) iterator.next().name[0]); + assertTrue(iterator.hasNext()); + assertEquals(3, (int) iterator.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); @@ -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 rows) { this.rows = rows.iterator(); @@ -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 rows; + private BackendColumn current; + + private AdvancingIterator(List 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 next() { + if (this.current == null) { + throw new NoSuchElementException(); + } + BackendColumn row = this.current; + this.current = null; + return (T) row; + } + + @Override + public void close() { } } }