You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "leekeiabstraction (via GitHub)" <gi...@apache.org> on 2025/12/01 11:30:55 UTC
[PR] [Log] Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer [fluss]
leekeiabstraction opened a new pull request, #2061:
URL: https://github.com/apache/fluss/pull/2061
### Purpose
Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer
Linked issue: close #2041
### Brief change log
Change the signatures and implementation that currently returns List<ScanRecord> to return CloseableIterator<ScanRecord>. This includes (not exhaustive):
- LogFetchCollector's Map<TableBucket, List<ScanRecord>> collectFetch(LogFetchBuffer) to Map<TableBucket, CloseableIterator<ScanRecord>> collectFetch(LogFetchBuffer)
- LogFetcher's Map<TableBucket, List<ScanRecord>> collectFetch(LogFetchBuffer) to Map<TableBucket, CloseableIterator <ScanRecord>> collectFetch(LogFetchBuffer)
Lazy initialisation of CompletedFetch and deserialisation of ScanRecord.
The closing of resources is done within:
- FetchCollector#ScanRecordIterator.close() and makeNext() (to ensure that concatenated CloseableIterators are closed upon finish consumption)
- LogScannerImpl.close() where LogScannerImpl will track unclosed CloseableIterators and close them.
- ScanRecords.close()
- FlinkRecordsWithSplitIds.recycle()
- TieringSplitReader.fetch()
### Tests
TODO: Address failing IT test cases and add additional tests.
### API and Format
TODO: Decide LogScanner/ScanRecords interface changes. See https://github.com/apache/fluss/issues/2041#issuecomment-3592642088
### Documentation
TODO: Update documentation
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Log] Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #2061:
URL: https://github.com/apache/fluss/pull/2061#issuecomment-3638424954
@wuchong
On further thoughts, I am more inclined towards having a different interface for LogScanner that supports Iterable/Iterator instead of changing current LogScanner interface so that current users are not impacted and do not need to migrate and risk resource leakage when forgetting to close.
Should I raise a FIP for this?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Log] Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #2061:
URL: https://github.com/apache/fluss/pull/2061#discussion_r2577871284
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java:
##########
@@ -50,12 +51,11 @@ public ScanRecords(Map<TableBucket, List<ScanRecord>> records) {
*
* @param scanBucket The bucket to get records for
*/
- public List<ScanRecord> records(TableBucket scanBucket) {
- List<ScanRecord> recs = records.get(scanBucket);
- if (recs == null) {
- return Collections.emptyList();
+ public Iterator<ScanRecord> records(TableBucket scanBucket) {
+ if (records.containsKey(scanBucket)) {
+ return records.get(scanBucket);
}
- return Collections.unmodifiableList(recs);
+ return CloseableIterator.emptyIterator();
Review Comment:
Discussed with @polyzos and came to the same page to leave ScanRecords interface unchanged.
To confirm: Iterators will only be used within internal components e.g. FlinkSourceSplitReader and TieringSplitReader
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Log] Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #2061:
URL: https://github.com/apache/fluss/pull/2061#issuecomment-3794740704
Deprecating this draft as its benefits are small (we use small batches of 200 IIRC) and there are higher priorities for 0.9 release and 2026 roadmap.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Log] Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction closed pull request #2061: [Log] Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer
URL: https://github.com/apache/fluss/pull/2061
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org