@Test public void testShowArchivedCommits() { CommandResult cr = getShell().executeCommand("show archived commit stats"); assertTrue(cr.isSuccess()); 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"); final List<Comparable[]> rows = new ArrayList<>(); for (int i = 100; i < 104; 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); } } String expectedResult = HoodiePrintHelper.print( header, new HashMap<>(), "", false, -1, false, rows); expectedResult = removeNonWordAndStripSpace(expectedResult); String got = removeNonWordAndStripSpace(cr.getResult().toString()); assertEquals(expectedResult, got); }
@CliCommand(value = "show archived commit stats", help = "Read commits from archived files and show details") public String showArchivedCommits( @CliOption(key = {"archiveFolderPattern"}, help = "Archive Folder", unspecifiedDefaultValue = "") String folder, @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { System.out.println("===============> Showing only " + limit + " archived commits <==============="); String basePath = HoodieCLI.getTableMetaClient().getBasePath(); Path archivePath = new Path(HoodieCLI.getTableMetaClient().getArchivePath() + "/.commits_.archive*"); if (folder != null && !folder.isEmpty()) { archivePath = new Path(basePath + "/.hoodie/" + folder); } FileStatus[] fsStatuses = FSUtils.getFs(basePath, HoodieCLI.conf).globStatus(archivePath); List<Comparable[]> allStats = new ArrayList<>(); for (FileStatus fs : fsStatuses) { Reader reader = HoodieLogFormat.newReader(FSUtils.getFs(basePath, HoodieCLI.conf), new HoodieLogFile(fs.getPath()), HoodieArchivedMetaEntry.getClassSchema()); List<IndexedRecord> readRecords = new ArrayList<>(); while (reader.hasNext()) { HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next(); List<IndexedRecord> records = blk.getRecords(); readRecords.addAll(records); } 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)) .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 -> { List<Comparable> row = new ArrayList<>(); row.add(action); row.add(instantTime); row.add(hoodieWriteStat.getPartitionPath()); row.add(hoodieWriteStat.getFileId()); row.add(hoodieWriteStat.getPrevCommit()); row.add(hoodieWriteStat.getNumWrites()); row.add(hoodieWriteStat.getNumInserts()); row.add(hoodieWriteStat.getNumDeletes()); row.add(hoodieWriteStat.getNumUpdateWrites()); row.add(hoodieWriteStat.getTotalLogFiles()); row.add(hoodieWriteStat.getTotalLogBlocks()); row.add(hoodieWriteStat.getTotalCorruptLogBlock()); row.add(hoodieWriteStat.getTotalRollbackBlocks()); row.add(hoodieWriteStat.getTotalLogRecords()); row.add(hoodieWriteStat.getTotalUpdatedRecordsCompacted()); row.add(hoodieWriteStat.getTotalWriteBytes()); row.add(hoodieWriteStat.getTotalWriteErrors()); return row; })).map(rowList -> rowList.toArray(new Comparable[0])); }).collect(Collectors.toList()); allStats.addAll(readCommits); reader.close(); } 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); }
ArchivedCommitsCommand implements CommandMarker { @CliCommand(value = "show archived commit stats", help = "Read commits from archived files and show details") public String showArchivedCommits( @CliOption(key = {"archiveFolderPattern"}, help = "Archive Folder", unspecifiedDefaultValue = "") String folder, @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { System.out.println("===============> Showing only " + limit + " archived commits <==============="); String basePath = HoodieCLI.getTableMetaClient().getBasePath(); Path archivePath = new Path(HoodieCLI.getTableMetaClient().getArchivePath() + "/.commits_.archive*"); if (folder != null && !folder.isEmpty()) { archivePath = new Path(basePath + "/.hoodie/" + folder); } FileStatus[] fsStatuses = FSUtils.getFs(basePath, HoodieCLI.conf).globStatus(archivePath); List<Comparable[]> allStats = new ArrayList<>(); for (FileStatus fs : fsStatuses) { Reader reader = HoodieLogFormat.newReader(FSUtils.getFs(basePath, HoodieCLI.conf), new HoodieLogFile(fs.getPath()), HoodieArchivedMetaEntry.getClassSchema()); List<IndexedRecord> readRecords = new ArrayList<>(); while (reader.hasNext()) { HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next(); List<IndexedRecord> records = blk.getRecords(); readRecords.addAll(records); } 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)) .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 -> { List<Comparable> row = new ArrayList<>(); row.add(action); row.add(instantTime); row.add(hoodieWriteStat.getPartitionPath()); row.add(hoodieWriteStat.getFileId()); row.add(hoodieWriteStat.getPrevCommit()); row.add(hoodieWriteStat.getNumWrites()); row.add(hoodieWriteStat.getNumInserts()); row.add(hoodieWriteStat.getNumDeletes()); row.add(hoodieWriteStat.getNumUpdateWrites()); row.add(hoodieWriteStat.getTotalLogFiles()); row.add(hoodieWriteStat.getTotalLogBlocks()); row.add(hoodieWriteStat.getTotalCorruptLogBlock()); row.add(hoodieWriteStat.getTotalRollbackBlocks()); row.add(hoodieWriteStat.getTotalLogRecords()); row.add(hoodieWriteStat.getTotalUpdatedRecordsCompacted()); row.add(hoodieWriteStat.getTotalWriteBytes()); row.add(hoodieWriteStat.getTotalWriteErrors()); return row; })).map(rowList -> rowList.toArray(new Comparable[0])); }).collect(Collectors.toList()); allStats.addAll(readCommits); reader.close(); } 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); } }
ArchivedCommitsCommand implements CommandMarker { @CliCommand(value = "show archived commit stats", help = "Read commits from archived files and show details") public String showArchivedCommits( @CliOption(key = {"archiveFolderPattern"}, help = "Archive Folder", unspecifiedDefaultValue = "") String folder, @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { System.out.println("===============> Showing only " + limit + " archived commits <==============="); String basePath = HoodieCLI.getTableMetaClient().getBasePath(); Path archivePath = new Path(HoodieCLI.getTableMetaClient().getArchivePath() + "/.commits_.archive*"); if (folder != null && !folder.isEmpty()) { archivePath = new Path(basePath + "/.hoodie/" + folder); } FileStatus[] fsStatuses = FSUtils.getFs(basePath, HoodieCLI.conf).globStatus(archivePath); List<Comparable[]> allStats = new ArrayList<>(); for (FileStatus fs : fsStatuses) { Reader reader = HoodieLogFormat.newReader(FSUtils.getFs(basePath, HoodieCLI.conf), new HoodieLogFile(fs.getPath()), HoodieArchivedMetaEntry.getClassSchema()); List<IndexedRecord> readRecords = new ArrayList<>(); while (reader.hasNext()) { HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next(); List<IndexedRecord> records = blk.getRecords(); readRecords.addAll(records); } 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)) .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 -> { List<Comparable> row = new ArrayList<>(); row.add(action); row.add(instantTime); row.add(hoodieWriteStat.getPartitionPath()); row.add(hoodieWriteStat.getFileId()); row.add(hoodieWriteStat.getPrevCommit()); row.add(hoodieWriteStat.getNumWrites()); row.add(hoodieWriteStat.getNumInserts()); row.add(hoodieWriteStat.getNumDeletes()); row.add(hoodieWriteStat.getNumUpdateWrites()); row.add(hoodieWriteStat.getTotalLogFiles()); row.add(hoodieWriteStat.getTotalLogBlocks()); row.add(hoodieWriteStat.getTotalCorruptLogBlock()); row.add(hoodieWriteStat.getTotalRollbackBlocks()); row.add(hoodieWriteStat.getTotalLogRecords()); row.add(hoodieWriteStat.getTotalUpdatedRecordsCompacted()); row.add(hoodieWriteStat.getTotalWriteBytes()); row.add(hoodieWriteStat.getTotalWriteErrors()); return row; })).map(rowList -> rowList.toArray(new Comparable[0])); }).collect(Collectors.toList()); allStats.addAll(readCommits); reader.close(); } 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); }  }
ArchivedCommitsCommand implements CommandMarker { @CliCommand(value = "show archived commit stats", help = "Read commits from archived files and show details") public String showArchivedCommits( @CliOption(key = {"archiveFolderPattern"}, help = "Archive Folder", unspecifiedDefaultValue = "") String folder, @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { System.out.println("===============> Showing only " + limit + " archived commits <==============="); String basePath = HoodieCLI.getTableMetaClient().getBasePath(); Path archivePath = new Path(HoodieCLI.getTableMetaClient().getArchivePath() + "/.commits_.archive*"); if (folder != null && !folder.isEmpty()) { archivePath = new Path(basePath + "/.hoodie/" + folder); } FileStatus[] fsStatuses = FSUtils.getFs(basePath, HoodieCLI.conf).globStatus(archivePath); List<Comparable[]> allStats = new ArrayList<>(); for (FileStatus fs : fsStatuses) { Reader reader = HoodieLogFormat.newReader(FSUtils.getFs(basePath, HoodieCLI.conf), new HoodieLogFile(fs.getPath()), HoodieArchivedMetaEntry.getClassSchema()); List<IndexedRecord> readRecords = new ArrayList<>(); while (reader.hasNext()) { HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next(); List<IndexedRecord> records = blk.getRecords(); readRecords.addAll(records); } 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)) .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 -> { List<Comparable> row = new ArrayList<>(); row.add(action); row.add(instantTime); row.add(hoodieWriteStat.getPartitionPath()); row.add(hoodieWriteStat.getFileId()); row.add(hoodieWriteStat.getPrevCommit()); row.add(hoodieWriteStat.getNumWrites()); row.add(hoodieWriteStat.getNumInserts()); row.add(hoodieWriteStat.getNumDeletes()); row.add(hoodieWriteStat.getNumUpdateWrites()); row.add(hoodieWriteStat.getTotalLogFiles()); row.add(hoodieWriteStat.getTotalLogBlocks()); row.add(hoodieWriteStat.getTotalCorruptLogBlock()); row.add(hoodieWriteStat.getTotalRollbackBlocks()); row.add(hoodieWriteStat.getTotalLogRecords()); row.add(hoodieWriteStat.getTotalUpdatedRecordsCompacted()); row.add(hoodieWriteStat.getTotalWriteBytes()); row.add(hoodieWriteStat.getTotalWriteErrors()); return row; })).map(rowList -> rowList.toArray(new Comparable[0])); }).collect(Collectors.toList()); allStats.addAll(readCommits); reader.close(); } 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); }  @CliCommand(value = "show archived commit stats", help = "Read commits from archived files and show details") String showArchivedCommits(
      @CliOption(key = {"archiveFolderPattern"}, help = "Archive Folder", unspecifiedDefaultValue = "") String folder,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); @CliCommand(value = "show archived commits", help = "Read commits from archived files and show details") String showCommits(
      @CliOption(key = {"skipMetadata"}, help = "Skip displaying commit metadata",
          unspecifiedDefaultValue = "true") boolean skipMetadata,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "10") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); }
