@ShellMethod(key = "show archived commit stats", value = "Read commits from archived files and show file group details")
  public String showArchivedCommits(
      @ShellOption(value = {"--archiveFolderPattern"}, help = "Archive Folder", defaultValue = "") String folder,
      @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 = folder != null && !folder.isEmpty()
        ? new StoragePath(metaClient.getMetaPath(), folder)
        : new StoragePath(metaClient.getArchivePath(), ".commits_.archive*");
    HoodieStorage storage = metaClient.getStorage();
    List<StoragePathInfo> pathInfoList = storage.globEntries(archivePath);
    List<Comparable[]> allStats = new ArrayList<>();
    for (StoragePathInfo pathInfo : pathInfoList) {
      // read the archived file
      try (Reader reader = HoodieLogFormat.newReader(storage, new HoodieLogFile(pathInfo.getPath()),
          HoodieArchivedMetaEntry.getClassSchema())) {
        List<IndexedRecord> readRecords = new ArrayList<>();
        // read the avro blocks
        while (reader.hasNext()) {
          HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next();
          blk.getRecordIterator(HoodieRecordType.AVRO).forEachRemaining(r -> readRecords.add((IndexedRecord) r.getData()));
        }
        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);
      }
    }
    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);
  }
@Test
  public void testShowArchivedCommits() {
    Object result = shell.evaluate(() -> "show archived commit stats");
    assertTrue(ShellEvaluationResultUtil.isSuccess(result));

    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");

    // Generate expected data
    final List<Comparable[]> rows = new ArrayList<>();
    for (int i = 100; i < 102; 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(result.toString());
    assertEquals(expectedResult, got);
  }