voonhous commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3888722359
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -107,9 +115,113 @@ public String showArchivedCommits(
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 {
+ allStats = readCommitStatsFromArchivedTimeline(metaClient);
+ }
+ 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();
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ List<Comparable[]> allCommits;
+ try {
+ if (!skipMetadata) {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ }
+ allCommits = archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .map(instant -> readArchivedCommit(archivedTimeline, instant,
skipMetadata))
+ .collect(Collectors.toList());
+ } finally {
+ if (!skipMetadata) {
+ // free the metadata that was loaded in memory for this command
+ archivedTimeline.getInstants().forEach(
+ instant ->
archivedTimeline.clearInstantDetailsFromMemory(instant.requestedTime()));
+ }
+ }
+
+ 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);
+ }
+
+ /**
+ * Reads the write stats of the archived commit and delta commit instants
through the
+ * archived timeline, which resolves the archive format from the table
version (LSM
+ * timeline for table version 8 and above, log format before that).
+ */
+ private List<Comparable[]>
readCommitStatsFromArchivedTimeline(HoodieTableMetaClient metaClient) {
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ try {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ return archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .filter(instant ->
HoodieTimeline.COMMIT_ACTION.equals(instant.getAction())
+ ||
HoodieTimeline.DELTA_COMMIT_ACTION.equals(instant.getAction()))
+ .flatMap(instant -> readWriteStatRows(archivedTimeline, instant))
+ .collect(Collectors.toList());
+ } finally {
+ // free the metadata that was loaded in memory for this command
+ archivedTimeline.getInstants().forEach(
+ instant ->
archivedTimeline.clearInstantDetailsFromMemory(instant.requestedTime()));
+ }
+ }
+
+ private Stream<Comparable[]> readWriteStatRows(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant) {
+ HoodieCommitMetadata metadata;
+ try {
+ metadata = archivedTimeline.readCommitMetadataToAvro(instant);
Review Comment:
This looks blocking for tables at version <= 7.
`ArchivedTimelineV1.readCommit` caches details as JSON text
(`actionData.toString().getBytes()`), and `readCommitMetadataToAvro` routes
through `CommitMetadataSerDeV1.deserialize`, which hands every non-POJO class
to `deserializeAvroMetadata` and throws `Unable to read commit metadata`. So
`show archived commit stats` fails on the first instant of a v6 table, where
the old glob printed. Could we read the POJO with `readCommitMetadata` and
`MetadataConversionUtils.convertCommitMetadataToAvro`, as
`CommitsCommand.java:77` already does? Do you agree this is blocking?
<details><summary>trace</summary>
`show archived commit stats` -> `readCommitStatsFromArchivedTimeline` ->
`metaClient.getArchivedTimeline()` is `ArchivedTimelineV1` for layout v1 ->
`loadCompletedInstantDetailsInMemory` fills `readCommits` with
`actionData.toString()` bytes (`ArchivedTimelineV1.java:343-347`) ->
`readCommitMetadataToAvro` -> `readInstantContent` ->
`TimelineLayout.fromVersion(LAYOUT_VERSION_1).getCommitMetadataSerDe()` ->
`CommitMetadataSerDeV1.deserialize`: `avro.model.HoodieCommitMetadata` is not a
`common.model.HoodieCommitMetadata`, so `deserializeAvroMetadata`
(DataFileStream, needs the Avro magic) reads `{...}` and fails ->
`IOException("Unable to read commit metadata for instant ...")` -> wrapped as
`HoodieException` at line 202.
The typed `readCleanMetadata` / `readRollbackMetadata` calls at lines 294
and 299 hit the same serde under `--skipMetadata false`. On the V1 layout the
cached bytes are already the JSON string the old command rendered, so returning
`new String(details.get(), UTF_8)` for that layout would keep those arms
working.
</details>
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchivedCommitsCommand.java:
##########
@@ -86,24 +84,15 @@ public void init() throws Exception {
.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
+ // Create six commits with metadata
Review Comment:
Only the `commit` arm of the rewritten `readArchivedMetadataString` runs
here: these six instants are all plain commits, so the clean, rollback and
replacecommit arms and the new `sortPartitions(HoodieReplaceCommitMetadata)`
helper have no test. The replacecommit arm broke twice before, both times with
no test: 51297736ca7f (HUDI-2844, #4091) and c63e04daa617 (HUDI-6011, #8345).
Could `init` add one clean and one replacecommit before archiving, and
`testShowCommits` assert those two rows under `--skipMetadata false`?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java:
##########
@@ -201,7 +201,11 @@ public void removeCorruptedPendingCleanAction() {
TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
instant, client.getInstantFileNameGenerator());
} catch (IOException ioe) {
- if (ioe.getMessage().contains("Not an Avro data file")) {
+ // An empty or truncated instant file does not reach Avro's magic-byte
check;
+ // the timeline reader reports it as "unable to read commit metadata"
instead.
+ if (ioe.getMessage() != null
+ && (ioe.getMessage().contains("Not an Avro data file")
+ || ioe.getMessage().contains("unable to read commit
metadata"))) {
Review Comment:
The V1 serde raises this with a capital U (`CommitMetadataSerDeV1.java:65`,
`"Unable to read commit metadata for instant "`), V2 lowercase
(`CommitMetadataSerDeV2.java:88`), both since d3d83020b77e (HUDI-8992).
`String.contains` is case-sensitive, so on a table-version-6 table the
corrupted clean instant is still rethrown. Could we match case-insensitively
(with `import java.util.Locale;`)?
```suggestion
||
ioe.getMessage().toLowerCase(Locale.ROOT).contains("unable to read commit
metadata"))) {
```
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -107,9 +115,113 @@ public String showArchivedCommits(
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 {
+ allStats = readCommitStatsFromArchivedTimeline(metaClient);
+ }
+ 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();
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
Review Comment:
`HoodieCLI.tableMetadata` is a static cached meta client and
`getArchivedTimeline()` memoises in `archivedTimelineMap`
(`HoodieTableMetaClient.java:619-627`), while `triggerArchival` forks
spark-submit and never refreshes. So `trigger archival` followed by `show
archived commits` in one session shows the pre-archival view; the glob this
replaces re-read storage each time. Same at line 181. Could we use the
non-caching overload `metaClient.getArchivedTimeline(StringUtils.EMPTY_STRING,
false)` in both paths, or call `HoodieCLI.refreshTableMetadata()` at the end of
`triggerArchival`?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -156,94 +268,67 @@ public String showArchivedCommits(
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 HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allStats);
+ return 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)
- throws IOException {
-
- 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);
- }
- }
-
- TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType");
-
+ private Comparable[] readArchivedCommit(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant, boolean skipMetadata) {
+ List<Comparable> commitDetails = new ArrayList<>();
+ commitDetails.add(instant.requestedTime());
+ commitDetails.add(instant.getAction());
if (!skipMetadata) {
- header = header.addTableHeaderField("CommitDetails");
- }
-
- 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());
+ commitDetails.add(readArchivedMetadataString(archivedTimeline, instant));
}
return commitDetails.toArray(new Comparable[commitDetails.size()]);
}
- 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 String readArchivedMetadataString(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant) {
+ 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 "{}";
+ }
+ try {
+ 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.SAVEPOINT_ACTION:
+ return archivedTimeline.readSavepointMetadata(instant).toString();
+ case HoodieTimeline.COMPACTION_ACTION:
Review Comment:
nit, feel free to ignore: after `.filter(HoodieInstant::isCompleted)` these
two arms (and `SAVEPOINT` / `CLUSTERING`) are unreachable. V2 archives a
completed compaction as `commit` and clustering as `replacecommit`
(`ArchivedTimelineV2.readCommit` hardcodes COMPLETED; `TimelineArchiverV2`
archives only clean, rollback and the write timeline), and if the compaction
arm were hit, `readCompactionPlan` would decode the bytes loaded by
`LoadMode.METADATA` as a plan. Could we drop the four arms, or comment that
they serve pre-v8 archives only?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java:
##########
@@ -635,7 +634,7 @@ public void testInflightCommand() throws Exception {
Object lookupBackIn70MinsResult = shell.evaluate(() -> "commits
show_inflights --lookbackInMins 70");
assertTrue(ShellEvaluationResultUtil.isSuccess(lookupBackIn70MinsResult));
output = lookupBackIn70MinsResult.toString();
- assertTrue(output.contains(oldInstantTime1));
+ assertFalse(output.contains(oldInstantTime1));
Review Comment:
The flip to `assertFalse` is right (`CommitsCommand` uses
`findInstantsBefore(now - lookback)`, so a 60-min-old inflight is hidden at
70), but the block now only asserts two negatives and passes even if the
fixture never landed. The command returns the literal `No inflight instants are
found.` here. Could we assert that string, and add
`assertFalse(output.contains(oldInstantTime3))` to the 15 and 50 cases?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java:
##########
@@ -623,7 +623,7 @@ private static void withMetaFieldsModeOf(JavaSparkContext
jsc, String basePath,
}
}
- private static int archive(JavaSparkContext jsc, int minCommits, int
maxCommits, int commitsRetained, boolean enableMetadata, String basePath) {
+ protected static int archive(JavaSparkContext jsc, int minCommits, int
maxCommits, int commitsRetained, boolean enableMetadata, String basePath) {
Review Comment:
nit, feel free to ignore: the widening matches `deleteMarker` and
`upgradeOrDowngradeTable`, but the only caller is the test, so
`@VisibleForTesting` (`org.apache.hudi.common.util.VisibleForTesting`) would
say why. Also `testArchiving` now repeats the scenario in
`TestArchiveCommitsProcedure.scala:53-70` (same 6 commits, 2/3/1, same
`ArchiveExecutorUtils.archive`), so its unique coverage is this wrapper; could
the test comment say so?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java:
##########
@@ -408,6 +425,7 @@ public void testRenamePartition() throws IOException {
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));
Review Comment:
nit, feel free to ignore: `assertEquals(totalRecs, totalRecsInOldPartition)`
at line 447 is satisfied by 0 == 0, so if `generateInserts` placed nothing in
`2016/03/15` the rename would pass without renaming anything. Could we add
`assertTrue(totalRecsInOldPartition > 0)` after line 433?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -107,9 +115,113 @@ public String showArchivedCommits(
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 {
+ allStats = readCommitStatsFromArchivedTimeline(metaClient);
+ }
+ 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();
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ List<Comparable[]> allCommits;
+ try {
+ if (!skipMetadata) {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ }
+ allCommits = archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .map(instant -> readArchivedCommit(archivedTimeline, instant,
skipMetadata))
+ .collect(Collectors.toList());
+ } finally {
+ if (!skipMetadata) {
+ // free the metadata that was loaded in memory for this command
+ archivedTimeline.getInstants().forEach(
+ instant ->
archivedTimeline.clearInstantDetailsFromMemory(instant.requestedTime()));
+ }
+ }
+
+ 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);
+ }
+
+ /**
+ * Reads the write stats of the archived commit and delta commit instants
through the
+ * archived timeline, which resolves the archive format from the table
version (LSM
+ * timeline for table version 8 and above, log format before that).
+ */
+ private List<Comparable[]>
readCommitStatsFromArchivedTimeline(HoodieTableMetaClient metaClient) {
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ try {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ return archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .filter(instant ->
HoodieTimeline.COMMIT_ACTION.equals(instant.getAction())
+ ||
HoodieTimeline.DELTA_COMMIT_ACTION.equals(instant.getAction()))
+ .flatMap(instant -> readWriteStatRows(archivedTimeline, instant))
+ .collect(Collectors.toList());
+ } finally {
+ // free the metadata that was loaded in memory for this command
+ archivedTimeline.getInstants().forEach(
+ instant ->
archivedTimeline.clearInstantDetailsFromMemory(instant.requestedTime()));
+ }
+ }
+
+ private 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 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 {
Review Comment:
This legacy reader was the default, tested path before this change; now it
is reachable only through `--archiveFolderPattern`, which no test passes (`grep
-rn archiveFolderPattern` hits only the `@ShellOption`). With the v1 serde
issue above it is also the only stats path that works on a v6 table. Could we
add a case that creates a table at version 6, archives with
`TimelineArchiverV1`, and asserts `show archived commit stats
--archiveFolderPattern "archived/.commits_.archive*"`, or at least record the
gap in the PR body?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/FileSystemViewCommand.java:
##########
@@ -150,10 +151,16 @@ public String showLatestFileSlices(
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()) {
Review Comment:
This fallback is not reached by any test: `TestFileSystemViewCommand` writes
`3.commit` / `4.commit`, which `InstantGeneratorV2.createInstant` accepts as
COMPLETED (legacy name, completion time from mtime), so `lastInstant()` is
always present. The Spark twin of this guard (`ShowFileSystemViewProcedure`,
HUDI-7845 #11418) returns an empty result on this path. Could we add one case
on a table with no completed instant asserting `show fsview latest` returns an
empty table, or drop the guard from this PR?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java:
##########
@@ -227,19 +222,29 @@ public void testOverwriteHoodieProperties() throws
IOException {
.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()));
+
+ // Add properties that are now present in Hudi 1.x by default
+ if (result.containsKey(HoodieTableConfig.TIMELINE_PATH.key())) {
+ expected.putIfAbsent(HoodieTableConfig.TIMELINE_PATH.key(),
result.get(HoodieTableConfig.TIMELINE_PATH.key()));
Review Comment:
These three `putIfAbsent(key, result.get(key))` lines copy the actual into
the expected, so those keys cannot fail, and the rendered check below is
key-presence over the union `RepairsCommand.overwriteHoodieProperties` always
prints (lines 171-185). The values are deterministic:
`TIMELINE_PATH.defaultValue()`, `TIMELINE_HISTORY_PATH.defaultValue()`, and
`String.valueOf(HoodieTableVersion.current().versionCode())` (INITIAL_VERSION
is copied from the old props). Could we assert those constants and rebuild the
exact-table comparison from the observed old/new key union, so the value
columns are checked again?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -107,9 +115,113 @@ public String showArchivedCommits(
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 {
+ allStats = readCommitStatsFromArchivedTimeline(metaClient);
+ }
+ 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();
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ List<Comparable[]> allCommits;
+ try {
+ if (!skipMetadata) {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ }
+ allCommits = archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .map(instant -> readArchivedCommit(archivedTimeline, instant,
skipMetadata))
+ .collect(Collectors.toList());
+ } finally {
+ if (!skipMetadata) {
+ // free the metadata that was loaded in memory for this command
+ archivedTimeline.getInstants().forEach(
+ instant ->
archivedTimeline.clearInstantDetailsFromMemory(instant.requestedTime()));
+ }
+ }
+
+ 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);
+ }
+
+ /**
+ * Reads the write stats of the archived commit and delta commit instants
through the
+ * archived timeline, which resolves the archive format from the table
version (LSM
+ * timeline for table version 8 and above, log format before that).
+ */
+ private List<Comparable[]>
readCommitStatsFromArchivedTimeline(HoodieTableMetaClient metaClient) {
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ try {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ return archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .filter(instant ->
HoodieTimeline.COMMIT_ACTION.equals(instant.getAction())
+ ||
HoodieTimeline.DELTA_COMMIT_ACTION.equals(instant.getAction()))
+ .flatMap(instant -> readWriteStatRows(archivedTimeline, instant))
+ .collect(Collectors.toList());
+ } finally {
+ // free the metadata that was loaded in memory for this command
+ archivedTimeline.getInstants().forEach(
+ instant ->
archivedTimeline.clearInstantDetailsFromMemory(instant.requestedTime()));
+ }
+ }
+
+ private 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 metadata.getPartitionToWriteStats().values().stream()
Review Comment:
`sortPartitions` sorts the metadata column, but the stat rows still come out
in `HashMap` order from `HoodieAvroUtils.convertToSpecificRecord`, so the
FIRST-then-SECOND expectation in `testShowArchivedCommits` rests on 16-bucket
iteration order. Could we sort here as well?
```suggestion
return sortByKey(metadata.getPartitionToWriteStats()).values().stream()
```
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestFileSystemViewCommand.java:
##########
@@ -110,8 +111,9 @@ private void createNonpartitionedTable() throws IOException
{
.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"));
+
Files.createDirectories(Paths.get(metaClient.getTimelinePath().toString()));
Review Comment:
nit, feel free to ignore: `FileCreateUtils.createCommit(metaClient,
commitTime)` already creates the directory and writes the native
`<requested>_<completion>.commit` name; the hand-rolled `N.commit` here (and at
line 158, and `TestRepairsCommand` lines 124 / 159) parses as a legacy instant
(`InstantGeneratorV2` sets `isLegacy = true` and takes the completion time from
mtime), so the fixture exercises the 0.x-compat arm. Could we use the helper at
the four sites?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java:
##########
@@ -104,26 +105,28 @@ public void init() throws Exception {
try (HoodieTableMetadataWriter metadataWriter =
SparkHoodieBackedTableMetadataWriter.create(metaClient.getStorageConf(),
config, context)) {
HoodieTestTable hoodieTestTable = HoodieMetadataTestTable.of(metaClient,
metadataWriter, Option.of(context))
.withPartitionMetaFiles(DEFAULT_PARTITION_PATHS)
- .addCommit("100")
+ .addCommit("100", Option.of("100001"), Option.empty())
.withBaseFilesInPartitions(partitionAndFileId).getLeft()
- .addCommit("101");
+ .addCommit("101", Option.of("101001"), Option.empty());
-
hoodieTestTable.addCommit("102").withBaseFilesInPartitions(partitionAndFileId);
+ hoodieTestTable.addCommit("102", Option.of("102001"),
Option.empty()).withBaseFilesInPartitions(partitionAndFileId);
HoodieSavepointMetadata savepointMetadata2 =
hoodieTestTable.doSavepoint("102");
- hoodieTestTable.addSavepoint("102", savepointMetadata2);
+ 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");
+ }
+
+ hoodieTestTable.addCommit("105", Option.of("105001"),
Option.empty()).withBaseFilesInPartitions(partitionAndFileId);
+ HoodieSavepointMetadata savepointMetadata =
hoodieTestTable.doSavepoint("105");
+ hoodieTestTable.addSavepointCommit("105", Option.of("105002"),
savepointMetadata);
-
hoodieTestTable.addCommit("105").withBaseFilesInPartitions(partitionAndFileId);
- HoodieSavepointMetadata savepointMetadata =
hoodieTestTable.doSavepoint("105");
- hoodieTestTable.addSavepoint("105", savepointMetadata);
+ hoodieTestTable.addCommit("106", Option.of("106001"),
Option.empty()).withBaseFilesInPartitions(partitionAndFileId);
Review Comment:
The expected rows in `testShowRestores` are built with
`readRestoreMetadata`, the same call the command makes, and there is no count
assertion, so the fixture is not pinned: dropping `client.rollback("106")` is
invisible, and a run that produced zero restores would still pass. Could we
assert the restore count (2) and the rolled-back instants (`103`, `106`) before
comparing the rendering?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java:
##########
@@ -257,16 +262,17 @@ public void testRemoveCorruptedPendingCleanAction()
throws IOException {
for (int i = 100; i < 104; i++) {
String timestamp = String.valueOf(i);
// Write corrupted requested Clean File
-
HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
timestamp, conf);
+ Path filePath = new Path(metaClient.getTimelinePath() + "/" + timestamp
+ ".clean.requested");
+ HoodieTestDataGenerator.createEmptyFile(tablePath, filePath, conf);
Review Comment:
nit, feel free to ignore: this inlines the body of
`HoodieTestDataGenerator.createEmptyCleanRequestedFile`, which now has no
callers repo-wide. Could we keep the helper?
```suggestion
HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
timestamp, conf);
```
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java:
##########
@@ -95,6 +95,7 @@ public void init() throws Exception {
// not valid (empty commit metadata, etc)
HoodieMetadataConfig.newBuilder()
.withMetadataIndexColumnStats(false)
+ .enable(false)
Review Comment:
nit, feel free to ignore: with `enable(false)` the
`SparkHoodieBackedTableMetadataWriter` initialises nothing
(`HoodieBackedTableMetadataWriter` gates on `isMetadataTableEnabled()`), and
none of `addCommit` / `doSavepoint` / `addSavepointCommit` /
`withBaseFilesInPartitions` is overridden by `HoodieMetadataTestTable`, so the
writer and the try-with-resources are inert. Could we use
`HoodieTestTable.of(metaClient)` and drop the writer plus
`withMetadataIndexColumnStats(false)`?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCleansCommand.java:
##########
@@ -86,7 +86,7 @@ public void init() throws Exception {
// 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");
Review Comment:
nit, feel free to ignore: `TableCommand` already has a 5-arg `createTable`
overload that defaults to `HoodieTableVersion.current().versionCode()`
(`TableCommand.java:114-121`) with zero callers, while all 32 CLI test sites
use the 6-arg form. Keeping the explicit argument is consistent with the
untouched `ITTest*` sites, so leaving it seems fine; should the dead overload
be removed instead, or used here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
