voonhous commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3888877373
##########
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:
Done in b85941ea: stat rows iterate the sorted map; the expectation is
SECOND then FIRST now.
##########
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:
Done in 059d0af2: two restores pinned, rolling back 103 and 106
(`rollback("103")` only un-publishes, so the restore still rolls back the
inflight 103).
##########
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:
Done in 059d0af2: exact `No inflight instants are found.` at 70, and
oldInstantTime3 asserted absent at 15 and 50.
--
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]
