Prune unused stats columns when reading Delta checkpoint

[findinpath]

Description

Follow-up work from #19588 used to retrieve from the checkpoint file only the add file statistics for the columns projected in the query.

This change can save up CPU and IO time from deserializing unnecessary add.stats_parsed.... columns in a Delta Lake query.

Used for testing a multi-part checkpoint file (25 parts , each around 12MB ~ 300MB in total) for testing this feature while storing the checkpoint in local MinIO and came up with the following results:

ADD without any checkpoint filtering
number of add entries: Optional[1235155]
checkpoint iterator completed positions: Optional[1235157]
checkpoint iterator completed bytes: Optional[323227977]
Elapsed Time in milliseconds: 20031

ADD with checkpoint filtering (no partition pruning) stats Projection
number of add entries: Optional[1235155]
checkpoint iterator completed positions : Optional[1235157]
checkpoint iterator completed bytes: Optional[290305358]
Elapsed Time in milliseconds: 12899

ADD with checkpoint filtering with partition pruning without stats projection
number of add entries: Optional[18578]
checkpoint iterator completed positions: Optional[49290]
checkpoint iterator completed bytes: Optional[14099674]
Elapsed Time in milliseconds: 1011

ADD with checkpoint filtering with partition pruning with stats projection
number of add entries: Optional[18578]
checkpoint iterator completed positions: Optional[49290]
checkpoint iterator completed bytes: Optional[13106732]
Elapsed Time in milliseconds: 802

As can be seen on the listing above, compared to the partition pruning change from #19588 , the change here does not provide an explosive improvement in efficiency, but it manages to shave off a noticeable wait time in processing the checkpoint.

Fixes #19733

Additional context and related issues

Same as in #19588 , the add stats projection functioanlity is effective only when the session setting checkpoint_filtering_enabled is set to true

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
* Improve performance when reading large checkpoint files on partitioned tables. ({issue}`issuenumber`)

[findinpath]

These changes were necessary to be able to showcase the effectiveness of the stats_parsed projection functionality.

[jkylling]

Thank you for adding this!

I mostly have nitpicks for you :)

We could perhaps just pass the underlying set for addStatsColumnFilter around, instead of a function. Then it could be renamed projectedColumns. Maybe we could even pass the projectedColumns to the CheckpointEntryIterator. That would set us up for supporting statistics projection of non-base columns in the future.

Perhaps unrelated to this PR, but it might be good to have a test case which tests performance of checkpoint reads from tables with a very wide schema (e.g., internally we have a table with 276 columns).

[findinpath]

We could perhaps just pass the underlying set for addStatsColumnFilter around, instead of a function. Then it could be renamed projectedColumns

My initial version of the code contained projectedColumns all the way down to CheckpointEntryIterator. However semantically it looked weird to deal with Optional.empty() for the situation where all the columns are to be retrieved.

See in TransactionLogAccess

    @Deprecated
    public List<AddFileEntry> getActiveFiles(TableSnapshot tableSnapshot, MetadataEntry metadataEntry, ProtocolEntry protocolEntry, ConnectorSession session)
    {
        return retrieveActiveFiles(tableSnapshot, metadataEntry, protocolEntry, TupleDomain.all(), Optional.empty(), session);
    }

It seems weird to pass Optional.empty() when we intend to select all the columns.

[findepi]

Normalized (lowercased) or not?

[findinpath]

We are using here the original column names - e.g a_NuMbEr instead of a_number.

See corresponding test io.trino.plugin.deltalake.TestDeltaLakeBasic#testCheckpointFilteringForParsedStatsWithCaseSensitiveColumnNames and test resource: databricks133/parsed_stats_case_sensitive/_delta_log/00000000000000000002.checkpoint.parquet

    optional group stats_parsed {
      optional int64 numRecords;
      optional group minValues {
        optional int32 a_NuMbEr;
        optional binary a_StRiNg (STRING);
      }