@ShellMethod(key = "stats filesizes", value = "File Sizes. Display summary stats on sizes of files")
  public String fileSizeStats(
      @ShellOption(value = {"--partitionPath"}, help = "regex to select files, eg: 2016/08/02",
          defaultValue = "*/*/*") final String globRegex,
      @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.storage;
    String globPath =
        String.format("%s/%s/*", HoodieCLI.getTableMetaClient().getBasePath(), globRegex);
    List<StoragePathInfo> pathInfoList = FSUtils.getGlobStatusExcludingMetaFolder(storage,
        new StoragePath(globPath));

    // max, min, #small files < 10MB, 50th, avg, 95th
    Histogram globalHistogram = new Histogram(new UniformReservoir(MAX_FILES));
    HashMap<String, Histogram> commitHistoMap = new HashMap<>();
    for (StoragePathInfo pathInfo : pathInfoList) {
      String instantTime = FSUtils.getCommitTime(pathInfo.getPath().getName());
      long sz = pathInfo.getLength();
      if (!commitHistoMap.containsKey(instantTime)) {
        commitHistoMap.put(instantTime, new Histogram(new UniformReservoir(MAX_FILES)));
      }
      commitHistoMap.get(instantTime).update(sz);
      globalHistogram.update(sz);
    }

    List<Comparable[]> rows = new ArrayList<>();
    for (Map.Entry<String, Histogram> entry : commitHistoMap.entrySet()) {
      Snapshot s = entry.getValue().getSnapshot();
      rows.add(printFileSizeHistogram(entry.getKey(), s));
    }
    Snapshot s = globalHistogram.getSnapshot();
    rows.add(printFileSizeHistogram("ALL", s));

    TableHeader header = new TableHeader()
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_MIN)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_10TH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_50TH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_AVG)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_95TH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_MAX)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_NUM_FILES)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_STD_DEV);
    return HoodiePrintHelper.print(header, getFieldNameToConverterMap(), sortByField, descending, limit, headerOnly, rows);
  }
@Test
  public void testFileSizeStats() throws Exception {
    String commit1 = "100";
    String commit2 = "101";
    Map<String, Integer[]> data = new LinkedHashMap<>();
    data.put(commit1, new Integer[] {100, 120, 150});
    data.put(commit2, new Integer[] {200, 180, 250, 300});

    // generate data file
    String partition1 = HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
    String partition2 = HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
    String partition3 = HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;

    HoodieTestTable testTable = HoodieTestTable.of(HoodieCLI.getTableMetaClient());
    Integer[] data1 = data.get(commit1);
    assertTrue(3 <= data1.length);
    testTable.addCommit(commit1)
        .withBaseFilesInPartition(partition1, data1[0])
        .withBaseFilesInPartition(partition2, data1[1])
        .withBaseFilesInPartition(partition3, data1[2]);

    Integer[] data2 = data.get(commit2);
    assertTrue(4 <= data2.length);
    testTable.addCommit(commit2)
        .withBaseFilesInPartition(partition1, data2[0])
        .withBaseFilesInPartition(partition2, data2[1], data2[2])
        .withBaseFilesInPartition(partition3, data2[3]);

    Object result = shell.evaluate(() -> "stats filesizes");
    assertTrue(ShellEvaluationResultUtil.isSuccess(result));

    Histogram globalHistogram = new Histogram(new UniformReservoir(StatsCommand.MAX_FILES));
    HashMap<String, Histogram> commitHistoMap = new HashMap<>();
    data.forEach((k, v) -> {
      commitHistoMap.put(k, new Histogram(new UniformReservoir(StatsCommand.MAX_FILES)));
      for (int value : v) {
        commitHistoMap.get(k).update(value);
        globalHistogram.update(value);
      }
    });

    // generate expect
    List<Comparable[]> rows = new ArrayList<>();
    for (Map.Entry<String, Histogram> entry : commitHistoMap.entrySet()) {
      Snapshot s = entry.getValue().getSnapshot();
      rows.add(new StatsCommand().printFileSizeHistogram(entry.getKey(), s));
    }
    Snapshot s = globalHistogram.getSnapshot();
    rows.add(new StatsCommand().printFileSizeHistogram("ALL", s));

    TableHeader header = new TableHeader()
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_MIN)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_10TH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_50TH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_AVG)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_95TH)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_MAX)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_NUM_FILES)
        .addTableHeaderField(HoodieTableHeaderFields.HEADER_HISTOGRAM_STD_DEV);
    String expect = HoodiePrintHelper.print(header, new StatsCommand().getFieldNameToConverterMap(),
        "", false, -1, false, rows);
    expect = removeNonWordAndStripSpace(expect);
    String got = removeNonWordAndStripSpace(result.toString());
    assertEquals(expect, got);
  }