wwj6591812 commented on code in PR #8116:
URL: https://github.com/apache/paimon/pull/8116#discussion_r3465051833
##########
paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java:
##########
@@ -129,11 +141,104 @@ public InnerTableRead withTopN(TopN topN) {
@Override
public InnerTableRead withLimit(int limit) {
- initialized().forEach(r -> r.withLimit(limit));
this.limit = limit;
+ refreshReaderConfig();
return this;
}
+ @Override
+ public RecordReader<InternalRow> createReader(List<Split> splits) throws
IOException {
+ if (splits.isEmpty()) {
+ return createConcatenatedReader(splits);
+ }
+
+ if (limit == null || limit <= 0) {
+ return createConcatenatedReader(splits);
+ }
+
+ if (needsTableLevelLimit(splits)) {
+ return createReaderWithGlobalLimit(splits);
+ }
+ return createConcatenatedReader(splits);
+ }
+
+ private boolean needsTableLevelLimit(List<Split> splits) {
+ if (splits.isEmpty()) {
+ return false;
+ }
+ if (splits.size() > 1) {
+ return true;
+ }
+ if (!isMergeReadLimitActive(splits.get(0))) {
+ return true;
+ }
+ return hasLateAppliedRowFilter(splits);
+ }
+
+ private static boolean hasLateAppliedRowFilter(List<Split> splits) {
+ for (Split split : splits) {
+ if (split instanceof QueryAuthSplit) {
+ TableQueryAuthResult authResult = ((QueryAuthSplit)
split).authResult();
+ if (authResult != null && authResult.extractPredicate() !=
null) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private RecordReader<InternalRow> createReaderWithGlobalLimit(List<Split>
splits)
+ throws IOException {
+ boolean previousApplyMergeReadLimit = applyMergeReadLimit;
+ applyMergeReadLimit = false;
Review Comment:
Thanks for the review, @JingsongLi .
You're right — keeping applyMergeReadLimit disabled until the returned
reader is closed was unsafe because this flag is shared by the reusable
KeyValueTableRead instance. A second reader created before the first one is
closed, or a failed close(), could leave unrelated reads with merge-read limit
disabled.
I've fixed this by materializing each split reader while merge-read limit is
disabled, wrapping them with the table-level LimitRecordReader, and restoring
applyMergeReadLimit in a finally block before returning. The lazy
ConcatRecordReader suppliers now return the pre-created split readers, so the
disablement is scoped to this fallback path without depending on reader close
ordering.
I also added testGlobalLimitReaderRestoresMergeReadLimitImmediately to
verify that applyMergeReadLimit is restored immediately after createReader()
returns on the multi-split global-limit path. Please take another look when you
have a chance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
