feat: Add SQL Support for MERGE INTO in Presto (engine + iceberg connector)

[acarpente-denodo]

Description

Engine support for SQL MERGE command. This command inserts or updates rows in a table based on specified conditions.
Iceberg connector support for the SQL MERGE INTO command.

Syntax:

MERGE INTO target_table [ [ AS ]  target_alias ]
USING { source_table | query } [ [ AS ] source_alias ]
ON search_condition
WHEN MATCHED THEN
    UPDATE SET ( column = expression [, ...] )
WHEN NOT MATCHED THEN
    INSERT [ column_list ]
    VALUES (expression, ...)

Example: MERGE INTO usage to update the sales information for existing products and insert the sales information for the new products in the market.

MERGE INTO product_sales AS s
    USING monthly_sales AS ms
    ON s.product_id = ms.product_id
WHEN MATCHED THEN
    UPDATE SET
        sales = sales + ms.sales
      , last_sale = ms.sale_date
      , current_price = ms.price
WHEN NOT MATCHED THEN
    INSERT (product_id, sales, last_sale, current_price)
    VALUES (ms.product_id, ms.sales, ms.sale_date, ms.price)

The Presto engine commit introduces an enum called RowChangeParadigm, which describes how a connector modifies rows. The iceberg connector will utilize the DELETE_ROW_AND_INSERT_ROW paradigm, as it represents an updated row as a combination of a deleted row followed by an inserted row. The CHANGE_ONLY_UPDATED_COLUMNS paradigm is meant for connectors that support updating individual columns of rows.

Note: Changes were made after reviewing the following Trino PR: trinodb/trino#7126
So, this PR is deeply inspired by Trino's implementation.

Motivation and Context

The MERGE INTO statement is commonly used to integrate data from two tables with different contents but similar structures.
For example, the source table could be part of a production transactional system, while the target table might be located in a data warehouse for analytics.
Regularly, MERGE operations are performed to update the analytics warehouse with the latest production data.
You can also use MERGE with tables that have different structures, as long as you can define a condition to match the rows between them.

Test Plan

Automated tests developed in:

• Engine: TestSqlParser, TestSqlParserErrorHandling, TestStatementBuilder, AbstractAnalyzerTest, TestAnalyzer, and TestClassLoaderSafeWrappers classes.
• Iceberg connector: IcebergDistributedTestBase

Contributor checklist

• Please make sure your submission complies with our contributing guide, in particular code style and commit standards.
• PR description addresses the issue accurately and concisely. If the change is non-trivial, a GitHub Issue is referenced.
• Documented new properties (with its default value), SQL syntax, functions, or other functionality.
• If release notes are required, they follow the release notes guidelines.
• Adequate tests were added if applicable.
• CI passed.

Release Notes

== RELEASE NOTES ==

General Changes
* Add support for the MERGE command in the Presto engine.

Iceberg Connector Changes
* Add support for MERGE command in the Iceberg connector.

[aditi-pandit]

@acarpente-denodo : Thanks for this contribution.

The code will need extensive changes for Presto C++ support. Have you investigated that ?

Did you try this code with a C++ worker (or side-car enabled ) ? We can do the full C++ support as a follow up, but its important the queries fail with a clean error message on Presto C++

[acarpente-denodo]

Hi @aditi-pandit,
Thanks for your feedback. No, I haven't investigated the support for Presto C++ yet. I also haven't tested this code with a C++ worker or with side-car enabled. At this point, I'm still focused on getting the MERGE INTO command fully working on Java. Once that's stable, I will plan for proper C++ support, including ensuring clean error messages when used with Presto C++.

[yingsu00]

Hi Adrian, were you able to build this PR? I got the following errors:

[ERROR] Failed to execute goal org.basepom.maven:duplicate-finder-maven-plugin:1.5.1:check (default) on project presto-main: Found duplicate classes/resources! -> [Help 1]

Also, how did you get the "handle not found" error? Were you running a query using the query runner? Were you able to get the correct plan? Would you please share the plan and full error stack?

[yingsu00]

At this commit, the merge into only supports parsing and not the full functionality. Let's move the documentation to the last commit when it's fully supported.

[acarpente-denodo]

OK, it's done.

[yingsu00]

Keep the original name testMerge()

[acarpente-denodo]

OK, it's done.

[yingsu00]

Hi Adrian, were you able to build this PR? The CI build failed with these:

In file included from /__w/presto/presto/presto-native-execution/./presto_cpp/main/PeriodicHeartbeatManager.h:17:
/__w/presto/presto/presto-native-execution/./presto_cpp/presto_protocol/core/presto_protocol_core.h:1684:3: error: unknown type name 'com'
  com.facebook.presto.spi.MergeHandle handle = {};
  ^
