@Override
  public List<HoodieColumnRangeMetadata<Comparable>> readColumnStatsFromMetadata(HoodieStorage storage,
                                                                                 StoragePath filePath,
                                                                                 List<String> columnList) {
    ParquetMetadata metadata = readMetadata(storage, filePath);

    // Collect stats from all individual Parquet blocks
    Stream<HoodieColumnRangeMetadata<Comparable>> hoodieColumnRangeMetadataStream =
        metadata.getBlocks().stream().sequential().flatMap(blockMetaData ->
            blockMetaData.getColumns().stream()
                    .filter(f -> columnList.contains(f.getPath().toDotString()))
                    .map(columnChunkMetaData -> {
                      Statistics stats = columnChunkMetaData.getStatistics();
                      return (HoodieColumnRangeMetadata<Comparable>) HoodieColumnRangeMetadata.<Comparable>create(
                          filePath.getName(),
                          columnChunkMetaData.getPath().toDotString(),
                          convertToNativeJavaType(
                              columnChunkMetaData.getPrimitiveType(),
                              stats.genericGetMin()),
                          convertToNativeJavaType(
                              columnChunkMetaData.getPrimitiveType(),
                              stats.genericGetMax()),
                          // NOTE: In case when column contains only nulls Parquet won't be creating
                          //       stats for it instead returning stubbed (empty) object. In that case
                          //       we have to equate number of nulls to the value count ourselves
                          stats.isEmpty() ? columnChunkMetaData.getValueCount() : stats.getNumNulls(),
                          columnChunkMetaData.getValueCount(),
                          columnChunkMetaData.getTotalSize(),
                          columnChunkMetaData.getTotalUncompressedSize());
                    })
        );

    Map<String, List<HoodieColumnRangeMetadata<Comparable>>> columnToStatsListMap =
        hoodieColumnRangeMetadataStream.collect(Collectors.groupingBy(HoodieColumnRangeMetadata::getColumnName));

    // Combine those into file-level statistics
    // NOTE: Inlining this var makes javac (1.8) upset (due to its inability to infer
    // expression type correctly)
    Stream<HoodieColumnRangeMetadata<Comparable>> stream = columnToStatsListMap.values()
        .stream()
        .map(this::getColumnRangeInFile);

    return stream.collect(Collectors.toList());
  }
@Test
  public void testReadColumnStatsFromMetadata() throws Exception {
    List<Pair<Pair<String, String>, Boolean>> valueList = new ArrayList<>();
    String minKey = "z";
    String maxKey = "0";
    String minValue = "z";
    String maxValue = "0";
    int nullValueCount = 0;
    int totalCount = 1000;
    String partitionPath = "path1";
    for (int i = 0; i < totalCount; i++) {
      boolean nullifyData = i % 3 == 0;
      String rowKey = UUID.randomUUID().toString();
      String value = String.valueOf(i);
      valueList.add(Pair.of(Pair.of(rowKey, value), nullifyData));
      minKey = (minKey.compareTo(rowKey) > 0) ? rowKey : minKey;
      maxKey = (maxKey.compareTo(rowKey) < 0) ? rowKey : maxKey;

      if (nullifyData) {
        nullValueCount++;
      } else {
        minValue = (minValue.compareTo(value) > 0) ? value : minValue;
        maxValue = (maxValue.compareTo(value) < 0) ? value : maxValue;
      }
    }

    String fileName = "test.parquet";
    String filePath = new StoragePath(basePath, fileName).toString();
    String recordKeyField = "id";
    String partitionPathField = "partition";
    String dataField = "data";
    Schema schema = getSchema(recordKeyField, partitionPathField, dataField);

    BloomFilter filter = BloomFilterFactory
        .createBloomFilter(1000, 0.0001, 10000, BloomFilterTypeCode.SIMPLE.name());
    HoodieAvroWriteSupport writeSupport =
        new HoodieAvroWriteSupport(new AvroSchemaConverter().convert(schema), schema, Option.of(filter), new Properties());
    try (ParquetWriter writer = new ParquetWriter(new Path(filePath), writeSupport, CompressionCodecName.GZIP,
        120 * 1024 * 1024, ParquetWriter.DEFAULT_PAGE_SIZE)) {
      valueList.forEach(entry -> {
        GenericRecord rec = new GenericData.Record(schema);
        rec.put(recordKeyField, entry.getLeft().getLeft());
        rec.put(partitionPathField, partitionPath);
        if (entry.getRight()) {
          rec.put(dataField, null);
        } else {
          rec.put(dataField, entry.getLeft().getRight());
        }
        try {
          writer.write(rec);
        } catch (IOException e) {
          throw new RuntimeException(e);
        }
        writeSupport.add(entry.getLeft().getLeft());
      });
    }

    List<String> columnList = new ArrayList<>();
    columnList.add(recordKeyField);
    columnList.add(partitionPathField);
    columnList.add(dataField);

    List<HoodieColumnRangeMetadata<Comparable>> columnRangeMetadataList = parquetUtils.readColumnStatsFromMetadata(
            HoodieTestUtils.getStorage(filePath), new StoragePath(filePath), columnList)
        .stream()
        .sorted(Comparator.comparing(HoodieColumnRangeMetadata::getColumnName))
        .collect(Collectors.toList());
    assertEquals(3, columnRangeMetadataList.size(), "Should return column stats of 3 columns");
    validateColumnRangeMetadata(columnRangeMetadataList.get(0),
        fileName, dataField, minValue, maxValue, nullValueCount, totalCount);
    validateColumnRangeMetadata(columnRangeMetadataList.get(1),
        fileName, recordKeyField, minKey, maxKey, 0, totalCount);
    validateColumnRangeMetadata(columnRangeMetadataList.get(2),
        fileName, partitionPathField, partitionPath, partitionPath, 0, totalCount);
  }