Enable failure recovery for Iceberg connector

[losipiuk]

Caveat:

With current logic it is possible that data file written during DML operation, which in the end is not part of the committed table snapshot, remains in the table directory on the distributed filesystem.
It should be a rare situation.

• If task which writes data file fails unfinished file is deleted by tasks itself.
• If there are two concurrently running attempts for a single task, when one completes, the other is killed. Yet killing is subject to race and if both tasks manage to complete successfully the extraneous file will remain in table directory.

Orphaned files can be clean via remove_orphan_files routing using Spark (https://iceberg.apache.org/#spark-procedures/).
Eventuall we want to have similar routine in Trino (#10623)

fixes: #10253

[findepi]

closeSuppressing

[findepi]

AbstractTest is an old naming convention, these days we call them Base...Test

[losipiuk]

Extracted to #10659

[findepi]

Sounds like you're working against testTableModification abstraction.

[losipiuk]

Yeah - I guess we do not need this test at all here. It would make sense to document behaviour difference for Iceberg if test was part of superclass.
I will drop for now, but it should be moved to superclass and we should start using TestingConnectorBehavior in super class

[findepi]

newline before .has...

[findepi]

nit: grammar is off

why not enable it for now here?
when is it going to be removed?

[losipiuk]

I will add it for now here. As extracting it out from Base class requires significant restructuring.

As for timeline: 🤷

[findepi]

Orphaned files can be clean via remove_orphan_files routing using Spark (https://iceberg.apache.org/#spark-procedures/).
Eventuall we want to have similar routine in Trino (#10623)

While i agree we should implement such a procedure, failure recovery should not leave garbage behind, in situations when this can be avoided.

Iceberg writes files with random names. The file name is determined here

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

Line 305 in 863da2b

String fileName = fileFormat.addExtension(randomUUID().toString());

we should consider

• add queryId to file name
• in io.trino.plugin.iceberg.IcebergMetadata#finishInsert, etc, when failure recovery is enabled, go over table destination folder (as determined by LocationProvider) and find left over files
  • we could do this only if there were some failures (but currently finish... doesn't know that)
  • we could try to limit ourselves to directories where we write some files -- this way we wouldn't scan all the sub-directories of a partitioned table, but only ones where we insert data into
    • this will work when LocationProvider is deterministic. for non-deterministic LocationProvider we would need quite different approach (eg registering write locations before creating a file), as remove_orphan_files-like procedure might have problems as well

cc @phd3 @alexjo2144

[findepi]

Current state of the PR - LGTM % comments.

#10622 (comment) fits well as a followup.

[losipiuk]

Orphaned files can be clean via remove_orphan_files routing using Spark (https://iceberg.apache.org/#spark-procedures/).
Eventuall we want to have similar routine in Trino (#10623)

While i agree we should implement such a procedure, failure recovery should not leave garbage behind, in situations when this can be avoided.

Iceberg writes files with random names. The file name is determined here

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

Line 305 in 863da2b

String fileName = fileFormat.addExtension(randomUUID().toString());

we should consider

• add queryId to file name

• in io.trino.plugin.iceberg.IcebergMetadata#finishInsert, etc, when failure recovery is enabled, go over table destination folder (as determined by LocationProvider) and find left over files

  • we could do this only if there were some failures (but currently finish... doesn't know that)

  • we could try to limit ourselves to directories where we write some files -- this way we wouldn't scan all the sub-directories of a partitioned table, but only ones where we insert data into

    • this will work when LocationProvider is deterministic. for non-deterministic LocationProvider we would need quite different approach (eg registering write locations before creating a file), as remove_orphan_files-like procedure might have problems as well

cc @phd3 @alexjo2144

@findepi I added some code for that. PTAL and tell me what you think.

[findepi]

"Cleanup extranous output files in Iceberg DML"

[findepi]

could the RetryMode be provided by the engine to the finish*() methods, so that we don't have to embed the info in handles?

[losipiuk]

We could - but I would opt for what we do now. It feels more natural - as we know information upfront. Also it allows for rejecting the request sooner if given retry mode is not supported by a connector.

[findepi]

allows for rejecting the request sooner if given retry mode is not supported by a connector.

i did not suggest not to provide it to begin methods.

[losipiuk]

I would be asymetric vs other "stuff" we pass to "begin*" methods (layout, list of columns). I will leave it as is.

[findepi]

maybe add a comment that one query id be cannot be a prefix of another query id

[losipiuk]

That is a good point actually.

Changed to

        verify(!queryId.contains("-"), "queryId(%s) should not contain hyphens", queryId);
        return fileName.startsWith(queryId + "-");

[losipiuk]

CI flake: #10631