/__w/presto/presto/presto-native-execution/./presto_cpp/presto_protocol/core/presto_protocol_core.h:1684:6: error: expected member name or ';' after declaration specifiers
  com.facebook.presto.spi.MergeHandle handle = {};
  ~~~^
/__w/presto/presto/presto-native-execution/./presto_cpp/presto_protocol/core/presto_protocol_core.h:1686:3: error: unknown type name 'ConnectorMergeTableHandle'; did you mean 'ConnectorDeleteTableHandle'?
  ConnectorMergeTableHandle connectorMergeTableHandle = {};
  ^~~~~~~~~~~~~~~~~~~~~~~~~
  ConnectorDeleteTableHandle
/__w/presto/presto/presto-native-execution/./presto_cpp/presto_protocol/core/presto_protocol_core.h:[313](https://github.com/prestodb/presto/actions/runs/16068848370/job/45348688297?pr=25470#step:17:314):8: note: 'ConnectorDeleteTableHandle' declared here
struct ConnectorDeleteTableHandle : public JsonEncodedSubclass {};
       ^
/__w/presto/presto/presto-native-execution/./presto_cpp/presto_protocol/core/presto_protocol_core.h:1709:3: error: unknown type name 'RowChangeParadigm'
  RowChangeParadigm paradigm = {};
  ^
4 errors generated.

Would you please fix the build errors first?

Also, how did you get the "handle not found" error? Were you running a query using the query runner? Since the build fails I wonder how you could do that. If you could run a query, were you able to get the correct plan? Would you please share the plan and full error stack? Thanks!

[acarpente-denodo]

Hi @yingsu00,

The PR is still in development. I'm still focused on getting the MERGE INTO command fully working on Java. The coordinator and workers are currently running on Java. I haven't investigated the MERGE INTO support for Presto C++ yet.

Automated tests have not been created yet. To test this enhancement, you need to perform a manual test. It is described at the end of the PR description in the section called "Test Plan".

I'm facing an error during the serialization of a new class called MergePartitioningHandle, which implements ConnectorPartitioningHandle interface

Caused by: java.lang.IllegalArgumentException: No connector for handle: MERGE [update = iceberg:INSTANCE]
at com.facebook.presto.metadata.HandleResolver.getId(HandleResolver.java:228)
at com.facebook.presto.metadata.HandleResolver.getId(HandleResolver.java:120)
at com.facebook.presto.metadata.AbstractTypedJacksonModule$InternalTypeResolver.idFromValueAndType(AbstractTypedJacksonModule.java:158)
at com.facebook.presto.metadata.AbstractTypedJacksonModule$InternalTypeResolver.idFromValue(AbstractTypedJacksonModule.java:150)
at com.fasterxml.jackson.databind.jsontype.impl.TypeSerializerBase.idFromValue(TypeSerializerBase.java:92)
at com.fasterxml.jackson.databind.jsontype.impl.TypeSerializerBase._generateTypeId(TypeSerializerBase.java:77)
at com.fasterxml.jackson.databind.jsontype.impl.TypeSerializerBase.writeTypePrefix(TypeSerializerBase.java:45)
at com.fasterxml.jackson.databind.ser.std.BeanSerializerBase.serializeWithType(BeanSerializerBase.java:650)
at com.facebook.presto.metadata.AbstractTypedJacksonModule$InternalTypeSerializer.serialize(AbstractTypedJacksonModule.java:115)
at com.fasterxml.jackson.databind.ser.BeanPropertyWriter.serializeAsField(BeanPropertyWriter.java:732)
at com.fasterxml.jackson.databind.ser.std.BeanSerializerBase.serializeFields(BeanSerializerBase.java:772)
... 100 more

I think there is an issue with com.facebook.presto.iceberg.IcebergUpdateHandle#INSTANCE. You can find a complete description of this problem in the Slack thread I created a few weeks ago: https://prestodb.slack.com/archives/CLYMEEJ6S/p1749466521358289

The goal of this draft PR is to get assistance in fixing this problem.

[steveburnett]

Format nit. Thanks for the doc!

[acarpente-denodo]

Hi @steveburnett! Thank you for reviewing the feature documentation.

[acarpente-denodo]

Hi @acarpente-denodo I want to check in with you and see if you need any help from my team. Please share your ask if you need anything so I can try to get one of our engineers to assist. Thank you.

Hi @ethanyzhang! Thanks for offering your help. The MERGE command currently works on Iceberg tables. I'm currently working on optimizing the workload distribution across the worker nodes. The current implementation does not use the table partitioning information to spread the workload efficiently. I would appreciate it if you could help me implement the MergePartitioningHandle, IcebergNodePartitioningProvider, IcebergBucketFunction, IcebergPartitioning, IcebergPartitioningHandle, etc.

[sourcery-ai]

Sorry @acarpente-denodo, your pull request is larger than the review limit of 150000 diff characters