ferenc-csaky commented on code in PR #17142:
URL: https://github.com/apache/iceberg/pull/17142#discussion_r3551905619
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java:
##########
@@ -318,4 +341,119 @@ public void testAllMaintenanceE2e(String lockType) throws
Exception {
getDataFiles(table.snapshot(table.currentSnapshot().parentId()),
table);
assertThat(preCompactDataFiles).hasSize(3);
}
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesOperatorAdded(String lockType) {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "hello"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal,
FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append();
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ boolean containsConvert = false;
+ for (JobVertex vertex : streamGraph.getJobGraph().getVertices()) {
+ if (vertex.getName().contains("EqConvert")) {
+ containsConvert = true;
+ break;
+ }
+ }
+
+ assertThat(containsConvert).isTrue();
+ }
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesRequiresEqualityFields(String
lockType) {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "hello"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal,
FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ // No equality field columns and no identifier fields on the schema, so
there is nothing to
+ // convert.
+ assertThatThrownBy(
+ () ->
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .setAll(flinkConf)
+ .append())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Equality field columns must be set");
+ }
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesE2e(String lockType) throws Exception {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+
+ // Pre-existing row on main. The sink then upserts the same key, writing
an equality delete that
+ // removes this row, which the converter resolves to a deletion vector in
a single cycle.
+ new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
+ .appendToTable(ImmutableList.of(SimpleDataUtil.createRecord(1,
"aaa")));
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "bbb"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal,
FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append();
+
+ JobClient jobClient = env.executeAsync("Test Convert Equality Deletes
E2E");
+ try {
+ Awaitility.await()
Review Comment:
nit: static import for `await()`, as it's a widely used lib and a common
practice
##########
docs/docs/flink-configuration.md:
##########
@@ -163,6 +163,20 @@ INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true')
*/
| shred-variants | Table write.parquet.shred-variants
| Overrides this table's shred variants for this write |
| variant-inference-buffer-size | Table
write.parquet.variant-inference-buffer-size | Overrides this table's variant
inference buffer size for this write |
+#### Post-commit maintenance options
+
+`IcebergSink` can run [table
maintenance](flink-maintenance.md#icebergsink-with-post-commit-integration)
+after each commit. The enable flags are listed below; see the maintenance docs
for the full per-task
+option set.
+
+| Flink option | Default | Description |
+|--------------|---------|-------------|
+| flink-maintenance.rewrite.enabled | false | Compact data files after commit |
+| flink-maintenance.expire-snapshots.enabled | false | Expire snapshots after
commit |
+| flink-maintenance.delete-orphan-files.enabled | false | Delete orphan files
after commit |
+| flink-maintenance.convert-equality-deletes.enabled | false | Convert
equality deletes to deletion vectors after commit (requires equality fields and
format version >= 3) |
+| flink-maintenance.convert-equality-deletes.target-branch | Write branch |
Branch the converted DVs are committed to; defaults to the write branch
(in-place conversion) |
Review Comment:
nit: The config option itself has no default value. Maybe it would be less
confusing to keep this table in sync with that and put `null` to the `Default`
table col (based on the read option table) and rephrase the description that
something like "if not specified, the write branch will be used (in-place
conversion)"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
