feat: Add full-featured Iceberg MERGE INTO conditional merges support and argument validation #3201

pedrorfdez · 2025-09-10T18:40:08Z

Feature or Bugfix

Feature

Detail

Implemented full support for Iceberg MERGE INTO operations in _write_iceberg.py/to_iceberg
Added new parameters for:
- merge_on_clause: Custom ON statement in the MERGE INTO ... USING ... ON [custom_expression] to allow <, <=, > and >= operators. Until now, only column equality was allowed. Risk of having more than one match in target table is warned in stringdocs.
- merge_condition: Added new accepted value conditional_merge
- merge_conditional_clauses: List of dictionaries specifying custom conditional clauses for the MERGE INTO statement.
  Each dictionary should have:
  - 'when': One of ['MATCHED', 'NOT MATCHED', 'NOT MATCHED BY SOURCE']
  - 'action': One of ['UPDATE', 'DELETE', 'INSERT']
  - 'condition': (optional) Additional SQL condition for the clause
  - 'columns': (optional) List of columns to update or insert
  Used only when merge_condition is 'conditional_merge'.
Added argument validation for mutually exclusive and required parameters merge_cols, merge_on_clause, merge_match_nulls, merge_condition, merge_conditional_clauses,
Added and updated unit tests to cover new validation logic and merge scenarios.
Updated docstrings for new parameters and behaviors.
Backward compatibility

Relates

This is the first draft of the implementation, feel free to suggest any changes in the approach. I am open to suggestions.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

pedrorfdez · 2025-09-10T19:12:37Z

Currently failing 1 Athena test:

FAILED tests/unit/test_athena_iceberg.py::test_to_iceberg_conditional_merge_happy_path - AssertionError: Attributes of DataFrame.iloc[:, 0] (column name="id") are different

Attribute "dtype" are different
[left]: int64
[right]: Int64

Plan to fix on next commit.

…ality

jaidisido · 2025-10-30T13:45:00Z

AWS CodeBuild CI Report

CodeBuild project: GitHubCodeBuild8756EF16-4rfo0GHQ0u9a
Commit ID: e7acaf0
Result: FAILED
Build Logs (available for 30 days)

Powered by github-codebuild-logs, available on the AWS Serverless Application Repository

kukushking · 2025-10-30T13:45:56Z

Hi @pedrorfdez it looks like there is an error in test_to_iceberg_conditional_merge_happy_path:

FAILED tests/unit/test_athena_iceberg.py::test_to_iceberg_conditional_merge_happy_path - AssertionError: Attributes of DataFrame.iloc[:, 0] (column name="id") are different
--
 
Attribute "dtype" are different
[left]:  int64
[right]: Int64
= 1 failed, 2312 passed, 72 skipped, 9 xfailed, 4 xpassed, 143 warnings, 1 rerun in 729.43s (0:12:09) =

jaidisido · 2025-10-30T14:47:17Z

AWS CodeBuild CI Report

CodeBuild project: GitHubDistributedCodeBuild6-jWcl5DLmvupS
Commit ID: e7acaf0
Result: FAILED
Build Logs (available for 30 days)

Powered by github-codebuild-logs, available on the AWS Serverless Application Repository

inbar-beehero · 2025-11-03T15:39:25Z

awswrangler/athena/_write_iceberg.py

+            "Cannot specify both merge_cols and merge_on_clause. Use either merge_cols for simple equality matching or merge_on_clause for custom logic."
+        )
+
+    if merge_on_clause and merge_match_nulls:


Possibly better to also check if merge_cols - as it will validate that merge cols are actually set

inbar-beehero · 2025-11-03T15:43:18Z

awswrangler/athena/_write_iceberg.py

+                raise exceptions.InvalidArgumentValue(f"merge_conditional_clauses[{i}] must contain 'action' field.")
+            if clause["when"] not in ["MATCHED", "NOT MATCHED", "NOT MATCHED BY SOURCE"]:
+                raise exceptions.InvalidArgumentValue(
+                    f"merge_conditional_clauses[{i}]['when'] must be one of ['MATCHED', 'NOT MATCHED', 'NOT MATCHED BY SOURCE']."


Maybe indicate in the error message that the operators are case sensitive

inbar-beehero · 2025-11-03T15:43:40Z

awswrangler/athena/_write_iceberg.py

+                )
+            if clause["action"] not in ["UPDATE", "DELETE", "INSERT"]:
+                raise exceptions.InvalidArgumentValue(
+                    f"merge_conditional_clauses[{i}]['action'] must be one of ['UPDATE', 'DELETE', 'INSERT']."


Maybe indicate in the error message that the action is case sensitive

inbar-beehero · 2025-11-03T15:47:54Z

awswrangler/athena/_write_iceberg.py

+        - 'update': Update matched rows and insert non-matched rows.
+        - 'ignore': Only insert non-matched rows.
+        - 'conditional_merge': Use custom conditional clauses for merge actions.
+    merge_conditional_clauses : List[dict], optional


Shouldn't the type hint be list[_MergeClause]?

inbar-beehero · 2025-11-03T15:54:52Z

awswrangler/athena/_write_iceberg.py

-                UPDATE SET {", ".join([f'"{x}" = source."{x}"' for x in df.columns])}"""
-        else:
-            match_condition = ""
+    if merge_cols or merge_on_clause:


The if code becomes unwieldy imho, maybe refactor to a helper function to formulate the conditions

pedrorfdez added 9 commits September 8, 2025 01:50

Added necesary parameters to 'to_iceberg' method to expand functionality

4e4936b

Removed unused import

1f56c72

Added validation for new parameters complying with new logic

14f3194

Draft implementation done. Pending tests and verify optimal approach

12802ad

Added additional validation and removed unhandled action.

991f610

Added new params to python docstrings

9b8cbf4

Fixed backslash in f-string

31d1a1d

Added tests

7685bc1

Updated stringdocs

eb25d2a

pedrorfdez changed the title ~~Add full-featured Iceberg MERGE INTO conditional merges support and argument validation~~ feat: Add full-featured Iceberg MERGE INTO conditional merges support and argument validation Sep 10, 2025

fix: ruff reformat

262bfc6