JingsongLi commented on code in PR #8116:
URL: https://github.com/apache/paimon/pull/8116#discussion_r3464830732
##########
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:
`createReaderWithGlobalLimit` keeps `applyMergeReadLimit` disabled until the
returned reader is closed, but this flag is shared by the reusable
`KeyValueTableRead` instance. A caller can create another reader from the same
table read before this global-limit reader is closed, and that second reader
will see `applyMergeReadLimit=false`, route through the fallback, and
initialize unrelated split readers with the merge-read limit disabled. If
`reader.close()` throws, the flag is also never restored. This makes the new
optimization depend on reader close ordering. Please scope the disablement to
the readers created for this fallback (for example, materialize/capture the
per-split readers while disabled and restore in a `finally`, or pass a per-call
flag to the suppliers) instead of leaving shared table-read state changed until
close.
--
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]
