Convert the retrieval of the active data files to streaming

[findinpath]

Description

Instead of waiting to get in bulk the collection of all
the active data files of a table relevant to a query,
provide a stream of active add entries so that the
split manager can start providing to the engine splits right away.

The active add entries stream is closeable in order to ensure that
the underlying checkpoint parquet page source gets closed
even when not all the entries from the checkpoint are consumed.

These changes should provide visible improvements in terms of reduced memory requirements when dealing with Delta Lake tables having many active add files, because as of now these files will not be held anymore in memory within DeltaLakeSplitManager due to the fact that the add files are processed as a consequence of the current changes now one by one.

Potential downside The checkpoint entry iterator will keep open the parquet page source corresponding to the checkpoint (multipart) file until all the add entries are processed by the DeltaLakeSplitSource.

Additional context and related issues

The access to the active add stream needs to be done within a try-with-resources statement to make sure that the checkpoint iterator does indeed release the opened resources.

The key output of these changes is that in the DeltaLakeSplitManager & DeltaLakeSplitSource don't block anymore until all the add entries relevant to the query are collected, but instead the splits are provided gradually (in parallel with the query execution).

This change should particularly speed up the planning of the queries involving Delta Lake tables where the table statistics are not required because the split generation can start right away.

Fixes #20088

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
(x) Release notes are required, with the following suggested text:

# Delta Lake
* Speed up the beginning of the query execution for queries which don't involve table statitstics. ({issue}`issuenumber`)

[alexjo2144]

We may want to consider standardizing all the code here on Iterators rather than Streams. Going back and forth has some overhead and generally I think we prefer iterators for performance sensitive things.

[findinpath]

Do we have a potential memory leak with this change?

trino/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/checkpoint/CheckpointEntryIterator.java

Line 221 in dc880e9

this.pageSource = (ParquetPageSource) pageSource.get();

The pageSource is opened, but if the entries are not consumed to the very last one, the pageSource is not being closed.

Do we maybe need a concept similar to https://github.com/apache/iceberg/blob/263b530502e5597b19b6b5e282917af8eede7600/api/src/main/java/org/apache/iceberg/io/CloseableIterator.java#L29 ?

[jkylling]

Do we have a potential memory leak with this change?

trino/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/checkpoint/CheckpointEntryIterator.java

Line 221 in dc880e9

this.pageSource = (ParquetPageSource) pageSource.get();

The pageSource is opened, but if the entries are not consumed to the very last one, the pageSource is not being closed.

Do we maybe need a concept similar to https://github.com/apache/iceberg/blob/263b530502e5597b19b6b5e282917af8eede7600/api/src/main/java/org/apache/iceberg/io/CloseableIterator.java#L29 ?

We should. One possible alternative could be to use that streams are AutoCloseable, and expose the CheckpointEntryIterator as a Stream, hook the closing of the stream up to the closing of the page source, and always consume the stream in a try-with statement.

[findepi]

"Rewrite data file filtering logic for OPTIMIZE statements"

[findinpath]

Rebased on master to address minor code conflicts

[findepi]

#20121 merged, please rebase.

[findepi]

"Add close logic for the Delta Lake splits"

-- add some color to the commit description

[findepi]

"Process lazily multi-part checkpoint files"

[jkylling]

LGTM

[jkylling]

Could there be a potential resource leak here (which might not be introduced as part of this PR)? We call TrinoInputFile.newInput() within ParquetPageSourceFactory.createPageSource as part of the constructor of CheckpointEntryIterator. If some later part of the constructor fails, we likely won't get to close the file.

[findinpath]

We call TrinoInputFile.newInput() ...

I'm assuming you are refferring to

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/parquet/TrinoParquetDataSource.java

Lines 39 to 47 in 875fe1b

	this.input = file.newInput();
	}
	@Override
	public void close()
	throws IOException
	{
	input.close();
	}

If some later part of the constructor fails, we likely won't get to close the file.

There is actually exception handling in place for this specific scenario. See

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/parquet/ParquetPageSourceFactory.java

Lines 285 to 290 in 875fe1b

	catch (Exception e) {
	try {
	if (dataSource != null) {
	dataSource.close();
	}
	}

I'm adding though exception handling in the CheckpointEntryIterator constructor to make sure that the page source gets closed in case that there are eventual exceptions happening after initializing the pageSource field.

[ebyhr]

Looks good to me. How about using @MustBeClosed annotation to enforce try-with-resources? The follow-up PR is fine though.

[findinpath]

How about using @MustBeClosed annotation to enforce try-with-resources?

I tried adding @MustBeClosed but I stumbled in DeltaLakeSplitManager because I need to pass further the stream to DeltaLakeSplitSource.

[findepi]

can @MustBeClosed resource be further returned if it's still @MustBeClosed?