This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2d4eb99f3c [flink] Only apply postpone merge-on-read to bounded
sources (#10521)
2d4eb99f3c is described below
commit 2d4eb99f3cee8abf16f7fcf583ddc195828f08b1
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 13 08:58:17 2026 +0800
[flink] Only apply postpone merge-on-read to bounded sources (#10521)
---
.../globalindex/GlobalIndexEvaluatorTest.java | 23 ++++++++++++----------
.../paimon/flink/source/FlinkSourceBuilder.java | 6 +-----
.../flink/source/FlinkSourceBuilderTest.java | 16 ++++++++++-----
3 files changed, 25 insertions(+), 20 deletions(-)
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
index 16d5e61584..e3d9942ff2 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
@@ -44,6 +44,7 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -869,25 +870,26 @@ class GlobalIndexEvaluatorTest {
executor = Executors.newFixedThreadPool(4);
RowType rowType = rowType();
- AtomicInteger concurrency = new AtomicInteger(0);
- AtomicInteger maxConcurrency = new AtomicInteger(0);
+ CountDownLatch concurrentReads = new CountDownLatch(3);
GlobalIndexReader concurrencyDetectingReader =
- new StubGlobalIndexReader(resultOf(1, 2, 3, 4, 5)) {
+ new StubGlobalIndexReader(null) {
@Override
public CompletableFuture<Optional<GlobalIndexResult>>
visitEqual(
FieldRef fieldRef, Object literal) {
return CompletableFuture.supplyAsync(
() -> {
- int c = concurrency.incrementAndGet();
- maxConcurrency.updateAndGet(cur ->
Math.max(cur, c));
+ concurrentReads.countDown();
try {
- Thread.sleep(50);
+ assertThat(concurrentReads.await(10,
TimeUnit.SECONDS))
+ .isTrue();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
}
- concurrency.decrementAndGet();
- return Optional.ofNullable(result);
+ // Roaring's cardinality cache is mutable,
so each asynchronous
+ // read must return its own bitmap.
+ return Optional.of(resultOf(1, 2, 3, 4,
5));
},
executor);
}
@@ -902,9 +904,10 @@ class GlobalIndexEvaluatorTest {
Predicate predicate =
PredicateBuilder.and(builder.equal(0, 1), builder.equal(0, 2),
builder.equal(0, 3));
- evaluator.evaluate(predicate);
+ Optional<GlobalIndexResult> result = evaluator.evaluate(predicate);
- assertThat(maxConcurrency.get()).isGreaterThan(1);
+ assertThat(result).isPresent();
+ assertBitmapContainsExactly(result.get().results(), 1L, 2L, 3L, 4L,
5L);
evaluator.close();
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index 4f8a10fd9f..0b73413b87 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -431,11 +431,7 @@ public class FlinkSourceBuilder {
+ " too many snapshots that could pose a risk to
the file system.");
}
- if (PostponeMergeOnRead.configured(table)) {
- if (!sourceBounded) {
- throw new UnsupportedOperationException(
- "Option 'postpone.merge-on-read' is only supported for
batch reads.");
- }
+ if (sourceBounded && PostponeMergeOnRead.configured(table)) {
DataStream<RowData> mergeSource = buildPostponeMergeSource();
if (mergeSource != null) {
return mergeSource;
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
index dd93f1bab9..9e891b12df 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
@@ -38,6 +38,7 @@ import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -304,7 +305,7 @@ public class FlinkSourceBuilderTest {
}
@Test
- public void testPostponeMergeOnReadRejectsContinuousSource() throws
Exception {
+ public void testPostponeMergeOnReadIgnoresContinuousSource() throws
Exception {
Identifier identifier = Identifier.create("default",
"postpone_merge_on_read");
catalog.createTable(
identifier,
@@ -318,10 +319,15 @@ public class FlinkSourceBuilderTest {
Table table = catalog.getTable(identifier);
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
- assertThatThrownBy(
- () -> new
FlinkSourceBuilder(table).env(env).sourceBounded(false).build())
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining("only supported for batch reads");
+ DataStream<RowData> dataStream =
+ new
FlinkSourceBuilder(table).env(env).sourceBounded(false).build();
+
+
assertThat(dataStream.getTransformation()).isInstanceOf(SourceTransformation.class);
+ SourceTransformation<?, ?, ?> transformation =
+ (SourceTransformation<?, ?, ?>) dataStream.getTransformation();
+
assertThat(transformation.getSource()).isInstanceOf(PaimonDataStreamSource.class);
+ assertThat(transformation.getSource().getBoundedness())
+ .isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
}
@Test
