@ShellMethod(key = "show logfile records", value = "Read records from log files")
  public String showLogFileRecords(
      @ShellOption(value = {"--limit"}, help = "Limit commits",
          defaultValue = "10") final Integer limit,
      @ShellOption(value = "--logFilePathPattern",
          help = "Fully qualified paths for the log files") final String logFilePathPattern,
      @ShellOption(value = "--mergeRecords", help = "If the records in the log files should be merged",
              defaultValue = "false") final Boolean shouldMerge)
      throws IOException {

    System.out.println("===============> Showing only " + limit + " records <===============");

    HoodieTableMetaClient client = HoodieCLI.getTableMetaClient();
    HoodieStorage storage = client.getStorage();
    List<String> logFilePaths = FSUtils.getGlobStatusExcludingMetaFolder(
        storage, new StoragePath(logFilePathPattern)).stream()
        .map(status -> status.getPath().toString()).sorted(Comparator.reverseOrder())
        .collect(Collectors.toList());

    // logFilePaths size must > 1
    checkArgument(logFilePaths.size() > 0, "There is no log file");

    // TODO : readerSchema can change across blocks/log files, fix this inside Scanner
    Schema readerSchema = null;
    // get schema from last log file
    for (int i = logFilePaths.size() - 1; i >= 0; i--) {
      Schema schema = TableSchemaResolver.readSchemaFromLogFile(
          storage, new StoragePath(logFilePaths.get(i)));
      if (schema != null) {
        readerSchema = schema;
        break;
      }
    }
    Objects.requireNonNull(readerSchema);
    List<IndexedRecord> allRecords = new ArrayList<>();

    if (shouldMerge) {
      System.out.println("===========================> MERGING RECORDS <===================");
      HoodieMergedLogRecordScanner scanner =
          HoodieMergedLogRecordScanner.newBuilder()
              .withStorage(storage)
              .withBasePath(client.getBasePath())
              .withLogFilePaths(logFilePaths)
              .withReaderSchema(readerSchema)
              .withLatestInstantTime(
                  client.getActiveTimeline()
                      .getCommitAndReplaceTimeline().lastInstant().get().getTimestamp())
              .withReverseReader(
                  Boolean.parseBoolean(
                      HoodieReaderConfig.COMPACTION_REVERSE_LOG_READ_ENABLE.defaultValue()))
              .withBufferSize(HoodieMemoryConfig.MAX_DFS_STREAM_BUFFER_SIZE.defaultValue())
              .withMaxMemorySizeInBytes(
                  HoodieMemoryConfig.DEFAULT_MAX_MEMORY_FOR_SPILLABLE_MAP_IN_BYTES)
              .withSpillableMapBasePath(FileIOUtils.getDefaultSpillableMapBasePath())
              .withDiskMapType(HoodieCommonConfig.SPILLABLE_DISK_MAP_TYPE.defaultValue())
              .withBitCaskDiskMapCompressionEnabled(HoodieCommonConfig.DISK_MAP_BITCASK_COMPRESSION_ENABLED.defaultValue())
              .withOptimizedLogBlocksScan(Boolean.parseBoolean(HoodieReaderConfig.ENABLE_OPTIMIZED_LOG_BLOCKS_SCAN.defaultValue()))
              .build();
      for (HoodieRecord hoodieRecord : scanner) {
        Option<HoodieAvroIndexedRecord> record = hoodieRecord.toIndexedRecord(readerSchema, new Properties());
        if (allRecords.size() < limit) {
          allRecords.add(record.get().getData());
        }
      }
    } else {
      for (String logFile : logFilePaths) {
        Schema writerSchema = TableSchemaResolver.readSchemaFromLogFile(
            client.getStorage(), new StoragePath(logFile));
        try (HoodieLogFormat.Reader reader =
                 HoodieLogFormat.newReader(storage, new HoodieLogFile(new StoragePath(logFile)), writerSchema)) {
          // read the avro blocks
          while (reader.hasNext()) {
            HoodieLogBlock n = reader.next();
            if (n instanceof HoodieDataBlock) {
              HoodieDataBlock blk = (HoodieDataBlock) n;
              try (ClosableIterator<HoodieRecord<IndexedRecord>> recordItr = blk.getRecordIterator(HoodieRecordType.AVRO)) {
                recordItr.forEachRemaining(record -> {
                  if (allRecords.size() < limit) {
                    allRecords.add(record.getData());
                  }
                });
              }
            }
          }
        }
        if (allRecords.size() >= limit) {
          break;
        }
      }
    }
    String[][] rows = new String[allRecords.size()][];
    int i = 0;
    for (IndexedRecord record : allRecords) {
      String[] data = new String[1];
      data[0] = record.toString();
      rows[i] = data;
      i++;
    }
    return HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows);
  }
@Test
  public void testShowLogFileRecords() throws IOException, URISyntaxException {
    Object result = shell.evaluate(() -> "show logfile records --logFilePathPattern " + partitionPath + "/*");
    assertTrue(ShellEvaluationResultUtil.isSuccess(result));

    // construct expect result, get 10 records.
    List<IndexedRecord> records = SchemaTestUtil.generateTestRecords(0, 10);
    String[][] rows = records.stream().map(r -> new String[] {r.toString()}).toArray(String[][]::new);
    String expected = HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows);
    expected = removeNonWordAndStripSpace(expected);
    String got = removeNonWordAndStripSpace(result.toString());
    assertEquals(expected, got);
  }