Support Iceberg Changelog Scan

[agnes-xinyi-lu]

Implement trino support for query changelog of iceberg tables. like SparkChangelogTable with Iceberg Changelog Schema.

To get all the inserts/deletes for a table:

SELECT * FROM db."tableName$changes"
ORDER BY _change_ordinal ASC

If we could agree on the syntax to specify start/end snapshot id for range query #17079 , then we could query changes for a range like this:

SELECT * FROM db."tableName$changes" FOR VERSION FROM startSnapshotId TO endSnapshotId

[alexjo2144]

Hey @agnes-xinyi-lu , thanks for taking a stab at this. I had put up a draft a while back with some parts of this #15677 but I think the Iceberg core code has changed a bit since then so it is likely pretty out of date.

Since then we've also added a CDC implementation for Delta Lake #16205 which uses a polymorphic table function (PTF) syntax rather than a system table. We probably want to go that route here as well.

I had to do a minor refactor on the IcebergSplitSource to make it accept both FileScanTask and ChangelogScanTask.

The PTF processor SPI should also let us get away with fewer changes in the main split generation path.

[alexjo2144]

I had been putting off working on this mostly because Trino uses merge-on-read for all write operations, and I don't believe those are supported by IncrementalChangeLogScan yet apache/iceberg#6182 and https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/BaseIncrementalChangelogScan.java#LL109C12-L109C12

[agnes-xinyi-lu]

I has been putting off working on this mostly because Trino uses merge-on-read for all write operations, and I don't believe those are supported by IncrementalChangeLogScan yet apache/iceberg#6182 and https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/BaseIncrementalChangelogScan.java#LL109C12-L109C12

Our pipelines use Spark for writing (copy on write, we mostly do insert overwrite) and Trino for read. And it needs this changelog feature during read. In Spark changelog read support it throws exception when there is row level deletes, and so is here in this PR.
Can we make this support available first like Spark?

[alexjo2144]

Yep, I think that's fine. Just calling out the limitation.

[agnes-xinyi-lu]

I had put up a draft a while back with some parts of this #15677 but I think the Iceberg core code has changed a bit since then so it is likely pretty out of date.
Since then we've also added a CDC implementation for Delta Lake #16205 which uses a polymorphic table function (PTF) syntax rather than a system table. We probably want to go that route here as well.

Your draft implementation looks very similar to mine, except for the syntax. Using table function like delta lake makes sense. Can we firstly create an issue or small PR to get the table function syntax approved?
Then we can reactivate your draft (or we can use my version and I can change it to table function instead of system table, either way is fine by me). Please let me know your preferences @alexjo2144.

[alexjo2144]

Let's resurrect my draft and pick things up there. I'll get that branch rebased in the next couple days.

[agnes-xinyi-lu]

Let's resurrect my draft and pick things up there. I'll get that branch rebased in the next couple days.

Great! Thanks!

[vinitamaloo-asu]

Hey @alexjo2144 any update on this?

[agnes-xinyi-lu]

Closing this issue since it's implemented as table function by @alexjo2144