Core, Data, Spark: Build equality delete map once per task

[kinolaev]

Equality delete files can be referenced by many data files. This PR allows to accumulate all equality deletes of a table snapshot in the structure spec id -> partition -> equality field ids -> equality key -> max data sequence number that can be reused for many data files. If passed to DeleteFilter, it fills the structure lazily.

I added shared equality deletes to Spark's RowDataReader and BatchDataReader by default, because the size of the shared structure is bounded by split size (multiplied by compression rate). To GenericReader I added a boolean parameter shareEqDeletes which is false by default, because the size is unbounded, and switching the parameter to true might cause OOMs. And I don't use Flink, so I hesitated to propose changing its behavior.

The non-shared DeleteFilter path is unchanged, so Flink and other engines are byte-for-byte identical. The shared path produces identical results (equivalence covered by tests).

Additionally, this PR changes task weight calculation algorithm - each delete file is accounted only once per bin even if it is referenced by several data files in that bin. This change can be moved into a separate PR if preferred.

In our case these changes reduced Spark memory consumption by 30-50% and, more important, prevented Spark from hanging for hours on building and evicting huge equality delete sets for each data file. For frequently updated tables (several thousands of equality delete files per day) ingested with Debezium Iceberg Sink rewrite_data_files execution time is reduced from several hours to several minutes.

Btw, I found that Trino uses a similar technique: EqualityDeleteFilter, DeleteManager.

[singhpk234]

wonder if you tried this already

#9563 ?

[kinolaev]

Hi @singhpk234 , I didn't try it, because I have only one big executor.