Support SQL MERGE in the Trino engine and five connectors

[djsagain]

This PR is a second take on implementing SQL MERGE. It consists commits that add support for SQL MERGE in the Trino engine and in the Hive, Kudu, Raptor, Iceberg and Delta Lake connectors. The implementation is structured so that most of the work happens in the Trino engine, so adding support in a connector is pretty simple.

The SQL MERGE implementation allows update of all columns, including partition or bucket columns, and the Trino engine performs redistribution to ensure that the updated rows end up on the appropriate nodes.

The Trino engine commit introduces an enum RowChangeParadigm, which characterizes how a connector modifies rows. Hive uses and Iceberg will use the DELETE_ROW_AND_INSERT_ROW paradigm, since they represent an updated row as a deleted row and an inserted row. Kudu uses the CHANGE_ONLY_UPDATED_COLUMNS paradigm.

Each paradigm corresponds to an implementation of the RowChangeProcessor interface. After this PR is merged, the intent is to retrofit SQL UPDATE to use the same RowChangeParadigm/Processor mechanism.

Extensive documentation on the internal MERGE architecture can be found in the developer doc supporting-merge.rst.

Fixes #7708

[electrum]

The Kudu commit looks good

[djsagain]

Thanks for the great comments, @electrum. I did everything you suggested.

[kasiafi]

A lot of questions and some comments. I've gone through the docs, and partially through the analysis.

[kasiafi]

Some more comments regarding the analyzer. Initial comments on the planner part.

[djsagain]

Thanks for the great first batch of comments, @kasiafi! I believe I've addressed the comments from yesterday except those listed below. It would be great if you could resolve the comments you think have been handled to your satisfaction.

I haven't addressed the more profound comments made 4 hours ago yet, and some of them will require coaching from you or @martint.

Here are the comments from yesterday that I haven't addressed:

• Does DuplicateRowFinder need to compare the writeRedistribution columns?
• Will matched target table rowIds really come out in order such that DuplicateRowFinder is guaranteed to identify them?
• Implementing multiple assignment.
• Addressing your comment: "Instead of assigning a scope to an Identifier, the aliased table should parse as AliasedRelation."
• Addressing your comment: "What if the table was a materialized view?"

[findepi]

re #7933 (comment)

target table rowIds would be partitioned among nodes

@djsstarburst can you please point me to a document outlining how MERGE interacts with connectors?

i would like to learn about the following

• what are the assumption on rowIds, can rowIds carry un-updated columns
• how should a connector construct rowIds if it needs to create deletion delta files for the sake of updates (e.g. a separate deletion file for an input file which would mark all the rows that got updated)
• what is table handle lifecycle for MERGE. for example, how MERGE interacts with partition, file and file chunk pruning

[kasiafi]

Here are some comments regarding the previously reviewed part. Additionally, I answered some of your replies directly. I resolved all conversations except those that require a follow-up.

I plan to review next portions of code, and put my comments in a new batch.

[findepi]

Why RowBlock is special-cased here?
What if underlyingBlock is a DictionaryBlock over a RowBlock? Would it require special-casing as well?

[djsagain]

I had endless trouble with this, and it's one of the main things I hoped review would shed light on.

I had hoped that I could just call rowIdBlock.getPositions(...) and end up with a consistent view of the resulting block. However, when I tried that, way downstream in the Driver I would see out-of-range array references. My assumption is that I'm doing something wrong, but I wasn't successful debugging the problem.

[findepi]

I had endless trouble with this, and it's one of the main things I hoped review would shed light on.

Sorry that i cannot help. Add a TODO comment here, warning the reader we don't exactly know why it's written the way it's written

[findepi]

Shouldn't this actually depend on rowIdType?

also, direct use of ArrayBlock is not correct. Typically you would use io.trino.spi.type.Type#createBlockBuilder(io.trino.spi.block.BlockBuilderStatus, int) to construct a block of values for given type.

Here, however, you actually want to create a single-value NULL block (nativeValueToBlock may be helpful) and wrap it in a RunLengthEncodedBlock instead

[kasiafi]

Some comments and questions regarding the planner part. I still have a few classes to review.

[findepi]

what I have understood, only a rebase to main is missing before pull request will be accepted.

@harlequin not exactly, but hopefully not very far from that. @electrum recently told me there is some correctness issue lingering somewhere, probably on the engine-side of the changes proposed here. Per my understanding David is going to take a stab and hunting it down.

[nicor88]

Any plan to re-work this feature soon? We want to use Trino as the main framework for our ETL, using Iceberg, and having MERGE operations will be indeed awesome, otherwise, we might need to fall back to Spark.

[djsagain]

Today @electrum and I found and fixed the long-standing problem with this PR that caused SQL MERGE tests with lots of modified/deleted rows to fail. The root cause of the failures was using Block.getChildren, and was fixed by using ColumnarRow in two places in the DeleteAndInsertRowProcessor. In the process, @electrum added a Raptor connector implementation of SQL MERGE, which is much easier to debug than the product tests for Hive SQL MERGE.

Here are the tasks I know of to finish off the work on SQL MERGE:

• Replicate the merge tests in TestHiveTransactionalTable for Raptor. This is mostly cut-and-paste/formatting (assuming the tests all pass).
• Eliminate the SPI call ConnectorMetadata.getWriteRedistributionColumnHandles - - replacing it with a call to the existing getInsertLayout method.
• Add the new SPI call ConnectorMetadata.getUpdateInsertLayout, or whatever we decide to call it. This layout is needed for unbucketed Raptor and Iceberg SQL MERGE operations, according to @electrum.
• Make sure that the internal developer documentation for SQL MERGE reflects reality.

[sopel39]

@djsstarburst Will merge work similar to INSERTS, e.g. allow complex plans that have UNIONS, joins or aggregations?

[djsagain]

Will merge work similar to INSERTS, e.g. allow complex plans that have UNIONS, joins or aggregations?

Hi Karol. Yes, I believe so. You can see the new tests in that PR in TestHiveTransactionalTable. The test names all start with testMerge.

If you see something untested, or have a test to propose, please send it my way.

[djsagain]

Eliminate the SPI call ConnectorMetadata.getWriteRedistributionColumnHandles - - replacing it with a call to the existing getInsertLayout method.

This is done - - getWriteRedistributionColumns is no longer mentioned in either the code or the developer documentation.

[nicor88]

@djsstarburst will this feature cover also DeltaLake connector?

[sopel39]

don't you need to add exchanges below MergeProcessorNode?

[electrum]

I don't understand the question, but the produced plan seems to be correct. We have these stages in the plan:

(scan[target], scan[source]) -> RightJoin -> MergeProcessor -> MergeWriter -> TableCommit

Take a look at the supporting-merge.rst which might help explain the structure.

[sopel39]

How do you make sure RightJoin does not participate in CBO and doesn't get reordered or flipped?

[dain]

Why would that matter? If the CBO is working properly it should not change the output of the join operation, so if it has a better plan that should be ok. right?

[dain]

Looks good to me

[dain]

Maybe a better message here

[djsagain]

Changed to:

The target table in Hive MERGE must be a transactional table

[dain]

Why would that matter? If the CBO is working properly it should not change the output of the join operation, so if it has a better plan that should be ok. right?

[dain]

We should mention what happens if not clauses match the row and there is no default WHEN MATCHED clause

[djsagain]

Added:

If a source row is not matched by any ``WHEN`` clause and there is no
``WHEN NOT MATCHED` clause, the source row is ignored.