[Log] Improve internals of LogScanner by lazily deserialising from record stream / arrow buffer

[leekeiabstraction]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Optimise/decrease memory footprint when LogScannerImpl.poll(Duration) is used by addressing TODO here: https://github.com/apache/fluss/blob/main/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java#L96-L97

Solution

Proposed changes as follow:

Change the signatures and implementation that currently returns List<ScanRecord> to return CloseableIterator<ScanRecord>. This includes (not exhaustive):

1. CompletedFetch's List<ScanRecord> fetchRecords(int) to CloseableIterator<ScanRecord> fetchRecords(). Current int argument is to indicate max number of records to fetch and deserialise, I propose removing it and let the eventual user of the iterator decide how many to fetch. An alternative here is to make CompletedFetch implements Iterable<ScanRecord>.
2. LogFetchCollector's Map<TableBucket, List<ScanRecord>> collectFetch(LogFetchBuffer) to Map<TableBucket, CloseableIterator<ScanRecord>> collectFetch(LogFetchBuffer)
3. LogFetcher's Map<TableBucket, List<ScanRecord>> collectFetch(LogFetchBuffer) to Map<TableBucket, CloseableIterator <ScanRecord>> collectFetch(LogFetchBuffer)

The closing of resources is done in two places:

1. Within ScanRecords.ConcatenatedIterable where CloseableIterators are closed when they have no next element.
2. Within LogScannerImpl.close() where LogScannerImpl will track unclosed CloseableIterators and close them.

Anything else?

Feedback and suggestion on alternative approaches welcome!

Willingness to contribute

• I'm willing to submit a PR!

[leekeiabstraction]

@wuchong Appreciate your feedback here and also if you can assign the issue to me. Thank you!

[wuchong]

Thank you @leekeiabstraction , this looks good to me. One more thing we need to take care of is the resource closing, the ScanRecords should also be closed if any failure happens, this can be done by closing ScanReocrds/CloseableIterators in org.apache.flink.connector.base.source.reader.RecordsWithSplitIds#recycle.

[leekeiabstraction]

@wuchong Would appreciate some input or suggestions from you on the following that may impact LogScanner.poll(int timeout) and whether we expose lazily initialised ScanRecords to direct users of LogScanner.

Currently ScanRecords.records(TableBucket scanBucket) returns List<ScanRecord>. To realise the benefits of lazy initialisation, I'd need to make the method return an Iterator<ScanRecord> instead. I can add ScanRecords.close() which user can call (automatic closing when iterator reaches end will also be added).

But question remains on the behaviour that I should be aiming for in LogScanner.poll(). Options are:

1. If previously returned ScanRecords are not fully consumed, poll() will always return ScanRecords.EMPTY. This is straightforward to implement but introduces issue where parallelisation of record consumption from multiple TableBucket is diminished.
2. If previously returned ScanRecords are not fully consumed, poll will only return ScanRecords which do not have records from similar TableBucket. Better parallelisation than Option 1 as consumption of records from other TableBucket is not blocked.

Of these two, I am inclined towards option 2 as I imagine that current FlinkSourceSplitReader and TieringSourceReader should work fine with it. Question arises however for users who use LogScanner directly. Unless they finish consuming (or call close()) on ScanRecords, they will end up with LogScanner returning ScanRecords.EMPTY. This deviates from existing behaviour where user can ignore previous results entirely and still get latest results using poll().

Finally there's a third option, which is to only expose and use iterator with lazy initialisation internally for components like FlinkSourceSplitReader or TieringSplitReader. Out of these three options, I prefer option 3 the most to protect LogScanner user from resource leak or deadlock when they do not close() or fully consume ScanRecords.

I suggest opt 3 where we do not expose lazy initialised iterators and only use lazy initialisation within internal components.