You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "mxm (via GitHub)" <gi...@apache.org> on 2026/03/09 15:21:44 UTC
[PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
mxm opened a new pull request, #15566:
URL: https://github.com/apache/iceberg/pull/15566
IcebergSink previously hardcoded RewriteDataFiles as the only post-commit table maintenance task, behind a boolean `compactMode` option. Users could not configure ExpireSnapshots, DeleteOrphanFiles, or any combination of tasks through the builder API.
This adds a maintenance(MaintenanceTaskBuilder) method to the IcebergSink builder that can be called multiple times. The compactMode config can still be used.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2941098528
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.FlinkConfParser;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+public class DeleteOrphanFilesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "delete-orphan-files.";
+
+ private static final Splitter COMMA_SPLITTER = Splitter.on(",");
+ private static final Splitter EQUALS_SPLITTER = Splitter.on("=").limit(2);
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .defaultValue(60 * 60L) // Default is 1 hour
+ .withDescription(
+ "The time interval (in seconds) between two consecutive delete orphan files operations.");
+
+ public static final String MIN_AGE_SECONDS = PREFIX + "min-age-seconds";
+ public static final ConfigOption<Long> MIN_AGE_SECONDS_OPTION =
+ ConfigOptions.key(MIN_AGE_SECONDS)
+ .longType()
+ .defaultValue(3L * 24 * 60 * 60) // Default is 3 days
+ .withDescription(
+ "The minimum age (in seconds) of files to be considered for deletion. "
+ + "Files newer than this will not be removed.");
+
+ public static final String DELETE_BATCH_SIZE = PREFIX + "delete-batch-size";
+ public static final ConfigOption<Integer> DELETE_BATCH_SIZE_OPTION =
+ ConfigOptions.key(DELETE_BATCH_SIZE)
+ .intType()
+ .defaultValue(1000)
+ .withDescription("The batch size used for deleting orphan files.");
+
+ public static final String LOCATION = PREFIX + "location";
+ public static final ConfigOption<String> LOCATION_OPTION =
+ ConfigOptions.key(LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location to start recursive listing of candidate files for removal. "
+ + "By default, the table location is used.");
+
+ public static final String USE_PREFIX_LISTING = PREFIX + "use-prefix-listing";
+ public static final ConfigOption<Boolean> USE_PREFIX_LISTING_OPTION =
+ ConfigOptions.key(USE_PREFIX_LISTING)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to use prefix listing when listing files from the file system.");
+
+ public static final String PLANNING_WORKER_POOL_SIZE = PREFIX + "planning-worker-pool-size";
+ public static final ConfigOption<Integer> PLANNING_WORKER_POOL_SIZE_OPTION =
+ ConfigOptions.key(PLANNING_WORKER_POOL_SIZE)
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The worker pool size used for planning the scan of the ALL_FILES table. "
+ + "If not set, the shared worker pool is used.");
+
+ public static final String EQUAL_SCHEMES = PREFIX + "equal-schemes";
+ public static final ConfigOption<String> EQUAL_SCHEMES_OPTION =
+ ConfigOptions.key(EQUAL_SCHEMES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Schemes that should be considered equal, in the format 'scheme1=scheme2,scheme3=scheme4'.");
+
+ public static final String EQUAL_AUTHORITIES = PREFIX + "equal-authorities";
+ public static final ConfigOption<String> EQUAL_AUTHORITIES_OPTION =
+ ConfigOptions.key(EQUAL_AUTHORITIES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Authorities that should be considered equal, in the format 'auth1=auth2,auth3=auth4'.");
+
+ public static final String PREFIX_MISMATCH_MODE = PREFIX + "prefix-mismatch-mode";
+ public static final ConfigOption<String> PREFIX_MISMATCH_MODE_OPTION =
+ ConfigOptions.key(PREFIX_MISMATCH_MODE)
+ .stringType()
+ .defaultValue(PrefixMismatchMode.ERROR.name())
+ .withDescription(
+ "Action behavior when location prefixes (schemes/authorities) mismatch. "
+ + "Valid values: ERROR, IGNORE, DELETE.");
+
+ private final FlinkConfParser confParser;
+
+ public DeleteOrphanFilesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ public long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .defaultValue(SCHEDULE_ON_INTERVAL_SECOND_OPTION.defaultValue())
+ .parse();
+ }
+
+ public long minAgeSeconds() {
+ return confParser
+ .longConf()
+ .option(MIN_AGE_SECONDS)
+ .flinkConfig(MIN_AGE_SECONDS_OPTION)
+ .defaultValue(MIN_AGE_SECONDS_OPTION.defaultValue())
+ .parse();
+ }
+
+ public int deleteBatchSize() {
+ return confParser
+ .intConf()
+ .option(DELETE_BATCH_SIZE)
+ .flinkConfig(DELETE_BATCH_SIZE_OPTION)
+ .defaultValue(DELETE_BATCH_SIZE_OPTION.defaultValue())
+ .parse();
+ }
+
+ public String location() {
+ return confParser.stringConf().option(LOCATION).flinkConfig(LOCATION_OPTION).parseOptional();
+ }
+
+ public boolean usePrefixListing() {
+ return confParser
+ .booleanConf()
+ .option(USE_PREFIX_LISTING)
+ .flinkConfig(USE_PREFIX_LISTING_OPTION)
+ .defaultValue(USE_PREFIX_LISTING_OPTION.defaultValue())
+ .parse();
+ }
+
+ public Integer planningWorkerPoolSize() {
+ return confParser
+ .intConf()
+ .option(PLANNING_WORKER_POOL_SIZE)
+ .flinkConfig(PLANNING_WORKER_POOL_SIZE_OPTION)
+ .parseOptional();
+ }
+
+ public Map<String, String> equalSchemes() {
+ String equalSchemes =
+ confParser
+ .stringConf()
+ .option(EQUAL_SCHEMES)
+ .flinkConfig(EQUAL_SCHEMES_OPTION)
+ .parseOptional();
+ if (equalSchemes != null) {
+ return parseKeyValuePairs(equalSchemes);
+ }
+
+ return Maps.newHashMap(DeleteOrphanFiles.DEFAULT_EQUAL_SCHEMES);
+ }
+
+ public Map<String, String> equalAuthorities() {
+ String value =
+ confParser
+ .stringConf()
+ .option(EQUAL_AUTHORITIES)
+ .flinkConfig(EQUAL_AUTHORITIES_OPTION)
+ .parseOptional();
+ return value != null ? parseKeyValuePairs(value) : null;
+ }
+
+ public PrefixMismatchMode prefixMismatchMode() {
+ String value =
+ confParser
+ .stringConf()
+ .option(PREFIX_MISMATCH_MODE)
+ .flinkConfig(PREFIX_MISMATCH_MODE_OPTION)
+ .defaultValue(PREFIX_MISMATCH_MODE_OPTION.defaultValue())
+ .parse();
+ return PrefixMismatchMode.valueOf(value);
+ }
+
+ private static Map<String, String> parseKeyValuePairs(String value) {
+ Map<String, String> result = Maps.newHashMap();
+ for (String pair : COMMA_SPLITTER.split(value)) {
+ List<String> parts = EQUALS_SPLITTER.splitToList(pair);
+ if (parts.size() == 2) {
Review Comment:
Throwing now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939796707
##########
docs/docs/flink-maintenance.md:
##########
@@ -361,9 +409,64 @@ CREATE TABLE db.tbl (
);
```
+### IcebergSink Maintenance Configuration (SQL)
+
+These keys are used in SQL (SET or table WITH options) or via `IcebergSink.Builder.set()` / `setAll()`.
+
+#### Enable Flags
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.rewrite.enabled` | Enable compaction (rewrite data files) | `false` |
+| `flink-maintenance.expire-snapshots.enabled` | Enable expire snapshots | `false` |
+| `flink-maintenance.delete-orphan-files.enabled` | Enable delete orphan files | `false` |
+
+#### Rewrite Data Files Configuration
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.rewrite.schedule.commit-count` | Trigger after N commits | `10` |
+| `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.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` |
+
+#### Expire Snapshots Configuration
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.expire-snapshots.schedule.commit-count` | Trigger after N commits | `10` |
+| `flink-maintenance.expire-snapshots.schedule.data-file-count` | Trigger after N data files | `1000` |
+| `flink-maintenance.expire-snapshots.schedule.data-file-size` | Trigger after total data file size (bytes) | `107374182400` (100GB) |
+| `flink-maintenance.expire-snapshots.schedule.interval-second` | Trigger after time interval (seconds) | `600` |
+| `flink-maintenance.expire-snapshots.max-snapshot-age-seconds` | Maximum age of snapshots to retain (seconds) | Not set |
+| `flink-maintenance.expire-snapshots.retain-last` | Minimum number of snapshots to retain | Not set |
+| `flink-maintenance.expire-snapshots.delete-batch-size` | Batch size for deleting expired files | `1000` |
+| `flink-maintenance.expire-snapshots.clean-expired-metadata` | Remove expired metadata (partition specs, schemas) | Not set |
+| `flink-maintenance.expire-snapshots.planning-worker-pool-size` | Worker pool size for planning | Shared pool |
+
+#### Delete Orphan Files Configuration
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.delete-orphan-files.schedule.commit-count` | Trigger after N commits | `10` |
+| `flink-maintenance.delete-orphan-files.schedule.data-file-count` | Trigger after N data files | `1000` |
+| `flink-maintenance.delete-orphan-files.schedule.data-file-size` | Trigger after total data file size (bytes) | `107374182400` (100GB) |
+| `flink-maintenance.delete-orphan-files.schedule.interval-second` | Trigger after time interval (seconds) | `600` |
+| `flink-maintenance.delete-orphan-files.min-age-seconds` | Minimum age of files to consider for deletion (seconds) | `259200` (3 days) |
+| `flink-maintenance.delete-orphan-files.delete-batch-size` | Batch size for deleting orphan files | `1000` |
+| `flink-maintenance.delete-orphan-files.location` | Location to start recursive listing | Table location |
+| `flink-maintenance.delete-orphan-files.use-prefix-listing` | Use prefix listing for file discovery | `false` |
+| `flink-maintenance.delete-orphan-files.planning-worker-pool-size` | Worker pool size for planning | Shared pool |
+| `flink-maintenance.delete-orphan-files.equal-schemes` | Equivalent schemes (format: `s3n=s3,s3a=s3`) | Not set |
Review Comment:
Adjusted.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2924293582
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles(boolean enabled) {
Review Comment:
maybe remove the boolean parameter from here too
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2906436628
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
What about configs for the other maintenance tasks.
Do we want to have a generic way to config those?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#issuecomment-4074569461
Merged to main.
Thanks @mxm for the PR and @Guosmilesmile for the review!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939989276
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,85 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles() {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(Map<String, String> config) {
+ rewriteDataFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables expire snapshots as a post-commit maintenance task.
+ *
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots() {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link ExpireSnapshotsConfig} for available
+ * keys
+ */
+ public Builder expireSnapshots(Map<String, String> config) {
+ expireSnapshots();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @see DeleteOrphanFilesConfig for the default config.
+ */
+ public Builder deleteOrphanFiles() {
+ writeOptions.put(FlinkWriteOptions.DELETE_ORPHAN_FILES_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link DeleteOrphanFilesConfig} for available
+ * keys.
+ */
+ public Builder deleteOrphanFiles(Map<String, String> config) {
+ deleteOrphanFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
@Override
public Builder toBranch(String branch) {
Review Comment:
We don't. This isn't related to my changes, but it probably makes sense to fix this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2924292065
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(boolean enabled, Map<String, String> config) {
Review Comment:
remove the boolean parameter. Doesn't add value
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940857469
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,35 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
+ this.location(loc);
+ }
+
+ Integer poolSize = deleteOrphanFilesConfig.planningWorkerPoolSize();
Review Comment:
If we set the config default to `ThreadPools.WORKER_THREAD_POOL_SIZE` then we don't need the `if`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2906501901
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
Eventually, yes. This will give users the option to use post-commit table maintenance outside of the Java API.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2906616594
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
I thought about that as well. We already allow configuring Iceberg via table options, Flink config, or via the builder method. We don't allow just one way. I think this a bit of a similar case.
Granted, if we treat maintenance tasks as a single configuration option, then we should not allow builder-based and table options at the same time. However, if we think about the maintenance tasks as separate options, then it makes sense to allow both configuration styles at the same time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2924241594
##########
docs/docs/flink-maintenance.md:
##########
@@ -302,19 +302,50 @@ public class TableMaintenanceJob {
Apache Iceberg Sink V2 for Flink allows automatic execution of maintenance tasks after data is committed to the table, using the addPostCommitTopology(...) method.
+#### DataStream API
+
+##### Builder
+
+```java
+IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .rewriteDataFiles(true, Map.of(
Review Comment:
does `true` add any value on the api?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2926132902
##########
docs/docs/flink-maintenance.md:
##########
@@ -302,19 +302,50 @@ public class TableMaintenanceJob {
Apache Iceberg Sink V2 for Flink allows automatic execution of maintenance tasks after data is committed to the table, using the addPostCommitTopology(...) method.
+#### DataStream API
+
+##### Builder
+
+```java
+IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .rewriteDataFiles(Map.of(
+ RewriteDataFilesConfig.MAX_BYTES, "1073741824"))
+ .expireSnapshots(Map.of(
+ ExpireSnapshotsConfig.RETAIN_LAST, "5",
+ ExpireSnapshotsConfig.MAX_SNAPSHOT_AGE_SECONDS, "604800"))
+ .deleteOrphanFiles(Map.of(
+ DeleteOrphanFilesConfig.MIN_AGE_SECONDS, "259200"))
+ .append();
+```
+
+##### Config
+
+All maintenance tasks are configured through string properties:
+
```java
Map<String, String> flinkConf = new HashMap<>();
-// Enable compaction and maintenance features
-flinkConf.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+// Enable maintenance tasks
+flinkConf.put("flink-maintenance.rewrite.enabled", "true");
+flinkConf.put("flink-maintenance.expire-snapshots.enabled", "true");
+flinkConf.put("flink-maintenance.delete-orphan-files.enabled", "true");
-// Configure JDBC lock settings
-flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
-flinkConf.put(LockConfig.JdbcLockConfig.JDBC_URI_OPTION.key(), "jdbc:postgresql://localhost:5432/iceberg");
-flinkConf.put(LockConfig.LOCK_ID_OPTION.key(), "catalog.db.table");
+// Configure rewrite data files
+flinkConf.put("flink-maintenance.rewrite.max-bytes", "1073741824");
-// Add any other maintenance-related options here as needed
-// ...
+// Configure expire snapshots
+flinkConf.put("flink-maintenance.expire-snapshots.retain-last", "5");
+flinkConf.put("flink-maintenance.expire-snapshots.max-snapshot-age-seconds", "604800");
Review Comment:
I see we're offering
```java
flinkConf.put("flink-maintenance.expire-snapshots.enabled", "true");
```
and we are also offering specific expire-snapshot configs like:
```java
flinkConf.put("flink-maintenance.expire-snapshots.retain-last", "5");
flinkConf.put("flink-maintenance.expire-snapshots.max-snapshot-age-seconds", "604800")
```
Does `flink-maintenance.expire-snapshots.retain-last` require `flink-maintenance.expire-snapshots.enabled` as true? (same question for the other config, and this comment in general for all the other maintenance tasks.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2935263805
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(boolean enabled, Map<String, String> config) {
+ rewriteDataFiles(enabled);
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable expire snapshots
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), Boolean.toString(enabled));
Review Comment:
**ExpireSnapshotsConfig**
- `schedule.commit-count`
- `schedule.interval-second`
- `max-snapshot-age-seconds`
- `retain-last`
- `delete-batch-size`
- `clean-expired-metadata`
- `planning-worker-pool-size`
Regarding `clean-expired-metadata`, I think the default value could be `true`.
For `max-snapshot-age-seconds` and `retain-last`, the defaults are both `null`—do we need to set a default value for them?
---
**DeleteOrphanFilesConfig**
- `schedule.interval-second`
- `min-age-seconds`
- `delete-batch-size`
- `use-prefix-listing`
- `planning-worker-pool-size`
- `equal-schemes`
- `equal-authorities`
- `prefix-mismatch-mode`
Is it enough to keep just these?
From my perspective, I’d like to set `use-prefix-listing` to `true`. We discussed before that we’d like to eventually drop the dependency on the Hadoop library, and setting this to `true` aligns better with that direction.
My concern is whether we might end up in a situation where the SQL defaults to `true`, but the default is `false` when someone manually creates a `DeleteOrphanFiles` task in single tablemainance job, which could be misleading. If the documentation clearly states the default value, that could avoid the issue, but I’d like to know your thoughts.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939868867
##########
docs/docs/flink-maintenance.md:
##########
@@ -207,7 +207,7 @@ env.execute("Table Maintenance Job");
| `retainLast(int)` | Minimum number of snapshots to retain | 1 | int |
| `deleteBatchSize(int)` | Number of files to delete in each batch | 1000 | int |
| `planningWorkerPoolSize(int)` | Number of worker threads for planning snapshot expiration | Shared worker pool | int |
-| `cleanExpiredMetadata(boolean)` | Remove expired metadata files when expiring snapshots | false | boolean |
+| `cleanExpiredMetadata(boolean)` | Remove expired metadata files when expiring snapshots | true | boolean |
Review Comment:
We should still highlight these in the release notes.
CC: @aihuaxu: What is the planned way to collect the release notes?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2909377891
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
Out of curiosity, is the scenario we’re trying to solve here to avoid having to maintain a separate maintenance job per table Or what scenario is this intended to address?
From my perspective, I have a few (still rough) questions would like to ask:
1. Do we allow multiple ways to configure the same maintenance task, and also allow multiple instances of the same type of task? For example, conflicts between the `compactMode` option and the `maintenance()` method, or adding multiple rewrite-data tasks with different configurations via `maintenance()`.
2. Regarding config injection, we can inject configs via the existing `flinkConf` approach, or bypass `flinkConf` and have `maintenance()` generate task configs directly. That could get confusing—can we unify this behind a single entry point?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2924288964
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(boolean enabled, Map<String, String> config) {
+ rewriteDataFiles(enabled);
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable expire snapshots
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), Boolean.toString(enabled));
Review Comment:
Do we have reasonable defaults?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2929086567
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.FlinkConfParser;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+public class DeleteOrphanFilesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "delete-orphan-files.";
+
+ private static final Splitter COMMA_SPLITTER = Splitter.on(",");
+ private static final Splitter EQUALS_SPLITTER = Splitter.on("=").limit(2);
+
+ public static final String SCHEDULE_ON_COMMIT_COUNT = PREFIX + "schedule.commit-count";
+ public static final ConfigOption<Integer> SCHEDULE_ON_COMMIT_COUNT_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_COMMIT_COUNT)
+ .intType()
+ .defaultValue(10)
+ .withDescription(
+ "The number of commits after which to trigger a new delete orphan files operation.");
+
+ public static final String SCHEDULE_ON_DATA_FILE_COUNT = PREFIX + "schedule.data-file-count";
+ public static final ConfigOption<Integer> SCHEDULE_ON_DATA_FILE_COUNT_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_DATA_FILE_COUNT)
+ .intType()
+ .defaultValue(1000)
+ .withDescription(
+ "The number of data files that should trigger a new delete orphan files operation.");
+
+ public static final String SCHEDULE_ON_DATA_FILE_SIZE = PREFIX + "schedule.data-file-size";
+ public static final ConfigOption<Long> SCHEDULE_ON_DATA_FILE_SIZE_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_DATA_FILE_SIZE)
+ .longType()
+ .defaultValue(100L * 1024 * 1024 * 1024) // Default is 100 GB
+ .withDescription(
+ "The total size of data files that should trigger a new delete orphan files operation.");
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .defaultValue(60 * 60L) // Default is 1 hour
+ .withDescription(
+ "The time interval (in seconds) between two consecutive delete orphan files operations.");
+
+ public static final String MIN_AGE_SECONDS = PREFIX + "min-age-seconds";
+ public static final ConfigOption<Long> MIN_AGE_SECONDS_OPTION =
+ ConfigOptions.key(MIN_AGE_SECONDS)
+ .longType()
+ .defaultValue(3L * 24 * 60 * 60) // Default is 3 days
+ .withDescription(
+ "The minimum age (in seconds) of files to be considered for deletion. "
+ + "Files newer than this will not be removed.");
+
+ public static final String DELETE_BATCH_SIZE = PREFIX + "delete-batch-size";
+ public static final ConfigOption<Integer> DELETE_BATCH_SIZE_OPTION =
+ ConfigOptions.key(DELETE_BATCH_SIZE)
+ .intType()
+ .defaultValue(1000)
+ .withDescription("The batch size used for deleting orphan files.");
+
+ public static final String LOCATION = PREFIX + "location";
+ public static final ConfigOption<String> LOCATION_OPTION =
+ ConfigOptions.key(LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location to start recursive listing of candidate files for removal. "
+ + "By default, the table location is used.");
+
+ public static final String USE_PREFIX_LISTING = PREFIX + "use-prefix-listing";
+ public static final ConfigOption<Boolean> USE_PREFIX_LISTING_OPTION =
+ ConfigOptions.key(USE_PREFIX_LISTING)
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to use prefix listing when listing files from the file system.");
+
+ public static final String PLANNING_WORKER_POOL_SIZE = PREFIX + "planning-worker-pool-size";
+ public static final ConfigOption<Integer> PLANNING_WORKER_POOL_SIZE_OPTION =
+ ConfigOptions.key(PLANNING_WORKER_POOL_SIZE)
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The worker pool size used for planning the scan of the ALL_FILES table. "
+ + "If not set, the shared worker pool is used.");
+
+ public static final String EQUAL_SCHEMES = PREFIX + "equal-schemes";
+ public static final ConfigOption<String> EQUAL_SCHEMES_OPTION =
+ ConfigOptions.key(EQUAL_SCHEMES)
+ .stringType()
+ .noDefaultValue()
Review Comment:
The save above
##########
docs/docs/flink-maintenance.md:
##########
@@ -361,9 +409,64 @@ CREATE TABLE db.tbl (
);
```
+### IcebergSink Maintenance Configuration (SQL)
+
+These keys are used in SQL (SET or table WITH options) or via `IcebergSink.Builder.set()` / `setAll()`.
+
+#### Enable Flags
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.rewrite.enabled` | Enable compaction (rewrite data files) | `false` |
+| `flink-maintenance.expire-snapshots.enabled` | Enable expire snapshots | `false` |
+| `flink-maintenance.delete-orphan-files.enabled` | Enable delete orphan files | `false` |
+
+#### Rewrite Data Files Configuration
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.rewrite.schedule.commit-count` | Trigger after N commits | `10` |
+| `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.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` |
+
+#### Expire Snapshots Configuration
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.expire-snapshots.schedule.commit-count` | Trigger after N commits | `10` |
+| `flink-maintenance.expire-snapshots.schedule.data-file-count` | Trigger after N data files | `1000` |
+| `flink-maintenance.expire-snapshots.schedule.data-file-size` | Trigger after total data file size (bytes) | `107374182400` (100GB) |
+| `flink-maintenance.expire-snapshots.schedule.interval-second` | Trigger after time interval (seconds) | `600` |
+| `flink-maintenance.expire-snapshots.max-snapshot-age-seconds` | Maximum age of snapshots to retain (seconds) | Not set |
+| `flink-maintenance.expire-snapshots.retain-last` | Minimum number of snapshots to retain | Not set |
+| `flink-maintenance.expire-snapshots.delete-batch-size` | Batch size for deleting expired files | `1000` |
+| `flink-maintenance.expire-snapshots.clean-expired-metadata` | Remove expired metadata (partition specs, schemas) | Not set |
+| `flink-maintenance.expire-snapshots.planning-worker-pool-size` | Worker pool size for planning | Shared pool |
+
+#### Delete Orphan Files Configuration
+
+| Key | Description | Default |
+|-----|-------------|---------|
+| `flink-maintenance.delete-orphan-files.schedule.commit-count` | Trigger after N commits | `10` |
+| `flink-maintenance.delete-orphan-files.schedule.data-file-count` | Trigger after N data files | `1000` |
+| `flink-maintenance.delete-orphan-files.schedule.data-file-size` | Trigger after total data file size (bytes) | `107374182400` (100GB) |
+| `flink-maintenance.delete-orphan-files.schedule.interval-second` | Trigger after time interval (seconds) | `600` |
+| `flink-maintenance.delete-orphan-files.min-age-seconds` | Minimum age of files to consider for deletion (seconds) | `259200` (3 days) |
+| `flink-maintenance.delete-orphan-files.delete-batch-size` | Batch size for deleting orphan files | `1000` |
+| `flink-maintenance.delete-orphan-files.location` | Location to start recursive listing | Table location |
+| `flink-maintenance.delete-orphan-files.use-prefix-listing` | Use prefix listing for file discovery | `false` |
+| `flink-maintenance.delete-orphan-files.planning-worker-pool-size` | Worker pool size for planning | Shared pool |
+| `flink-maintenance.delete-orphan-files.equal-schemes` | Equivalent schemes (format: `s3n=s3,s3a=s3`) | Not set |
Review Comment:
We have default value
```
private Map<String, String> equalSchemes =
Maps.newHashMap(
ImmutableMap.of(
"s3n", "s3",
"s3a", "s3"));
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +189,41 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnCommitCount(deleteOrphanFilesConfig.scheduleOnCommitCount())
+ .scheduleOnDataFileCount(deleteOrphanFilesConfig.scheduleOnDataFileCount())
+ .scheduleOnDataFileSize(deleteOrphanFilesConfig.scheduleOnDataFileSize())
+ .scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
+ this.location(loc);
+ }
+
+ Integer poolSize = deleteOrphanFilesConfig.planningWorkerPoolSize();
+ if (poolSize != null) {
+ this.planningWorkerPoolSize(poolSize);
+ }
+
+ Map<String, String> schemes = deleteOrphanFilesConfig.equalSchemes();
+ if (schemes != null) {
+ this.equalSchemes(schemes);
+ }
Review Comment:
We have default value in https://github.com/apache/iceberg/blob/bd54026d732a1eac49d0a6852036d1cd113ea65d/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java#L83-L87.
Does this need to be changed to an append instead of an overwrite here?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939779340
##########
docs/docs/flink-maintenance.md:
##########
@@ -207,7 +207,7 @@ env.execute("Table Maintenance Job");
| `retainLast(int)` | Minimum number of snapshots to retain | 1 | int |
| `deleteBatchSize(int)` | Number of files to delete in each batch | 1000 | int |
| `planningWorkerPoolSize(int)` | Number of worker threads for planning snapshot expiration | Shared worker pool | int |
-| `cleanExpiredMetadata(boolean)` | Remove expired metadata files when expiring snapshots | false | boolean |
+| `cleanExpiredMetadata(boolean)` | Remove expired metadata files when expiring snapshots | true | boolean |
Review Comment:
I see that we forget to set `TableMaintenance` `@Experimantal`. What does Flink says about config changes in minor releases?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939794451
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(boolean enabled, Map<String, String> config) {
+ rewriteDataFiles(enabled);
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable expire snapshots
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), Boolean.toString(enabled));
Review Comment:
That makes sense. I've made these adjustments based on your feedback:
ExpireSnapshotsConfig
- schedule.commit-count: 10
- schedule.interval-second: 3600 (1 hour)
- max-snapshot-age-seconds: not set
- retain-last: not set
- delete-batch-size: 1000
- clean-expired-metadata: true
- planning-worker-pool-size: not set (shared pool)
DeleteOrphanFilesConfig
- schedule.interval-second: 3600 (1 hour)
- min-age-seconds: 259200 (3 days)
- delete-batch-size: 1000
- location: not set (table location)
- use-prefix-listing: true
- planning-worker-pool-size: not set (shared pool)
- equal-schemes: s3n=s3, s3a=s3
- equal-authorities: not set
- prefix-mismatch-mode: ERROR
> For max-snapshot-age-seconds and retain-last, the defaults are both null—do we need to set a default value for them?
I find it difficult to set defaults for these, this is usually configured per table. If we don't set that, the table values will be used, which looks like the right thing to do.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2941096699
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,35 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
Review Comment:
No, removed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2924631426
##########
docs/docs/flink-maintenance.md:
##########
@@ -302,19 +302,50 @@ public class TableMaintenanceJob {
Apache Iceberg Sink V2 for Flink allows automatic execution of maintenance tasks after data is committed to the table, using the addPostCommitTopology(...) method.
+#### DataStream API
+
+##### Builder
+
+```java
+IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .rewriteDataFiles(true, Map.of(
Review Comment:
Not really, it is an existing pattern, but I don't know why you would turn off the maintenance task again, and the default will always be false.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2926024978
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
Review Comment:
+1 on this. I think the best API would be to let the user choose what kind of "compaction task" they want to enable. I'd even suggest to mark for deletion in the next release (or at least the next to next release)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940866264
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.FlinkConfParser;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+public class DeleteOrphanFilesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "delete-orphan-files.";
+
+ private static final Splitter COMMA_SPLITTER = Splitter.on(",");
+ private static final Splitter EQUALS_SPLITTER = Splitter.on("=").limit(2);
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .defaultValue(60 * 60L) // Default is 1 hour
+ .withDescription(
+ "The time interval (in seconds) between two consecutive delete orphan files operations.");
+
+ public static final String MIN_AGE_SECONDS = PREFIX + "min-age-seconds";
+ public static final ConfigOption<Long> MIN_AGE_SECONDS_OPTION =
+ ConfigOptions.key(MIN_AGE_SECONDS)
+ .longType()
+ .defaultValue(3L * 24 * 60 * 60) // Default is 3 days
+ .withDescription(
+ "The minimum age (in seconds) of files to be considered for deletion. "
+ + "Files newer than this will not be removed.");
+
+ public static final String DELETE_BATCH_SIZE = PREFIX + "delete-batch-size";
+ public static final ConfigOption<Integer> DELETE_BATCH_SIZE_OPTION =
+ ConfigOptions.key(DELETE_BATCH_SIZE)
+ .intType()
+ .defaultValue(1000)
+ .withDescription("The batch size used for deleting orphan files.");
+
+ public static final String LOCATION = PREFIX + "location";
+ public static final ConfigOption<String> LOCATION_OPTION =
+ ConfigOptions.key(LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location to start recursive listing of candidate files for removal. "
+ + "By default, the table location is used.");
+
+ public static final String USE_PREFIX_LISTING = PREFIX + "use-prefix-listing";
+ public static final ConfigOption<Boolean> USE_PREFIX_LISTING_OPTION =
+ ConfigOptions.key(USE_PREFIX_LISTING)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to use prefix listing when listing files from the file system.");
+
+ public static final String PLANNING_WORKER_POOL_SIZE = PREFIX + "planning-worker-pool-size";
+ public static final ConfigOption<Integer> PLANNING_WORKER_POOL_SIZE_OPTION =
+ ConfigOptions.key(PLANNING_WORKER_POOL_SIZE)
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The worker pool size used for planning the scan of the ALL_FILES table. "
+ + "If not set, the shared worker pool is used.");
+
+ public static final String EQUAL_SCHEMES = PREFIX + "equal-schemes";
+ public static final ConfigOption<String> EQUAL_SCHEMES_OPTION =
+ ConfigOptions.key(EQUAL_SCHEMES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Schemes that should be considered equal, in the format 'scheme1=scheme2,scheme3=scheme4'.");
+
+ public static final String EQUAL_AUTHORITIES = PREFIX + "equal-authorities";
+ public static final ConfigOption<String> EQUAL_AUTHORITIES_OPTION =
+ ConfigOptions.key(EQUAL_AUTHORITIES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Authorities that should be considered equal, in the format 'auth1=auth2,auth3=auth4'.");
+
+ public static final String PREFIX_MISMATCH_MODE = PREFIX + "prefix-mismatch-mode";
+ public static final ConfigOption<String> PREFIX_MISMATCH_MODE_OPTION =
+ ConfigOptions.key(PREFIX_MISMATCH_MODE)
+ .stringType()
+ .defaultValue(PrefixMismatchMode.ERROR.name())
+ .withDescription(
+ "Action behavior when location prefixes (schemes/authorities) mismatch. "
+ + "Valid values: ERROR, IGNORE, DELETE.");
+
+ private final FlinkConfParser confParser;
+
+ public DeleteOrphanFilesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ public long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .defaultValue(SCHEDULE_ON_INTERVAL_SECOND_OPTION.defaultValue())
+ .parse();
+ }
+
+ public long minAgeSeconds() {
+ return confParser
+ .longConf()
+ .option(MIN_AGE_SECONDS)
+ .flinkConfig(MIN_AGE_SECONDS_OPTION)
+ .defaultValue(MIN_AGE_SECONDS_OPTION.defaultValue())
+ .parse();
+ }
+
+ public int deleteBatchSize() {
+ return confParser
+ .intConf()
+ .option(DELETE_BATCH_SIZE)
+ .flinkConfig(DELETE_BATCH_SIZE_OPTION)
+ .defaultValue(DELETE_BATCH_SIZE_OPTION.defaultValue())
+ .parse();
+ }
+
+ public String location() {
+ return confParser.stringConf().option(LOCATION).flinkConfig(LOCATION_OPTION).parseOptional();
+ }
+
+ public boolean usePrefixListing() {
+ return confParser
+ .booleanConf()
+ .option(USE_PREFIX_LISTING)
+ .flinkConfig(USE_PREFIX_LISTING_OPTION)
+ .defaultValue(USE_PREFIX_LISTING_OPTION.defaultValue())
+ .parse();
+ }
+
+ public Integer planningWorkerPoolSize() {
+ return confParser
+ .intConf()
+ .option(PLANNING_WORKER_POOL_SIZE)
+ .flinkConfig(PLANNING_WORKER_POOL_SIZE_OPTION)
+ .parseOptional();
+ }
+
+ public Map<String, String> equalSchemes() {
+ String equalSchemes =
+ confParser
+ .stringConf()
+ .option(EQUAL_SCHEMES)
+ .flinkConfig(EQUAL_SCHEMES_OPTION)
+ .parseOptional();
+ if (equalSchemes != null) {
+ return parseKeyValuePairs(equalSchemes);
+ }
+
+ return Maps.newHashMap(DeleteOrphanFiles.DEFAULT_EQUAL_SCHEMES);
+ }
+
+ public Map<String, String> equalAuthorities() {
+ String value =
+ confParser
+ .stringConf()
+ .option(EQUAL_AUTHORITIES)
+ .flinkConfig(EQUAL_AUTHORITIES_OPTION)
+ .parseOptional();
+ return value != null ? parseKeyValuePairs(value) : null;
+ }
+
+ public PrefixMismatchMode prefixMismatchMode() {
+ String value =
+ confParser
+ .stringConf()
+ .option(PREFIX_MISMATCH_MODE)
+ .flinkConfig(PREFIX_MISMATCH_MODE_OPTION)
+ .defaultValue(PREFIX_MISMATCH_MODE_OPTION.defaultValue())
+ .parse();
+ return PrefixMismatchMode.valueOf(value);
+ }
+
+ private static Map<String, String> parseKeyValuePairs(String value) {
+ Map<String, String> result = Maps.newHashMap();
+ for (String pair : COMMA_SPLITTER.split(value)) {
+ List<String> parts = EQUALS_SPLITTER.splitToList(pair);
+ if (parts.size() == 2) {
Review Comment:
What if we don't have 2 parts. Should we throw?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940818914
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,85 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles() {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(Map<String, String> config) {
+ rewriteDataFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables expire snapshots as a post-commit maintenance task.
+ *
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots() {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link ExpireSnapshotsConfig} for available
+ * keys
+ */
+ public Builder expireSnapshots(Map<String, String> config) {
+ expireSnapshots();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @see DeleteOrphanFilesConfig for the default config.
+ */
+ public Builder deleteOrphanFiles() {
+ writeOptions.put(FlinkWriteOptions.DELETE_ORPHAN_FILES_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link DeleteOrphanFilesConfig} for available
+ * keys.
+ */
+ public Builder deleteOrphanFiles(Map<String, String> config) {
+ deleteOrphanFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
@Override
public Builder toBranch(String branch) {
Review Comment:
+1 on follow-up
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2941097509
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,35 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
+ this.location(loc);
+ }
+
+ Integer poolSize = deleteOrphanFilesConfig.planningWorkerPoolSize();
+ if (poolSize != null) {
+ this.planningWorkerPoolSize(poolSize);
+ }
+
+ this.equalSchemes(deleteOrphanFilesConfig.equalSchemes());
+
+ Map<String, String> authorities = deleteOrphanFilesConfig.equalAuthorities();
+ if (authorities != null) {
Review Comment:
No, removed.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,35 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
+ this.location(loc);
+ }
+
+ Integer poolSize = deleteOrphanFilesConfig.planningWorkerPoolSize();
Review Comment:
+1
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2945998189
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,23 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ this.location(deleteOrphanFilesConfig.location());
+ this.planningWorkerPoolSize(deleteOrphanFilesConfig.planningWorkerPoolSize());
+ this.equalSchemes(deleteOrphanFilesConfig.equalSchemes());
+ this.equalAuthorities(deleteOrphanFilesConfig.equalAuthorities());
+
+ return this;
Review Comment:
```suggestion
return this.scheduleOnInterval(
Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()))
.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
.deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
.usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
.prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode())
.location(deleteOrphanFilesConfig.location())
.planningWorkerPoolSize(deleteOrphanFilesConfig.planningWorkerPoolSize())
.equalSchemes(deleteOrphanFilesConfig.equalSchemes())
.equalAuthorities(deleteOrphanFilesConfig.equalAuthorities());
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2917598835
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java:
##########
@@ -82,7 +85,18 @@ private FlinkWriteOptions() {}
ConfigOptions.key("write-parallelism").intType().noDefaultValue();
public static final ConfigOption<Boolean> COMPACTION_ENABLE =
- ConfigOptions.key("compaction-enabled").booleanType().defaultValue(false);
+ ConfigOptions.key(RewriteDataFilesConfig.PREFIX + "enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDeprecatedKeys("compaction-enabled");
Review Comment:
I moved the enabled flag behind the config prefix.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
Review Comment:
I deprecated this method in favor of `rewriteDataFiles`. I think it's better to use the actual maintenance task names, to avoid confusion.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#issuecomment-4045894770
Rebased after conflicts with #15575.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #15566:
URL: https://github.com/apache/iceberg/pull/15566
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2941099686
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java:
##########
@@ -107,6 +107,35 @@ public Builder cleanExpiredMetadata(boolean newCleanExpiredMetadata) {
return this;
}
+ public Builder config(ExpireSnapshotsConfig expireSnapshotsConfig) {
+ this.scheduleOnCommitCount(expireSnapshotsConfig.scheduleOnCommitCount())
+ .scheduleOnInterval(Duration.ofSeconds(expireSnapshotsConfig.scheduleOnIntervalSecond()));
+
+ this.deleteBatchSize(expireSnapshotsConfig.deleteBatchSize());
+
+ Long maxSnapshotAgeSec = expireSnapshotsConfig.maxSnapshotAgeSeconds();
Review Comment:
Adjusted accordingly.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.FlinkConfParser;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+public class DeleteOrphanFilesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "delete-orphan-files.";
+
+ private static final Splitter COMMA_SPLITTER = Splitter.on(",");
+ private static final Splitter EQUALS_SPLITTER = Splitter.on("=").limit(2);
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .defaultValue(60 * 60L) // Default is 1 hour
+ .withDescription(
+ "The time interval (in seconds) between two consecutive delete orphan files operations.");
+
+ public static final String MIN_AGE_SECONDS = PREFIX + "min-age-seconds";
+ public static final ConfigOption<Long> MIN_AGE_SECONDS_OPTION =
+ ConfigOptions.key(MIN_AGE_SECONDS)
+ .longType()
+ .defaultValue(3L * 24 * 60 * 60) // Default is 3 days
+ .withDescription(
+ "The minimum age (in seconds) of files to be considered for deletion. "
+ + "Files newer than this will not be removed.");
+
+ public static final String DELETE_BATCH_SIZE = PREFIX + "delete-batch-size";
+ public static final ConfigOption<Integer> DELETE_BATCH_SIZE_OPTION =
+ ConfigOptions.key(DELETE_BATCH_SIZE)
+ .intType()
+ .defaultValue(1000)
+ .withDescription("The batch size used for deleting orphan files.");
+
+ public static final String LOCATION = PREFIX + "location";
+ public static final ConfigOption<String> LOCATION_OPTION =
+ ConfigOptions.key(LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location to start recursive listing of candidate files for removal. "
+ + "By default, the table location is used.");
+
+ public static final String USE_PREFIX_LISTING = PREFIX + "use-prefix-listing";
+ public static final ConfigOption<Boolean> USE_PREFIX_LISTING_OPTION =
+ ConfigOptions.key(USE_PREFIX_LISTING)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to use prefix listing when listing files from the file system.");
+
+ public static final String PLANNING_WORKER_POOL_SIZE = PREFIX + "planning-worker-pool-size";
+ public static final ConfigOption<Integer> PLANNING_WORKER_POOL_SIZE_OPTION =
+ ConfigOptions.key(PLANNING_WORKER_POOL_SIZE)
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The worker pool size used for planning the scan of the ALL_FILES table. "
+ + "If not set, the shared worker pool is used.");
+
+ public static final String EQUAL_SCHEMES = PREFIX + "equal-schemes";
+ public static final ConfigOption<String> EQUAL_SCHEMES_OPTION =
+ ConfigOptions.key(EQUAL_SCHEMES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Schemes that should be considered equal, in the format 'scheme1=scheme2,scheme3=scheme4'.");
+
+ public static final String EQUAL_AUTHORITIES = PREFIX + "equal-authorities";
+ public static final ConfigOption<String> EQUAL_AUTHORITIES_OPTION =
+ ConfigOptions.key(EQUAL_AUTHORITIES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Authorities that should be considered equal, in the format 'auth1=auth2,auth3=auth4'.");
+
+ public static final String PREFIX_MISMATCH_MODE = PREFIX + "prefix-mismatch-mode";
+ public static final ConfigOption<String> PREFIX_MISMATCH_MODE_OPTION =
+ ConfigOptions.key(PREFIX_MISMATCH_MODE)
+ .stringType()
+ .defaultValue(PrefixMismatchMode.ERROR.name())
+ .withDescription(
+ "Action behavior when location prefixes (schemes/authorities) mismatch. "
+ + "Valid values: ERROR, IGNORE, DELETE.");
+
+ private final FlinkConfParser confParser;
+
+ public DeleteOrphanFilesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ public long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .defaultValue(SCHEDULE_ON_INTERVAL_SECOND_OPTION.defaultValue())
+ .parse();
+ }
+
+ public long minAgeSeconds() {
+ return confParser
+ .longConf()
+ .option(MIN_AGE_SECONDS)
+ .flinkConfig(MIN_AGE_SECONDS_OPTION)
+ .defaultValue(MIN_AGE_SECONDS_OPTION.defaultValue())
+ .parse();
+ }
+
+ public int deleteBatchSize() {
+ return confParser
+ .intConf()
+ .option(DELETE_BATCH_SIZE)
+ .flinkConfig(DELETE_BATCH_SIZE_OPTION)
+ .defaultValue(DELETE_BATCH_SIZE_OPTION.defaultValue())
+ .parse();
+ }
+
+ public String location() {
+ return confParser.stringConf().option(LOCATION).flinkConfig(LOCATION_OPTION).parseOptional();
+ }
+
+ public boolean usePrefixListing() {
+ return confParser
+ .booleanConf()
+ .option(USE_PREFIX_LISTING)
+ .flinkConfig(USE_PREFIX_LISTING_OPTION)
+ .defaultValue(USE_PREFIX_LISTING_OPTION.defaultValue())
+ .parse();
+ }
+
+ public Integer planningWorkerPoolSize() {
+ return confParser
+ .intConf()
+ .option(PLANNING_WORKER_POOL_SIZE)
+ .flinkConfig(PLANNING_WORKER_POOL_SIZE_OPTION)
+ .parseOptional();
+ }
+
+ public Map<String, String> equalSchemes() {
+ String equalSchemes =
+ confParser
+ .stringConf()
+ .option(EQUAL_SCHEMES)
+ .flinkConfig(EQUAL_SCHEMES_OPTION)
+ .parseOptional();
+ if (equalSchemes != null) {
+ return parseKeyValuePairs(equalSchemes);
+ }
+
+ return Maps.newHashMap(DeleteOrphanFiles.DEFAULT_EQUAL_SCHEMES);
Review Comment:
Changed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2946021149
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestDeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.Map;
+import org.apache.flink.configuration.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ThreadPools;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestDeleteOrphanFilesConfig extends OperatorTestBase {
+ private Table table;
+ private Map<String, String> input = Maps.newHashMap();
+
+ @BeforeEach
+ public void before() {
+ this.table = createTable();
+ input.put(DeleteOrphanFilesConfig.SCHEDULE_ON_INTERVAL_SECOND, "60");
+ input.put(DeleteOrphanFilesConfig.MIN_AGE_SECONDS, "86400");
+ input.put(DeleteOrphanFilesConfig.DELETE_BATCH_SIZE, "500");
+ input.put(DeleteOrphanFilesConfig.LOCATION, "/tmp/test-location");
+ input.put(DeleteOrphanFilesConfig.USE_PREFIX_LISTING, "true");
+ input.put(DeleteOrphanFilesConfig.PLANNING_WORKER_POOL_SIZE, "4");
+ input.put(DeleteOrphanFilesConfig.EQUAL_SCHEMES, "s3n=s3,s3a=s3");
+ input.put(DeleteOrphanFilesConfig.EQUAL_AUTHORITIES, "auth1=auth2");
+ input.put(DeleteOrphanFilesConfig.PREFIX_MISMATCH_MODE, "IGNORE");
+ input.put("other.config", "should-be-ignored");
+ }
+
+ @AfterEach
+ public void after() {
+ input.clear();
+ }
+
+ @Test
+ void testConfigParsing() {
+ DeleteOrphanFilesConfig config = new DeleteOrphanFilesConfig(table, input, new Configuration());
+
+ assertThat(config.scheduleOnIntervalSecond()).isEqualTo(60);
+ assertThat(config.minAgeSeconds()).isEqualTo(86400L);
+ assertThat(config.deleteBatchSize()).isEqualTo(500);
+ assertThat(config.location()).isEqualTo("/tmp/test-location");
+ assertThat(config.usePrefixListing()).isTrue();
+ assertThat(config.planningWorkerPoolSize()).isEqualTo(4);
+ assertThat(config.equalSchemes()).containsEntry("s3n", "s3").containsEntry("s3a", "s3");
+ assertThat(config.equalAuthorities()).containsEntry("auth1", "auth2");
+ assertThat(config.prefixMismatchMode()).isEqualTo(PrefixMismatchMode.IGNORE);
+ }
+
+ @Test
+ void testConfigDefaults() {
+ DeleteOrphanFilesConfig config =
+ new DeleteOrphanFilesConfig(table, Maps.newHashMap(), new Configuration());
+
+ assertThat(config.scheduleOnIntervalSecond())
+ .isEqualTo(DeleteOrphanFilesConfig.SCHEDULE_ON_INTERVAL_SECOND_OPTION.defaultValue());
+ assertThat(config.minAgeSeconds())
+ .isEqualTo(DeleteOrphanFilesConfig.MIN_AGE_SECONDS_OPTION.defaultValue());
+ assertThat(config.deleteBatchSize())
+ .isEqualTo(DeleteOrphanFilesConfig.DELETE_BATCH_SIZE_OPTION.defaultValue());
+ assertThat(config.location()).isNull();
+ assertThat(config.usePrefixListing()).isTrue();
+ assertThat(config.planningWorkerPoolSize()).isEqualTo(ThreadPools.WORKER_THREAD_POOL_SIZE);
+ assertThat(config.equalSchemes()).containsEntry("s3n", "s3").containsEntry("s3a", "s3");
+ assertThat(config.equalAuthorities()).isEqualTo(Collections.emptyMap());
Review Comment:
Could we use `Map.of()`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940842279
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,35 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
+ this.location(loc);
+ }
+
+ Integer poolSize = deleteOrphanFilesConfig.planningWorkerPoolSize();
+ if (poolSize != null) {
+ this.planningWorkerPoolSize(poolSize);
+ }
+
+ this.equalSchemes(deleteOrphanFilesConfig.equalSchemes());
+
+ Map<String, String> authorities = deleteOrphanFilesConfig.equalAuthorities();
+ if (authorities != null) {
Review Comment:
Do we need this null check?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2917591978
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
Update pushed via:
- [fixup! Configure ExpireSnapshots and DeleteOrphanFiles via config](https://github.com/apache/iceberg/pull/15566/commits/9068b8ad309d25769330d866a8c4d1b8887e91bc)
[9068b8a](https://github.com/apache/iceberg/pull/15566/commits/9068b8ad309d25769330d866a8c4d1b8887e91bc)
- [fixup! docs](https://github.com/apache/iceberg/pull/15566/commits/04d190d18f421e648d83c516a454ffed920b423a)
[04d190d](https://github.com/apache/iceberg/pull/15566/commits/04d190d18f421e648d83c516a454ffed920b423a)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#issuecomment-4038450912
Build is currently failing with
```
[Incubating] Problems report is available at: file:///home/runner/work/iceberg/iceberg/build/reports/problems/problems-report.html
* What went wrong:
Execution failed for task ':iceberg-bigquery:compileJava'.
> Could not resolve all files for configuration ':iceberg-bigquery:compileClasspath'.
> Could not resolve com.google.cloud.opentelemetry:exporter-metrics:0.33.0.
Required by:
project :iceberg-bigquery > com.google.cloud:google-cloud-storage:2.64.0
> Could not resolve com.google.cloud.opentelemetry:exporter-metrics:0.33.0.
> Could not parse POM https://repo.maven.apache.org/maven2/com/google/cloud/opentelemetry/exporter-metrics/0.33.0/exporter-metrics-0.33.0.pom
```
https://github.com/apache/iceberg/actions/runs/22949338400/job/66609723276?pr=15566
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940895536
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java:
##########
@@ -107,6 +107,35 @@ public Builder cleanExpiredMetadata(boolean newCleanExpiredMetadata) {
return this;
}
+ public Builder config(ExpireSnapshotsConfig expireSnapshotsConfig) {
+ this.scheduleOnCommitCount(expireSnapshotsConfig.scheduleOnCommitCount())
+ .scheduleOnInterval(Duration.ofSeconds(expireSnapshotsConfig.scheduleOnIntervalSecond()));
+
+ this.deleteBatchSize(expireSnapshotsConfig.deleteBatchSize());
+
+ Long maxSnapshotAgeSec = expireSnapshotsConfig.maxSnapshotAgeSeconds();
Review Comment:
We own `ExpireSnapshots` and `ExpireSnapshotsConfig` as well. Shall we synchronize the configs and send null instead of checking it every time?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2946019398
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.FlinkConfParser;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ThreadPools;
+
+public class DeleteOrphanFilesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "delete-orphan-files.";
+
+ private static final Splitter COMMA_SPLITTER = Splitter.on(",");
+ private static final Splitter EQUALS_SPLITTER = Splitter.on("=").limit(2);
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .defaultValue(60 * 60L) // Default is 1 hour
+ .withDescription(
+ "The time interval (in seconds) between two consecutive delete orphan files operations.");
+
+ public static final String MIN_AGE_SECONDS = PREFIX + "min-age-seconds";
+ public static final ConfigOption<Long> MIN_AGE_SECONDS_OPTION =
+ ConfigOptions.key(MIN_AGE_SECONDS)
+ .longType()
+ .defaultValue(3L * 24 * 60 * 60) // Default is 3 days
+ .withDescription(
+ "The minimum age (in seconds) of files to be considered for deletion. "
+ + "Files newer than this will not be removed.");
+
+ public static final String DELETE_BATCH_SIZE = PREFIX + "delete-batch-size";
+ public static final ConfigOption<Integer> DELETE_BATCH_SIZE_OPTION =
+ ConfigOptions.key(DELETE_BATCH_SIZE)
+ .intType()
+ .defaultValue(1000)
+ .withDescription("The batch size used for deleting orphan files.");
+
+ public static final String LOCATION = PREFIX + "location";
+ public static final ConfigOption<String> LOCATION_OPTION =
+ ConfigOptions.key(LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location to start recursive listing of candidate files for removal. "
+ + "By default, the table location is used.");
+
+ public static final String USE_PREFIX_LISTING = PREFIX + "use-prefix-listing";
+ public static final ConfigOption<Boolean> USE_PREFIX_LISTING_OPTION =
+ ConfigOptions.key(USE_PREFIX_LISTING)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to use prefix listing when listing files from the file system.");
+
+ public static final String PLANNING_WORKER_POOL_SIZE = PREFIX + "planning-worker-pool-size";
+ public static final ConfigOption<Integer> PLANNING_WORKER_POOL_SIZE_OPTION =
+ ConfigOptions.key(PLANNING_WORKER_POOL_SIZE)
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The worker pool size used for planning the scan of the ALL_FILES table. "
+ + "If not set, the shared worker pool is used.");
+
+ public static final String EQUAL_SCHEMES = PREFIX + "equal-schemes";
+ public static final ConfigOption<String> EQUAL_SCHEMES_OPTION =
+ ConfigOptions.key(EQUAL_SCHEMES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Schemes that should be considered equal, in the format 'scheme1=scheme2,scheme3=scheme4'.");
+
+ public static final String EQUAL_AUTHORITIES = PREFIX + "equal-authorities";
+ public static final ConfigOption<String> EQUAL_AUTHORITIES_OPTION =
+ ConfigOptions.key(EQUAL_AUTHORITIES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Authorities that should be considered equal, in the format 'auth1=auth2,auth3=auth4'.");
+
+ public static final String PREFIX_MISMATCH_MODE = PREFIX + "prefix-mismatch-mode";
+ public static final ConfigOption<String> PREFIX_MISMATCH_MODE_OPTION =
+ ConfigOptions.key(PREFIX_MISMATCH_MODE)
+ .stringType()
+ .defaultValue(PrefixMismatchMode.ERROR.name())
+ .withDescription(
+ "Action behavior when location prefixes (schemes/authorities) mismatch. "
+ + "Valid values: ERROR, IGNORE, DELETE.");
+
+ private final FlinkConfParser confParser;
+
+ public DeleteOrphanFilesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ public long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .defaultValue(SCHEDULE_ON_INTERVAL_SECOND_OPTION.defaultValue())
+ .parse();
+ }
+
+ public long minAgeSeconds() {
+ return confParser
+ .longConf()
+ .option(MIN_AGE_SECONDS)
+ .flinkConfig(MIN_AGE_SECONDS_OPTION)
+ .defaultValue(MIN_AGE_SECONDS_OPTION.defaultValue())
+ .parse();
+ }
+
+ public int deleteBatchSize() {
+ return confParser
+ .intConf()
+ .option(DELETE_BATCH_SIZE)
+ .flinkConfig(DELETE_BATCH_SIZE_OPTION)
+ .defaultValue(DELETE_BATCH_SIZE_OPTION.defaultValue())
+ .parse();
+ }
+
+ public String location() {
+ return confParser.stringConf().option(LOCATION).flinkConfig(LOCATION_OPTION).parseOptional();
+ }
+
+ public boolean usePrefixListing() {
+ return confParser
+ .booleanConf()
+ .option(USE_PREFIX_LISTING)
+ .flinkConfig(USE_PREFIX_LISTING_OPTION)
+ .defaultValue(USE_PREFIX_LISTING_OPTION.defaultValue())
+ .parse();
+ }
+
+ public Integer planningWorkerPoolSize() {
+ return confParser
+ .intConf()
+ .option(PLANNING_WORKER_POOL_SIZE)
+ .flinkConfig(PLANNING_WORKER_POOL_SIZE_OPTION)
+ .defaultValue(ThreadPools.WORKER_THREAD_POOL_SIZE)
+ .parse();
+ }
+
+ public Map<String, String> equalSchemes() {
+ String equalSchemes =
+ confParser
+ .stringConf()
+ .option(EQUAL_SCHEMES)
+ .flinkConfig(EQUAL_SCHEMES_OPTION)
+ .parseOptional();
+
+ return equalSchemes != null
+ ? parseKeyValuePairs(equalSchemes)
+ : Maps.newHashMap(DeleteOrphanFiles.DEFAULT_EQUAL_SCHEMES);
+ }
+
+ public Map<String, String> equalAuthorities() {
+ String equalAuthorities =
+ confParser
+ .stringConf()
+ .option(EQUAL_AUTHORITIES)
+ .flinkConfig(EQUAL_AUTHORITIES_OPTION)
+ .parseOptional();
+
+ return equalAuthorities != null ? parseKeyValuePairs(equalAuthorities) : Collections.emptyMap();
Review Comment:
Could we use `Map.of()`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2931095026
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(boolean enabled, Map<String, String> config) {
+ rewriteDataFiles(enabled);
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable expire snapshots
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), Boolean.toString(enabled));
Review Comment:
> Defaults are hard. These are the current defaults:
>
> ExpireSnapshotsConfig defaults:
>
> * schedule.commit-count: 10
> * schedule.data-file-count: 1,000
> * schedule.data-file-size: 100 GB
> * schedule.interval-second: 3600 (1 hour)
> * max-snapshot-age-seconds: none
> * retain-last: none
> * delete-batch-size: 1,000
> * clean-expired-metadata: false
> * planning-worker-pool-size: none (shared pool)
Can we remove `schedule.data-file-count`, `schedule.data-file-size` - I don't thing we need those for expire snapshots.
Maybe we should set `clean-expired-metadata`? I think keeping is only makes sense when we want to keep the old functionality. We are introducing a new one. We should remove unused metadata as soon as possible.
> DeleteOrphanFilesConfig defaults:
>
> * schedule.commit-count: 10
> * schedule.data-file-count: 1,000
> * schedule.data-file-size: 100 GB
> * schedule.interval-second: 3600 (1 hour)
> * min-age-seconds: 259,200 (3 days)
> * delete-batch-size: 1,000
> * use-prefix-listing: false
> * planning-worker-pool-size: none (shared pool)
> * equal-schemes: none
> * equal-authorities: none
> * prefix-mismatch-mode: ERROR
Maybe we could remove all of the `schedule` configs other than the interval. The others shouldn't have to do anything with the delete orphan files
Maybe setting `use-prefix-listing` to true would be better. It should perform better in most cases, and only edge cases need the recursive listing.
What do you think about these @Guosmilesmile?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940874366
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.FlinkConfParser;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+public class DeleteOrphanFilesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "delete-orphan-files.";
+
+ private static final Splitter COMMA_SPLITTER = Splitter.on(",");
+ private static final Splitter EQUALS_SPLITTER = Splitter.on("=").limit(2);
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .defaultValue(60 * 60L) // Default is 1 hour
+ .withDescription(
+ "The time interval (in seconds) between two consecutive delete orphan files operations.");
+
+ public static final String MIN_AGE_SECONDS = PREFIX + "min-age-seconds";
+ public static final ConfigOption<Long> MIN_AGE_SECONDS_OPTION =
+ ConfigOptions.key(MIN_AGE_SECONDS)
+ .longType()
+ .defaultValue(3L * 24 * 60 * 60) // Default is 3 days
+ .withDescription(
+ "The minimum age (in seconds) of files to be considered for deletion. "
+ + "Files newer than this will not be removed.");
+
+ public static final String DELETE_BATCH_SIZE = PREFIX + "delete-batch-size";
+ public static final ConfigOption<Integer> DELETE_BATCH_SIZE_OPTION =
+ ConfigOptions.key(DELETE_BATCH_SIZE)
+ .intType()
+ .defaultValue(1000)
+ .withDescription("The batch size used for deleting orphan files.");
+
+ public static final String LOCATION = PREFIX + "location";
+ public static final ConfigOption<String> LOCATION_OPTION =
+ ConfigOptions.key(LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location to start recursive listing of candidate files for removal. "
+ + "By default, the table location is used.");
+
+ public static final String USE_PREFIX_LISTING = PREFIX + "use-prefix-listing";
+ public static final ConfigOption<Boolean> USE_PREFIX_LISTING_OPTION =
+ ConfigOptions.key(USE_PREFIX_LISTING)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to use prefix listing when listing files from the file system.");
+
+ public static final String PLANNING_WORKER_POOL_SIZE = PREFIX + "planning-worker-pool-size";
+ public static final ConfigOption<Integer> PLANNING_WORKER_POOL_SIZE_OPTION =
+ ConfigOptions.key(PLANNING_WORKER_POOL_SIZE)
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The worker pool size used for planning the scan of the ALL_FILES table. "
+ + "If not set, the shared worker pool is used.");
+
+ public static final String EQUAL_SCHEMES = PREFIX + "equal-schemes";
+ public static final ConfigOption<String> EQUAL_SCHEMES_OPTION =
+ ConfigOptions.key(EQUAL_SCHEMES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Schemes that should be considered equal, in the format 'scheme1=scheme2,scheme3=scheme4'.");
+
+ public static final String EQUAL_AUTHORITIES = PREFIX + "equal-authorities";
+ public static final ConfigOption<String> EQUAL_AUTHORITIES_OPTION =
+ ConfigOptions.key(EQUAL_AUTHORITIES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Authorities that should be considered equal, in the format 'auth1=auth2,auth3=auth4'.");
+
+ public static final String PREFIX_MISMATCH_MODE = PREFIX + "prefix-mismatch-mode";
+ public static final ConfigOption<String> PREFIX_MISMATCH_MODE_OPTION =
+ ConfigOptions.key(PREFIX_MISMATCH_MODE)
+ .stringType()
+ .defaultValue(PrefixMismatchMode.ERROR.name())
+ .withDescription(
+ "Action behavior when location prefixes (schemes/authorities) mismatch. "
+ + "Valid values: ERROR, IGNORE, DELETE.");
+
+ private final FlinkConfParser confParser;
+
+ public DeleteOrphanFilesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ public long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .defaultValue(SCHEDULE_ON_INTERVAL_SECOND_OPTION.defaultValue())
+ .parse();
+ }
+
+ public long minAgeSeconds() {
+ return confParser
+ .longConf()
+ .option(MIN_AGE_SECONDS)
+ .flinkConfig(MIN_AGE_SECONDS_OPTION)
+ .defaultValue(MIN_AGE_SECONDS_OPTION.defaultValue())
+ .parse();
+ }
+
+ public int deleteBatchSize() {
+ return confParser
+ .intConf()
+ .option(DELETE_BATCH_SIZE)
+ .flinkConfig(DELETE_BATCH_SIZE_OPTION)
+ .defaultValue(DELETE_BATCH_SIZE_OPTION.defaultValue())
+ .parse();
+ }
+
+ public String location() {
+ return confParser.stringConf().option(LOCATION).flinkConfig(LOCATION_OPTION).parseOptional();
+ }
+
+ public boolean usePrefixListing() {
+ return confParser
+ .booleanConf()
+ .option(USE_PREFIX_LISTING)
+ .flinkConfig(USE_PREFIX_LISTING_OPTION)
+ .defaultValue(USE_PREFIX_LISTING_OPTION.defaultValue())
+ .parse();
+ }
+
+ public Integer planningWorkerPoolSize() {
+ return confParser
+ .intConf()
+ .option(PLANNING_WORKER_POOL_SIZE)
+ .flinkConfig(PLANNING_WORKER_POOL_SIZE_OPTION)
+ .parseOptional();
+ }
+
+ public Map<String, String> equalSchemes() {
+ String equalSchemes =
+ confParser
+ .stringConf()
+ .option(EQUAL_SCHEMES)
+ .flinkConfig(EQUAL_SCHEMES_OPTION)
+ .parseOptional();
+ if (equalSchemes != null) {
+ return parseKeyValuePairs(equalSchemes);
+ }
+
+ return Maps.newHashMap(DeleteOrphanFiles.DEFAULT_EQUAL_SCHEMES);
Review Comment:
Why not:
```suggestion
return equalSchemes != null ? parseKeyValuePairs(equalSchemes) : DeleteOrphanFiles.DEFAULT_EQUAL_SCHEMES;
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2946006072
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java:
##########
@@ -107,6 +108,22 @@ public Builder cleanExpiredMetadata(boolean newCleanExpiredMetadata) {
return this;
}
+ public Builder config(ExpireSnapshotsConfig expireSnapshotsConfig) {
+ this.scheduleOnCommitCount(expireSnapshotsConfig.scheduleOnCommitCount())
+ .scheduleOnInterval(Duration.ofSeconds(expireSnapshotsConfig.scheduleOnIntervalSecond()));
+
+ this.deleteBatchSize(expireSnapshotsConfig.deleteBatchSize());
+ this.maxSnapshotAge(
+ Optional.ofNullable(expireSnapshotsConfig.maxSnapshotAgeSeconds())
+ .map(Duration::ofSeconds)
+ .orElse(null));
+ this.retainLast(expireSnapshotsConfig.retainLast());
+ this.cleanExpiredMetadata(expireSnapshotsConfig.cleanExpiredMetadata());
+ this.planningWorkerPoolSize(expireSnapshotsConfig.planningWorkerPoolSize());
+
+ return this;
Review Comment:
```suggestion
return this.scheduleOnCommitCount(expireSnapshotsConfig.scheduleOnCommitCount())
.scheduleOnInterval(Duration.ofSeconds(expireSnapshotsConfig.scheduleOnIntervalSecond()))
.deleteBatchSize(expireSnapshotsConfig.deleteBatchSize())
.maxSnapshotAge(
Optional.ofNullable(expireSnapshotsConfig.maxSnapshotAgeSeconds())
.map(Duration::ofSeconds)
.orElse(null));
.retainLast(expireSnapshotsConfig.retainLast())
.cleanExpiredMetadata(expireSnapshotsConfig.cleanExpiredMetadata())
.planningWorkerPoolSize(expireSnapshotsConfig.planningWorkerPoolSize());
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939785036
##########
docs/docs/flink-maintenance.md:
##########
@@ -302,19 +302,50 @@ public class TableMaintenanceJob {
Apache Iceberg Sink V2 for Flink allows automatic execution of maintenance tasks after data is committed to the table, using the addPostCommitTopology(...) method.
+#### DataStream API
+
+##### Builder
+
+```java
+IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .rewriteDataFiles(Map.of(
+ RewriteDataFilesConfig.MAX_BYTES, "1073741824"))
+ .expireSnapshots(Map.of(
+ ExpireSnapshotsConfig.RETAIN_LAST, "5",
+ ExpireSnapshotsConfig.MAX_SNAPSHOT_AGE_SECONDS, "604800"))
+ .deleteOrphanFiles(Map.of(
+ DeleteOrphanFilesConfig.MIN_AGE_SECONDS, "259200"))
+ .append();
+```
+
+##### Config
+
+All maintenance tasks are configured through string properties:
+
```java
Map<String, String> flinkConf = new HashMap<>();
-// Enable compaction and maintenance features
-flinkConf.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+// Enable maintenance tasks
+flinkConf.put("flink-maintenance.rewrite.enabled", "true");
+flinkConf.put("flink-maintenance.expire-snapshots.enabled", "true");
+flinkConf.put("flink-maintenance.delete-orphan-files.enabled", "true");
-// Configure JDBC lock settings
-flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
-flinkConf.put(LockConfig.JdbcLockConfig.JDBC_URI_OPTION.key(), "jdbc:postgresql://localhost:5432/iceberg");
-flinkConf.put(LockConfig.LOCK_ID_OPTION.key(), "catalog.db.table");
+// Configure rewrite data files
+flinkConf.put("flink-maintenance.rewrite.max-bytes", "1073741824");
+
+// Configure expire snapshots
+flinkConf.put("flink-maintenance.expire-snapshots.retain-last", "5");
+flinkConf.put("flink-maintenance.expire-snapshots.max-snapshot-age-seconds", "604800");
-// Add any other maintenance-related options here as needed
-// ...
+// Configure delete orphan files
+flinkConf.put("flink-maintenance.delete-orphan-files.min-age-seconds", "259200");
+
+// Configure JDBC lock settings
Review Comment:
Do we want to say that the locks are deprecated?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940006961
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,85 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles() {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(Map<String, String> config) {
+ rewriteDataFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables expire snapshots as a post-commit maintenance task.
+ *
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots() {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link ExpireSnapshotsConfig} for available
+ * keys
+ */
+ public Builder expireSnapshots(Map<String, String> config) {
+ expireSnapshots();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @see DeleteOrphanFilesConfig for the default config.
+ */
+ public Builder deleteOrphanFiles() {
+ writeOptions.put(FlinkWriteOptions.DELETE_ORPHAN_FILES_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link DeleteOrphanFilesConfig} for available
+ * keys.
+ */
+ public Builder deleteOrphanFiles(Map<String, String> config) {
+ deleteOrphanFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
@Override
public Builder toBranch(String branch) {
Review Comment:
Should we do this in a follow-up?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939823246
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,85 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles() {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(Map<String, String> config) {
+ rewriteDataFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables expire snapshots as a post-commit maintenance task.
+ *
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots() {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link ExpireSnapshotsConfig} for available
+ * keys
+ */
+ public Builder expireSnapshots(Map<String, String> config) {
+ expireSnapshots();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @see DeleteOrphanFilesConfig for the default config.
+ */
+ public Builder deleteOrphanFiles() {
+ writeOptions.put(FlinkWriteOptions.DELETE_ORPHAN_FILES_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link DeleteOrphanFilesConfig} for available
+ * keys.
+ */
+ public Builder deleteOrphanFiles(Map<String, String> config) {
+ deleteOrphanFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
@Override
public Builder toBranch(String branch) {
Review Comment:
Do we compact the correct branch, if the `toBranch` is set?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939824693
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,85 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles() {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables rewriting data files (compaction) as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(Map<String, String> config) {
+ rewriteDataFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables expire snapshots as a post-commit maintenance task.
+ *
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots() {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link ExpireSnapshotsConfig} for available
+ * keys
+ */
+ public Builder expireSnapshots(Map<String, String> config) {
+ expireSnapshots();
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @see DeleteOrphanFilesConfig for the default config.
+ */
+ public Builder deleteOrphanFiles() {
+ writeOptions.put(FlinkWriteOptions.DELETE_ORPHAN_FILES_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables delete orphan files as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link DeleteOrphanFilesConfig} for available
+ * keys.
+ */
+ public Builder deleteOrphanFiles(Map<String, String> config) {
+ deleteOrphanFiles();
+ writeOptions.putAll(config);
+ return this;
+ }
+
@Override
public Builder toBranch(String branch) {
Review Comment:
Somewhat unrelated 😄
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939809617
##########
docs/docs/flink-maintenance.md:
##########
@@ -207,7 +207,7 @@ env.execute("Table Maintenance Job");
| `retainLast(int)` | Minimum number of snapshots to retain | 1 | int |
| `deleteBatchSize(int)` | Number of files to delete in each batch | 1000 | int |
| `planningWorkerPoolSize(int)` | Number of worker threads for planning snapshot expiration | Shared worker pool | int |
-| `cleanExpiredMetadata(boolean)` | Remove expired metadata files when expiring snapshots | false | boolean |
+| `cleanExpiredMetadata(boolean)` | Remove expired metadata files when expiring snapshots | true | boolean |
Review Comment:
Flink has some API guarantees listed here: https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/upgrading/, but those don't apply directly to configs. I think it should be fine.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2917591978
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
Update pushed via:
- [fixup! Configure ExpireSnapshots and DeleteOrphanFiles via config](https://github.com/apache/iceberg/pull/15566/commits/746902cb643981af293e096a0c75835cc2bfcfec)
- [fixup! docs](https://github.com/apache/iceberg/pull/15566/commits/65cef00889e626f99a4bb20d881ffad0a0a8a485)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2906564461
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
I think this solution conflicts with "add only a single way to do a single thing".
The user can enable compaction by adding `COMPACTION_ENABLE` but the same could be archived with setting the compaction in the `maintenance`. Do we really want to do this?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2906797077
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
What is the typical use-case?
The most wide config settings are like this:
```
public String orcCompressionStrategy() {
return confParser
.stringConf()
.option(FlinkWriteOptions.COMPRESSION_STRATEGY.key())
.flinkConfig(FlinkWriteOptions.COMPRESSION_STRATEGY)
.tableProperty(TableProperties.ORC_COMPRESSION_STRATEGY)
.defaultValue(TableProperties.ORC_COMPRESSION_STRATEGY_DEFAULT)
.parse();
}
```
Sometimes we allow the user to override the settings by manually adding options:
```
@Override
public Builder overwrite(boolean newOverwrite) {
writeOptions.put(FlinkWriteOptions.OVERWRITE_MODE.key(), Boolean.toString(newOverwrite));
return this;
}
```
We might even validate against setting them twice with a different java method:
```
public Builder<T> watermarkColumn(String columnName) {
Preconditions.checkArgument(
splitAssignerFactory == null,
"Watermark column and SplitAssigner should not be set in the same source");
readOptions.put(FlinkReadOptions.WATERMARK_COLUMN, columnName);
return this;
}
```
To support the property based solutions we likely to have:
- ExpireSnapshotsConfig
- DeleteOrphanFilesConfig
This means that setting `ExpireSnapshotsConfig` and also adding a new ExpireSnapshots tasks should only effect each other. If we create a generic `maintenance` method on `IcebergSink`, then we will have hard time matching those.
I think we need to think a bit more about this.
CC: @Guosmilesmile
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2924635384
##########
docs/docs/flink-maintenance.md:
##########
@@ -302,19 +302,50 @@ public class TableMaintenanceJob {
Apache Iceberg Sink V2 for Flink allows automatic execution of maintenance tasks after data is committed to the table, using the addPostCommitTopology(...) method.
+#### DataStream API
+
+##### Builder
+
+```java
+IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .rewriteDataFiles(true, Map.of(
Review Comment:
Removed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2946104049
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,23 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ this.location(deleteOrphanFilesConfig.location());
+ this.planningWorkerPoolSize(deleteOrphanFilesConfig.planningWorkerPoolSize());
+ this.equalSchemes(deleteOrphanFilesConfig.equalSchemes());
+ this.equalAuthorities(deleteOrphanFilesConfig.equalAuthorities());
+
+ return this;
Review Comment:
Yes!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2940837813
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +190,35 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
Review Comment:
Do we need this `if`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939795746
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFiles.java:
##########
@@ -189,6 +189,41 @@ public Builder deleteBatchSize(int newDeleteBatchSize) {
return this;
}
+ public Builder config(DeleteOrphanFilesConfig deleteOrphanFilesConfig) {
+ this.scheduleOnCommitCount(deleteOrphanFilesConfig.scheduleOnCommitCount())
+ .scheduleOnDataFileCount(deleteOrphanFilesConfig.scheduleOnDataFileCount())
+ .scheduleOnDataFileSize(deleteOrphanFilesConfig.scheduleOnDataFileSize())
+ .scheduleOnInterval(
+ Duration.ofSeconds(deleteOrphanFilesConfig.scheduleOnIntervalSecond()));
+
+ this.minAge(Duration.ofSeconds(deleteOrphanFilesConfig.minAgeSeconds()))
+ .deleteBatchSize(deleteOrphanFilesConfig.deleteBatchSize())
+ .usePrefixListing(deleteOrphanFilesConfig.usePrefixListing())
+ .prefixMismatchMode(deleteOrphanFilesConfig.prefixMismatchMode());
+
+ String loc = deleteOrphanFilesConfig.location();
+ if (loc != null) {
+ this.location(loc);
+ }
+
+ Integer poolSize = deleteOrphanFilesConfig.planningWorkerPoolSize();
+ if (poolSize != null) {
+ this.planningWorkerPoolSize(poolSize);
+ }
+
+ Map<String, String> schemes = deleteOrphanFilesConfig.equalSchemes();
+ if (schemes != null) {
+ this.equalSchemes(schemes);
+ }
Review Comment:
Changed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2939811044
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +768,26 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
+
+ if (flinkWriteConf.expireSnapshotsMode()) {
+ ExpireSnapshotsConfig expireSnapshotsConfig =
+ flinkMaintenanceConfig.createExpireSnapshotsConfig();
+ maintenanceTasks.add(ExpireSnapshots.builder().config(expireSnapshotsConfig));
+ }
+
+ if (flinkWriteConf.deleteOrphanFilesMode()) {
+ DeleteOrphanFilesConfig deleteOrphanFilesConfig =
+ flinkMaintenanceConfig.createDeleteOrphanFilesConfig();
+ maintenanceTasks.add(
+ org.apache.iceberg.flink.maintenance.api.DeleteOrphanFiles.builder()
Review Comment:
nit: import
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2946084590
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/DeleteOrphanFilesConfig.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.api;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
+import org.apache.iceberg.flink.FlinkConfParser;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ThreadPools;
+
+public class DeleteOrphanFilesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "delete-orphan-files.";
+
+ private static final Splitter COMMA_SPLITTER = Splitter.on(",");
+ private static final Splitter EQUALS_SPLITTER = Splitter.on("=").limit(2);
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .defaultValue(60 * 60L) // Default is 1 hour
+ .withDescription(
+ "The time interval (in seconds) between two consecutive delete orphan files operations.");
+
+ public static final String MIN_AGE_SECONDS = PREFIX + "min-age-seconds";
+ public static final ConfigOption<Long> MIN_AGE_SECONDS_OPTION =
+ ConfigOptions.key(MIN_AGE_SECONDS)
+ .longType()
+ .defaultValue(3L * 24 * 60 * 60) // Default is 3 days
+ .withDescription(
+ "The minimum age (in seconds) of files to be considered for deletion. "
+ + "Files newer than this will not be removed.");
+
+ public static final String DELETE_BATCH_SIZE = PREFIX + "delete-batch-size";
+ public static final ConfigOption<Integer> DELETE_BATCH_SIZE_OPTION =
+ ConfigOptions.key(DELETE_BATCH_SIZE)
+ .intType()
+ .defaultValue(1000)
+ .withDescription("The batch size used for deleting orphan files.");
+
+ public static final String LOCATION = PREFIX + "location";
+ public static final ConfigOption<String> LOCATION_OPTION =
+ ConfigOptions.key(LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location to start recursive listing of candidate files for removal. "
+ + "By default, the table location is used.");
+
+ public static final String USE_PREFIX_LISTING = PREFIX + "use-prefix-listing";
+ public static final ConfigOption<Boolean> USE_PREFIX_LISTING_OPTION =
+ ConfigOptions.key(USE_PREFIX_LISTING)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to use prefix listing when listing files from the file system.");
+
+ public static final String PLANNING_WORKER_POOL_SIZE = PREFIX + "planning-worker-pool-size";
+ public static final ConfigOption<Integer> PLANNING_WORKER_POOL_SIZE_OPTION =
+ ConfigOptions.key(PLANNING_WORKER_POOL_SIZE)
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The worker pool size used for planning the scan of the ALL_FILES table. "
+ + "If not set, the shared worker pool is used.");
+
+ public static final String EQUAL_SCHEMES = PREFIX + "equal-schemes";
+ public static final ConfigOption<String> EQUAL_SCHEMES_OPTION =
+ ConfigOptions.key(EQUAL_SCHEMES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Schemes that should be considered equal, in the format 'scheme1=scheme2,scheme3=scheme4'.");
+
+ public static final String EQUAL_AUTHORITIES = PREFIX + "equal-authorities";
+ public static final ConfigOption<String> EQUAL_AUTHORITIES_OPTION =
+ ConfigOptions.key(EQUAL_AUTHORITIES)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Authorities that should be considered equal, in the format 'auth1=auth2,auth3=auth4'.");
+
+ public static final String PREFIX_MISMATCH_MODE = PREFIX + "prefix-mismatch-mode";
+ public static final ConfigOption<String> PREFIX_MISMATCH_MODE_OPTION =
+ ConfigOptions.key(PREFIX_MISMATCH_MODE)
+ .stringType()
+ .defaultValue(PrefixMismatchMode.ERROR.name())
+ .withDescription(
+ "Action behavior when location prefixes (schemes/authorities) mismatch. "
+ + "Valid values: ERROR, IGNORE, DELETE.");
+
+ private final FlinkConfParser confParser;
+
+ public DeleteOrphanFilesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ public long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .defaultValue(SCHEDULE_ON_INTERVAL_SECOND_OPTION.defaultValue())
+ .parse();
+ }
+
+ public long minAgeSeconds() {
+ return confParser
+ .longConf()
+ .option(MIN_AGE_SECONDS)
+ .flinkConfig(MIN_AGE_SECONDS_OPTION)
+ .defaultValue(MIN_AGE_SECONDS_OPTION.defaultValue())
+ .parse();
+ }
+
+ public int deleteBatchSize() {
+ return confParser
+ .intConf()
+ .option(DELETE_BATCH_SIZE)
+ .flinkConfig(DELETE_BATCH_SIZE_OPTION)
+ .defaultValue(DELETE_BATCH_SIZE_OPTION.defaultValue())
+ .parse();
+ }
+
+ public String location() {
+ return confParser.stringConf().option(LOCATION).flinkConfig(LOCATION_OPTION).parseOptional();
+ }
+
+ public boolean usePrefixListing() {
+ return confParser
+ .booleanConf()
+ .option(USE_PREFIX_LISTING)
+ .flinkConfig(USE_PREFIX_LISTING_OPTION)
+ .defaultValue(USE_PREFIX_LISTING_OPTION.defaultValue())
+ .parse();
+ }
+
+ public Integer planningWorkerPoolSize() {
+ return confParser
+ .intConf()
+ .option(PLANNING_WORKER_POOL_SIZE)
+ .flinkConfig(PLANNING_WORKER_POOL_SIZE_OPTION)
+ .defaultValue(ThreadPools.WORKER_THREAD_POOL_SIZE)
+ .parse();
+ }
+
+ public Map<String, String> equalSchemes() {
+ String equalSchemes =
+ confParser
+ .stringConf()
+ .option(EQUAL_SCHEMES)
+ .flinkConfig(EQUAL_SCHEMES_OPTION)
+ .parseOptional();
+
+ return equalSchemes != null
+ ? parseKeyValuePairs(equalSchemes)
+ : Maps.newHashMap(DeleteOrphanFiles.DEFAULT_EQUAL_SCHEMES);
+ }
+
+ public Map<String, String> equalAuthorities() {
+ String equalAuthorities =
+ confParser
+ .stringConf()
+ .option(EQUAL_AUTHORITIES)
+ .flinkConfig(EQUAL_AUTHORITIES_OPTION)
+ .parseOptional();
+
+ return equalAuthorities != null ? parseKeyValuePairs(equalAuthorities) : Collections.emptyMap();
Review Comment:
Yes, I thought `Map.of()` was forbidden as per Checkstyle rule, but I think this has been lifted recently.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2910298870
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
Fair, point taken.
>Out of curiosity, is the scenario we’re trying to solve here to avoid having to maintain a separate maintenance job per table Or what scenario is this intended to address?
We're trying to make all table maintenance available post-commit. This is way simpler than having multiple jobs and avoids conflicts with the writer.
I'm going to change the implementation:
- I'll add a similar configuration as for RewriteDataFiles ("compactMode")
- I'll remove the generic `maintenance(..)` builder method.
- Change the code to only allow one maintenance for each type. Builder will have precedence over the property-based configuration.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2906633622
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -682,6 +698,14 @@ IcebergSink build() {
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+ // If the "compactMode" option is configured,
+ // add it as a separate maintenance task
+ if (flinkWriteConf.compactMode()) {
+ RewriteDataFilesConfig rewriteDataFilesConfig =
+ flinkMaintenanceConfig.createRewriteDataFilesConfig();
+ maintenanceTasks.add(RewriteDataFiles.builder().config(rewriteDataFilesConfig));
+ }
Review Comment:
It would be sensible to allow only one of each maintenance types, i.e. let that be the unit of configuration. Consequently, when `RewriteDataFiles` is configured, it should supersede "compaction mode".
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2926024978
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
Review Comment:
+1 on this. I think the best API would be to let the user choose what kind of "compaction task" they want to enable. I'd even suggest to mark this `compaction` method for deletion in the next release (or at least the next to next release)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2924975956
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -626,6 +633,92 @@ public Builder setSnapshotProperty(String property, String value) {
return this;
}
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ * @deprecated See {@code rewriteDatafiles(..)}
+ */
+ @Deprecated
+ public Builder compaction(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @see RewriteDataFilesConfig for the default config.
+ */
+ public Builder rewriteDataFiles(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), Boolean.toString(enabled));
+ return this;
+ }
+
+ /**
+ * Enables or disables compaction (rewriting data files) as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable compaction
+ * @param config task-specific configuration, see {@link RewriteDataFilesConfig} for available
+ * keys
+ */
+ public Builder rewriteDataFiles(boolean enabled, Map<String, String> config) {
+ rewriteDataFiles(enabled);
+ writeOptions.putAll(config);
+ return this;
+ }
+
+ /**
+ * Enables or disables expire snapshots as a post-commit maintenance task.
+ *
+ * @param enabled whether to enable expire snapshots
+ * @see ExpireSnapshotsConfig for the default config.
+ */
+ public Builder expireSnapshots(boolean enabled) {
+ writeOptions.put(FlinkWriteOptions.EXPIRE_SNAPSHOTS_ENABLE.key(), Boolean.toString(enabled));
Review Comment:
Defaults are hard. These are the current defaults:
ExpireSnapshotsConfig defaults:
- schedule.commit-count: 10
- schedule.data-file-count: 1,000
- schedule.data-file-size: 100 GB
- schedule.interval-second: 3600 (1 hour)
- max-snapshot-age-seconds: none
- retain-last: none
- delete-batch-size: 1,000
- clean-expired-metadata: false
- planning-worker-pool-size: none (shared pool)
DeleteOrphanFilesConfig defaults:
- schedule.commit-count: 10
- schedule.data-file-count: 1,000
- schedule.data-file-size: 100 GB
- schedule.interval-second: 3600 (1 hour)
- min-age-seconds: 259,200 (3 days)
- delete-batch-size: 1,000
- use-prefix-listing: false
- planning-worker-pool-size: none (shared pool)
- equal-schemes: none
- equal-authorities: none
- prefix-mismatch-mode: ERROR
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Allow arbitrary post-commit maintenance tasks via IcebergSink Builder [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15566:
URL: https://github.com/apache/iceberg/pull/15566#discussion_r2926769556
##########
docs/docs/flink-maintenance.md:
##########
@@ -302,19 +302,50 @@ public class TableMaintenanceJob {
Apache Iceberg Sink V2 for Flink allows automatic execution of maintenance tasks after data is committed to the table, using the addPostCommitTopology(...) method.
+#### DataStream API
+
+##### Builder
+
+```java
+IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .rewriteDataFiles(Map.of(
+ RewriteDataFilesConfig.MAX_BYTES, "1073741824"))
+ .expireSnapshots(Map.of(
+ ExpireSnapshotsConfig.RETAIN_LAST, "5",
+ ExpireSnapshotsConfig.MAX_SNAPSHOT_AGE_SECONDS, "604800"))
+ .deleteOrphanFiles(Map.of(
+ DeleteOrphanFilesConfig.MIN_AGE_SECONDS, "259200"))
+ .append();
+```
+
+##### Config
+
+All maintenance tasks are configured through string properties:
+
```java
Map<String, String> flinkConf = new HashMap<>();
-// Enable compaction and maintenance features
-flinkConf.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
+// Enable maintenance tasks
+flinkConf.put("flink-maintenance.rewrite.enabled", "true");
+flinkConf.put("flink-maintenance.expire-snapshots.enabled", "true");
+flinkConf.put("flink-maintenance.delete-orphan-files.enabled", "true");
-// Configure JDBC lock settings
-flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
-flinkConf.put(LockConfig.JdbcLockConfig.JDBC_URI_OPTION.key(), "jdbc:postgresql://localhost:5432/iceberg");
-flinkConf.put(LockConfig.LOCK_ID_OPTION.key(), "catalog.db.table");
+// Configure rewrite data files
+flinkConf.put("flink-maintenance.rewrite.max-bytes", "1073741824");
-// Add any other maintenance-related options here as needed
-// ...
+// Configure expire snapshots
+flinkConf.put("flink-maintenance.expire-snapshots.retain-last", "5");
+flinkConf.put("flink-maintenance.expire-snapshots.max-snapshot-age-seconds", "604800");
Review Comment:
Yes, exactly.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org