You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2021/09/28 07:10:59 UTC
[GitHub] [hudi] yihua opened a new pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
yihua opened a new pull request #3727:
URL: https://github.com/apache/hudi/pull/3727
## What is the purpose of the pull request
This PR refactors the clean, restore, and compaction actions in hudi-client module to extract common logic into hudi-client-common module and reduce LoC.
## Brief change log
## Verify this pull request
This pull request is already covered by existing tests, such as *(please describe tests)*.
## Committer checklist
- [ ] Has a corresponding JIRA in PR title & commit
- [ ] Commit message is descriptive of the change
- [ ] CI is green
- [ ] Necessary doc changes done or have another open PR
- [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
--
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] yihua commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719752579
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/function/SerializablePairFlatMapFunction.java
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.hudi.common.function;
+
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.io.Serializable;
+import java.util.stream.Stream;
+
+/**
+ * A function that returns a stream of key-value pairs (Pair<K, V>).
Review comment:
Done.
--
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 edited a comment on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot edited a comment on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448",
"triggerID" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"triggerType" : "PUSH"
}, {
"hash" : "10081946cf12f3c7116ee5192ef0a44709b19021",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "10081946cf12f3c7116ee5192ef0a44709b19021",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448)
* 10081946cf12f3c7116ee5192ef0a44709b19021 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run travis` re-run the last Travis build
- `@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] yihua closed pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua closed pull request #3727:
URL: https://github.com/apache/hudi/pull/3727
--
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] yihua commented on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-931656856
@nsivabalan I addressed your comments in this PR on refactoring compaction actions: https://github.com/apache/hudi/pull/3741
--
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] yihua commented on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-930614516
@nsivabalan I'll address your comments in the PR of refactoring 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] hudi-bot commented on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run travis` re-run the last Travis build
- `@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 edited a comment on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot edited a comment on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448",
"triggerID" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run travis` re-run the last Travis build
- `@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 edited a comment on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot edited a comment on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448",
"triggerID" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"triggerType" : "PUSH"
}, {
"hash" : "10081946cf12f3c7116ee5192ef0a44709b19021",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2454",
"triggerID" : "10081946cf12f3c7116ee5192ef0a44709b19021",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10081946cf12f3c7116ee5192ef0a44709b19021 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2454)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run travis` re-run the last Travis build
- `@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 edited a comment on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot edited a comment on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448",
"triggerID" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"triggerType" : "PUSH"
}, {
"hash" : "10081946cf12f3c7116ee5192ef0a44709b19021",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2454",
"triggerID" : "10081946cf12f3c7116ee5192ef0a44709b19021",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448)
* 10081946cf12f3c7116ee5192ef0a44709b19021 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2454)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run travis` re-run the last Travis build
- `@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] yihua commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719758782
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
##########
@@ -68,6 +77,21 @@ public HoodieFlinkEngineContext(SerializableConfiguration hadoopConf, TaskContex
this.runtimeContext = ((FlinkTaskContextSupplier) taskContextSupplier).getFlinkRuntimeContext();
}
+ @Override
+ public HoodieAccumulator createNewAccumulator() {
Review comment:
The abstract method is named like that.
--
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] yihua commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719752579
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/function/SerializablePairFlatMapFunction.java
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.hudi.common.function;
+
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.io.Serializable;
+import java.util.stream.Stream;
+
+/**
+ * A function that returns a stream of key-value pairs (Pair<K, V>).
Review comment:
These are actually automatically converted to conform Javadocs format since entities for the less than symbol (<) and the greater than symbol (>) should be written as `<` and `>`.
--
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 edited a comment on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot edited a comment on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
--
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 edited a comment on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot edited a comment on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448",
"triggerID" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=2448)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run travis` re-run the last Travis build
- `@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] nsivabalan commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r718849052
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactHelpers.java
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.hudi.table.action.compact;
+
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.client.AbstractHoodieWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.CompactionOperation;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieCompactionException;
+import org.apache.hudi.table.HoodieCopyOnWriteTableOperation;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
+
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Base class helps to perform compact.
+ *
+ * @param <T> Type of payload in {@link org.apache.hudi.common.model.HoodieRecord}
+ */
+public class CompactHelpers<T extends HoodieRecordPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactHelpers.class);
+
+ private CompactHelpers() {
+ }
+
+ private static class CompactHelperHolder {
+ private static final CompactHelpers COMPACT_HELPERS = new CompactHelpers();
+ }
+
+ public static CompactHelpers newInstance() {
+ return CompactHelperHolder.COMPACT_HELPERS;
+ }
+
+ public HoodieWriteMetadata<HoodieData<WriteStatus>> compact(
+ HoodieEngineContext context, HoodieTable table,
+ HoodieCopyOnWriteTableOperation copyOnWriteTableOperation, HoodieWriteConfig config,
+ String compactionInstantTime, AbstractHoodieWriteClient writeClient,
+ TaskContextSupplier taskContextSupplier) {
+ HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
+ HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
+ if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
+ writeClient.rollbackInflightCompaction(inflightInstant, table);
+ table.getMetaClient().reloadActiveTimeline();
+ }
+
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata = new HoodieWriteMetadata<>();
+ compactionMetadata.setWriteStatuses(context.createEmptyHoodieData());
+ try {
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ if (compactionPlan == null || (compactionPlan.getOperations() == null)
+ || (compactionPlan.getOperations().isEmpty())) {
+ // do nothing.
+ LOG.info("No compaction plan for instant " + compactionInstantTime);
+ return compactionMetadata;
+ }
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+ table.getMetaClient().reloadActiveTimeline();
+
+ HoodieTableMetaClient metaClient = table.getMetaClient();
Review comment:
this method is too lengthy. Can we try to add 1 or 2 private methods.
##########
File path: hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
##########
@@ -321,15 +324,14 @@ protected void completeCompaction(HoodieCommitMetadata metadata, JavaRDD<WriteSt
protected JavaRDD<WriteStatus> compact(String compactionInstantTime, boolean shouldComplete) {
HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
preWrite(compactionInstantTime, WriteOperationType.COMPACT, table.getMetaClient());
- HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
- HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
- if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
- rollbackInflightCompaction(inflightInstant, table);
- table.getMetaClient().reloadActiveTimeline();
- }
compactionTimer = metrics.getCompactionCtx();
- HoodieWriteMetadata<JavaRDD<WriteStatus>> compactionMetadata = table.compact(context, compactionInstantTime);
- JavaRDD<WriteStatus> statuses = compactionMetadata.getWriteStatuses();
+ HoodieSparkCopyOnWriteTable sparkCopyOnWriteTable =
+ new HoodieSparkCopyOnWriteTable(config, context, table.getMetaClient());
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata =
+ CompactHelpers.newInstance().compact(context, table, sparkCopyOnWriteTable,
+ config, compactionInstantTime, this, new SparkTaskContextSupplier());
Review comment:
same here, why not use existing context ?
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactHelpers.java
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.hudi.table.action.compact;
+
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.client.AbstractHoodieWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.CompactionOperation;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieCompactionException;
+import org.apache.hudi.table.HoodieCopyOnWriteTableOperation;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
+
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Base class helps to perform compact.
+ *
+ * @param <T> Type of payload in {@link org.apache.hudi.common.model.HoodieRecord}
+ */
+public class CompactHelpers<T extends HoodieRecordPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactHelpers.class);
+
+ private CompactHelpers() {
+ }
+
+ private static class CompactHelperHolder {
+ private static final CompactHelpers COMPACT_HELPERS = new CompactHelpers();
+ }
+
+ public static CompactHelpers newInstance() {
+ return CompactHelperHolder.COMPACT_HELPERS;
+ }
+
+ public HoodieWriteMetadata<HoodieData<WriteStatus>> compact(
+ HoodieEngineContext context, HoodieTable table,
+ HoodieCopyOnWriteTableOperation copyOnWriteTableOperation, HoodieWriteConfig config,
+ String compactionInstantTime, AbstractHoodieWriteClient writeClient,
+ TaskContextSupplier taskContextSupplier) {
+ HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
+ HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
+ if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
+ writeClient.rollbackInflightCompaction(inflightInstant, table);
+ table.getMetaClient().reloadActiveTimeline();
+ }
+
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata = new HoodieWriteMetadata<>();
+ compactionMetadata.setWriteStatuses(context.createEmptyHoodieData());
+ try {
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ if (compactionPlan == null || (compactionPlan.getOperations() == null)
+ || (compactionPlan.getOperations().isEmpty())) {
+ // do nothing.
+ LOG.info("No compaction plan for instant " + compactionInstantTime);
+ return compactionMetadata;
+ }
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+ table.getMetaClient().reloadActiveTimeline();
+
+ HoodieTableMetaClient metaClient = table.getMetaClient();
+ TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
+
+ // Here we firstly use the table schema as the reader schema to read
+ // log file.That is because in the case of MergeInto, the config.getSchema may not
+ // the same with the table schema.
+ try {
Review comment:
I guess these are not invoked in flink engine? can you double check that please. HoodieFlinkMergeOnReadTableCompactor
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
##########
@@ -68,6 +77,21 @@ public HoodieFlinkEngineContext(SerializableConfiguration hadoopConf, TaskContex
this.runtimeContext = ((FlinkTaskContextSupplier) taskContextSupplier).getFlinkRuntimeContext();
}
+ @Override
+ public HoodieAccumulator createNewAccumulator() {
Review comment:
createNewLongAccumulator
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/function/SerializablePairFlatMapFunction.java
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.hudi.common.function;
+
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.io.Serializable;
+import java.util.stream.Stream;
+
+/**
+ * A function that returns a stream of key-value pairs (Pair<K, V>).
Review comment:
minor: can you fix java docs "<" ">"
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
##########
@@ -86,6 +110,17 @@ public RuntimeContext getRuntimeContext() {
.collect(Collectors.toList());
}
+ @Override
+ public <I, K, V> Stream<ImmutablePair<K, V>> mapPartitionsToPairAndReduceByKey(
+ Stream<I> data, SerializablePairFlatMapFunction<Iterator<I>, K, V> flatMapToPairFunc,
+ SerializableBiFunction<V, V, V> reduceFunc, int parallelism) {
+ return throwingFlatMapToPairWrapper(flatMapToPairFunc).apply(data.parallel().iterator())
+ .collect(Collectors.groupingBy(Pair::getKey)).entrySet().stream()
Review comment:
may I know where have we used the "parallelism" within this method?
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
##########
@@ -369,7 +372,14 @@ public void completeCompaction(
protected List<WriteStatus> compact(String compactionInstantTime, boolean shouldComplete) {
// only used for metadata table, the compaction happens in single thread
try {
- List<WriteStatus> writeStatuses = FlinkCompactHelpers.compact(compactionInstantTime, this);
+ HoodieFlinkCopyOnWriteTable flinkCopyOnWriteTable = new HoodieFlinkCopyOnWriteTable(
+ getConfig(), getEngineContext(), getHoodieTable().getMetaClient());
+ HoodieWriteMetadata<HoodieData<WriteStatus>> writeMetadata =
+ CompactHelpers.newInstance().compact(context, getHoodieTable(), flinkCopyOnWriteTable,
+ config, compactionInstantTime, this, new FlinkTaskContextSupplier(null));
Review comment:
why create a new FlinkTaskContextSupplier? dont' we have one already locally ?
--
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] yihua commented on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-930614238
--
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] yihua commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719756887
##########
File path: hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
##########
@@ -321,15 +324,14 @@ protected void completeCompaction(HoodieCommitMetadata metadata, JavaRDD<WriteSt
protected JavaRDD<WriteStatus> compact(String compactionInstantTime, boolean shouldComplete) {
HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
preWrite(compactionInstantTime, WriteOperationType.COMPACT, table.getMetaClient());
- HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
- HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
- if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
- rollbackInflightCompaction(inflightInstant, table);
- table.getMetaClient().reloadActiveTimeline();
- }
compactionTimer = metrics.getCompactionCtx();
- HoodieWriteMetadata<JavaRDD<WriteStatus>> compactionMetadata = table.compact(context, compactionInstantTime);
- JavaRDD<WriteStatus> statuses = compactionMetadata.getWriteStatuses();
+ HoodieSparkCopyOnWriteTable sparkCopyOnWriteTable =
+ new HoodieSparkCopyOnWriteTable(config, context, table.getMetaClient());
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata =
+ CompactHelpers.newInstance().compact(context, table, sparkCopyOnWriteTable,
+ config, compactionInstantTime, this, new SparkTaskContextSupplier());
Review comment:
Right. Fixed.
--
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] yihua commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719752331
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
##########
@@ -86,6 +110,17 @@ public RuntimeContext getRuntimeContext() {
.collect(Collectors.toList());
}
+ @Override
+ public <I, K, V> Stream<ImmutablePair<K, V>> mapPartitionsToPairAndReduceByKey(
+ Stream<I> data, SerializablePairFlatMapFunction<Iterator<I>, K, V> flatMapToPairFunc,
+ SerializableBiFunction<V, V, V> reduceFunc, int parallelism) {
+ return throwingFlatMapToPairWrapper(flatMapToPairFunc).apply(data.parallel().iterator())
+ .collect(Collectors.groupingBy(Pair::getKey)).entrySet().stream()
Review comment:
For Flink it's not used since it's not relevant. It's there only because of the same interface to be compatible with Spark.
--
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] nsivabalan commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r718849052
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactHelpers.java
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.hudi.table.action.compact;
+
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.client.AbstractHoodieWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.CompactionOperation;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieCompactionException;
+import org.apache.hudi.table.HoodieCopyOnWriteTableOperation;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
+
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Base class helps to perform compact.
+ *
+ * @param <T> Type of payload in {@link org.apache.hudi.common.model.HoodieRecord}
+ */
+public class CompactHelpers<T extends HoodieRecordPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactHelpers.class);
+
+ private CompactHelpers() {
+ }
+
+ private static class CompactHelperHolder {
+ private static final CompactHelpers COMPACT_HELPERS = new CompactHelpers();
+ }
+
+ public static CompactHelpers newInstance() {
+ return CompactHelperHolder.COMPACT_HELPERS;
+ }
+
+ public HoodieWriteMetadata<HoodieData<WriteStatus>> compact(
+ HoodieEngineContext context, HoodieTable table,
+ HoodieCopyOnWriteTableOperation copyOnWriteTableOperation, HoodieWriteConfig config,
+ String compactionInstantTime, AbstractHoodieWriteClient writeClient,
+ TaskContextSupplier taskContextSupplier) {
+ HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
+ HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
+ if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
+ writeClient.rollbackInflightCompaction(inflightInstant, table);
+ table.getMetaClient().reloadActiveTimeline();
+ }
+
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata = new HoodieWriteMetadata<>();
+ compactionMetadata.setWriteStatuses(context.createEmptyHoodieData());
+ try {
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ if (compactionPlan == null || (compactionPlan.getOperations() == null)
+ || (compactionPlan.getOperations().isEmpty())) {
+ // do nothing.
+ LOG.info("No compaction plan for instant " + compactionInstantTime);
+ return compactionMetadata;
+ }
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+ table.getMetaClient().reloadActiveTimeline();
+
+ HoodieTableMetaClient metaClient = table.getMetaClient();
Review comment:
this method is too lengthy. Can we try to add 1 or 2 private methods.
##########
File path: hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
##########
@@ -321,15 +324,14 @@ protected void completeCompaction(HoodieCommitMetadata metadata, JavaRDD<WriteSt
protected JavaRDD<WriteStatus> compact(String compactionInstantTime, boolean shouldComplete) {
HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
preWrite(compactionInstantTime, WriteOperationType.COMPACT, table.getMetaClient());
- HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
- HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
- if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
- rollbackInflightCompaction(inflightInstant, table);
- table.getMetaClient().reloadActiveTimeline();
- }
compactionTimer = metrics.getCompactionCtx();
- HoodieWriteMetadata<JavaRDD<WriteStatus>> compactionMetadata = table.compact(context, compactionInstantTime);
- JavaRDD<WriteStatus> statuses = compactionMetadata.getWriteStatuses();
+ HoodieSparkCopyOnWriteTable sparkCopyOnWriteTable =
+ new HoodieSparkCopyOnWriteTable(config, context, table.getMetaClient());
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata =
+ CompactHelpers.newInstance().compact(context, table, sparkCopyOnWriteTable,
+ config, compactionInstantTime, this, new SparkTaskContextSupplier());
Review comment:
same here, why not use existing context ?
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactHelpers.java
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.hudi.table.action.compact;
+
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.client.AbstractHoodieWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.CompactionOperation;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieCompactionException;
+import org.apache.hudi.table.HoodieCopyOnWriteTableOperation;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
+
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Base class helps to perform compact.
+ *
+ * @param <T> Type of payload in {@link org.apache.hudi.common.model.HoodieRecord}
+ */
+public class CompactHelpers<T extends HoodieRecordPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactHelpers.class);
+
+ private CompactHelpers() {
+ }
+
+ private static class CompactHelperHolder {
+ private static final CompactHelpers COMPACT_HELPERS = new CompactHelpers();
+ }
+
+ public static CompactHelpers newInstance() {
+ return CompactHelperHolder.COMPACT_HELPERS;
+ }
+
+ public HoodieWriteMetadata<HoodieData<WriteStatus>> compact(
+ HoodieEngineContext context, HoodieTable table,
+ HoodieCopyOnWriteTableOperation copyOnWriteTableOperation, HoodieWriteConfig config,
+ String compactionInstantTime, AbstractHoodieWriteClient writeClient,
+ TaskContextSupplier taskContextSupplier) {
+ HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
+ HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
+ if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
+ writeClient.rollbackInflightCompaction(inflightInstant, table);
+ table.getMetaClient().reloadActiveTimeline();
+ }
+
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata = new HoodieWriteMetadata<>();
+ compactionMetadata.setWriteStatuses(context.createEmptyHoodieData());
+ try {
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ if (compactionPlan == null || (compactionPlan.getOperations() == null)
+ || (compactionPlan.getOperations().isEmpty())) {
+ // do nothing.
+ LOG.info("No compaction plan for instant " + compactionInstantTime);
+ return compactionMetadata;
+ }
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+ table.getMetaClient().reloadActiveTimeline();
+
+ HoodieTableMetaClient metaClient = table.getMetaClient();
+ TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
+
+ // Here we firstly use the table schema as the reader schema to read
+ // log file.That is because in the case of MergeInto, the config.getSchema may not
+ // the same with the table schema.
+ try {
Review comment:
I guess these are not invoked in flink engine? can you double check that please. HoodieFlinkMergeOnReadTableCompactor
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
##########
@@ -68,6 +77,21 @@ public HoodieFlinkEngineContext(SerializableConfiguration hadoopConf, TaskContex
this.runtimeContext = ((FlinkTaskContextSupplier) taskContextSupplier).getFlinkRuntimeContext();
}
+ @Override
+ public HoodieAccumulator createNewAccumulator() {
Review comment:
createNewLongAccumulator
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/function/SerializablePairFlatMapFunction.java
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.hudi.common.function;
+
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.io.Serializable;
+import java.util.stream.Stream;
+
+/**
+ * A function that returns a stream of key-value pairs (Pair<K, V>).
Review comment:
minor: can you fix java docs "<" ">"
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
##########
@@ -86,6 +110,17 @@ public RuntimeContext getRuntimeContext() {
.collect(Collectors.toList());
}
+ @Override
+ public <I, K, V> Stream<ImmutablePair<K, V>> mapPartitionsToPairAndReduceByKey(
+ Stream<I> data, SerializablePairFlatMapFunction<Iterator<I>, K, V> flatMapToPairFunc,
+ SerializableBiFunction<V, V, V> reduceFunc, int parallelism) {
+ return throwingFlatMapToPairWrapper(flatMapToPairFunc).apply(data.parallel().iterator())
+ .collect(Collectors.groupingBy(Pair::getKey)).entrySet().stream()
Review comment:
may I know where have we used the "parallelism" within this method?
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
##########
@@ -369,7 +372,14 @@ public void completeCompaction(
protected List<WriteStatus> compact(String compactionInstantTime, boolean shouldComplete) {
// only used for metadata table, the compaction happens in single thread
try {
- List<WriteStatus> writeStatuses = FlinkCompactHelpers.compact(compactionInstantTime, this);
+ HoodieFlinkCopyOnWriteTable flinkCopyOnWriteTable = new HoodieFlinkCopyOnWriteTable(
+ getConfig(), getEngineContext(), getHoodieTable().getMetaClient());
+ HoodieWriteMetadata<HoodieData<WriteStatus>> writeMetadata =
+ CompactHelpers.newInstance().compact(context, getHoodieTable(), flinkCopyOnWriteTable,
+ config, compactionInstantTime, this, new FlinkTaskContextSupplier(null));
Review comment:
why create a new FlinkTaskContextSupplier? dont' we have one already locally ?
--
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] yihua commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719749416
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactHelpers.java
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.hudi.table.action.compact;
+
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.client.AbstractHoodieWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.CompactionOperation;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieCompactionException;
+import org.apache.hudi.table.HoodieCopyOnWriteTableOperation;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
+
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Base class helps to perform compact.
+ *
+ * @param <T> Type of payload in {@link org.apache.hudi.common.model.HoodieRecord}
+ */
+public class CompactHelpers<T extends HoodieRecordPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactHelpers.class);
+
+ private CompactHelpers() {
+ }
+
+ private static class CompactHelperHolder {
+ private static final CompactHelpers COMPACT_HELPERS = new CompactHelpers();
+ }
+
+ public static CompactHelpers newInstance() {
+ return CompactHelperHolder.COMPACT_HELPERS;
+ }
+
+ public HoodieWriteMetadata<HoodieData<WriteStatus>> compact(
+ HoodieEngineContext context, HoodieTable table,
+ HoodieCopyOnWriteTableOperation copyOnWriteTableOperation, HoodieWriteConfig config,
+ String compactionInstantTime, AbstractHoodieWriteClient writeClient,
+ TaskContextSupplier taskContextSupplier) {
+ HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
+ HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
+ if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
+ writeClient.rollbackInflightCompaction(inflightInstant, table);
+ table.getMetaClient().reloadActiveTimeline();
+ }
+
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata = new HoodieWriteMetadata<>();
+ compactionMetadata.setWriteStatuses(context.createEmptyHoodieData());
+ try {
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ if (compactionPlan == null || (compactionPlan.getOperations() == null)
+ || (compactionPlan.getOperations().isEmpty())) {
+ // do nothing.
+ LOG.info("No compaction plan for instant " + compactionInstantTime);
+ return compactionMetadata;
+ }
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+ table.getMetaClient().reloadActiveTimeline();
+
+ HoodieTableMetaClient metaClient = table.getMetaClient();
Review comment:
Yes. I refactor the logic in a different way, into CompactHelpers and RunCompactionActionExecutor: https://github.com/apache/hudi/pull/3741
--
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] nsivabalan commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719331449
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactHelpers.java
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.hudi.table.action.compact;
+
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.client.AbstractHoodieWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.CompactionOperation;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieCompactionException;
+import org.apache.hudi.table.HoodieCopyOnWriteTableOperation;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
+
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Base class helps to perform compact.
+ *
+ * @param <T> Type of payload in {@link org.apache.hudi.common.model.HoodieRecord}
+ */
+public class CompactHelpers<T extends HoodieRecordPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactHelpers.class);
+
+ private CompactHelpers() {
+ }
+
+ private static class CompactHelperHolder {
+ private static final CompactHelpers COMPACT_HELPERS = new CompactHelpers();
+ }
+
+ public static CompactHelpers newInstance() {
+ return CompactHelperHolder.COMPACT_HELPERS;
+ }
+
+ public HoodieWriteMetadata<HoodieData<WriteStatus>> compact(
+ HoodieEngineContext context, HoodieTable table,
+ HoodieCopyOnWriteTableOperation copyOnWriteTableOperation, HoodieWriteConfig config,
+ String compactionInstantTime, AbstractHoodieWriteClient writeClient,
+ TaskContextSupplier taskContextSupplier) {
+ HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
+ HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
+ if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
+ writeClient.rollbackInflightCompaction(inflightInstant, table);
+ table.getMetaClient().reloadActiveTimeline();
+ }
+
+ HoodieWriteMetadata<HoodieData<WriteStatus>> compactionMetadata = new HoodieWriteMetadata<>();
+ compactionMetadata.setWriteStatuses(context.createEmptyHoodieData());
+ try {
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ if (compactionPlan == null || (compactionPlan.getOperations() == null)
+ || (compactionPlan.getOperations().isEmpty())) {
+ // do nothing.
+ LOG.info("No compaction plan for instant " + compactionInstantTime);
+ return compactionMetadata;
+ }
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+ table.getMetaClient().reloadActiveTimeline();
+
+ HoodieTableMetaClient metaClient = table.getMetaClient();
+ TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
+
+ // Here we firstly use the table schema as the reader schema to read
+ // log file.That is because in the case of MergeInto, the config.getSchema may not
+ // the same with the table schema.
+ try {
Review comment:
@danny0405 : need some help on understanding some behavior diff between spark and flink wrt compation.
there is a code block which does some schema inference in spark which is skipped in flink.
[Flink link](https://github.com/apache/hudi/blob/47ed91799943271f219419cf209793a98b3f09b5/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/compact/HoodieFlinkMergeOnReadTableCompactor.java#L88), [spark link](https://github.com/apache/hudi/blob/47ed91799943271f219419cf209793a98b3f09b5/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/HoodieSparkMergeOnReadTableCompactor.java#L90).
In other words, this below code block is missing in flink.
```
HoodieTableMetaClient metaClient = hoodieTable.getMetaClient();
TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
// Here we firstly use the table schema as the reader schema to read
// log file.That is because in the case of MergeInto, the config.getSchema may not
// the same with the table schema.
try {
Schema readerSchema = schemaUtil.getTableAvroSchema(false);
config.setSchema(readerSchema.toString());
} catch (Exception e) {
// If there is no commit in the table, just ignore the exception.
}
```
Is this intentional? can you help us understand the behavior difference.
--
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] yihua closed pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua closed pull request #3727:
URL: https://github.com/apache/hudi/pull/3727
--
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 #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-928920877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f37aaed5a8a14fcad20eb2e761f05c0e1dfa4b0 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run travis` re-run the last Travis build
- `@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] yihua commented on a change in pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on a change in pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#discussion_r719751200
##########
File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
##########
@@ -369,7 +372,14 @@ public void completeCompaction(
protected List<WriteStatus> compact(String compactionInstantTime, boolean shouldComplete) {
// only used for metadata table, the compaction happens in single thread
try {
- List<WriteStatus> writeStatuses = FlinkCompactHelpers.compact(compactionInstantTime, this);
+ HoodieFlinkCopyOnWriteTable flinkCopyOnWriteTable = new HoodieFlinkCopyOnWriteTable(
+ getConfig(), getEngineContext(), getHoodieTable().getMetaClient());
+ HoodieWriteMetadata<HoodieData<WriteStatus>> writeMetadata =
+ CompactHelpers.newInstance().compact(context, getHoodieTable(), flinkCopyOnWriteTable,
+ config, compactionInstantTime, this, new FlinkTaskContextSupplier(null));
Review comment:
A new FlinkTaskContextSupplier instance is created in the original code. Since the table already has it, I changed it to reuse the existing FlinkTaskContextSupplier.
--
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] yihua commented on pull request #3727: [HUDI-2497] Refactor clean, restore, and compaction actions in hudi-client module
Posted by GitBox <gi...@apache.org>.
yihua commented on pull request #3727:
URL: https://github.com/apache/hudi/pull/3727#issuecomment-930614238
I'm going to split this PR into two given the complexity of the 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