This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 634110752b2f test(cli): cover timeline, export, metadata commands and
the SparkMain helpers (#19877)
634110752b2f is described below
commit 634110752b2f0bd1fbcba265bb3e4f3252368269
Author: voonhous <[email protected]>
AuthorDate: Thu Sep 10 16:56:41 2026 +0800
test(cli): cover timeline, export, metadata commands and the SparkMain
helpers (#19877)
Second round of hudi-cli coverage after #18816 wired the module into
CI. TimelineCommand, ExportCommand and most of MetadataCommand had no
tests, and the SparkMain helpers were only reached through the
spark-submit child process the shell commands launch, which JaCoCo
never sees.
- TestTimelineCommand (new, 8): timeline show active / incomplete and
metadata timeline show active / incomplete with every option; the
fixture holds completed, requested and rolled-back commits plus a
pending rollback so both rollback annotations render.
- TestExportCommand (new, 2): export instants, decoding the exported
files through the table serde and asserting the written partitions,
and the invalid-folder guard.
- TestMetadataCommand (+4): stats, list-partitions, list-files with
and without a partition, validate-files in both verbose arms with a
planted stray base file, create and init, set.
- TestCompactionCommand (+6): compaction validate on a healthy plan
and on one with a deleted log file, repair, unschedule plan and
unschedule file with and without dryRun, and schedule-and-execute
compaction on a MOR table with log files.
- TestClusteringCommand (new, 1), TestRollbacksCommand (+2) and
TestSavepointsCommand (+1): the cluster, rollback and savepoint
entry points, including the failure return codes.
- CLIFunctionalTestHarness gains renderedRows, a parser for
HoodiePrintHelper tables so tests assert on cell values, and
createTableAndConnect, which skips the five-second existing-table
probe in TableCommand.createTable.
The only production change is visibility: ten SparkMain helpers go
from private static to package-private @VisibleForTesting static,
matching the existing archive helper, so they can be called
in-process. No behaviour change.
---
.../org/apache/hudi/cli/commands/SparkMain.java | 46 +--
.../hudi/cli/commands/TestClusteringCommand.java | 106 +++++++
.../hudi/cli/commands/TestCompactionCommand.java | 232 +++++++++++++++
.../hudi/cli/commands/TestExportCommand.java | 169 +++++++++++
.../hudi/cli/commands/TestMetadataCommand.java | 229 +++++++++++++++
.../hudi/cli/commands/TestRollbacksCommand.java | 39 ++-
.../hudi/cli/commands/TestSavepointsCommand.java | 62 ++++
.../hudi/cli/commands/TestTimelineCommand.java | 316 +++++++++++++++++++++
.../cli/functional/CLIFunctionalTestHarness.java | 99 +++++++
9 files changed, 1279 insertions(+), 19 deletions(-)
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
index 121c1f8b4231..8a64538eadc2 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
@@ -286,8 +286,9 @@ public class SparkMain {
}
}
- private static void doCompactValidate(JavaSparkContext jsc, String basePath,
String compactionInstant,
- String outputPath, int parallelism)
throws Exception {
+ @VisibleForTesting
+ static void doCompactValidate(JavaSparkContext jsc, String basePath, String
compactionInstant,
+ String outputPath, int parallelism) throws
Exception {
HoodieCompactionAdminTool.Config cfg = new
HoodieCompactionAdminTool.Config();
cfg.basePath = basePath;
cfg.operation = Operation.VALIDATE;
@@ -297,8 +298,9 @@ public class SparkMain {
new HoodieCompactionAdminTool(cfg).run(jsc);
}
- private static void doCompactRepair(JavaSparkContext jsc, String basePath,
String compactionInstant,
- String outputPath, int parallelism,
boolean dryRun) throws Exception {
+ @VisibleForTesting
+ static void doCompactRepair(JavaSparkContext jsc, String basePath, String
compactionInstant,
+ String outputPath, int parallelism, boolean
dryRun) throws Exception {
HoodieCompactionAdminTool.Config cfg = new
HoodieCompactionAdminTool.Config();
cfg.basePath = basePath;
cfg.operation = Operation.REPAIR;
@@ -309,8 +311,9 @@ public class SparkMain {
new HoodieCompactionAdminTool(cfg).run(jsc);
}
- private static void doCompactUnschedule(JavaSparkContext jsc, String
basePath, String compactionInstant,
- String outputPath, int parallelism,
boolean skipValidation, boolean dryRun) throws Exception {
+ @VisibleForTesting
+ static void doCompactUnschedule(JavaSparkContext jsc, String basePath,
String compactionInstant,
+ String outputPath, int parallelism, boolean
skipValidation, boolean dryRun) throws Exception {
HoodieCompactionAdminTool.Config cfg = new
HoodieCompactionAdminTool.Config();
cfg.basePath = basePath;
cfg.operation = Operation.UNSCHEDULE_PLAN;
@@ -322,8 +325,9 @@ public class SparkMain {
new HoodieCompactionAdminTool(cfg).run(jsc);
}
- private static void doCompactUnscheduleFile(JavaSparkContext jsc, String
basePath, String fileId, String partitionPath,
- String outputPath, int
parallelism, boolean skipValidation, boolean dryRun)
+ @VisibleForTesting
+ static void doCompactUnscheduleFile(JavaSparkContext jsc, String basePath,
String fileId, String partitionPath,
+ String outputPath, int parallelism,
boolean skipValidation, boolean dryRun)
throws Exception {
HoodieCompactionAdminTool.Config cfg = new
HoodieCompactionAdminTool.Config();
cfg.basePath = basePath;
@@ -337,9 +341,10 @@ public class SparkMain {
new HoodieCompactionAdminTool(cfg).run(jsc);
}
- private static int compact(JavaSparkContext jsc, String basePath, String
tableName, String compactionInstant,
- int parallelism, String schemaFile, int retry,
String mode, String propsFilePath,
- List<String> configs) {
+ @VisibleForTesting
+ static int compact(JavaSparkContext jsc, String basePath, String tableName,
String compactionInstant,
+ int parallelism, String schemaFile, int retry, String
mode, String propsFilePath,
+ List<String> configs) {
HoodieCompactor.Config cfg = new HoodieCompactor.Config();
cfg.basePath = basePath;
cfg.tableName = tableName;
@@ -354,8 +359,9 @@ public class SparkMain {
return new HoodieCompactor(jsc, cfg).compact(retry);
}
- private static int cluster(JavaSparkContext jsc, String basePath, String
tableName, String clusteringInstant,
- int parallelism, String sparkMemory, int retry,
String runningMode, String propsFilePath, List<String> configs) {
+ @VisibleForTesting
+ static int cluster(JavaSparkContext jsc, String basePath, String tableName,
String clusteringInstant,
+ int parallelism, String sparkMemory, int retry, String
runningMode, String propsFilePath, List<String> configs) {
HoodieClusteringJob.Config cfg = new HoodieClusteringJob.Config();
cfg.basePath = basePath;
cfg.tableName = tableName;
@@ -504,7 +510,8 @@ public class SparkMain {
return 0;
}
- private static int rollback(JavaSparkContext jsc, String instantTime, String
basePath, Boolean rollbackUsingMarkers) throws Exception {
+ @VisibleForTesting
+ static int rollback(JavaSparkContext jsc, String instantTime, String
basePath, Boolean rollbackUsingMarkers) throws Exception {
SparkRDDWriteClient client = createHoodieClient(jsc, basePath,
rollbackUsingMarkers, false);
if (client.rollback(instantTime)) {
log.info("The commit \"{}\" rolled back.", instantTime);
@@ -515,8 +522,9 @@ public class SparkMain {
}
}
- private static int createSavepoint(JavaSparkContext jsc, String commitTime,
String user,
- String comments, String basePath) throws
Exception {
+ @VisibleForTesting
+ static int createSavepoint(JavaSparkContext jsc, String commitTime, String
user,
+ String comments, String basePath) throws
Exception {
try (SparkRDDWriteClient client = createHoodieClient(jsc, basePath,
false)) {
client.savepoint(commitTime, user, comments);
log.info("The commit \"{}\" has been savepointed.", commitTime);
@@ -527,7 +535,8 @@ public class SparkMain {
}
}
- private static int rollbackToSavepoint(JavaSparkContext jsc, String
savepointTime, String basePath, boolean lazyCleanPolicy) throws Exception {
+ @VisibleForTesting
+ static int rollbackToSavepoint(JavaSparkContext jsc, String savepointTime,
String basePath, boolean lazyCleanPolicy) throws Exception {
try (SparkRDDWriteClient client = createHoodieClient(jsc, basePath,
lazyCleanPolicy)) {
client.restoreToSavepoint(savepointTime);
log.info("The commit \"{}\" rolled back.", savepointTime);
@@ -538,7 +547,8 @@ public class SparkMain {
}
}
- private static int deleteSavepoint(JavaSparkContext jsc, String
savepointTime, String basePath) throws Exception {
+ @VisibleForTesting
+ static int deleteSavepoint(JavaSparkContext jsc, String savepointTime,
String basePath) throws Exception {
try (SparkRDDWriteClient client = createHoodieClient(jsc, basePath,
false)) {
client.deleteSavepoint(savepointTime);
log.info("Savepoint \"{}\" deleted.", savepointTime);
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestClusteringCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestClusteringCommand.java
new file mode 100644
index 000000000000..5f8014fedf8d
--- /dev/null
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestClusteringCommand.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.cli.commands;
+
+import org.apache.hudi.cli.HoodieCLI;
+import org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.testutils.Assertions;
+import org.apache.hudi.utilities.UtilHelpers;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test cases for the clustering entry point of {@link SparkMain}, which the
clustering commands
+ * reach through a spark-submit of their own.
+ */
+@Tag("functional")
+@SpringBootTest(properties = {"spring.shell.interactive.enabled=false",
"spring.shell.command.script.enabled=false"})
+public class TestClusteringCommand extends CLIFunctionalTestHarness {
+
+ private String tableName;
+ private String tablePath;
+
+ @BeforeEach
+ public void init() throws IOException {
+ HoodieCLI.conf = storageConf();
+ tableName = tableName();
+ tablePath = tablePath(tableName);
+
+ createTableAndConnect(tablePath, tableName, HoodieTableType.COPY_ON_WRITE,
HoodieAvroPayload.class.getName());
+ }
+
+ @Test
+ public void testSparkMainClusterScheduleAndExecute() throws Exception {
+ writeCommits();
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
+ assertEquals(0,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
+
+ int returnCode = SparkMain.cluster(jsc(), tablePath, tableName, null, 1,
"1g", 0,
+ UtilHelpers.SCHEDULE_AND_EXECUTE, null, Collections.emptyList());
+
+ assertEquals(0, returnCode);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ // the plan it scheduled ran to completion, leaving a replace commit and
nothing pending
+ assertEquals(0,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
+ assertEquals(1,
metaClient.getActiveTimeline().getCompletedReplaceTimeline().countInstants());
+ }
+
+ /**
+ * Writes two commits of small files into the table, which is what the
clustering plan strategy
+ * looks for.
+ */
+ private void writeCommits() {
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(new String[]
{DEFAULT_FIRST_PARTITION_PATH});
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath(tablePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA).withParallelism(1,
1).forTable(tableName).build();
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(context(),
config)) {
+ String firstCommit = client.startCommit();
+ writeAndCommit(client, dataGen.generateInserts(firstCommit, 10),
firstCommit);
+ String secondCommit = client.startCommit();
+ writeAndCommit(client, dataGen.generateInserts(secondCommit, 10),
secondCommit);
+ }
+ }
+
+ private void writeAndCommit(SparkRDDWriteClient client, List<HoodieRecord>
records, String commitTime) {
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+ List<WriteStatus> result = client.insert(writeRecords,
commitTime).collect();
+ client.commit(commitTime, jsc().parallelize(result));
+ Assertions.assertNoWriteErrors(result);
+ }
+}
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCompactionCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCompactionCommand.java
index d25abe1789ad..3e849367ebe0 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCompactionCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCompactionCommand.java
@@ -18,16 +18,22 @@
package org.apache.hudi.cli.commands;
+import org.apache.hudi.avro.model.HoodieCompactionOperation;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
import org.apache.hudi.cli.HoodieCLI;
import org.apache.hudi.cli.HoodiePrintHelper;
import org.apache.hudi.cli.TableHeader;
import org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
import org.apache.hudi.cli.testutils.HoodieTestCommitMetadataGenerator;
+import org.apache.hudi.client.CompactionAdminClient.ValidationOpResult;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.timeline.HoodieTimelineArchiver;
import org.apache.hudi.client.timeline.TimelineArchiverV2;
import org.apache.hudi.common.model.HoodieAvroPayload;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
@@ -35,35 +41,53 @@ import
org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.testutils.CompactionTestUtils;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieArchivalConfig;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.testutils.Assertions;
+import org.apache.hudi.utilities.UtilHelpers;
+import org.apache.spark.api.java.JavaRDD;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.shell.Shell;
import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Test Cases for {@link CompactionCommand}.
@@ -75,6 +99,8 @@ public class TestCompactionCommand extends
CLIFunctionalTestHarness {
@Autowired
private Shell shell;
+ private static final String PENDING_COMPACTION_INSTANT = "001";
+
private String tableName;
private String tablePath;
@@ -146,6 +172,212 @@ public class TestCompactionCommand extends
CLIFunctionalTestHarness {
assertNotNull(result);
}
+ /**
+ * Test case of the compaction validation entry point of {@link SparkMain},
which the
+ * 'compaction validate' command reaches through a spark-submit of its own.
+ */
+ @Test
+ public void testSparkMainCompactValidate() throws Exception {
+ createPendingCompactions();
+ String outputPath = outputPath("validate");
+
+ SparkMain.doCompactValidate(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2);
+
+ List<ValidationOpResult> results = readOperationResults(outputPath);
+ assertEquals(operationsOf(PENDING_COMPACTION_INSTANT).size(),
results.size());
+ assertTrue(results.stream().allMatch(ValidationOpResult::isSuccess),
results.toString());
+ assertEquals(fileIdsOf(PENDING_COMPACTION_INSTANT),
+ results.stream().map(result ->
result.getOperation().getFileId()).collect(Collectors.toSet()));
+ }
+
+ @Test
+ public void testSparkMainCompactValidateReportsMissingLogFile() throws
Exception {
+ createPendingCompactions();
+ HoodieCompactionOperation broken =
operationsOf(PENDING_COMPACTION_INSTANT).get(0);
+ // a log file the plan reads is gone, so that operation can no longer be
compacted
+ Files.delete(Paths.get(tablePath, broken.getPartitionPath(),
broken.getDeltaFilePaths().get(0)));
+ String outputPath = outputPath("validate-broken");
+
+ SparkMain.doCompactValidate(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2);
+
+ List<ValidationOpResult> results = readOperationResults(outputPath);
+ assertEquals(operationsOf(PENDING_COMPACTION_INSTANT).size(),
results.size());
+ List<ValidationOpResult> failed = results.stream().filter(result ->
!result.isSuccess()).collect(Collectors.toList());
+ assertEquals(1, failed.size(), results.toString());
+ assertEquals(broken.getFileId(), failed.get(0).getOperation().getFileId());
+ assertTrue(failed.get(0).getException().isPresent());
+ }
+
+ /**
+ * Repair runs the plan validation and returns an empty result: the log file
renaming it was
+ * written for is gone from the admin client, which leaves the plan
untouched and never reads
+ * the dry run flag, so there is only one arm to exercise. See
+ * https://github.com/apache/hudi/issues/19881.
+ */
+ @Test
+ public void testSparkMainCompactRepair() throws Exception {
+ createPendingCompactions();
+ Set<String> fileIdsBefore = fileIdsOf(PENDING_COMPACTION_INSTANT);
+ String outputPath = outputPath("repair");
+
+ SparkMain.doCompactRepair(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2, false);
+
+ assertTrue(readOperationResults(outputPath).isEmpty());
+
assertTrue(pendingCompactionInstants().contains(PENDING_COMPACTION_INSTANT));
+ assertEquals(fileIdsBefore, fileIdsOf(PENDING_COMPACTION_INSTANT));
+ }
+
+ /**
+ * Unscheduling a plan takes the requested compaction instant off the
timeline, unless this is a
+ * dry run. The other pending plans are left alone either way. Skip
validation is held at false:
+ * the admin client takes the flag but never reads it, so toggling it
repeats the same run.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSparkMainCompactUnschedulePlan(boolean dryRun) throws
Exception {
+ createPendingCompactions();
+ Set<String> pendingBefore = pendingCompactionInstants();
+ String outputPath = outputPath("unschedule-" + dryRun);
+
+ SparkMain.doCompactUnschedule(jsc(), tablePath,
PENDING_COMPACTION_INSTANT, outputPath, 2, false, dryRun);
+
+ assertTrue(readOperationResults(outputPath).isEmpty());
+ Set<String> pendingAfter = pendingCompactionInstants();
+ if (dryRun) {
+ assertEquals(pendingBefore, pendingAfter);
+ } else {
+ assertFalse(pendingAfter.contains(PENDING_COMPACTION_INSTANT),
pendingAfter.toString());
+ pendingBefore.remove(PENDING_COMPACTION_INSTANT);
+ assertEquals(pendingBefore, pendingAfter);
+ }
+ }
+
+ /**
+ * Unscheduling a single file group rewrites the plan without it, unless
this is a dry run. Skip
+ * validation is held at false: the admin client takes the flag but never
reads it, so toggling
+ * it repeats the same run.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSparkMainCompactUnscheduleFile(boolean dryRun) throws
Exception {
+ Map<HoodieFileGroupId, Pair<String, HoodieCompactionOperation>>
pendingOperations = createPendingCompactions();
+ HoodieFileGroupId unscheduled = pendingOperations.entrySet().stream()
+ .filter(entry ->
entry.getValue().getKey().equals(PENDING_COMPACTION_INSTANT))
+ .map(Map.Entry::getKey).findFirst().get();
+ String outputPath = outputPath("unschedule-file-" + dryRun);
+ // both operations of this plan sit in the same partition, so there is a
sibling to keep
+ Set<String> fileIdsBefore = fileIdsOf(PENDING_COMPACTION_INSTANT);
+ assertEquals(2, fileIdsBefore.size());
+ assertTrue(fileIdsBefore.contains(unscheduled.getFileId()));
+
+ SparkMain.doCompactUnscheduleFile(jsc(), tablePath,
unscheduled.getFileId(), unscheduled.getPartitionPath(),
+ outputPath, 2, false, dryRun);
+
+ assertTrue(readOperationResults(outputPath).isEmpty());
+ // the plan itself stays pending either way, only its operations change
+
assertTrue(pendingCompactionInstants().contains(PENDING_COMPACTION_INSTANT));
+ if (dryRun) {
+ assertEquals(fileIdsBefore, fileIdsOf(PENDING_COMPACTION_INSTANT));
+ } else {
+ // The admin client keeps the operations that differ from the
unscheduled one in file id AND
+ // in partition path, so the sibling operation goes with it and the plan
is left with no
+ // operations at all (https://github.com/apache/hudi/issues/19881). When
that is fixed this
+ // expectation has to become the sibling on its own:
+ // fileIdsBefore minus the unscheduled file id.
+ assertEquals(Collections.emptySet(),
fileIdsOf(PENDING_COMPACTION_INSTANT));
+ }
+ }
+
+ /**
+ * A MOR table with four pending compaction plans, of which {@link
#PENDING_COMPACTION_INSTANT}
+ * holds more than one operation.
+ *
+ * @return The pending compaction operations, by file group.
+ */
+ private Map<HoodieFileGroupId, Pair<String, HoodieCompactionOperation>>
createPendingCompactions() throws IOException {
+ createTableAndConnect(tablePath, tableName, HoodieTableType.MERGE_ON_READ,
HoodieAvroPayload.class.getName());
+ Map<HoodieFileGroupId, Pair<String, HoodieCompactionOperation>> operations
=
+
CompactionTestUtils.setupAndValidateCompactionOperations(HoodieCLI.getTableMetaClient(),
false, 2, 1, 1, 1);
+ HoodieCLI.getTableMetaClient().reloadActiveTimeline();
+ return operations;
+ }
+
+ private String outputPath(String name) {
+ return Paths.get(basePath(), "compaction-admin-" + name).toString();
+ }
+
+ private Set<String> pendingCompactionInstants() {
+ return
HoodieCLI.getTableMetaClient().reloadActiveTimeline().filterPendingCompactionTimeline()
+
.getInstantsAsStream().map(HoodieInstant::requestedTime).collect(Collectors.toSet());
+ }
+
+ private List<HoodieCompactionOperation> operationsOf(String
compactionInstant) throws IOException {
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ metaClient.reloadActiveTimeline();
+ return CompactionUtils.getCompactionPlan(metaClient,
compactionInstant).getOperations();
+ }
+
+ private Set<String> fileIdsOf(String compactionInstant) throws IOException {
+ return operationsOf(compactionInstant).stream()
+ .map(HoodieCompactionOperation::getFileId).collect(Collectors.toSet());
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T> List<T> readOperationResults(String outputPath) throws Exception
{
+ try (ObjectInputStream in = new
ObjectInputStream(Files.newInputStream(Paths.get(outputPath)))) {
+ return (List<T>) in.readObject();
+ }
+ }
+
+ /**
+ * Test case of the compaction entry point of {@link SparkMain}, which the
'compaction run' and
+ * 'compaction scheduleAndExecute' commands reach through a spark-submit of
their own.
+ */
+ @Test
+ public void testSparkMainCompact() throws Exception {
+ writeDeltaCommits();
+ assertEquals(0,
HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient())
+
.getActiveTimeline().filterPendingCompactionTimeline().countInstants());
+
+ int returnCode = SparkMain.compact(jsc(), tablePath, tableName, null, 1,
"", 0,
+ UtilHelpers.SCHEDULE_AND_EXECUTE, null,
+
Collections.singletonList("hoodie.compact.inline.max.delta.commits=1"));
+
+ assertEquals(0, returnCode);
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
+ // the plan it scheduled ran to completion, leaving a commit and nothing
pending
+ assertEquals(0,
metaClient.getActiveTimeline().filterPendingCompactionTimeline().countInstants());
+ assertEquals(1, metaClient.getActiveTimeline().filterCompletedInstants()
+ .filter(instant ->
COMMIT_ACTION.equals(instant.getAction())).countInstants());
+ }
+
+ /**
+ * Writes two delta commits into a new MOR table at {@link #tablePath}, the
second one updating
+ * the records of the first so that the file groups have log files to
compact.
+ */
+ private void writeDeltaCommits() throws IOException {
+ createTableAndConnect(tablePath, tableName, HoodieTableType.MERGE_ON_READ,
HoodieAvroPayload.class.getName());
+
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(new String[]
{DEFAULT_FIRST_PARTITION_PATH});
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath(tablePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA).withParallelism(1,
1).forTable(tableName).build();
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(context(),
config)) {
+ String firstCommit = client.startCommit();
+ List<HoodieRecord> records = dataGen.generateInserts(firstCommit, 10);
+ writeAndCommit(client, records, firstCommit);
+
+ String secondCommit = client.startCommit();
+ writeAndCommit(client, dataGen.generateUpdates(secondCommit, 5),
secondCommit);
+ }
+ }
+
+ private void writeAndCommit(SparkRDDWriteClient client, List<HoodieRecord>
records, String commitTime) {
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+ List<WriteStatus> result = client.upsert(writeRecords,
commitTime).collect();
+ client.commit(commitTime, jsc().parallelize(result));
+ Assertions.assertNoWriteErrors(result);
+ }
+
private void generateCompactionInstances() throws IOException {
// create MOR table.
new TableCommand().createTable(
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestExportCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestExportCommand.java
new file mode 100644
index 000000000000..c0128d70a4a4
--- /dev/null
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestExportCommand.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.cli.commands;
+
+import org.apache.hudi.cli.HoodieCLI;
+import org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
+import org.apache.hudi.cli.testutils.HoodieTestCommitMetadataGenerator;
+import org.apache.hudi.cli.testutils.ShellEvaluationResultUtil;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.exception.HoodieException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.shell.Shell;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test cases for {@link ExportCommand}.
+ */
+@Tag("functional")
+@SpringBootTest(properties = {"spring.shell.interactive.enabled=false",
"spring.shell.command.script.enabled=false"})
+public class TestExportCommand extends CLIFunctionalTestHarness {
+
+ private static final String[] COMMIT_TIMES = new String[] {"101", "102",
"103"};
+
+ @Autowired
+ private Shell shell;
+
+ private String tablePath;
+ private Path exportFolder;
+
+ @BeforeEach
+ public void init() throws Exception {
+ HoodieCLI.conf = storageConf();
+ String tableName = tableName();
+ tablePath = tablePath(tableName);
+ exportFolder = Files.createDirectories(Paths.get(basePath(),
"exported-instants"));
+
+ createTableAndConnect(tablePath, tableName, HoodieTableType.COPY_ON_WRITE,
HoodieAvroPayload.class.getName());
+ for (String commitTime : COMMIT_TIMES) {
+
HoodieTestCommitMetadataGenerator.createCommitFileWithMetadata(tablePath,
commitTime, storageConf());
+ }
+ HoodieCLI.refreshTableMetadata();
+ }
+
+ /**
+ * Exports the whole timeline. The instant count is passed as the limit and
the ordering is
+ * descending on purpose: that is the one shape in which the export does not
walk the archived
+ * timeline, whose reader cannot open the LSM timeline history directory of
a table of version
+ * eight or above. The limit is not honoured for active instants either.
Both are tracked in
+ * https://github.com/apache/hudi/issues/19879; once fixed, this test can
drop the workaround.
+ */
+ @Test
+ public void testExportInstants() throws Exception {
+ Object result = shell.evaluate(
+ () -> "export instants --desc true --limit " + COMMIT_TIMES.length + "
--localFolder " + exportFolder);
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result),
String.valueOf(result));
+ assertEquals("Exported " + COMMIT_TIMES.length + " Instants to " +
exportFolder, result.toString());
+
+ // one file per completed instant, named after the instant file it was
read from
+ assertEquals(instantFileNames(COMMIT_TIMES), exportedFiles());
+
+ // The export copies the instant file off the timeline as it stands, in
whatever format the
+ // table writes its commit metadata in, so it is read back through the
table's own serde and
+ // compared with what the fixture wrote.
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ Set<String> writtenPartitions = new HashSet<>(
+ Arrays.asList(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH));
+ for (HoodieInstant instant :
metaClient.getActiveTimeline().filterCompletedInstants().getInstants()) {
+ Map<String, List<HoodieWriteStat>> writeStats =
readExportedCommit(metaClient, instant).getPartitionToWriteStats();
+ assertEquals(writtenPartitions, writeStats.keySet());
+ for (List<HoodieWriteStat> partitionStats : writeStats.values()) {
+ assertEquals(1, partitionStats.size());
+ assertEquals(HoodieTestCommitMetadataGenerator.DEFAULT_NUM_WRITES,
partitionStats.get(0).getNumWrites());
+ assertEquals(HoodieTestCommitMetadataGenerator.DEFAULT_PRE_COMMIT,
partitionStats.get(0).getPrevCommit());
+ }
+ }
+ }
+
+ /**
+ * Reads back the file the export wrote for one instant.
+ *
+ * @param metaClient Meta client of the exported table.
+ * @param instant Instant whose exported file to read.
+ * @return The commit metadata the file holds.
+ */
+ private HoodieCommitMetadata readExportedCommit(HoodieTableMetaClient
metaClient, HoodieInstant instant)
+ throws IOException {
+ String fileName =
metaClient.getInstantFileNameGenerator().getFileName(instant);
+ try (InputStream exported =
Files.newInputStream(exportFolder.resolve(fileName))) {
+ return metaClient.getCommitMetadataSerDe().deserialize(instant,
exported, () -> false, HoodieCommitMetadata.class);
+ }
+ }
+
+ @Test
+ public void testExportInstantsToInvalidFolder() {
+ String missingFolder = Paths.get(basePath(), "no-such-folder").toString();
+ Object result = shell.evaluate(() -> "export instants --localFolder " +
missingFolder);
+ assertFalse(ShellEvaluationResultUtil.isSuccess(result));
+ assertEquals(HoodieException.class, result.getClass());
+ assertTrue(result.toString().contains(missingFolder + " is not a valid
local directory"), result.toString());
+ assertFalse(Files.exists(Paths.get(missingFolder)));
+ }
+
+ private Set<String> exportedFiles() {
+ try (Stream<Path> files = Files.list(exportFolder)) {
+ return files.map(file ->
file.getFileName().toString()).collect(Collectors.toSet());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * The names of the instant files of the given commits, which are the names
the export uses.
+ *
+ * @param commitTimes Requested times of the commits.
+ * @return The instant file names.
+ */
+ private Set<String> instantFileNames(String... commitTimes) {
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ List<String> wanted = Arrays.asList(commitTimes);
+ return
metaClient.getActiveTimeline().filterCompletedInstants().getInstantsAsStream()
+ .filter(instant -> wanted.contains(instant.requestedTime()))
+ .map(instant ->
metaClient.getInstantFileNameGenerator().getFileName(instant))
+ .collect(Collectors.toSet());
+ }
+}
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestMetadataCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestMetadataCommand.java
index 90bd4bb7b59b..0de8620cca0c 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestMetadataCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestMetadataCommand.java
@@ -34,6 +34,8 @@ import
org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.keygen.SimpleKeyGenerator;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.testutils.Assertions;
import org.apache.spark.api.java.JavaRDD;
@@ -45,12 +47,21 @@ import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.shell.Shell;
import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -224,6 +235,224 @@ public class TestMetadataCommand extends
CLIFunctionalTestHarness {
}
}
+ @Test
+ public void testMetadataStatsAndFileListing() throws Exception {
+ writeOneCommit(true);
+ connectToTable();
+
+ // The command opens the reader with metadata metrics off, so there is
nothing to report on,
+ // but the stat table is still rendered. Tracked in
https://github.com/apache/hudi/issues/19880;
+ // once fixed, assert on the rows instead.
+ Object stats = shell.evaluate(() -> "metadata stats");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(stats));
+ assertTrue(stats.toString().contains("stat key"), stats.toString());
+ assertTrue(renderedRows(stats.toString()).isEmpty(), stats.toString());
+
+ Object partitions = shell.evaluate(() -> "metadata list-partitions");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(partitions));
+ Set<String> written = writtenPartitions();
+ assertFalse(written.isEmpty());
+ assertEquals(written, renderedRows(partitions.toString()).stream()
+ .map(row -> row.get(0)).collect(Collectors.toSet()));
+
+ // The files of one partition, as the metadata table has them.
+ Object files = shell.evaluate(() -> "metadata list-files --partition " +
DEFAULT_FIRST_PARTITION_PATH);
+ assertTrue(ShellEvaluationResultUtil.isSuccess(files));
+ Set<String> baseFiles = baseFilesOf(DEFAULT_FIRST_PARTITION_PATH);
+ assertFalse(baseFiles.isEmpty());
+ assertEquals(baseFiles.size(), renderedRows(files.toString()).size(),
files.toString());
+ for (String baseFile : baseFiles) {
+ assertTrue(files.toString().contains(baseFile), files.toString());
+ }
+
+ // Without --partition the lookup key is the non-partitioned name ("."),
which the files index
+ // of a partitioned table has no record for, so the lookup misses and
nothing is listed.
+ Object rootFiles = shell.evaluate(() -> "metadata list-files");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(rootFiles));
+ assertTrue(renderedRows(rootFiles.toString()).isEmpty(),
rootFiles.toString());
+ }
+
+ @Test
+ public void testMetadataValidateFiles() throws Exception {
+ writeOneCommit(true);
+ connectToTable();
+
+ // Every file on disk is in the metadata table, so nothing is reported
without --verbose.
+ Object matching = shell.evaluate(() -> "metadata validate-files");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(matching));
+ assertTrue(renderedRows(matching.toString()).isEmpty(),
matching.toString());
+
+ // --verbose reports every file, present on both sides and of the same
size.
+ Object verbose = shell.evaluate(() -> "metadata validate-files --verbose
true");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(verbose));
+ List<List<String>> rows = renderedRows(verbose.toString());
+ int filesOnDisk = writtenPartitions().stream().mapToInt(partition ->
baseFilesOf(partition).size()).sum();
+ assertEquals(filesOnDisk, rows.size(), verbose.toString());
+ for (List<String> row : rows) {
+ assertEquals("true", row.get(2), row.toString());
+ assertEquals("true", row.get(3), row.toString());
+ assertEquals(row.get(4), row.get(5), row.toString());
+ }
+
+ // A base file the metadata table does not know about is reported even
without --verbose.
+ String strayFile = writeStrayBaseFile(DEFAULT_FIRST_PARTITION_PATH);
+ Object mismatching = shell.evaluate(() -> "metadata validate-files");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(mismatching));
+ List<List<String>> mismatchingRows = renderedRows(mismatching.toString());
+ assertEquals(1, mismatchingRows.size(), mismatching.toString());
+ assertEquals(DEFAULT_FIRST_PARTITION_PATH, mismatchingRows.get(0).get(0));
+ assertEquals(strayFile, mismatchingRows.get(0).get(1));
+ assertEquals("true", mismatchingRows.get(0).get(2));
+ assertEquals("false", mismatchingRows.get(0).get(3));
+ }
+
+ @Test
+ public void testMetadataCreateAndInit() throws Exception {
+ // Written with the metadata table off, so that it does not exist yet.
+ writeOneCommit(false);
+ connectToTable();
+ StoragePath metadataPath = new
StoragePath(HoodieTableMetadata.getMetadataTableBasePath(tablePath));
+ assertFalse(HoodieCLI.storage.exists(metadataPath));
+
+ // There is nothing to update yet.
+ Object tooEarly = shell.evaluate(() -> "metadata init");
+ assertFalse(ShellEvaluationResultUtil.isSuccess(tooEarly));
+ assertTrue(tooEarly.toString().contains("does not exist"),
tooEarly.toString());
+
+ Object created = shell.evaluate(() -> "metadata create");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(created));
+ assertTrue(created.toString().startsWith("Created Metadata Table in " +
metadataPath), created.toString());
+ HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder()
+
.setConf(HoodieCLI.conf.newInstance()).setBasePath(metadataPath.toString()).build();
+ int instantsAfterCreate =
metadataMetaClient.getActiveTimeline().countInstants();
+ assertTrue(instantsAfterCreate > 0);
+
+ // The second create finds the directory of the first one.
+ Object again = shell.evaluate(() -> "metadata create");
+ assertFalse(ShellEvaluationResultUtil.isSuccess(again));
+ assertTrue(again.toString().contains("not empty"), again.toString());
+
+ // Read only init opens the table without writing to it.
+ Object opened = shell.evaluate(() -> "metadata init --readonly true");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(opened));
+ assertTrue(opened.toString().startsWith("Opened Metadata Table in " +
metadataPath), opened.toString());
+ assertEquals(instantsAfterCreate,
+
HoodieTableMetaClient.reload(metadataMetaClient).getActiveTimeline().countInstants());
+
+ Object initialized = shell.evaluate(() -> "metadata init");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(initialized));
+ assertTrue(initialized.toString().startsWith("Initialized Metadata Table
in " + metadataPath),
+ initialized.toString());
+ }
+
+ @Test
+ public void testMetadataSetDirectory() throws Exception {
+ String customDirectory = tablePath + "-metadata";
+ try {
+ // An empty directory leaves the default location of the metadata table
in place.
+ Object unset = shell.evaluate(() -> "metadata set");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(unset));
+ assertEquals("Ok", unset.toString());
+ assertEquals(HoodieTableMetadata.getMetadataTableBasePath(tablePath),
+ MetadataCommand.getMetadataTableBasePath(tablePath));
+
+ Object set = shell.evaluate(() -> "metadata set --metadataDir " +
customDirectory);
+ assertTrue(ShellEvaluationResultUtil.isSuccess(set));
+ assertEquals("Ok", set.toString());
+ assertEquals(customDirectory,
MetadataCommand.getMetadataTableBasePath(tablePath));
+
+ // The directory is global to the session and can only be set once.
+ Object reset = shell.evaluate(() -> "metadata set --metadataDir " +
customDirectory + "-other");
+ assertFalse(ShellEvaluationResultUtil.isSuccess(reset));
+ assertEquals(customDirectory,
MetadataCommand.getMetadataTableBasePath(tablePath));
+ } finally {
+ clearMetadataBaseDirectory();
+ }
+ }
+
+ /**
+ * Writes one commit of ten records over two partitions into a new table at
{@link #tablePath}.
+ *
+ * @param enableMetadataTable Whether the write maintains the metadata table.
+ */
+ private void writeOneCommit(boolean enableMetadataTable) throws Exception {
+
writeOneCommit(HoodieMetadataConfig.newBuilder().enable(enableMetadataTable).build());
+ }
+
+ /**
+ * Writes one commit of ten records over two partitions into a new table at
{@link #tablePath}.
+ *
+ * @param metadataConfig Metadata table configuration of the write.
+ */
+ private void writeOneCommit(HoodieMetadataConfig metadataConfig) throws
Exception {
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableType(HoodieTableType.COPY_ON_WRITE.name())
+ .setTableName(tableName)
+
.setArchiveLogFolder(HoodieTableConfig.TIMELINE_HISTORY_PATH.defaultValue())
+ .setPayloadClassName("org.apache.hudi.common.model.HoodieAvroPayload")
+ .setPartitionFields("partition_path")
+ .setRecordKeyFields("_row_key")
+ .setKeyGeneratorClassProp(SimpleKeyGenerator.class.getCanonicalName())
+ .initTable(HoodieCLI.conf.newInstance(), tablePath);
+
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(
+ new String[] {DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH});
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath(tablePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withMetadataConfig(metadataConfig)
+ .build();
+
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(context(),
config)) {
+ String commitTime = client.startCommit();
+ List<HoodieRecord> records = dataGen.generateInserts(commitTime, 10);
+ JavaRDD<HoodieRecord> writeRecords =
context().getJavaSparkContext().parallelize(records, 1);
+ List<WriteStatus> result = client.upsert(writeRecords,
commitTime).collect();
+ client.commit(commitTime, jsc().parallelize(result));
+ Assertions.assertNoWriteErrors(result);
+ }
+ }
+
+ private void connectToTable() throws IOException {
+ new TableCommand().connect(tablePath, false, 0, 0, 0,
"WAIT_TO_ADJUST_SKEW", 200L, false);
+ }
+
+ private Set<String> writtenPartitions() {
+ return Stream.of(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH)
+ .filter(partition -> Files.isDirectory(Paths.get(tablePath,
partition)))
+ .collect(Collectors.toSet());
+ }
+
+ private Set<String> baseFilesOf(String partition) {
+ try (Stream<Path> files = Files.list(Paths.get(tablePath, partition))) {
+ return files.map(file -> file.getFileName().toString())
+ .filter(name -> name.endsWith(BASE_FILE_EXTENSION))
+ .collect(Collectors.toSet());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Adds a base file the metadata table has never heard of to a partition of
the table.
+ *
+ * @param partition Partition to add the file to.
+ * @return The name of the file created.
+ */
+ private String writeStrayBaseFile(String partition) throws IOException {
+ String existing = baseFilesOf(partition).iterator().next();
+ // <fileId>_<writeToken>_<instantTime>.parquet, only the file id has to be
new
+ String strayFile = UUID.randomUUID() +
existing.substring(existing.indexOf('_'));
+ Files.createFile(Paths.get(tablePath, partition, strayFile));
+ return strayFile;
+ }
+
+ private static void clearMetadataBaseDirectory() throws Exception {
+ Field field =
MetadataCommand.class.getDeclaredField("metadataBaseDirectory");
+ field.setAccessible(true);
+ field.set(null, null);
+ }
+
private void validateRecordIndexOutput(String recordKey, Option<String>
partitionPathOp,
String expectedInstantTime, String
expectedPartitionPath) {
// Execute the metadata lookup-record-index command
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 951213b856ed..304459bcf730 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
@@ -61,6 +61,7 @@ import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_P
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -74,10 +75,12 @@ public class TestRollbacksCommand extends
CLIFunctionalTestHarness {
@Autowired
private Shell shell;
+ private String tablePath;
+
@BeforeEach
public void init() throws Exception {
String tableName = tableName();
- String tablePath = tablePath(tableName);
+ tablePath = tablePath(tableName);
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.MERGE_ON_READ.name(),
"", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
@@ -205,4 +208,38 @@ public class TestRollbacksCommand extends
CLIFunctionalTestHarness {
String got = removeNonWordAndStripSpace(result.toString());
assertEquals(expected, got);
}
+
+ /**
+ * Test case of the rollback entry point of {@link SparkMain}, which the
'commit rollback'
+ * command reaches through a spark-submit of its own. The fixture leaves
commit 101 in the
+ * inflight state, which is the failed write such a rollback is meant to
clean up.
+ */
+ @Test
+ public void testSparkMainRollback() throws Exception {
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ HoodieActiveTimeline timeline = metaClient.reloadActiveTimeline();
+
assertTrue(timeline.getCommitsTimeline().filterInflightsAndRequested().containsInstant("101"));
+ int rollbacksBefore =
timeline.getRollbackTimeline().filterCompletedInstants().countInstants();
+
+ assertEquals(0, SparkMain.rollback(jsc(), "101", tablePath, false));
+
+ timeline = metaClient.reloadActiveTimeline();
+ assertFalse(timeline.getCommitsTimeline().containsInstant("101"));
+ assertEquals(rollbacksBefore + 1,
+
timeline.getRollbackTimeline().filterCompletedInstants().countInstants());
+ }
+
+ /**
+ * An instant that is not on the timeline cannot be rolled back, and nothing
on the timeline
+ * moves because of the attempt.
+ */
+ @Test
+ public void testSparkMainRollbackOfUnknownInstant() throws Exception {
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ List<HoodieInstant> before =
metaClient.reloadActiveTimeline().getInstants();
+
+ assertEquals(-1, SparkMain.rollback(jsc(), "999", tablePath, false));
+
+ assertEquals(before, metaClient.reloadActiveTimeline().getInstants());
+ }
}
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestSavepointsCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestSavepointsCommand.java
index 4700b2366319..124686d9976a 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestSavepointsCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestSavepointsCommand.java
@@ -23,11 +23,18 @@ import org.apache.hudi.cli.HoodiePrintHelper;
import org.apache.hudi.cli.HoodieTableHeaderFields;
import org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
import org.apache.hudi.cli.testutils.ShellEvaluationResultUtil;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.testutils.Assertions;
+import org.apache.spark.api.java.JavaRDD;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
@@ -37,9 +44,13 @@ import org.springframework.shell.Shell;
import java.io.IOException;
import java.util.Comparator;
+import java.util.List;
import java.util.stream.Stream;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -117,4 +128,55 @@ public class TestSavepointsCommand extends
CLIFunctionalTestHarness {
// After refresh, there are 4 instants
assertEquals(4, timeline.countInstants(), "there should have 4 instants");
}
+
+ /**
+ * Test case of the savepoint entry points of {@link SparkMain}, which the
savepoint commands
+ * reach through a spark-submit of their own.
+ */
+ @Test
+ public void testSparkMainSavepointLifecycle() throws Exception {
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(new String[]
{DEFAULT_FIRST_PARTITION_PATH});
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath(tablePath).withSchema(TRIP_EXAMPLE_SCHEMA).build();
+ String firstCommit;
+ String secondCommit;
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(context(),
config)) {
+ firstCommit = writeInserts(client, dataGen);
+ secondCommit = writeInserts(client, dataGen);
+ }
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ assertEquals(0, SparkMain.createSavepoint(jsc(), firstCommit, "test-user",
"test-comment", tablePath));
+ HoodieTimeline savepoints =
metaClient.reloadActiveTimeline().getSavePointTimeline().filterCompletedInstants();
+ assertEquals(1, savepoints.countInstants());
+ assertEquals(firstCommit, savepoints.firstInstant().get().requestedTime());
+ assertEquals("test-user",
savepoints.readSavepointMetadata(savepoints.firstInstant().get()).getSavepointedBy());
+
+ // a commit that is not on the timeline cannot be savepointed
+ assertEquals(-1, SparkMain.createSavepoint(jsc(), "001", "test-user",
"test-comment", tablePath));
+ assertEquals(1,
metaClient.reloadActiveTimeline().getSavePointTimeline().countInstants());
+
+ // restoring to the savepoint takes the commits made after it off the
timeline
+ assertEquals(0, SparkMain.rollbackToSavepoint(jsc(), firstCommit,
tablePath, false));
+ HoodieTimeline commits =
metaClient.reloadActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ assertTrue(commits.containsInstant(firstCommit));
+ assertFalse(commits.containsInstant(secondCommit));
+ assertEquals(-1, SparkMain.rollbackToSavepoint(jsc(), "001", tablePath,
false));
+
+ assertEquals(0, SparkMain.deleteSavepoint(jsc(), firstCommit, tablePath));
+ assertEquals(0,
metaClient.reloadActiveTimeline().getSavePointTimeline().countInstants());
+ // deleting a savepoint that is already gone is a no-op rather than a
failure
+ assertEquals(0, SparkMain.deleteSavepoint(jsc(), firstCommit, tablePath));
+ assertEquals(0,
metaClient.reloadActiveTimeline().getSavePointTimeline().countInstants());
+ }
+
+ private String writeInserts(SparkRDDWriteClient client,
HoodieTestDataGenerator dataGen) {
+ String commitTime = client.startCommit();
+ List<HoodieRecord> records = dataGen.generateInserts(commitTime, 10);
+ JavaRDD<HoodieRecord> writeRecords =
context().getJavaSparkContext().parallelize(records, 1);
+ List<WriteStatus> result = client.upsert(writeRecords,
commitTime).collect();
+ client.commit(commitTime, jsc().parallelize(result));
+ Assertions.assertNoWriteErrors(result);
+ return commitTime;
+ }
}
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestTimelineCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestTimelineCommand.java
new file mode 100644
index 000000000000..2b2f4ec96b0e
--- /dev/null
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestTimelineCommand.java
@@ -0,0 +1,316 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.cli.commands;
+
+import org.apache.hudi.avro.model.HoodieInstantInfo;
+import org.apache.hudi.avro.model.HoodieRollbackPlan;
+import org.apache.hudi.cli.HoodieCLI;
+import org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
+import org.apache.hudi.cli.testutils.ShellEvaluationResultUtil;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.testutils.HoodieMetadataTestTable;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.SparkHoodieBackedTableMetadataWriter;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.shell.Shell;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test cases for {@link TimelineCommand}.
+ */
+@Tag("functional")
+@SpringBootTest(properties = {"spring.shell.interactive.enabled=false",
"spring.shell.command.script.enabled=false"})
+public class TestTimelineCommand extends CLIFunctionalTestHarness {
+
+ // Column offsets of the data table part of the rendered timeline, row
number included.
+ private static final int COL_INSTANT = 1;
+ private static final int COL_ACTION = 2;
+ private static final int COL_STATE = 3;
+ private static final int COL_REQUESTED_TIME = 4;
+ private static final int COL_INFLIGHT_TIME = 5;
+ private static final int COL_COMPLETED_TIME = 6;
+ // Column offsets of the metadata table part, only rendered with
--with-metadata-table.
+ private static final int COL_MT_ACTION = 7;
+ private static final int COL_MT_STATE = 8;
+
+ // The commit left in the requested state, and the rollback scheduled
against it.
+ private static final String REQUESTED_COMMIT = "103";
+ private static final String PENDING_ROLLBACK_INSTANT = "104";
+ private static final String ROLLED_BACK_COMMIT = "102";
+
+ private static final String DATE_NO_SECONDS = "\\d{2}-\\d{2} \\d{2}:\\d{2}";
+ private static final String DATE_WITH_SECONDS = "\\d{2}-\\d{2}
\\d{2}:\\d{2}:\\d{2}";
+
+ @Autowired
+ private Shell shell;
+
+ private String tablePath;
+ private HoodieTableMetaClient metaClient;
+ private String rollbackInstantTime;
+
+ /**
+ * Builds a table whose active timeline holds two completed commits, a
completed rollback of a
+ * third commit, one commit left in the requested state and a rollback
scheduled against that
+ * commit, with the metadata table enabled so that the metadata table
timeline is populated too.
+ */
+ @BeforeEach
+ public void init() throws Exception {
+ HoodieCLI.conf = storageConf();
+ String tableName = tableName();
+ tablePath = tablePath(tableName);
+
+ createTableAndConnect(tablePath, tableName, HoodieTableType.COPY_ON_WRITE,
HoodieAvroPayload.class.getName());
+ metaClient = HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
+
+ Map<String, String> partitionAndFileId = new HashMap<>();
+ partitionAndFileId.put(DEFAULT_FIRST_PARTITION_PATH, "file-1");
+ partitionAndFileId.put(DEFAULT_SECOND_PARTITION_PATH, "file-2");
+
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath(tablePath)
+ .withMetadataConfig(
+ // Column Stats Index is disabled, since this table is built with
empty commit metadata
+
HoodieMetadataConfig.newBuilder().withMetadataIndexColumnStats(false).build())
+ .withRollbackUsingMarkers(false)
+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ .build();
+
+ try (HoodieTableMetadataWriter metadataWriter =
SparkHoodieBackedTableMetadataWriter.create(
+ metaClient.getStorageConf(), config, context)) {
+ HoodieTestTable testTable = HoodieMetadataTestTable.of(metaClient,
metadataWriter, Option.of(context))
+ .withPartitionMetaFiles(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH)
+
.addCommit("100").withBaseFilesInPartitions(partitionAndFileId).getLeft()
+
.addCommit("101").withBaseFilesInPartitions(partitionAndFileId).getLeft()
+ .addInflightCommit(ROLLED_BACK_COMMIT);
+ testTable.withBaseFilesInPartitions(partitionAndFileId);
+
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(context(),
config)) {
+ client.rollback(ROLLED_BACK_COMMIT);
+ }
+ // left behind on the timeline so that the incomplete timeline is not
empty
+ testTable.addRequestedCommit(REQUESTED_COMMIT);
+
+ // A rollback that is scheduled but has not run yet. The completed
rollback above deleted the
+ // commit it targeted, so on this table the scheduled one is what makes
the "rolled back by"
+ // annotation reachable: it leaves its target on the timeline. Added
straight through the test
+ // table so that it stays on the data table timeline only.
+ HoodieRollbackPlan rollbackPlan = new HoodieRollbackPlan();
+ rollbackPlan.setRollbackRequests(Collections.emptyList());
+ rollbackPlan.setInstantToRollback(new
HoodieInstantInfo(REQUESTED_COMMIT, HoodieTimeline.COMMIT_ACTION));
+ testTable.addRequestedRollback(PENDING_ROLLBACK_INSTANT, rollbackPlan);
+ testTable.addInflightRollback(PENDING_ROLLBACK_INSTANT);
+ }
+
+ HoodieCLI.refreshTableMetadata();
+ metaClient = HoodieCLI.getTableMetaClient();
+ rollbackInstantTime = metaClient.getActiveTimeline().getRollbackTimeline()
+ .filterCompletedInstants().lastInstant().get().requestedTime();
+ }
+
+ @Test
+ public void testShowActive() {
+ Object result = shell.evaluate(() -> "timeline show active");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ List<List<String>> rows = renderedRows(result.toString());
+ assertEquals(instantTimes(metaClient), rows.stream().map(r ->
r.get(COL_INSTANT)).collect(Collectors.toSet()));
+ assertEquals(metaClient.getActiveTimeline().countInstants(), rows.size());
+
+ List<String> commit100 = rowOf(rows, "100");
+ assertEquals("commit", commit100.get(COL_ACTION));
+ assertEquals(HoodieInstant.State.COMPLETED.toString(),
commit100.get(COL_STATE));
+ // a completed commit has all three instant files, so all three
modification times are rendered
+ assertTrue(commit100.get(COL_REQUESTED_TIME).matches(DATE_NO_SECONDS),
commit100.toString());
+ assertTrue(commit100.get(COL_INFLIGHT_TIME).matches(DATE_NO_SECONDS),
commit100.toString());
+ assertTrue(commit100.get(COL_COMPLETED_TIME).matches(DATE_NO_SECONDS),
commit100.toString());
+
+ List<String> commit103 = rowOf(rows, REQUESTED_COMMIT);
+ assertEquals("commit", commit103.get(COL_ACTION));
+ assertEquals(HoodieInstant.State.REQUESTED.toString(),
commit103.get(COL_STATE));
+ // only the requested file exists for it, the other two states render as a
dash
+ assertTrue(commit103.get(COL_REQUESTED_TIME).matches(DATE_NO_SECONDS),
commit103.toString());
+ assertEquals("-", commit103.get(COL_INFLIGHT_TIME));
+ assertEquals("-", commit103.get(COL_COMPLETED_TIME));
+
+ assertEquals("rollback", rowOf(rows, rollbackInstantTime).get(COL_ACTION));
+ }
+
+ @Test
+ public void testShowActiveWithLimitAndSorting() {
+ Object result = shell.evaluate(() -> "timeline show active --limit 2
--sortBy Instant --desc true");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ List<List<String>> rows = renderedRows(result.toString());
+ assertEquals(2, rows.size());
+ List<String> allInstants = new ArrayList<>(instantTimes(metaClient));
+ allInstants.sort(String::compareTo);
+ // descending order on the instant time, cut to the first two rows
+ assertEquals(allInstants.get(allInstants.size() - 1),
rows.get(0).get(COL_INSTANT));
+ assertEquals(allInstants.get(allInstants.size() - 2),
rows.get(1).get(COL_INSTANT));
+ }
+
+ @Test
+ public void testShowActiveHeaderOnly() {
+ Object result = shell.evaluate(() -> "timeline show active --headeronly
true");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ assertTrue(renderedRows(result.toString()).isEmpty(), result.toString());
+ assertTrue(result.toString().contains("Instant"), result.toString());
+ assertTrue(result.toString().contains(EMPTY_TABLE_CELL),
result.toString());
+ }
+
+ @Test
+ public void testShowActiveWithRollbackInfoAndSeconds() {
+ Object result = shell.evaluate(
+ () -> "timeline show active --show-rollback-info true
--show-time-seconds true");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ List<List<String>> rows = renderedRows(result.toString());
+ // the completed rollback is annotated with the commit it rolled back,
read from its metadata
+ assertEquals("rollback Rolls back " + ROLLED_BACK_COMMIT, rowOf(rows,
rollbackInstantTime).get(COL_ACTION));
+ // the scheduled one with the commit its plan targets
+ assertEquals("rollback Rolls back " + REQUESTED_COMMIT,
+ rowOf(rows, PENDING_ROLLBACK_INSTANT).get(COL_ACTION));
+ // and that commit is annotated back with the rollback scheduled against it
+ assertEquals("commit Rolled back by " + PENDING_ROLLBACK_INSTANT,
+ rowOf(rows, REQUESTED_COMMIT).get(COL_ACTION));
+ // instants that no rollback refers to carry no annotation
+ assertEquals("commit", rowOf(rows, "100").get(COL_ACTION));
+ assertTrue(rowOf(rows,
"100").get(COL_COMPLETED_TIME).matches(DATE_WITH_SECONDS),
+ rowOf(rows, "100").toString());
+ }
+
+ @Test
+ public void testShowIncomplete() {
+ Object result = shell.evaluate(() -> "timeline show incomplete");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ List<List<String>> rows = renderedRows(result.toString());
+ assertEquals(2, rows.size(), result.toString());
+
+ List<String> requestedCommit = rowOf(rows, REQUESTED_COMMIT);
+ assertEquals("commit", requestedCommit.get(COL_ACTION));
+ assertEquals(HoodieInstant.State.REQUESTED.toString(),
requestedCommit.get(COL_STATE));
+ assertEquals("-", requestedCommit.get(COL_COMPLETED_TIME));
+
+ List<String> pendingRollback = rowOf(rows, PENDING_ROLLBACK_INSTANT);
+ assertEquals("rollback", pendingRollback.get(COL_ACTION));
+ assertEquals(HoodieInstant.State.INFLIGHT.toString(),
pendingRollback.get(COL_STATE));
+ assertEquals("-", pendingRollback.get(COL_COMPLETED_TIME));
+ }
+
+ @Test
+ public void testShowActiveWithMetadataTable() {
+ Object result = shell.evaluate(
+ () -> "timeline show active --with-metadata-table true
--show-rollback-info true --limit 50");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ HoodieTableMetaClient mtMetaClient = metadataTableMetaClient();
+ Set<String> expected = instantTimes(metaClient);
+ expected.addAll(instantTimes(mtMetaClient));
+
+ List<List<String>> rows = renderedRows(result.toString());
+ assertEquals(expected, rows.stream().map(r ->
r.get(COL_INSTANT)).collect(Collectors.toSet()));
+
+ // the data table columns of a data table only instant, and its empty
metadata table columns
+ List<String> commit103 = rowOf(rows, REQUESTED_COMMIT);
+ assertEquals("commit Rolled back by " + PENDING_ROLLBACK_INSTANT,
commit103.get(COL_ACTION));
+ assertEquals("-", commit103.get(COL_MT_ACTION));
+ assertEquals("-", commit103.get(COL_MT_STATE));
+
+ // every metadata table instant is rendered with its action and state in
the metadata columns
+ for (HoodieInstant instant :
mtMetaClient.getActiveTimeline().getInstants()) {
+ List<String> row = rowOf(rows, instant.requestedTime());
+ assertEquals(instant.getAction(), row.get(COL_MT_ACTION));
+ assertEquals(instant.getState().toString(), row.get(COL_MT_STATE));
+ }
+ }
+
+ @Test
+ public void testMetadataShowActive() {
+ Object result = shell.evaluate(() -> "metadata timeline show active
--limit 50");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ HoodieTableMetaClient mtMetaClient = metadataTableMetaClient();
+ List<List<String>> rows = renderedRows(result.toString());
+ assertFalse(rows.isEmpty(), result.toString());
+ assertEquals(instantTimes(mtMetaClient), rows.stream().map(r ->
r.get(COL_INSTANT)).collect(Collectors.toSet()));
+ for (HoodieInstant instant :
mtMetaClient.getActiveTimeline().getInstants()) {
+ List<String> row = rowOf(rows, instant.requestedTime());
+ assertEquals(instant.getAction(), row.get(COL_ACTION));
+ assertEquals(instant.getState().toString(), row.get(COL_STATE));
+ }
+ }
+
+ @Test
+ public void testMetadataShowIncomplete() {
+ Object result = shell.evaluate(() -> "metadata timeline show incomplete");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ // the metadata table is written synchronously, so nothing is left
incomplete on its timeline
+ assertEquals(0,
metadataTableMetaClient().getActiveTimeline().filterInflightsAndRequested().countInstants());
+ assertTrue(renderedRows(result.toString()).isEmpty(), result.toString());
+ }
+
+ private HoodieTableMetaClient metadataTableMetaClient() {
+ return HoodieTableMetaClient.builder()
+ .setConf(HoodieCLI.conf.newInstance())
+ .setBasePath(HoodieTableMetadata.getMetadataTableBasePath(tablePath))
+ .build();
+ }
+
+ private static Set<String> instantTimes(HoodieTableMetaClient metaClient) {
+ return
HoodieTableMetaClient.reload(metaClient).getActiveTimeline().getInstantsAsStream()
+ .map(HoodieInstant::requestedTime).collect(Collectors.toSet());
+ }
+
+ private static List<String> rowOf(List<List<String>> rows, String
instantTime) {
+ return rows.stream().filter(r ->
r.get(COL_INSTANT).equals(instantTime)).findFirst()
+ .orElseThrow(() -> new AssertionError("No rendered row for instant " +
instantTime + " in " + rows));
+ }
+}
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/functional/CLIFunctionalTestHarness.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/functional/CLIFunctionalTestHarness.java
index 17584caa3d8a..8bde9cdd06ad 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/functional/CLIFunctionalTestHarness.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/functional/CLIFunctionalTestHarness.java
@@ -19,9 +19,14 @@
package org.apache.hudi.cli.functional;
+import org.apache.hudi.cli.HoodieCLI;
+import org.apache.hudi.cli.commands.TableCommand;
import org.apache.hudi.client.SparkRDDReadClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.StorageConfiguration;
@@ -41,11 +46,20 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
public class CLIFunctionalTestHarness implements SparkProvider {
protected static final String BASE_FILE_EXTENSION =
HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().getFileExtension();
+ // Box drawing characters of a rendered table, kept as escapes so that the
source stays ASCII.
+ private static final char TABLE_ROW_START = '\u2551'; // double vertical,
starts and ends a row
+ private static final char TABLE_HEADER_DIVIDER = '\u2560'; // double
vertical and right, under the header
+ private static final char TABLE_ROW_DIVIDER = '\u255F'; // double vertical
and single right, between rows
+ private static final String TABLE_CELL_SEPARATORS = "[\u2551\u2502]"; //
double and single vertical
+ protected static final String EMPTY_TABLE_CELL = "(empty)";
+
protected static int timelineServicePort =
FileSystemViewStorageConfig.REMOTE_PORT_NUM.defaultValue();
protected static transient TimelineService timelineService;
@@ -139,6 +153,91 @@ public class CLIFunctionalTestHarness implements
SparkProvider {
return str.replaceAll("[\\s]+", ",").replaceAll("[\\W]+", ",");
}
+ /**
+ * Initializes a table and connects the CLI to it, which is what the
'create' command does once
+ * its check for an already existing table comes back empty. That check is
what this skips: on a
+ * path that holds no table it spends five seconds in the hoodie.properties
read retry loop (five
+ * attempts, one second apart) before it concludes there is nothing there,
and a fixture creating
+ * its own table knows that already.
+ *
+ * @param tablePath Base path of the table to create.
+ * @param tableName Name of the table.
+ * @param tableType Type of the table.
+ * @param payloadClass Payload class of the table.
+ */
+ protected void createTableAndConnect(String tablePath, String tableName,
HoodieTableType tableType,
+ String payloadClass) throws IOException
{
+ boolean initialized = HoodieCLI.initConf();
+ HoodieCLI.initFS(initialized);
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableType(tableType.name())
+ .setTableName(tableName)
+ .setPayloadClassName(payloadClass)
+ .setTableVersion(HoodieTableVersion.current().versionCode())
+ .initTable(HoodieCLI.conf.newInstance(), tablePath);
+ new TableCommand().connect(tablePath, false, 0, 0, 0,
"WAIT_TO_ADJUST_SKEW", 200L, true);
+ }
+
+ /**
+ * Splits a table rendered by {@link org.apache.hudi.cli.HoodiePrintHelper}
into its data rows,
+ * each row being the list of its trimmed cell values. Cells spanning
several rendered lines are
+ * joined back into a single cell, separated by a space. The header and an
empty table yield no
+ * rows.
+ *
+ * @param rendered Rendered table.
+ * @return One list of cell values per data row.
+ */
+ protected static List<List<String>> renderedRows(String rendered) {
+ List<List<String>> rows = new ArrayList<>();
+ boolean inData = false;
+ boolean startOfRow = false;
+ for (String line : rendered.split("\n")) {
+ if (line.isEmpty()) {
+ continue;
+ }
+ char first = line.charAt(0);
+ if (first == TABLE_HEADER_DIVIDER || first == TABLE_ROW_DIVIDER) {
+ inData = true;
+ startOfRow = true;
+ continue;
+ }
+ if (first != TABLE_ROW_START || !inData) {
+ continue;
+ }
+ List<String> cells = renderedCells(line);
+ if (cells.size() == 1 && EMPTY_TABLE_CELL.equals(cells.get(0))) {
+ continue;
+ }
+ if (startOfRow) {
+ rows.add(cells);
+ startOfRow = false;
+ } else {
+ List<String> previous = rows.get(rows.size() - 1);
+ for (int i = 0; i < cells.size(); i++) {
+ if (!cells.get(i).isEmpty()) {
+ previous.set(i, (previous.get(i) + " " + cells.get(i)).trim());
+ }
+ }
+ }
+ }
+ return rows;
+ }
+
+ /**
+ * Splits a single rendered line into its trimmed cell values.
+ *
+ * @param line One line of a rendered table.
+ * @return The cell values of that line.
+ */
+ protected static List<String> renderedCells(String line) {
+ String[] parts = line.split(TABLE_CELL_SEPARATORS, -1);
+ List<String> cells = new ArrayList<>();
+ for (int i = 1; i < parts.length - 1; i++) {
+ cells.add(parts[i].trim());
+ }
+ return cells;
+ }
+
protected int incrementTimelineServicePortToUse() {
// Increment the timeline service port for each individual test
// to avoid port reuse causing failures
