mxm commented on code in PR #17142:
URL: https://github.com/apache/iceberg/pull/17142#discussion_r3556386243
##########
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:
While working on the doc changes, I realized that we don't list the
maintenance config options on the Flink config page. This may warrants another
PR if we decide to list maintenance options under the config page. I've removed
the changes here for now.
##########
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:
Thanks! Searching through the code base, I've found 190 results for the use
of `Awaitility.await()`, but none for the static import. For consistency with
the existing code style, I'd like to keep the non-static version.
##########
docs/docs/flink-maintenance.md:
##########
Review Comment:
Thanks @Guosmilesmile! That's a better placement. I've moved the config here.
##########
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:
Thanks! Based don the PR feedback, I've removed the section here.
##########
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:
I've applied your suggestion when moving these configs to the
`flink-maintenance.md` page.
##########
docs/docs/flink-maintenance.md:
##########
@@ -397,6 +411,10 @@
flinkConf.put("flink-maintenance.expire-snapshots.max-snapshot-age-seconds", "60
// Configure delete orphan files
flinkConf.put("flink-maintenance.delete-orphan-files.min-age-seconds",
"259200");
+// Configure convert equality deletes. Converts in place on the write branch
by default;
+// set a target branch to instead promote the converted deletion vectors there.
Review Comment:
Thanks @ferenc-csaky! Updated.
##########
docs/docs/flink-writes.md:
##########
@@ -398,6 +398,32 @@ To use SinkV2 based implementation, replace `FlinkSink`
with `IcebergSink` in th
- The `RANGE` distribution mode is not yet available for the `IcebergSink`
- When using `IcebergSink` use `uidSuffix` instead of the `uidPrefix`
+### Post-commit table maintenance
Review Comment:
With the addition of the post-commit maintenance trigger, table maintenance
can now intertwined with the write process. IMHO we should at least mention
this feature in the write section.
--
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]
