Fix Iceberg merge, update, delete, for tables with equality deletes #24062

dain · 2024-11-07T20:52:25Z

Description

Rewrite the creation of merge row id to avoid duplicate key exception.
Simplify by having data source produce row position for row id and then wrappering this block into a row id in IcebergPageSource
Remove unused update row id. This should have be removed during convertion to merge row ids.

Fixes #15952
Superseeds #16216

Release notes

(x) Release notes are required, with the following suggested text:

## Section
* Fix Iceberg merge, update, delete, for tables with equality deletes. ({issue}`15952`)

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

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

findinpath · 2024-11-08T14:02:46Z

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

+        public RowBlock apply(Block rowPosition)
+        {
+            return RowBlock.fromFieldBlocks(rowPosition.getPositionCount(), new Block[] {
+                    RunLengthEncodedBlock.create(filePath, rowPosition.getPositionCount()),


Pre-existing

I am wondering whether the components (_file, _pos, partition_spec_id, partition_data) can be actually considered the row identifier.

With add_files we can add exactly the same file twice in an Iceberg table.

When doing the sequence of operations:

add_file('myfile')

add equality delete affecting the file 'myfile'

add_file('myfile')

The second addition of 'myfile' is affected by the previous equality delete applied on 'myfile'

cc @ebyhr

That seems like a bug in add_files. IMO, we should disallow that as I would guess there are other things that depend on files being unique.

findinpath · 2024-11-08T14:41:45Z

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

@@ -1149,19 +1091,14 @@ else if (column.isFileModifiedTimeColumn()) {
                    constantPopulatingPageSourceBuilder.addConstantColumn(nativeValueToBlock(FILE_MODIFIED_TIME.getType(), packDateTimeWithZone(fileModifiedTime.orElseThrow(), UTC_KEY)));
                }
                // For delete
-                else if (column.isRowPositionColumn()) {
+                else if (column.isMergeRowIdColumn() || column.isRowPositionColumn()) {


i find this mixing a bit misleading to read.
(unfortunately i don't have a better suggestion)
maybe similarly to what is done for parquet and orc with a duplicated if

else if (column.isMergeRowIdColumn()) { // The merge $row_id is a composite of the row position and constant file information. The final value is assembled in IcebergPageSource pageSourceBuilder.addRowIndexColumn(); } else if (column.isRowPositionColumn()) { pageSourceBuilder.addRowIndexColumn(); }

Don't worry about it . I am mostly rewriting this class in this PR

findinpath · 2024-11-08T14:47:28Z

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

@@ -880,7 +880,7 @@ private static ReaderPageSourceWithRowPositions createParquetPageSource(

            List<org.apache.parquet.schema.Type> parquetFields = readBaseColumns.stream()
                    .map(column -> parquetIdToField.get(column.getId()))
-                    .collect(toList());
+                    .toList();


Suggested change

.toList();

.collect(toImmutableList());

This can't use immutable list because this list contains nulls.

findinpath · 2024-11-08T14:50:39Z

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergV2.java

+        assertThat(icebergTable.currentSnapshot().summary()).containsEntry("total-equality-deletes", "0");
+        writeEqualityDeleteToNationTable(icebergTable);
+        assertThat(icebergTable.currentSnapshot().summary()).containsEntry("total-equality-deletes", "1");
+        assertUpdate("UPDATE " + tableName + " SET comment = 'test'", 20);


Confirming here that the test reproduces the fixed problem.

Stacktrace of the test without the other changes in this PR:

io.trino.testing.QueryFailedException: Multiple entries with same key: 2147483645=$row_id._pos and 2147483645=_pos at io.trino.testing.AbstractTestingTrinoClient.execute(AbstractTestingTrinoClient.java:138) at io.trino.testing.DistributedQueryRunner.executeInternal(DistributedQueryRunner.java:565) at io.trino.testing.DistributedQueryRunner.executeWithPlan(DistributedQueryRunner.java:554) at io.trino.testing.QueryAssertions.assertDistributedUpdate(QueryAssertions.java:106) at io.trino.testing.QueryAssertions.assertUpdate(QueryAssertions.java:60) at io.trino.testing.AbstractTestQueryFramework.assertUpdate(AbstractTestQueryFramework.java:420) at io.trino.testing.AbstractTestQueryFramework.assertUpdate(AbstractTestQueryFramework.java:415) at io.trino.plugin.iceberg.TestIcebergV2.testUpdateAfterEqualityDelete(TestIcebergV2.java:1439) at java.base/java.lang.reflect.Method.invoke(Method.java:580) at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:507) at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1458) at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:2034) at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:189) Suppressed: java.lang.Exception: SQL: UPDATE test_update_after_equality_delete_y36815gpdg SET comment = 'test' at io.trino.testing.DistributedQueryRunner.executeInternal(DistributedQueryRunner.java:572) ... 11 more Caused by: io.trino.spi.TrinoException: Multiple entries with same key: 2147483645=$row_id._pos and 2147483645=_pos at io.trino.plugin.iceberg.IcebergPageSource.getNextPage(IcebergPageSource.java:147) at io.trino.operator.ScanFilterAndProjectOperator$ConnectorPageSourceToPages.process(ScanFilterAndProjectOperator.java:381)

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergV2.java

Update row id was replaced with merge row id

Rewrite the creation of merge row id to avoid duplicate key exception. This also simplifies and consolidates the merge row id code.

findinpath · 2024-11-11T10:47:23Z

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

-    public static final int TRINO_MERGE_PARTITION_SPEC_ID = Integer.MIN_VALUE + 2;
-    public static final int TRINO_MERGE_PARTITION_DATA = Integer.MIN_VALUE + 3;
+    public static final int TRINO_MERGE_PARTITION_SPEC_ID = Integer.MIN_VALUE + 1;
+    public static final int TRINO_MERGE_PARTITION_DATA = Integer.MIN_VALUE + 2;

    public static final String DATA_CHANGE_TYPE_NAME = "_change_type";


Pre-existing / nit : it's not clear to me why did we have the gap between Integer.MIN_VALUE + 3 and Integer.MIN_VALUE + 5

cla-bot bot added the cla-signed label Nov 7, 2024

github-actions bot added the iceberg Iceberg connector label Nov 7, 2024

dain force-pushed the update-after-equality-delete branch 2 times, most recently from b87430a to a080a7a Compare November 7, 2024 21:54

ebyhr mentioned this pull request Nov 7, 2024

Fix iceberg v2 table with equality delete can't update #16216

Closed

ebyhr requested review from ebyhr, Heltman and findinpath November 7, 2024 22:51

dain force-pushed the update-after-equality-delete branch from a080a7a to dddde02 Compare November 7, 2024 22:54

findinpath reviewed Nov 8, 2024

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergColumnHandle.java Outdated Show resolved Hide resolved

findinpath reviewed Nov 8, 2024

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSource.java Show resolved Hide resolved

findinpath reviewed Nov 8, 2024

View reviewed changes

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergV2.java Show resolved Hide resolved

dain added 2 commits November 9, 2024 13:16

Remove unused Iceberg update row id

ebc35f1

Update row id was replaced with merge row id

Fix Iceberg merge, update, delete, for tables with equality deletes

2607ab4

Rewrite the creation of merge row id to avoid duplicate key exception. This also simplifies and consolidates the merge row id code.

dain force-pushed the update-after-equality-delete branch from dddde02 to 89482d6 Compare November 9, 2024 21:27

Fix IntelliJ warnings in IcebergPageSourceProvider

7a06693

dain force-pushed the update-after-equality-delete branch from 89482d6 to 7a06693 Compare November 9, 2024 22:16

dain mentioned this pull request Nov 11, 2024

Add SourcePage interface for delayed materialization of ConnectorSourceData #24011

Open

findinpath approved these changes Nov 11, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix Iceberg merge, update, delete, for tables with equality deletes #24062

Fix Iceberg merge, update, delete, for tables with equality deletes #24062

dain commented Nov 7, 2024

findinpath Nov 8, 2024

dain Nov 8, 2024

findinpath Nov 8, 2024 •

edited

Loading

dain Nov 9, 2024

findinpath Nov 8, 2024

dain Nov 9, 2024

findinpath Nov 8, 2024 •

edited

Loading

findinpath Nov 11, 2024

Fix Iceberg merge, update, delete, for tables with equality deletes #24062

Are you sure you want to change the base?

Fix Iceberg merge, update, delete, for tables with equality deletes #24062

Conversation

dain commented Nov 7, 2024

Description

Release notes

findinpath Nov 8, 2024

Choose a reason for hiding this comment

dain Nov 8, 2024

Choose a reason for hiding this comment

findinpath Nov 8, 2024 • edited Loading

Choose a reason for hiding this comment

dain Nov 9, 2024

Choose a reason for hiding this comment

findinpath Nov 8, 2024

Choose a reason for hiding this comment

dain Nov 9, 2024

Choose a reason for hiding this comment

findinpath Nov 8, 2024 • edited Loading

Choose a reason for hiding this comment

findinpath Nov 11, 2024

Choose a reason for hiding this comment

findinpath Nov 8, 2024 •

edited

Loading

findinpath Nov 8, 2024 •

edited

Loading