From 9f1603d1dd9d1ef09772c38f2d2cc82312c195a8 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 6 Apr 2022 15:02:14 -0600 Subject: [PATCH 01/16] Flink - Disallow upsert mode for Flink 1.12 due to correctness issues --- .../apache/iceberg/flink/sink/FlinkSink.java | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index 24e23941f6d7..fb84e9ec4b6b 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -375,19 +375,13 @@ private SingleOutputStreamOperator appendWriter(DataStream boolean upsertMode = upsert || PropertyUtil.propertyAsBoolean(table.properties(), UPSERT_ENABLED, UPSERT_ENABLED_DEFAULT); - // Validate the equality fields and partition fields if we enable the upsert mode. + // `upsert` mode is not allowed in Flink 1.12 due to correctness issues. + // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) { - Preconditions.checkState(!overwrite, - "OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream."); - Preconditions.checkState(!equalityFieldIds.isEmpty(), - "Equality field columns shouldn't be empty when configuring to use UPSERT data stream."); - if (!table.spec().isUnpartitioned()) { - for (PartitionField partitionField : table.spec().fields()) { - Preconditions.checkState(equalityFieldIds.contains(partitionField.sourceId()), - "In UPSERT mode, partition field '%s' should be included in equality fields: '%s'", - partitionField, equalityFieldColumns); - } - } + throw new UnsupportedOperationException( + "upsert mode is not supported in Flink 1.12 due to correctness issues. Please upgrade to Flink 1.13+ if " + + "upsert mode is needed. If `upsert` mode is not needed, set the table's `write.upsert.enabled` " + + "property to 'false'."); } IcebergStreamWriter streamWriter = createStreamWriter(table, flinkRowType, equalityFieldIds, upsertMode); From afbfd657855caa8adc6e88cc229ade12701a492a Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Mon, 11 Apr 2022 15:17:54 -0700 Subject: [PATCH 02/16] Use proper sentence structure --- .../main/java/org/apache/iceberg/flink/sink/FlinkSink.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index fb84e9ec4b6b..c007c3c4556c 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -379,8 +379,8 @@ private SingleOutputStreamOperator appendWriter(DataStream // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) { throw new UnsupportedOperationException( - "upsert mode is not supported in Flink 1.12 due to correctness issues. Please upgrade to Flink 1.13+ if " + - "upsert mode is needed. If `upsert` mode is not needed, set the table's `write.upsert.enabled` " + + "Upsert mode is not supported in Flink 1.12 due to correctness issues. Please upgrade to Flink 1.13+ if " + + "upsert mode is needed. If upsert mode is not needed, set the table property `write.upsert.enabled` " + "property to 'false'."); } From a0319f6c44a44a06ee8987df5d21b62cab4ed667 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Mon, 11 Apr 2022 15:56:45 -0700 Subject: [PATCH 03/16] Remove upsert related tests in Flink 1.12 and add Assume statement that we're not in upsert --- .../flink/sink/TestFlinkIcebergSinkV2.java | 123 ++---------------- 1 file changed, 12 insertions(+), 111 deletions(-) diff --git a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java index 23169d1a59c6..c2401e539272 100644 --- a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java +++ b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java @@ -53,6 +53,7 @@ import org.apache.iceberg.types.Types; import org.apache.iceberg.util.StructLikeSet; import org.junit.Assert; +import org.junit.Assume; import org.junit.Before; import org.junit.ClassRule; import org.junit.Test; @@ -159,6 +160,8 @@ private void testChangeLogs(List equalityFieldColumns, boolean insertAsUpsert, List> elementsPerCheckpoint, List> expectedRecordsPerCheckpoint) throws Exception { + Assume.assumeFalse("Upsert mode is not supported in Flink 1.12", insertAsUpsert); + DataStream dataStream = env.addSource(new BoundedTestSource<>(elementsPerCheckpoint), ROW_TYPE_INFO); FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA) @@ -376,118 +379,16 @@ public void testChangeLogOnSameKey() throws Exception { } @Test - public void testUpsertModeCheck() throws Exception { + public void testUpsertModeIsDisabled() throws Exception { DataStream dataStream = env.addSource(new BoundedTestSource<>(ImmutableList.of()), ROW_TYPE_INFO); - FlinkSink.Builder builder = FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA) - .tableLoader(tableLoader) - .tableSchema(SimpleDataUtil.FLINK_SCHEMA) - .writeParallelism(parallelism) - .upsert(true); - - AssertHelpers.assertThrows("Should be error because upsert mode and overwrite mode enable at the same time.", - IllegalStateException.class, "OVERWRITE mode shouldn't be enable", - () -> builder.equalityFieldColumns(ImmutableList.of("id", "data")).overwrite(true).append() - ); - - AssertHelpers.assertThrows("Should be error because equality field columns are empty.", - IllegalStateException.class, "Equality field columns shouldn't be empty", - () -> builder.equalityFieldColumns(ImmutableList.of()).overwrite(false).append() - ); - } - - @Test - public void testUpsertOnIdKey() throws Exception { - List> elementsPerCheckpoint = ImmutableList.of( - ImmutableList.of( - row("+I", 1, "aaa"), - row("+U", 1, "bbb") - ), - ImmutableList.of( - row("+I", 1, "ccc") - ), - ImmutableList.of( - row("+U", 1, "ddd"), - row("+I", 1, "eee") - ) - ); - - List> expectedRecords = ImmutableList.of( - ImmutableList.of(record(1, "bbb")), - ImmutableList.of(record(1, "ccc")), - ImmutableList.of(record(1, "eee")) - ); - - if (!partitioned) { - testChangeLogs(ImmutableList.of("id"), row -> row.getField(ROW_ID_POS), true, - elementsPerCheckpoint, expectedRecords); - } else { - AssertHelpers.assertThrows("Should be error because equality field columns don't include all partition keys", - IllegalStateException.class, "should be included in equality fields", - () -> { - testChangeLogs(ImmutableList.of("id"), row -> row.getField(ROW_ID_POS), true, - elementsPerCheckpoint, expectedRecords); - return null; - }); - } - } - - @Test - public void testUpsertOnDataKey() throws Exception { - List> elementsPerCheckpoint = ImmutableList.of( - ImmutableList.of( - row("+I", 1, "aaa"), - row("+I", 2, "aaa"), - row("+I", 3, "bbb") - ), - ImmutableList.of( - row("+U", 4, "aaa"), - row("-U", 3, "bbb"), - row("+U", 5, "bbb") - ), - ImmutableList.of( - row("+I", 6, "aaa"), - row("+U", 7, "bbb") - ) - ); - - List> expectedRecords = ImmutableList.of( - ImmutableList.of(record(2, "aaa"), record(3, "bbb")), - ImmutableList.of(record(4, "aaa"), record(5, "bbb")), - ImmutableList.of(record(6, "aaa"), record(7, "bbb")) - ); - - testChangeLogs(ImmutableList.of("data"), row -> row.getField(ROW_DATA_POS), true, - elementsPerCheckpoint, expectedRecords); - } - - @Test - public void testUpsertOnIdDataKey() throws Exception { - List> elementsPerCheckpoint = ImmutableList.of( - ImmutableList.of( - row("+I", 1, "aaa"), - row("+U", 1, "aaa"), - row("+I", 2, "bbb") - ), - ImmutableList.of( - row("+I", 1, "aaa"), - row("-D", 2, "bbb"), - row("+I", 2, "ccc") - ), - ImmutableList.of( - row("+U", 1, "bbb"), - row("-U", 1, "ccc"), - row("-D", 1, "aaa") - ) - ); - - List> expectedRecords = ImmutableList.of( - ImmutableList.of(record(1, "aaa"), record(2, "bbb")), - ImmutableList.of(record(1, "aaa"), record(2, "ccc")), - ImmutableList.of(record(1, "bbb"), record(2, "ccc")) - ); - - testChangeLogs(ImmutableList.of("id", "data"), row -> Row.of(row.getField(ROW_ID_POS), row.getField(ROW_DATA_POS)), - true, elementsPerCheckpoint, expectedRecords); + AssertHelpers.assertThrows("Upsert mode is not supported in Flink 1.12", + UnsupportedOperationException.class, "Upsert mode is not supported in Flink 1.12", + () -> FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA) + .tableLoader(tableLoader) + .tableSchema(SimpleDataUtil.FLINK_SCHEMA) + .writeParallelism(parallelism) + .upsert(true) + .append()); } private StructLikeSet expectedRowSet(Record... records) { From 27bc9c5c90cf0ba990fced9120dc6baa1cb0b866 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Mon, 11 Apr 2022 15:58:50 -0700 Subject: [PATCH 04/16] Use Assert and not Assume so we don't skip tests that have upsert enabeld in them --- .../org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java index c2401e539272..e0685dbb2227 100644 --- a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java +++ b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java @@ -53,7 +53,6 @@ import org.apache.iceberg.types.Types; import org.apache.iceberg.util.StructLikeSet; import org.junit.Assert; -import org.junit.Assume; import org.junit.Before; import org.junit.ClassRule; import org.junit.Test; @@ -160,7 +159,7 @@ private void testChangeLogs(List equalityFieldColumns, boolean insertAsUpsert, List> elementsPerCheckpoint, List> expectedRecordsPerCheckpoint) throws Exception { - Assume.assumeFalse("Upsert mode is not supported in Flink 1.12", insertAsUpsert); + Assert.assertFalse("Upsert mode is not supported in Flink 1.12", insertAsUpsert); DataStream dataStream = env.addSource(new BoundedTestSource<>(elementsPerCheckpoint), ROW_TYPE_INFO); From 133070871262dc8f55bb56c599047ed67fd8c034 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 19:35:29 -0700 Subject: [PATCH 05/16] Revert V1 FlinkSink from throwing to logging an error message --- .../apache/iceberg/flink/sink/FlinkSink.java | 30 +++++++++++++++---- 1 file changed, 25 insertions(+), 5 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index c007c3c4556c..e271a1766911 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -375,13 +375,33 @@ private SingleOutputStreamOperator appendWriter(DataStream boolean upsertMode = upsert || PropertyUtil.propertyAsBoolean(table.properties(), UPSERT_ENABLED, UPSERT_ENABLED_DEFAULT); - // `upsert` mode is not allowed in Flink 1.12 due to correctness issues. + // `upsert` mode should not be used in Flink 1.12 due to correctness issues. + // As part of a patch release, apachee-iceberg-flink-runtime_1.12:0.13.2, + // it has been decided the best course of action would be to log a warning + // asking people to upgrade, as Flink 1.12 is deprecated from both upstream + // Flink as well as Iceberg as of Iceberg 0.14.0. But we allow the configuration + // given that it's a patch release and the change would be otherwise very breaking. + // // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) { - throw new UnsupportedOperationException( - "Upsert mode is not supported in Flink 1.12 due to correctness issues. Please upgrade to Flink 1.13+ if " + - "upsert mode is needed. If upsert mode is not needed, set the table property `write.upsert.enabled` " + - "property to 'false'."); + String deprecationNotice = + "This job is running in upsert mode. Upsert mode should not be used with Flink 1.12 due to correctness " + + "issues. Please upgrade to Flink 1.13+ if upsert mode is truly needed. If upsert mode is not needed, " + + "set the table property `write.upsert.enabled` to 'false' and don't use `upsert(true)` while " + + "using the Flink Sink builder"; + LOG.error(deprecationNotice); + + Preconditions.checkState(!overwrite, + "OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream."); + Preconditions.checkState(!equalityFieldIds.isEmpty(), + "Equality field columns shouldn't be empty when configuring to use UPSERT data stream."); + if (!table.spec().isUnpartitioned()) { + for (PartitionField partitionField : table.spec().fields()) { + Preconditions.checkState(equalityFieldIds.contains(partitionField.sourceId()), + "In UPSERT mode, partition field '%s' should be included in equality fields: '%s'", + partitionField, equalityFieldColumns); + } + } } IcebergStreamWriter streamWriter = createStreamWriter(table, flinkRowType, equalityFieldIds, upsertMode); From 3d9483b81d258a4936d74be121f528e3843b11e5 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 19:37:41 -0700 Subject: [PATCH 06/16] Remove test for checking if upsert mode is disabled --- .../iceberg/flink/sink/TestFlinkIcebergSinkV2.java | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java index e0685dbb2227..a86575475b41 100644 --- a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java +++ b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java @@ -377,19 +377,6 @@ public void testChangeLogOnSameKey() throws Exception { false, elementsPerCheckpoint, expectedRecords); } - @Test - public void testUpsertModeIsDisabled() throws Exception { - DataStream dataStream = env.addSource(new BoundedTestSource<>(ImmutableList.of()), ROW_TYPE_INFO); - AssertHelpers.assertThrows("Upsert mode is not supported in Flink 1.12", - UnsupportedOperationException.class, "Upsert mode is not supported in Flink 1.12", - () -> FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA) - .tableLoader(tableLoader) - .tableSchema(SimpleDataUtil.FLINK_SCHEMA) - .writeParallelism(parallelism) - .upsert(true) - .append()); - } - private StructLikeSet expectedRowSet(Record... records) { return SimpleDataUtil.expectedRowSet(table, records); } From 57f41d08ac524abce640421501c758bd0e327127 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 19:38:55 -0700 Subject: [PATCH 07/16] Remove test assumption that upsert mode is not enabled in 1.12 --- .../org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java index a86575475b41..10a70e78df4a 100644 --- a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java +++ b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java @@ -159,8 +159,6 @@ private void testChangeLogs(List equalityFieldColumns, boolean insertAsUpsert, List> elementsPerCheckpoint, List> expectedRecordsPerCheckpoint) throws Exception { - Assert.assertFalse("Upsert mode is not supported in Flink 1.12", insertAsUpsert); - DataStream dataStream = env.addSource(new BoundedTestSource<>(elementsPerCheckpoint), ROW_TYPE_INFO); FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA) From 9c0e69bf629831165a1429a5a5e9a03e4d467657 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 19:45:12 -0700 Subject: [PATCH 08/16] Add back in original tests and line them up --- .../flink/sink/TestFlinkIcebergSinkV2.java | 115 ++++++++++++++++++ 1 file changed, 115 insertions(+) diff --git a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java index 10a70e78df4a..c8363e57b77a 100644 --- a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java +++ b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java @@ -375,6 +375,121 @@ public void testChangeLogOnSameKey() throws Exception { false, elementsPerCheckpoint, expectedRecords); } + @Test + public void testUpsertModeCheck() throws Exception { + DataStream dataStream = env.addSource(new BoundedTestSource<>(ImmutableList.of()), ROW_TYPE_INFO); + FlinkSink.Builder builder = FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA) + .tableLoader(tableLoader) + .tableSchema(SimpleDataUtil.FLINK_SCHEMA) + .writeParallelism(parallelism) + .upsert(true); + + AssertHelpers.assertThrows("Should be error because upsert mode and overwrite mode enable at the same time.", + IllegalStateException.class, "OVERWRITE mode shouldn't be enable", + () -> builder.equalityFieldColumns(ImmutableList.of("id", "data")).overwrite(true).append() + ); + + AssertHelpers.assertThrows("Should be error because equality field columns are empty.", + IllegalStateException.class, "Equality field columns shouldn't be empty", + () -> builder.equalityFieldColumns(ImmutableList.of()).overwrite(false).append() + ); + } + + @Test + public void testUpsertOnIdKey() throws Exception { + List> elementsPerCheckpoint = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("+U", 1, "bbb") + ), + ImmutableList.of( + row("+I", 1, "ccc") + ), + ImmutableList.of( + row("+U", 1, "ddd"), + row("+I", 1, "eee") + ) + ); + + List> expectedRecords = ImmutableList.of( + ImmutableList.of(record(1, "bbb")), + ImmutableList.of(record(1, "ccc")), + ImmutableList.of(record(1, "eee")) + ); + + if (!partitioned) { + testChangeLogs(ImmutableList.of("id"), row -> row.getField(ROW_ID_POS), true, + elementsPerCheckpoint, expectedRecords); + } else { + AssertHelpers.assertThrows("Should be error because equality field columns don't include all partition keys", + IllegalStateException.class, "should be included in equality fields", + () -> { + testChangeLogs(ImmutableList.of("id"), row -> row.getField(ROW_ID_POS), true, + elementsPerCheckpoint, expectedRecords); + return null; + }); + } + } + + @Test + public void testUpsertOnDataKey() throws Exception { + List> elementsPerCheckpoint = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("+I", 2, "aaa"), + row("+I", 3, "bbb") + ), + ImmutableList.of( + row("+U", 4, "aaa"), + row("-U", 3, "bbb"), + row("+U", 5, "bbb") + ), + ImmutableList.of( + row("+I", 6, "aaa"), + row("+U", 7, "bbb") + ) + ); + + List> expectedRecords = ImmutableList.of( + ImmutableList.of(record(2, "aaa"), record(3, "bbb")), + ImmutableList.of(record(4, "aaa"), record(5, "bbb")), + ImmutableList.of(record(6, "aaa"), record(7, "bbb")) + ); + + testChangeLogs(ImmutableList.of("data"), row -> row.getField(ROW_DATA_POS), true, + elementsPerCheckpoint, expectedRecords); + } + + @Test + public void testUpsertOnIdDataKey() throws Exception { + List> elementsPerCheckpoint = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("+U", 1, "aaa"), + row("+I", 2, "bbb") + ), + ImmutableList.of( + row("+I", 1, "aaa"), + row("-D", 2, "bbb"), + row("+I", 2, "ccc") + ), + ImmutableList.of( + row("+U", 1, "bbb"), + row("-U", 1, "ccc"), + row("-D", 1, "aaa") + ) + ); + + List> expectedRecords = ImmutableList.of( + ImmutableList.of(record(1, "aaa"), record(2, "bbb")), + ImmutableList.of(record(1, "aaa"), record(2, "ccc")), + ImmutableList.of(record(1, "bbb"), record(2, "ccc")) + ); + + testChangeLogs(ImmutableList.of("id", "data"), row -> Row.of(row.getField(ROW_ID_POS), row.getField(ROW_DATA_POS)), + true, elementsPerCheckpoint, expectedRecords); + } + private StructLikeSet expectedRowSet(Record... records) { return SimpleDataUtil.expectedRowSet(table, records); } From d7b7a37a51951fce84da1b118a154936f9bf396a Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 19:50:08 -0700 Subject: [PATCH 09/16] checkstyle --- .../flink/sink/TestFlinkIcebergSinkV2.java | 36 +++++++++---------- 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java index c8363e57b77a..ad0af05bb090 100644 --- a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java +++ b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java @@ -434,30 +434,30 @@ public void testUpsertOnIdKey() throws Exception { @Test public void testUpsertOnDataKey() throws Exception { List> elementsPerCheckpoint = ImmutableList.of( - ImmutableList.of( - row("+I", 1, "aaa"), - row("+I", 2, "aaa"), - row("+I", 3, "bbb") - ), - ImmutableList.of( - row("+U", 4, "aaa"), - row("-U", 3, "bbb"), - row("+U", 5, "bbb") - ), - ImmutableList.of( - row("+I", 6, "aaa"), - row("+U", 7, "bbb") - ) + ImmutableList.of( + row("+I", 1, "aaa"), + row("+I", 2, "aaa"), + row("+I", 3, "bbb") + ), + ImmutableList.of( + row("+U", 4, "aaa"), + row("-U", 3, "bbb"), + row("+U", 5, "bbb") + ), + ImmutableList.of( + row("+I", 6, "aaa"), + row("+U", 7, "bbb") + ) ); List> expectedRecords = ImmutableList.of( - ImmutableList.of(record(2, "aaa"), record(3, "bbb")), - ImmutableList.of(record(4, "aaa"), record(5, "bbb")), - ImmutableList.of(record(6, "aaa"), record(7, "bbb")) + ImmutableList.of(record(2, "aaa"), record(3, "bbb")), + ImmutableList.of(record(4, "aaa"), record(5, "bbb")), + ImmutableList.of(record(6, "aaa"), record(7, "bbb")) ); testChangeLogs(ImmutableList.of("data"), row -> row.getField(ROW_DATA_POS), true, - elementsPerCheckpoint, expectedRecords); + elementsPerCheckpoint, expectedRecords); } @Test From 6e014f406c81c41b28d16db192d7bfa1de4aa664 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 19:52:00 -0700 Subject: [PATCH 10/16] finish checkstyle --- .../flink/sink/TestFlinkIcebergSinkV2.java | 38 +++++++++---------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java index ad0af05bb090..23169d1a59c6 100644 --- a/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java +++ b/flink/v1.12/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java @@ -463,31 +463,31 @@ public void testUpsertOnDataKey() throws Exception { @Test public void testUpsertOnIdDataKey() throws Exception { List> elementsPerCheckpoint = ImmutableList.of( - ImmutableList.of( - row("+I", 1, "aaa"), - row("+U", 1, "aaa"), - row("+I", 2, "bbb") - ), - ImmutableList.of( - row("+I", 1, "aaa"), - row("-D", 2, "bbb"), - row("+I", 2, "ccc") - ), - ImmutableList.of( - row("+U", 1, "bbb"), - row("-U", 1, "ccc"), - row("-D", 1, "aaa") - ) + ImmutableList.of( + row("+I", 1, "aaa"), + row("+U", 1, "aaa"), + row("+I", 2, "bbb") + ), + ImmutableList.of( + row("+I", 1, "aaa"), + row("-D", 2, "bbb"), + row("+I", 2, "ccc") + ), + ImmutableList.of( + row("+U", 1, "bbb"), + row("-U", 1, "ccc"), + row("-D", 1, "aaa") + ) ); List> expectedRecords = ImmutableList.of( - ImmutableList.of(record(1, "aaa"), record(2, "bbb")), - ImmutableList.of(record(1, "aaa"), record(2, "ccc")), - ImmutableList.of(record(1, "bbb"), record(2, "ccc")) + ImmutableList.of(record(1, "aaa"), record(2, "bbb")), + ImmutableList.of(record(1, "aaa"), record(2, "ccc")), + ImmutableList.of(record(1, "bbb"), record(2, "ccc")) ); testChangeLogs(ImmutableList.of("id", "data"), row -> Row.of(row.getField(ROW_ID_POS), row.getField(ROW_DATA_POS)), - true, elementsPerCheckpoint, expectedRecords); + true, elementsPerCheckpoint, expectedRecords); } private StructLikeSet expectedRowSet(Record... records) { From 4cb0939d5698f105910f508a4717cdf2e2768219 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 19:53:36 -0700 Subject: [PATCH 11/16] Final checkstyle --- .../main/java/org/apache/iceberg/flink/sink/FlinkSink.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index e271a1766911..d7a5344132b6 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -398,8 +398,8 @@ private SingleOutputStreamOperator appendWriter(DataStream if (!table.spec().isUnpartitioned()) { for (PartitionField partitionField : table.spec().fields()) { Preconditions.checkState(equalityFieldIds.contains(partitionField.sourceId()), - "In UPSERT mode, partition field '%s' should be included in equality fields: '%s'", - partitionField, equalityFieldColumns); + "In UPSERT mode, partition field '%s' should be included in equality fields: '%s'", + partitionField, equalityFieldColumns); } } } From e35cdb7e9abcdbe1a8b6ed6aa45e83001d0233d7 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 20:51:07 -0700 Subject: [PATCH 12/16] Change warning message to be more direct --- .../main/java/org/apache/iceberg/flink/sink/FlinkSink.java | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index d7a5344132b6..20914ac33422 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -385,10 +385,9 @@ private SingleOutputStreamOperator appendWriter(DataStream // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) { String deprecationNotice = - "This job is running in upsert mode. Upsert mode should not be used with Flink 1.12 due to correctness " + - "issues. Please upgrade to Flink 1.13+ if upsert mode is truly needed. If upsert mode is not needed, " + - "set the table property `write.upsert.enabled` to 'false' and don't use `upsert(true)` while " + - "using the Flink Sink builder"; + "Upsert mode should not be used with Flink 1.12 because it will write incorrect delete file metadata, " + + "which could prevent deletes from being correctly applied. Upgrading to Flink 1.13+ is recommended. " + + "To safely use Flink 1.12, set manifest metrics to counts only."; LOG.error(deprecationNotice); Preconditions.checkState(!overwrite, From f711cde7eafbe25c31bdfd09c66ceeca38c2e0b6 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 20:52:43 -0700 Subject: [PATCH 13/16] Merge old warning message and new --- .../main/java/org/apache/iceberg/flink/sink/FlinkSink.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index 20914ac33422..d8fbf29612ba 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -385,8 +385,9 @@ private SingleOutputStreamOperator appendWriter(DataStream // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) { String deprecationNotice = - "Upsert mode should not be used with Flink 1.12 because it will write incorrect delete file metadata, " + - "which could prevent deletes from being correctly applied. Upgrading to Flink 1.13+ is recommended. " + + "This job is running in upsert mode. Upsert mode should not be used with Flink 1.12 because it will " + + "write incorrect delete file metadata, which could prevent deletes from being correctly applied. " + + "Upgrading to Flink 1.13+ is recommended. " + "To safely use Flink 1.12, set manifest metrics to counts only."; LOG.error(deprecationNotice); From 161e16da511fd8ff1eebcae8eba20f86e7d47745 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 20:54:42 -0700 Subject: [PATCH 14/16] Use table sink in place of job --- .../src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index d8fbf29612ba..b87d5829cbb5 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -385,7 +385,7 @@ private SingleOutputStreamOperator appendWriter(DataStream // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) { String deprecationNotice = - "This job is running in upsert mode. Upsert mode should not be used with Flink 1.12 because it will " + + "This table sink is running in upsert mode. Upsert mode should not be used with Flink 1.12 because it will " + "write incorrect delete file metadata, which could prevent deletes from being correctly applied. " + "Upgrading to Flink 1.13+ is recommended. " + "To safely use Flink 1.12, set manifest metrics to counts only."; From 211fa7adb13871ca284708c9405937a1391fd6d7 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 21:03:53 -0700 Subject: [PATCH 15/16] Dont use a variable for logging as it breaks checkstyle --- .../java/org/apache/iceberg/flink/sink/FlinkSink.java | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index b87d5829cbb5..9a3f2f3de491 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -384,12 +384,11 @@ private SingleOutputStreamOperator appendWriter(DataStream // // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) { - String deprecationNotice = - "This table sink is running in upsert mode. Upsert mode should not be used with Flink 1.12 because it will " + - "write incorrect delete file metadata, which could prevent deletes from being correctly applied. " + - "Upgrading to Flink 1.13+ is recommended. " + - "To safely use Flink 1.12, set manifest metrics to counts only."; - LOG.error(deprecationNotice); + LOG.error( + "This table sink is running in upsert mode. Upsert mode should not be used with Flink 1.12 because " + + "it will write incorrect delete file metadata, which could prevent deletes from being correctly" + + "applied. Upgrading to Flink 1.13+ is recommended. " + + "To safely use Flink 1.12, set manifest metrics to counts only."); Preconditions.checkState(!overwrite, "OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream."); From 15319b53f48dbec7dcacc6428c21342a5818ffb8 Mon Sep 17 00:00:00 2001 From: Kyle Bendickson Date: Wed, 11 May 2022 22:12:08 -0700 Subject: [PATCH 16/16] Fix typo in comment --- .../java/org/apache/iceberg/flink/sink/FlinkSink.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java index 9a3f2f3de491..bae585f247e3 100644 --- a/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java +++ b/flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java @@ -376,11 +376,13 @@ private SingleOutputStreamOperator appendWriter(DataStream UPSERT_ENABLED, UPSERT_ENABLED_DEFAULT); // `upsert` mode should not be used in Flink 1.12 due to correctness issues. - // As part of a patch release, apachee-iceberg-flink-runtime_1.12:0.13.2, + // As part of a patch release, apache-iceberg-flink-runtime_1.12:0.13.2, // it has been decided the best course of action would be to log a warning - // asking people to upgrade, as Flink 1.12 is deprecated from both upstream - // Flink as well as Iceberg as of Iceberg 0.14.0. But we allow the configuration - // given that it's a patch release and the change would be otherwise very breaking. + // asking people to upgrade, as Flink 1.12 has been deprecated in upstream Apache Flink + // for some time as well as will be removed in the next major Iceberg release, Iceberg 0.14.0. + // + // But we allow the configuration given that it's a patch release and the change would be otherwise + // too breaking for a patch release. // // See in https://github.com/apache/iceberg/pull/4364 for more information. if (upsertMode) {