You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by sa...@apache.org on 2022/12/13 08:04:36 UTC
[hudi] 01/22: [HUDI-5241] Optimize HoodieDefaultTimeline API (#7241)
This is an automated email from the ASF dual-hosted git repository.
satish pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e6520bbb365ee0c04dbee2ba872c949cec9af0e7
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Mon Nov 28 17:41:20 2022 +0800
[HUDI-5241] Optimize HoodieDefaultTimeline API (#7241)
---
.../apache/hudi/cli/commands/CommitsCommand.java | 12 ++--
.../hudi/cli/commands/CompactionCommand.java | 2 +-
.../org/apache/hudi/cli/commands/DiffCommand.java | 2 +-
.../apache/hudi/cli/commands/ExportCommand.java | 2 +-
.../hudi/cli/commands/FileSystemViewCommand.java | 2 +-
.../cli/commands/HoodieSyncValidateCommand.java | 2 +-
.../apache/hudi/cli/commands/RepairsCommand.java | 2 +-
.../hudi/cli/commands/SavepointsCommand.java | 2 +-
.../org/apache/hudi/cli/commands/StatsCommand.java | 3 +-
.../apache/hudi/cli/commands/TimelineCommand.java | 5 +-
.../hudi/cli/commands/TestArchiveCommand.java | 4 +-
.../hudi/cli/commands/TestCleansCommand.java | 8 +--
.../hudi/cli/commands/TestCommitsCommand.java | 2 +-
.../hudi/cli/commands/TestRepairsCommand.java | 6 +-
.../hudi/cli/commands/TestRollbacksCommand.java | 4 +-
.../hudi/cli/integ/ITTestClusteringCommand.java | 6 +-
.../hudi/cli/integ/ITTestCompactionCommand.java | 4 +-
.../apache/hudi/client/BaseHoodieWriteClient.java | 8 +--
.../apache/hudi/client/HoodieTimelineArchiver.java | 6 +-
...urrentFileWritesConflictResolutionStrategy.java | 4 +-
.../apache/hudi/client/utils/TransactionUtils.java | 4 +-
.../metadata/HoodieBackedTableMetadataWriter.java | 6 +-
.../java/org/apache/hudi/table/HoodieTable.java | 2 +-
.../table/action/clean/CleanActionExecutor.java | 2 +-
.../hudi/table/action/clean/CleanPlanner.java | 4 +-
.../compact/ScheduleCompactionActionExecutor.java | 2 +-
.../table/action/index/RunIndexActionExecutor.java | 8 +--
.../rollback/BaseRollbackActionExecutor.java | 4 +-
.../org/apache/hudi/table/repair/RepairUtils.java | 4 +-
.../table/upgrade/ZeroToOneUpgradeHandler.java | 2 +-
.../org/apache/hudi/client/TestClientRollback.java | 12 ++--
.../hudi/client/TestHoodieClientMultiWriter.java | 4 +-
.../TestHoodieClientOnCopyOnWriteStorage.java | 2 +-
.../apache/hudi/io/TestHoodieTimelineArchiver.java | 15 ++--
.../clean/TestCleanerInsertAndCleanByCommits.java | 4 +-
.../clean/TestCleanerInsertAndCleanByVersions.java | 2 +-
.../table/action/compact/CompactionTestBase.java | 4 +-
.../table/action/compact/TestAsyncCompaction.java | 4 +-
.../table/action/compact/TestInlineCompaction.java | 2 +-
.../hudi/testutils/HoodieClientTestUtils.java | 4 +-
.../hudi/common/table/HoodieTableMetaClient.java | 2 +-
.../table/timeline/HoodieActiveTimeline.java | 4 +-
.../table/timeline/HoodieArchivedTimeline.java | 4 +-
.../table/timeline/HoodieDefaultTimeline.java | 83 +++++++++++-----------
.../hudi/common/table/timeline/HoodieTimeline.java | 28 ++++++--
.../common/table/timeline/TimelineDiffHelper.java | 11 +--
.../hudi/common/table/timeline/TimelineUtils.java | 4 +-
.../common/table/timeline/dto/TimelineDTO.java | 2 +-
.../table/view/AbstractTableFileSystemView.java | 4 +-
.../apache/hudi/common/util/ClusteringUtils.java | 4 +-
.../apache/hudi/common/util/CompactionUtils.java | 7 +-
.../hudi/common/util/InternalSchemaCache.java | 2 +-
.../io/FileBasedInternalSchemaStorageManager.java | 2 +-
.../hudi/metadata/HoodieBackedTableMetadata.java | 4 +-
.../common/table/TestHoodieTableMetaClient.java | 4 +-
.../table/timeline/TestHoodieActiveTimeline.java | 16 ++---
.../table/view/TestIncrementalFSViewSync.java | 2 +-
.../hudi/common/util/TestCompactionUtils.java | 6 +-
.../compact/strategy/CompactionPlanStrategies.java | 7 +-
.../sink/partitioner/profile/WriteProfile.java | 2 +-
.../apache/hudi/source/IncrementalInputSplits.java | 6 +-
.../sink/TestStreamWriteOperatorCoordinator.java | 16 ++---
.../apache/hudi/table/ITTestHoodieDataSource.java | 2 +-
.../apache/hudi/table/format/TestInputFormat.java | 6 +-
.../org/apache/hudi/utils/TestCompactionUtil.java | 12 ++--
.../HoodieMergeOnReadTableInputFormat.java | 2 +-
.../hudi/hadoop/utils/HoodieInputFormatUtils.java | 3 +-
.../dag/nodes/ValidateAsyncOperations.java | 2 +-
.../testsuite/job/TestHoodieTestSuiteJob.java | 10 +--
.../main/scala/org/apache/hudi/DefaultSource.scala | 2 +-
.../org/apache/hudi/IncrementalRelation.scala | 6 +-
.../hudi/MergeOnReadIncrementalRelation.scala | 4 +-
.../org/apache/hudi/HoodieDataSourceHelpers.java | 2 +-
.../procedures/CommitsCompareProcedure.scala | 4 +-
.../procedures/RunCompactionProcedure.scala | 2 +-
.../procedures/ValidateHoodieSyncProcedure.scala | 2 +-
.../src/test/java/HoodieJavaStreamingApp.java | 9 ++-
.../apache/hudi/functional/TestCOWDataSource.scala | 2 +-
.../TestIncrementalReadWithFullTableScan.scala | 2 +-
.../sql/hudi/procedure/TestRepairsProcedure.scala | 4 +-
.../org/apache/hudi/hive/HoodieHiveSyncClient.java | 2 +-
.../org/apache/hudi/hive/TestHiveSyncTool.java | 4 +-
.../hudi/timeline/service/RequestHandler.java | 7 +-
.../hudi/utilities/HiveIncrementalPuller.java | 4 +-
.../hudi/utilities/HoodieDataTableValidator.java | 2 +-
.../utilities/HoodieMetadataTableValidator.java | 2 +-
.../hudi/utilities/deltastreamer/DeltaSync.java | 4 +-
.../sources/helpers/IncrSourceHelper.java | 2 +-
.../functional/TestHoodieDeltaStreamer.java | 46 ++++++------
.../TestHoodieDeltaStreamerWithMultiWriter.java | 3 +-
90 files changed, 283 insertions(+), 266 deletions(-)
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CommitsCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CommitsCommand.java
index 75f7b9a8b86..60f4de5e7ce 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CommitsCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CommitsCommand.java
@@ -65,7 +65,7 @@ public class CommitsCommand {
final List<Comparable[]> rows = new ArrayList<>();
final List<HoodieInstant> commits = timeline.getCommitsTimeline().filterCompletedInstants()
- .getInstants().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
+ .getInstantsAsStream().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
for (final HoodieInstant commit : commits) {
if (timeline.getInstantDetails(commit).isPresent()) {
@@ -103,7 +103,7 @@ public class CommitsCommand {
final List<Comparable[]> rows = new ArrayList<>();
final List<HoodieInstant> commits = timeline.getCommitsTimeline().filterCompletedInstants()
- .getInstants().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
+ .getInstantsAsStream().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
for (final HoodieInstant commit : commits) {
if (timeline.getInstantDetails(commit).isPresent()) {
@@ -372,20 +372,20 @@ public class CommitsCommand {
HoodieTimeline targetTimeline = target.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
HoodieTimeline sourceTimeline = source.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
String targetLatestCommit =
- targetTimeline.getInstants().iterator().hasNext() ? targetTimeline.lastInstant().get().getTimestamp() : "0";
+ targetTimeline.getInstantsAsStream().iterator().hasNext() ? targetTimeline.lastInstant().get().getTimestamp() : "0";
String sourceLatestCommit =
- sourceTimeline.getInstants().iterator().hasNext() ? sourceTimeline.lastInstant().get().getTimestamp() : "0";
+ sourceTimeline.getInstantsAsStream().iterator().hasNext() ? sourceTimeline.lastInstant().get().getTimestamp() : "0";
if (sourceLatestCommit != null
&& HoodieTimeline.compareTimestamps(targetLatestCommit, HoodieTimeline.GREATER_THAN, sourceLatestCommit)) {
// source is behind the target
List<String> commitsToCatchup = targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE)
- .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
return "Source " + source.getTableConfig().getTableName() + " is behind by " + commitsToCatchup.size()
+ " commits. Commits to catch up - " + commitsToCatchup;
} else {
List<String> commitsToCatchup = sourceTimeline.findInstantsAfter(targetLatestCommit, Integer.MAX_VALUE)
- .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
return "Source " + source.getTableConfig().getTableName() + " is ahead by " + commitsToCatchup.size()
+ " commits. Commits to catch up - " + commitsToCatchup;
}
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CompactionCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CompactionCommand.java
index cb24f56236c..b7e550a9aae 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CompactionCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CompactionCommand.java
@@ -317,7 +317,7 @@ public class CompactionCommand {
.collect(Collectors.toList());
Set<String> committedInstants = timeline.getCommitTimeline().filterCompletedInstants()
- .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
List<Comparable[]> rows = new ArrayList<>();
for (Pair<HoodieInstant, HoodieCompactionPlan> compactionPlan : compactionPlans) {
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/DiffCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/DiffCommand.java
index 07d21fe0226..01e6da421a6 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/DiffCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/DiffCommand.java
@@ -137,7 +137,7 @@ public class DiffCommand {
BiFunction<HoodieWriteStat, String, Boolean> diffEntityChecker) throws IOException {
List<Comparable[]> rows = new ArrayList<>();
List<HoodieInstant> commits = timeline.getCommitsTimeline().filterCompletedInstants()
- .getInstants().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
+ .getInstantsAsStream().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
for (final HoodieInstant commit : commits) {
Option<byte[]> instantDetails = timeline.getInstantDetails(commit);
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ExportCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ExportCommand.java
index 2406eddacf3..333eb11b0c0 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ExportCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ExportCommand.java
@@ -87,7 +87,7 @@ public class ExportCommand {
// The non archived instants can be listed from the Timeline.
HoodieTimeline timeline = HoodieCLI.getTableMetaClient().getActiveTimeline().filterCompletedInstants()
.filter(i -> actionSet.contains(i.getAction()));
- List<HoodieInstant> nonArchivedInstants = timeline.getInstants().collect(Collectors.toList());
+ List<HoodieInstant> nonArchivedInstants = timeline.getInstants();
// Archived instants are in the commit archive files
FileStatus[] statuses = FSUtils.getFs(basePath, HoodieCLI.conf).globStatus(archivePath);
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/FileSystemViewCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/FileSystemViewCommand.java
index 78e7d90195d..08c892dde4b 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/FileSystemViewCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/FileSystemViewCommand.java
@@ -256,7 +256,7 @@ public class FileSystemViewCommand {
timeline = timeline.filterCompletedInstants();
}
- instantsStream = timeline.getInstants();
+ instantsStream = timeline.getInstantsAsStream();
if (!maxInstant.isEmpty()) {
final BiPredicate<String, String> predicate;
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieSyncValidateCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieSyncValidateCommand.java
index 0fc26a55b89..f24e1655971 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieSyncValidateCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieSyncValidateCommand.java
@@ -90,7 +90,7 @@ public class HoodieSyncValidateCommand {
private String getString(HoodieTableMetaClient target, HoodieTimeline targetTimeline, HoodieTableMetaClient source, long sourceCount, long targetCount, String sourceLatestCommit)
throws IOException {
List<HoodieInstant> commitsToCatchup = targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE)
- .getInstants().collect(Collectors.toList());
+ .getInstants();
if (commitsToCatchup.isEmpty()) {
return "Count difference now is (count(" + target.getTableConfig().getTableName() + ") - count("
+ source.getTableConfig().getTableName() + ") == " + (targetCount - sourceCount);
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java
index 644fe610045..4b817473f05 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java
@@ -211,7 +211,7 @@ public class RepairsCommand {
public void showFailedCommits() {
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
- activeTimeline.filterCompletedInstants().getInstants().filter(activeTimeline::isEmpty).forEach(hoodieInstant -> LOG.warn("Empty Commit: " + hoodieInstant.toString()));
+ activeTimeline.filterCompletedInstants().getInstantsAsStream().filter(activeTimeline::isEmpty).forEach(hoodieInstant -> LOG.warn("Empty Commit: " + hoodieInstant.toString()));
}
@ShellMethod(key = "repair migrate-partition-meta", value = "Migrate all partition meta file currently stored in text format "
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SavepointsCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SavepointsCommand.java
index 73f94acda87..2d5bae423e2 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SavepointsCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SavepointsCommand.java
@@ -106,7 +106,7 @@ public class SavepointsCommand {
}
HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
HoodieTimeline timeline = activeTimeline.getCommitsTimeline().filterCompletedInstants();
- List<HoodieInstant> instants = timeline.getInstants().filter(instant -> instant.getTimestamp().equals(instantTime)).collect(Collectors.toList());
+ List<HoodieInstant> instants = timeline.getInstantsAsStream().filter(instant -> instant.getTimestamp().equals(instantTime)).collect(Collectors.toList());
if (instants.isEmpty()) {
return "Commit " + instantTime + " not found in Commits " + timeline;
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/StatsCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/StatsCommand.java
index c9034d03d5f..2c8ab342f31 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/StatsCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/StatsCommand.java
@@ -45,7 +45,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
-import java.util.stream.Collectors;
/**
* CLI command to displays stats options.
@@ -73,7 +72,7 @@ public class StatsCommand {
List<Comparable[]> rows = new ArrayList<>();
DecimalFormat df = new DecimalFormat("#.00");
- for (HoodieInstant instantTime : timeline.getInstants().collect(Collectors.toList())) {
+ for (HoodieInstant instantTime : timeline.getInstants()) {
String waf = "0";
HoodieCommitMetadata commit = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(instantTime).get(),
HoodieCommitMetadata.class);
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/TimelineCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/TimelineCommand.java
index bf7e5397cab..eab61e5dfaa 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/TimelineCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/TimelineCommand.java
@@ -208,7 +208,7 @@ public class TimelineCommand {
Integer limit, String sortByField, boolean descending, boolean headerOnly, boolean withRowNo,
boolean showTimeSeconds, boolean showRollbackInfo) {
Map<String, List<String>> rollbackInfo = getRolledBackInstantInfo(timeline);
- final List<Comparable[]> rows = timeline.getInstants().map(instant -> {
+ final List<Comparable[]> rows = timeline.getInstantsAsStream().map(instant -> {
int numColumns = showRollbackInfo ? 7 : 6;
Comparable[] row = new Comparable[numColumns];
String instantTimestamp = instant.getTimestamp();
@@ -343,8 +343,7 @@ public class TimelineCommand {
// Instant rolled back or to roll back -> rollback instants
Map<String, List<String>> rollbackInfoMap = new HashMap<>();
List<HoodieInstant> rollbackInstants = timeline.filter(instant ->
- HoodieTimeline.ROLLBACK_ACTION.equalsIgnoreCase(instant.getAction()))
- .getInstants().collect(Collectors.toList());
+ HoodieTimeline.ROLLBACK_ACTION.equalsIgnoreCase(instant.getAction())).getInstants();
rollbackInstants.forEach(rollbackInstant -> {
try {
if (rollbackInstant.isInflight()) {
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchiveCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchiveCommand.java
index 1a10d41c9ac..16e203730c1 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchiveCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchiveCommand.java
@@ -66,13 +66,13 @@ public class TestArchiveCommand extends CLIFunctionalTestHarness {
//get instants in the active timeline only returns the latest state of the commit
//therefore we expect 2 instants because minCommits is 2
- assertEquals(2, metaClient.getActiveTimeline().getInstants().count());
+ assertEquals(2, metaClient.getActiveTimeline().countInstants());
//get instants in the archived timeline returns all instants in the commit
//therefore we expect 12 instants because 6 commits - 2 commits in active timeline = 4 in archived
//since each commit is completed, there are 3 instances per commit (requested, inflight, completed)
//and 3 instances per commit * 4 commits = 12 instances
- assertEquals(12, metaClient.getArchivedTimeline().getInstants().count());
+ assertEquals(12, metaClient.getArchivedTimeline().countInstants());
}
}
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCleansCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCleansCommand.java
index f0ed1787e21..1b45fdd4d07 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCleansCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCleansCommand.java
@@ -127,13 +127,13 @@ public class TestCleansCommand extends CLIFunctionalTestHarness {
// First, run clean
SparkMain.clean(jsc(), HoodieCLI.basePath, propsFilePath.getPath(), new ArrayList<>());
- assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().count(),
+ assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().countInstants(),
"Loaded 1 clean and the count should match");
Object result = shell.evaluate(() -> "cleans show");
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
- HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().findFirst().orElse(null);
+ HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstantsAsStream().findFirst().orElse(null);
assertNotNull(clean);
TableHeader header =
@@ -163,10 +163,10 @@ public class TestCleansCommand extends CLIFunctionalTestHarness {
// First, run clean with two partition
SparkMain.clean(jsc(), HoodieCLI.basePath, propsFilePath.toString(), new ArrayList<>());
- assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().count(),
+ assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().countInstants(),
"Loaded 1 clean and the count should match");
- HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().findFirst().get();
+ HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstantsAsStream().findFirst().get();
Object result = shell.evaluate(() -> "clean showpartitions --clean " + clean.getTimestamp());
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
index 7e504488a2d..bfce9b3f0a6 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
@@ -521,7 +521,7 @@ public class TestCommitsCommand extends CLIFunctionalTestHarness {
// the latest instant of test_table2 is 101
List<String> commitsToCatchup = metaClient.getActiveTimeline().findInstantsAfter("101", Integer.MAX_VALUE)
- .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
String expected = String.format("Source %s is ahead by %d commits. Commits to catch up - %s",
tableName1, commitsToCatchup.size(), commitsToCatchup);
assertEquals(expected, result.toString());
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java
index c7492bba2fe..b10fab891e2 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java
@@ -255,14 +255,14 @@ public class TestRepairsCommand extends CLIFunctionalTestHarness {
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
// first, there are four instants
- assertEquals(4, metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants().count());
+ assertEquals(4, metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
Object result = shell.evaluate(() -> "repair corrupted clean files");
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
- assertEquals(0, metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants().count());
+ assertEquals(0, metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
}
/**
@@ -283,7 +283,7 @@ public class TestRepairsCommand extends CLIFunctionalTestHarness {
HoodieTestCommitMetadataGenerator.createCommitFile(tablePath, timestamp, conf);
}
- metaClient.getActiveTimeline().getInstants().filter(hoodieInstant -> Integer.parseInt(hoodieInstant.getTimestamp()) % 4 == 0).forEach(hoodieInstant -> {
+ metaClient.getActiveTimeline().getInstantsAsStream().filter(hoodieInstant -> Integer.parseInt(hoodieInstant.getTimestamp()) % 4 == 0).forEach(hoodieInstant -> {
metaClient.getActiveTimeline().deleteInstantFileIfExists(hoodieInstant);
metaClient.getActiveTimeline().createNewInstant(hoodieInstant);
});
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRollbacksCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRollbacksCommand.java
index a4144937621..e8dee58ec43 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRollbacksCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRollbacksCommand.java
@@ -128,7 +128,7 @@ public class TestRollbacksCommand extends CLIFunctionalTestHarness {
// get rollback instants
HoodieActiveTimeline activeTimeline = new RollbacksCommand.RollbackTimeline(HoodieCLI.getTableMetaClient());
- Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstants();
+ Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstantsAsStream();
List<Comparable[]> rows = new ArrayList<>();
rollback.sorted().forEach(instant -> {
@@ -169,7 +169,7 @@ public class TestRollbacksCommand extends CLIFunctionalTestHarness {
public void testShowRollback() throws IOException {
// get instant
HoodieActiveTimeline activeTimeline = new RollbacksCommand.RollbackTimeline(HoodieCLI.getTableMetaClient());
- Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstants();
+ Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstantsAsStream();
HoodieInstant instant = rollback.findFirst().orElse(null);
assertNotNull(instant, "The instant can not be null.");
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
index f81133aca00..22dbbe1b34b 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
@@ -128,12 +128,12 @@ public class ITTestClusteringCommand extends HoodieCLIIntegrationTestBase {
// assert clustering complete
assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
- .filterCompletedInstants().getInstants()
+ .filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList()).contains(instance),
"Pending clustering must be completed");
assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
- .getCompletedReplaceTimeline().getInstants()
+ .getCompletedReplaceTimeline().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList()).contains(instance),
"Pending clustering must be completed");
}
@@ -156,7 +156,7 @@ public class ITTestClusteringCommand extends HoodieCLIIntegrationTestBase {
// assert clustering complete
assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
- .getCompletedReplaceTimeline().getInstants()
+ .getCompletedReplaceTimeline().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).count() > 0,
"Completed clustering couldn't be 0");
}
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java
index b8294f9e334..4e03efe4aae 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java
@@ -136,7 +136,7 @@ public class ITTestCompactionCommand extends HoodieCLIIntegrationTestBase {
// assert compaction complete
assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
- .filterCompletedInstants().getInstants()
+ .filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList()).contains(instance),
"Pending compaction must be completed");
}
@@ -164,7 +164,7 @@ public class ITTestCompactionCommand extends HoodieCLIIntegrationTestBase {
// assert compaction complete
assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
- .filterCompletedInstants().getInstants()
+ .filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).count() > 0,
"Completed compaction couldn't be 0");
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index f9017534b39..50337d56d91 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -557,7 +557,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
// if just inline schedule is enabled
if (!config.inlineCompactionEnabled() && config.scheduleInlineCompaction()
- && !table.getActiveTimeline().getWriteTimeline().filterPendingCompactionTimeline().getInstants().findAny().isPresent()) {
+ && table.getActiveTimeline().getWriteTimeline().filterPendingCompactionTimeline().empty()) {
// proceed only if there are no pending compactions
metadata.addMetadata(HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT.key(), "true");
inlineScheduleCompaction(extraMetadata);
@@ -574,7 +574,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
// if just inline schedule is enabled
if (!config.inlineClusteringEnabled() && config.scheduleInlineClustering()
- && !table.getActiveTimeline().filterPendingReplaceTimeline().getInstants().findAny().isPresent()) {
+ && table.getActiveTimeline().filterPendingReplaceTimeline().empty()) {
// proceed only if there are no pending clustering
metadata.addMetadata(HoodieClusteringConfig.SCHEDULE_INLINE_CLUSTERING.key(), "true");
inlineScheduleClustering(extraMetadata);
@@ -749,7 +749,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
final Timer.Context timerContext = this.metrics.getRollbackCtx();
try {
HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
- Option<HoodieInstant> commitInstantOpt = Option.fromJavaOptional(table.getActiveTimeline().getCommitsTimeline().getInstants()
+ Option<HoodieInstant> commitInstantOpt = Option.fromJavaOptional(table.getActiveTimeline().getCommitsTimeline().getInstantsAsStream()
.filter(instant -> HoodieActiveTimeline.EQUALS.test(instant.getTimestamp(), commitInstantTime))
.findFirst());
if (commitInstantOpt.isPresent() || pendingRollbackInfo.isPresent()) {
@@ -1114,7 +1114,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
* @return map of pending commits to be rolled-back instants to Rollback Instant and Rollback plan Pair.
*/
protected Map<String, Option<HoodiePendingRollbackInfo>> getPendingRollbackInfos(HoodieTableMetaClient metaClient, boolean ignoreCompactionAndClusteringInstants) {
- List<HoodieInstant> instants = metaClient.getActiveTimeline().filterPendingRollbackTimeline().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> instants = metaClient.getActiveTimeline().filterPendingRollbackTimeline().getInstants();
Map<String, Option<HoodiePendingRollbackInfo>> infoMap = new HashMap<>();
for (HoodieInstant rollbackInstant : instants) {
HoodieRollbackPlan rollbackPlan;
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
index 5b50195d6d8..bb814f817d0 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
@@ -385,7 +385,7 @@ public class HoodieTimelineArchiver<T extends HoodieAvroPayload, I, K, O> {
private Stream<HoodieInstant> getCleanInstantsToArchive() {
HoodieTimeline cleanAndRollbackTimeline = table.getActiveTimeline()
.getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.CLEAN_ACTION, HoodieTimeline.ROLLBACK_ACTION)).filterCompletedInstants();
- return cleanAndRollbackTimeline.getInstants()
+ return cleanAndRollbackTimeline.getInstantsAsStream()
.collect(Collectors.groupingBy(HoodieInstant::getAction)).values().stream()
.map(hoodieInstants -> {
if (hoodieInstants.size() > this.maxInstantsToKeep) {
@@ -430,7 +430,7 @@ public class HoodieTimelineArchiver<T extends HoodieAvroPayload, I, K, O> {
: Option.empty();
// Actually do the commits
- Stream<HoodieInstant> instantToArchiveStream = commitTimeline.getInstants()
+ Stream<HoodieInstant> instantToArchiveStream = commitTimeline.getInstantsAsStream()
.filter(s -> {
if (config.shouldArchiveBeyondSavepoint()) {
// skip savepoint commits and proceed further
@@ -473,7 +473,7 @@ public class HoodieTimelineArchiver<T extends HoodieAvroPayload, I, K, O> {
// For archiving and cleaning instants, we need to include intermediate state files if they exist
HoodieActiveTimeline rawActiveTimeline = new HoodieActiveTimeline(metaClient, false);
- Map<Pair<String, String>, List<HoodieInstant>> groupByTsAction = rawActiveTimeline.getInstants()
+ Map<Pair<String, String>, List<HoodieInstant>> groupByTsAction = rawActiveTimeline.getInstantsAsStream()
.collect(Collectors.groupingBy(i -> Pair.of(i.getTimestamp(),
HoodieInstant.getComparableAction(i.getAction()))));
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
index 938a40684a0..c096b2de134 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -59,13 +59,13 @@ public class SimpleConcurrentFileWritesConflictResolutionStrategy
.getCommitsTimeline()
.filterCompletedInstants()
.findInstantsAfter(lastSuccessfulInstant.isPresent() ? lastSuccessfulInstant.get().getTimestamp() : HoodieTimeline.INIT_INSTANT_TS)
- .getInstants();
+ .getInstantsAsStream();
Stream<HoodieInstant> compactionAndClusteringPendingTimeline = activeTimeline
.getTimelineOfActions(CollectionUtils.createSet(REPLACE_COMMIT_ACTION, COMPACTION_ACTION))
.findInstantsAfter(currentInstant.getTimestamp())
.filterInflightsAndRequested()
- .getInstants();
+ .getInstantsAsStream();
return Stream.concat(completedCommitsInstantStream, compactionAndClusteringPendingTimeline);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
index 01d37eaf32a..afc6feb6db0 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
@@ -131,7 +131,7 @@ public class TransactionUtils {
.getActiveTimeline()
.getTimelineOfActions(timelineActions)
.filterInflightsAndRequested()
- .getInstants()
+ .getInstantsAsStream()
.map(HoodieInstant::getTimestamp)
.collect(Collectors.toSet());
}
@@ -144,7 +144,7 @@ public class TransactionUtils {
.reloadActiveTimeline()
.getCommitsTimeline()
.filterCompletedInstants()
- .getInstants()
+ .getInstantsAsStream()
.filter(f -> pendingInstants.contains(f.getTimestamp()));
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
index d5c11713216..42e35cfa30f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
@@ -575,7 +575,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
// We can only initialize if there are no pending operations on the dataset
List<HoodieInstant> pendingDataInstant = dataMetaClient.getActiveTimeline()
- .getInstants().filter(i -> !i.isCompleted())
+ .getInstantsAsStream().filter(i -> !i.isCompleted())
.filter(i -> !inflightInstantTimestamp.isPresent() || !i.getTimestamp().equals(inflightInstantTimestamp.get()))
// regular writers should not be blocked due to pending indexing action
.filter(i -> !HoodieTimeline.INDEXING_ACTION.equals(i.getAction()))
@@ -759,7 +759,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
* if the partition path in the plan matches with the given partition path.
*/
private static void deletePendingIndexingInstant(HoodieTableMetaClient metaClient, String partitionPath) {
- metaClient.reloadActiveTimeline().filterPendingIndexTimeline().getInstants().filter(instant -> REQUESTED.equals(instant.getState()))
+ metaClient.reloadActiveTimeline().filterPendingIndexTimeline().getInstantsAsStream().filter(instant -> REQUESTED.equals(instant.getState()))
.forEach(instant -> {
try {
HoodieIndexPlan indexPlan = deserializeIndexPlan(metaClient.getActiveTimeline().readIndexPlanAsBytes(instant).get());
@@ -1014,7 +1014,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
String latestDeltaCommitTime = metadataMetaClient.reloadActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().lastInstant()
.get().getTimestamp();
List<HoodieInstant> pendingInstants = dataMetaClient.reloadActiveTimeline().filterInflightsAndRequested()
- .findInstantsBefore(instantTime).getInstants().collect(Collectors.toList());
+ .findInstantsBefore(instantTime).getInstants();
if (!pendingInstants.isEmpty()) {
LOG.info(String.format("Cannot compact metadata table as there are %d inflight instants before latest deltacommit %s: %s",
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
index 5ca3aee764a..b09d488e7d2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
@@ -371,7 +371,7 @@ public abstract class HoodieTable<T extends HoodieRecordPayload, I, K, O> implem
* Get the list of savepoint timestamps in this table.
*/
public Set<String> getSavepointTimestamps() {
- return getCompletedSavepointTimeline().getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
+ return getCompletedSavepointTimeline().getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
}
public HoodieActiveTimeline getActiveTimeline() {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
index 750e687947b..c0a4f773168 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
@@ -238,7 +238,7 @@ public class CleanActionExecutor<T extends HoodieRecordPayload, I, K, O> extends
List<HoodieCleanMetadata> cleanMetadataList = new ArrayList<>();
// If there are inflight(failed) or previously requested clean operation, first perform them
List<HoodieInstant> pendingCleanInstants = table.getCleanTimeline()
- .filterInflightsAndRequested().getInstants().collect(Collectors.toList());
+ .filterInflightsAndRequested().getInstants();
if (pendingCleanInstants.size() > 0) {
// try to clean old history schema.
try {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
index 64e69b1d2a9..9e11e79da02 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
@@ -179,7 +179,7 @@ public class CleanPlanner<T extends HoodieRecordPayload, I, K, O> implements Ser
LOG.info("Incremental Cleaning mode is enabled. Looking up partition-paths that have since changed "
+ "since last cleaned at " + cleanMetadata.getEarliestCommitToRetain()
+ ". New Instant to retain : " + newInstantToRetain);
- return hoodieTable.getCompletedCommitsTimeline().getInstants().filter(
+ return hoodieTable.getCompletedCommitsTimeline().getInstantsAsStream().filter(
instant -> HoodieTimeline.compareTimestamps(instant.getTimestamp(), HoodieTimeline.GREATER_THAN_OR_EQUALS,
cleanMetadata.getEarliestCommitToRetain()) && HoodieTimeline.compareTimestamps(instant.getTimestamp(),
HoodieTimeline.LESSER_THAN, newInstantToRetain.get().getTimestamp())).flatMap(instant -> {
@@ -470,7 +470,7 @@ public class CleanPlanner<T extends HoodieRecordPayload, I, K, O> implements Ser
Instant instant = Instant.now();
ZonedDateTime currentDateTime = ZonedDateTime.ofInstant(instant, ZoneId.systemDefault());
String earliestTimeToRetain = HoodieActiveTimeline.formatDate(Date.from(currentDateTime.minusHours(hoursRetained).toInstant()));
- earliestCommitToRetain = Option.fromJavaOptional(commitTimeline.getInstants().filter(i -> HoodieTimeline.compareTimestamps(i.getTimestamp(),
+ earliestCommitToRetain = Option.fromJavaOptional(commitTimeline.getInstantsAsStream().filter(i -> HoodieTimeline.compareTimestamps(i.getTimestamp(),
HoodieTimeline.GREATER_THAN_OR_EQUALS, earliestTimeToRetain)).findFirst());
}
return earliestCommitToRetain;
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
index 4fb5f9f7ddb..3fd6e91eeff 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
@@ -81,7 +81,7 @@ public class ScheduleCompactionActionExecutor<T extends HoodieRecordPayload, I,
}
// Committed and pending compaction instants should have strictly lower timestamps
List<HoodieInstant> conflictingInstants = table.getActiveTimeline()
- .getWriteTimeline().filterCompletedAndCompactionInstants().getInstants()
+ .getWriteTimeline().filterCompletedAndCompactionInstants().getInstantsAsStream()
.filter(instant -> HoodieTimeline.compareTimestamps(
instant.getTimestamp(), HoodieTimeline.GREATER_THAN_OR_EQUALS, instantTime))
.collect(Collectors.toList());
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java
index 43c7ed459d2..0d43df70c74 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java
@@ -278,11 +278,11 @@ public class RunIndexActionExecutor<T extends HoodieRecordPayload, I, K, O> exte
}
private static List<HoodieInstant> getRemainingArchivedAndActiveInstantsSince(String instant, HoodieTableMetaClient metaClient) {
- List<HoodieInstant> remainingInstantsToIndex = metaClient.getArchivedTimeline().getInstants()
+ List<HoodieInstant> remainingInstantsToIndex = metaClient.getArchivedTimeline().getInstantsAsStream()
.filter(i -> HoodieTimeline.compareTimestamps(i.getTimestamp(), GREATER_THAN_OR_EQUALS, instant))
.filter(i -> !INDEXING_ACTION.equals(i.getAction()))
.collect(Collectors.toList());
- remainingInstantsToIndex.addAll(metaClient.getActiveTimeline().findInstantsAfter(instant).getInstants()
+ remainingInstantsToIndex.addAll(metaClient.getActiveTimeline().findInstantsAfter(instant).getInstantsAsStream()
.filter(i -> HoodieTimeline.compareTimestamps(i.getTimestamp(), GREATER_THAN_OR_EQUALS, instant))
.filter(i -> !INDEXING_ACTION.equals(i.getAction()))
.collect(Collectors.toList()));
@@ -291,9 +291,9 @@ public class RunIndexActionExecutor<T extends HoodieRecordPayload, I, K, O> exte
private static List<HoodieInstant> getCompletedArchivedAndActiveInstantsAfter(String instant, HoodieTableMetaClient metaClient) {
List<HoodieInstant> completedInstants = metaClient.getArchivedTimeline().filterCompletedInstants().findInstantsAfter(instant)
- .getInstants().filter(i -> !INDEXING_ACTION.equals(i.getAction())).collect(Collectors.toList());
+ .getInstantsAsStream().filter(i -> !INDEXING_ACTION.equals(i.getAction())).collect(Collectors.toList());
completedInstants.addAll(metaClient.reloadActiveTimeline().filterCompletedInstants().findInstantsAfter(instant)
- .getInstants().filter(i -> !INDEXING_ACTION.equals(i.getAction())).collect(Collectors.toList()));
+ .getInstantsAsStream().filter(i -> !INDEXING_ACTION.equals(i.getAction())).collect(Collectors.toList()));
return completedInstants;
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
index ef2c7905161..eb888bdec77 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
@@ -140,7 +140,7 @@ public abstract class BaseRollbackActionExecutor<T extends HoodieRecordPayload,
private void validateSavepointRollbacks() {
// Check if any of the commits is a savepoint - do not allow rollback on those commits
- List<String> savepoints = table.getCompletedSavepointTimeline().getInstants()
+ List<String> savepoints = table.getCompletedSavepointTimeline().getInstantsAsStream()
.map(HoodieInstant::getTimestamp)
.collect(Collectors.toList());
savepoints.forEach(s -> {
@@ -176,7 +176,7 @@ public abstract class BaseRollbackActionExecutor<T extends HoodieRecordPayload,
}
}
- List<String> inflights = inflightAndRequestedCommitTimeline.getInstants().filter(instant -> {
+ List<String> inflights = inflightAndRequestedCommitTimeline.getInstantsAsStream().filter(instant -> {
if (!instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
return true;
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java
index 5aa03a4bddf..81ccb0a620a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java
@@ -121,8 +121,8 @@ public final class RepairUtils {
String instantToRepair, List<String> baseAndLogFilesFromFs,
HoodieActiveTimeline activeTimeline, HoodieArchivedTimeline archivedTimeline) {
// Skips the instant if it is requested or inflight in active timeline
- if (activeTimeline.filter(instant -> instant.getTimestamp().equals(instantToRepair)
- && !instant.isCompleted()).getInstants().findAny().isPresent()) {
+ if (!activeTimeline.filter(instant -> instant.getTimestamp().equals(instantToRepair)
+ && !instant.isCompleted()).empty()) {
return Collections.emptyList();
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java
index 95f22bba27d..ca605970395 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java
@@ -86,7 +86,7 @@ public class ZeroToOneUpgradeHandler implements UpgradeHandler {
int parallelism) throws HoodieRollbackException {
try {
// fetch hoodie instant
- Option<HoodieInstant> commitInstantOpt = Option.fromJavaOptional(table.getActiveTimeline().getCommitsTimeline().getInstants()
+ Option<HoodieInstant> commitInstantOpt = Option.fromJavaOptional(table.getActiveTimeline().getCommitsTimeline().getInstantsAsStream()
.filter(instant -> HoodieActiveTimeline.EQUALS.test(instant.getTimestamp(), commitInstantTime))
.findFirst());
if (commitInstantOpt.isPresent()) {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java
index 92eeac85535..c810ff4f968 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java
@@ -161,7 +161,7 @@ public class TestClientRollback extends HoodieClientTestBase {
}, "Rolling back to non-existent savepoint should not be allowed");
// rollback to savepoint 002
- HoodieInstant savepoint = table.getCompletedSavepointTimeline().getInstants().findFirst().get();
+ HoodieInstant savepoint = table.getCompletedSavepointTimeline().getInstantsAsStream().findFirst().get();
client.restoreToSavepoint(savepoint.getTimestamp());
metaClient = HoodieTableMetaClient.reload(metaClient);
@@ -259,7 +259,7 @@ public class TestClientRollback extends HoodieClientTestBase {
assertEquals(3, dataFiles.size(), "The data files for commit 004 should be present");
// rollback to savepoint 002
- HoodieInstant savepoint = table.getCompletedSavepointTimeline().getInstants().findFirst().get();
+ HoodieInstant savepoint = table.getCompletedSavepointTimeline().getInstantsAsStream().findFirst().get();
client.restoreToSavepoint(savepoint.getTimestamp());
metaClient = HoodieTableMetaClient.reload(metaClient);
@@ -453,7 +453,7 @@ public class TestClientRollback extends HoodieClientTestBase {
assertTrue(testTable.baseFilesExist(partitionAndFileId2, commitTime2));
metaClient.reloadActiveTimeline();
- List<HoodieInstant> rollbackInstants = metaClient.getActiveTimeline().getRollbackTimeline().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> rollbackInstants = metaClient.getActiveTimeline().getRollbackTimeline().getInstants();
assertEquals(rollbackInstants.size(), 1);
HoodieInstant rollbackInstant = rollbackInstants.get(0);
@@ -470,7 +470,7 @@ public class TestClientRollback extends HoodieClientTestBase {
// verify there are no extra rollback instants
metaClient.reloadActiveTimeline();
- rollbackInstants = metaClient.getActiveTimeline().getRollbackTimeline().getInstants().collect(Collectors.toList());
+ rollbackInstants = metaClient.getActiveTimeline().getRollbackTimeline().getInstants();
assertEquals(rollbackInstants.size(), 1);
assertEquals(rollbackInstants.get(0), rollbackInstant);
@@ -492,7 +492,7 @@ public class TestClientRollback extends HoodieClientTestBase {
client.rollback(commitTime4);
metaClient.reloadActiveTimeline();
- rollbackInstants = metaClient.reloadActiveTimeline().getRollbackTimeline().getInstants().collect(Collectors.toList());
+ rollbackInstants = metaClient.reloadActiveTimeline().getRollbackTimeline().getInstants();
assertEquals(2, rollbackInstants.size());
}
}
@@ -670,7 +670,7 @@ public class TestClientRollback extends HoodieClientTestBase {
assertTrue(testTable.baseFilesExist(partitionAndFileId2, commitTime2));
metaClient.reloadActiveTimeline();
- List<HoodieInstant> rollbackInstants = metaClient.getActiveTimeline().getRollbackTimeline().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> rollbackInstants = metaClient.getActiveTimeline().getRollbackTimeline().getInstants();
// Corrupted requested rollback plan should be deleted before scheduling a new one
assertEquals(rollbackInstants.size(), 1);
HoodieInstant rollbackInstant = rollbackInstants.get(0);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
index 6ad8666a0fa..73fefc70ca6 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
@@ -414,9 +414,9 @@ public class TestHoodieClientMultiWriter extends HoodieClientTestBase {
validInstants.addAll(
metaClient.reloadActiveTimeline().getCompletedReplaceTimeline()
- .filterCompletedInstants().getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toSet()));
+ .filterCompletedInstants().getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toSet()));
Set<String> completedInstants = metaClient.reloadActiveTimeline().getCommitsTimeline()
- .filterCompletedInstants().getInstants().map(HoodieInstant::getTimestamp)
+ .filterCompletedInstants().getInstantsAsStream().map(HoodieInstant::getTimestamp)
.collect(Collectors.toSet());
assertTrue(validInstants.containsAll(completedInstants));
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
index 0da9e890b3c..fd8013c1505 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
@@ -621,7 +621,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
0, 150);
HoodieActiveTimeline activeTimeline = new HoodieActiveTimeline(metaClient, false);
- List<HoodieInstant> instants = activeTimeline.getCommitTimeline().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> instants = activeTimeline.getCommitTimeline().getInstants();
assertEquals(5, instants.size());
assertEquals(new HoodieInstant(COMPLETED, COMMIT_ACTION, "001"),
instants.get(0));
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
index 0af05b2d6b0..76b51e6970b 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
@@ -864,7 +864,7 @@ public class TestHoodieTimelineArchiver extends HoodieClientTestHarness {
HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
List<HoodieInstant> archivedInstants = Arrays.asList(instant1, instant2, instant3);
assertEquals(new HashSet<>(archivedInstants),
- archivedTimeline.filterCompletedInstants().getInstants().collect(Collectors.toSet()));
+ archivedTimeline.filterCompletedInstants().getInstantsAsStream().collect(Collectors.toSet()));
assertFalse(wrapperFs.exists(markerPath));
}
@@ -1030,7 +1030,7 @@ public class TestHoodieTimelineArchiver extends HoodieClientTestHarness {
archiver.archiveIfRequired(context);
- Stream<HoodieInstant> currentInstants = metaClient.getActiveTimeline().reload().getInstants();
+ Stream<HoodieInstant> currentInstants = metaClient.getActiveTimeline().reload().getInstantsAsStream();
Map<Object, List<HoodieInstant>> actionInstantMap = currentInstants.collect(Collectors.groupingBy(HoodieInstant::getAction));
assertTrue(actionInstantMap.containsKey("clean"), "Clean Action key must be preset");
@@ -1080,7 +1080,7 @@ public class TestHoodieTimelineArchiver extends HoodieClientTestHarness {
archiver.archiveIfRequired(context);
- List<HoodieInstant> notArchivedInstants = metaClient.getActiveTimeline().reload().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> notArchivedInstants = metaClient.getActiveTimeline().reload().getInstants();
assertEquals(3, notArchivedInstants.size(), "Not archived instants should be 3");
assertEquals(notArchivedInstants, Arrays.asList(notArchivedInstant1, notArchivedInstant2, notArchivedInstant3), "");
}
@@ -1285,8 +1285,7 @@ public class TestHoodieTimelineArchiver extends HoodieClientTestHarness {
metadataTableMetaClient = HoodieTableMetaClient.reload(metadataTableMetaClient);
List<HoodieInstant> metadataTableInstants = metadataTableMetaClient.getActiveTimeline()
- .getCommitsTimeline().filterCompletedInstants().getInstants()
- .collect(Collectors.toList());
+ .getCommitsTimeline().filterCompletedInstants().getInstants();
if (i <= 7) {
// In the metadata table timeline, the first delta commit is "00000000000000"
@@ -1370,12 +1369,12 @@ public class TestHoodieTimelineArchiver extends HoodieClientTestHarness {
private Pair<List<HoodieInstant>, List<HoodieInstant>> archiveAndGetCommitsList(HoodieWriteConfig writeConfig) throws IOException {
metaClient.reloadActiveTimeline();
HoodieTimeline timeline = metaClient.getActiveTimeline().reload().getAllCommitsTimeline().filterCompletedInstants();
- List<HoodieInstant> originalCommits = timeline.getInstants().collect(Collectors.toList());
+ List<HoodieInstant> originalCommits = timeline.getInstants();
HoodieTable table = HoodieSparkTable.create(writeConfig, context, metaClient);
HoodieTimelineArchiver archiver = new HoodieTimelineArchiver(writeConfig, table);
archiver.archiveIfRequired(context);
timeline = metaClient.getActiveTimeline().reload().getAllCommitsTimeline().filterCompletedInstants();
- List<HoodieInstant> commitsAfterArchival = timeline.getInstants().collect(Collectors.toList());
+ List<HoodieInstant> commitsAfterArchival = timeline.getInstants();
return Pair.of(originalCommits, commitsAfterArchival);
}
@@ -1385,7 +1384,7 @@ public class TestHoodieTimelineArchiver extends HoodieClientTestHarness {
assertEquals(expectedActiveInstants, commitsAfterArchival);
expectedArchivedInstants.forEach(entry -> assertFalse(commitsAfterArchival.contains(entry)));
HoodieArchivedTimeline archivedTimeline = new HoodieArchivedTimeline(metaClient);
- List<HoodieInstant> actualArchivedInstants = archivedTimeline.getInstants().collect(Collectors.toList());
+ List<HoodieInstant> actualArchivedInstants = archivedTimeline.getInstants();
Collections.sort(actualArchivedInstants, Comparator.comparing(HoodieInstant::getTimestamp));
Collections.sort(expectedArchivedInstants, Comparator.comparing(HoodieInstant::getTimestamp));
assertEquals(actualArchivedInstants, expectedArchivedInstants);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByCommits.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByCommits.java
index 816a9371870..f6de3bd2805 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByCommits.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByCommits.java
@@ -158,11 +158,11 @@ public class TestCleanerInsertAndCleanByCommits extends SparkClientFunctionalTes
// NOTE: See CleanPlanner#getFilesToCleanKeepingLatestCommits. We explicitly keep one commit before earliest
// commit
Option<HoodieInstant> earliestRetainedCommit = activeTimeline.nthFromLastInstant(maxCommits);
- Set<HoodieInstant> acceptableCommits = activeTimeline.getInstants().collect(Collectors.toSet());
+ Set<HoodieInstant> acceptableCommits = activeTimeline.getInstantsAsStream().collect(Collectors.toSet());
if (earliestRetainedCommit.isPresent()) {
acceptableCommits
.removeAll(activeTimeline.findInstantsInRange("000", earliestRetainedCommit.get().getTimestamp())
- .getInstants().collect(Collectors.toSet()));
+ .getInstantsAsStream().collect(Collectors.toSet()));
acceptableCommits.add(earliestRetainedCommit.get());
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java
index e9c74936f3c..b6e19941bcd 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java
@@ -190,7 +190,7 @@ public class TestCleanerInsertAndCleanByVersions extends SparkClientFunctionalTe
for (String partitionPath : dataGen.getPartitionPaths()) {
// compute all the versions of all files, from time 0
HashMap<String, TreeSet<String>> fileIdToVersions = new HashMap<>();
- for (HoodieInstant entry : timeline.getInstants().collect(Collectors.toList())) {
+ for (HoodieInstant entry : timeline.getInstants()) {
HoodieCommitMetadata commitMetadata =
HoodieCommitMetadata.fromBytes(timeline.getInstantDetails(entry).get(), HoodieCommitMetadata.class);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java
index a571a6f4732..034f239318e 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java
@@ -154,7 +154,7 @@ public class CompactionTestBase extends HoodieClientTestBase {
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(cfg.getBasePath()).build();
HoodieInstant compactionInstant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
metaClient.getActiveTimeline().transitionCompactionRequestedToInflight(compactionInstant);
- HoodieInstant instant = metaClient.getActiveTimeline().reload().filterPendingCompactionTimeline().getInstants()
+ HoodieInstant instant = metaClient.getActiveTimeline().reload().filterPendingCompactionTimeline().getInstantsAsStream()
.filter(in -> in.getTimestamp().equals(compactionInstantTime)).findAny().get();
assertTrue(instant.isInflight(), "Instant must be marked inflight");
}
@@ -219,7 +219,7 @@ public class CompactionTestBase extends HoodieClientTestBase {
table = getHoodieTable(HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(cfg.getBasePath()).setLoadActiveTimelineOnLoad(true).build(), cfg);
HoodieTimeline timeline = table.getMetaClient().getCommitTimeline().filterCompletedInstants();
// verify compaction commit is visible in timeline
- assertTrue(timeline.filterCompletedInstants().getInstants()
+ assertTrue(timeline.filterCompletedInstants().getInstantsAsStream()
.filter(instant -> compactionInstantTime.equals(instant.getTimestamp())).findFirst().isPresent());
for (String partition: partitions) {
table.getSliceView().getLatestFileSlicesBeforeOrOn(partition, compactionInstantTime, true).forEach(fs -> {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestAsyncCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestAsyncCompaction.java
index f673872804a..18d6e5a261e 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestAsyncCompaction.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestAsyncCompaction.java
@@ -92,7 +92,7 @@ public class TestAsyncCompaction extends CompactionTestBase {
new HoodieInstant(State.INFLIGHT, HoodieTimeline.COMPACTION_ACTION, compactionInstantTime));
metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(cfg.getBasePath()).build();
pendingCompactionInstant = metaClient.getCommitsAndCompactionTimeline().filterPendingCompactionTimeline()
- .getInstants().findFirst().get();
+ .getInstantsAsStream().findFirst().get();
assertEquals("compaction", pendingCompactionInstant.getAction());
assertEquals(State.REQUESTED, pendingCompactionInstant.getState());
assertEquals(compactionInstantTime, pendingCompactionInstant.getTimestamp());
@@ -148,7 +148,7 @@ public class TestAsyncCompaction extends CompactionTestBase {
inflightInstant = metaClient.getActiveTimeline().filterPendingExcludingCompaction().firstInstant().get();
assertEquals(inflightInstant.getTimestamp(), nextInflightInstantTime, "inflight instant has expected instant time");
assertEquals(1, metaClient.getActiveTimeline()
- .filterPendingExcludingCompaction().getInstants().count(),
+ .filterPendingExcludingCompaction().countInstants(),
"Expect only one inflight instant");
// Expect pending Compaction to be present
pendingCompactionInstant = metaClient.getActiveTimeline().filterPendingCompactionTimeline().firstInstant().get();
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java
index 1797f61ed2a..ba8ca389309 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java
@@ -126,7 +126,7 @@ public class TestInlineCompaction extends CompactionTestBase {
scheduleCompaction(requestInstant, writeClient, cfg);
metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(cfg.getBasePath()).build();
- assertEquals(metaClient.getActiveTimeline().getInstants()
+ assertEquals(metaClient.getActiveTimeline().getInstantsAsStream()
.filter(hoodieInstant -> hoodieInstant.getAction().equals(HoodieTimeline.COMPACTION_ACTION)
&& hoodieInstant.getState() == HoodieInstant.State.REQUESTED).count(), 1);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
index 6cc936df980..e2e764e15b8 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
@@ -157,8 +157,8 @@ public class HoodieClientTestUtils {
public static long countRecordsOptionallySince(JavaSparkContext jsc, String basePath, SQLContext sqlContext,
HoodieTimeline commitTimeline, Option<String> lastCommitTimeOpt) {
List<HoodieInstant> commitsToReturn =
- lastCommitTimeOpt.isPresent() ? commitTimeline.findInstantsAfter(lastCommitTimeOpt.get(), Integer.MAX_VALUE).getInstants().collect(Collectors.toList()) :
- commitTimeline.getInstants().collect(Collectors.toList());
+ lastCommitTimeOpt.isPresent() ? commitTimeline.findInstantsAfter(lastCommitTimeOpt.get(), Integer.MAX_VALUE).getInstants() :
+ commitTimeline.getInstants();
try {
// Go over the commit metadata, and obtain the new files that need to be read.
HashMap<String, String> fileIdToFullPath = getLatestFileIDsToFullPath(basePath, commitTimeline, commitsToReturn);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
index 7b085be7b20..a155955e5ec 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
@@ -520,7 +520,7 @@ public class HoodieTableMetaClient implements Serializable {
* @return {@code true} if any commits are found, else {@code false}.
*/
public boolean isTimelineNonEmpty() {
- return getCommitsTimeline().filterCompletedInstants().getInstants().collect(Collectors.toList()).size() > 0;
+ return !getCommitsTimeline().filterCompletedInstants().empty();
}
/**
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
index 2b27d3ab5e5..96e22e9dac0 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
@@ -225,7 +225,7 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
public HoodieInstant revertToInflight(HoodieInstant instant) {
LOG.info("Reverting instant to inflight " + instant);
- HoodieInstant inflight = HoodieTimeline.getInflightInstant(instant, metaClient.getTableType());
+ HoodieInstant inflight = HoodieTimeline.getInflightInstant(instant, metaClient);
revertCompleteToInflight(instant, inflight);
LOG.info("Reverted " + instant + " to inflight " + inflight);
return inflight;
@@ -336,7 +336,7 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
private Stream<Pair<HoodieInstant, HoodieCommitMetadata>> getCommitMetadataStream() {
// NOTE: Streams are lazy
return getCommitsTimeline().filterCompletedInstants()
- .getInstants()
+ .getInstantsAsStream()
.sorted(Comparator.comparing(HoodieInstant::getTimestamp).reversed())
.map(instant -> {
try {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
index 4df30b115e0..aab94c72e2f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
@@ -363,8 +363,8 @@ public class HoodieArchivedTimeline extends HoodieDefaultTimeline {
@Override
public HoodieDefaultTimeline getWriteTimeline() {
// filter in-memory instants
- Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, REPLACE_COMMIT_ACTION);
- return new HoodieDefaultTimeline(getInstants().filter(i ->
+ Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, LOG_COMPACTION_ACTION, REPLACE_COMMIT_ACTION);
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(i ->
readCommits.containsKey(i.getTimestamp()))
.filter(s -> validActions.contains(s.getAction())), details);
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
index 7324421894c..7867ce18089 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
@@ -27,6 +27,7 @@ import org.apache.hudi.exception.HoodieException;
import java.io.Serializable;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
@@ -34,7 +35,6 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static java.util.Collections.reverse;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.compareTimestamps;
/**
@@ -81,37 +81,37 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
@Override
public HoodieTimeline filterInflights() {
- return new HoodieDefaultTimeline(instants.stream().filter(HoodieInstant::isInflight), details);
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(HoodieInstant::isInflight), details);
}
@Override
public HoodieTimeline filterInflightsAndRequested() {
return new HoodieDefaultTimeline(
- instants.stream().filter(i -> i.getState().equals(State.REQUESTED) || i.getState().equals(State.INFLIGHT)),
+ getInstantsAsStream().filter(i -> i.getState().equals(State.REQUESTED) || i.getState().equals(State.INFLIGHT)),
details);
}
@Override
public HoodieTimeline filterPendingExcludingCompaction() {
- return new HoodieDefaultTimeline(instants.stream().filter(instant -> (!instant.isCompleted())
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(instant -> (!instant.isCompleted())
&& (!instant.getAction().equals(HoodieTimeline.COMPACTION_ACTION))), details);
}
@Override
public HoodieTimeline filterCompletedInstants() {
- return new HoodieDefaultTimeline(instants.stream().filter(HoodieInstant::isCompleted), details);
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(HoodieInstant::isCompleted), details);
}
@Override
public HoodieTimeline filterCompletedAndCompactionInstants() {
- return new HoodieDefaultTimeline(instants.stream().filter(s -> s.isCompleted()
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(s -> s.isCompleted()
|| s.getAction().equals(HoodieTimeline.COMPACTION_ACTION)), details);
}
@Override
public HoodieDefaultTimeline getWriteTimeline() {
- Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, REPLACE_COMMIT_ACTION);
- return new HoodieDefaultTimeline(instants.stream().filter(s -> validActions.contains(s.getAction())), details);
+ Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, LOG_COMPACTION_ACTION, REPLACE_COMMIT_ACTION);
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(s -> validActions.contains(s.getAction())), details);
}
@Override
@@ -127,80 +127,80 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
@Override
public HoodieTimeline getCompletedReplaceTimeline() {
return new HoodieDefaultTimeline(
- instants.stream().filter(s -> s.getAction().equals(REPLACE_COMMIT_ACTION)).filter(HoodieInstant::isCompleted), details);
+ getInstantsAsStream().filter(s -> s.getAction().equals(REPLACE_COMMIT_ACTION)).filter(HoodieInstant::isCompleted), details);
}
@Override
public HoodieTimeline filterPendingReplaceTimeline() {
- return new HoodieDefaultTimeline(instants.stream().filter(
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(
s -> s.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION) && !s.isCompleted()), details);
}
@Override
public HoodieTimeline filterPendingRollbackTimeline() {
- return new HoodieDefaultTimeline(instants.stream().filter(
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(
s -> s.getAction().equals(HoodieTimeline.ROLLBACK_ACTION) && !s.isCompleted()), details);
}
@Override
public HoodieTimeline filterPendingCompactionTimeline() {
return new HoodieDefaultTimeline(
- instants.stream().filter(s -> s.getAction().equals(HoodieTimeline.COMPACTION_ACTION) && !s.isCompleted()), details);
+ getInstantsAsStream().filter(s -> s.getAction().equals(HoodieTimeline.COMPACTION_ACTION) && !s.isCompleted()), details);
}
@Override
public HoodieDefaultTimeline findInstantsInRange(String startTs, String endTs) {
return new HoodieDefaultTimeline(
- instants.stream().filter(s -> HoodieTimeline.isInRange(s.getTimestamp(), startTs, endTs)), details);
+ getInstantsAsStream().filter(s -> HoodieTimeline.isInRange(s.getTimestamp(), startTs, endTs)), details);
}
@Override
public HoodieDefaultTimeline findInstantsAfter(String instantTime, int numCommits) {
- return new HoodieDefaultTimeline(instants.stream()
+ return new HoodieDefaultTimeline(getInstantsAsStream()
.filter(s -> compareTimestamps(s.getTimestamp(), GREATER_THAN, instantTime)).limit(numCommits),
details);
}
@Override
public HoodieTimeline findInstantsAfter(String instantTime) {
- return new HoodieDefaultTimeline(instants.stream()
+ return new HoodieDefaultTimeline(getInstantsAsStream()
.filter(s -> compareTimestamps(s.getTimestamp(), GREATER_THAN, instantTime)), details);
}
@Override
public HoodieDefaultTimeline findInstantsAfterOrEquals(String commitTime, int numCommits) {
- return new HoodieDefaultTimeline(instants.stream()
+ return new HoodieDefaultTimeline(getInstantsAsStream()
.filter(s -> compareTimestamps(s.getTimestamp(), GREATER_THAN_OR_EQUALS, commitTime))
.limit(numCommits), details);
}
@Override
public HoodieDefaultTimeline findInstantsBefore(String instantTime) {
- return new HoodieDefaultTimeline(instants.stream()
+ return new HoodieDefaultTimeline(getInstantsAsStream()
.filter(s -> compareTimestamps(s.getTimestamp(), LESSER_THAN, instantTime)),
details);
}
@Override
public HoodieDefaultTimeline findInstantsBeforeOrEquals(String instantTime) {
- return new HoodieDefaultTimeline(instants.stream()
+ return new HoodieDefaultTimeline(getInstantsAsStream()
.filter(s -> compareTimestamps(s.getTimestamp(), LESSER_THAN_OR_EQUALS, instantTime)),
details);
}
@Override
public HoodieTimeline filter(Predicate<HoodieInstant> filter) {
- return new HoodieDefaultTimeline(instants.stream().filter(filter), details);
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(filter), details);
}
@Override
public HoodieTimeline filterPendingIndexTimeline() {
- return new HoodieDefaultTimeline(instants.stream().filter(s -> s.getAction().equals(INDEXING_ACTION) && !s.isCompleted()), details);
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(s -> s.getAction().equals(INDEXING_ACTION) && !s.isCompleted()), details);
}
@Override
public HoodieTimeline filterCompletedIndexTimeline() {
- return new HoodieDefaultTimeline(instants.stream().filter(s -> s.getAction().equals(INDEXING_ACTION) && s.isCompleted()), details);
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(s -> s.getAction().equals(INDEXING_ACTION) && s.isCompleted()), details);
}
/**
@@ -240,7 +240,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
* @param actions actions allowed in the timeline
*/
public HoodieTimeline getTimelineOfActions(Set<String> actions) {
- return new HoodieDefaultTimeline(getInstants().filter(s -> actions.contains(s.getAction())),
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(s -> actions.contains(s.getAction())),
(Function<HoodieInstant, Option<byte[]>> & Serializable) this::getInstantDetails);
}
@@ -284,27 +284,27 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
}
protected Stream<HoodieInstant> filterInstantsByAction(String action) {
- return getInstants().filter(s -> s.getAction().equals(action));
+ return getInstantsAsStream().filter(s -> s.getAction().equals(action));
}
@Override
public boolean empty() {
- return !instants.stream().findFirst().isPresent();
+ return getInstants().isEmpty();
}
@Override
public int countInstants() {
- return instants.size();
+ return getInstants().size();
}
@Override
public Option<HoodieInstant> firstInstant() {
- return Option.fromJavaOptional(instants.stream().findFirst());
+ return Option.fromJavaOptional(getInstantsAsStream().findFirst());
}
@Override
public Option<HoodieInstant> firstInstant(String action, State state) {
- return Option.fromJavaOptional(instants.stream()
+ return Option.fromJavaOptional(getInstantsAsStream()
.filter(s -> action.equals(s.getAction()) && state.equals(s.getState())).findFirst());
}
@@ -313,7 +313,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
if (empty() || n >= countInstants()) {
return Option.empty();
}
- return Option.of(instants.get(n));
+ return Option.of(getInstants().get(n));
}
@Override
@@ -331,17 +331,17 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
@Override
public boolean containsInstant(HoodieInstant instant) {
- return instants.stream().anyMatch(s -> s.equals(instant));
+ return getInstantsAsStream().anyMatch(s -> s.equals(instant));
}
@Override
public boolean containsInstant(String ts) {
- return instants.stream().anyMatch(s -> s.getTimestamp().equals(ts));
+ return getInstantsAsStream().anyMatch(s -> s.getTimestamp().equals(ts));
}
@Override
public boolean containsOrBeforeTimelineStarts(String instant) {
- return instants.stream().anyMatch(s -> s.getTimestamp().equals(instant)) || isBeforeTimelineStarts(instant);
+ return getInstantsAsStream().anyMatch(s -> s.getTimestamp().equals(instant)) || isBeforeTimelineStarts(instant);
}
@Override
@@ -350,15 +350,18 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
}
@Override
- public Stream<HoodieInstant> getInstants() {
+ public Stream<HoodieInstant> getInstantsAsStream() {
return instants.stream();
}
+ @Override
+ public List<HoodieInstant> getInstants() {
+ return new ArrayList<>(instants);
+ }
+
@Override
public Stream<HoodieInstant> getReverseOrderedInstants() {
- List<HoodieInstant> instants = getInstants().collect(Collectors.toList());
- reverse(instants);
- return instants.stream();
+ return getInstantsAsStream().sorted(HoodieInstant.COMPARATOR.reversed());
}
@Override
@@ -370,7 +373,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
public Option<HoodieInstant> getFirstNonSavepointCommit() {
Option<HoodieInstant> firstCommit = firstInstant();
- Set<String> savepointTimestamps = instants.stream()
+ Set<String> savepointTimestamps = getInstantsAsStream()
.filter(entry -> entry.getAction().equals(HoodieTimeline.SAVEPOINT_ACTION))
.map(HoodieInstant::getTimestamp)
.collect(Collectors.toSet());
@@ -378,7 +381,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
if (!savepointTimestamps.isEmpty()) {
// There are chances that there could be holes in the timeline due to archival and savepoint interplay.
// So, the first non-savepoint commit is considered as beginning of the active timeline.
- firstNonSavepointCommit = Option.fromJavaOptional(instants.stream()
+ firstNonSavepointCommit = Option.fromJavaOptional(getInstantsAsStream()
.filter(entry -> !savepointTimestamps.contains(entry.getTimestamp()))
.findFirst());
}
@@ -397,16 +400,16 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
@Override
public String toString() {
- return this.getClass().getName() + ": " + instants.stream().map(Object::toString).collect(Collectors.joining(","));
+ return this.getClass().getName() + ": " + getInstantsAsStream().map(Object::toString).collect(Collectors.joining(","));
}
/**
* Merge this timeline with the given timeline.
*/
public HoodieDefaultTimeline mergeTimeline(HoodieDefaultTimeline timeline) {
- Stream<HoodieInstant> instantStream = Stream.concat(instants.stream(), timeline.getInstants()).sorted();
+ Stream<HoodieInstant> instantStream = Stream.concat(getInstantsAsStream(), timeline.getInstantsAsStream()).sorted();
Function<HoodieInstant, Option<byte[]>> details = instant -> {
- if (instants.stream().anyMatch(i -> i.equals(instant))) {
+ if (getInstantsAsStream().anyMatch(i -> i.equals(instant))) {
return this.getInstantDetails(instant);
} else {
return timeline.getInstantDetails(instant);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
index e52a2795969..5a3b31cb2b5 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import java.io.Serializable;
+import java.util.List;
import java.util.function.BiPredicate;
import java.util.function.Predicate;
import java.util.stream.Stream;
@@ -53,6 +54,7 @@ public interface HoodieTimeline extends Serializable {
// With Async Compaction, compaction instant can be in 3 states :
// (compaction-requested), (compaction-inflight), (completed)
String COMPACTION_ACTION = "compaction";
+ String LOG_COMPACTION_ACTION = "logcompaction";
String REQUESTED_EXTENSION = ".requested";
String RESTORE_ACTION = "restore";
String INDEXING_ACTION = "indexing";
@@ -292,7 +294,12 @@ public interface HoodieTimeline extends Serializable {
/**
* @return Get the stream of completed instants
*/
- Stream<HoodieInstant> getInstants();
+ Stream<HoodieInstant> getInstantsAsStream();
+
+ /**
+ * @return Get tht list of instants
+ */
+ List<HoodieInstant> getInstants();
/**
* @return Get the stream of completed instants in reverse order TODO Change code references to getInstants() that
@@ -390,12 +397,23 @@ public interface HoodieTimeline extends Serializable {
* Returns the inflight instant corresponding to the instant being passed. Takes care of changes in action names
* between inflight and completed instants (compaction <=> commit).
* @param instant Hoodie Instant
- * @param tableType Hoodie Table Type
* @return Inflight Hoodie Instant
*/
- static HoodieInstant getInflightInstant(final HoodieInstant instant, final HoodieTableType tableType) {
- if ((tableType == HoodieTableType.MERGE_ON_READ) && instant.getAction().equals(COMMIT_ACTION)) {
- return new HoodieInstant(true, COMPACTION_ACTION, instant.getTimestamp());
+ static HoodieInstant getInflightInstant(final HoodieInstant instant, final HoodieTableMetaClient metaClient) {
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ if (instant.getAction().equals(COMMIT_ACTION)) {
+ return new HoodieInstant(true, COMPACTION_ACTION, instant.getTimestamp());
+ } else if (instant.getAction().equals(DELTA_COMMIT_ACTION)) {
+ // Deltacommit is used by both ingestion and logcompaction.
+ // So, distinguish both of them check for the inflight file being present.
+ HoodieActiveTimeline rawActiveTimeline = new HoodieActiveTimeline(metaClient, false);
+ Option<HoodieInstant> logCompactionInstant = Option.fromJavaOptional(rawActiveTimeline.getInstantsAsStream()
+ .filter(hoodieInstant -> hoodieInstant.getTimestamp().equals(instant.getTimestamp())
+ && LOG_COMPACTION_ACTION.equals(hoodieInstant.getAction())).findFirst());
+ if (logCompactionInstant.isPresent()) {
+ return new HoodieInstant(true, LOG_COMPACTION_ACTION, instant.getTimestamp());
+ }
+ }
}
return new HoodieInstant(true, instant.getAction(), instant.getTimestamp());
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java
index 5659dcbdf7f..e03aa523921 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java
@@ -52,7 +52,7 @@ public class TimelineDiffHelper {
// The last seen instant is no longer in the timeline. Do not incrementally Sync.
return TimelineDiffResult.UNSAFE_SYNC_RESULT;
}
- Set<HoodieInstant> oldTimelineInstants = oldT.getInstants().collect(Collectors.toSet());
+ Set<HoodieInstant> oldTimelineInstants = oldT.getInstantsAsStream().collect(Collectors.toSet());
List<HoodieInstant> newInstants = new ArrayList<>();
@@ -72,7 +72,7 @@ public class TimelineDiffHelper {
&& instantPair.getValue().isCompleted())
.map(Pair::getKey).collect(Collectors.toList());
- newT.getInstants().filter(instant -> !oldTimelineInstants.contains(instant)).forEach(newInstants::add);
+ newT.getInstantsAsStream().filter(instant -> !oldTimelineInstants.contains(instant)).forEach(newInstants::add);
return new TimelineDiffResult(newInstants, finishedCompactionInstants, true);
} else {
// One or more timelines is empty
@@ -81,11 +81,14 @@ public class TimelineDiffHelper {
}
}
+ /**
+ * Getting pending compaction transitions.
+ */
private static List<Pair<HoodieInstant, HoodieInstant>> getPendingCompactionTransitions(HoodieTimeline oldTimeline,
HoodieTimeline newTimeline) {
- Set<HoodieInstant> newTimelineInstants = newTimeline.getInstants().collect(Collectors.toSet());
+ Set<HoodieInstant> newTimelineInstants = newTimeline.getInstantsAsStream().collect(Collectors.toSet());
- return oldTimeline.filterPendingCompactionTimeline().getInstants().map(instant -> {
+ return oldTimeline.filterPendingCompactionTimeline().getInstantsAsStream().map(instant -> {
if (newTimelineInstants.contains(instant)) {
return Pair.of(instant, instant);
} else {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
index 965870d0e9c..6b517d6022f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
@@ -65,7 +65,7 @@ public class TimelineUtils {
public static List<String> getDroppedPartitions(HoodieTimeline timeline) {
HoodieTimeline replaceCommitTimeline = timeline.getWriteTimeline().filterCompletedInstants().getCompletedReplaceTimeline();
- return replaceCommitTimeline.getInstants().flatMap(instant -> {
+ return replaceCommitTimeline.getInstantsAsStream().flatMap(instant -> {
try {
HoodieReplaceCommitMetadata commitMetadata = HoodieReplaceCommitMetadata.fromBytes(
replaceCommitTimeline.getInstantDetails(instant).get(), HoodieReplaceCommitMetadata.class);
@@ -85,7 +85,7 @@ public class TimelineUtils {
* Returns partitions that have been modified including internal operations such as clean in the passed timeline.
*/
public static List<String> getAffectedPartitions(HoodieTimeline timeline) {
- return timeline.filterCompletedInstants().getInstants().flatMap(s -> {
+ return timeline.filterCompletedInstants().getInstantsAsStream().flatMap(s -> {
switch (s.getAction()) {
case HoodieTimeline.COMMIT_ACTION:
case HoodieTimeline.DELTA_COMMIT_ACTION:
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/TimelineDTO.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/TimelineDTO.java
index b25b8e4bc51..740dcea5bfe 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/TimelineDTO.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/TimelineDTO.java
@@ -39,7 +39,7 @@ public class TimelineDTO {
public static TimelineDTO fromTimeline(HoodieTimeline timeline) {
TimelineDTO dto = new TimelineDTO();
- dto.instants = timeline.getInstants().map(InstantDTO::fromInstant).collect(Collectors.toList());
+ dto.instants = timeline.getInstantsAsStream().map(InstantDTO::fromInstant).collect(Collectors.toList());
return dto;
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
index 60104354db3..a29c197565e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
@@ -214,7 +214,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV
HoodieTimer hoodieTimer = HoodieTimer.start();
// for each REPLACE instant, get map of (partitionPath -> deleteFileGroup)
HoodieTimeline replacedTimeline = timeline.getCompletedReplaceTimeline();
- Stream<Map.Entry<HoodieFileGroupId, HoodieInstant>> resultStream = replacedTimeline.getInstants().flatMap(instant -> {
+ Stream<Map.Entry<HoodieFileGroupId, HoodieInstant>> resultStream = replacedTimeline.getInstantsAsStream().flatMap(instant -> {
try {
HoodieReplaceCommitMetadata replaceMetadata = HoodieReplaceCommitMetadata.fromBytes(metaClient.getActiveTimeline().getInstantDetails(instant).get(),
HoodieReplaceCommitMetadata.class);
@@ -389,7 +389,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV
*/
protected boolean isBaseFileDueToPendingClustering(HoodieBaseFile baseFile) {
List<String> pendingReplaceInstants =
- metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+ metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
return !pendingReplaceInstants.isEmpty() && pendingReplaceInstants.contains(baseFile.getCommitTime());
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
index b415b3bb64d..3b7d43cb6db 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
@@ -66,7 +66,7 @@ public class ClusteringUtils {
public static Stream<Pair<HoodieInstant, HoodieClusteringPlan>> getAllPendingClusteringPlans(
HoodieTableMetaClient metaClient) {
List<HoodieInstant> pendingReplaceInstants =
- metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstants().collect(Collectors.toList());
+ metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstants();
return pendingReplaceInstants.stream().map(instant -> getClusteringPlan(metaClient, instant))
.filter(Option::isPresent).map(Option::get);
}
@@ -216,7 +216,7 @@ public class ClusteringUtils {
}
public static List<HoodieInstant> getPendingClusteringInstantTimes(HoodieTableMetaClient metaClient) {
- return metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstants()
+ return metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstantsAsStream()
.filter(instant -> isPendingClusteringInstant(metaClient, instant))
.collect(Collectors.toList());
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java
index cf9b5fb3ced..4722995ae16 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java
@@ -132,7 +132,7 @@ public class CompactionUtils {
public static List<Pair<HoodieInstant, HoodieCompactionPlan>> getAllPendingCompactionPlans(
HoodieTableMetaClient metaClient) {
List<HoodieInstant> pendingCompactionInstants =
- metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstants().collect(Collectors.toList());
+ metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstants();
return pendingCompactionInstants.stream().map(instant -> {
try {
return Pair.of(instant, getCompactionPlan(metaClient, instant.getTimestamp()));
@@ -140,6 +140,7 @@ public class CompactionUtils {
throw new HoodieException(e);
}
}).collect(Collectors.toList());
+
}
public static HoodieCompactionPlan getCompactionPlan(HoodieTableMetaClient metaClient, String compactionInstant)
@@ -200,7 +201,7 @@ public class CompactionUtils {
* @return
*/
public static List<HoodieInstant> getPendingCompactionInstantTimes(HoodieTableMetaClient metaClient) {
- return metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstants().collect(Collectors.toList());
+ return metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstants();
}
/**
@@ -291,7 +292,7 @@ public class CompactionUtils {
return Option.of(deltaCommitsInfo.getRight());
} else {
// delta commits with the last one to keep
- List<HoodieInstant> instants = deltaCommitTimeline.getInstants()
+ List<HoodieInstant> instants = deltaCommitTimeline.getInstantsAsStream()
.limit(numDeltaCommits - maxDeltaCommits + 1).collect(Collectors.toList());
return Option.of(instants.get(instants.size() - 1));
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java b/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
index a848fd42e2e..846309b7b67 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
@@ -117,7 +117,7 @@ public class InternalSchemaCache {
private static Option<InternalSchema> getSchemaByReadingCommitFile(long versionID, HoodieTableMetaClient metaClient) {
try {
HoodieTimeline timeline = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
- List<HoodieInstant> instants = timeline.getInstants().filter(f -> f.getTimestamp().equals(String.valueOf(versionID))).collect(Collectors.toList());
+ List<HoodieInstant> instants = timeline.getInstantsAsStream().filter(f -> f.getTimestamp().equals(String.valueOf(versionID))).collect(Collectors.toList());
if (instants.isEmpty()) {
return Option.empty();
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/internal/schema/io/FileBasedInternalSchemaStorageManager.java b/hudi-common/src/main/java/org/apache/hudi/internal/schema/io/FileBasedInternalSchemaStorageManager.java
index 81814a3e6fc..3cd72886c48 100644
--- a/hudi-common/src/main/java/org/apache/hudi/internal/schema/io/FileBasedInternalSchemaStorageManager.java
+++ b/hudi-common/src/main/java/org/apache/hudi/internal/schema/io/FileBasedInternalSchemaStorageManager.java
@@ -131,7 +131,7 @@ public class FileBasedInternalSchemaStorageManager extends AbstractInternalSchem
private List<String> getValidInstants() {
return getMetaClient().getCommitsTimeline()
- .filterCompletedInstants().getInstants().map(f -> f.getTimestamp()).collect(Collectors.toList());
+ .filterCompletedInstants().getInstantsAsStream().map(f -> f.getTimestamp()).collect(Collectors.toList());
}
@Override
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
index 8c55c27a42f..11b25b4aad0 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
@@ -448,14 +448,14 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// Only those log files which have a corresponding completed instant on the dataset should be read
// This is because the metadata table is updated before the dataset instants are committed.
HoodieActiveTimeline datasetTimeline = dataMetaClient.getActiveTimeline();
- Set<String> validInstantTimestamps = datasetTimeline.filterCompletedInstants().getInstants()
+ Set<String> validInstantTimestamps = datasetTimeline.filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
// For any rollbacks and restores, we cannot neglect the instants that they are rolling back.
// The rollback instant should be more recent than the start of the timeline for it to have rolled back any
// instant which we have a log block for.
final String earliestInstantTime = validInstantTimestamps.isEmpty() ? SOLO_COMMIT_TIMESTAMP : Collections.min(validInstantTimestamps);
- datasetTimeline.getRollbackAndRestoreTimeline().filterCompletedInstants().getInstants()
+ datasetTimeline.getRollbackAndRestoreTimeline().filterCompletedInstants().getInstantsAsStream()
.filter(instant -> HoodieTimeline.compareTimestamps(instant.getTimestamp(), HoodieTimeline.GREATER_THAN, earliestInstantTime))
.forEach(instant -> {
validInstantTimestamps.addAll(getRollbackedCommits(instant, datasetTimeline));
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java b/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
index 5c227c67dd4..9f780727f11 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
@@ -76,7 +76,7 @@ public class TestHoodieTableMetaClient extends HoodieCommonTestHarness {
commitTimeline.saveAsComplete(instant, Option.of("test-detail".getBytes()));
commitTimeline = commitTimeline.reload();
HoodieInstant completedInstant = HoodieTimeline.getCompletedInstant(instant);
- assertEquals(completedInstant, commitTimeline.getInstants().findFirst().get(),
+ assertEquals(completedInstant, commitTimeline.getInstantsAsStream().findFirst().get(),
"Commit should be 1 and completed");
assertArrayEquals("test-detail".getBytes(), commitTimeline.getInstantDetails(completedInstant).get(),
"Commit value should be \"test-detail\"");
@@ -101,7 +101,7 @@ public class TestHoodieTableMetaClient extends HoodieCommonTestHarness {
activeTimeline = activeTimeline.reload();
activeCommitTimeline = activeTimeline.getCommitTimeline();
assertFalse(activeCommitTimeline.empty(), "Should be the 1 commit we made");
- assertEquals(completedInstant, activeCommitTimeline.getInstants().findFirst().get(),
+ assertEquals(completedInstant, activeCommitTimeline.getInstantsAsStream().findFirst().get(),
"Commit should be 1");
assertArrayEquals("test-detail".getBytes(), activeCommitTimeline.getInstantDetails(completedInstant).get(),
"Commit value should be \"test-detail\"");
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
index b4591d9aff4..e40341a2033 100755
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
@@ -113,16 +113,16 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
assertEquals(5, timeline.countInstants(), "Total instants should be 5");
assertStreamEquals(
Stream.of(instant1Complete, instant2Complete, instant3Complete, instant4Complete, instant5),
- timeline.getInstants(), "Check the instants stream");
+ timeline.getInstantsAsStream(), "Check the instants stream");
assertStreamEquals(
Stream.of(instant1Complete, instant2Complete, instant3Complete, instant4Complete, instant5),
- timeline.getCommitTimeline().getInstants(), "Check the instants stream");
+ timeline.getCommitTimeline().getInstantsAsStream(), "Check the instants stream");
assertStreamEquals(
Stream.of(instant1Complete, instant2Complete, instant3Complete, instant4Complete),
- timeline.getCommitTimeline().filterCompletedInstants().getInstants(),
+ timeline.getCommitTimeline().filterCompletedInstants().getInstantsAsStream(),
"Check the instants stream");
assertStreamEquals(Stream.of(instant5),
- timeline.getCommitTimeline().filterPendingExcludingCompaction().getInstants(),
+ timeline.getCommitTimeline().filterPendingExcludingCompaction().getInstantsAsStream(),
"Check the instants stream");
// Backwards compatibility testing for reading compaction plans
@@ -175,15 +175,15 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
Stream.of("21", "23"));
assertStreamEquals(Stream.of("05", "07", "09", "11"),
timeline.getCommitTimeline().filterCompletedInstants().findInstantsInRange("04", "11")
- .getInstants().map(HoodieInstant::getTimestamp),
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp),
"findInstantsInRange should return 4 instants");
assertStreamEquals(Stream.of("09", "11"),
timeline.getCommitTimeline().filterCompletedInstants().findInstantsAfter("07", 2)
- .getInstants().map(HoodieInstant::getTimestamp),
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp),
"findInstantsAfter 07 should return 2 instants");
assertStreamEquals(Stream.of("01", "03", "05"),
timeline.getCommitTimeline().filterCompletedInstants().findInstantsBefore("07")
- .getInstants().map(HoodieInstant::getTimestamp),
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp),
"findInstantsBefore 07 should return 3 instants");
assertFalse(timeline.empty());
assertFalse(timeline.getCommitTimeline().filterPendingExcludingCompaction().empty());
@@ -557,7 +557,7 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
timeline = new HoodieActiveTimeline(metaClient);
timeline.setInstants(allInstants);
List<HoodieInstant> validReplaceInstants =
- timeline.getCompletedReplaceTimeline().getInstants().collect(Collectors.toList());
+ timeline.getCompletedReplaceTimeline().getInstants();
assertEquals(1, validReplaceInstants.size());
assertEquals(instant.getTimestamp(), validReplaceInstants.get(0).getTimestamp());
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java b/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java
index 1851e7d8b2e..d0ec6af5f76 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java
@@ -761,7 +761,7 @@ public class TestIncrementalFSViewSync extends HoodieCommonTestHarness {
assertEquals(State.COMPLETED, view.getLastInstant().get().getState());
assertEquals(lastInstant.getAction(), view.getLastInstant().get().getAction(),
"Expected Last=" + lastInstant + ", Found Instants="
- + view.getTimeline().getInstants().collect(Collectors.toList()));
+ + view.getTimeline().getInstants());
partitions.forEach(p -> assertEquals(fileIdsPerPartition.size(), view.getLatestFileSlices(p).count()));
final long expTotalFileSlicesPerPartition = fileIdsPerPartition.size() * multiple;
partitions.forEach(p -> assertEquals(expTotalFileSlicesPerPartition, view.getAllFileSlices(p).count()));
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
index 30abe48cb4e..b7855bec767 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
@@ -245,14 +245,14 @@ public class TestCompactionUtils extends HoodieCommonTestHarness {
Pair<HoodieTimeline, HoodieInstant> actual =
CompactionUtils.getDeltaCommitsSinceLatestCompaction(timeline).get();
if (hasCompletedCompaction) {
- Stream<HoodieInstant> instants = actual.getLeft().getInstants();
+ Stream<HoodieInstant> instants = actual.getLeft().getInstantsAsStream();
assertEquals(
Stream.of(
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "07"),
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "08"),
new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, "09"))
.collect(Collectors.toList()),
- actual.getLeft().getInstants().collect(Collectors.toList()));
+ actual.getLeft().getInstants());
assertEquals(
new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, "06"),
actual.getRight());
@@ -268,7 +268,7 @@ public class TestCompactionUtils extends HoodieCommonTestHarness {
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "08"),
new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, "09"))
.collect(Collectors.toList()),
- actual.getLeft().getInstants().collect(Collectors.toList()));
+ actual.getLeft().getInstants());
assertEquals(
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "01"),
actual.getRight());
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/strategy/CompactionPlanStrategies.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/strategy/CompactionPlanStrategies.java
index 662dcabda32..5c9a1a7b335 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/strategy/CompactionPlanStrategies.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/strategy/CompactionPlanStrategies.java
@@ -19,6 +19,7 @@
package org.apache.hudi.sink.compact.strategy;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.sink.compact.FlinkCompactionConfig;
import org.apache.hudi.util.CompactionUtil;
@@ -44,7 +45,7 @@ public class CompactionPlanStrategies {
public static CompactionPlanStrategy getStrategy(FlinkCompactionConfig config) {
switch (config.compactionPlanSelectStrategy.toLowerCase(Locale.ROOT)) {
case CompactionPlanStrategy.ALL:
- return pendingCompactionTimeline -> pendingCompactionTimeline.getInstants().collect(Collectors.toList());
+ return HoodieTimeline::getInstants;
case CompactionPlanStrategy.INSTANTS:
return pendingCompactionTimeline -> {
if (StringUtils.isNullOrEmpty(config.compactionPlanInstant)) {
@@ -52,13 +53,13 @@ public class CompactionPlanStrategies {
return Collections.emptyList();
}
List<String> instants = Arrays.asList(config.compactionPlanInstant.split(","));
- return pendingCompactionTimeline.getInstants()
+ return pendingCompactionTimeline.getInstantsAsStream()
.filter(instant -> instants.contains(instant.getTimestamp()))
.collect(Collectors.toList());
};
case CompactionPlanStrategy.NUM_INSTANTS:
return pendingCompactionTimeline -> {
- List<HoodieInstant> pendingCompactionPlanInstants = pendingCompactionTimeline.getInstants().collect(Collectors.toList());
+ List<HoodieInstant> pendingCompactionPlanInstants = pendingCompactionTimeline.getInstants();
if (CompactionUtil.isLIFO(config.compactionSeq)) {
Collections.reverse(pendingCompactionPlanInstants);
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
index db13a6c2ae5..45a3e0bf837 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
@@ -262,7 +262,7 @@ public class WriteProfile {
oldFsView.close();
recordProfile();
- cleanMetadataCache(this.metaClient.getCommitsTimeline().filterCompletedInstants().getInstants());
+ cleanMetadataCache(this.metaClient.getCommitsTimeline().filterCompletedInstants().getInstantsAsStream());
this.smallFilesMap.clear();
this.reloadedCheckpointId = checkpointId;
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
index 874158e3f4a..3455417b6db 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
@@ -422,7 +422,7 @@ public class IncrementalInputSplits implements Serializable {
// read the archived metadata if the start instant is archived.
HoodieTimeline archivedTimeline = getArchivedReadTimeline(metaClient, instantRange.getStartInstant());
if (!archivedTimeline.empty()) {
- return archivedTimeline.getInstants()
+ return archivedTimeline.getInstantsAsStream()
.map(instant -> WriteProfiles.getCommitMetadata(tableName, path, instant, archivedTimeline)).collect(Collectors.toList());
}
}
@@ -454,12 +454,12 @@ public class IncrementalInputSplits implements Serializable {
if (issuedInstant != null) {
// returns early for streaming mode
return commitTimeline
- .getInstants()
+ .getInstantsAsStream()
.filter(s -> HoodieTimeline.compareTimestamps(s.getTimestamp(), GREATER_THAN, issuedInstant))
.collect(Collectors.toList());
}
- Stream<HoodieInstant> instantStream = completedTimeline.getInstants();
+ Stream<HoodieInstant> instantStream = completedTimeline.getInstantsAsStream();
if (OptionsResolver.isSpecificStartCommit(this.conf)) {
final String startCommit = this.conf.get(FlinkOptions.READ_START_COMMIT);
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
index d5d35f7494f..070937a1ab4 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
@@ -210,7 +210,7 @@ public class TestStreamWriteOperatorCoordinator {
final String metadataTableBasePath = HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
HoodieTableMetaClient metadataTableMetaClient = StreamerUtil.createMetaClient(metadataTableBasePath, HadoopConfigurations.getHadoopConf(conf));
HoodieTimeline completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(1L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP));
// test metadata table compaction
@@ -219,14 +219,14 @@ public class TestStreamWriteOperatorCoordinator {
instant = mockWriteWithMetadata();
metadataTableMetaClient.reloadActiveTimeline();
completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(i + 1L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(i + 1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(instant));
}
// the 5th commit triggers the compaction
mockWriteWithMetadata();
metadataTableMetaClient.reloadActiveTimeline();
completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedAndCompactionInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(7L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(7));
assertThat(completedTimeline.nthFromLastInstant(1).get().getTimestamp(), is(instant + "001"));
assertThat(completedTimeline.nthFromLastInstant(1).get().getAction(), is(HoodieTimeline.COMMIT_ACTION));
// write another 2 commits
@@ -234,7 +234,7 @@ public class TestStreamWriteOperatorCoordinator {
instant = mockWriteWithMetadata();
metadataTableMetaClient.reloadActiveTimeline();
completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(i + 1L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(i + 1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(instant));
}
@@ -242,7 +242,7 @@ public class TestStreamWriteOperatorCoordinator {
instant = mockWriteWithMetadata();
metadataTableMetaClient.reloadActiveTimeline();
completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedAndCompactionInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(10L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(10));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(instant + "002"));
assertThat(completedTimeline.lastInstant().get().getAction(), is(HoodieTimeline.CLEAN_ACTION));
@@ -254,7 +254,7 @@ public class TestStreamWriteOperatorCoordinator {
mockWriteWithMetadata();
metadataTableMetaClient.reloadActiveTimeline();
completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedAndCompactionInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(14L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(14));
assertThat(completedTimeline.nthFromLastInstant(1).get().getTimestamp(), is(instant + "001"));
assertThat(completedTimeline.nthFromLastInstant(1).get().getAction(), is(HoodieTimeline.COMMIT_ACTION));
}
@@ -281,7 +281,7 @@ public class TestStreamWriteOperatorCoordinator {
final String metadataTableBasePath = HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
HoodieTableMetaClient metadataTableMetaClient = StreamerUtil.createMetaClient(metadataTableBasePath, HadoopConfigurations.getHadoopConf(conf));
HoodieTimeline completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(1L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP));
// writes a normal commit
@@ -298,7 +298,7 @@ public class TestStreamWriteOperatorCoordinator {
metadataTableMetaClient.reloadActiveTimeline();
completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
- assertThat("One instant need to sync to metadata table", completedTimeline.getInstants().count(), is(3L));
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(3));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(instant));
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
index 27aa48981e6..a2a0f985e6a 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -342,7 +342,7 @@ public class ITTestHoodieDataSource {
Configuration conf = Configuration.fromMap(options1);
HoodieTimeline timeline = StreamerUtil.createMetaClient(conf).getActiveTimeline();
assertTrue(timeline.filterCompletedInstants()
- .getInstants().anyMatch(instant -> instant.getAction().equals("clean")),
+ .getInstantsAsStream().anyMatch(instant -> instant.getAction().equals("clean")),
"some commits should be cleaned");
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
index e3392748ea7..4faff4a67c8 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
@@ -546,7 +546,7 @@ public class TestInputFormat {
}
HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(tempFile.getAbsolutePath(), HadoopConfigurations.getHadoopConf(conf));
- List<String> commits = metaClient.getCommitsTimeline().filterCompletedInstants().getInstants()
+ List<String> commits = metaClient.getCommitsTimeline().filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList());
assertThat(commits.size(), is(3));
@@ -663,13 +663,13 @@ public class TestInputFormat {
writeClient.clean();
HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(tempFile.getAbsolutePath(), HadoopConfigurations.getHadoopConf(conf));
- List<String> commits = metaClient.getCommitsTimeline().filterCompletedInstants().getInstants()
+ List<String> commits = metaClient.getCommitsTimeline().filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList());
assertThat(commits.size(), is(4));
List<String> archivedCommits = metaClient.getArchivedTimeline().getCommitsTimeline().filterCompletedInstants()
- .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+ .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
assertThat(archivedCommits.size(), is(6));
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java
index e8d28858548..68ac7114c54 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java
@@ -94,15 +94,14 @@ public class TestCompactionUtil {
List<HoodieInstant> instants = metaClient.getActiveTimeline()
.filterPendingCompactionTimeline()
.filter(instant -> instant.getState() == HoodieInstant.State.INFLIGHT)
- .getInstants()
- .collect(Collectors.toList());
+ .getInstants();
assertThat("all the instants should be in pending state", instants.size(), is(3));
CompactionUtil.rollbackCompaction(table);
- boolean allRolledBack = metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstants()
+ boolean allRolledBack = metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstantsAsStream()
.allMatch(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
assertTrue(allRolledBack, "all the instants should be rolled back");
List<String> actualInstants = metaClient.getActiveTimeline()
- .filterPendingCompactionTimeline().getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+ .filterPendingCompactionTimeline().getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
assertThat(actualInstants, is(oriInstants));
}
@@ -115,11 +114,10 @@ public class TestCompactionUtil {
List<HoodieInstant> instants = metaClient.getActiveTimeline()
.filterPendingCompactionTimeline()
.filter(instant -> instant.getState() == HoodieInstant.State.INFLIGHT)
- .getInstants()
- .collect(Collectors.toList());
+ .getInstants();
assertThat("all the instants should be in pending state", instants.size(), is(3));
CompactionUtil.rollbackEarliestCompaction(table, conf);
- long requestedCnt = metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstants()
+ long requestedCnt = metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstantsAsStream()
.filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED).count();
assertThat("Only the first instant expects to be rolled back", requestedCnt, is(1L));
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java
index b5c73037093..630ef398dfc 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java
@@ -148,7 +148,7 @@ public class HoodieMergeOnReadTableInputFormat extends HoodieCopyOnWriteTableInp
return result;
}
HoodieTimeline commitsTimelineToReturn = HoodieInputFormatUtils.getHoodieTimelineForIncrementalQuery(jobContext, incrementalTableName, timeline.get());
- Option<List<HoodieInstant>> commitsToCheck = Option.of(commitsTimelineToReturn.getInstants().collect(Collectors.toList()));
+ Option<List<HoodieInstant>> commitsToCheck = Option.of(commitsTimelineToReturn.getInstants());
if (!commitsToCheck.isPresent()) {
return result;
}
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
index 7b47ffa75f3..889446a1de5 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
@@ -283,8 +283,7 @@ public class HoodieInputFormatUtils {
* @return
*/
public static Option<List<HoodieInstant>> getCommitsForIncrementalQuery(Job job, String tableName, HoodieTimeline timeline) {
- return Option.of(getHoodieTimelineForIncrementalQuery(job, tableName, timeline)
- .getInstants().collect(Collectors.toList()));
+ return Option.of(getHoodieTimelineForIncrementalQuery(job, tableName, timeline).getInstants());
}
/**
diff --git a/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/ValidateAsyncOperations.java b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/ValidateAsyncOperations.java
index 714f3bf6cac..9c8dc4d82c7 100644
--- a/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/ValidateAsyncOperations.java
+++ b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/ValidateAsyncOperations.java
@@ -69,7 +69,7 @@ public class ValidateAsyncOperations extends DagNode<Option<String>> {
String earliestCommitToRetain = cleanMetadata.getEarliestCommitToRetain();
log.warn("Earliest commit to retain : " + earliestCommitToRetain);
long unCleanedInstants = metaClient.getActiveTimeline().filterCompletedInstants().filter(instant ->
- HoodieTimeline.compareTimestamps(instant.getTimestamp(), HoodieTimeline.GREATER_THAN_OR_EQUALS, earliestCommitToRetain)).getInstants().count();
+ HoodieTimeline.compareTimestamps(instant.getTimestamp(), HoodieTimeline.GREATER_THAN_OR_EQUALS, earliestCommitToRetain)).countInstants();
ValidationUtils.checkArgument(unCleanedInstants >= (maxCommitsRetained + 1), "Total uncleaned instants " + unCleanedInstants
+ " mismatched with max commits retained " + (maxCommitsRetained + 1));
}
diff --git a/hudi-integ-test/src/test/java/org/apache/hudi/integ/testsuite/job/TestHoodieTestSuiteJob.java b/hudi-integ-test/src/test/java/org/apache/hudi/integ/testsuite/job/TestHoodieTestSuiteJob.java
index 0ce0076f888..2bc86f2f8d0 100644
--- a/hudi-integ-test/src/test/java/org/apache/hudi/integ/testsuite/job/TestHoodieTestSuiteJob.java
+++ b/hudi-integ-test/src/test/java/org/apache/hudi/integ/testsuite/job/TestHoodieTestSuiteJob.java
@@ -212,7 +212,7 @@ public class TestHoodieTestSuiteJob extends UtilitiesTestBase {
HoodieTestSuiteJob hoodieTestSuiteJob = new HoodieTestSuiteJob(cfg, jsc);
hoodieTestSuiteJob.runTestSuite();
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(new Configuration()).setBasePath(cfg.targetBasePath).build();
- assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().getInstants().count(), 2);
+ assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().countInstants(), 2);
}
@Test
@@ -231,7 +231,7 @@ public class TestHoodieTestSuiteJob extends UtilitiesTestBase {
HoodieTestSuiteJob hoodieTestSuiteJob = new HoodieTestSuiteJob(cfg, jsc);
hoodieTestSuiteJob.runTestSuite();
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(new Configuration()).setBasePath(cfg.targetBasePath).build();
- assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().getInstants().count(), 1);
+ assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().countInstants(), 1);
}
@Test
@@ -246,7 +246,7 @@ public class TestHoodieTestSuiteJob extends UtilitiesTestBase {
HoodieTestSuiteJob hoodieTestSuiteJob = new HoodieTestSuiteJob(cfg, jsc);
hoodieTestSuiteJob.runTestSuite();
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(new Configuration()).setBasePath(cfg.targetBasePath).build();
- //assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().getInstants().count(), 5);
+ //assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().countInstants(), 5);
}
@Test
@@ -261,7 +261,7 @@ public class TestHoodieTestSuiteJob extends UtilitiesTestBase {
HoodieTestSuiteJob hoodieTestSuiteJob = new HoodieTestSuiteJob(cfg, jsc);
hoodieTestSuiteJob.runTestSuite();
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(new Configuration()).setBasePath(cfg.targetBasePath).build();
- //assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().getInstants().count(), 7);
+ //assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().countInstants(), 7);
}
@Test
@@ -282,7 +282,7 @@ public class TestHoodieTestSuiteJob extends UtilitiesTestBase {
HoodieTestSuiteJob hoodieTestSuiteJob = new HoodieTestSuiteJob(cfg, jsc);
hoodieTestSuiteJob.runTestSuite();
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(new Configuration()).setBasePath(cfg.targetBasePath).build();
- assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().getInstants().count(), 3);
+ assertEquals(metaClient.getActiveTimeline().getCommitsTimeline().countInstants(), 3);
}
@Test
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala
index 45bb52243b7..5641a622cbc 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala
@@ -113,7 +113,7 @@ class DefaultSource extends RelationProvider
def getValidCommits(metaClient: HoodieTableMetaClient): String = {
metaClient
- .getCommitsAndCompactionTimeline.filterCompletedInstants.getInstants.toArray().map(_.asInstanceOf[HoodieInstant].getFileName).mkString(",")
+ .getCommitsAndCompactionTimeline.filterCompletedInstants.getInstantsAsStream.toArray().map(_.asInstanceOf[HoodieInstant].getFileName).mkString(",")
}
/**
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala
index 260be7530bf..4c763b054ad 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala
@@ -83,7 +83,7 @@ class IncrementalRelation(val sqlContext: SQLContext,
private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
optParams(DataSourceReadOptions.BEGIN_INSTANTTIME.key),
optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME.key(), lastInstant.getTimestamp))
- private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val commitsToReturn = commitsTimelineToReturn.getInstantsAsStream.iterator().toList
// use schema from a file produced in the end/latest instant
@@ -131,7 +131,7 @@ class IncrementalRelation(val sqlContext: SQLContext,
// create Replaced file group
val replacedTimeline = commitsTimelineToReturn.getCompletedReplaceTimeline
- val replacedFile = replacedTimeline.getInstants.collect(Collectors.toList[HoodieInstant]).flatMap { instant =>
+ val replacedFile = replacedTimeline.getInstants.flatMap { instant =>
val replaceMetadata = HoodieReplaceCommitMetadata.
fromBytes(metaClient.getActiveTimeline.getInstantDetails(instant).get, classOf[HoodieReplaceCommitMetadata])
replaceMetadata.getPartitionToReplaceFileIds.entrySet().flatMap { entry =>
@@ -177,7 +177,7 @@ class IncrementalRelation(val sqlContext: SQLContext,
}
// pass internalSchema to hadoopConf, so it can be used in executors.
val validCommits = metaClient
- .getCommitsAndCompactionTimeline.filterCompletedInstants.getInstants.toArray().map(_.asInstanceOf[HoodieInstant].getFileName).mkString(",")
+ .getCommitsAndCompactionTimeline.filterCompletedInstants.getInstantsAsStream.toArray().map(_.asInstanceOf[HoodieInstant].getFileName).mkString(",")
sqlContext.sparkContext.hadoopConfiguration.set(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA, SerDeHelper.toJson(internalSchema))
sqlContext.sparkContext.hadoopConfiguration.set(SparkInternalSchemaConverter.HOODIE_TABLE_PATH, metaClient.getBasePath)
sqlContext.sparkContext.hadoopConfiguration.set(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST, validCommits)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
index 0b310c8e282..cefd3197805 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
@@ -150,9 +150,9 @@ trait HoodieIncrementalRelationTrait extends HoodieBaseRelation {
if (!startInstantArchived || !endInstantArchived) {
// If endTimestamp commit is not archived, will filter instants
// before endTimestamp.
- super.timeline.findInstantsInRange(startTimestamp, endTimestamp).getInstants.iterator().asScala.toList
+ super.timeline.findInstantsInRange(startTimestamp, endTimestamp).getInstants.asScala.toList
} else {
- super.timeline.getInstants.iterator().asScala.toList
+ super.timeline.getInstants.asScala.toList
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java b/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java
index 9491e43e21f..7ddf348c22b 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java
+++ b/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java
@@ -55,7 +55,7 @@ public class HoodieDataSourceHelpers {
@PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
public static List<String> listCommitsSince(FileSystem fs, String basePath, String instantTimestamp) {
HoodieTimeline timeline = allCompletedCommitsCompactions(fs, basePath);
- return timeline.findInstantsAfter(instantTimestamp, Integer.MAX_VALUE).getInstants()
+ return timeline.findInstantsAfter(instantTimestamp, Integer.MAX_VALUE).getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList());
}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/CommitsCompareProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/CommitsCompareProcedure.scala
index 9cb03bffc9e..379b38e5921 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/CommitsCompareProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/CommitsCompareProcedure.scala
@@ -55,9 +55,9 @@ class CommitsCompareProcedure() extends BaseProcedure with ProcedureBuilder {
val sourceTimeline = source.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
val targetTimeline = target.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
val targetLatestCommit =
- if (targetTimeline.getInstants.iterator.hasNext) targetTimeline.lastInstant.get.getTimestamp else "0"
+ if (targetTimeline.getInstants.iterator().hasNext) targetTimeline.lastInstant.get.getTimestamp else "0"
val sourceLatestCommit =
- if (sourceTimeline.getInstants.iterator.hasNext) sourceTimeline.lastInstant.get.getTimestamp else "0"
+ if (sourceTimeline.getInstants.iterator().hasNext) sourceTimeline.lastInstant.get.getTimestamp else "0"
if (sourceLatestCommit != null && HoodieTimeline.compareTimestamps(targetLatestCommit, HoodieTimeline.GREATER_THAN, sourceLatestCommit)) { // source is behind the target
val commitsToCatchup = targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE).getInstants.iterator().asScala.map(instant => instant.getTimestamp).toList.asJava
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
index bd2e863b1ea..3c51d7d8b29 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
@@ -118,7 +118,7 @@ class RunCompactionProcedure extends BaseProcedure with ProcedureBuilder with Sp
case _ => throw new UnsupportedOperationException(s"Unsupported compaction operation: $operation")
}
- val compactionInstants = metaClient.reloadActiveTimeline().getInstants.iterator().asScala
+ val compactionInstants = metaClient.reloadActiveTimeline().getInstantsAsStream.iterator().asScala
.filter(instant => willCompactionInstants.contains(instant.getTimestamp))
.toSeq
.sortBy(p => p.getTimestamp)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ValidateHoodieSyncProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ValidateHoodieSyncProcedure.scala
index 77dd4f3ee00..76c6b471740 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ValidateHoodieSyncProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ValidateHoodieSyncProcedure.scala
@@ -114,7 +114,7 @@ class ValidateHoodieSyncProcedure extends BaseProcedure with ProcedureBuilder wi
sourceCount: Long, targetCount: Long, sourceLatestCommit: String): String = {
val commitsToCatchup: List[HoodieInstant] =
- targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE).getInstants.iterator().asScala.toList
+ targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE).getInstantsAsStream.iterator().asScala.toList
if (commitsToCatchup.isEmpty) {
s"Count difference now is count(${target.getTableConfig.getTableName}) - count(${source.getTableConfig.getTableName}) == ${targetCount - sourceCount}"
} else {
diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java b/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java
index e716d34bd5e..3f26b3d036c 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java
+++ b/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java
@@ -51,7 +51,6 @@ import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import java.util.stream.Collectors;
import static org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_PASS;
@@ -198,9 +197,9 @@ public class HoodieJavaStreamingApp {
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(jssc.hadoopConfiguration()).setBasePath(tablePath).build();
if (tableType.equals(HoodieTableType.MERGE_ON_READ.name())) {
// Ensure we have successfully completed one compaction commit
- ValidationUtils.checkArgument(metaClient.getActiveTimeline().getCommitTimeline().getInstants().count() == 1);
+ ValidationUtils.checkArgument(metaClient.getActiveTimeline().getCommitTimeline().countInstants() == 1);
} else {
- ValidationUtils.checkArgument(metaClient.getActiveTimeline().getCommitTimeline().getInstants().count() >= 1);
+ ValidationUtils.checkArgument(metaClient.getActiveTimeline().getCommitTimeline().countInstants() >= 1);
}
// Deletes Stream
@@ -253,12 +252,12 @@ public class HoodieJavaStreamingApp {
while ((currTime - beginTime) < timeoutMsecs) {
try {
HoodieTimeline timeline = HoodieDataSourceHelpers.allCompletedCommitsCompactions(fs, tablePath);
- LOG.info("Timeline :" + timeline.getInstants().collect(Collectors.toList()));
+ LOG.info("Timeline :" + timeline.getInstants());
if (timeline.countInstants() >= numCommits) {
return;
}
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).setLoadActiveTimelineOnLoad(true).build();
- System.out.println("Instants :" + metaClient.getActiveTimeline().getInstants().collect(Collectors.toList()));
+ System.out.println("Instants :" + metaClient.getActiveTimeline().getInstants());
} catch (TableNotFoundException te) {
LOG.info("Got table not found exception. Retrying");
} finally {
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
index 8ba5db5e5ff..a82c7f658e6 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
@@ -312,7 +312,7 @@ class TestCOWDataSource extends HoodieClientTestBase {
val metaClient = HoodieTableMetaClient.builder().setConf(spark.sparkContext.hadoopConfiguration).setBasePath(basePath)
.setLoadActiveTimelineOnLoad(true).build()
- val instantTime = metaClient.getActiveTimeline.filterCompletedInstants().getInstants.findFirst().get().getTimestamp
+ val instantTime = metaClient.getActiveTimeline.filterCompletedInstants().getInstantsAsStream.findFirst().get().getTimestamp
val record1FilePaths = fs.listStatus(new Path(basePath, dataGen.getPartitionPaths.head))
.filter(!_.getPath.getName.contains("hoodie_partition_metadata"))
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
index fb0cf5e179d..b828a0626bb 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
@@ -105,7 +105,7 @@ class TestIncrementalReadWithFullTableScan extends HoodieClientTestBase {
val completedCommits = hoodieMetaClient.getCommitsTimeline.filterCompletedInstants() // C4 to C9
val archivedInstants = hoodieMetaClient.getArchivedTimeline.filterCompletedInstants()
- .getInstants.distinct().toArray // C0 to C3
+ .getInstantsAsStream.distinct().toArray // C0 to C3
//Anything less than 2 is a valid commit in the sense no cleanup has been done for those commit files
val startUnarchivedCommitTs = completedCommits.nthInstant(0).get().getTimestamp //C4
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairsProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairsProcedure.scala
index 86dbae5b1e8..52570596e17 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairsProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairsProcedure.scala
@@ -178,7 +178,7 @@ class TestRepairsProcedure extends HoodieSparkProcedureTestBase {
metaClient = HoodieTableMetaClient.reload(metaClient)
// first, there are four instants
assertResult(4) {
- metaClient.getActiveTimeline.filterInflightsAndRequested.getInstants.count
+ metaClient.getActiveTimeline.filterInflightsAndRequested.countInstants
}
checkAnswer(s"""call repair_corrupted_clean_files(table => '$tableName')""")(Seq(true))
@@ -187,7 +187,7 @@ class TestRepairsProcedure extends HoodieSparkProcedureTestBase {
metaClient = HoodieTableMetaClient.reload(metaClient)
// after clearing, there should be 0 instant
assertResult(0) {
- metaClient.getActiveTimeline.filterInflightsAndRequested.getInstants.count
+ metaClient.getActiveTimeline.filterInflightsAndRequested.getInstantsAsStream.count
}
}
}
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
index 1bdc87ab112..5e31f88d486 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
@@ -249,7 +249,7 @@ public class HoodieHiveSyncClient extends HoodieSyncClient {
}
public void updateLastReplicatedTimeStamp(String tableName, String timeStamp) {
- if (getActiveTimeline().getInstants().noneMatch(i -> i.getTimestamp().equals(timeStamp))) {
+ if (getActiveTimeline().getInstantsAsStream().noneMatch(i -> i.getTimestamp().equals(timeStamp))) {
throw new HoodieHiveSyncException(
"Not a valid completed timestamp " + timeStamp + " for table " + tableName);
}
diff --git a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index acd75595fb8..98c40c3e0bc 100644
--- a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++ b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -1044,7 +1044,7 @@ public class TestHiveSyncTool {
HiveSyncTool tool = new HiveSyncTool(hiveSyncProps, getHiveConf());
// now delete the evolved commit instant
Path fullPath = new Path(HiveTestUtil.basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
- + hiveClient.getActiveTimeline().getInstants()
+ + hiveClient.getActiveTimeline().getInstantsAsStream()
.filter(inst -> inst.getTimestamp().equals(commitTime2))
.findFirst().get().getFileName());
assertTrue(HiveTestUtil.fileSystem.delete(fullPath, false));
@@ -1088,7 +1088,7 @@ public class TestHiveSyncTool {
reinitHiveSyncClient();
// now delete the evolved commit instant
Path fullPath = new Path(HiveTestUtil.basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
- + hiveClient.getActiveTimeline().getInstants()
+ + hiveClient.getActiveTimeline().getInstantsAsStream()
.filter(inst -> inst.getTimestamp().equals(commitTime2))
.findFirst().get().getFileName());
assertTrue(HiveTestUtil.fileSystem.delete(fullPath, false));
diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
index ff334113946..590b8cacd9d 100644
--- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
+++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
@@ -60,7 +60,6 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.stream.Collectors;
/**
* Main REST Handler class that handles and delegates calls to timeline relevant handlers.
@@ -159,10 +158,10 @@ public class RequestHandler {
viewManager.getFileSystemView(basePath).getTimeline().filterCompletedAndCompactionInstants();
if (LOG.isDebugEnabled()) {
LOG.debug("Client [ LastTs=" + lastKnownInstantFromClient + ", TimelineHash=" + timelineHashFromClient
- + "], localTimeline=" + localTimeline.getInstants().collect(Collectors.toList()));
+ + "], localTimeline=" + localTimeline.getInstants());
}
- if ((!localTimeline.getInstants().findAny().isPresent())
+ if ((!localTimeline.getInstantsAsStream().findAny().isPresent())
&& HoodieTimeline.INVALID_INSTANT_TS.equals(lastKnownInstantFromClient)) {
return false;
}
@@ -522,7 +521,7 @@ public class RequestHandler {
"Last known instant from client was "
+ lastKnownInstantFromClient
+ " but server has the following timeline "
- + localTimeline.getInstants().collect(Collectors.toList());
+ + localTimeline.getInstants();
throw new BadRequestResponse(errMsg);
}
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HiveIncrementalPuller.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HiveIncrementalPuller.java
index 2e66a2275af..52e2ce55a46 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HiveIncrementalPuller.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HiveIncrementalPuller.java
@@ -315,13 +315,13 @@ public class HiveIncrementalPuller {
private String getLastCommitTimePulled(FileSystem fs, String sourceTableLocation) {
HoodieTableMetaClient metadata = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(sourceTableLocation).build();
List<String> commitsToSync = metadata.getActiveTimeline().getCommitsTimeline().filterCompletedInstants()
- .findInstantsAfter(config.fromCommitTime, config.maxCommits).getInstants().map(HoodieInstant::getTimestamp)
+ .findInstantsAfter(config.fromCommitTime, config.maxCommits).getInstantsAsStream().map(HoodieInstant::getTimestamp)
.collect(Collectors.toList());
if (commitsToSync.isEmpty()) {
LOG.warn(
"Nothing to sync. All commits in "
+ config.sourceTable + " are " + metadata.getActiveTimeline().getCommitsTimeline()
- .filterCompletedInstants().getInstants().collect(Collectors.toList())
+ .filterCompletedInstants().getInstants()
+ " and from commit time is " + config.fromCommitTime);
return null;
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java
index ef05bdc03ad..b5924deeed4 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java
@@ -321,7 +321,7 @@ public class HoodieDataTableValidator implements Serializable {
Map<String, List<String>> instantToFilesMap = RepairUtils.tagInstantsOfBaseAndLogFiles(
metaClient.getBasePath(), allDataFilePaths);
HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
- List<HoodieInstant> hoodieInstants = activeTimeline.filterCompletedInstants().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> hoodieInstants = activeTimeline.filterCompletedInstants().getInstants();
List<String> danglingFiles = engineContext.flatMap(hoodieInstants, instant -> {
Option<Set<String>> filesFromTimeline = RepairUtils.getBaseAndLogFilePathsFromTimeline(
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
index 9d1b0cbd873..6566f0c029a 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
@@ -407,7 +407,7 @@ public class HoodieMetadataTableValidator implements Serializable {
if (cfg.skipDataFilesForCleaning) {
HoodieTimeline inflightCleaningTimeline = metaClient.getActiveTimeline().getCleanerTimeline().filterInflights();
- baseFilesForCleaning = inflightCleaningTimeline.getInstants().flatMap(instant -> {
+ baseFilesForCleaning = inflightCleaningTimeline.getInstantsAsStream().flatMap(instant -> {
try {
// convert inflight instant to requested and get clean plan
instant = new HoodieInstant(HoodieInstant.State.REQUESTED, instant.getAction(), instant.getTimestamp());
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
index df11e693d68..5b17a9f85be 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
@@ -282,7 +282,7 @@ public class DeltaSync implements Serializable {
initializeEmptyTable();
// reload the timeline from metaClient and validate that its empty table. If there are any instants found, then we should fail the pipeline, bcoz hoodie.properties got deleted by mistake.
HoodieTableMetaClient metaClientToValidate = HoodieTableMetaClient.builder().setConf(new Configuration(fs.getConf())).setBasePath(cfg.targetBasePath).build();
- if (metaClientToValidate.reloadActiveTimeline().getInstants().count() > 0) {
+ if (metaClientToValidate.reloadActiveTimeline().countInstants() > 0) {
// Deleting the recreated hoodie.properties and throwing exception.
fs.delete(new Path(String.format("%s%s/%s", basePathWithForwardSlash, HoodieTableMetaClient.METAFOLDER_NAME, HoodieTableConfig.HOODIE_PROPERTIES_FILE)));
throw new HoodieIOException("hoodie.properties is missing. Likely due to some external entity. Please populate the hoodie.properties and restart the pipeline. ",
@@ -557,7 +557,7 @@ public class DeltaSync implements Serializable {
throw new HoodieDeltaStreamerException(
"Unable to find previous checkpoint. Please double check if this table "
+ "was indeed built via delta streamer. Last Commit :" + lastCommit + ", Instants :"
- + commitTimelineOpt.get().getInstants().collect(Collectors.toList()) + ", CommitMetadata="
+ + commitTimelineOpt.get().getInstants() + ", CommitMetadata="
+ commitMetadata.toJsonString());
}
// KAFKA_CHECKPOINT_TYPE will be honored only for first batch.
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
index eca3439e62f..67ab0f5b2ba 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
@@ -106,7 +106,7 @@ public class IncrSourceHelper {
if (missingCheckpointStrategy == MissingCheckpointStrategy.READ_LATEST || !activeCommitTimeline.isBeforeTimelineStarts(beginInstantTime)) {
Option<HoodieInstant> nthInstant = Option.fromJavaOptional(activeCommitTimeline
- .findInstantsAfter(beginInstantTime, numInstantsPerFetch).getInstants().reduce((x, y) -> y));
+ .findInstantsAfter(beginInstantTime, numInstantsPerFetch).getInstantsAsStream().reduce((x, y) -> y));
return Pair.of(DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL(), Pair.of(beginInstantTime, nthInstant.map(HoodieInstant::getTimestamp).orElse(beginInstantTime)));
} else {
// when MissingCheckpointStrategy is set to read everything until latest, trigger snapshot query.
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamer.java
index 538f194ad97..ab0d8b0ba32 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamer.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamer.java
@@ -338,32 +338,32 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
static void assertAtleastNCompactionCommits(int minExpected, String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).build();
HoodieTimeline timeline = meta.getActiveTimeline().getCommitTimeline().filterCompletedInstants();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numCompactionCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numCompactionCommits = timeline.countInstants();
assertTrue(minExpected <= numCompactionCommits, "Got=" + numCompactionCommits + ", exp >=" + minExpected);
}
static void assertAtleastNDeltaCommits(int minExpected, String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).build();
HoodieTimeline timeline = meta.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numDeltaCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numDeltaCommits = timeline.countInstants();
assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ", exp >=" + minExpected);
}
static void assertAtleastNCompactionCommitsAfterCommit(int minExpected, String lastSuccessfulCommit, String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).build();
HoodieTimeline timeline = meta.getActiveTimeline().getCommitTimeline().findInstantsAfter(lastSuccessfulCommit).filterCompletedInstants();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numCompactionCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numCompactionCommits = timeline.countInstants();
assertTrue(minExpected <= numCompactionCommits, "Got=" + numCompactionCommits + ", exp >=" + minExpected);
}
static void assertAtleastNDeltaCommitsAfterCommit(int minExpected, String lastSuccessfulCommit, String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).build();
HoodieTimeline timeline = meta.reloadActiveTimeline().getDeltaCommitTimeline().findInstantsAfter(lastSuccessfulCommit).filterCompletedInstants();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numDeltaCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numDeltaCommits = timeline.countInstants();
assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ", exp >=" + minExpected);
}
@@ -399,48 +399,48 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
static void assertAtLeastNCommits(int minExpected, String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).build();
HoodieTimeline timeline = meta.getActiveTimeline().filterCompletedInstants();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numDeltaCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numDeltaCommits = timeline.countInstants();
assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ", exp >=" + minExpected);
}
static void assertAtLeastNReplaceCommits(int minExpected, String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).setLoadActiveTimelineOnLoad(true).build();
HoodieTimeline timeline = meta.getActiveTimeline().getCompletedReplaceTimeline();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numDeltaCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numDeltaCommits = timeline.countInstants();
assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ", exp >=" + minExpected);
}
static void assertPendingIndexCommit(String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).setLoadActiveTimelineOnLoad(true).build();
HoodieTimeline timeline = meta.getActiveTimeline().getAllCommitsTimeline().filterPendingIndexTimeline();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numIndexCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numIndexCommits = timeline.countInstants();
assertEquals(1, numIndexCommits, "Got=" + numIndexCommits + ", exp=1");
}
static void assertCompletedIndexCommit(String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).setLoadActiveTimelineOnLoad(true).build();
HoodieTimeline timeline = meta.getActiveTimeline().getAllCommitsTimeline().filterCompletedIndexTimeline();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numIndexCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numIndexCommits = timeline.countInstants();
assertEquals(1, numIndexCommits, "Got=" + numIndexCommits + ", exp=1");
}
static void assertNoReplaceCommits(String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).setLoadActiveTimelineOnLoad(true).build();
HoodieTimeline timeline = meta.getActiveTimeline().getCompletedReplaceTimeline();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numDeltaCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numDeltaCommits = timeline.countInstants();
assertEquals(0, numDeltaCommits, "Got=" + numDeltaCommits + ", exp =" + 0);
}
static void assertAtLeastNReplaceRequests(int minExpected, String tablePath, FileSystem fs) {
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tablePath).setLoadActiveTimelineOnLoad(true).build();
HoodieTimeline timeline = meta.getActiveTimeline().filterPendingReplaceTimeline();
- LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants().collect(Collectors.toList()));
- int numDeltaCommits = (int) timeline.getInstants().count();
+ LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
+ int numDeltaCommits = timeline.countInstants();
assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ", exp >=" + minExpected);
}
}
@@ -936,7 +936,7 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
HoodieClusteringJob clusteringJob = initialHoodieClusteringJob(tableBasePath, null, false, "schedule");
clusteringJob.cluster(0);
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tableBasePath).build();
- List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingReplaceTimeline().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingReplaceTimeline().getInstants();
HoodieInstant clusteringRequest = hoodieClusteringInstants.get(0);
meta.getActiveTimeline().transitionReplaceRequestedToInflight(clusteringRequest, Option.empty());
@@ -1034,7 +1034,7 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
ds.sync();
// Step 5 : Make sure that firstReplaceHoodieInstant is archived.
- long count = meta.reloadActiveTimeline().getCompletedReplaceTimeline().getInstants().filter(instant -> firstReplaceHoodieInstant.get().equals(instant)).count();
+ long count = meta.reloadActiveTimeline().getCompletedReplaceTimeline().getInstantsAsStream().filter(instant -> firstReplaceHoodieInstant.get().equals(instant)).count();
assertEquals(0, count);
// Step 6 : All the replaced files in firstReplaceHoodieInstant should be deleted through sync/async cleaner.
@@ -1282,7 +1282,7 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
// convert clustering request into inflight, Simulate the last clustering failed scenario
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(tableBasePath).build();
- List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingReplaceTimeline().getInstants().collect(Collectors.toList());
+ List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingReplaceTimeline().getInstants();
HoodieInstant clusteringRequest = hoodieClusteringInstants.get(0);
HoodieInstant hoodieInflightInstant = meta.getActiveTimeline().transitionReplaceRequestedToInflight(clusteringRequest, Option.empty());
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamerWithMultiWriter.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamerWithMultiWriter.java
index abe1994a3dd..389c687619d 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamerWithMultiWriter.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamerWithMultiWriter.java
@@ -428,8 +428,7 @@ public class TestHoodieDeltaStreamerWithMultiWriter extends SparkClientFunctiona
long getCommitsAfterInstant() {
HoodieTimeline timeline1 = meta.reloadActiveTimeline().getAllCommitsTimeline().findInstantsAfter(lastSuccessfulCommit);
- // LOG.info("Timeline Instants=" + meta1.getActiveTimeline().getInstants().collect(Collectors.toList()));
- return timeline1.getInstants().count();
+ return timeline1.countInstants();
}
}