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 08af5869addb fix(cli): repair the hudi-cli tests and re-enable the
module in CI (#18816)
08af5869addb is described below
commit 08af5869addbdf3249464cf8771f60150e0bd0d2
Author: Kavya Popat <[email protected]>
AuthorDate: Tue Sep 8 18:20:05 2026 +0530
fix(cli): repair the hudi-cli tests and re-enable the module in CI (#18816)
The hudi-cli tests have not run in CI since #10805 excluded the module.
#11866 then repurposed the fifth parameter of TableCommand.createTable
from --layoutVersion to --tableVersion. Both are Integer, so the 14 call
sites passing TimelineLayoutVersion.VERSION_1 kept compiling and quietly
created table-version-1 tables, which Hudi 1.x rejects.
Tests: pass the current table version at those call sites, and fix the
timeline path, file-name generator and sort order that the v9 layout
changed. New coverage for archived clean and replacecommit rows, a v6
table read over the legacy archive, an empty timeline under show fsview
latest, and the clean-plan corruption cases below.
RepairsCommand: repair corrupted clean files left empty .clean.requested
files in place, because reading one raises "unable to read commit
metadata" rather than the "Not an Avro data file" it matched. That
message cannot be matched either, since the serde wraps a transient read
failure in it too. The command now reads the plan bytes in full, so a
read failure propagates and the instant stays, and only the in-memory
decode judges corruption -- through any runtime failure, since Avro
raises UnsupportedOperationException past its array ceiling and a
versionless plan leaves the migrator unboxing a null. A corrupt plan now
takes the whole pending action with it, inflight file first, instead of
leaving the requested file behind for a second pass.
ArchivedCommitsCommand: read through the archived timeline on the v8+
LSM layout instead of globbing .commits_.archive*, uncached so that a
trigger archival earlier in the session is visible. Tables below v8 keep
the legacy reader, which now renders the completed entry of each instant
and skips entries with no metadata, where a writer's empty requested
file used to NPE it. Actions with no typed reader render through the
schema embedded in their payload instead of failing the listing, and
without --sortBy the printer keeps timeline order and cuts at --limit,
so only the instants that can be rendered are loaded. Output change: the
legacy format archived each state separately, so show archived commits
printed three rows per commit and now prints one on both layouts.
CleansCommand renders clean partitions in partition order rather than
map iteration order, and FileSystemViewCommand guards the active
timeline lookup so view commands do not throw on an empty timeline.
CI: drop !hudi-cli from azure-pipelines-20230430.yml, and run the module
inside test-spark-client-and-hadoop-common in bot.yml, since Codecov is
fed only by bot.yml. That job's -am reactor grows from 10 to 23 modules
and it runs about 43 min, roughly 30 min under the critical path.
The hudi-cli functional suite goes from 113 tests with 12 failures and
24 errors to 118 with none, all 20 classes green, and every HUDI-7614
disable lifted.
Closes #16448
---
.github/workflows/bot.yml | 26 +-
azure-pipelines-20230430.yml | 2 -
.../hudi/cli/commands/ArchivedCommitsCommand.java | 357 +++++++++++++----
.../apache/hudi/cli/commands/CleansCommand.java | 6 +-
.../hudi/cli/commands/FileSystemViewCommand.java | 13 +-
.../apache/hudi/cli/commands/RepairsCommand.java | 77 +++-
.../org/apache/hudi/cli/commands/SparkMain.java | 4 +-
.../hudi/cli/commands/TestArchiveCommand.java | 39 +-
.../cli/commands/TestArchivedCommitsCommand.java | 426 ++++++++++++++++++---
.../hudi/cli/commands/TestCleansCommand.java | 8 +-
.../hudi/cli/commands/TestCommitsCommand.java | 15 +-
.../hudi/cli/commands/TestCompactionCommand.java | 10 +-
.../apache/hudi/cli/commands/TestDiffCommand.java | 4 +-
.../cli/commands/TestFileSystemViewCommand.java | 62 ++-
.../cli/commands/TestHoodieLogFileCommand.java | 4 +-
.../hudi/cli/commands/TestRepairsCommand.java | 325 ++++++++++++++--
.../hudi/cli/commands/TestRestoresCommand.java | 85 ++--
.../hudi/cli/commands/TestRollbacksCommand.java | 4 +-
.../hudi/cli/commands/TestSavepointsCommand.java | 4 +-
.../apache/hudi/cli/commands/TestStatsCommand.java | 4 +-
.../cli/commands/TestUpgradeDowngradeCommand.java | 3 +-
21 files changed, 1177 insertions(+), 301 deletions(-)
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
index dceae0761bfb..1887ac7acbc2 100644
--- a/.github/workflows/bot.yml
+++ b/.github/workflows/bot.yml
@@ -114,10 +114,16 @@ jobs:
- name: RAT check
run: ./scripts/release/validate_source_rat.sh
+ # hudi-cli rides along with hudi-spark-client here rather than joining the
catch-all
+ # module lists in test-common-and-other-modules: its functional suite starts
a
+ # SparkSession per test class, and this job has the headroom for it while
that one does
+ # not. hudi-cli's dependency closure is a superset of hudi-spark-client's,
so it shares
+ # this job's build instead of paying for a second one. It stays listed in
the catch-all
+ # job's UT_MODULES/FT_MODULES exclusions so the tests are not run twice.
test-spark-client-and-hadoop-common:
runs-on: ubuntu-latest
needs: changes
- timeout-minutes: 60
+ timeout-minutes: 75
strategy:
matrix:
include:
@@ -143,7 +149,7 @@ jobs:
SPARK_PROFILE: ${{ matrix.sparkProfile }}
FLINK_PROFILE: ${{ matrix.flinkProfile }}
run:
- mvn clean install -T 2 -D"$SCALA_PROFILE" -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -DskipTests=true -Phudi-platform-service $MVN_ARGS -am -pl
hudi-client/hudi-spark-client
+ mvn clean install -T 2 -D"$SCALA_PROFILE" -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -DskipTests=true -Phudi-platform-service $MVN_ARGS -am -pl
hudi-client/hudi-spark-client,hudi-cli
- name: UT - hudi-hadoop-common
if: needs.changes.outputs.relevant == 'true'
env:
@@ -168,6 +174,22 @@ jobs:
FLINK_PROFILE: ${{ matrix.flinkProfile }}
run:
mvn test -Pfunctional-tests -D"$SCALA_PROFILE" -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -fae -pl hudi-client/hudi-spark-client $MVN_ARGS
-Djacoco.skip=false
+ - name: UT - hudi-cli
+ if: needs.changes.outputs.relevant == 'true'
+ env:
+ SCALA_PROFILE: ${{ matrix.scalaProfile }}
+ SPARK_PROFILE: ${{ matrix.sparkProfile }}
+ FLINK_PROFILE: ${{ matrix.flinkProfile }}
+ run:
+ mvn test -Punit-tests -D"$SCALA_PROFILE" -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -fae -pl hudi-cli $MVN_ARGS -Djacoco.skip=false
+ - name: FT - hudi-cli
+ if: needs.changes.outputs.relevant == 'true'
+ env:
+ SCALA_PROFILE: ${{ matrix.scalaProfile }}
+ SPARK_PROFILE: ${{ matrix.sparkProfile }}
+ FLINK_PROFILE: ${{ matrix.flinkProfile }}
+ run:
+ mvn test -Pfunctional-tests -D"$SCALA_PROFILE" -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -fae -pl hudi-cli $MVN_ARGS -Djacoco.skip=false
- name: Generate merged coverage report
if: always() && needs.changes.outputs.relevant == 'true'
run: ./scripts/jacoco/generate_merged_coverage_report.sh
$GITHUB_WORKSPACE
diff --git a/azure-pipelines-20230430.yml b/azure-pipelines-20230430.yml
index b863f65bb1c5..0e3667a238d7 100644
--- a/azure-pipelines-20230430.yml
+++ b/azure-pipelines-20230430.yml
@@ -43,7 +43,6 @@ parameters:
- '!hudi-hadoop-mr'
- '!hudi-client/hudi-java-client'
- '!hudi-client/hudi-spark-client'
- - '!hudi-cli'
- '!hudi-examples'
- '!hudi-examples/hudi-examples-common'
- '!hudi-examples/hudi-examples-flink'
@@ -59,7 +58,6 @@ parameters:
type: object
default:
- '!hudi-client/hudi-spark-client'
- - '!hudi-cli'
- '!hudi-examples'
- '!hudi-examples/hudi-examples-common'
- '!hudi-examples/hudi-examples-flink'
diff --git
a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java
b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java
index c2cbe30cd0b5..2a50ca0a586f 100644
---
a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java
+++
b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java
@@ -20,6 +20,7 @@ package org.apache.hudi.cli.commands;
import org.apache.hudi.avro.model.HoodieArchivedMetaEntry;
import org.apache.hudi.avro.model.HoodieCommitMetadata;
+import org.apache.hudi.avro.model.HoodieReplaceCommitMetadata;
import org.apache.hudi.cli.HoodieCLI;
import org.apache.hudi.cli.HoodiePrintHelper;
import org.apache.hudi.cli.TableHeader;
@@ -27,22 +28,27 @@ import org.apache.hudi.cli.commands.SparkMain.SparkCommand;
import org.apache.hudi.cli.utils.InputStreamConsumer;
import org.apache.hudi.cli.utils.SparkUtil;
import org.apache.hudi.common.model.HoodieLogFile;
-import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormat.Reader;
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
+import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
import lombok.extern.slf4j.Slf4j;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
import org.apache.avro.specific.SpecificData;
@@ -52,11 +58,16 @@ import org.springframework.shell.standard.ShellComponent;
import org.springframework.shell.standard.ShellMethod;
import org.springframework.shell.standard.ShellOption;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static
org.apache.hudi.util.JavaScalaConverters.convertJavaPropertiesToScalaMap;
@@ -98,7 +109,12 @@ public class ArchivedCommitsCommand {
@ShellMethod(key = "show archived commit stats", value = "Read commits from
archived files and show file group details")
public String showArchivedCommits(
- @ShellOption(value = {"--archiveFolderPattern"}, help = "Archive
Folder", defaultValue = "") String folder,
+ @ShellOption(value = {"--archiveFolderPattern"},
+ help = "Archive Folder, a folder under the meta path holding archive
files in the"
+ + " legacy log format written before table version 8. When
absent, a table"
+ + " written before table version 8 falls back to its default
archive folder"
+ + " and any other table is read through its archived timeline",
+ defaultValue = "") String folder,
@ShellOption(value = {"--limit"}, help = "Limit commits", defaultValue =
"10") final Integer limit,
@ShellOption(value = {"--sortBy"}, help = "Sorting Field", defaultValue
= "") final String sortByField,
@ShellOption(value = {"--desc"}, help = "Ordering", defaultValue =
"false") final boolean descending,
@@ -107,9 +123,186 @@ public class ArchivedCommitsCommand {
throws IOException {
System.out.println("===============> Showing only " + limit + " archived
commits <===============");
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
- StoragePath archivePath = folder != null && !folder.isEmpty()
- ? new StoragePath(metaClient.getMetaPath(), folder)
- : new StoragePath(metaClient.getArchivePath(), ".commits_.archive*");
+ List<Comparable[]> allStats;
+ if (folder != null && !folder.isEmpty()) {
+ allStats = readCommitStatsFromLegacyArchive(metaClient, new
StoragePath(metaClient.getMetaPath(), folder));
+ } else if (isLegacyArchive(metaClient)) {
+ allStats = readCommitStatsFromLegacyArchive(
+ metaClient, new StoragePath(metaClient.getArchivePath(),
".commits_.archive*"));
+ } else {
+ allStats =
readCommitStatsFromArchivedTimeline(newArchivedTimeline(metaClient),
sortByField, limit);
+ }
+ TableHeader header = new
TableHeader().addTableHeaderField("action").addTableHeaderField("instant")
+
.addTableHeaderField("partition").addTableHeaderField("file_id").addTableHeaderField("prev_instant")
+
.addTableHeaderField("num_writes").addTableHeaderField("num_inserts").addTableHeaderField("num_deletes")
+
.addTableHeaderField("num_update_writes").addTableHeaderField("total_log_files")
+
.addTableHeaderField("total_log_blocks").addTableHeaderField("total_corrupt_log_blocks")
+
.addTableHeaderField("total_rollback_blocks").addTableHeaderField("total_log_records")
+
.addTableHeaderField("total_updated_records_compacted").addTableHeaderField("total_write_bytes")
+ .addTableHeaderField("total_write_errors");
+
+ return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allStats);
+ }
+
+ @ShellMethod(key = "show archived commits", value = "Read commits from
archived files and show details")
+ public String showCommits(
+ @ShellOption(value = {"--skipMetadata"}, help = "Skip displaying commit
metadata",
+ defaultValue = "true") boolean skipMetadata,
+ @ShellOption(value = {"--limit"}, help = "Limit commits", defaultValue =
"10") final Integer limit,
+ @ShellOption(value = {"--sortBy"}, help = "Sorting Field", defaultValue
= "") final String sortByField,
+ @ShellOption(value = {"--desc"}, help = "Ordering", defaultValue =
"false") final boolean descending,
+ @ShellOption(value = {"--headeronly"}, help = "Print Header Only",
+ defaultValue = "false") final boolean headerOnly) {
+
+ System.out.println("===============> Showing only " + limit + " archived
commits <===============");
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ List<Comparable[]> allCommits = readArchivedCommits(
+ newArchivedTimeline(metaClient), skipMetadata,
isLegacyArchive(metaClient), sortByField, limit);
+
+ TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType");
+
+ if (!skipMetadata) {
+ header = header.addTableHeaderField("CommitDetails");
+ }
+
+ return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allCommits);
+ }
+
+ /**
+ * Renders the completed archived instants as rows, loading metadata only
for the rows that
+ * can reach the output.
+ * <p>
+ * Without a sort field the printer keeps the timeline order and cuts at the
limit, so only
+ * the leading instants can be shown, and an archive that has grown for
years holds far more
+ * payload than those few rows need. With a sort field, or no limit, every
row takes part
+ * and everything has to be loaded.
+ */
+ @VisibleForTesting
+ static List<Comparable[]> readArchivedCommits(HoodieArchivedTimeline
archivedTimeline, boolean skipMetadata,
+ boolean legacyArchive, String
sortByField, int limit) {
+ List<HoodieInstant> completed = archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .collect(Collectors.toList());
+ List<HoodieInstant> shown = boundedByLimit(completed, sortByField, limit);
+ if (!skipMetadata) {
+ loadInstantDetails(archivedTimeline, shown, shown.size() <
completed.size());
+ }
+ return shown.stream()
+ .map(instant -> readArchivedCommit(archivedTimeline, instant,
skipMetadata, legacyArchive))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Returns the leading instants that the printer can render, or all of them
when a sort field
+ * or a non-positive limit makes every row a candidate.
+ */
+ private static List<HoodieInstant> boundedByLimit(List<HoodieInstant>
instants, String sortByField, int limit) {
+ if (!sortByField.isEmpty() || limit <= 0 || limit >= instants.size()) {
+ return instants;
+ }
+ return instants.subList(0, limit);
+ }
+
+ /**
+ * Loads the details of the given instants, through the closed time range
they span when they
+ * are a strict subset of the archive, so that the payloads outside the
range are never held.
+ * <p>
+ * On the LSM timeline that table version 8 and above are written with, the
range prunes the
+ * archive files by the instant range in their names, so the files outside
it are never opened.
+ * The legacy layout carries no such range in its file names and reads every
archive file
+ * either way, keeping only the records that fall in the range.
+ */
+ private static void loadInstantDetails(HoodieArchivedTimeline
archivedTimeline, List<HoodieInstant> instants,
+ boolean subset) {
+ if (instants.isEmpty()) {
+ return;
+ }
+ if (subset) {
+ archivedTimeline.loadCompletedInstantDetailsInMemory(
+ instants.get(0).requestedTime(), instants.get(instants.size() -
1).requestedTime());
+ } else {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ }
+ }
+
+ /**
+ * Reads the write stats of the archived commit and delta commit instants
through the
+ * archived timeline, the LSM timeline that table version 8 and above are
written with.
+ * <p>
+ * Without a sort field the printer keeps the timeline order and cuts at the
limit, so the
+ * rows come from a leading run of the write instants. Each of them
contributes a row per
+ * write stat, so their details are loaded a limit-sized window of instants
at a time until
+ * the rows reach the limit, instead of materializing every archived
payload. With a sort
+ * field, or no limit, every row takes part and everything has to be loaded.
+ */
+ @VisibleForTesting
+ static List<Comparable[]>
readCommitStatsFromArchivedTimeline(HoodieArchivedTimeline archivedTimeline,
+ String
sortByField, int limit) {
+ List<HoodieInstant> writes = archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .filter(instant ->
HoodieTimeline.COMMIT_ACTION.equals(instant.getAction())
+ || HoodieTimeline.DELTA_COMMIT_ACTION.equals(instant.getAction()))
+ .collect(Collectors.toList());
+ if (!sortByField.isEmpty() || limit <= 0) {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ return writes.stream()
+ .flatMap(instant -> readWriteStatRows(archivedTimeline, instant))
+ .collect(Collectors.toList());
+ }
+ List<Comparable[]> rows = new ArrayList<>();
+ for (int from = 0; from < writes.size() && rows.size() < limit; from +=
limit) {
+ List<HoodieInstant> window = writes.subList(from, Math.min(from + limit,
writes.size()));
+ loadInstantDetails(archivedTimeline, window, window.size() <
writes.size());
+ window.forEach(instant -> readWriteStatRows(archivedTimeline,
instant).forEach(rows::add));
+ }
+ return rows;
+ }
+
+ /**
+ * Returns whether the table keeps its archived instants in the legacy log
format, the archive
+ * layout of the timeline layout version 1 that table versions before eight
are written with.
+ */
+ private static boolean isLegacyArchive(HoodieTableMetaClient metaClient) {
+ return
metaClient.getTimelineLayoutVersion().compareTo(TimelineLayoutVersion.LAYOUT_VERSION_2)
< 0;
+ }
+
+ /**
+ * Builds an archived timeline that bypasses the meta client cache, so that
an archival
+ * triggered earlier in the same CLI session is visible. The instance is
discarded with the
+ * command, so the instant details it loads need no explicit eviction.
+ */
+ private static HoodieArchivedTimeline
newArchivedTimeline(HoodieTableMetaClient metaClient) {
+ return metaClient.getArchivedTimeline(StringUtils.EMPTY_STRING, false);
+ }
+
+ private static Stream<Comparable[]> readWriteStatRows(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant) {
+ HoodieCommitMetadata metadata;
+ try {
+ metadata = archivedTimeline.readCommitMetadataToAvro(instant);
+ } catch (IOException e) {
+ throw new HoodieException("Failed to read the archived commit metadata
of instant " + instant, e);
+ }
+ if (metadata == null || metadata.getPartitionToWriteStats() == null) {
+ return Stream.empty();
+ }
+ final String action = instant.getAction();
+ final String instantTime = instant.requestedTime();
+ return sortByKey(metadata.getPartitionToWriteStats()).values().stream()
+ .flatMap(List::stream)
+ .map(writeStat -> new Comparable[] {action, instantTime,
writeStat.getPartitionPath(),
+ writeStat.getFileId(), writeStat.getPrevCommit(),
writeStat.getNumWrites(),
+ writeStat.getNumInserts(), writeStat.getNumDeletes(),
writeStat.getNumUpdateWrites(),
+ writeStat.getTotalLogFiles(), writeStat.getTotalLogBlocks(),
writeStat.getTotalCorruptLogBlock(),
+ writeStat.getTotalRollbackBlocks(), writeStat.getTotalLogRecords(),
+ writeStat.getTotalUpdatedRecordsCompacted(),
writeStat.getTotalWriteBytes(),
+ writeStat.getTotalWriteErrors()});
+ }
+
+ /**
+ * Reads the write stats of the archived commit and delta commit instants
from the given
+ * folder of archive files in the legacy log format written before table
version 8.
+ */
+ private List<Comparable[]>
readCommitStatsFromLegacyArchive(HoodieTableMetaClient metaClient, StoragePath
archivePath) throws IOException {
HoodieStorage storage = metaClient.getStorage();
List<StoragePathInfo> pathInfoList = storage.globEntries(archivePath);
List<Comparable[]> allStats = new ArrayList<>();
@@ -126,12 +319,15 @@ public class ArchivedCommitsCommand {
List<Comparable[]> readCommits = readRecords.stream().map(r ->
(GenericRecord) r)
.filter(r ->
r.get("actionType").toString().equals(HoodieTimeline.COMMIT_ACTION)
||
r.get("actionType").toString().equals(HoodieTimeline.DELTA_COMMIT_ACTION))
+ // the legacy archive holds an entry per instant state; only the
completed one carries
+ // the final write stats, and the requested entry of a writer
carries no metadata at all
+ .filter(r -> isCompletedEntry(r) && r.get("hoodieCommitMetadata")
!= null)
.flatMap(r -> {
HoodieCommitMetadata metadata = (HoodieCommitMetadata)
SpecificData.get()
.deepCopy(HoodieCommitMetadata.SCHEMA$,
r.get("hoodieCommitMetadata"));
final String instantTime = r.get("commitTime").toString();
final String action = r.get("actionType").toString();
- return
metadata.getPartitionToWriteStats().values().stream().flatMap(hoodieWriteStats
-> hoodieWriteStats.stream().map(hoodieWriteStat -> {
+ return
sortByKey(metadata.getPartitionToWriteStats()).values().stream().flatMap(hoodieWriteStats
-> hoodieWriteStats.stream().map(hoodieWriteStat -> {
List<Comparable> row = new ArrayList<>();
row.add(action);
row.add(instantTime);
@@ -156,94 +352,89 @@ public class ArchivedCommitsCommand {
allStats.addAll(readCommits);
}
}
- TableHeader header = new
TableHeader().addTableHeaderField("action").addTableHeaderField("instant")
-
.addTableHeaderField("partition").addTableHeaderField("file_id").addTableHeaderField("prev_instant")
-
.addTableHeaderField("num_writes").addTableHeaderField("num_inserts").addTableHeaderField("num_deletes")
-
.addTableHeaderField("num_update_writes").addTableHeaderField("total_log_files")
-
.addTableHeaderField("total_log_blocks").addTableHeaderField("total_corrupt_log_blocks")
-
.addTableHeaderField("total_rollback_blocks").addTableHeaderField("total_log_records")
-
.addTableHeaderField("total_updated_records_compacted").addTableHeaderField("total_write_bytes")
- .addTableHeaderField("total_write_errors");
+ return allStats;
+ }
- return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allStats);
+ private static boolean isCompletedEntry(GenericRecord archivedEntry) {
+ // entries written before the state was recorded are completed ones
+ Object actionState = archivedEntry.get("actionState");
+ return actionState == null ||
HoodieInstant.State.COMPLETED.name().equals(actionState.toString());
}
- @ShellMethod(key = "show archived commits", value = "Read commits from
archived files and show details")
- public String showCommits(
- @ShellOption(value = {"--skipMetadata"}, help = "Skip displaying commit
metadata",
- defaultValue = "true") boolean skipMetadata,
- @ShellOption(value = {"--limit"}, help = "Limit commits", defaultValue =
"10") final Integer limit,
- @ShellOption(value = {"--sortBy"}, help = "Sorting Field", defaultValue
= "") final String sortByField,
- @ShellOption(value = {"--desc"}, help = "Ordering", defaultValue =
"false") final boolean descending,
- @ShellOption(value = {"--headeronly"}, help = "Print Header Only",
- defaultValue = "false") final boolean headerOnly)
- throws IOException {
+ private static Comparable[] readArchivedCommit(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant,
+ boolean skipMetadata, boolean
legacyArchive) {
+ List<Comparable> commitDetails = new ArrayList<>();
+ commitDetails.add(instant.requestedTime());
+ commitDetails.add(instant.getAction());
+ if (!skipMetadata) {
+ commitDetails.add(readArchivedMetadataString(archivedTimeline, instant,
legacyArchive));
+ }
+ return commitDetails.toArray(new Comparable[commitDetails.size()]);
+ }
- System.out.println("===============> Showing only " + limit + " archived
commits <===============");
- HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
- StoragePath archivePath =
- new StoragePath(metaClient.getArchivePath(), ".commits_.archive*");
- HoodieStorage storage = metaClient.getStorage();
- List<StoragePathInfo> pathInfoList = storage.globEntries(archivePath);
- List<Comparable[]> allCommits = new ArrayList<>();
- for (StoragePathInfo pathInfo : pathInfoList) {
- // read the archived file
- try (HoodieLogFormat.Reader reader =
HoodieLogFormat.newReader(metaClient,
- new HoodieLogFile(pathInfo.getPath()),
HoodieSchema.fromAvroSchema(HoodieArchivedMetaEntry.getClassSchema()))) {
- List<IndexedRecord> readRecords = new ArrayList<>();
- // read the avro blocks
- while (reader.hasNext()) {
- HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next();
- try (ClosableIterator<HoodieRecord<IndexedRecord>> recordItr =
blk.getRecordIterator(HoodieRecordType.AVRO)) {
- recordItr.forEachRemaining(r -> readRecords.add(r.getData()));
- }
- }
- List<Comparable[]> readCommits = readRecords.stream().map(r ->
(GenericRecord) r)
- .map(r -> readCommit(r,
skipMetadata)).collect(Collectors.toList());
- allCommits.addAll(readCommits);
+ private static String readArchivedMetadataString(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant,
+ boolean legacyArchive) {
+ Option<byte[]> details = archivedTimeline.getInstantDetails(instant);
+ if (!details.isPresent() || details.get().length == 0) {
+ // instants can be archived with no metadata, e.g. from an empty
completed
+ // meta file that a writer failure left behind
+ return "{}";
+ }
+ if (legacyArchive) {
+ // ArchivedTimelineV1 caches the JSON rendering of each archived entry,
which is what the
+ // legacy reader printed, and the v1 serde cannot decode it back into
the Avro classes
+ return new String(details.get(), StandardCharsets.UTF_8);
+ }
+ try {
+ // TimelineArchiverV2 writes only the actions with a typed reader below,
but the upgrade to the
+ // LSM timeline copies every entry of a legacy archive across,
savepoints, compaction plans and
+ // index metadata included, and takes an entry written before the action
state was recorded as
+ // completed. Every such payload is an Avro file that carries its own
schema
+ switch (instant.getAction()) {
+ case HoodieTimeline.CLEAN_ACTION:
+ return archivedTimeline.readCleanMetadata(instant).toString();
+ case HoodieTimeline.COMMIT_ACTION:
+ case HoodieTimeline.DELTA_COMMIT_ACTION:
+ return
sortPartitions(archivedTimeline.readCommitMetadataToAvro(instant)).toString();
+ case HoodieTimeline.ROLLBACK_ACTION:
+ return archivedTimeline.readRollbackMetadata(instant).toString();
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ return
sortPartitions(archivedTimeline.readReplaceCommitMetadataToAvro(instant)).toString();
+ default:
+ return readAvroPayload(details.get());
}
+ } catch (IOException e) {
+ throw new HoodieException("Failed to read the archived metadata of
instant " + instant, e);
}
+ }
- TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType");
-
- if (!skipMetadata) {
- header = header.addTableHeaderField("CommitDetails");
+ /**
+ * Renders an archived payload through the schema embedded in it, which is
the rendering the
+ * typed Avro records give of themselves.
+ */
+ private static String readAvroPayload(byte[] payload) throws IOException {
+ try (DataFileStream<GenericRecord> reader =
+ new DataFileStream<>(new ByteArrayInputStream(payload), new
GenericDatumReader<>())) {
+ return reader.hasNext() ? reader.next().toString() : "{}";
}
-
- return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allCommits);
}
- private Comparable[] commitDetail(GenericRecord record, String metadataName,
boolean skipMetadata) {
- List<Object> commitDetails = new ArrayList<>();
- commitDetails.add(record.get("commitTime"));
- commitDetails.add(record.get("actionType").toString());
- if (!skipMetadata) {
-
commitDetails.add(Option.ofNullable(record.get(metadataName)).orElse("{}").toString());
- }
- return commitDetails.toArray(new Comparable[commitDetails.size()]);
+ /**
+ * Orders the partition keyed maps of the metadata so that the rendering
does not
+ * depend on the iteration order of the map the Avro decoder happens to
build.
+ */
+ private static HoodieCommitMetadata sortPartitions(HoodieCommitMetadata
metadata) {
+
metadata.setPartitionToWriteStats(sortByKey(metadata.getPartitionToWriteStats()));
+ return metadata;
}
- private Comparable[] readCommit(GenericRecord record, boolean skipMetadata) {
- String actionType = record.get("actionType").toString();
- switch (actionType) {
- case HoodieTimeline.CLEAN_ACTION:
- return commitDetail(record, "hoodieCleanMetadata", skipMetadata);
- case HoodieTimeline.COMMIT_ACTION:
- case HoodieTimeline.DELTA_COMMIT_ACTION:
- return commitDetail(record, "hoodieCommitMetadata", skipMetadata);
- case HoodieTimeline.ROLLBACK_ACTION:
- return commitDetail(record, "hoodieRollbackMetadata", skipMetadata);
- case HoodieTimeline.SAVEPOINT_ACTION:
- return commitDetail(record, "hoodieSavePointMetadata", skipMetadata);
- case HoodieTimeline.COMPACTION_ACTION:
- return commitDetail(record, "hoodieCompactionMetadata", skipMetadata);
- case HoodieTimeline.REPLACE_COMMIT_ACTION:
- case HoodieTimeline.CLUSTERING_ACTION:
- return commitDetail(record, "hoodieReplaceCommitMetadata",
skipMetadata);
- default: {
- throw new HoodieException("Unexpected action type: " + actionType);
- }
- }
+ private static HoodieReplaceCommitMetadata
sortPartitions(HoodieReplaceCommitMetadata metadata) {
+
metadata.setPartitionToWriteStats(sortByKey(metadata.getPartitionToWriteStats()));
+
metadata.setPartitionToReplaceFileIds(sortByKey(metadata.getPartitionToReplaceFileIds()));
+ return metadata;
}
+ private static <T> Map<String, T> sortByKey(Map<String, T>
partitionKeyedMap) {
+ return partitionKeyedMap == null ? null : new TreeMap<>(partitionKeyedMap);
+ }
}
diff --git
a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CleansCommand.java
b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CleansCommand.java
index 704e11719791..519f56236863 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CleansCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/CleansCommand.java
@@ -45,6 +45,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.TreeMap;
import java.util.stream.Collectors;
import scala.collection.JavaConverters;
@@ -108,7 +109,10 @@ public class CleansCommand {
HoodieCleanMetadata cleanMetadata =
timeline.readCleanMetadata(cleanInstant);
List<Comparable[]> rows = new ArrayList<>();
- for (Map.Entry<String, HoodieCleanPartitionMetadata> entry :
cleanMetadata.getPartitionMetadata().entrySet()) {
+ // the metadata's partition map carries no order of its own, and the
printer sorts only when
+ // --sortBy is given, so the rows are keyed in partition order to keep the
output stable
+ for (Map.Entry<String, HoodieCleanPartitionMetadata> entry :
+ new TreeMap<>(cleanMetadata.getPartitionMetadata()).entrySet()) {
String path = entry.getKey();
HoodieCleanPartitionMetadata stats = entry.getValue();
String policy = stats.getPolicy();
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 1f3fbb51d9e6..28fca03f0558 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
@@ -31,6 +31,7 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.TimelineFactory;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.NumericUtils;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
@@ -150,10 +151,16 @@ public class FileSystemViewCommand {
fileSliceStream = fsView.getLatestFileSlices(partition);
} else {
if (maxInstant.isEmpty()) {
- maxInstant =
HoodieCLI.getTableMetaClient().getActiveTimeline().filterCompletedAndCompactionInstants().lastInstant()
- .get().requestedTime();
+ Option<HoodieInstant> lastInstant =
HoodieCLI.getTableMetaClient().getActiveTimeline().filterCompletedAndCompactionInstants().lastInstant();
+ if (lastInstant.isPresent()) {
+ maxInstant = lastInstant.get().requestedTime();
+ }
+ }
+ if (maxInstant.isEmpty()) {
+ fileSliceStream = fsView.getLatestFileSlices(partition);
+ } else {
+ fileSliceStream =
fsView.getLatestMergedFileSlicesBeforeOrOn(partition, maxInstant);
}
- fileSliceStream = fsView.getLatestMergedFileSlicesBeforeOrOn(partition,
maxInstant);
}
fileSliceStream.forEach(fs -> {
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 872ff342dce5..5f7bb2525485 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
@@ -29,6 +29,7 @@ import org.apache.hudi.common.model.HoodiePartitionMetadata;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.TimelineUtils;
import org.apache.hudi.common.util.CleanerUtils;
@@ -36,10 +37,10 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.PartitionPathEncodeUtils;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.io.util.FileIOUtils;
import org.apache.hudi.storage.StoragePath;
import lombok.extern.slf4j.Slf4j;
-import org.apache.avro.AvroRuntimeException;
import org.apache.spark.launcher.SparkLauncher;
import org.apache.spark.sql.hudi.DeDupeType;
import org.apache.spark.util.Utils;
@@ -47,8 +48,10 @@ import org.springframework.shell.standard.ShellComponent;
import org.springframework.shell.standard.ShellMethod;
import org.springframework.shell.standard.ShellOption;
+import java.io.ByteArrayInputStream;
import java.io.FileInputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -189,29 +192,71 @@ public class RepairsCommand {
@ShellMethod(key = "repair corrupted clean files", value = "repair corrupted
clean files")
public void removeCorruptedPendingCleanAction() {
+ removeCorruptedPendingCleanAction(HoodieCLI.getTableMetaClient());
+ }
- HoodieTableMetaClient client = HoodieCLI.getTableMetaClient();
- HoodieTimeline cleanerTimeline =
HoodieCLI.getTableMetaClient().getActiveTimeline().getCleanerTimeline();
+ /**
+ * Removes the pending clean instants whose plan is verifiably empty or
corrupt.
+ * <p>
+ * The plan bytes are read in full before anything is judged. The timeline
serde wraps every
+ * exception raised while it streams an instant file, a transient read
failure included, in
+ * the same "unable to read commit metadata" IOException that an empty or
truncated file
+ * raises, so the message cannot tell a storage outage from corruption. A
failure of the
+ * read itself is therefore propagated, and only the in-memory decode, which
no I/O can
+ * disturb, decides that the plan is corrupt.
+ */
+ static void removeCorruptedPendingCleanAction(HoodieTableMetaClient client) {
+ HoodieActiveTimeline activeTimeline = client.getActiveTimeline();
+ HoodieTimeline cleanerTimeline = activeTimeline.getCleanerTimeline();
log.info("Inspecting pending clean metadata in timeline for corrupted
files");
cleanerTimeline.filterInflightsAndRequested().getInstants().forEach(instant -> {
- try {
- CleanerUtils.getCleanerPlan(client, instant);
- } catch (AvroRuntimeException e) {
- log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
+ HoodieInstant planInstant = CleanerUtils.getCleanRequestInstant(client,
instant);
+ byte[] plan;
+ try (InputStream in =
activeTimeline.getInstantContentStream(planInstant)) {
+ plan = FileIOUtils.readAsByteArray(in);
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read the plan of pending clean
instant " + instant, e);
+ }
+ if (plan.length > 0 && isReadableCleanerPlan(client, plan)) {
+ return;
+ }
+ log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
+ // An inflight clean keeps its plan in the requested file, so removing
only the instant the
+ // timeline listed would leave the corrupt plan behind for the next
clean to fail on. The
+ // inflight file goes first: a failure between the two deletes leaves
the action requested,
+ // which is a state this command already handles, whereas the reverse
order would leave an
+ // inflight action with no plan at all.
+ TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
+ instant, client.getInstantFileNameGenerator());
+ if (!instant.equals(planInstant)) {
TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
- instant, client.getInstantFileNameGenerator());
- } catch (IOException ioe) {
- if (ioe.getMessage().contains("Not an Avro data file")) {
- log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
- TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
- instant, client.getInstantFileNameGenerator());
- } else {
- throw new HoodieIOException(ioe.getMessage(), ioe);
- }
+ planInstant, client.getInstantFileNameGenerator());
}
});
}
+ /**
+ * Decodes a clean plan held in memory, which fails only on the content
itself: bytes that are not
+ * an Avro file or that stop short fail the read, and an Avro container that
holds no record, which
+ * a writer killed between opening and closing it leaves behind, fails the
serde's argument check.
+ * <p>
+ * Every failure is caught, because corrupt bytes do not reach the decoder
through one exception
+ * type: a length that survives as far as Avro's own ceiling raises an {@code
+ * UnsupportedOperationException} rather than an {@code
AvroRuntimeException}, and a plan that
+ * decodes with no version leaves the migrator to unbox a null. Nothing
inside the try touches
+ * storage -- the bytes are already in memory and both clean plan migration
handlers are pure
+ * transforms -- so there is no I/O failure here to mistake for corruption,
and letting one of
+ * these escape would abandon the repair of every instant behind this one.
+ */
+ private static boolean isReadableCleanerPlan(HoodieTableMetaClient client,
byte[] plan) {
+ try {
+ CleanerUtils.getCleanerPlan(client, new ByteArrayInputStream(plan));
+ return true;
+ } catch (IOException | RuntimeException e) {
+ return false;
+ }
+ }
+
@ShellMethod(key = "repair show empty commit metadata", value = "show failed
commits")
public void showFailedCommits() {
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
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 9f1c94af6026..121c1f8b4231 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
@@ -37,6 +37,7 @@ import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.PartitionPathEncodeUtils;
import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.config.HoodieBootstrapConfig;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieIndexConfig;
@@ -623,7 +624,8 @@ public class SparkMain {
}
}
- private static int archive(JavaSparkContext jsc, int minCommits, int
maxCommits, int commitsRetained, boolean enableMetadata, String basePath) {
+ @VisibleForTesting
+ protected static int archive(JavaSparkContext jsc, int minCommits, int
maxCommits, int commitsRetained, boolean enableMetadata, String basePath) {
try {
return ArchiveExecutorUtils.archive(jsc, minCommits, maxCommits,
commitsRetained, enableMetadata, basePath, new HashMap<>());
} catch (IOException ex) {
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 209d3744dfcd..6c2d829708c5 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
@@ -21,25 +21,17 @@ 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.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
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 static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
@Tag("functional")
-@SpringBootTest(properties = {"spring.shell.interactive.enabled=false",
"spring.shell.command.script.enabled=false"})
public class TestArchiveCommand extends CLIFunctionalTestHarness {
- @Autowired
- private Shell shell;
-
@Test
public void testArchiving() throws Exception {
HoodieCLI.conf = storageConf();
@@ -50,7 +42,7 @@ public class TestArchiveCommand extends
CLIFunctionalTestHarness {
new TableCommand().createTable(
tablePath, tableName,
- "COPY_ON_WRITE", "", 1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "COPY_ON_WRITE", "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
@@ -60,20 +52,27 @@ public class TestArchiveCommand extends
CLIFunctionalTestHarness {
HoodieTestCommitMetadataGenerator.createCommitFileWithMetadata(tablePath,
timestamp, storageConf());
}
- Object cmdResult = shell.evaluate(() -> "trigger archival --minCommits 2
--maxCommits 3 --commitsRetainedByCleaner 1 --enableMetadata false");
- assertTrue(ShellEvaluationResultUtil.isSuccess(cmdResult));
+ // The shell command "trigger archival" launches SparkMain in a separate
spark-submit
+ // process, which needs SPARK_HOME and the jars packaged under target/lib
(see the
+ // ITTest classes), neither of which exists when the functional suite
runs. Invoke the
+ // entry point that process dispatches to for the ARCHIVE command instead,
with the
+ // arguments the shell invocation used:
+ // trigger archival --minCommits 2 --maxCommits 3
--commitsRetainedByCleaner 1 --enableMetadata false
+ // The same 6-commit / 2-3-1 scenario against ArchiveExecutorUtils.archive
is already covered by
+ // TestArchiveCommitsProcedure in hudi-spark, so what this test adds is
coverage of the CLI wrapper.
+ assertEquals(0, SparkMain.archive(jsc(), 2, 3, 1, false, tablePath));
+
metaClient = HoodieTableMetaClient.reload(metaClient);
- //get instants in the active timeline only returns the latest state of the
commit
- //therefore we expect 2 instants because minCommits is 2
+ // 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().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().countInstants());
+ // 6 commits - 2 kept in the active timeline = 4 archived. The LSM
archived timeline
+ // of table version 8 and above holds a single entry per instant (the
legacy log
+ // format archived requested, inflight and completed as separate entries),
so the
+ // archived timeline counts 4 instants.
+ assertEquals(4, metaClient.getArchivedTimeline().countInstants());
}
}
-
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchivedCommitsCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchivedCommitsCommand.java
index 28b1ecd8a8aa..7d476c51dc96 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchivedCommitsCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchivedCommitsCommand.java
@@ -18,6 +18,11 @@
package org.apache.hudi.cli.commands;
+import org.apache.hudi.avro.model.HoodieActionInstant;
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
+import org.apache.hudi.avro.model.HoodieRequestedReplaceMetadata;
+import org.apache.hudi.avro.model.HoodieSavepointMetadata;
import org.apache.hudi.cli.HoodieCLI;
import org.apache.hudi.cli.HoodiePrintHelper;
import org.apache.hudi.cli.TableHeader;
@@ -25,17 +30,37 @@ import
org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
import org.apache.hudi.cli.testutils.HoodieTestCommitMetadataGenerator;
import org.apache.hudi.cli.testutils.HoodieTestCommitUtilities;
import org.apache.hudi.cli.testutils.ShellEvaluationResultUtil;
+import org.apache.hudi.client.timeline.ActiveActionWithDetails;
import org.apache.hudi.client.timeline.HoodieTimelineArchiver;
+import org.apache.hudi.client.timeline.LSMTimelineWriter;
+import org.apache.hudi.client.timeline.TimelineArchiverV1;
import org.apache.hudi.client.timeline.TimelineArchiverV2;
+import org.apache.hudi.common.HoodieCleanStat;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
+import org.apache.hudi.common.model.HoodieCleaningPolicy;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.MetadataConversionUtils;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import
org.apache.hudi.common.table.timeline.versioning.clean.CleanPlanV2MigrationHandler;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
-import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.util.Option;
+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.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieSparkTable;
import org.junit.jupiter.api.BeforeEach;
@@ -45,12 +70,22 @@ 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.OutputStream;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
-import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
+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.util.CleanerUtils.convertCleanMetadata;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -60,10 +95,22 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@SpringBootTest(properties = {"spring.shell.interactive.enabled=false",
"spring.shell.command.script.enabled=false"})
public class TestArchivedCommitsCommand extends CLIFunctionalTestHarness {
+ /**
+ * The clean and the replace commit that the fixture archives along with the
plain commits.
+ * The clean has to be older than the newest archived write instant for
TimelineArchiverV2 to
+ * pick it up, and its earliest instant to retain has to be past the replace
commit so that the
+ * clustering guard of the archiver does not hold the replace commit back.
+ */
+ private static final String CLEAN_INSTANT = "101";
+ private static final String REPLACE_COMMIT_INSTANT = "102";
+ private static final String EARLIEST_COMMIT_TO_RETAIN = "103";
+
@Autowired
private Shell shell;
private String tablePath;
+ private HoodieCleanMetadata cleanMetadata;
+ private HoodieReplaceCommitMetadata replaceCommitMetadata;
@BeforeEach
public void init() throws Exception {
@@ -75,7 +122,7 @@ public class TestArchivedCommitsCommand extends
CLIFunctionalTestHarness {
new TableCommand().createTable(
tablePath, tableName,
- "COPY_ON_WRITE", "", 1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "COPY_ON_WRITE", "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
@@ -86,24 +133,20 @@ public class TestArchivedCommitsCommand extends
CLIFunctionalTestHarness {
.withCleanConfig(HoodieCleanConfig.newBuilder().retainCommits(1).build())
.withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
.withRemoteServerPort(timelineServicePort).build())
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
.forTable("test-trip-table").build();
- // Create six commits
- for (int i = 100; i < 106; i++) {
+ // Create seven write instants (a commit, a replace commit and five more
commits) plus one
+ // clean, so that archival keeps the four newest write instants and
archives 100, 101, 102
+ // and 103
+ HoodieTestCommitMetadataGenerator.createCommitFileWithMetadata(tablePath,
"100", storageConf());
+ cleanMetadata = addClean(metaClient, CLEAN_INSTANT,
EARLIEST_COMMIT_TO_RETAIN);
+ replaceCommitMetadata = addReplaceCommit(metaClient,
REPLACE_COMMIT_INSTANT);
+ for (int i = 103; i < 108; i++) {
String timestamp = String.valueOf(i);
- // Requested Compaction
-
HoodieTestCommitMetadataGenerator.createCompactionAuxiliaryMetadata(tablePath,
- INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.COMPACTION_ACTION, timestamp), storageConf());
- // Inflight Compaction
-
HoodieTestCommitMetadataGenerator.createCompactionAuxiliaryMetadata(tablePath,
- INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.COMPACTION_ACTION, timestamp), storageConf());
HoodieTestCommitMetadataGenerator.createCommitFileWithMetadata(tablePath,
timestamp, storageConf());
}
- // Simulate a compaction commit in metadata table timeline
- // so the archival in data table can happen
- HoodieTestUtils.createCompactionCommitInMetadataTable(storageConf(),
tablePath, "105");
-
metaClient = HoodieTableMetaClient.reload(metaClient);
// reload the timeline and get all the commits before archive
metaClient.getActiveTimeline().reload().getAllCommitsTimeline().filterCompletedInstants();
@@ -131,31 +174,13 @@ public class TestArchivedCommitsCommand extends
CLIFunctionalTestHarness {
.addTableHeaderField("total_updated_records_compacted").addTableHeaderField("total_write_bytes")
.addTableHeaderField("total_write_errors");
- // Generate expected data
+ // Generate expected data: the stats only cover the archived commit and
delta commit
+ // instants, so the archived clean and replace commit contribute no row.
The archived
+ // timeline holds one entry per instant, with one write stat for each of
the two
+ // partitions, ordered by partition path
final List<Comparable[]> rows = new ArrayList<>();
- for (int i = 100; i < 102; i++) {
- String instant = String.valueOf(i);
- for (int j = 0; j < 3; j++) {
- Comparable[] defaultComp = new Comparable[] {"commit", instant,
- HoodieTestCommitMetadataGenerator.DEFAULT_SECOND_PARTITION_PATH,
- HoodieTestCommitMetadataGenerator.DEFAULT_FILEID,
- HoodieTestCommitMetadataGenerator.DEFAULT_PRE_COMMIT,
- HoodieTestCommitMetadataGenerator.DEFAULT_NUM_WRITES,
- HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
- HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
- HoodieTestCommitMetadataGenerator.DEFAULT_NUM_UPDATE_WRITES,
- HoodieTestCommitMetadataGenerator.DEFAULT_NULL_VALUE,
- HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_LOG_BLOCKS,
- HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
- HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
- HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_LOG_RECORDS,
- HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
- HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_WRITE_BYTES,
- HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE};
- rows.add(defaultComp.clone());
- defaultComp[2] =
HoodieTestCommitMetadataGenerator.DEFAULT_FIRST_PARTITION_PATH;
- rows.add(defaultComp);
- }
+ for (String instant : Arrays.asList("100", "103")) {
+ rows.addAll(writeStatRows("commit", instant));
}
String expectedResult = HoodiePrintHelper.print(
@@ -174,15 +199,13 @@ public class TestArchivedCommitsCommand extends
CLIFunctionalTestHarness {
assertTrue(ShellEvaluationResultUtil.isSuccess(cmdResult));
final List<Comparable[]> rows = new ArrayList<>();
- // Test default skipMetadata and limit 5
+ // Test default skipMetadata and limit 5. The archived timeline holds one
entry
+ // per instant, so each instant shows up as a single row
TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType");
- Comparable[] result1 = new Comparable[] {"100", "commit"};
- Comparable[] result2 = new Comparable[] {"101", "commit"};
- rows.add(result1);
- rows.add(result1);
- rows.add(result1);
- rows.add(result2);
- rows.add(result2);
+ rows.add(new Comparable[] {"100", "commit"});
+ rows.add(new Comparable[] {CLEAN_INSTANT, "clean"});
+ rows.add(new Comparable[] {REPLACE_COMMIT_INSTANT, "replacecommit"});
+ rows.add(new Comparable[] {"103", "commit"});
String expected = HoodiePrintHelper.print(header, new HashMap<>(), "",
false, 5, false, rows);
expected = removeNonWordAndStripSpace(expected);
String got = removeNonWordAndStripSpace(cmdResult.toString());
@@ -194,20 +217,309 @@ public class TestArchivedCommitsCommand extends
CLIFunctionalTestHarness {
rows.clear();
- for (int i = 100; i < 102; i++) {
- String instant = String.valueOf(i);
- // Since HoodiePrintHelper order data by default, need to order
commitMetadata
- HoodieCommitMetadata metadata =
HoodieTestCommitMetadataGenerator.generateCommitMetadata(tablePath, instant);
- Comparable[] result = new Comparable[] {
- instant, "commit",
HoodieTestCommitUtilities.convertAndOrderCommitMetadata(metadata)};
- rows.add(result);
- rows.add(result);
- rows.add(result);
- }
+ // Since HoodiePrintHelper order data by default, need to order
commitMetadata
+ rows.add(new Comparable[] {"100", "commit", commitDetails("100")});
+ rows.add(new Comparable[] {CLEAN_INSTANT, "clean", cleanMetadata});
+ rows.add(new Comparable[] {REPLACE_COMMIT_INSTANT, "replacecommit",
orderedAvroReplaceCommitMetadata()});
+ rows.add(new Comparable[] {"103", "commit", commitDetails("103")});
header = header.addTableHeaderField("CommitDetails");
expected = HoodiePrintHelper.print(header, new HashMap<>(), "", false, 0,
false, rows);
expected = removeNonWordAndStripSpace(expected);
got = removeNonWordAndStripSpace(cmdResult.toString());
assertEquals(expected, got);
+
+ // Test with Metadata and a limit, the path that loads only the rendered
instants
+ cmdResult = shell.evaluate(() -> "show archived commits --skipMetadata
false --limit 2");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(cmdResult));
+
+ rows.clear();
+ rows.add(new Comparable[] {"100", "commit", commitDetails("100")});
+ rows.add(new Comparable[] {CLEAN_INSTANT, "clean", cleanMetadata});
+ expected = HoodiePrintHelper.print(header, new HashMap<>(), "", false, 2,
false, rows);
+ expected = removeNonWordAndStripSpace(expected);
+ got = removeNonWordAndStripSpace(cmdResult.toString());
+ assertEquals(expected, got);
+ }
+
+ /**
+ * With no sort field the printer only renders the leading rows, so the
metadata of the
+ * instants past the limit is never loaded. isEmpty(instant) on the archived
timeline reports
+ * whether that instant's details are held in memory.
+ */
+ @Test
+ public void testShowCommitsLoadsOnlyTheRenderedInstants() {
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ HoodieArchivedTimeline archivedTimeline =
metaClient.getArchivedTimeline("", false);
+ Map<String, HoodieInstant> byTime = archivedTimeline.getInstants().stream()
+ .collect(Collectors.toMap(HoodieInstant::requestedTime, instant ->
instant));
+
+ List<Comparable[]> rows =
ArchivedCommitsCommand.readArchivedCommits(archivedTimeline, false, false, "",
2);
+ assertEquals(Arrays.asList("100", CLEAN_INSTANT),
+ rows.stream().map(row -> row[0]).collect(Collectors.toList()));
+ assertFalse(archivedTimeline.isEmpty(byTime.get("100")));
+ assertFalse(archivedTimeline.isEmpty(byTime.get(CLEAN_INSTANT)));
+ assertTrue(archivedTimeline.isEmpty(byTime.get(REPLACE_COMMIT_INSTANT)));
+ assertTrue(archivedTimeline.isEmpty(byTime.get("103")));
+
+ // a sort field makes every row a candidate, so everything is loaded
+ archivedTimeline = metaClient.getArchivedTimeline("", false);
+ rows = ArchivedCommitsCommand.readArchivedCommits(archivedTimeline, false,
false, "CommitTime", 2);
+ assertEquals(4, rows.size());
+ assertFalse(archivedTimeline.isEmpty(byTime.get("103")));
+
+ // the stats fill a limit of one row from the two write stats of instant
100 and stop there
+ archivedTimeline = metaClient.getArchivedTimeline("", false);
+ List<Comparable[]> stats =
ArchivedCommitsCommand.readCommitStatsFromArchivedTimeline(archivedTimeline,
"", 1);
+ assertEquals(2, stats.size());
+ assertTrue(archivedTimeline.isEmpty(byTime.get("103")));
+ }
+
+ /**
+ * The upgrade to the LSM timeline copies every entry of a legacy archive
across, so the archived
+ * timeline can hold actions that TimelineArchiverV2 never writes, such as a
savepoint. Their
+ * payload renders through the schema it carries instead of failing the
listing.
+ */
+ @Test
+ public void testShowCommitsRendersActionsWithoutTypedReader() throws
Exception {
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ HoodieSavepointMetadata savepointMetadata =
TimelineMetadataUtils.convertSavepointMetadata(
+ "user", "savepoint copied across by the upgrade",
+ Collections.singletonMap(DEFAULT_FIRST_PARTITION_PATH,
Collections.singletonList("file-1")));
+ HoodieInstant savepoint =
metaClient.getInstantGenerator().createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.SAVEPOINT_ACTION, "099",
"099");
+ byte[] payload =
TimelineMetadataUtils.serializeAvroMetadata(savepointMetadata,
HoodieSavepointMetadata.class).get();
+ HoodieWriteConfig cfg = HoodieWriteConfig.newBuilder().withPath(tablePath)
+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ .withMarkersType("DIRECT").build();
+ LSMTimelineWriter.getInstance(cfg, new LocalTaskContextSupplier(),
metaClient).write(
+
Collections.singletonList(ActiveActionWithDetails.fromInstantAndDetails(
+ Collections.singletonList(Pair.of(savepoint,
Option.of(payload))))),
+ Option.empty(), Option.empty());
+
+ Object cmdResult = shell.evaluate(() -> "show archived commits
--skipMetadata false --limit 0");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(cmdResult));
+
+ TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType")
+ .addTableHeaderField("CommitDetails");
+ final List<Comparable[]> rows = new ArrayList<>();
+ rows.add(new Comparable[] {"099", "savepoint", savepointMetadata});
+ rows.add(new Comparable[] {"100", "commit", commitDetails("100")});
+ rows.add(new Comparable[] {CLEAN_INSTANT, "clean", cleanMetadata});
+ rows.add(new Comparable[] {REPLACE_COMMIT_INSTANT, "replacecommit",
orderedAvroReplaceCommitMetadata()});
+ rows.add(new Comparable[] {"103", "commit", commitDetails("103")});
+ String expected = removeNonWordAndStripSpace(
+ HoodiePrintHelper.print(header, new HashMap<>(), "", false, 0, false,
rows));
+ assertEquals(expected, removeNonWordAndStripSpace(cmdResult.toString()));
+ }
+
+ /**
+ * Test for both show archived commands against a table created at version
6, whose archived
+ * instants live in the legacy log format under the archive folder instead
of in an LSM
+ * timeline.
+ */
+ @Test
+ public void testShowArchivedCommitsOnLegacyArchive() throws Exception {
+ String legacyTableName = tableName("_legacy_table");
+ String legacyTablePath = tablePath(legacyTableName);
+
+ // createTable also connects the CLI to the new table
+ new TableCommand().createTable(
+ legacyTablePath, legacyTableName, "COPY_ON_WRITE", "",
+ HoodieTableVersion.SIX.versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
+
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ HoodieWriteConfig cfg =
HoodieWriteConfig.newBuilder().withPath(legacyTablePath)
+
.withSchema(HoodieTestCommitMetadataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2,
2)
+ .withWriteTableVersion(HoodieTableVersion.SIX.versionCode())
+ .withAutoUpgradeVersion(false)
+
.withArchivalConfig(HoodieArchivalConfig.newBuilder().archiveCommitsWith(4,
5).build())
+
.withCleanConfig(HoodieCleanConfig.newBuilder().retainCommits(1).build())
+ .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
+ .withRemoteServerPort(timelineServicePort).build())
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
+ .forTable(legacyTableName).build();
+
+ // Six commits, of which archival keeps the four newest and archives 100
and 101
+ List<HoodieCommitMetadata> commitMetadata = new ArrayList<>();
+ for (int i = 100; i < 106; i++) {
+ commitMetadata.add(createLegacyCommit(metaClient, legacyTablePath,
String.valueOf(i)));
+ }
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieSparkTable table = HoodieSparkTable.create(cfg, context(),
metaClient);
+ HoodieTimelineArchiver archiver = new TimelineArchiverV1(cfg, table);
+ archiver.archiveIfRequired(context());
+
+ TableHeader statsHeader = new
TableHeader().addTableHeaderField("action").addTableHeaderField("instant")
+
.addTableHeaderField("partition").addTableHeaderField("file_id").addTableHeaderField("prev_instant")
+
.addTableHeaderField("num_writes").addTableHeaderField("num_inserts").addTableHeaderField("num_deletes")
+
.addTableHeaderField("num_update_writes").addTableHeaderField("total_log_files")
+
.addTableHeaderField("total_log_blocks").addTableHeaderField("total_corrupt_log_blocks")
+
.addTableHeaderField("total_rollback_blocks").addTableHeaderField("total_log_records")
+
.addTableHeaderField("total_updated_records_compacted").addTableHeaderField("total_write_bytes")
+ .addTableHeaderField("total_write_errors");
+
+ // The pre table version 8 archive keeps one entry per instant state. The
requested entry
+ // of a commit carries no metadata and the inflight one only the
in-progress stats, so the
+ // legacy stats reader renders the completed entry of each archived
instant, with its stat
+ // rows ordered by partition path like the archived timeline path
+ final List<Comparable[]> statsRows = new ArrayList<>();
+ for (String instant : Arrays.asList("100", "101")) {
+ statsRows.addAll(writeStatRows("commit", instant));
+ }
+ String expectedStats = removeNonWordAndStripSpace(
+ HoodiePrintHelper.print(statsHeader, new HashMap<>(), "", false, 0,
false, statsRows));
+
+ Object result = shell.evaluate(() -> "show archived commit stats");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+ assertEquals(expectedStats, removeNonWordAndStripSpace(result.toString()));
+
+ // the explicit folder pattern reads the very same archive files
+ result = shell.evaluate(() -> "show archived commit stats
--archiveFolderPattern archived/.commits_.archive*");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+ assertEquals(expectedStats, removeNonWordAndStripSpace(result.toString()));
+
+ // show archived commits reads the same archive through
ArchivedTimelineV1, which keeps only
+ // the completed entry of each instant and caches its metadata as the
archived JSON rendering
+ TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType")
+ .addTableHeaderField("CommitDetails");
+ final List<Comparable[]> rows = new ArrayList<>();
+ rows.add(new Comparable[] {"100", "commit",
+
HoodieTestCommitUtilities.convertAndOrderCommitMetadata(commitMetadata.get(0))});
+ rows.add(new Comparable[] {"101", "commit",
+
HoodieTestCommitUtilities.convertAndOrderCommitMetadata(commitMetadata.get(1))});
+ String expected = removeNonWordAndStripSpace(
+ HoodiePrintHelper.print(header, new HashMap<>(), "", false, 0, false,
rows));
+
+ result = shell.evaluate(() -> "show archived commits --skipMetadata
false");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+ assertEquals(expected, removeNonWordAndStripSpace(result.toString()));
+ }
+
+ /**
+ * The metadata of one of the plain commits of the fixture, rendered the way
the command
+ * renders it, with the partition keyed map ordered by partition path.
+ */
+ private Comparable commitDetails(String instantTime) throws Exception {
+ return HoodieTestCommitUtilities.convertAndOrderCommitMetadata(
+ HoodieTestCommitMetadataGenerator.generateCommitMetadata(tablePath,
instantTime));
+ }
+
+ /**
+ * The archived replace commit rendered the way the command renders it, with
both partition
+ * keyed maps ordered by partition path.
+ */
+ private org.apache.hudi.avro.model.HoodieReplaceCommitMetadata
orderedAvroReplaceCommitMetadata() {
+ org.apache.hudi.avro.model.HoodieReplaceCommitMetadata avroMetadata =
+
MetadataConversionUtils.convertCommitMetadataToAvro(replaceCommitMetadata);
+ avroMetadata.setPartitionToWriteStats(new
TreeMap<>(avroMetadata.getPartitionToWriteStats()));
+ avroMetadata.setPartitionToReplaceFileIds(new
TreeMap<>(avroMetadata.getPartitionToReplaceFileIds()));
+ return avroMetadata;
+ }
+
+ /**
+ * The stat rows of a commit written by {@link
HoodieTestCommitMetadataGenerator}, ordered by
+ * partition path.
+ */
+ private static List<Comparable[]> writeStatRows(String action, String
instant) {
+ return Arrays.asList(
+ writeStatRow(action, instant, DEFAULT_SECOND_PARTITION_PATH),
+ writeStatRow(action, instant, DEFAULT_FIRST_PARTITION_PATH));
+ }
+
+ private static Comparable[] writeStatRow(String action, String instant,
String partitionPath) {
+ return new Comparable[] {action, instant, partitionPath,
+ HoodieTestCommitMetadataGenerator.DEFAULT_FILEID,
+ HoodieTestCommitMetadataGenerator.DEFAULT_PRE_COMMIT,
+ HoodieTestCommitMetadataGenerator.DEFAULT_NUM_WRITES,
+ HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
+ HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
+ HoodieTestCommitMetadataGenerator.DEFAULT_NUM_UPDATE_WRITES,
+ HoodieTestCommitMetadataGenerator.DEFAULT_NULL_VALUE,
+ HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_LOG_BLOCKS,
+ HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
+ HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
+ HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_LOG_RECORDS,
+ HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE,
+ HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_WRITE_BYTES,
+ HoodieTestCommitMetadataGenerator.DEFAULT_OTHER_VALUE};
+ }
+
+ /**
+ * Writes a completed clean whose plan retains from {@code
earliestCommitToRetain}.
+ */
+ private static HoodieCleanMetadata addClean(HoodieTableMetaClient
metaClient, String instantTime,
+ String earliestCommitToRetain)
throws IOException {
+ HoodieCleanerPlan cleanerPlan = new HoodieCleanerPlan(
+ new HoodieActionInstant(earliestCommitToRetain,
HoodieTimeline.COMMIT_ACTION, HoodieInstant.State.COMPLETED.name()),
+ "", HoodieCleaningPolicy.KEEP_LATEST_COMMITS.name(), new HashMap<>(),
+ CleanPlanV2MigrationHandler.VERSION, new HashMap<>(), new
ArrayList<>(), Collections.emptyMap());
+ HoodieCleanStat cleanStat = HoodieCleanStat.builder()
+ .withPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS)
+ .withPartitionPath(DEFAULT_FIRST_PARTITION_PATH)
+ .withEarliestCommitToRetain(earliestCommitToRetain)
+ .withLastCompletedCommitTimestamp("")
+ .build();
+ HoodieCleanMetadata metadata = convertCleanMetadata(
+ instantTime, Option.of(0L), Collections.singletonList(cleanStat),
Collections.emptyMap());
+ HoodieTestTable.of(metaClient).addClean(instantTime, cleanerPlan,
metadata);
+ return metadata;
+ }
+
+ /**
+ * Writes a completed insert overwrite replace commit over the two default
partitions.
+ */
+ private static HoodieReplaceCommitMetadata
addReplaceCommit(HoodieTableMetaClient metaClient,
+ String
instantTime) throws Exception {
+ HoodieReplaceCommitMetadata metadata = new HoodieReplaceCommitMetadata();
+ metadata.setOperationType(WriteOperationType.INSERT_OVERWRITE);
+ for (String partitionPath : Arrays.asList(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH)) {
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setPartitionPath(partitionPath);
+ writeStat.setPath(HoodieTestCommitMetadataGenerator.DEFAULT_PATH);
+ writeStat.setFileId(HoodieTestCommitMetadataGenerator.DEFAULT_FILEID);
+
writeStat.setTotalWriteBytes(HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_WRITE_BYTES);
+
writeStat.setPrevCommit(HoodieTestCommitMetadataGenerator.DEFAULT_PRE_COMMIT);
+
writeStat.setNumWrites(HoodieTestCommitMetadataGenerator.DEFAULT_NUM_WRITES);
+
writeStat.setNumUpdateWrites(HoodieTestCommitMetadataGenerator.DEFAULT_NUM_UPDATE_WRITES);
+
writeStat.setTotalLogBlocks(HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_LOG_BLOCKS);
+
writeStat.setTotalLogRecords(HoodieTestCommitMetadataGenerator.DEFAULT_TOTAL_LOG_RECORDS);
+ metadata.addWriteStat(partitionPath, writeStat);
+ metadata.addReplaceFileId(partitionPath,
HoodieTestCommitMetadataGenerator.DEFAULT_FILEID);
+ }
+ HoodieRequestedReplaceMetadata requestedReplaceMetadata =
HoodieRequestedReplaceMetadata.newBuilder()
+ .setOperationType(WriteOperationType.INSERT_OVERWRITE.name())
+ .setVersion(1)
+ .setExtraMetadata(Collections.emptyMap())
+ .build();
+ HoodieTestTable.of(metaClient)
+ .addReplaceCommit(instantTime, Option.of(requestedReplaceMetadata),
Option.empty(), metadata);
+ return metadata;
+ }
+
+ /**
+ * Writes the requested, the inflight and the completed file of one commit
through the meta
+ * client of the table, so that the files follow the layout of the table
version and look like
+ * what a writer leaves behind: an empty requested file, and the commit
metadata in the inflight
+ * and the completed file.
+ */
+ private static HoodieCommitMetadata createLegacyCommit(HoodieTableMetaClient
metaClient, String basePath,
+ String instantTime)
throws Exception {
+ HoodieCommitMetadata metadata =
HoodieTestCommitMetadataGenerator.generateCommitMetadata(basePath, instantTime);
+ // the archiver drops the commit metadata of an entry whose operation type
is UNKNOWN
+ metadata.setOperationType(WriteOperationType.INSERT);
+ metaClient.getStorage().create(new
StoragePath(metaClient.getTimelinePath(),
+
metaClient.getInstantFileNameGenerator().makeRequestedCommitFileName(instantTime)),
true).close();
+ List<String> fileNames = Arrays.asList(
+
metaClient.getInstantFileNameGenerator().makeInflightCommitFileName(instantTime),
+
metaClient.getInstantFileNameGenerator().makeCommitFileName(instantTime));
+ for (String fileName : fileNames) {
+ StoragePath path = new StoragePath(metaClient.getTimelinePath(),
fileName);
+ try (OutputStream os = metaClient.getStorage().create(path, true)) {
+
metaClient.getCommitMetadataSerDe().getInstantWriter(metadata).get().writeToStream(os);
+ }
+ }
+ return metadata;
}
}
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 c1972c6e488a..18f364ca90d0 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
@@ -30,10 +30,10 @@ import org.apache.hudi.common.model.HoodieCleaningPolicy;
import org.apache.hudi.common.model.HoodieCommitMetadata;
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.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestTable;
import org.apache.hudi.common.util.HoodieStorageUtils;
@@ -86,7 +86,7 @@ public class TestCleansCommand extends
CLIFunctionalTestHarness {
// Create table and connect
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.COPY_ON_WRITE.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
StorageConfiguration<?> conf = HoodieCLI.conf;
@@ -177,12 +177,10 @@ public class TestCleansCommand extends
CLIFunctionalTestHarness {
.addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_FILES_SUCCESSFULLY_DELETED)
.addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_FAILED_DELETIONS);
- // There should be two partition path
+ // There should be two partition paths, rendered in partition order
List<Comparable[]> rows = new ArrayList<>();
rows.add(new Comparable[]
{HoodieTestCommitMetadataGenerator.DEFAULT_SECOND_PARTITION_PATH,
HoodieCleaningPolicy.KEEP_LATEST_COMMITS, "1", "0"});
- rows.add(new Comparable[]
{HoodieTestCommitMetadataGenerator.DEFAULT_THIRD_PARTITION_PATH,
- HoodieCleaningPolicy.KEEP_LATEST_COMMITS, "0", "0"});
rows.add(new Comparable[]
{HoodieTestCommitMetadataGenerator.DEFAULT_FIRST_PARTITION_PATH,
HoodieCleaningPolicy.KEEP_LATEST_COMMITS, "1", "0"});
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 54f2de8bc835..2166448ffd10 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
@@ -32,12 +32,11 @@ import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.fs.FSUtils;
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.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.InstantFileNameGenerator;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
-import
org.apache.hudi.common.table.timeline.versioning.v1.InstantFileNameGeneratorV1;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestUtils;
@@ -105,7 +104,7 @@ public class TestCommitsCommand extends
CLIFunctionalTestHarness {
// Create table and connect
new TableCommand().createTable(
tablePath1, tableName1, HoodieTableType.COPY_ON_WRITE.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
metaClient = HoodieCLI.getTableMetaClient();
}
@@ -590,7 +589,7 @@ public class TestCommitsCommand extends
CLIFunctionalTestHarness {
metaClient = HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
// Create inflight commits
- InstantFileNameGenerator v1InstantNameGenerator = new
InstantFileNameGeneratorV1();
+ InstantFileNameGenerator v1InstantNameGenerator =
metaClient.getInstantFileNameGenerator();
List<String> fileNames = Arrays.asList(
v1InstantNameGenerator.makeInflightCommitFileName(oldInstantTime1),
v1InstantNameGenerator.makeCommitFileName(
@@ -598,7 +597,7 @@ public class TestCommitsCommand extends
CLIFunctionalTestHarness {
v1InstantNameGenerator.makeInflightCommitFileName(oldInstantTime3));
fileNames.forEach(name -> {
try {
- Path filePath = new Path(basePath() + "/" +
HoodieTableMetaClient.METAFOLDER_NAME + "/" + name);
+ Path filePath = new Path(metaClient.getTimelinePath() + "/" + name);
HoodieTestDataGenerator.createEmptyFile(basePath(), filePath,
storageConf());
} catch (IOException ignored) {
// Exception ignored.
@@ -618,11 +617,13 @@ public class TestCommitsCommand extends
CLIFunctionalTestHarness {
assertTrue(output.contains(oldInstantTime3));
// Only one instants should be shown when duration is 15 since 2nd commit
is a completed commit.
+ // The 3rd instant started 5 mins ago, so it is not older than the 15 min
lookback window either.
Object lookupBackIn15MinsResult = shell.evaluate(() -> "commits
show_inflights --lookbackInMins 15");
assertTrue(ShellEvaluationResultUtil.isSuccess(lookupBackIn15MinsResult));
output = lookupBackIn15MinsResult.toString();
assertTrue(output.contains(oldInstantTime1));
assertFalse(output.contains(oldInstantTime2));
+ assertFalse(output.contains(oldInstantTime3));
// Only one instant should be shown when duration is 50
Object lookupBackIn50MinsResult = shell.evaluate(() -> "commits
show_inflights --lookbackInMins 50");
@@ -630,12 +631,12 @@ public class TestCommitsCommand extends
CLIFunctionalTestHarness {
output = lookupBackIn50MinsResult.toString();
assertTrue(output.contains(oldInstantTime1));
assertFalse(output.contains(oldInstantTime2));
+ assertFalse(output.contains(oldInstantTime3));
// No instants should be shown when duration is > 60
Object lookupBackIn70MinsResult = shell.evaluate(() -> "commits
show_inflights --lookbackInMins 70");
assertTrue(ShellEvaluationResultUtil.isSuccess(lookupBackIn70MinsResult));
output = lookupBackIn70MinsResult.toString();
- assertTrue(output.contains(oldInstantTime1));
- assertFalse(output.contains(oldInstantTime2));
+ assertEquals("No inflight instants are found.", output);
}
}
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 001b5181e022..d25abe1789ad 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
@@ -30,9 +30,9 @@ 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.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.testutils.CompactionTestUtils;
import org.apache.hudi.common.testutils.HoodieTestUtils;
@@ -89,7 +89,7 @@ public class TestCompactionCommand extends
CLIFunctionalTestHarness {
// create COW table.
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.COPY_ON_WRITE.name(),
- "", TimelineLayoutVersion.VERSION_1,
HoodieAvroPayload.class.getName());
+ "", HoodieTableVersion.current().versionCode(),
HoodieAvroPayload.class.getName());
// expect HoodieException for COPY_ON_WRITE table.
assertThrows(HoodieException.class,
@@ -104,7 +104,7 @@ public class TestCompactionCommand extends
CLIFunctionalTestHarness {
// create MOR table.
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.MERGE_ON_READ.name(),
- "", TimelineLayoutVersion.VERSION_1,
HoodieAvroPayload.class.getName());
+ "", HoodieTableVersion.current().versionCode(),
HoodieAvroPayload.class.getName());
CompactionTestUtils.setupAndValidateCompactionOperations(HoodieCLI.getTableMetaClient(),
false, 3, 4, 3, 3);
@@ -136,7 +136,7 @@ public class TestCompactionCommand extends
CLIFunctionalTestHarness {
// create MOR table.
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.MERGE_ON_READ.name(),
- "", TimelineLayoutVersion.VERSION_1,
HoodieAvroPayload.class.getName());
+ "", HoodieTableVersion.current().versionCode(),
HoodieAvroPayload.class.getName());
CompactionTestUtils.setupAndValidateCompactionOperations(HoodieCLI.getTableMetaClient(),
false, 3, 4, 3, 3);
@@ -150,7 +150,7 @@ public class TestCompactionCommand extends
CLIFunctionalTestHarness {
// create MOR table.
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.MERGE_ON_READ.name(),
- "", TimelineLayoutVersion.VERSION_1,
HoodieAvroPayload.class.getName());
+ "", HoodieTableVersion.current().versionCode(),
HoodieAvroPayload.class.getName());
CompactionTestUtils.setupAndValidateCompactionOperations(HoodieCLI.getTableMetaClient(),
true, 1, 2, 3, 4);
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestDiffCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestDiffCommand.java
index d4b240b67c57..c80a8f8d954b 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestDiffCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestDiffCommand.java
@@ -30,9 +30,9 @@ 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.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestTable;
import org.apache.hudi.common.util.HoodieStorageUtils;
@@ -86,7 +86,7 @@ public class TestDiffCommand extends CLIFunctionalTestHarness
{
// create COW table.
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.COPY_ON_WRITE.name(),
- "", TimelineLayoutVersion.VERSION_1,
HoodieAvroPayload.class.getName());
+ "", HoodieTableVersion.current().versionCode(),
HoodieAvroPayload.class.getName());
StorageConfiguration<?> conf = HoodieCLI.conf;
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestFileSystemViewCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestFileSystemViewCommand.java
index 1eaa08959ac3..a495d37fffaa 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestFileSystemViewCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestFileSystemViewCommand.java
@@ -30,8 +30,10 @@ import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieFileGroup;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.table.view.SyncableFileSystemView;
+import org.apache.hudi.common.testutils.FileCreateUtils;
import org.apache.hudi.common.util.NumericUtils;
import org.junit.jupiter.api.BeforeEach;
@@ -86,7 +88,7 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
nonpartitionedTablePath = tablePath(nonpartitionedTableName);
new TableCommand().createTable(
nonpartitionedTablePath, nonpartitionedTableName,
- "COPY_ON_WRITE", "", 1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "COPY_ON_WRITE", "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
@@ -110,8 +112,8 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
.makeInlineLogFileName(fileId1, HoodieLogFile.DELTA_EXTENSION,
commitTime2, 0, testWriteToken)));
// Write commit files
- Files.createFile(Paths.get(nonpartitionedTablePath, ".hoodie", commitTime1
+ ".commit"));
- Files.createFile(Paths.get(nonpartitionedTablePath, ".hoodie", commitTime2
+ ".commit"));
+ FileCreateUtils.createCommit(metaClient, commitTime1);
+ FileCreateUtils.createCommit(metaClient, commitTime2);
// Reload meta client and create fsView
metaClient = HoodieTableMetaClient.reload(metaClient);
@@ -127,7 +129,7 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
partitionedTablePath = tablePath(partitionedTableName);
new TableCommand().createTable(
partitionedTablePath, partitionedTableName,
- "COPY_ON_WRITE", "", 1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "COPY_ON_WRITE", "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
@@ -153,8 +155,8 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
.makeInlineLogFileName(fileId1, HoodieLogFile.DELTA_EXTENSION,
commitTime2, 0, testWriteToken)));
// Write commit files
- Files.createFile(Paths.get(partitionedTablePath, ".hoodie", commitTime1 +
".commit"));
- Files.createFile(Paths.get(partitionedTablePath, ".hoodie", commitTime2 +
".commit"));
+ FileCreateUtils.createCommit(metaClient, commitTime1);
+ FileCreateUtils.createCommit(metaClient, commitTime2);
// Reload meta client and create fsView
metaClient = HoodieTableMetaClient.reload(metaClient);
@@ -294,12 +296,7 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
return rows;
}
- /**
- * (
- * Test case for command 'show fsview latest'.
- */
- @Test
- public void testShowLatestFileSlices() throws IOException {
+ private static Map<String, Function<Object, String>>
latestFileSlicesConverterMap() {
Function<Object, String> converterFunction =
entry ->
NumericUtils.humanReadableByteCount((Double.parseDouble(entry.toString())));
Map<String, Function<Object, String>> fieldNameToConverterMap = new
HashMap<>();
@@ -307,8 +304,11 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
fieldNameToConverterMap.put(HoodieTableHeaderFields.HEADER_TOTAL_DELTA_SIZE,
converterFunction);
fieldNameToConverterMap.put(HoodieTableHeaderFields.HEADER_DELTA_SIZE_SCHEDULED,
converterFunction);
fieldNameToConverterMap.put(HoodieTableHeaderFields.HEADER_DELTA_SIZE_UNSCHEDULED,
converterFunction);
+ return fieldNameToConverterMap;
+ }
- TableHeader header = new
TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_PARTITION)
+ private static TableHeader latestFileSlicesHeader() {
+ return new
TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_PARTITION)
.addTableHeaderField(HoodieTableHeaderFields.HEADER_FILE_ID)
.addTableHeaderField(HoodieTableHeaderFields.HEADER_BASE_INSTANT)
.addTableHeaderField(HoodieTableHeaderFields.HEADER_DATA_FILE)
@@ -321,6 +321,16 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
.addTableHeaderField(HoodieTableHeaderFields.HEADER_DELTA_BASE_UNSCHEDULED)
.addTableHeaderField(HoodieTableHeaderFields.HEADER_DELTA_FILES_SCHEDULED)
.addTableHeaderField(HoodieTableHeaderFields.HEADER_DELTA_FILES_UNSCHEDULED);
+ }
+
+ /**
+ * (
+ * Test case for command 'show fsview latest'.
+ */
+ @Test
+ public void testShowLatestFileSlices() throws IOException {
+ Map<String, Function<Object, String>> fieldNameToConverterMap =
latestFileSlicesConverterMap();
+ TableHeader header = latestFileSlicesHeader();
// Test show with partition path '2016/03/15'
new TableCommand().connect(partitionedTablePath, false, 0, 0, 0,
@@ -351,4 +361,30 @@ public class TestFileSystemViewCommand extends
CLIFunctionalTestHarness {
String nonpartitionedResults =
removeNonWordAndStripSpace(nonpartitionedTable.toString());
assertEquals(nonpartitionedExpected, nonpartitionedResults);
}
+
+ /**
+ * Test case for command 'show fsview latest' on a table whose timeline
holds no completed
+ * instant. The merge branch has nothing to pick a max instant from, so it
has to fall back to
+ * the unmerged latest file slices instead of dereferencing an empty Option.
+ */
+ @Test
+ public void testShowLatestFileSlicesWithEmptyTimeline() throws IOException {
+ HoodieCLI.conf = storageConf();
+
+ // Create table and connect. No commit is ever written to it.
+ String emptyTableName = "empty_" + tableName();
+ String emptyTablePath = tablePath(emptyTableName);
+ new TableCommand().createTable(
+ emptyTablePath, emptyTableName,
+ "COPY_ON_WRITE", "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
+
+ // --merge defaults to true, so this takes the branch that resolves a max
instant
+ Object result = shell.evaluate(() -> "show fsview latest");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+
+ // header only, no file slice rows
+ String expected = HoodiePrintHelper.print(latestFileSlicesHeader(),
latestFileSlicesConverterMap(),
+ "", false, -1, false, new ArrayList<>());
+ assertEquals(removeNonWordAndStripSpace(expected),
removeNonWordAndStripSpace(result.toString()));
+ }
}
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestHoodieLogFileCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestHoodieLogFileCommand.java
index cb0c4c742b78..ac7ca0998d1a 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestHoodieLogFileCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestHoodieLogFileCommand.java
@@ -35,13 +35,13 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.SchemaTestUtil;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
@@ -105,7 +105,7 @@ public class TestHoodieLogFileCommand extends
CLIFunctionalTestHarness {
partitionPath = Paths.get(tablePath,
DEFAULT_FIRST_PARTITION_PATH).toString();
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.MERGE_ON_READ.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
Files.createDirectories(Paths.get(partitionPath));
storage = HoodieStorageUtils.getStorage(tablePath, storageConf());
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 ea8c4b8817ed..b49325159d62 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
@@ -18,6 +18,9 @@
package org.apache.hudi.cli.commands;
+import org.apache.hudi.avro.model.HoodieActionInstant;
+import org.apache.hudi.avro.model.HoodieCleanFileInfo;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
import org.apache.hudi.cli.HoodieCLI;
import org.apache.hudi.cli.HoodiePrintHelper;
import org.apache.hudi.cli.HoodieTableHeaderFields;
@@ -29,22 +32,33 @@ import org.apache.hudi.client.WriteClientTestUtils;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
+import org.apache.hudi.common.model.HoodieCleaningPolicy;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
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.timeline.versioning.TimelineLayoutVersion;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import
org.apache.hudi.common.table.timeline.versioning.clean.CleanPlanV2MigrationHandler;
+import org.apache.hudi.common.testutils.FileCreateUtils;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.PartitionPathEncodeUtils;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.io.util.FileIOUtils;
import org.apache.hudi.keygen.SimpleKeyGenerator;
import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.testutils.Assertions;
+import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.logging.log4j.Level;
@@ -63,31 +77,33 @@ import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.shell.Shell;
import java.io.FileInputStream;
+import java.io.FilterInputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.SocketTimeoutException;
import java.net.URL;
-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.List;
import java.util.Map;
import java.util.Properties;
+import java.util.TreeSet;
import java.util.UUID;
import java.util.stream.Collectors;
import static
org.apache.hudi.common.table.HoodieTableConfig.DROP_PARTITION_COLUMNS;
-import static org.apache.hudi.common.table.HoodieTableConfig.NAME;
import static org.apache.hudi.common.table.HoodieTableConfig.TABLE_CHECKSUM;
-import static
org.apache.hudi.common.table.HoodieTableConfig.TIMELINE_HISTORY_PATH;
-import static
org.apache.hudi.common.table.HoodieTableConfig.TIMELINE_LAYOUT_VERSION;
-import static org.apache.hudi.common.table.HoodieTableConfig.TYPE;
-import static org.apache.hudi.common.table.HoodieTableConfig.VERSION;
import static org.apache.hudi.common.table.HoodieTableConfig.generateChecksum;
import static org.apache.hudi.common.table.HoodieTableConfig.validateChecksum;
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.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -112,7 +128,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
// Create table and connect
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.COPY_ON_WRITE.name(),
- HoodieTableConfig.TIMELINE_HISTORY_PATH.defaultValue(),
TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ HoodieTableConfig.TIMELINE_HISTORY_PATH.defaultValue(),
HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
}
@AfterEach
@@ -126,7 +142,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
@Test
public void testAddPartitionMetaWithDryRun() throws IOException {
// create commit instant
- Files.createFile(Paths.get(tablePath, ".hoodie/timeline/", "100.commit"));
+ FileCreateUtils.createCommit(HoodieCLI.getTableMetaClient(), "100");
// create partition path
String partition1 = Paths.get(tablePath,
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH).toString();
@@ -161,7 +177,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
@Test
public void testAddPartitionMetaWithRealRun() throws IOException {
// create commit instant
- Files.createFile(Paths.get(tablePath, ".hoodie", "100.commit"));
+ FileCreateUtils.createCommit(HoodieCLI.getTableMetaClient(), "100");
// create partition path
String partition1 = Paths.get(tablePath,
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH).toString();
@@ -205,7 +221,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
* Test case for 'repair overwrite-hoodie-props'.
*/
@Test
- public void testOverwriteHoodieProperties() throws IOException {
+ public void testOverwriteHoodieProperties() throws Exception {
URL newProps =
this.getClass().getClassLoader().getResource("table-config.properties");
assertNotNull(newProps, "New property file must exist");
@@ -227,19 +243,26 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
.collect(Collectors.toMap(e -> String.valueOf(e.getKey()), e ->
String.valueOf(e.getValue())));
expected.putIfAbsent(TABLE_CHECKSUM.key(),
String.valueOf(generateChecksum(tableConfig.getProps())));
expected.putIfAbsent(DROP_PARTITION_COLUMNS.key(),
String.valueOf(DROP_PARTITION_COLUMNS.defaultValue()));
+
+ // Properties that Hudi 1.x fills in on its own: the new-props file sets
none of the three, so
+ // HoodieTableConfig.create writes its own defaults for the two paths and
the command carries
+ // the initial table version over from the old properties. All three are
fixed values here,
+ // never read back out of what the command wrote.
+ expected.putIfAbsent(HoodieTableConfig.TIMELINE_PATH.key(),
HoodieTableConfig.TIMELINE_PATH.defaultValue());
+ expected.putIfAbsent(HoodieTableConfig.TIMELINE_HISTORY_PATH.key(),
HoodieTableConfig.TIMELINE_HISTORY_PATH.defaultValue());
+ expected.putIfAbsent(HoodieTableConfig.INITIAL_VERSION.key(),
String.valueOf(HoodieTableVersion.current().versionCode()));
+
assertEquals(expected, result);
- // check result
- List<String> allPropsStr = Arrays.asList(NAME.key(), TYPE.key(),
VERSION.key(),
- TIMELINE_HISTORY_PATH.key(), TIMELINE_LAYOUT_VERSION.key(),
TABLE_CHECKSUM.key(), DROP_PARTITION_COLUMNS.key());
- String[][] rows = allPropsStr.stream().sorted().map(key -> new String[]
{key,
- oldProps.getOrDefault(key, "null"), result.getOrDefault(key,
"null")})
+ // the rendered table lists one row per property, with its old and new
value
+ TreeSet<String> allPropKeys = new TreeSet<>(oldProps.keySet());
+ allPropKeys.addAll(result.keySet());
+ String[][] rows = allPropKeys.stream()
+ .map(key -> new String[] {key, oldProps.getOrDefault(key, "null"),
result.getOrDefault(key, "null")})
.toArray(String[][]::new);
String expect = HoodiePrintHelper.print(new String[]
{HoodieTableHeaderFields.HEADER_HOODIE_PROPERTY,
HoodieTableHeaderFields.HEADER_OLD_VALUE,
HoodieTableHeaderFields.HEADER_NEW_VALUE}, rows);
- expect = removeNonWordAndStripSpace(expect);
- String got = removeNonWordAndStripSpace(cmdResult.toString());
- assertEquals(expect, got);
+ assertEquals(removeNonWordAndStripSpace(expect),
removeNonWordAndStripSpace(cmdResult.toString()));
}
/**
@@ -256,21 +279,186 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
// Create four requested files
for (int i = 100; i < 104; i++) {
String timestamp = String.valueOf(i);
- // Write corrupted requested Clean File
+ // Write an empty requested Clean File
HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
timestamp, conf);
}
+ // A plan cut short mid-write, so that its bytes stop inside the Avro
header
+ FileCreateUtils.createRequestedCleanFile(metaClient, "104",
validCleanerPlan());
+ truncateInHalf(metaClient, requestedCleanPath(metaClient, "104"));
+
+ // A plan whose writer was killed between opening and closing the Avro
container, which leaves a
+ // complete header and no record behind
+ try (DataFileWriter<HoodieCleanerPlan> writer =
+ new DataFileWriter<>(new
SpecificDatumWriter<>(HoodieCleanerPlan.class))) {
+ writer.create(HoodieCleanerPlan.getClassSchema(),
+ metaClient.getStorage().create(requestedCleanPath(metaClient,
"105"), true));
+ }
+
+ // A plan that decodes, which the command has to leave in place
+ FileCreateUtils.createRequestedCleanFile(metaClient, "106",
validCleanerPlan());
+
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
- // first, there are four instants
- assertEquals(4,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+ // first, there are seven pending instants
+ assertEquals(7,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
- Object result = shell.evaluate(() -> "repair corrupted clean files");
- assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+ Object cleanResult = shell.evaluate(() -> "repair corrupted clean files");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(cleanResult));
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
- assertEquals(0,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+ // the empty, the truncated and the record-less plans are gone and the
readable one is untouched
+ List<HoodieInstant> remaining =
+
metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, remaining.size());
+ assertEquals("106", remaining.get(0).requestedTime());
+ }
+
+ /**
+ * A transient read failure on a valid pending clean plan must not be taken
for corruption:
+ * the timeline serde wraps any exception raised while it streams the plan
file in the same
+ * "unable to read commit metadata" IOException that an empty or truncated
file raises.
+ */
+ @Test
+ public void testRemoveCorruptedPendingCleanActionKeepsPlanOnReadFailure()
throws IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ FileCreateUtils.createRequestedCleanFile(metaClient, "100",
validCleanerPlan());
+ StoragePath planPath = requestedCleanPath(metaClient, "100");
+
+ HoodieTableMetaClient timingOutClient = HoodieTableMetaClient.builder()
+ .setStorage(new TimingOutStorage(fs,
planPath)).setBasePath(tablePath).build();
+ HoodieIOException thrown = assertThrows(HoodieIOException.class,
+ () ->
RepairsCommand.removeCorruptedPendingCleanAction(timingOutClient));
+ assertInstanceOf(SocketTimeoutException.class, thrown.getCause());
+ assertTrue(metaClient.getStorage().exists(planPath));
+
+ // the same plan read through a healthy storage is left alone as well
+
RepairsCommand.removeCorruptedPendingCleanAction(HoodieTableMetaClient.reload(metaClient));
+ assertEquals(1,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .filterInflightsAndRequested().countInstants());
+ }
+
+ /**
+ * A clean that reached inflight keeps its plan in the requested file, so a
corrupt plan has to
+ * take both files with it. The timeline collapses the two states into the
inflight instant
+ * alone, so removing only the instant it listed would leave the corrupt
plan behind for the
+ * next clean to fail on, and take a second run of this command to clear.
+ */
+ @Test
+ public void testRemoveCorruptedPendingCleanActionRemovesInflightAndItsPlan()
throws IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ // a clean that was scheduled and started, whose plan was then cut short
+ FileCreateUtils.createRequestedCleanFile(metaClient, "100",
validCleanerPlan());
+ // the inflight file a clean leaves behind carries no plan of its own
+ FileCreateUtils.createInflightCleanFile(metaClient, "100", null, true);
+ truncateInHalf(metaClient, requestedCleanPath(metaClient, "100"));
+
+ StoragePath inflightPath = new StoragePath(metaClient.getTimelinePath(),
+
metaClient.getInstantFileNameGenerator().makeInflightCleanerFileName("100"));
+ assertTrue(metaClient.getStorage().exists(inflightPath));
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ List<HoodieInstant> pending =
metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, pending.size());
+ assertTrue(pending.get(0).isInflight());
+
+ RepairsCommand.removeCorruptedPendingCleanAction(metaClient);
+
+ // one pass takes the whole action, not just the file the timeline listed
+ assertFalse(metaClient.getStorage().exists(inflightPath));
+ assertFalse(metaClient.getStorage().exists(requestedCleanPath(metaClient,
"100")));
+ assertEquals(0,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .filterInflightsAndRequested().countInstants());
+ }
+
+ /**
+ * Corrupt bytes do not always reach the decoder as an Avro failure: a plan
that decodes with no
+ * version leaves the migrator to unbox a null, and a length that survives
as far as Avro's own
+ * ceiling raises an {@code UnsupportedOperationException}. Such an instant
has to be judged
+ * corrupt like any other, and the instants behind it still repaired, rather
than the failure
+ * escaping and abandoning the rest of the timeline.
+ */
+ @Test
+ public void
testRemoveCorruptedPendingCleanActionRepairsPastAnUndecodablePlan() throws
IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ HoodieCleanerPlan versionless = validCleanerPlan();
+ versionless.setVersion(null);
+ writeCleanerPlan(metaClient, "100", versionless);
+
+ // an ordinary corruption behind it, which is only reached if the first
one does not escape
+ HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
"101", HoodieCLI.conf);
+ // and a readable plan the command has to leave in place
+ FileCreateUtils.createRequestedCleanFile(metaClient, "102",
validCleanerPlan());
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(3,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+
+ RepairsCommand.removeCorruptedPendingCleanAction(metaClient);
+
+ List<HoodieInstant> remaining = HoodieTableMetaClient.reload(metaClient)
+ .getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, remaining.size());
+ assertEquals("102", remaining.get(0).requestedTime());
+ }
+
+ /**
+ * Cuts a file's bytes in half in place. The file is overwritten rather than
deleted and
+ * rewritten, because both the emptiness check and the file name generator
resolve the file
+ * from the instant itself.
+ */
+ private static void truncateInHalf(HoodieTableMetaClient metaClient,
StoragePath path) throws IOException {
+ byte[] bytes;
+ try (InputStream in = metaClient.getStorage().open(path)) {
+ bytes = FileIOUtils.readAsByteArray(in);
+ }
+ try (OutputStream out = metaClient.getStorage().create(path, true)) {
+ out.write(bytes, 0, bytes.length / 2);
+ }
+ }
+
+ /**
+ * Writes a clean plan straight into the requested file of an instant, so
that a plan the
+ * timeline's own writer would not produce still reaches the command.
+ */
+ private static void writeCleanerPlan(HoodieTableMetaClient metaClient,
String instantTime,
+ HoodieCleanerPlan plan) throws
IOException {
+ try (DataFileWriter<HoodieCleanerPlan> writer =
+ new DataFileWriter<>(new
SpecificDatumWriter<>(HoodieCleanerPlan.class))) {
+ writer.create(HoodieCleanerPlan.getClassSchema(),
+ metaClient.getStorage().create(requestedCleanPath(metaClient,
instantTime), true));
+ writer.append(plan);
+ }
+ }
+
+ /**
+ * The path of the requested file of a clean instant.
+ */
+ private static StoragePath requestedCleanPath(HoodieTableMetaClient
metaClient, String instantTime) {
+ return new StoragePath(metaClient.getTimelinePath(),
+
metaClient.getInstantFileNameGenerator().makeRequestedCleanerFileName(instantTime));
+ }
+
+ /**
+ * A clean plan that decodes into the latest plan version.
+ */
+ private static HoodieCleanerPlan validCleanerPlan() {
+ return HoodieCleanerPlan.newBuilder()
+ .setEarliestInstantToRetain(HoodieActionInstant.newBuilder()
+ .setAction(HoodieTimeline.COMMIT_ACTION).setTimestamp("001")
+ .setState(HoodieInstant.State.COMPLETED.name()).build())
+
.setFilesToBeDeletedPerPartition(Collections.singletonMap("partition1",
Collections.singletonList("file1")))
+
.setFilePathsToBeDeletedPerPartition(Collections.singletonMap("partition1",
+
Collections.singletonList(HoodieCleanFileInfo.newBuilder().setFilePath("file1").build())))
+ .setLastCompletedCommitTimestamp("002")
+ .setPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS.name())
+ .setVersion(CleanPlanV2MigrationHandler.VERSION)
+ .build();
}
/**
@@ -278,27 +466,33 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
*
*/
@Test
- public void testShowFailedCommits() {
+ public void testShowFailedCommits() throws Exception {
HoodieCLI.conf = storageConf();
StorageConfiguration<?> conf = HoodieCLI.conf;
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ // Every commit starts with real metadata, so none of them is empty yet.
for (int i = 1; i < 20; i++) {
- String timestamp = String.valueOf(i);
- // Write corrupted requested Clean File
- HoodieTestCommitMetadataGenerator.createCommitFile(tablePath, timestamp,
conf);
+
HoodieTestCommitMetadataGenerator.createCommitFileWithMetadata(tablePath,
String.valueOf(i), conf);
}
- metaClient.getActiveTimeline().getInstantsAsStream().filter(hoodieInstant
-> Integer.parseInt(hoodieInstant.requestedTime()) % 4 ==
0).forEach(hoodieInstant -> {
- metaClient.getActiveTimeline().deleteInstantFileIfExists(hoodieInstant);
- if (hoodieInstant.isCompleted()) {
- metaClient.getActiveTimeline().createCompleteInstant(hoodieInstant);
- } else {
- metaClient.getActiveTimeline().createNewInstant(hoodieInstant);
- }
- });
+ HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient);
+ // Truncate a subset in place. Rewriting via createCompleteInstant would
mint a fresh
+ // completion time and leave the instant name pointing at a file that no
longer exists.
+ List<HoodieInstant> allInstants =
reloaded.getActiveTimeline().getInstantsAsStream().collect(Collectors.toList());
+ List<HoodieInstant> emptied = allInstants.stream()
+ .filter(instant -> Integer.parseInt(instant.requestedTime()) % 4 == 0)
+ .collect(Collectors.toList());
+ for (HoodieInstant instant : emptied) {
+ Path instantPath = new Path(reloaded.getTimelinePath().toString(),
+ reloaded.getInstantFileNameGenerator().getFileName(instant));
+ HoodieTestDataGenerator.createEmptyFile(tablePath, instantPath, conf);
+ }
+ // A proper subset, so the command has to filter rather than report
everything.
+ assertTrue(emptied.size() > 0 && emptied.size() < allInstants.size(),
+ "truncated " + emptied.size() + " of " + allInstants.size() + "
instants; expected a proper subset");
final TestLogAppender appender = new TestLogAppender();
final Logger logger = (Logger) LogManager.getLogger(RepairsCommand.class);
@@ -308,7 +502,8 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
Object result = shell.evaluate(() -> "repair show empty commit
metadata");
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
final List<LogEvent> log = appender.getLog();
- assertEquals(log.size(),4);
+ // only the instants truncated above should be flagged as empty
+ assertEquals(emptied.size(), log.size());
log.forEach(LoggingEvent -> {
assertEquals(LoggingEvent.getLevel(), Level.WARN);
assertTrue(LoggingEvent.getMessage().getFormattedMessage().contains("Empty
Commit: "));
@@ -332,6 +527,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
.setPartitionFields("partition_path")
.setRecordKeyFields("_row_key")
.setKeyGeneratorClassProp(SimpleKeyGenerator.class.getCanonicalName())
+ .setTableVersion(HoodieTableVersion.current().versionCode())
.initTable(HoodieCLI.conf.newInstance(), tablePath);
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
@@ -346,6 +542,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
JavaRDD<HoodieRecord> writeRecords =
context().getJavaSparkContext().parallelize(records, 1);
List<WriteStatus> result = client.upsert(writeRecords,
newCommitTime).collect();
Assertions.assertNoWriteErrors(result);
+ client.commit(newCommitTime, jsc().parallelize(result));
newCommitTime = "002";
// Generate HoodieRecords w/ null values for partition path field.
@@ -363,6 +560,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
JavaRDD<HoodieRecord> writeRecords2 =
context().getJavaSparkContext().parallelize(records2, 1);
List<WriteStatus> result2 = client.bulkInsert(writeRecords2,
newCommitTime).collect();
Assertions.assertNoWriteErrors(result2);
+ client.commit(newCommitTime, jsc().parallelize(result2));
SQLContext sqlContext = context().getSqlContext();
long totalRecs =
sqlContext.read().format("hudi").load(tablePath).count();
@@ -394,6 +592,7 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
.setPartitionFields("partition_path")
.setRecordKeyFields("_row_key")
.setKeyGeneratorClassProp(SimpleKeyGenerator.class.getCanonicalName())
+ .setTableVersion(HoodieTableVersion.current().versionCode())
.initTable(HoodieCLI.conf.newInstance(), tablePath);
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
@@ -408,12 +607,15 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
JavaRDD<HoodieRecord> writeRecords =
context().getJavaSparkContext().parallelize(records, 1);
List<WriteStatus> result = client.upsert(writeRecords,
newCommitTime).collect();
Assertions.assertNoWriteErrors(result);
+ client.commit(newCommitTime, jsc().parallelize(result));
SQLContext sqlContext = context().getSqlContext();
long totalRecs =
sqlContext.read().format("hudi").load(tablePath).count();
assertEquals(totalRecs, 20);
long totalRecsInOldPartition =
sqlContext.read().format("hudi").load(tablePath)
.filter(HoodieRecord.PARTITION_PATH_METADATA_FIELD + " == '" +
DEFAULT_FIRST_PARTITION_PATH + "'").count();
+ // otherwise the final assertion below would be satisfied by 0 == 0
+ assertTrue(totalRecsInOldPartition > 0);
// Execute rename partition command
assertEquals(0, SparkMain.renamePartition(jsc(), tablePath,
DEFAULT_FIRST_PARTITION_PATH, "2016/03/18"));
@@ -430,6 +632,51 @@ public class TestRepairsCommand extends
CLIFunctionalTestHarness {
}
}
+ /**
+ * Storage whose read of one file times out part-way, the way a remote store
does under a
+ * transient outage: the open succeeds and the stream fails after the first
bytes.
+ */
+ private static class TimingOutStorage extends HoodieHadoopStorage {
+ private final StoragePath timingOutPath;
+
+ TimingOutStorage(FileSystem fs, StoragePath timingOutPath) {
+ super(fs);
+ this.timingOutPath = timingOutPath;
+ }
+
+ @Override
+ public InputStream open(StoragePath path) throws IOException {
+ InputStream in = super.open(path);
+ if (!path.equals(timingOutPath)) {
+ return in;
+ }
+ return new FilterInputStream(in) {
+ private int remaining = 32;
+
+ @Override
+ public int read() throws IOException {
+ if (remaining <= 0) {
+ throw new SocketTimeoutException("Read timed out");
+ }
+ remaining--;
+ return super.read();
+ }
+
+ @Override
+ public int read(byte[] buffer, int offset, int length) throws
IOException {
+ if (remaining <= 0) {
+ throw new SocketTimeoutException("Read timed out");
+ }
+ int read = super.read(buffer, offset, Math.min(length, remaining));
+ if (read > 0) {
+ remaining -= read;
+ }
+ return read;
+ }
+ };
+ }
+ }
+
class TestLogAppender extends AbstractAppender {
private final List<LogEvent> log = new ArrayList<>();
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java
index 0ff5b3b428b5..982593bbd1d1 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java
@@ -31,17 +31,14 @@ import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.common.config.HoodieMetadataConfig;
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.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
-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.HoodieTableMetadataWriter;
-import org.apache.hudi.metadata.SparkHoodieBackedTableMetadataWriter;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
@@ -52,17 +49,17 @@ import org.springframework.shell.Shell;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.stream.Stream;
+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_PARTITION_PATHS;
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.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@Tag("functional")
@@ -78,7 +75,7 @@ public class TestRestoresCommand extends
CLIFunctionalTestHarness {
String tablePath = tablePath(tableName);
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.MERGE_ON_READ.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
//Create some commits files and base files
Map<String, String> partitionAndFileId = new HashMap<String, String>() {
@@ -91,55 +88,65 @@ public class TestRestoresCommand extends
CLIFunctionalTestHarness {
HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath(tablePath)
.withMetadataConfig(
- // Column Stats Index is disabled, since these tests
construct tables which are
+ // The metadata table is disabled, since these tests
construct tables which are
// not valid (empty commit metadata, etc)
HoodieMetadataConfig.newBuilder()
- .withMetadataIndexColumnStats(false)
+ .enable(false)
.build()
)
.withRollbackUsingMarkers(false)
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
.build();
- try (HoodieTableMetadataWriter metadataWriter =
SparkHoodieBackedTableMetadataWriter.create(metaClient.getStorageConf(),
config, context)) {
- HoodieTestTable hoodieTestTable = HoodieMetadataTestTable.of(metaClient,
metadataWriter, Option.of(context))
- .withPartitionMetaFiles(DEFAULT_PARTITION_PATHS)
- .addCommit("100")
- .withBaseFilesInPartitions(partitionAndFileId).getLeft()
- .addCommit("101");
+ HoodieTestTable hoodieTestTable = HoodieTestTable.of(metaClient)
+ .withPartitionMetaFiles(DEFAULT_PARTITION_PATHS)
+ .addCommit("100", Option.of("100001"), Option.empty())
+ .withBaseFilesInPartitions(partitionAndFileId).getLeft()
+ .addCommit("101", Option.of("101001"), Option.empty());
-
hoodieTestTable.addCommit("102").withBaseFilesInPartitions(partitionAndFileId);
- HoodieSavepointMetadata savepointMetadata2 =
hoodieTestTable.doSavepoint("102");
- hoodieTestTable.addSavepoint("102", savepointMetadata2);
+ hoodieTestTable.addCommit("102", Option.of("102001"),
Option.empty()).withBaseFilesInPartitions(partitionAndFileId);
+ HoodieSavepointMetadata savepointMetadata2 =
hoodieTestTable.doSavepoint("102");
+ hoodieTestTable.addSavepointCommit("102", Option.of("102002"),
savepointMetadata2);
-
hoodieTestTable.addCommit("103").withBaseFilesInPartitions(partitionAndFileId);
+ hoodieTestTable.addCommit("103", Option.of("103001"),
Option.empty()).withBaseFilesInPartitions(partitionAndFileId);
- try (BaseHoodieWriteClient client = new SparkRDDWriteClient(context(),
config)) {
- client.rollback("103");
- client.restoreToSavepoint("102");
+ try (BaseHoodieWriteClient client = new SparkRDDWriteClient(context(),
config)) {
+ client.rollback("103");
+ client.restoreToSavepoint("102");
+ }
-
hoodieTestTable.addCommit("105").withBaseFilesInPartitions(partitionAndFileId);
- HoodieSavepointMetadata savepointMetadata =
hoodieTestTable.doSavepoint("105");
- hoodieTestTable.addSavepoint("105", savepointMetadata);
+ hoodieTestTable.addCommit("105", Option.of("105001"),
Option.empty()).withBaseFilesInPartitions(partitionAndFileId);
+ HoodieSavepointMetadata savepointMetadata =
hoodieTestTable.doSavepoint("105");
+ hoodieTestTable.addSavepointCommit("105", Option.of("105002"),
savepointMetadata);
-
hoodieTestTable.addCommit("106").withBaseFilesInPartitions(partitionAndFileId);
- client.rollback("106");
- client.restoreToSavepoint("105");
- }
+ hoodieTestTable.addCommit("106", Option.of("106001"),
Option.empty()).withBaseFilesInPartitions(partitionAndFileId);
+
+ try (BaseHoodieWriteClient client = new SparkRDDWriteClient(context(),
config)) {
+ client.restoreToSavepoint("105");
}
}
@Test
- public void testShowRestores() {
+ public void testShowRestores() throws IOException {
Object result = shell.evaluate(() -> "show restores");
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
// get restored instants
HoodieActiveTimeline activeTimeline =
HoodieCLI.getTableMetaClient().getActiveTimeline();
- Stream<HoodieInstant> restores =
activeTimeline.getRestoreTimeline().filterCompletedInstants().getInstantsAsStream();
+ List<HoodieInstant> restores =
activeTimeline.getRestoreTimeline().filterCompletedInstants()
+ .getInstantsAsStream().sorted().collect(Collectors.toList());
+ // init() runs exactly two restores: restoreToSavepoint("102") and
restoreToSavepoint("105")
+ assertEquals(2, restores.size());
+ // The restore to savepoint 102 still rolls back commit 103: the preceding
client.rollback("103")
+ // only un-publishes it, leaving 103 behind as an inflight commit for the
restore to clean up
+ assertEquals(Collections.singletonList("103"),
+
activeTimeline.readRestoreMetadata(restores.get(0)).getInstantsToRollback());
+ // The restore to savepoint 105 rolls back commit 106, the only commit
written after that savepoint
+ assertEquals(Collections.singletonList("106"),
+
activeTimeline.readRestoreMetadata(restores.get(1)).getInstantsToRollback());
List<Comparable[]> rows = new ArrayList<>();
- restores.sorted().forEach(instant -> {
+ restores.forEach(instant -> {
try {
HoodieRestoreMetadata metadata =
activeTimeline.readRestoreMetadata(instant);
metadata.getInstantsToRollback().forEach(c -> {
@@ -155,6 +162,9 @@ public class TestRestoresCommand extends
CLIFunctionalTestHarness {
}
});
+ // One row per rolled back instant, one from each of the two restores
+ assertEquals(2, rows.size());
+
TableHeader header = new TableHeader()
.addTableHeaderField(HoodieTableHeaderFields.HEADER_INSTANT)
.addTableHeaderField(HoodieTableHeaderFields.HEADER_RESTORE_INSTANT)
@@ -171,15 +181,20 @@ public class TestRestoresCommand extends
CLIFunctionalTestHarness {
public void testShowRestore() throws IOException {
// get instant
HoodieActiveTimeline activeTimeline =
HoodieCLI.getTableMetaClient().getActiveTimeline();
- Stream<HoodieInstant> restores =
activeTimeline.getRestoreTimeline().filterCompletedInstants().getInstantsAsStream();
- HoodieInstant instant = restores.findFirst().orElse(null);
- assertNotNull(instant, "The instant can not be null.");
+ List<HoodieInstant> restores =
activeTimeline.getRestoreTimeline().filterCompletedInstants()
+ .getInstantsAsStream().sorted().collect(Collectors.toList());
+ // init() runs exactly two restores: restoreToSavepoint("102") and
restoreToSavepoint("105")
+ assertEquals(2, restores.size());
+ HoodieInstant instant = restores.get(0);
Object result = shell.evaluate(() -> "show restore --instant " +
instant.requestedTime());
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
// get metadata of instant
HoodieRestoreMetadata instantMetadata =
activeTimeline.readRestoreMetadata(instant);
+ // The earliest of the two restores done by init() is the one to savepoint
102, and it rolls back
+ // commit 103, which the preceding client.rollback("103") only
un-published rather than removed
+ assertEquals(Collections.singletonList("103"),
instantMetadata.getInstantsToRollback());
// generate expected result
TableHeader header = new TableHeader()
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 baf5db208cea..951213b856ed 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
@@ -30,9 +30,9 @@ import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.common.config.HoodieMetadataConfig;
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.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.HoodieMetadataTestTable;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -80,7 +80,7 @@ public class TestRollbacksCommand extends
CLIFunctionalTestHarness {
String tablePath = tablePath(tableName);
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.MERGE_ON_READ.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
//Create some commits files and base files
Map<String, String> partitionAndFileId = new HashMap<String, String>() {
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 abc9a6141d19..4700b2366319 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
@@ -24,8 +24,8 @@ import org.apache.hudi.cli.HoodieTableHeaderFields;
import org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
import org.apache.hudi.cli.testutils.ShellEvaluationResultUtil;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.junit.jupiter.api.BeforeEach;
@@ -62,7 +62,7 @@ public class TestSavepointsCommand extends
CLIFunctionalTestHarness {
// Create table and connect
new TableCommand().createTable(
tablePath, "test_table", HoodieTableType.COPY_ON_WRITE.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
}
/**
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestStatsCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestStatsCommand.java
index 8558d4dd4c67..58e47805b1ea 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestStatsCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestStatsCommand.java
@@ -26,7 +26,7 @@ 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.HoodieTableType;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestTable;
import org.apache.hudi.common.util.Option;
@@ -73,7 +73,7 @@ public class TestStatsCommand extends
CLIFunctionalTestHarness {
// Create table and connect
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.COPY_ON_WRITE.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
}
/**
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestUpgradeDowngradeCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestUpgradeDowngradeCommand.java
index 0aeb3baa1765..f38406cbcc04 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestUpgradeDowngradeCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestUpgradeDowngradeCommand.java
@@ -27,7 +27,6 @@ import org.apache.hudi.common.model.IOType;
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.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.testutils.FileCreateUtilsLegacy;
import org.apache.hudi.common.testutils.HoodieTestTable;
@@ -70,7 +69,7 @@ public class TestUpgradeDowngradeCommand extends
CLIFunctionalTestHarness {
tablePath = tablePath(tableName);
new TableCommand().createTable(
tablePath, tableName, HoodieTableType.COPY_ON_WRITE.name(),
- "", TimelineLayoutVersion.VERSION_1,
"org.apache.hudi.common.model.HoodieAvroPayload");
+ "", HoodieTableVersion.current().versionCode(),
"org.apache.hudi.common.model.HoodieAvroPayload");
timelineService = HoodieClientTestUtils.initTimelineService(
context, basePath(),
FileSystemViewStorageConfig.REMOTE_PORT_NUM.defaultValue());
metaClient = HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
