You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org> on 2023/04/06 07:28:54 UTC
[GitHub] [hudi] zhuanshenbsj1 opened a new pull request, #8394: adjust HoodieTableSink for clean
zhuanshenbsj1 opened a new pull request, #8394:
URL: https://github.com/apache/hudi/pull/8394
### Change Logs
MOR table in upsert scenario, not need to clean when online async compaction is turned off.
In scenarios with a large number of files and partitions, turning off clean will improve performance
### Impact
_Describe any public API or user-facing feature change or any performance impact._
### Risk level (write none, low medium or high below)
_If medium or high, explain what verification was done to mitigate the risks._
### Documentation Update
_Describe any necessary documentation update if there is any new feature, config, or user-facing change_
- _The config description must be updated if new configs are added or the default value of the configs are changed_
- _Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
ticket number here and follow the [instruction](https://hudi.apache.org/contribute/developer-setup#website) to make
changes to the website._
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1167417375
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java:
##########
@@ -358,6 +358,9 @@ private void cluster() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_clustering_" + clusteringInstant.getTimestamp());
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CompactionCommitSink#open should already handle the cleaning.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119639
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java:
##########
@@ -358,6 +358,9 @@ private void cluster() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_clustering_" + clusteringInstant.getTimestamp());
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in ClusteringCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, HoodieClusteringPlan clusteringPlan, List<ClusteringCommitEvent> events) {
List<WriteStatus> statuses = events.stream()
.map(ClusteringCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieWriteMetadata<List<WriteStatus>> writeMetadata = new HoodieWriteMetadata<>();
writeMetadata.setWriteStatuses(statuses);
writeMetadata.setWriteStats(statuses.stream().map(WriteStatus::getStat).collect(Collectors.toList()));
writeMetadata.setPartitionToReplaceFileIds(getPartitionToReplacedFileIds(clusteringPlan, writeMetadata));
validateWriteResult(clusteringPlan, instant, writeMetadata);
if (!writeMetadata.getCommitMetadata().isPresent()) {
HoodieCommitMetadata commitMetadata = CommitUtils.buildMetadata(
writeMetadata.getWriteStats().get(),
writeMetadata.getPartitionToReplaceFileIds(),
Option.empty(),
WriteOperationType.CLUSTER,
this.writeClient.getConfig().getSchema(),
HoodieTimeline.REPLACE_COMMIT_ACTION);
writeMetadata.setCommitMetadata(Option.of(commitMetadata));
}
// commit the clustering
this.table.getMetaClient().reloadActiveTimeline();
this.writeClient.completeTableService(
TableServiceType.CLUSTER, writeMetadata.getCommitMetadata().get(), table, instant);
// whether to clean up the input base parquet files used for clustering
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
LOG.info("Running inline clean");
this.writeClient.clean();
}
}
```
So I added configuration CLEAN_OFFLINE_ENABLE to distinguish it from configuration CLEAN_ASYNC_ENABLED.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CompactionCommitSink#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in CompactionCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, Collection<CompactionCommitEvent> events) throws IOException {
List<WriteStatus> statuses = events.stream()
.map(CompactionCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieCommitMetadata metadata = CompactHelpers.getInstance().createCompactionMetadata(
table, instant, HoodieListData.eager(statuses), writeClient.getConfig().getSchema());
// commit the compaction
this.writeClient.commitCompaction(instant, metadata, Option.empty());
// Whether to clean up the old log file when compaction
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient.clean();
}
}
```
So I added configuration CLEAN_OFFLINE_ENABLE to distinguish it from configuration CLEAN_ASYNC_ENABLED.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119639
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java:
##########
@@ -358,6 +358,9 @@ private void cluster() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_clustering_" + clusteringInstant.getTimestamp());
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
Oh, I see. My Hudi version does not include this PR https://github.com/apache/hudi/pull/6515, So cleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1164056771
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Add a sync-clean config verification for 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1172020216
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java:
##########
@@ -320,7 +322,9 @@ protected HoodieWriteMetadata<JavaRDD<WriteStatus>> logCompact(String logCompact
public HoodieWriteMetadata<JavaRDD<WriteStatus>> cluster(String clusteringInstant, boolean shouldComplete) {
HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
preWrite(clusteringInstant, WriteOperationType.CLUSTER, table.getMetaClient());
- return tableServiceClient.cluster(clusteringInstant, shouldComplete);
+ HoodieWriteMetadata<JavaRDD<WriteStatus>> clusteringMetadata = tableServiceClient.cluster(clusteringInstant, shouldComplete);
+ autoCleanOnCommit();
+ return clusteringMetadata;
Review Comment:
Do not introduce unrelated changes in one patch, if we want to add cleaning procedure for Spark compaction and cleaning, fire another PR and let's discuss with it.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-2019221360
@zhuanshenbsj1 @danny0405 is this PR still relevant?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1160347951
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Offline compaction do clean，Consistent with clustering.
// Append mode
if (OptionsResolver.isAppendMode(conf)) {
DataStream<Object> pipeline = Pipelines.append(conf, rowType, dataStream, context.isBounded());
if (OptionsResolver.needsAsyncClustering(conf)) {
return Pipelines.cluster(conf, rowType, pipeline);
} else {
return Pipelines.dummySink(pipeline);// If execute-clustering=true, no clean.
}
}
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1514468055
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
}, {
"hash" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382",
"triggerID" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"triggerType" : "PUSH"
}, {
"hash" : "9edae8190c33edf9cdc05568ffe3b923069ca19e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16459",
"triggerID" : "9edae8190c33edf9cdc05568ffe3b923069ca19e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 79ec658c4c5280dc2ba0fbc5d4570a76fe09952a Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382)
* 9edae8190c33edf9cdc05568ffe3b923069ca19e Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16459)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1505192289
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 67fc8b9e83ad777e860be2faf86e7d5969583a6f Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168)
* 47cb215abcd658fec147a77db248740e9496e931 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1510686209
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
}, {
"hash" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382",
"triggerID" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 36fc037a6ef0e2c6c3409694a50a196b738d4e4d Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307)
* 79ec658c4c5280dc2ba0fbc5d4570a76fe09952a Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in CompactionCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, Collection<CompactionCommitEvent> events) throws IOException {
List<WriteStatus> statuses = events.stream()
.map(CompactionCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieCommitMetadata metadata = CompactHelpers.getInstance().createCompactionMetadata(
table, instant, HoodieListData.eager(statuses), writeClient.getConfig().getSchema());
// commit the compaction
this.writeClient.commitCompaction(instant, metadata, Option.empty());
// Whether to clean up the old log file when compaction
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient.clean();
}
}
```
So online async-clean and offline clean use the same config, So I added configuration CLEAN_OFFLINE_ENABLE to distinguish it from configuration CLEAN_ASYNC_ENABLED.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java:
##########
@@ -358,6 +358,9 @@ private void cluster() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_clustering_" + clusteringInstant.getTimestamp());
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in ClusteringCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, HoodieClusteringPlan clusteringPlan, List<ClusteringCommitEvent> events) {
List<WriteStatus> statuses = events.stream()
.map(ClusteringCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieWriteMetadata<List<WriteStatus>> writeMetadata = new HoodieWriteMetadata<>();
writeMetadata.setWriteStatuses(statuses);
writeMetadata.setWriteStats(statuses.stream().map(WriteStatus::getStat).collect(Collectors.toList()));
writeMetadata.setPartitionToReplaceFileIds(getPartitionToReplacedFileIds(clusteringPlan, writeMetadata));
validateWriteResult(clusteringPlan, instant, writeMetadata);
if (!writeMetadata.getCommitMetadata().isPresent()) {
HoodieCommitMetadata commitMetadata = CommitUtils.buildMetadata(
writeMetadata.getWriteStats().get(),
writeMetadata.getPartitionToReplaceFileIds(),
Option.empty(),
WriteOperationType.CLUSTER,
this.writeClient.getConfig().getSchema(),
HoodieTimeline.REPLACE_COMMIT_ACTION);
writeMetadata.setCommitMetadata(Option.of(commitMetadata));
}
// commit the clustering
this.table.getMetaClient().reloadActiveTimeline();
this.writeClient.completeTableService(
TableServiceType.CLUSTER, writeMetadata.getCommitMetadata().get(), table, instant);
// whether to clean up the input base parquet files used for clustering
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
LOG.info("Running inline clean");
this.writeClient.clean();
}
}
```
So online async-clean and offline clean use the same config, So I added configuration CLEAN_OFFLINE_ENABLE to distinguish it from configuration CLEAN_ASYNC_ENABLED.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119639
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java:
##########
@@ -358,6 +358,9 @@ private void cluster() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_clustering_" + clusteringInstant.getTimestamp());
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CompactionCommitSink#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1514456123
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
}, {
"hash" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382",
"triggerID" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"triggerType" : "PUSH"
}, {
"hash" : "9edae8190c33edf9cdc05568ffe3b923069ca19e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9edae8190c33edf9cdc05568ffe3b923069ca19e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 79ec658c4c5280dc2ba0fbc5d4570a76fe09952a Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382)
* 9edae8190c33edf9cdc05568ffe3b923069ca19e UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168688393
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
The spark offline compaction Job does not take care of cleaning, could you make it clrear how user can handle the cleaning when they use the flink streaming ingestion and spark offline compaction.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1160347951
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Offline compaction do clean，Consistent with clustering.
// Append mode
if (OptionsResolver.isAppendMode(conf)) {
DataStream<Object> pipeline = Pipelines.append(conf, rowType, dataStream, context.isBounded());
if (OptionsResolver.needsAsyncClustering(conf)) {
return Pipelines.cluster(conf, rowType, pipeline);
} else {
return Pipelines.dummySink(pipeline);// If async-clustering=true, no clean operator.
}
}
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1506576401
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 47cb215abcd658fec147a77db248740e9496e931 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285)
* 36fc037a6ef0e2c6c3409694a50a196b738d4e4d Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1498655831
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 67fc8b9e83ad777e860be2faf86e7d5969583a6f UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled [hudi]
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 closed pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled
URL: https://github.com/apache/hudi/pull/8394
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-2019224041
cc @zhuanshenbsj1 I think we can cloase it, because we already have #8505 .
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1505204028
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 67fc8b9e83ad777e860be2faf86e7d5969583a6f Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168)
* 47cb215abcd658fec147a77db248740e9496e931 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in CompactionCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, Collection<CompactionCommitEvent> events) throws IOException {
List<WriteStatus> statuses = events.stream()
.map(CompactionCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieCommitMetadata metadata = CompactHelpers.getInstance().createCompactionMetadata(
table, instant, HoodieListData.eager(statuses), writeClient.getConfig().getSchema());
// commit the compaction
this.writeClient.commitCompaction(instant, metadata, Option.empty());
// Whether to clean up the old log file when compaction
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient.clean();
}
}
```
So I added configuration CLEAN_OFFLINE_ENABLE to distinguish it from configuration CLEAN_ASYNC_ENABLED.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CompactionCommitSink#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in CompactionCommitSink#doCommit.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1167417320
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java:
##########
@@ -640,6 +640,12 @@ private FlinkOptions() {
.defaultValue(true)
.withDescription("Whether to cleanup the old commits immediately on new commits, enabled by default");
+ public static final ConfigOption<Boolean> CLEAN_OFFLINE_ENABLE = ConfigOptions
+ .key("clean.offline.enable")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription("Whether to cleanup the old commits immediately in offline mode, enabled by default");
+
Review Comment:
No need to introduce this option.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CompactionCommitSink#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in CompactionCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, Collection<CompactionCommitEvent> events) throws IOException {
List<WriteStatus> statuses = events.stream()
.map(CompactionCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieCommitMetadata metadata = CompactHelpers.getInstance().createCompactionMetadata(
table, instant, HoodieListData.eager(statuses), writeClient.getConfig().getSchema());
// commit the compaction
this.writeClient.commitCompaction(instant, metadata, Option.empty());
// Whether to clean up the old log file when compaction
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient.clean();
}
}
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
Oh, I see. My Hudi version does not include this PR https://github.com/apache/hudi/pull/6515, So cleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1159879765
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
So what component is responsible for data cleaning then?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1164056771
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Add a sync-clean config verification for 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1165154162
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Current online async-clean and offline clean use the same configuration ：clean.async.enabled. Add a new configuration clean.offline.enable，making the configuration description clearer.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1506565557
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 47cb215abcd658fec147a77db248740e9496e931 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285)
* 36fc037a6ef0e2c6c3409694a50a196b738d4e4d UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1164035974
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Offline clustering/compaction will do clean by default , add --clean-async-enabled config will close it.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CompactionCommitSink#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1171085867
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Adjust the cleaning operation in SparkRDDWriteClient#cluster/compact, when ASYNC_CLEAN is true will do asynchronous clean in prewrite， otherwise will do synchronous clean in autoCleanOnCommit().
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1167417258
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
`CompactionCommitSink#open` should already handle the cleaning.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1510869676
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
}, {
"hash" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382",
"triggerID" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 79ec658c4c5280dc2ba0fbc5d4570a76fe09952a Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1498665513
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 67fc8b9e83ad777e860be2faf86e7d5969583a6f Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1499289129
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 67fc8b9e83ad777e860be2faf86e7d5969583a6f Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1506800465
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 36fc037a6ef0e2c6c3409694a50a196b738d4e4d Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: adjust HoodieTableSink for clean
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1505666062
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 47cb215abcd658fec147a77db248740e9496e931 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1193270939
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
We will not schedule cleaning task if async cleaning is disabled: https://github.com/apache/hudi/blob/67ae0c8e7e4e58454cce18a8f58bfa43f67c1183/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java#L72, so 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1193373503
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
If CLEAN_ASYNC_ENABLED = true，a schedule will still be executed. I think cluster and compact should be consistent here(If the cow table async cluster is closed, there will be no clean operator). And now both Spark and Flink will clean up when executing offline jobs, unless forcibly closed. What do you think?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1510656865
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
}, {
"hash" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 36fc037a6ef0e2c6c3409694a50a196b738d4e4d Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307)
* 79ec658c4c5280dc2ba0fbc5d4570a76fe09952a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1172031561
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java:
##########
@@ -320,7 +322,9 @@ protected HoodieWriteMetadata<JavaRDD<WriteStatus>> logCompact(String logCompact
public HoodieWriteMetadata<JavaRDD<WriteStatus>> cluster(String clusteringInstant, boolean shouldComplete) {
HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
preWrite(clusteringInstant, WriteOperationType.CLUSTER, table.getMetaClient());
- return tableServiceClient.cluster(clusteringInstant, shouldComplete);
+ HoodieWriteMetadata<JavaRDD<WriteStatus>> clusteringMetadata = tableServiceClient.cluster(clusteringInstant, shouldComplete);
+ autoCleanOnCommit();
+ return clusteringMetadata;
Review Comment:
Removed Spark related modifications and migrated to a new PR. https://github.com/apache/hudi/pull/8505
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119189
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java:
##########
@@ -314,6 +314,9 @@ private void compact() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_compaction_" + String.join(",", compactionInstantTimes));
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in CompactionCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, Collection<CompactionCommitEvent> events) throws IOException {
List<WriteStatus> statuses = events.stream()
.map(CompactionCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieCommitMetadata metadata = CompactHelpers.getInstance().createCompactionMetadata(
table, instant, HoodieListData.eager(statuses), writeClient.getConfig().getSchema());
// commit the compaction
this.writeClient.commitCompaction(instant, metadata, Option.empty());
// Whether to clean up the old log file when compaction
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient.clean();
}
}
```
Online async-clean and offline clean use the same config, So I added configuration CLEAN_OFFLINE_ENABLE to distinguish it from configuration CLEAN_ASYNC_ENABLED.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java:
##########
@@ -358,6 +358,9 @@ private void cluster() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_clustering_" + clusteringInstant.getTimestamp());
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
I think the real cleaning is actually in ClusteringCommitSink#doCommit， and it depends on whether to enable configuration CLEAN_ASYNC_ENABLED.
```
private void doCommit(String instant, HoodieClusteringPlan clusteringPlan, List<ClusteringCommitEvent> events) {
List<WriteStatus> statuses = events.stream()
.map(ClusteringCommitEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
HoodieWriteMetadata<List<WriteStatus>> writeMetadata = new HoodieWriteMetadata<>();
writeMetadata.setWriteStatuses(statuses);
writeMetadata.setWriteStats(statuses.stream().map(WriteStatus::getStat).collect(Collectors.toList()));
writeMetadata.setPartitionToReplaceFileIds(getPartitionToReplacedFileIds(clusteringPlan, writeMetadata));
validateWriteResult(clusteringPlan, instant, writeMetadata);
if (!writeMetadata.getCommitMetadata().isPresent()) {
HoodieCommitMetadata commitMetadata = CommitUtils.buildMetadata(
writeMetadata.getWriteStats().get(),
writeMetadata.getPartitionToReplaceFileIds(),
Option.empty(),
WriteOperationType.CLUSTER,
this.writeClient.getConfig().getSchema(),
HoodieTimeline.REPLACE_COMMIT_ACTION);
writeMetadata.setCommitMetadata(Option.of(commitMetadata));
}
// commit the clustering
this.table.getMetaClient().reloadActiveTimeline();
this.writeClient.completeTableService(
TableServiceType.CLUSTER, writeMetadata.getCommitMetadata().get(), table, instant);
// whether to clean up the input base parquet files used for clustering
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
LOG.info("Running inline clean");
this.writeClient.clean();
}
}
```
Online async-clean and offline clean use the same config, So I added configuration CLEAN_OFFLINE_ENABLE to distinguish it from configuration CLEAN_ASYNC_ENABLED.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1168119639
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java:
##########
@@ -358,6 +358,9 @@ private void cluster() throws Exception {
.setParallelism(1);
env.execute("flink_hudi_clustering_" + clusteringInstant.getTimestamp());
+ if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && conf.getBoolean(FlinkOptions.CLEAN_OFFLINE_ENABLE)){
+ writeClient.clean();
+ }
Review Comment:
CleanFunction#open only do clean when OptionsResolver.isInsertOverwrite(conf) is true.
```
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
}
}
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1171085867
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
Adjust the cleaning operation in SparkRDDWriteClient#cluster/compact, when ASYNC_CLEAN is true will do clean in prewrite， otherwise will do clean in autoCleanOnCommit().
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1172046008
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
We need to think through the e2e use case for flink streaming ingestion and spark offline compaction, we should add cleaning for spark compaction and clustering first which is a block change of this patch.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if async cleaning is disabled
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #8394:
URL: https://github.com/apache/hudi/pull/8394#issuecomment-1515240120
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16168",
"triggerID" : "67fc8b9e83ad777e860be2faf86e7d5969583a6f",
"triggerType" : "PUSH"
}, {
"hash" : "47cb215abcd658fec147a77db248740e9496e931",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16285",
"triggerID" : "47cb215abcd658fec147a77db248740e9496e931",
"triggerType" : "PUSH"
}, {
"hash" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16307",
"triggerID" : "36fc037a6ef0e2c6c3409694a50a196b738d4e4d",
"triggerType" : "PUSH"
}, {
"hash" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16382",
"triggerID" : "79ec658c4c5280dc2ba0fbc5d4570a76fe09952a",
"triggerType" : "PUSH"
}, {
"hash" : "9edae8190c33edf9cdc05568ffe3b923069ca19e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16459",
"triggerID" : "9edae8190c33edf9cdc05568ffe3b923069ca19e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9edae8190c33edf9cdc05568ffe3b923069ca19e Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=16459)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #8394: adjust HoodieTableSink for clean
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1161181943
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
The cleaning can take effect finally right? Because the table would get compacted for sometime anyway.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] zhuanshenbsj1 commented on a diff in pull request #8394: [HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled
Posted by "zhuanshenbsj1 (via GitHub)" <gi...@apache.org>.
zhuanshenbsj1 commented on code in PR #8394:
URL: https://github.com/apache/hudi/pull/8394#discussion_r1193373503
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
+ } else if (OptionsResolver.isMorTable(conf)) {
+ return Pipelines.dummySink(pipeline);
} else {
Review Comment:
If CLEAN_ASYNC_ENABLED = true，a schedule will still be executed. I think cluster and compact should be consistent here. And now both Spark and Flink will clean up when executing offline jobs, unless forcibly closed. What do you think?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org