@ShellMethod(key = "show logfile metadata", value = "Read commit metadata from log files")
  public String showLogFileCommits(
      @ShellOption(value = "--logFilePathPattern",
          help = "Fully qualified path for the log file") final String logFilePathPattern,
      @ShellOption(value = {"--limit"}, help = "Limit commits", defaultValue = "-1") 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 {

    HoodieStorage storage = HoodieCLI.getTableMetaClient().getStorage();
    List<String> logFilePaths = FSUtils.getGlobStatusExcludingMetaFolder(
        storage, new StoragePath(logFilePathPattern)).stream()
        .map(status -> status.getPath().toString()).collect(Collectors.toList());
    Map<String, List<Tuple3<Tuple2<String, HoodieLogBlockType>, Tuple2<Map<HeaderMetadataType, String>,
        Map<HeaderMetadataType, String>>, Integer>>> commitCountAndMetadata =
        new HashMap<>();
    int numCorruptBlocks = 0;
    int dummyInstantTimeCount = 0;
    String basePath = HoodieCLI.basePath;

    for (String logFilePath : logFilePaths) {
      StoragePath path = new StoragePath(logFilePath);
      String pathString = path.toString();
      String fileName;
      if (pathString.contains(basePath)) {
        String[] split = pathString.split(basePath);
        fileName = split[split.length - 1];
      } else {
        fileName = path.getName();
      }
      Schema writerSchema = TableSchemaResolver.readSchemaFromLogFile(storage, path);
      try (Reader reader = HoodieLogFormat.newReader(storage, new HoodieLogFile(path), writerSchema)) {

        // read the avro blocks
        while (reader.hasNext()) {
          HoodieLogBlock n = reader.next();
          String instantTime;
          AtomicInteger recordCount = new AtomicInteger(0);
          if (n instanceof HoodieCorruptBlock) {
            try {
              instantTime = n.getLogBlockHeader().get(HeaderMetadataType.INSTANT_TIME);
              if (instantTime == null) {
                throw new Exception("Invalid instant time " + instantTime);
              }
            } catch (Exception e) {
              numCorruptBlocks++;
              instantTime = "corrupt_block_" + numCorruptBlocks;
              // could not read metadata for corrupt block
            }
          } else {
            instantTime = n.getLogBlockHeader().get(HeaderMetadataType.INSTANT_TIME);
            if (instantTime == null) {
              // This can happen when reading archived commit files since they were written without any instant time
              dummyInstantTimeCount++;
              instantTime = "dummy_instant_time_" + dummyInstantTimeCount;
            }
            if (n instanceof HoodieDataBlock) {
              try (ClosableIterator<HoodieRecord<IndexedRecord>> recordItr = ((HoodieDataBlock) n).getRecordIterator(HoodieRecordType.AVRO)) {
                recordItr.forEachRemaining(r -> recordCount.incrementAndGet());
              }
            }
          }
          if (commitCountAndMetadata.containsKey(instantTime)) {
            commitCountAndMetadata.get(instantTime).add(
                new Tuple3<>(new Tuple2<>(fileName, n.getBlockType()),
                    new Tuple2<>(n.getLogBlockHeader(), n.getLogBlockFooter()), recordCount.get()));
          } else {
            List<Tuple3<Tuple2<String, HoodieLogBlockType>, Tuple2<Map<HeaderMetadataType, String>,
                Map<HeaderMetadataType, String>>, Integer>> list =
                new ArrayList<>();
            list.add(
                new Tuple3<>(new Tuple2<>(fileName, n.getBlockType()),
                    new Tuple2<>(n.getLogBlockHeader(), n.getLogBlockFooter()), recordCount.get()));
            commitCountAndMetadata.put(instantTime, list);
          }
        }
      }
    }
    List<Comparable[]> rows = new ArrayList<>();
    ObjectMapper objectMapper = new ObjectMapper();
    for (Map.Entry<String, List<Tuple3<Tuple2<String, HoodieLogBlockType>, Tuple2<Map<HeaderMetadataType, String>,
        Map<HeaderMetadataType, String>>, Integer>>> entry : commitCountAndMetadata
        .entrySet()) {
      String instantTime = entry.getKey();
      for (Tuple3<Tuple2<String, HoodieLogBlockType>, Tuple2<Map<HeaderMetadataType, String>,
          Map<HeaderMetadataType, String>>, Integer> tuple3 : entry
          .getValue()) {
        Comparable[] output = new Comparable[6];
        output[0] = tuple3._1()._1();
        output[1] = instantTime;
        output[2] = tuple3._3();
        output[3] = tuple3._1()._2().toString();
        output[4] = objectMapper.writeValueAsString(tuple3._2()._1());
        output[5] = objectMapper.writeValueAsString(tuple3._2()._2());
        rows.add(output);
      }
    }

    TableHeader header = new TableHeader()
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_FILE_PATH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_INSTANT_TIME)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_RECORD_COUNT)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_BLOCK_TYPE)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HEADER_METADATA)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_FOOTER_METADATA);

    return HoodiePrintHelper.print(header, new HashMap<>(), sortByField, descending, limit, headerOnly, rows);
  }
@Test
  public void testShowLogFileCommits() throws JsonProcessingException {
    Object result = shell.evaluate(() -> "show logfile metadata --logFilePathPattern " + partitionPath + "/*");
    assertTrue(ShellEvaluationResultUtil.isSuccess(result));

    TableHeader header = new TableHeader()
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_FILE_PATH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_INSTANT_TIME)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_RECORD_COUNT)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_BLOCK_TYPE)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HEADER_METADATA)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_FOOTER_METADATA);

    // construct expect result, there is only 1 line.
    List<Comparable[]> rows = new ArrayList<>();
    ObjectMapper objectMapper = new ObjectMapper();
    String logFileNamePrefix = DEFAULT_FIRST_PARTITION_PATH + "/test-log-fileid1_" + INSTANT_TIME + ".log";
    rows.add(new Comparable[] {
        logFileNamePrefix + ".1_1-0-1", INSTANT_TIME, 100, dataBlock.getBlockType(),
        objectMapper.writeValueAsString(dataBlock.getLogBlockHeader()),
        objectMapper.writeValueAsString(dataBlock.getLogBlockFooter())});
    rows.add(new Comparable[] {
        logFileNamePrefix + ".2_1-0-1", "103", 0, commandBlock.getBlockType(),
        objectMapper.writeValueAsString(commandBlock.getLogBlockHeader()),
        objectMapper.writeValueAsString(commandBlock.getLogBlockFooter())});

    String expected = HoodiePrintHelper.print(header, new HashMap<>(), "", false, -1, false, rows);
    expected = removeNonWordAndStripSpace(expected);
    String got = removeNonWordAndStripSpace(result.toString());
    assertEquals(expected, got);
  }