mxm commented on code in PR #18661:
URL: https://github.com/apache/iceberg/pull/18661#discussion_r4412661589
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -847,6 +851,24 @@ IcebergSink build() {
equalityFieldColumnsSet);
}
+ private void validateCompactionBranch(
+ FlinkWriteConf flinkWriteConf,
+ FlinkMaintenanceConfig flinkMaintenanceConfig,
+ RewriteDataFilesConfig rewriteDataFilesConfig) {
+ String convertTargetBranch =
+
flinkMaintenanceConfig.createConvertEqualityDeletesConfig().targetBranch();
+ Preconditions.checkState(
+ convertTargetBranch == null
+ || convertTargetBranch.equals(flinkWriteConf.branch())
+ || rewriteDataFilesConfig.branch() != null,
Review Comment:
Should we check the resolved rewrite branch? If
`flink-maintenance.rewrite.branch` is set to the sink's write branch, this
check passes, but the converter still fails on the rewrite snapshot.
##########
docs/docs/flink-maintenance.md:
##########
@@ -505,10 +505,13 @@ These keys are used in SQL (SET or table WITH options) or
via `IcebergSink.Build
| `flink-maintenance.rewrite.schedule.data-file-count` | Trigger after N data
files | `1000` |
| `flink-maintenance.rewrite.schedule.data-file-size` | Trigger after total
data file size (bytes) | `107374182400` (100GB) |
| `flink-maintenance.rewrite.schedule.interval-second` | Trigger after time
interval (seconds) | `600` |
+| `flink-maintenance.rewrite.branch` | Branch to compact. Set it to compact a
different branch than the one the sink writes to, e.g. `main` when the sink
writes to a staging branch | Not set |
| `flink-maintenance.rewrite.max-bytes` | Maximum bytes to rewrite per
execution | `Long.MAX_VALUE` |
| `flink-maintenance.rewrite.partial-progress.enabled` | Enable partial
progress commits | `false` |
| `flink-maintenance.rewrite.partial-progress.max-commits` | Maximum commits
for partial progress | `10` |
+When rewrite data files and convert equality deletes are both enabled and
`flink-maintenance.convert-equality-deletes.target-branch` is set to a branch
other than the sink's write branch, set `flink-maintenance.rewrite.branch`
explicitly to the branch to rewrite.
Review Comment:
NIT: Should this be:
```suggestion
When rewrite data files and convert equality deletes are both enabled and
`flink-maintenance.convert-equality-deletes.target-branch` is set to a branch
other than the sink's write branch, set `flink-maintenance.rewrite.branch` to a
branch other than the sink's write branch, usually the convert target branch.
```
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFilesConfig.java:
##########
@@ -30,6 +30,13 @@
public class RewriteDataFilesConfig {
public static final String PREFIX = FlinkMaintenanceConfig.PREFIX +
"rewrite.";
+ public static final String BRANCH = PREFIX + "branch";
+ public static final ConfigOption<String> BRANCH_OPTION =
+ ConfigOptions.key(BRANCH)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The branch to compact. Defaults to the sink's
write branch.");
Review Comment:
Thanks for updating the code to reject invalid configurations. I think we
can also default the rewrite branch to the target branch of the conversion,
unless the user has manually configured the rewrite branch. In any case, we
should reject invalid configurations.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java:
##########
@@ -436,6 +438,33 @@ public void testConvertEqualityDeletesE2e(String lockType)
throws Exception {
}
}
+ @Test
+ void testCompactionBranchRequiredWhenConvertTargetsOtherBranch() {
+ setupLockConfig("");
+ setupCompactionConfig();
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+ flinkConf.put(FlinkWriteOptions.BRANCH.key(), "staging");
+ flinkConf.put(ConvertEqualityDeletesConfig.TARGET_BRANCH, "main");
+
+ 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));
+
+ assertThatThrownBy(
+ () ->
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Missing config " +
RewriteDataFilesConfig.BRANCH);
Review Comment:
Could we assert the full message with `hasMessage`? That also checks the
branch names.
--
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]
