voonhous commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3888875957
##########
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:
Done in b85941ea: testShowArchivedCommitsOnLegacyArchive covers both paths;
the reader now skips entries without metadata and renders the completed entry
only.
##########
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:
Done in e721392e: case-insensitive match.
--
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]
