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 ca3c7d3b0b [core][flink] Support long limits in the read path (#10028)
ca3c7d3b0b is described below
commit ca3c7d3b0b07a8ded664349e0f3012fe6b1eaa12
Author: Akash Reddy Jammula <[email protected]>
AuthorDate: Sun Sep 20 19:38:51 2026 -0700
[core][flink] Support long limits in the read path (#10028)
---
.../apache/paimon/reader/LimitRecordReader.java | 2 +-
.../paimon/reader/LimitRecordReaderTest.java | 20 +++++-
.../paimon/globalindex/DataEvolutionBatchScan.java | 4 +-
.../apache/paimon/operation/RawFileSplitRead.java | 11 ++-
.../org/apache/paimon/operation/SplitRead.java | 6 +-
.../paimon/table/FallbackReadFileStoreTable.java | 2 +-
.../paimon/table/format/FormatReadBuilder.java | 19 ++++-
.../paimon/table/format/FormatTableRead.java | 4 +-
.../paimon/table/format/FormatTableScan.java | 6 +-
.../table/source/AbstractBatchTableScan.java | 9 +--
.../paimon/table/source/AppendTableRead.java | 4 +-
.../table/source/BlobViewTableReadSupport.java | 2 +-
.../apache/paimon/table/source/InnerTableRead.java | 2 +-
.../apache/paimon/table/source/InnerTableScan.java | 2 +-
.../paimon/table/source/KeyValueTableRead.java | 4 +-
.../apache/paimon/table/source/ReadBuilder.java | 13 ++++
.../paimon/table/source/ReadBuilderImpl.java | 24 ++++++-
.../table/source/snapshot/SnapshotReader.java | 2 +-
.../table/source/snapshot/SnapshotReaderImpl.java | 2 +-
.../apache/paimon/table/system/AuditLogTable.java | 4 +-
.../paimon/table/format/FormatReadBuilderTest.java | 7 ++
.../paimon/table/source/ReadBuilderImplTest.java | 83 ++++++++++++++++++++++
.../paimon/flink/source/FlinkSourceBuilder.java | 2 +-
.../flink/source/FlinkSourceBuilderTest.java | 26 +++++++
24 files changed, 224 insertions(+), 36 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
b/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
index 2f0766e29a..836ae12b1e 100644
---
a/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
@@ -40,7 +40,7 @@ public final class LimitRecordReader<T> implements
RecordReader<T> {
this.limit = limit;
}
- public static <T> RecordReader<T> limit(RecordReader<T> reader, @Nullable
Integer limit) {
+ public static <T> RecordReader<T> limit(RecordReader<T> reader, @Nullable
Long limit) {
if (limit == null || limit <= 0) {
return reader;
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/reader/LimitRecordReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/reader/LimitRecordReaderTest.java
index 6e5224d98d..53571c62cb 100644
---
a/paimon-common/src/test/java/org/apache/paimon/reader/LimitRecordReaderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/reader/LimitRecordReaderTest.java
@@ -25,10 +25,28 @@ import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/** Tests for {@link LimitRecordReader}. */
public class LimitRecordReaderTest {
+ @Test
+ public void testLimitAboveIntegerRangeDoesNotNarrow() throws Exception {
+ RecordReader<Integer> input = mock(RecordReader.class);
+ RecordReader.RecordIterator<Integer> inputBatch =
mock(RecordReader.RecordIterator.class);
+ when(input.readBatch()).thenReturn(inputBatch, null);
+ when(inputBatch.next()).thenReturn(0, 1, 2, null);
+
+ try (RecordReader<Integer> reader = LimitRecordReader.limit(input,
4294967297L)) {
+ RecordReader.RecordIterator<Integer> batch = reader.readBatch();
+ assertThat(batch.next()).isEqualTo(0);
+ assertThat(batch.next()).isEqualTo(1);
+ assertThat(batch.next()).isEqualTo(2);
+ assertThat(batch.next()).isNull();
+ }
+ }
+
@Test
public void testPreservesFileRecordIterator() throws Exception {
FileRecordIterator<Integer> fileIterator =
@@ -74,7 +92,7 @@ public class LimitRecordReaderTest {
public void close() {}
};
- try (RecordReader<Integer> reader =
LimitRecordReader.limit(fileReader, 2)) {
+ try (RecordReader<Integer> reader =
LimitRecordReader.limit(fileReader, 2L)) {
RecordReader.RecordIterator<Integer> batch = reader.readBatch();
assertThat(batch).isInstanceOf(FileRecordIterator.class);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
index 1a408c87fa..ed3bf2f623 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
@@ -70,7 +70,7 @@ public class DataEvolutionBatchScan implements DataTableScan {
private Predicate filter;
private TopN topN;
- private Integer pushDownLimit;
+ private Long pushDownLimit;
// set when part of the filter reaches the reader only, so limit/TopN must
not prune ahead of it
private boolean rowIdFilterDeferred;
private RowRangeIndex pushedRowRangeIndex;
@@ -184,7 +184,7 @@ public class DataEvolutionBatchScan implements
DataTableScan {
}
@Override
- public InnerTableScan withLimit(int limit) {
+ public InnerTableScan withLimit(long limit) {
// forwarded in plan(), once withFilter has said whether a row-id part
was deferred
this.pushDownLimit = limit;
return this;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
index 0c79a11db2..84b32eaa1d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
@@ -92,7 +92,7 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
private RowType readRowType;
@Nullable private List<Predicate> filters;
@Nullable private TopN topN;
- @Nullable private Integer limit;
+ @Nullable private Long limit;
@Nullable private ReadBatchSizer readBatchSizer;
public RawFileSplitRead(
@@ -151,7 +151,7 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
}
@Override
- public SplitRead<InternalRow> withLimit(@Nullable Integer limit) {
+ public SplitRead<InternalRow> withLimit(@Nullable Long limit) {
this.limit = limit;
return this;
}
@@ -210,7 +210,7 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
RowType outputRowType = readRowType;
Builder formatReaderMappingBuilder =
- createFormatReaderMappingBuilder(outputRowType, topN, limit);
+ createFormatReaderMappingBuilder(outputRowType, topN,
intLimit(limit));
for (DataFileMeta file : files) {
suppliers.add(
@@ -277,6 +277,11 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
nestedFieldEnabled);
}
+ @Nullable
+ private static Integer intLimit(@Nullable Long limit) {
+ return limit != null && limit <= Integer.MAX_VALUE ? limit.intValue()
: null;
+ }
+
private ReaderSupplier<InternalRow> createFileReader(
BinaryRow partition,
DataFilePathFactory dataFilePathFactory,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
index 6a1ed5fc00..22fa440169 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
@@ -50,10 +50,14 @@ public interface SplitRead<T> {
return this;
}
- default SplitRead<T> withLimit(@Nullable Integer limit) {
+ default SplitRead<T> withLimit(@Nullable Long limit) {
return this;
}
+ default SplitRead<T> withLimit(int limit) {
+ return withLimit(Long.valueOf(limit));
+ }
+
default SplitRead<T> withReadBatchSizer(ReadBatchSizer sizer) {
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
index cbdf01fd42..2863320050 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
@@ -468,7 +468,7 @@ public class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
}
@Override
- public FallbackReadScan withLimit(int limit) {
+ public FallbackReadScan withLimit(long limit) {
mainScan.withLimit(limit);
fallbackScan.withLimit(limit);
return this;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index 97762e225c..1bdc1dbb6b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -78,7 +78,9 @@ public class FormatReadBuilder implements ReadBuilder {
private final CoreOptions options;
@Nullable private Predicate filter;
@Nullable private PartitionPredicate partitionFilter;
+ // Keep the legacy field for Java serialization compatibility.
@Nullable private Integer limit;
+ @Nullable private Long longLimit;
@Nullable private transient FormatTableFileIOResolver fileIOResolver;
public FormatReadBuilder(FormatTable table) {
@@ -161,6 +163,14 @@ public class FormatReadBuilder implements ReadBuilder {
@Override
public ReadBuilder withLimit(int limit) {
this.limit = limit;
+ this.longLimit = (long) limit;
+ return this;
+ }
+
+ @Override
+ public ReadBuilder withLimit(long limit) {
+ this.longLimit = limit;
+ this.limit = limit >= Integer.MIN_VALUE && limit <= Integer.MAX_VALUE
? (int) limit : null;
return this;
}
@@ -176,12 +186,17 @@ public class FormatReadBuilder implements ReadBuilder {
partitionFilter = partitionPredicateOpt.get();
}
}
- return new FormatTableScan(table, partitionFilter, limit);
+ return new FormatTableScan(table, partitionFilter, effectiveLimit());
}
@Override
public TableRead newRead() {
- return new FormatTableRead(readType(), table.rowType(), this, filter,
limit);
+ return new FormatTableRead(readType(), table.rowType(), this, filter,
effectiveLimit());
+ }
+
+ @Nullable
+ private Long effectiveLimit() {
+ return longLimit != null ? longLimit : limit == null ? null :
limit.longValue();
}
protected RecordReader<InternalRow> createReader(FormatDataSplit
dataSplit) throws IOException {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
index 51f3d2b1e4..d3a2eb23e9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
@@ -43,7 +43,7 @@ public class FormatTableRead implements TableRead {
private final RowType tableRowType;
private final Predicate predicate;
private final FormatReadBuilder read;
- private final Integer limit;
+ private final Long limit;
private boolean executeFilter = false;
@Nullable private ReadBatchSizer readBatchSizer;
@@ -53,7 +53,7 @@ public class FormatTableRead implements TableRead {
RowType tableRowType,
FormatReadBuilder read,
Predicate predicate,
- Integer limit) {
+ Long limit) {
this.tableRowType = tableRowType;
this.readType = readType == null ? tableRowType : readType;
this.read = read;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
index 32cd464315..4ad70d2509 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
@@ -78,12 +78,10 @@ public class FormatTableScan implements InnerTableScan {
final CoreOptions coreOptions;
@Nullable private PartitionPredicate partitionFilter;
private final SplitEnumerator splitEnumerator;
- @Nullable private final Integer limit;
+ @Nullable private final Long limit;
public FormatTableScan(
- FormatTable table,
- @Nullable PartitionPredicate partitionFilter,
- @Nullable Integer limit) {
+ FormatTable table, @Nullable PartitionPredicate partitionFilter,
@Nullable Long limit) {
this.table = table;
this.coreOptions = new CoreOptions(table.options());
this.partitionFilter = partitionFilter;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
index c95fb56a36..03a154ce86 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
@@ -56,7 +56,7 @@ public abstract class AbstractBatchTableScan extends
AbstractDataTableScan {
private StartingScanner startingScanner;
private boolean hasNext;
- private Integer pushDownLimit;
+ private Long pushDownLimit;
private TopN topN;
@Nullable private String readProtectionTagName;
@@ -97,7 +97,7 @@ public abstract class AbstractBatchTableScan extends
AbstractDataTableScan {
}
@Override
- public InnerTableScan withLimit(int limit) {
+ public InnerTableScan withLimit(long limit) {
// Record it; applyPushDownLimit pushes the file-store limit only when
safe.
this.pushDownLimit = limit;
return this;
@@ -196,8 +196,8 @@ public abstract class AbstractBatchTableScan extends
AbstractDataTableScan {
OptionalLong mergedRowCount = split.mergedRowCount();
if (mergedRowCount.isPresent()) {
limitedSplits.add(split);
- scannedRowCount += mergedRowCount.getAsLong();
- if (scannedRowCount >= pushDownLimit) {
+ long splitRowCount = mergedRowCount.getAsLong();
+ if (scannedRowCount >= pushDownLimit - splitRowCount) {
SnapshotReader.Plan newPlan =
new PlanImpl(plan.watermark(), plan.snapshotId(),
limitedSplits);
LOG.info(
@@ -207,6 +207,7 @@ public abstract class AbstractBatchTableScan extends
AbstractDataTableScan {
pushDownLimit);
return Optional.of(new ScannedResult(newPlan));
}
+ scannedRowCount += splitRowCount;
}
}
return Optional.of(result);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java
index 99561b7551..ac5f9e3741 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java
@@ -48,7 +48,7 @@ public class AppendTableRead extends AbstractDataTableRead {
@Nullable private RowType readType = null;
private Predicate predicate = null;
protected TopN topN = null;
- protected Integer limit = null;
+ protected Long limit = null;
@Nullable private ReadBatchSizer readBatchSizer;
public AppendTableRead(
@@ -104,7 +104,7 @@ public class AppendTableRead extends AbstractDataTableRead {
}
@Override
- public InnerTableRead withLimit(int limit) {
+ public InnerTableRead withLimit(long limit) {
initialized().forEach(r -> r.withLimit(limit));
this.limit = limit;
return this;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java
index 6602d40a6e..5371ec8092 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java
@@ -74,7 +74,7 @@ final class BlobViewTableReadSupport {
RowType readType,
@Nullable Predicate predicate,
@Nullable TopN topN,
- @Nullable Integer limit,
+ @Nullable Long limit,
boolean executeFilter,
RecordReaderSupplier dataReaderSupplier,
Supplier<InnerTableRead> prescanReadSupplier)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableRead.java
index 0a325e70e9..4f005fa4e8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableRead.java
@@ -56,7 +56,7 @@ public interface InnerTableRead extends TableRead {
return this;
}
- default InnerTableRead withLimit(int limit) {
+ default InnerTableRead withLimit(long limit) {
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
index 38278d4620..6c4b18ec9b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
@@ -43,7 +43,7 @@ public interface InnerTableScan extends TableScan {
return this;
}
- default InnerTableScan withLimit(int limit) {
+ default InnerTableScan withLimit(long limit) {
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
index 117798d562..167e287faa 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
@@ -68,7 +68,7 @@ public final class KeyValueTableRead extends
AbstractDataTableRead {
private Predicate predicate = null;
private IOManager ioManager = null;
@Nullable private TopN topN = null;
- @Nullable private Integer limit = null;
+ @Nullable private Long limit = null;
@Nullable private ReadBatchSizer readBatchSizer;
public KeyValueTableRead(
@@ -146,7 +146,7 @@ public final class KeyValueTableRead extends
AbstractDataTableRead {
}
@Override
- public InnerTableRead withLimit(int limit) {
+ public InnerTableRead withLimit(long limit) {
this.limit = limit;
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilder.java
index ee94211a7f..a83dd59de9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilder.java
@@ -135,6 +135,19 @@ public interface ReadBuilder extends Serializable {
/** the row number pushed down. */
ReadBuilder withLimit(int limit);
+ /**
+ * The row number pushed down.
+ *
+ * <p>The default implementation preserves compatibility with
implementations which only support
+ * integer limits. A limit outside the integer range is left unpushed
instead of being narrowed.
+ */
+ default ReadBuilder withLimit(long limit) {
+ if (limit < Integer.MIN_VALUE || limit > Integer.MAX_VALUE) {
+ return this;
+ }
+ return withLimit(Math.toIntExact(limit));
+ }
+
/**
* Push TopN filter. Will filter the data as much as possible, but it is
not guaranteed that it
* is a complete filter.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
index ee6949eddb..51c76d7183 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
@@ -58,7 +58,9 @@ public class ReadBuilderImpl implements ReadBuilder {
private Predicate filter;
+ // Keep the legacy field for Java serialization compatibility.
private Integer limit = null;
+ private Long longLimit = null;
private TopN topN = null;
private Integer shardIndexOfThisSubtask;
@@ -137,6 +139,14 @@ public class ReadBuilderImpl implements ReadBuilder {
@Override
public ReadBuilder withLimit(int limit) {
this.limit = limit;
+ this.longLimit = (long) limit;
+ return this;
+ }
+
+ @Override
+ public ReadBuilder withLimit(long limit) {
+ this.longLimit = limit;
+ this.limit = limit >= Integer.MIN_VALUE && limit <= Integer.MAX_VALUE
? (int) limit : null;
return this;
}
@@ -189,6 +199,7 @@ public class ReadBuilderImpl implements ReadBuilder {
@Override
public TableScan newScan() {
+ Long limit = effectiveLimit();
InnerTableScan tableScan = configureScan(table.newScan());
if (limit != null) {
tableScan.withLimit(limit);
@@ -240,6 +251,7 @@ public class ReadBuilderImpl implements ReadBuilder {
@Override
public TableRead newRead() {
+ Long limit = effectiveLimit();
InnerTableRead read = table.newRead().withFilter(filter);
if (readType != null) {
read.withReadType(readType);
@@ -252,6 +264,7 @@ public class ReadBuilderImpl implements ReadBuilder {
}
return read;
}
+
if (topN != null) {
read.withTopN(topN);
}
@@ -261,6 +274,11 @@ public class ReadBuilderImpl implements ReadBuilder {
return read;
}
+ @Nullable
+ private Long effectiveLimit() {
+ return longLimit != null ? longLimit : limit == null ? null :
limit.longValue();
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -290,13 +308,13 @@ public class ReadBuilderImpl implements ReadBuilder {
private static class LimitTableRead implements TableRead {
private final TableRead delegate;
- private final int limit;
+ private final long limit;
// with a filter the reader only evaluates it once executeFilter() is
requested;
// otherwise the engine does, after this limit, so capping here would
drop matches
private final boolean filterPresent;
private boolean filterExecutedByReader = false;
- private LimitTableRead(TableRead delegate, int limit, boolean
filterPresent) {
+ private LimitTableRead(TableRead delegate, long limit, boolean
filterPresent) {
this.delegate = delegate;
this.limit = limit;
this.filterPresent = filterPresent;
@@ -350,7 +368,7 @@ public class ReadBuilderImpl implements ReadBuilder {
// Stop reading once the limit is reached (return EOF), rather
than filtering and
// draining the rest of the data.
return new RecordReader<InternalRow>() {
- private int count;
+ private long count;
@Nullable
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index 7bc7150556..e4e252c7ab 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -132,7 +132,7 @@ public interface SnapshotReader {
SnapshotReader withReadType(RowType readType);
- SnapshotReader withLimit(int limit);
+ SnapshotReader withLimit(long limit);
/** Whether the pushed filter still contains non-partition predicates. */
boolean hasNonPartitionFilter();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index cb338c8570..095959ddd3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -353,7 +353,7 @@ public class SnapshotReaderImpl implements SnapshotReader {
}
@Override
- public SnapshotReader withLimit(int limit) {
+ public SnapshotReader withLimit(long limit) {
scan.withLimit(limit);
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index 8f259502ff..5ccc84886e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -515,7 +515,7 @@ public class AuditLogTable implements DataTable,
ReadonlyTable {
}
@Override
- public SnapshotReader withLimit(int limit) {
+ public SnapshotReader withLimit(long limit) {
wrapped.withLimit(limit);
return this;
}
@@ -582,7 +582,7 @@ public class AuditLogTable implements DataTable,
ReadonlyTable {
}
@Override
- public InnerTableScan withLimit(int limit) {
+ public InnerTableScan withLimit(long limit) {
batchScan.withLimit(limit);
return this;
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
index 7b3f7b0eff..3e48b5e9b2 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
@@ -53,6 +53,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.io.ObjectStreamClass;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
@@ -69,6 +70,12 @@ public class FormatReadBuilderTest {
@TempDir java.nio.file.Path tempPath;
+ @Test
+ public void testSerializedLimitFieldCompatibility() {
+
assertThat(ObjectStreamClass.lookup(FormatReadBuilder.class).getField("limit").getType())
+ .isEqualTo(Integer.class);
+ }
+
@Test
public void testSerializeAndDeserialize() throws IOException,
ClassNotFoundException {
RowType rowType =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/ReadBuilderImplTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/ReadBuilderImplTest.java
new file mode 100644
index 0000000000..c080905f03
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/ReadBuilderImplTest.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.table.InnerTable;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ObjectStreamClass;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Answers.CALLS_REAL_METHODS;
+import static org.mockito.Answers.RETURNS_SELF;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link ReadBuilderImpl}. */
+public class ReadBuilderImplTest {
+
+ @Test
+ public void testLongLimitCompatibilityBridge() {
+ ReadBuilder legacyBuilder = mock(ReadBuilder.class,
CALLS_REAL_METHODS);
+
+ legacyBuilder.withLimit((long) Integer.MAX_VALUE);
+ verify(legacyBuilder).withLimit(Integer.MAX_VALUE);
+
+ clearInvocations(legacyBuilder);
+
assertThat(legacyBuilder.withLimit(4294967297L)).isSameAs(legacyBuilder);
+ verify(legacyBuilder, never()).withLimit(anyInt());
+ }
+
+ @Test
+ public void testSerializedLimitFieldCompatibility() {
+
assertThat(ObjectStreamClass.lookup(ReadBuilderImpl.class).getField("limit").getType())
+ .isEqualTo(Integer.class);
+ }
+
+ @Test
+ public void testLongLimitForwardedToScanAndRead() {
+ InnerTable table = mock(InnerTable.class);
+ when(table.name()).thenReturn("table");
+ when(table.options()).thenReturn(Collections.emptyMap());
+ when(table.partitionKeys()).thenReturn(Collections.emptyList());
+ when(table.rowType()).thenReturn(RowType.of(DataTypes.INT()));
+
+ InnerTableScan scan = mock(InnerTableScan.class, RETURNS_SELF);
+ InnerTableRead read = mock(InnerTableRead.class, RETURNS_SELF);
+ when(table.newScan()).thenReturn(scan);
+ when(table.newRead()).thenReturn(read);
+
+ long limit = 4294967297L;
+ ReadBuilderImpl builder = new ReadBuilderImpl(table);
+ builder.withLimit(limit);
+ builder.newScan();
+ builder.newRead();
+
+ verify(scan).withLimit(limit);
+ verify(read).withLimit(limit);
+ }
+}
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 eb3da3e163..ec09badf9e 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
@@ -212,7 +212,7 @@ public class FlinkSourceBuilder {
readBuilder.withPartitionFilter(partitionPredicate);
}
if (limit != null) {
- readBuilder.withLimit(limit.intValue());
+ readBuilder.withLimit(limit);
}
return readBuilder.dropStats();
}
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 709b0e733b..d2f708f2d1 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
@@ -31,8 +31,10 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.IncrementalSplit;
import org.apache.paimon.table.source.QueryAuthSplit;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.runtime.plugable.SerializationDelegate;
@@ -58,6 +60,10 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Answers.RETURNS_SELF;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Test for {@link FlinkSourceBuilder}. */
public class FlinkSourceBuilderTest {
@@ -153,6 +159,26 @@ public class FlinkSourceBuilderTest {
assertThat(transformation.getSource()).isInstanceOf(PaimonDataStreamSource.class);
}
+ @Test
+ public void testLongLimitForwardedToReadBuilder() {
+ Table table = mock(Table.class);
+ when(table.name()).thenReturn("table");
+ when(table.options())
+ .thenReturn(Collections.singletonMap("path",
tempDir.toUri().toString()));
+ when(table.primaryKeys()).thenReturn(Collections.emptyList());
+ when(table.rowType()).thenReturn(RowType.of(DataTypes.INT()));
+ ReadBuilder readBuilder = mock(ReadBuilder.class, RETURNS_SELF);
+ when(table.newReadBuilder()).thenReturn(readBuilder);
+
+ new FlinkSourceBuilder(table)
+ .env(StreamExecutionEnvironment.getExecutionEnvironment())
+ .sourceBounded(true)
+ .limit(4294967297L)
+ .build();
+
+ verify(readBuilder).withLimit(4294967297L);
+ }
+
@Test
public void testBuildWrapsContinuousSourceWithPaimonDataStreamSource()
throws Exception {
Table table = createTable("continuous_source", false, -1, true);
