Expire Snapshot and Remove Orphan files for Iceberg together with coordinator-only execute

[homar]

Two new procedures are added to Iceberg connector:

1. REMOVE_ORPHAN_FILES - removes files from table folder that don't belong to any tables and that are older than minimum retention
2. EXPIRE_SNAPSHOTS - expires snapshots older than provided minimum retention

Trino lacked infrastructure to run a specific task on a single node tough the necessity for such a behaviour was predicted when table execute was introduced.
This PR adds necessary changes to run tasks on coordinator only.

[electrum]

Import directly

[electrum]

The "else" is redundant since the if block returns

[electrum]

"else" is redundant

[electrum]

Format intentionally empty methods like

public void finish() {}

[electrum]

Check for null

[electrum]

Not needed

[electrum]

expireTimestamp

[electrum]

table.expireSnapshots()
        .expireOlderThan(expireTimestamp)
        .cleanExpiredFiles(true)
        .commit();

[electrum]

Move to previous line

[electrum]

We normally do throws Exception for test methods that throw any checked exceptions

[electrum]

This is flaky, as it relies on the INSERT, SELECTs and EXECUTE all running in under 1 second. Or if we don't care, then there is no need to sleep or have a threshold greater than zero.

[homar]

That is not correct.
The sleep is there to make sure at least 1 second passed after we inserted first row.

[losipiuk]

Can you just drop the sleep and make retention 0?

[losipiuk]

It feels to me output schema of all ALTER TABLE ... EXECUTE statements should be the same. So I would make it single BIGINT column and return no rows.

[losipiuk]

I do not think we need status. Either operation completes correctly. Or we get an error, which is passed via exception.

[losipiuk]

no need for separate class - you can inline fields imo

[losipiuk]

Name is cumbersome. Maybe NonReadingTableExecuteNode?

[findepi]

continued at #10810 (comment)

[losipiuk]

node.getTarget does not have toStriong - but I suggest to drop the class altogether

[losipiuk]

You have analysis.getTableExecuteHandle().orElseThrow() assigned to variable one line above

[losipiuk]

"Remove only files older than threshold"

[findepi]

continued at #10810 (comment)

[losipiuk]

First pass done

[findinpath]

can you add a check at the file level to see whether the vacuum operation was effective?

You could use $files metadata table (while the snapshot to be expired is still active) to gather the file(s) relevant for the snapshot.
You could use $snapshots / $history metadata tables to actually check that the snapshot was expired.
Subsequently you could check at the file level that the files you previously recorded are also removed.

[homar]

I added counting physical files as it is more reliable

[findinpath]

either here or in a follow-up PR it would be beneficial to have documentation for this newly introduced operation.

[findinpath]

I am having a hard time to understand what readsData is about and why the method io.trino.spi.connector.TableProcedureExecutionMode#coordinatorOnly uses readsData set to false

[homar]

it is removed already from here

[electrum]

Let's make the constructor private and only have static factories. The boolean arguments are confusing, especially because certain combinations are not valid.

[electrum]

You can add Javadoc to the coordinatorOnly() factory:

Table procedure that does not read any table data and only executes on the coordinator.
Such procedures are useful for custom DDL-type operations.

[electrum]

I find the name distributedWithFilteringAndRepartitioning() confusing. Would supportsFiltering=false also do repartitioning? More generally, how does filtering relate to repartitioning?

[findinpath]

separate commit?

[findinpath]

is the field name still consistent with the type?

[findinpath]

is the javadoc change intentional?

[homar]

@losipiuk I addressed all the comments, improved existing test and added a new one

[findepi]

URI.getPath is bad conversion (lossy), is it file:// URI?

Path.of(tableDataDir) should suffice

[homar]

This is for local file system. Path is from java.nio. It doesn't work with protocol I've just checked

[homar]

I've added 2 tests: testTrinoReadsTrinoTableWithSparkDeletesAfterOptimizeAndCleanUp and testCleaningUpIcebergTableFailsForTableV2.
First checks if performing a delete doesn't break the table. It looks like fine. Tough I found that spark has some issues when queries are run both on spark and on trino - as this is not related to this change at all I created separate pr to show the problem here #11891.
Second test makes sure that neither EXPIRE_SNAPSHOT nor DELETE_OPRHAN_FILES works do any work on the table with iceberg version 2.
@findepi @alexjo2144 WDYT ?

[synhershko]

But does it make sense to let the coordinator (which should coordinate Trino tasks) execute actual data or metadata mutations? wouldn't it be more in-line with Trino's design to just create a task and assign it to a single worker node?

[homar]

But does it make sense to let the coordinator (which should coordinate Trino tasks) execute actual data or metadata mutations? wouldn't it be more in-line with Trino's design to just create a task and assign it to a single worker node?

@synhershko Take a look at this comment #10810 (comment)

[findepi]

@homar please rebase, there are conflicts.

[homar]

@findepi done

[findepi]

Add a comment why we decided to use URI.getPath only here. Perhaps a reference to iceberg issue.

[alexjo2144]

Need to update the name

[kekwan]

Does expire_snapshots execute in parallel like the Spark version?

[homar]

Does expire_snapshots execute in parallel like the Spark version?

Currently it is not parallel. It may change in the future