Add support for concurrent write on Iceberg transformed column

[pajaks]

Description

Add concurrent write to scenarios like:

-- create test table in iceberg
create table all_defaults_partitioned
with (
  partitioning = array['month(shipdate)']
)
as select * from tpch.sf1000.lineitem

-- first session
update all_defaults_partitioned
set orderkey = 654
where shipdate = date '1995-01-01'

-- second session 
update all_defaults_partitioned
set orderkey = 765
where shipdate = date '1996-12-01'

Release notes

## Iceberg
* Fix failure when writing concurrently with [transformed partition](https://iceberg.apache.org/spec/#partition-transforms) columns

[findinpath]

Why do we intersect enforced with the unenforced predicate?
They are rather unrelated.

[pajaks]

My understanding is that both carriers information about predicate.
The difference is that taken from @raunaqmorarka explanation:

enforced is the part of the predicate which is guaranteed to be satisfied by the connector, so the engine will not apply it on it's side.
unenforced is the part of the predicate which connector cannot guarantee even if it is able to use it to reduce output, so the engine will apply it on the connector output

@raunaqmorarka Correct me if I'm wrong here

[findinpath]

{2:part:date=[ SortedRangeSet[type=date, ranges=1, {[2024-01-01]}] ]}

I'm checking io.trino.plugin.iceberg.TestIcebergLocalConcurrentWritesTest#testConcurrentUpdateWithOverlappingPartitionTransformation (BTW really cool that we have this battery of new concurrency tests)

It's unclear to me why the partition predicate is not an "enforced predicate" while debugging your code.

[pajaks]

Adding such predicates for transformed columns as enforced, would mean pushdowns of those values and connector would need to filter files and rows also during reading from such table as well. It's not supported right now and would require more work.

[pajaks]

They are rather unrelated.

We already join those in IcebergSplitSource

trino/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

Line 259 in 61a79dd

TupleDomain<IcebergColumnHandle> effectivePredicate = TupleDomain.intersect(

[raunaqmorarka]

We are able to prune files in IcebergSplitSource based on unenforced filter, but not rows (hence "unenforced"). Filters on identity partitioned columns are enforced at the row level due to the data being partitioned by value and PartitionConstraintMatcher in IcebergSplitSource.
I think it's valid to intersect unenforced filter here because org.apache.iceberg.MergingSnapshotProducer is internally checking whether any files matching given filter have been added to the table.

[raunaqmorarka]

We are able to prune files in IcebergSplitSource based on unenforced filter, but not rows (hence "unenforced"). Filters on identity partitioned columns are enforced at the row level due to the data being partitioned by value and PartitionConstraintMatcher in IcebergSplitSource.
I think it's valid to intersect unenforced filter here because org.apache.iceberg.MergingSnapshotProducer is internally checking whether any files matching given filter have been added to the table.

[raunaqmorarka]

lgtm
But I would like a second opinion on it
@alexjo2144 @findepi @losipiuk @hashhar @electrum are any of you able to weigh in here ?

[findinpath]

Pretty cool @pajaks 🎉

Outstanding test coverage.

[pajaks]

@ebyhr @raunaqmorarka Can this be merged?