You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/05/10 07:57:00 UTC
[incubator-paimon] branch master updated: [core][fix] reset the iterator in loserTree to null after releaseBatch. (#1116)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 98342e2f0 [core][fix] reset the iterator in loserTree to null after releaseBatch. (#1116)
98342e2f0 is described below
commit 98342e2f01d37c82fe69eb23b3c80fd2323a80bc
Author: liming.1018 <69...@users.noreply.github.com>
AuthorDate: Wed May 10 15:56:55 2023 +0800
[core][fix] reset the iterator in loserTree to null after releaseBatch. (#1116)
---
.../apache/paimon/mergetree/compact/LoserTree.java | 2 +
.../paimon/mergetree/compact/LoserTreeTest.java | 97 ++++++++++++++++------
.../paimon/utils/TestReusingRecordReader.java | 27 +++++-
3 files changed, 98 insertions(+), 28 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LoserTree.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LoserTree.java
index b35be96f9..4f9063014 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LoserTree.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LoserTree.java
@@ -309,6 +309,7 @@ public class LoserTree<T> implements Closeable {
while (!endOfInput) {
if (iterator != null) {
iterator.releaseBatch();
+ iterator = null;
}
iterator = reader.readBatch();
@@ -327,6 +328,7 @@ public class LoserTree<T> implements Closeable {
public void close() throws IOException {
if (this.iterator != null) {
this.iterator.releaseBatch();
+ this.iterator = null;
}
this.reader.close();
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LoserTreeTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LoserTreeTest.java
index d1547ed02..5eeb65ab8 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LoserTreeTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LoserTreeTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.mergetree.compact;
import org.apache.paimon.KeyValue;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.testutils.assertj.AssertionUtils;
import org.apache.paimon.utils.ReusingTestData;
import org.apache.paimon.utils.TestReusingRecordReader;
@@ -32,50 +33,92 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
/** Test for {@link LoserTree}. */
public class LoserTreeTest {
private static final Comparator<KeyValue> KEY_COMPARATOR =
- Comparator.comparingInt(o -> o.key().getInt(0));
+ Comparator.<KeyValue>comparingInt(o -> o.key().getInt(0)).reversed();
private static final Comparator<KeyValue> SEQUENCE_COMPARATOR =
- Comparator.comparingLong(KeyValue::sequenceNumber);
+ Comparator.comparingLong(KeyValue::sequenceNumber).reversed();
@RepeatedTest(100)
public void testLoserTreeIsOrdered() throws IOException {
- List<ReusingTestData> reusingTestData = ReusingTestData.generateData(1000, false);
+ List<ReusingTestData> reusingTestData = new ArrayList<>();
+ List<RecordReader<KeyValue>> sortedTestReaders =
+ createSortedTestReaders(reusingTestData, 0, () -> Function::identity);
+ Collections.sort(reusingTestData);
+ try (LoserTree<KeyValue> loserTree =
+ new LoserTree<>(sortedTestReaders, KEY_COMPARATOR, SEQUENCE_COMPARATOR)) {
+ Iterator<ReusingTestData> expectedIterator = reusingTestData.iterator();
+ checkLoserTree(
+ loserTree,
+ kv -> {
+ assertThat(expectedIterator.hasNext());
+ expectedIterator.next().assertEquals(kv);
+ });
+ assertThat(expectedIterator.hasNext()).isFalse();
+ }
+ }
+
+ @RepeatedTest(100)
+ public void testLoserTreeCloseNormally() {
+ List<RecordReader<KeyValue>> sortedTestReaders =
+ createSortedTestReaders(
+ new ArrayList<>(),
+ 1,
+ () ->
+ new TestReusingRecordReader.ThrowingRunnable<IOException>() {
+ private boolean initialized = false;
+
+ @Override
+ public void run() throws IOException {
+ if (initialized) {
+ throw new IOException("ingest exception");
+ }
+ initialized = true;
+ }
+ });
+ LoserTree<KeyValue> loserTree =
+ new LoserTree<>(sortedTestReaders, KEY_COMPARATOR, SEQUENCE_COMPARATOR);
+ assertThatThrownBy(() -> checkLoserTree(loserTree, kv -> {}))
+ .satisfies(AssertionUtils.anyCauseMatches(IOException.class));
+ assertDoesNotThrow(loserTree::close);
+ }
+
+ private List<RecordReader<KeyValue>> createSortedTestReaders(
+ List<ReusingTestData> reusingTestData,
+ int minNumberRecords,
+ Supplier<TestReusingRecordReader.ThrowingRunnable<IOException>> beforeReadBatch) {
List<RecordReader<KeyValue>> sortedTestReaders = new ArrayList<>();
ThreadLocalRandom random = ThreadLocalRandom.current();
int numberReaders = random.nextInt(20) + 1;
- int lowerBound = 0, upperBound = reusingTestData.size();
for (int i = 0; i < numberReaders; i++) {
- int subUpperBound = random.nextInt(lowerBound, upperBound);
- List<ReusingTestData> subReusingTestData =
- reusingTestData.subList(lowerBound, subUpperBound);
- Collections.sort(subReusingTestData);
- sortedTestReaders.add(new TestReusingRecordReader(subReusingTestData));
- lowerBound = subUpperBound;
+ int numberRecords = Math.max(random.nextInt(100), minNumberRecords);
+ List<ReusingTestData> testData =
+ ReusingTestData.generateOrderedNoDuplicatedKeys(numberRecords, false);
+ reusingTestData.addAll(testData);
+ sortedTestReaders.add(new TestReusingRecordReader(testData, beforeReadBatch.get()));
}
- Collections.sort(reusingTestData);
- checkLoserTree(sortedTestReaders, reusingTestData);
+ return sortedTestReaders;
}
- private void checkLoserTree(
- List<RecordReader<KeyValue>> sortedTestReaders, List<ReusingTestData> expectedData)
+ private void checkLoserTree(LoserTree<KeyValue> loserTree, Consumer<KeyValue> consumer)
throws IOException {
- try (LoserTree<KeyValue> loserTree =
- new LoserTree<>(sortedTestReaders, KEY_COMPARATOR, SEQUENCE_COMPARATOR)) {
- Iterator<ReusingTestData> expectedIterator = expectedData.iterator();
- do {
- loserTree.adjustForNextLoop();
- for (KeyValue winner = loserTree.popWinner();
- winner != null;
- winner = loserTree.popWinner()) {
- assertThat(expectedIterator.hasNext());
- expectedIterator.next().assertEquals(winner);
- }
- } while (loserTree.peekWinner() != null && expectedIterator.hasNext());
- }
+ loserTree.initializeIfNeeded();
+ do {
+ for (KeyValue winner = loserTree.popWinner();
+ winner != null;
+ winner = loserTree.popWinner()) {
+ consumer.accept(winner);
+ }
+ loserTree.adjustForNextLoop();
+ } while (loserTree.peekWinner() != null);
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/TestReusingRecordReader.java b/paimon-core/src/test/java/org/apache/paimon/utils/TestReusingRecordReader.java
index dd4fd441e..95b9e6b61 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/TestReusingRecordReader.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/TestReusingRecordReader.java
@@ -42,16 +42,29 @@ public class TestReusingRecordReader implements RecordReader<KeyValue> {
private final List<TestRecordIterator> producedBatches;
private final Random random;
-
+ private final ThrowingRunnable<RuntimeException> beforeReadBatch;
private int nextLowerBound;
private boolean closed;
public TestReusingRecordReader(List<ReusingTestData> testData) {
+ this(testData, () -> {});
+ }
+
+ public TestReusingRecordReader(
+ List<ReusingTestData> testData, ThrowingRunnable<?> beforeReadBatch) {
this.testData = testData;
this.reuse = new ReusingKeyValue();
this.producedBatches = new ArrayList<>();
this.random = new Random();
+ this.beforeReadBatch =
+ () -> {
+ try {
+ beforeReadBatch.run();
+ } catch (Throwable t) {
+ throw new RuntimeException(t);
+ }
+ };
this.nextLowerBound = 0;
this.closed = false;
@@ -61,6 +74,7 @@ public class TestReusingRecordReader implements RecordReader<KeyValue> {
@Override
public RecordIterator<KeyValue> readBatch() {
assertThat(nextLowerBound != -1).isTrue();
+ beforeReadBatch.run();
if (nextLowerBound == testData.size() && random.nextBoolean()) {
nextLowerBound = -1;
return null;
@@ -112,7 +126,18 @@ public class TestReusingRecordReader implements RecordReader<KeyValue> {
@Override
public void releaseBatch() {
+ if (released) {
+ throw new UnsupportedOperationException("Unsupported release twice.");
+ }
this.released = true;
}
}
+
+ /**
+ * Similar to a {@link Runnable}, this interface is used to capture a block of code to be
+ * executed. In contrast to {@code Runnable}, this interface allows throwing checked exceptions.
+ */
+ public interface ThrowingRunnable<E extends Throwable> {
+ void run() throws E;
+ }
}