ArchivedCommitsCommand implements CommandMarker { @CliCommand(value = "show archived commit stats", help = "Read commits from archived files and show details") public String showArchivedCommits( @CliOption(key = {"archiveFolderPattern"}, help = "Archive Folder", unspecifiedDefaultValue = "") String folder, @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { System.out.println("===============> Showing only " + limit + " archived commits <==============="); String basePath = HoodieCLI.getTableMetaClient().getBasePath(); Path archivePath = new Path(HoodieCLI.getTableMetaClient().getArchivePath() + "/.commits_.archive*"); if (folder != null && !folder.isEmpty()) { archivePath = new Path(basePath + "/.hoodie/" + folder); } FileStatus[] fsStatuses = FSUtils.getFs(basePath, HoodieCLI.conf).globStatus(archivePath); List<Comparable[]> allStats = new ArrayList<>(); for (FileStatus fs : fsStatuses) { Reader reader = HoodieLogFormat.newReader(FSUtils.getFs(basePath, HoodieCLI.conf), new HoodieLogFile(fs.getPath()), HoodieArchivedMetaEntry.getClassSchema()); List<IndexedRecord> readRecords = new ArrayList<>(); while (reader.hasNext()) { HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next(); List<IndexedRecord> records = blk.getRecords(); readRecords.addAll(records); } 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)) .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 -> { List<Comparable> row = new ArrayList<>(); row.add(action); row.add(instantTime); row.add(hoodieWriteStat.getPartitionPath()); row.add(hoodieWriteStat.getFileId()); row.add(hoodieWriteStat.getPrevCommit()); row.add(hoodieWriteStat.getNumWrites()); row.add(hoodieWriteStat.getNumInserts()); row.add(hoodieWriteStat.getNumDeletes()); row.add(hoodieWriteStat.getNumUpdateWrites()); row.add(hoodieWriteStat.getTotalLogFiles()); row.add(hoodieWriteStat.getTotalLogBlocks()); row.add(hoodieWriteStat.getTotalCorruptLogBlock()); row.add(hoodieWriteStat.getTotalRollbackBlocks()); row.add(hoodieWriteStat.getTotalLogRecords()); row.add(hoodieWriteStat.getTotalUpdatedRecordsCompacted()); row.add(hoodieWriteStat.getTotalWriteBytes()); row.add(hoodieWriteStat.getTotalWriteErrors()); return row; })).map(rowList -> rowList.toArray(new Comparable[0])); }).collect(Collectors.toList()); allStats.addAll(readCommits); reader.close(); } 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); }  @CliCommand(value = "show archived commit stats", help = "Read commits from archived files and show details") String showArchivedCommits(
      @CliOption(key = {"archiveFolderPattern"}, help = "Archive Folder", unspecifiedDefaultValue = "") String folder,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); @CliCommand(value = "show archived commits", help = "Read commits from archived files and show details") String showCommits(
      @CliOption(key = {"skipMetadata"}, help = "Skip displaying commit metadata",
          unspecifiedDefaultValue = "true") boolean skipMetadata,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "10") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly);  }