voonhous commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3949858235
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -107,9 +120,182 @@ 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 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 archive files outside the
range are never
+ * opened and the payloads outside it never held.
+ */
+ 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 {
Review Comment:
Fixed: the legacy reader goes through `sortByKey` as well, so both stats
paths render partitions in key order and the shared expectation holds by
construction rather than by `HashMap` order.
--
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]
