Improve performance for Equality Delete files in Iceberg connector

[jasonf20]

Description

The goal of this PR is to improve the performance of queries that contain Equality Delete files.

Before this commit equality delete files were re-loaded for every split that needed them. Now, each delete file is loaded once per execution node.

In addition, the equality deletes are stored in a single map (per delete schema) with the data sequence number in which the row was deleted. This allows us to merge rows that were deleted multiple times (common in upsert use cases) into a single entry instead of holding an entry per delete file.

Additional context and related issues

Fixes #18396

Changes in this commit:

• Create a DeleteManager class to manage the above logic
• Read every delete file only once
• Merge deletes into single map
• Opportunistic delete file load parallelization

Release notes

( ) This is not user-visible or 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:

# Iceberg
* Improved performance and memory usage when [Equality Delete](https://iceberg.apache.org/spec/#equality-delete-files)  files are used ({issue}`18396`)

[cla-bot]

Thank you for your pull request and welcome to the Trino community. We require contributors to sign our Contributor License Agreement, and we don't seem to have you on file. Continue to work with us on the review and improvements in this PR, and submit the signed CLA to cla@trino.io. Photos, scans, or digitally-signed PDF files are all suitable. Processing may take a few days. The CLA needs to be on file before we merge your changes. For more information, see https://github.com/trinodb/cla

[cla-bot]

Thank you for your pull request and welcome to the Trino community. We require contributors to sign our Contributor License Agreement, and we don't seem to have you on file. Continue to work with us on the review and improvements in this PR, and submit the signed CLA to cla@trino.io. Photos, scans, or digitally-signed PDF files are all suitable. Processing may take a few days. The CLA needs to be on file before we merge your changes. For more information, see https://github.com/trinodb/cla

[cla-bot]

Thank you for your pull request and welcome to the Trino community. We require contributors to sign our Contributor License Agreement, and we don't seem to have you on file. Continue to work with us on the review and improvements in this PR, and submit the signed CLA to cla@trino.io. Photos, scans, or digitally-signed PDF files are all suitable. Processing may take a few days. The CLA needs to be on file before we merge your changes. For more information, see https://github.com/trinodb/cla

[cla-bot]

Thank you for your pull request and welcome to the Trino community. We require contributors to sign our Contributor License Agreement, and we don't seem to have you on file. Continue to work with us on the review and improvements in this PR, and submit the signed CLA to cla@trino.io. Photos, scans, or digitally-signed PDF files are all suitable. Processing may take a few days. The CLA needs to be on file before we merge your changes. For more information, see https://github.com/trinodb/cla

[yoniiny]

@findinpath This PR is the fix for the equality deletes issue we discussed. We have another PR coming that should further improve for situations where there are a lot of deletes (>~200M).

[findinpath]

@jasonf20 does your contribution overlap with #17115 ?

Note that there is in the above mentioned PR scaffolding for testing. Please use it here as well to ensure the validity of your changes.

Read every delete file only once

Is this something we can verify as part of this PR (before & after your change) ?
Have a look at https://github.com/trinodb/trino/blob/18b47f0223c7d7cb74fbb8c16f9b3871ad79690b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergMetadataFileOperations.java

[jasonf20]

Hi @findinpath

It does seem like this PR overlaps with #17115. This PR should do the same map compaction that is done in that PR. However, it includes more optimizations on top of that including:

• Reading delete files only once
  • This adds some complexity since the shared state must eventually be cleaned up. This is done with a WeakReference in this PR.
  • Also the data sequence number needs to be taken into account since we share the deletes map across splits
• Optimistically loading delete files in parallel (despite reading each file only once, we will allow different splits to load the files in a different order so that multiple files are loaded at once)

I think that it should suffice to merge only one of these PRs. If the other one is merged I will need to rebase and add the above improvements on top of the previous PR.

I have added two tests based on the templates/suggestion I saw in the other PR.

I was able to use the MetadataFileOperations spec as a guideline for adding a test that validates the delete file is loaded only once. Added as a new commit.

[findinpath]

@jasonf20 let's concentrate on landing first the PR #17115 which comes with less changes.
There are a few corner cases that are not fully obvious which need to be covered in detail before landing these changes. Feel free to contribute with suggestions on the above mentioned PR to accelerate its landing.

[jasonf20]

@findinpath Happy to assist with the merging of #17115. Is there anything specifically there? Looking at the PR it seems most action items have been handled.

Once that's merged I'll rebase on top of that with the remaining changes.

Keep in mind that as it stands the other PR isn't enough for the use case of a large table with constant upserts to actually work since each split loads all the delete files sequentially.

[findinpath]

Happy to assist with the merging of #17115.
Is there anything specifically there?

We stumbled on a situation related to using nested fields for equality deletes. The PR will be adding proper handling for such situations.
It seems that, at the moment, Trino is not capable to read such delete files.
In any case, adding support in Trino for dealing with nested fields in equality deletes is not in the scope of the above mentioned PR.

[findinpath]

@jasonf20 #17115 has been merged.
Feel free to continue the work on this PR.

[yoniiny]

@findinpath This PR is ready for review

[sopel39]

@jasonf20 IIUC this PR is caching deletion files. However, it seems that these can be cached per split rather than per query.
In that case ConnectorPageSourceProviderFactory seems like too narrow scope.

[jasonf20]

@jasonf20 IIUC this PR is caching deletion files. However, it seems that these can be cached per split rather than per query. In that case ConnectorPageSourceProviderFactory seems like too narrow scope.

The delete files are already read only once per split. The issue is that caching per task/query is required otherwise the delete files are read in each split leading to unusable performance. This approach was designed with @pettyjamesm as the cleanest way to cache at the right level.

[sopel39]

@jasonf20 Would it make sense to cache deletion files across queries too?

[jasonf20]

@sopel39

It could be useful for performance when querying the same tables but the same can be said for data files. Perhaps a generic cache of this sort would be implemented in Trino at some point but it's probably a much more involved task, considering memory usage and such.

For this PR specifically it would be interesting to consider caching at the query level (this PR does it at the Task Level). But for most queries caching at the task level should produce the same performance. We discussed query level caches and decided it can probably be built on top of this Task Level caching with reference counting or something like that in the future if needed. So the first step is to solve this at this level which should have the largest impact and potentially expand the caching scope later.

The main motivation here is to make queries on tables with equality deletes complete in a reasonable amount of time. The current implementation doesn't really work with more than a couple equality delete files

[jasonf20]

@dain @pettyjamesm Should the PageSourceProviderFactory be closeable? It acts as a singleton now (like PageSourceProvider did before). Should we instead make PageSourceProvider closeable? If so when would be the right place to close it? I don't think OperatorFactory::noMoreOperators is the right place since IIUC that is called before the splits are read.

I don't think the use case here really has anything to close, it should all be cleaned up by GC, but if we are creating a stateful instance we might want it as part of the interface.

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[yoniiny]

@findinpath we'd love to get this PR merged. Are there any outstanding action items here?

[mosabua]

Replaced by #21441