You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "zhangyue19921010 (via GitHub)" <gi...@apache.org> on 2025/01/08 10:06:08 UTC
[PR] [HUDI-8780][RFC-83][WIP] Incremental Table Service [hudi]
zhangyue19921010 opened a new pull request, #12601:
URL: https://github.com/apache/hudi/pull/12601
### Change Logs
In Hudi, when scheduling Compaction and Clustering, the default behavior is to scan all partitions under the current table. When there are many historical partitions, such as 640,000 in our production environment, this scanning and planning operation becomes very inefficient. For Flink, it often leads to checkpoint timeouts, resulting in data delays.
As for cleaning, we already have the ability to do cleaning for incremental partitions.
This RFC will draw on the design of Incremental Clean to generalize the capability of processing incremental partitions to all table services, such as Clustering and Compaction.
### Impact
compaction and clustering
### Risk level (write none, low medium or high below)
medium
### Documentation Update
_Describe any necessary documentation update if there is any new feature, config, or user-facing change. If not, put "none"._
- _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
Re: [PR] [HUDI-8780] Incremental Table Service [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608758968
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"triggerType" : "PUSH"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 1a7290f66c485eaf59964b30d25ccfaf02003f82 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031) Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032)
<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-8780] Incremental Table Service [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608818634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"triggerType" : "PUSH"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3035",
"triggerID" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a7290f66c485eaf59964b30d25ccfaf02003f82 Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032) Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031)
* 02fe10bf4122514d17cabcafe33eed70e5d2df1b Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3035)
<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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1923034258
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
Review Comment:
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
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1920932312
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -116,7 +116,7 @@ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableSe
String rightBoundary = instantTime;
// compute [leftBoundary, rightBoundary) as time window
HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
- Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().findInstantsAfter(leftBoundary).filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
Review Comment:
why filtering the instant time by `leftBoundary`?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915925767
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/IncrementalPartitionAwareStrategy.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.List;
+
+/**
+ * Marking strategy interface.
+ *
+ * Any Strategy implement this `IncrementalPartitionAwareStrategy` could have the ability to perform incremental partitions processing.
+ * At this time, Incremental partitions should be passed to the current strategy.
+ */
+public interface IncrementalPartitionAwareStrategy {
+
+ /**
+ * Filter the given incremental partitions.
+ * @param writeConfig
+ * @param incrementalPartitions
Review Comment:
changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917613978
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/action/compact/TestHoodieCompactor.java:
##########
@@ -344,6 +350,82 @@ public void testCompactionSpecifyPartition(String regex, List<String> expectedCo
}
}
+ @Test
+ public void testPartitionsForIncrCompaction() throws Exception {
+ HoodieWriteConfig config = getConfigBuilder()
+ .withIncrementalTableServiceEnable(true)
+ .withCompactionConfig(HoodieCompactionConfig
+ .newBuilder()
+ .withCompactionStrategy(new SmallBoundedIOCompactionStrategy())
+ .withMaxNumDeltaCommitsBeforeCompaction(2).build())
+ .build();
+ String[] partitions = {"20250115"};
+
+ HoodieWriteConfig config2 = getConfigBuilder()
+ .withIncrementalTableServiceEnable(true)
+ .withCompactionConfig(HoodieCompactionConfig
+ .newBuilder()
+ .withMaxNumDeltaCommitsBeforeCompaction(2).build())
+ .build();
+ String[] partitions2 = {"20250116"};
+
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
+ prepareRecords(writeClient, config, partitions);
+ prepareRecords(writeClient, config, partitions);
+ compact(writeClient, writeClient.createNewInstantTime());
+ HoodieCompactionPlan compactionPlan1 = getLatestCompactionPlan();
+
+ List<String> affectedPartitions = compactionPlan1.getOperations().stream()
+ .map(HoodieCompactionOperation::getPartitionPath).collect(Collectors.toList());
+ // compaction including 20250115
+ assertTrue(affectedPartitions.contains(partitions[0]));
+ List<String> missingPartitions = compactionPlan1.getMissingSchedulePartitions();
+ // compaction missing 20250115 because not all the fileSlices are all processed.
+ assertTrue(missingPartitions.contains(partitions[0]));
+ }
+
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config2)) {
+ prepareRecords(writeClient, config2, partitions2);
+ prepareRecords(writeClient, config2, partitions2);
+ compact(writeClient, writeClient.createNewInstantTime());
+ HoodieCompactionPlan compactionPlan2 = getLatestCompactionPlan();
+ List<String> affectedPartitions2 = compactionPlan2.getOperations().stream()
+ .map(HoodieCompactionOperation::getPartitionPath).collect(Collectors.toList());
+ // compaction including 20250115 (fetched from recorded missing partitions)
+ assertTrue(affectedPartitions2.contains(partitions[0]));
Review Comment:
do we test the fallback for archived commits?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2603801011
@hudi-bot run azure
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1921747178
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -116,7 +116,7 @@ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableSe
String rightBoundary = instantTime;
// compute [leftBoundary, rightBoundary) as time window
HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
- Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().findInstantsAfter(leftBoundary).filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
Review Comment:
@danny0405 Nice catch Danny.
At first I just wanted to filter out the left boundary instant itself, but obviously I called the API `findInstantsAfter` by mistake. Here we should get and filter the entire active time line complete instants
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1921747178
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -116,7 +116,7 @@ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableSe
String rightBoundary = instantTime;
// compute [leftBoundary, rightBoundary) as time window
HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
- Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().findInstantsAfter(leftBoundary).filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
Review Comment:
@danny0405 Nice catch Danny.
At first I just wanted to filter out the left boundary instant itself, but obviously I called wrong API `findInstantsAfter` by mistake. Here we should get and filter the entire active time line complete instants
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912714671
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/IncrementalPartitionAwareStrategy.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.List;
+
+/**
+ * Marking strategy interface.
+ *
+ * Any Strategy implement this `IncrementalPartitionAwareStrategy` could have the ability to perform incremental partitions processing.
+ * At this time, Incremental partitions should be passed to the current strategy.
Review Comment:
```java
<p> Any strategy class that implements this `IncrementalPartitionAwareStrategy` could have the ability to perform incremental partitions processing.
Currently, Incremental partitions will be passed to the strategy instance as a best-effort. In the following cases, the partitions would fallback to full partition list:
<ul>
<li></li>
...
</ul>
```
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912735867
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -68,6 +72,7 @@ protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String
- (o1.getBaseFile().isPresent() ? o1.getBaseFile().get().getFileSize() : writeConfig.getParquetMaxFileSize())));
long totalSizeSoFar = 0;
+ boolean isAllSlicesIncluded = true;
Review Comment:
isAllSlicesIncluded -> partitialScheduled?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1923049710
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
Review Comment:
changed all the related `enable` to `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
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915928949
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/BaseConsistentHashingBucketClusteringPlanStrategy.java:
##########
@@ -128,7 +129,7 @@ protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String
.build();
}).collect(Collectors.toList()));
}
- return ret.stream();
+ return Pair.of(ret.stream(), true);
Review Comment:
This boolean value is used to indicate whether all candidate fileslice under the current partition have been processed. The external world uses this as a basis to determine whether the current partition is added to missingPartitions.
For `BaseConsistentHashingBucketClusteringPlanStrategy` it always return true.
For other `PartitionAwareClusteringPlanStrategy` like `SparkSizeBasedClusteringPlanStrategy` it may return false.
Such as , 100 fileSlices are passed in, but due to the limitation of `writeConfig.getClusteringMaxNumGroups()`, only 10 of them are processed. In this case, false should be returned.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915974387
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (incrementalPartitions.isEmpty()) {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ } else if (lastCompleteTableServiceInstant.isPresent()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
Review Comment:
why ignore cluster?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915980054
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -68,6 +75,7 @@ protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String
- (o1.getBaseFile().isPresent() ? o1.getBaseFile().get().getFileSize() : writeConfig.getParquetMaxFileSize())));
long totalSizeSoFar = 0;
+ boolean partialScheduled = true;
Review Comment:
The default value should be false I guess.
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912698539
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
+ .key("hoodie.incremental.tableservice.enable")
+ .defaultValue(true)
+ .markAdvanced()
+ .sinceVersion("1.0.0")
+ .withDocumentation("Whether to enable incremental table service.");
Review Comment:
Can we elaborate the scope of the kinds of table services that are affected by this option.
enable -> 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
Re: [PR] [HUDI-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912738022
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java:
##########
@@ -182,12 +186,14 @@ public HoodieCompactionPlan generateCompactionPlan(String compactionInstant) thr
return compactionPlan;
}
- protected abstract HoodieCompactionPlan getCompactionPlan(HoodieTableMetaClient metaClient, List<HoodieCompactionOperation> operations);
+ protected abstract List<String> getPartitions();
+
+ protected abstract HoodieCompactionPlan getCompactionPlan(HoodieTableMetaClient metaClient, List<HoodieCompactionOperation> operations, Pair<List<String>,List<String>> partitionPair);
Review Comment:
Do we need the missing partitions to generate the plan?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915925564
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/IncrementalPartitionAwareStrategy.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.List;
+
+/**
+ * Marking strategy interface.
+ *
+ * Any Strategy implement this `IncrementalPartitionAwareStrategy` could have the ability to perform incremental partitions processing.
+ * At this time, Incremental partitions should be passed to the current strategy.
Review Comment:
changed.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/IncrementalPartitionAwareStrategy.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.List;
+
+/**
+ * Marking strategy interface.
+ *
+ * Any Strategy implement this `IncrementalPartitionAwareStrategy` could have the ability to perform incremental partitions processing.
+ * At this time, Incremental partitions should be passed to the current strategy.
Review Comment:
changed.
```
/**
* Marking interface for table service strategy that utilize incremental partitions.
*
* <p> Any strategy class that implements this `IncrementalPartitionAwareStrategy` could have the ability to perform incremental partitions processing.
* Currently, Incremental partitions will be passed to the strategy instance as a best-effort. In the following cases, the partitions would fallback to full partition list:
*
* <ul>
* <li> Executing Table Service for the first time. </li>
* <li> The last completed table service instant is archived. </li>
* <li> Any exception thrown during retrieval of incremental partitions. </li>
* </ul>
*/
```
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917565080
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
+ .key("hoodie.incremental.tableservice.enabled")
Review Comment:
Incremental processing is used everywhere including UT. There is a switch here (default True) just in case. In some scenarios, users need full processing. Users can manually turn off the logic of incremental processing (similar to the switch between MDT and 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
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1919673549
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/TestBaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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;
+
+import org.apache.hudi.HoodieTestCommitGenerator;
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieClusteringStrategy;
+import org.apache.hudi.avro.model.HoodieRequestedReplaceMetadata;
+import org.apache.hudi.avro.model.HoodieSliceInfo;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.testutils.FileCreateUtils;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.action.BaseTableServicePlanActionExecutor;
+import org.apache.hudi.table.action.IncrementalPartitionAwareStrategy;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestBaseTableServicePlanActionExecutor {
+
+ @TempDir
+ File tempFile;
+ private static final StorageConfiguration<?> CONF = getDefaultStorageConf();
+ private final HoodieEngineContext context = new HoodieLocalEngineContext(CONF);
+
+ /**
+ * Clustering commit : cl
+ * Commit : c
+ * |requestTime<---commit type--->completionTime|(written partitions in current commit)
+ * ---------------------------------------------------------------------------------------------------------------------------------------------> timeline
+ * |0<-c1->1|(part=0000)
+ * |2<------cl2------>4|
+ * |3<-----------c3--------->6|(part=0003)
+ * |5<-----------c4--------->7|(part=0005)
+ * |8<-------------------c5----------------->10|(part=0008)
+ * | start incr clustering at 9
+ */
+ @Test
+ public void testExecutorWithMultiWriter() throws Exception {
+ String tableName = "testTable";
+ String tablePath = tempFile.getAbsolutePath() + StoragePath.SEPARATOR + tableName;
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(
+ HoodieTestUtils.getDefaultStorageConf(), tablePath, HoodieTableType.COPY_ON_WRITE, tableName);
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder().withPath(tablePath)
+ .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ .withMarkersType("DIRECT")
+ .build();
+
+ // <requestTime, CompletionTime>
+ HashMap<String, String> instants = new HashMap<>();
+ instants.put("0000", "0001"); // commit 1
+ instants.put("0002", "0004"); // clustering 1
+ instants.put("0003", "0006"); // commit 2
+ instants.put("0005", "0007"); // commit 3
+ instants.put("0008", "0010"); // commit 4
+
+ prepareTimeline(metaClient, instants);
+
+ DummyTableServicePlanActionExecutor executor = new DummyTableServicePlanActionExecutor(context, writeConfig, getMockHoodieTable(metaClient), "0009");
+ Set<String> incrementalPartitions = (Set<String>)executor.getIncrementalPartitions(TableServiceType.CLUSTER).getRight();
+
+ assertEquals(incrementalPartitions.size(), 2);
+ assertTrue(incrementalPartitions.contains("0003"));
+ assertTrue(incrementalPartitions.contains("0005"));
+ assertFalse(incrementalPartitions.contains("0008"));
+ }
+
+ @Test
+ public void testGetPartitionsFallbackToFullScan() throws Exception {
+ String tableName = "testTable";
+ String tablePath = tempFile.getAbsolutePath() + StoragePath.SEPARATOR + tableName;
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(
+ HoodieTestUtils.getDefaultStorageConf(), tablePath, HoodieTableType.COPY_ON_WRITE, tableName);
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder().withPath(tablePath)
+ .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ .withMarkersType("DIRECT")
+ .build();
+
+ // <requestTime, CompletionTime>
+ HashMap<String, String> instants = new HashMap<>();
+ instants.put("0000", "0001"); // commit 1
+ instants.put("0002", "0004"); // clustering 1
+ instants.put("0003", "0006"); // commit 2
+ instants.put("0005", "0007"); // commit 3
+ instants.put("0008", "0010"); // commit 4
+
+ prepareTimeline(metaClient, instants);
+ instants.keySet().forEach(instant -> {
+ try {
+ FileCreateUtils.createPartitionMetaFile(tablePath, instant);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ DummyStrategy incrementalStrategy = new DummyStrategy();
+ DummyTableServicePlanActionExecutor executor = new DummyTableServicePlanActionExecutor(context, writeConfig, getMockHoodieTable(metaClient), "0009");
+ List<String> partitions = (List<String>) executor.getPartitions(incrementalStrategy, TableServiceType.CLUSTER);
+ assertEquals(partitions.size(), 2);
+
+ // Simulation archive commit instant 0000 and clustering instant 0002
+ Path path = new Path(tablePath);
+ FileSystem fs = path.getFileSystem(new Configuration());
+ StoragePath timelinePath = metaClient.getTimelinePath();
+ Arrays.stream(fs.listStatus(new Path(timelinePath.toString()))).forEach(instant -> {
+ if (instant.getPath().toString().contains("0000") || instant.getPath().toString().contains("0002")) {
+ try {
+ fs.delete(instant.getPath(), true);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+
+ // fall back to get all partitions
+ List<String> allPartitions = (List<String>) executor.getPartitions(incrementalStrategy, TableServiceType.CLUSTER);
+ assertEquals(allPartitions.stream().sorted().collect(Collectors.toList()), instants.keySet().stream().sorted().collect(Collectors.toList()));
+ }
+
+ @Test
+ public void testContinuousEmptyCommits() throws Exception {
Review Comment:
testContinuousEmptyCommits
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1919720469
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -107,6 +107,7 @@ public static HoodieWriteConfig createMetadataWriteConfig(
// Create the write config for the metadata table by borrowing options from the main write config.
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
+ .withIncrementalTableServiceEnable(false)
Review Comment:
Or we could deprecat right boundary which means [left-boundary, +∞)
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -107,6 +107,7 @@ public static HoodieWriteConfig createMetadataWriteConfig(
// Create the write config for the metadata table by borrowing options from the main write config.
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
+ .withIncrementalTableServiceEnable(false)
Review Comment:
Or we could deprecate right boundary which means [left-boundary, +∞)
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2606095692
@hudi-bot run azure
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915978093
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (incrementalPartitions.isEmpty()) {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ } else if (lastCompleteTableServiceInstant.isPresent()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
+ return metadata.getWriteStats().stream().map(HoodieWriteStat::getPartitionPath);
+ }
+ }
+ return Stream.empty();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to get commit meta " + instant, e);
+ }
+ }).collect(Collectors.toSet());
+
+ partitionsInCommitMeta.addAll(missingPartitions);
+ return Pair.of(lastCompleteTableServiceInstant, partitionsInCommitMeta);
+ }
+
+ private boolean filterCommitByTableType(HoodieInstant instant) {
+ // for completed replace commit, need CommitMetadata to distinguish clustering or insert overwrite
+ switch (table.getMetaClient().getTableType()) {
+ case MERGE_ON_READ: {
+ // for mor only take cares of delta commit and replace commit
+ Set<String> operations = CollectionUtils.createSet(DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION);
+ return operations.contains(instant.getAction());
+ }
+ case COPY_ON_WRITE: {
+ // for mor only take cares of commit and replace commit
+ Set<String> operations = CollectionUtils.createSet(COMMIT_ACTION, REPLACE_COMMIT_ACTION);
+ return operations.contains(instant.getAction());
+ }
+ default:
+ throw new HoodieException("Un-supported table type " + table.getMetaClient().getTableType());
+ }
+ }
+
+ public Pair<Option<HoodieInstant>, List<String>> fetchMissingPartitions(TableServiceType tableServiceType) {
+ if (!config.isIncrementalTableServiceEnable()) {
+ return Pair.of(Option.empty(), new ArrayList<>());
Review Comment:
Collections.emptyList?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915930128
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/CompactionStrategy.java:
##########
@@ -71,10 +76,14 @@ public Map<String, Double> captureMetrics(HoodieWriteConfig writeConfig, FileSli
* @return Compaction plan to be scheduled.
*/
public HoodieCompactionPlan generateCompactionPlan(HoodieWriteConfig writeConfig,
- List<HoodieCompactionOperation> operations, List<HoodieCompactionPlan> pendingCompactionPlans) {
+ List<HoodieCompactionOperation> operations, List<HoodieCompactionPlan> pendingCompactionPlans, Map<String, String> params, Pair<List<String>, List<String>> partitionPair) {
// Strategy implementation can overload this method to set specific compactor-id
+ Set<String> missingPartitions = new HashSet<>(partitionPair.getRight());
+ List<HoodieCompactionOperation> operationsToProcess = orderAndFilter(writeConfig, operations, pendingCompactionPlans, missingPartitions);
+ List<String> res = writeConfig.isIncrementalTableServiceEnable() ? new ArrayList<>(missingPartitions) : new ArrayList<>();
Review Comment:
chenged.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917569458
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
+ .key("hoodie.incremental.tableservice.enabled")
Review Comment:
changed to `hoodie.table.services.incremental.enabled`
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (lastCompleteTableServiceInstant.isPresent()) {
+ if (!incrementalPartitions.isEmpty()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ }
+ }
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
+ return metadata.getWriteStats().stream().map(HoodieWriteStat::getPartitionPath);
+ }
+ }
+ return Stream.empty();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to get commit meta " + instant, e);
+ }
+ }).collect(Collectors.toSet());
+
+ partitionsInCommitMeta.addAll(missingPartitions);
+ return Pair.of(lastCompleteTableServiceInstant, partitionsInCommitMeta);
+ }
+
+ private boolean filterCommitByTableType(HoodieInstant instant) {
+ // for completed replace commit, need CommitMetadata to distinguish clustering or insert overwrite
+ switch (table.getMetaClient().getTableType()) {
+ case MERGE_ON_READ: {
+ // for mor only take cares of delta commit and replace commit
+ Set<String> operations = CollectionUtils.createSet(DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION);
Review Comment:
changed. Thanks
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917568634
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -131,36 +138,74 @@ public Option<HoodieClusteringPlan> generateClusteringPlan() {
String partitionSelected = config.getClusteringPartitionSelected();
LOG.info("Scheduling clustering partitionSelected: {}", partitionSelected);
- List<String> partitionPaths;
+ List<String> partitionPaths = new ArrayList<>();
+ List<String> missingPartitions = new ArrayList<>();
+ List<HoodieClusteringGroup> clusteringGroups;
+ int clusteringMaxNumGroups = getWriteConfig().getClusteringMaxNumGroups();
if (StringUtils.isNullOrEmpty(partitionSelected)) {
// get matched partitions if set
- partitionPaths = getRegexPatternMatchedPartitions(config, FSUtils.getAllPartitionPaths(
- getEngineContext(), metaClient.getStorage(), config.getMetadataConfig(), metaClient.getBasePath()));
+ partitionPaths = getRegexPatternMatchedPartitions(config, partitions.get());
// filter the partition paths if needed to reduce list status
} else {
partitionPaths = Arrays.asList(partitionSelected.split(","));
+ // Users may temporarily set specific partitions for clustering.
+ // Ensure the coherence of the missing partitions.
+ missingPartitions = (List<String>)executor.fetchMissingPartitions(TableServiceType.CLUSTER).getRight();
}
- partitionPaths = filterPartitionPaths(partitionPaths);
+ Pair<List<String>, List<String>> partitionsPair = filterPartitionPaths(getWriteConfig(), partitionPaths);
+ partitionPaths = partitionsPair.getLeft();
+ missingPartitions.addAll(partitionsPair.getRight());
LOG.info("Scheduling clustering partitionPaths: {}", partitionPaths);
+ LOG.info("Missing Scheduled clustering partitionPaths: {}", missingPartitions);
if (partitionPaths.isEmpty()) {
// In case no partitions could be picked, return no clustering plan
return Option.empty();
}
- List<HoodieClusteringGroup> clusteringGroups = getEngineContext()
- .flatMap(
- partitionPaths,
- partitionPath -> {
- List<FileSlice> fileSlicesEligible = getFileSlicesEligibleForClustering(partitionPath).collect(Collectors.toList());
- return buildClusteringGroupsForPartition(partitionPath, fileSlicesEligible).limit(getWriteConfig().getClusteringMaxNumGroups());
- },
- partitionPaths.size())
- .stream()
- .limit(getWriteConfig().getClusteringMaxNumGroups())
- .collect(Collectors.toList());
+ List<Pair<List<HoodieClusteringGroup>, String>> res = getEngineContext().map(partitionPaths, partitionPath -> {
+ List<FileSlice> fileSlicesEligible = getFileSlicesEligibleForClustering(partitionPath).collect(Collectors.toList());
+ Pair<Stream<HoodieClusteringGroup>, Boolean> groupPair = buildClusteringGroupsForPartition(partitionPath, fileSlicesEligible);
+ List<HoodieClusteringGroup> clusteringGroupsPartition = groupPair.getLeft().collect(Collectors.toList());
+ Boolean allProcessed = groupPair.getRight();
Review Comment:
changed. Sorry for missing 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
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1919720469
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -107,6 +107,7 @@ public static HoodieWriteConfig createMetadataWriteConfig(
// Create the write config for the metadata table by borrowing options from the main write config.
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
+ .withIncrementalTableServiceEnable(false)
Review Comment:
Or we could deprecate right boundary which means [left-boundary, +∞)
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608669743
Trigger CI using https://github.com/apache/hudi/pull/12693
--
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-8780] Incremental Table Service [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608782629
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"triggerType" : "PUSH"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a7290f66c485eaf59964b30d25ccfaf02003f82 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032) Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031)
* 02fe10bf4122514d17cabcafe33eed70e5d2df1b 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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912699776
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Set<String> incrementalPartitions = getIncrementalPartitions(type);
+ if (!incrementalPartitions.isEmpty()) {
Review Comment:
Why fall back to full partitions if the incremental partition list is empty?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915915684
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
+ .key("hoodie.incremental.tableservice.enable")
+ .defaultValue(true)
+ .markAdvanced()
+ .sinceVersion("1.0.0")
+ .withDocumentation("Whether to enable incremental table service.");
Review Comment:
changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915929990
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java:
##########
@@ -182,12 +186,14 @@ public HoodieCompactionPlan generateCompactionPlan(String compactionInstant) thr
return compactionPlan;
}
- protected abstract HoodieCompactionPlan getCompactionPlan(HoodieTableMetaClient metaClient, List<HoodieCompactionOperation> operations);
+ protected abstract List<String> getPartitions();
+
+ protected abstract HoodieCompactionPlan getCompactionPlan(HoodieTableMetaClient metaClient, List<HoodieCompactionOperation> operations, Pair<List<String>,List<String>> partitionPair);
Review Comment:
because we need to record this missing partitions(`.setMissingSchedulePartitions(res)`) in plan during `getCompactionPlan function`
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/BoundedIOCompactionStrategy.java:
##########
@@ -46,7 +47,7 @@ public List<HoodieCompactionOperation> orderAndFilter(HoodieWriteConfig writeCon
targetIORemaining -= opIo;
finalOperations.add(op);
if (targetIORemaining <= 0) {
- return finalOperations;
+ missingPartitions.add(op.getPartitionPath());
Review Comment:
changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2591633203
Hi @danny0405 Thanks for your review. All comments are changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915918399
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Set<String> incrementalPartitions = getIncrementalPartitions(type);
+ if (!incrementalPartitions.isEmpty()) {
Review Comment:
changed.
For now, there are two situations fall back to get all partitions
1. incremental partitions getting throws exceptions.
2. last completed table service instant is archived.
AND
when incremental partition list is empty , we will skip current schedule.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915980851
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -131,36 +138,74 @@ public Option<HoodieClusteringPlan> generateClusteringPlan() {
String partitionSelected = config.getClusteringPartitionSelected();
LOG.info("Scheduling clustering partitionSelected: {}", partitionSelected);
- List<String> partitionPaths;
+ List<String> partitionPaths = new ArrayList<>();
+ List<String> missingPartitions = new ArrayList<>();
+ List<HoodieClusteringGroup> clusteringGroups;
+ int clusteringMaxNumGroups = getWriteConfig().getClusteringMaxNumGroups();
if (StringUtils.isNullOrEmpty(partitionSelected)) {
// get matched partitions if set
- partitionPaths = getRegexPatternMatchedPartitions(config, FSUtils.getAllPartitionPaths(
- getEngineContext(), metaClient.getStorage(), config.getMetadataConfig(), metaClient.getBasePath()));
+ partitionPaths = getRegexPatternMatchedPartitions(config, partitions.get());
// filter the partition paths if needed to reduce list status
} else {
partitionPaths = Arrays.asList(partitionSelected.split(","));
+ // Users may temporarily set specific partitions for clustering.
+ // Ensure the coherence of the missing partitions.
+ missingPartitions = (List<String>)executor.fetchMissingPartitions(TableServiceType.CLUSTER).getRight();
}
- partitionPaths = filterPartitionPaths(partitionPaths);
+ Pair<List<String>, List<String>> partitionsPair = filterPartitionPaths(getWriteConfig(), partitionPaths);
+ partitionPaths = partitionsPair.getLeft();
+ missingPartitions.addAll(partitionsPair.getRight());
LOG.info("Scheduling clustering partitionPaths: {}", partitionPaths);
+ LOG.info("Missing Scheduled clustering partitionPaths: {}", missingPartitions);
if (partitionPaths.isEmpty()) {
// In case no partitions could be picked, return no clustering plan
return Option.empty();
}
- List<HoodieClusteringGroup> clusteringGroups = getEngineContext()
- .flatMap(
- partitionPaths,
- partitionPath -> {
- List<FileSlice> fileSlicesEligible = getFileSlicesEligibleForClustering(partitionPath).collect(Collectors.toList());
- return buildClusteringGroupsForPartition(partitionPath, fileSlicesEligible).limit(getWriteConfig().getClusteringMaxNumGroups());
- },
- partitionPaths.size())
- .stream()
- .limit(getWriteConfig().getClusteringMaxNumGroups())
- .collect(Collectors.toList());
+ List<Pair<List<HoodieClusteringGroup>, String>> res = getEngineContext().map(partitionPaths, partitionPath -> {
+ List<FileSlice> fileSlicesEligible = getFileSlicesEligibleForClustering(partitionPath).collect(Collectors.toList());
+ Pair<Stream<HoodieClusteringGroup>, Boolean> groupPair = buildClusteringGroupsForPartition(partitionPath, fileSlicesEligible);
+ List<HoodieClusteringGroup> clusteringGroupsPartition = groupPair.getLeft().collect(Collectors.toList());
+ Boolean allProcessed = groupPair.getRight();
Review Comment:
partialScheduled?
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912740846
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/CompactionStrategy.java:
##########
@@ -71,10 +76,14 @@ public Map<String, Double> captureMetrics(HoodieWriteConfig writeConfig, FileSli
* @return Compaction plan to be scheduled.
*/
public HoodieCompactionPlan generateCompactionPlan(HoodieWriteConfig writeConfig,
- List<HoodieCompactionOperation> operations, List<HoodieCompactionPlan> pendingCompactionPlans) {
+ List<HoodieCompactionOperation> operations, List<HoodieCompactionPlan> pendingCompactionPlans, Map<String, String> params, Pair<List<String>, List<String>> partitionPair) {
// Strategy implementation can overload this method to set specific compactor-id
+ Set<String> missingPartitions = new HashSet<>(partitionPair.getRight());
+ List<HoodieCompactionOperation> operationsToProcess = orderAndFilter(writeConfig, operations, pendingCompactionPlans, missingPartitions);
+ List<String> res = writeConfig.isIncrementalTableServiceEnable() ? new ArrayList<>(missingPartitions) : new ArrayList<>();
Review Comment:
Maybe null as the default value to avoid deserialization.
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912729316
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -55,7 +59,7 @@ public PartitionAwareClusteringPlanStrategy(HoodieTable table, HoodieEngineConte
/**
* Create Clustering group based on files eligible for clustering in the partition.
*/
- protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String partitionPath, List<FileSlice> fileSlices) {
+ protected Pair<Stream<HoodieClusteringGroup>, Boolean> buildClusteringGroupsForPartition(String partitionPath, List<FileSlice> fileSlices) {
Review Comment:
Can we add some doc on the boolean flag.
--
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-8780] Incremental Table Service [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608757585
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a7290f66c485eaf59964b30d25ccfaf02003f82 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-8780] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2612968213
> @danny0405 @zhangyue19921010 `TestHoodieCompactionStrategy.testBoundedIOSimple` is flaky after this PR was merged - [HUDI-8910](https://issues.apache.org/jira/browse/HUDI-8910). Could you please take a look?
ack
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917568212
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (incrementalPartitions.isEmpty()) {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ } else if (lastCompleteTableServiceInstant.isPresent()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
+ return metadata.getWriteStats().stream().map(HoodieWriteStat::getPartitionPath);
+ }
+ }
+ return Stream.empty();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to get commit meta " + instant, e);
+ }
+ }).collect(Collectors.toSet());
+
+ partitionsInCommitMeta.addAll(missingPartitions);
+ return Pair.of(lastCompleteTableServiceInstant, partitionsInCommitMeta);
+ }
+
+ private boolean filterCommitByTableType(HoodieInstant instant) {
+ // for completed replace commit, need CommitMetadata to distinguish clustering or insert overwrite
+ switch (table.getMetaClient().getTableType()) {
+ case MERGE_ON_READ: {
+ // for mor only take cares of delta commit and replace commit
+ Set<String> operations = CollectionUtils.createSet(DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION);
+ return operations.contains(instant.getAction());
+ }
+ case COPY_ON_WRITE: {
+ // for mor only take cares of commit and replace commit
+ Set<String> operations = CollectionUtils.createSet(COMMIT_ACTION, REPLACE_COMMIT_ACTION);
+ return operations.contains(instant.getAction());
+ }
+ default:
+ throw new HoodieException("Un-supported table type " + table.getMetaClient().getTableType());
+ }
+ }
+
+ public Pair<Option<HoodieInstant>, List<String>> fetchMissingPartitions(TableServiceType tableServiceType) {
+ if (!config.isIncrementalTableServiceEnable()) {
+ return Pair.of(Option.empty(), new ArrayList<>());
Review Comment:
changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917568421
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -68,6 +75,7 @@ protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String
- (o1.getBaseFile().isPresent() ? o1.getBaseFile().get().getFileSize() : writeConfig.getParquetMaxFileSize())));
long totalSizeSoFar = 0;
+ boolean partialScheduled = true;
Review Comment:
exactly Missing here. changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912715243
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/IncrementalPartitionAwareStrategy.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.List;
+
+/**
+ * Marking strategy interface.
+ *
+ * Any Strategy implement this `IncrementalPartitionAwareStrategy` could have the ability to perform incremental partitions processing.
+ * At this time, Incremental partitions should be passed to the current strategy.
+ */
+public interface IncrementalPartitionAwareStrategy {
+
+ /**
+ * Filter the given incremental partitions.
+ * @param writeConfig
+ * @param incrementalPartitions
Review Comment:
partitions
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912707203
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/IncrementalPartitionAwareStrategy.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.List;
+
+/**
+ * Marking strategy interface.
Review Comment:
Marking interface for table service srategy that utilitize incremental partitions.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917568081
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (incrementalPartitions.isEmpty()) {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ } else if (lastCompleteTableServiceInstant.isPresent()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
Review Comment:
No need to ignore actually. `filterCommitByTableType` is good enough.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (incrementalPartitions.isEmpty()) {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ } else if (lastCompleteTableServiceInstant.isPresent()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
Review Comment:
No need to ignore actually. Maybe `filterCommitByTableType` is good enough.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1919703175
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -107,6 +107,7 @@ public static HoodieWriteConfig createMetadataWriteConfig(
// Create the write config for the metadata table by borrowing options from the main write config.
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
+ .withIncrementalTableServiceEnable(false)
Review Comment:
Hi @danny0405 Maybe we need to disable incremental compaction for MDT. Since MDT compaction instant is not based on processing time but based on `https://github.com/apache/hudi/blob/dee6274e56d8b7d3852104d3e45377f4909466cd/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java#L1576`
Which could cause selecting instants based on window is inaccurate, thus missing compaction
what do u think?
This is a screenshot of the debug process during the specific test.
<img width="1339" alt="截屏2025-01-17 15 46 49" src="https://github.com/user-attachments/assets/e9ae1edd-5d26-4df8-b6a0-7fc8c62e6d28" />
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915929252
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -55,7 +59,7 @@ public PartitionAwareClusteringPlanStrategy(HoodieTable table, HoodieEngineConte
/**
* Create Clustering group based on files eligible for clustering in the partition.
*/
- protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String partitionPath, List<FileSlice> fileSlices) {
+ protected Pair<Stream<HoodieClusteringGroup>, Boolean> buildClusteringGroupsForPartition(String partitionPath, List<FileSlice> fileSlices) {
Review Comment:
added.
```
/**
* Create Clustering group based on files eligible for clustering in the partition.
* return stream of HoodieClusteringGroup and boolean partial Scheduled indicating whether all given fileSlices in the current partition have been processed.
* For example, if some file slices will not be processed due to writeConfig.getClusteringMaxNumGroups(), then return false
*/
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/PartitionAwareClusteringPlanStrategy.java:
##########
@@ -68,6 +72,7 @@ protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String
- (o1.getBaseFile().isPresent() ? o1.getBaseFile().get().getFileSize() : writeConfig.getParquetMaxFileSize())));
long totalSizeSoFar = 0;
+ boolean isAllSlicesIncluded = true;
Review Comment:
changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915936308
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
+ .key("hoodie.incremental.tableservice.enabled")
Review Comment:
Does this option exist for some corner cases, we should always try incremental scheduling I 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
Re: [PR] [HUDI-8780] Incremental Table Service [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608783918
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"triggerType" : "PUSH"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a7290f66c485eaf59964b30d25ccfaf02003f82 Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032) Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031)
* 02fe10bf4122514d17cabcafe33eed70e5d2df1b 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-8780] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 merged PR #12601:
URL: https://github.com/apache/hudi/pull/12601
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2581604354
Hi @TheR1sing3un ， Thanks for your attention.
> @zhangyue19921010 Hi, judging from the rfc content, the goal this time is to use incremental `compaction` and `clustering`. Should we consider incorporating the incremental processing logic for `clean` as well?
Sure thing we can build a unified incremental policy. But as we know, clean action is not a Strategy-Coding-Style-Action, even though we have many different cleaning strategies(clean by commits or clean by versions). So that we may need to reconstruct the clean plan phase and abstract it into different strategy objects. We can do it in the next PR.
> By the way, whether we can solve this problem: #11647 , it seems that we can solve it at the time of implementation.
Unfortunately, this PR shouldn't solve the problem. This PR solves the problem of incremental processing, that is, how to process only incremental partitions next time after the last table service is completed. #11647 focuses on how to trigger the first action in an elegant way
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912739466
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/BoundedIOCompactionStrategy.java:
##########
@@ -46,7 +47,7 @@ public List<HoodieCompactionOperation> orderAndFilter(HoodieWriteConfig writeCon
targetIORemaining -= opIo;
finalOperations.add(op);
if (targetIORemaining <= 0) {
- return finalOperations;
+ missingPartitions.add(op.getPartitionPath());
Review Comment:
Usually we do not modify the passed in collections.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915974387
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (incrementalPartitions.isEmpty()) {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ } else if (lastCompleteTableServiceInstant.isPresent()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
Review Comment:
why ignore cluster? Maybe we should ingore based on the passed in table service type.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915983411
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/BoundedIOCompactionStrategy.java:
##########
@@ -46,7 +47,7 @@ public List<HoodieCompactionOperation> orderAndFilter(HoodieWriteConfig writeCon
targetIORemaining -= opIo;
finalOperations.add(op);
if (targetIORemaining <= 0) {
- return finalOperations;
+ missingPartitions.add(op.getPartitionPath());
Review Comment:
Can we add some doc around the return values of `#orderAndFilter`
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915918399
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Set<String> incrementalPartitions = getIncrementalPartitions(type);
+ if (!incrementalPartitions.isEmpty()) {
Review Comment:
changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915925940
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanPartitionFilter.java:
##########
@@ -42,13 +42,13 @@
*/
public class ClusteringPlanPartitionFilter {
- public static List<String> filter(List<String> partitions, HoodieWriteConfig config) {
+ public static List<String> filter(List<String> partitions, HoodieWriteConfig config, ArrayList<String> missingPartitions) {
Review Comment:
make sense.
changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780] Incremental Table Service [hudi]
Posted by "codope (via GitHub)" <gi...@apache.org>.
codope commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2612935968
@danny0405 @zhangyue19921010 `TestHoodieCompactionStrategy.testBoundedIOSimple` is flaky after this PR was merged - HUDI-8910. Could you please take a look?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1919720469
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -107,6 +107,7 @@ public static HoodieWriteConfig createMetadataWriteConfig(
// Create the write config for the metadata table by borrowing options from the main write config.
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
+ .withIncrementalTableServiceEnable(false)
Review Comment:
Or we could deprecated right boundary which means [left-boundary, +∞)
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1923034488
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -3341,6 +3352,11 @@ public Builder withWriteRecordPositionsEnabled(boolean shouldWriteRecordPosition
return this;
}
+ public Builder withIncrementalTableServiceEnable(boolean incrementalTableServiceEnable) {
Review Comment:
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
Re: [PR] [HUDI-8780] Incremental Table Service [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608760502
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"triggerType" : "PUSH"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 1a7290f66c485eaf59964b30d25ccfaf02003f82 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032) Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031)
<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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1919703175
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -107,6 +107,7 @@ public static HoodieWriteConfig createMetadataWriteConfig(
// Create the write config for the metadata table by borrowing options from the main write config.
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
+ .withIncrementalTableServiceEnable(false)
Review Comment:
Hi @danny0405 Maybe we need to disable incremental compaction for MDT. Since MDT compaction instant is not based on processing time but based on `https://github.com/apache/hudi/blob/dee6274e56d8b7d3852104d3e45377f4909466cd/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java#L1576`
Which could cause selecting instants based on window is inaccurate, thus missing compaction
what do u think?
This is a screenshot of the debug process during the specific test.
<img width="1339" alt="截屏2025-01-17 15 46 49" src="https://github.com/user-attachments/assets/e9ae1edd-5d26-4df8-b6a0-7fc8c62e6d28" />
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1921747178
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -116,7 +116,7 @@ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableSe
String rightBoundary = instantTime;
// compute [leftBoundary, rightBoundary) as time window
HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
- Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().findInstantsAfter(leftBoundary).filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
Review Comment:
@danny0405 Nice catch Danny.
At first I just wanted to filter out the left boundary instant itself, but obviously I called the API `findInstantsAfter `by mistake. Here we should get and filter the entire active time line complete instants
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1923034345
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -2179,6 +2186,10 @@ public String getPartialUpdateSchema() {
return getString(WRITE_PARTIAL_UPDATE_SCHEMA);
}
+ public boolean isIncrementalTableServiceEnable() {
Review Comment:
enabled
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -2179,6 +2186,10 @@ public String getPartialUpdateSchema() {
return getString(WRITE_PARTIAL_UPDATE_SCHEMA);
}
+ public boolean isIncrementalTableServiceEnable() {
+ return getBoolean(INCREMENTAL_TABLE_SERVICE_ENABLE);
Review Comment:
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
Re: [PR] [HUDI-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915924556
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Set<String> incrementalPartitions = getIncrementalPartitions(type);
+ if (!incrementalPartitions.isEmpty()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions);
+ return new ArrayList<>(incrementalPartitions);
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // fall back to get all partitions
+ LOG.info("Start to fetch all partitions for " + type);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ private Set<String> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ if (!missingPair.getLeft().isPresent()) {
+ // Last complete table service commit maybe archived.
+ return Collections.emptySet();
Review Comment:
make senses. changed
For now, there are two situations fall back to get all partitions
incremental partitions getting throws exceptions.
last completed table service instant is archived.
AND
when incremental partition list is empty , we will skip current schedule.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/IncrementalPartitionAwareStrategy.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.List;
+
+/**
+ * Marking strategy interface.
Review Comment:
changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912722714
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanPartitionFilter.java:
##########
@@ -42,13 +42,13 @@
*/
public class ClusteringPlanPartitionFilter {
- public static List<String> filter(List<String> partitions, HoodieWriteConfig config) {
+ public static List<String> filter(List<String> partitions, HoodieWriteConfig config, ArrayList<String> missingPartitions) {
Review Comment:
Maybe we return a Pair for both missing partitions and filtered partitions.
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912724772
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/BaseConsistentHashingBucketClusteringPlanStrategy.java:
##########
@@ -128,7 +129,7 @@ protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String
.build();
}).collect(Collectors.toList()));
}
- return ret.stream();
+ return Pair.of(ret.stream(), true);
Review Comment:
In which case the flag is false.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1912703717
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Set<String> incrementalPartitions = getIncrementalPartitions(type);
+ if (!incrementalPartitions.isEmpty()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions);
+ return new ArrayList<>(incrementalPartitions);
+ }
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // fall back to get all partitions
+ LOG.info("Start to fetch all partitions for " + type);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ private Set<String> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ if (!missingPair.getLeft().isPresent()) {
+ // Last complete table service commit maybe archived.
+ return Collections.emptySet();
Review Comment:
How do we handle the case the writer just commits the empty commits continuously? I think we still should trigger incremental partiton fetching. If the incremental partition list is empty we can just skip the scheduling?
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1919673168
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/TestBaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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;
+
+import org.apache.hudi.HoodieTestCommitGenerator;
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieClusteringStrategy;
+import org.apache.hudi.avro.model.HoodieRequestedReplaceMetadata;
+import org.apache.hudi.avro.model.HoodieSliceInfo;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.testutils.FileCreateUtils;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.action.BaseTableServicePlanActionExecutor;
+import org.apache.hudi.table.action.IncrementalPartitionAwareStrategy;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestBaseTableServicePlanActionExecutor {
+
+ @TempDir
+ File tempFile;
+ private static final StorageConfiguration<?> CONF = getDefaultStorageConf();
+ private final HoodieEngineContext context = new HoodieLocalEngineContext(CONF);
+
+ /**
+ * Clustering commit : cl
+ * Commit : c
+ * |requestTime<---commit type--->completionTime|(written partitions in current commit)
+ * ---------------------------------------------------------------------------------------------------------------------------------------------> timeline
+ * |0<-c1->1|(part=0000)
+ * |2<------cl2------>4|
+ * |3<-----------c3--------->6|(part=0003)
+ * |5<-----------c4--------->7|(part=0005)
+ * |8<-------------------c5----------------->10|(part=0008)
+ * | start incr clustering at 9
+ */
+ @Test
+ public void testExecutorWithMultiWriter() throws Exception {
+ String tableName = "testTable";
+ String tablePath = tempFile.getAbsolutePath() + StoragePath.SEPARATOR + tableName;
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(
+ HoodieTestUtils.getDefaultStorageConf(), tablePath, HoodieTableType.COPY_ON_WRITE, tableName);
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder().withPath(tablePath)
+ .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ .withMarkersType("DIRECT")
+ .build();
+
+ // <requestTime, CompletionTime>
+ HashMap<String, String> instants = new HashMap<>();
+ instants.put("0000", "0001"); // commit 1
+ instants.put("0002", "0004"); // clustering 1
+ instants.put("0003", "0006"); // commit 2
+ instants.put("0005", "0007"); // commit 3
+ instants.put("0008", "0010"); // commit 4
+
+ prepareTimeline(metaClient, instants);
+
+ DummyTableServicePlanActionExecutor executor = new DummyTableServicePlanActionExecutor(context, writeConfig, getMockHoodieTable(metaClient), "0009");
+ Set<String> incrementalPartitions = (Set<String>)executor.getIncrementalPartitions(TableServiceType.CLUSTER).getRight();
+
+ assertEquals(incrementalPartitions.size(), 2);
+ assertTrue(incrementalPartitions.contains("0003"));
+ assertTrue(incrementalPartitions.contains("0005"));
+ assertFalse(incrementalPartitions.contains("0008"));
+ }
+
+ @Test
+ public void testGetPartitionsFallbackToFullScan() throws Exception {
Review Comment:
@danny0405 This UT is test fall back to full scan when table service instant is archived.
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1917803271
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/BoundedIOCompactionStrategy.java:
##########
@@ -46,7 +47,7 @@ public List<HoodieCompactionOperation> orderAndFilter(HoodieWriteConfig writeCon
targetIORemaining -= opIo;
finalOperations.add(op);
if (targetIORemaining <= 0) {
- return finalOperations;
+ missingPartitions.add(op.getPartitionPath());
Review Comment:
added
--
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-8780][RFC-83] Incremental Table Service [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on code in PR #12601:
URL: https://github.com/apache/hudi/pull/12601#discussion_r1915863817
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -808,6 +808,13 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Avro schema of the partial updates. This is automatically set by the "
+ "Hudi write client and user is not expected to manually change the value.");
+ public static final ConfigProperty<Boolean> INCREMENTAL_TABLE_SERVICE_ENABLE = ConfigProperty
+ .key("hoodie.incremental.tableservice.enabled")
Review Comment:
tableservice -> table.service to keep consistent with other table service conf?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseTableServicePlanActionExecutor.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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;
+
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteOperationType;
+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.TimelineUtils;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.table.HoodieTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+public abstract class BaseTableServicePlanActionExecutor<T, I, K, O, R> extends BaseActionExecutor<T, I, K, O, R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseTableServicePlanActionExecutor.class);
+
+ public BaseTableServicePlanActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
+ HoodieTable<T, I, K, O> table, String instantTime) {
+ super(context, config, table, instantTime);
+ }
+
+ /**
+ * Get partitions, if strategy implement `IncrementalPartitionAwareStrategy` then return incremental partitions,
+ * otherwise return all partitions of the table
+ * @param strategy
+ * @return
+ */
+ public List<String> getPartitions(Object strategy, TableServiceType type) {
+ if (config.isIncrementalTableServiceEnable() && strategy instanceof IncrementalPartitionAwareStrategy) {
+ try {
+ // get incremental partitions.
+ LOG.info("Start to fetch incremental partitions for " + type);
+ Pair<Option<HoodieInstant>, Set<String>> lastInstantAndIncrPartitions = getIncrementalPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = lastInstantAndIncrPartitions.getLeft();
+ Set<String> incrementalPartitions = lastInstantAndIncrPartitions.getRight();
+
+ if (lastCompleteTableServiceInstant.isPresent()) {
+ if (!incrementalPartitions.isEmpty()) {
+ LOG.info("Fetched incremental partitions for " + type + ". " + incrementalPartitions + ". Instant " + instantTime);
+ return new ArrayList<>(incrementalPartitions);
+ } else {
+ // handle the case the writer just commits the empty commits continuously
+ // the incremental partition list is empty we just skip the scheduling
+ LOG.info("Incremental partitions are empty. Skip current schedule " + instantTime);
+ return Collections.emptyList();
+ }
+ }
+ // Last complete table service commit maybe archived.
+ // fall back to get all partitions.
+ LOG.info("No previous completed table service instant, fall back to get all partitions");
+ } catch (Exception ex) {
+ LOG.warn("Failed to get incremental partitions", ex);
+ }
+ }
+
+ // get all partitions
+ LOG.info("Start to fetch all partitions for " + type + ". Instant " + instantTime);
+ return FSUtils.getAllPartitionPaths(context, table.getMetaClient().getStorage(),
+ config.getMetadataConfig(), table.getMetaClient().getBasePath());
+ }
+
+ public Pair<Option<HoodieInstant>, Set<String>> getIncrementalPartitions(TableServiceType type) {
+ Pair<Option<HoodieInstant>, List<String>> missingPair = fetchMissingPartitions(type);
+ Option<HoodieInstant> lastCompleteTableServiceInstant = missingPair.getLeft();
+ List<String> missingPartitions = missingPair.getRight();
+
+ String leftBoundary = lastCompleteTableServiceInstant.isPresent()
+ ? missingPair.getLeft().get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS;
+ String rightBoundary = instantTime;
+ // compute [leftBoundary, rightBoundary) as time window
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ Set<String> partitionsInCommitMeta = table.getActiveTimeline().filterCompletedInstants().getCommitsTimeline().getInstantsAsStream()
+ .filter(this::filterCommitByTableType).flatMap(instant -> {
+ try {
+ String completionTime = instant.getCompletionTime();
+ if (completionTime.compareTo(leftBoundary) >= 0 && completionTime.compareTo(rightBoundary) < 0) {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, activeTimeline);
+ // ignore all the clustering operation for both mor and cow table
+ if (!metadata.getOperationType().equals(WriteOperationType.CLUSTER)) {
+ return metadata.getWriteStats().stream().map(HoodieWriteStat::getPartitionPath);
+ }
+ }
+ return Stream.empty();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to get commit meta " + instant, e);
+ }
+ }).collect(Collectors.toSet());
+
+ partitionsInCommitMeta.addAll(missingPartitions);
+ return Pair.of(lastCompleteTableServiceInstant, partitionsInCommitMeta);
+ }
+
+ private boolean filterCommitByTableType(HoodieInstant instant) {
+ // for completed replace commit, need CommitMetadata to distinguish clustering or insert overwrite
+ switch (table.getMetaClient().getTableType()) {
+ case MERGE_ON_READ: {
+ // for mor only take cares of delta commit and replace commit
+ Set<String> operations = CollectionUtils.createSet(DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION);
Review Comment:
better not create constant set per instant
--
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-8780][RFC-83][WIP] Incremental Table Service [hudi]
Posted by "TheR1sing3un (via GitHub)" <gi...@apache.org>.
TheR1sing3un commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2579158414
@zhangyue19921010 Hi, judging from the rfc content, the goal this time is to use incremental `compaction` and `clustering`. Should we consider incorporating the incremental processing logic for `clean` as well? By the way, whether we can solve this problem: https://github.com/apache/hudi/issues/11647 , it seems that we can solve it at the time of implementation.
--
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-8780] Incremental Table Service [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12601:
URL: https://github.com/apache/hudi/pull/12601#issuecomment-2608920746
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"triggerType" : "PUSH"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3032",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "1a7290f66c485eaf59964b30d25ccfaf02003f82",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3031",
"triggerID" : "2606095692",
"triggerType" : "MANUAL"
}, {
"hash" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3035",
"triggerID" : "02fe10bf4122514d17cabcafe33eed70e5d2df1b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 02fe10bf4122514d17cabcafe33eed70e5d2df1b Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=3035)
<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