You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2021/11/21 10:24:40 UTC
[GitHub] [hudi] xiarixiaoyao commented on a change in pull request #4060: [HUDI-2814] Addressing issues w/ Z-order Layout Optimization
xiarixiaoyao commented on a change in pull request #4060:
URL: https://github.com/apache/hudi/pull/4060#discussion_r753778621
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
##########
@@ -284,55 +288,102 @@ public Boolean apply(String recordKey) {
/**
* Parse min/max statistics stored in parquet footers for all columns.
*/
- public Collection<HoodieColumnRangeMetadata<Comparable>> readRangeFromParquetMetadata(Configuration conf, Path parquetFilePath, List<String> cols) {
+ public List<HoodieColumnRangeMetadata<Comparable>> readRangeFromParquetMetadata(
+ @Nonnull Configuration conf,
+ @Nonnull Path parquetFilePath,
+ @Nonnull List<String> cols
+ ) {
ParquetMetadata metadata = readMetadata(conf, parquetFilePath);
- // collect stats from all parquet blocks
- Map<String, List<HoodieColumnRangeMetadata<Comparable>>> columnToStatsListMap = metadata.getBlocks().stream().flatMap(blockMetaData -> {
- return blockMetaData.getColumns().stream().filter(f -> cols.contains(f.getPath().toDotString())).map(columnChunkMetaData ->
- new HoodieColumnRangeMetadata<>(parquetFilePath.getName(), columnChunkMetaData.getPath().toDotString(),
- columnChunkMetaData.getStatistics().genericGetMin(),
- columnChunkMetaData.getStatistics().genericGetMax(),
- columnChunkMetaData.getStatistics().getNumNulls(),
- columnChunkMetaData.getPrimitiveType().stringifier()));
- }).collect(Collectors.groupingBy(e -> e.getColumnName()));
-
- // we only intend to keep file level statistics.
- return new ArrayList<>(columnToStatsListMap.values().stream()
- .map(blocks -> getColumnRangeInFile(blocks))
- .collect(Collectors.toList()));
+ // Collect stats from all individual Parquet blocks
+ Map<String, List<HoodieColumnRangeMetadata<Comparable>>> columnToStatsListMap =
+ metadata.getBlocks()
+ .stream()
+ .sequential()
+ .flatMap(blockMetaData ->
+ blockMetaData.getColumns()
+ .stream()
+ .filter(f -> cols.contains(f.getPath().toDotString()))
+ .map(columnChunkMetaData ->
+ new HoodieColumnRangeMetadata<Comparable>(
+ parquetFilePath.getName(),
+ columnChunkMetaData.getPath().toDotString(),
+ convertToNativeJavaType(
+ columnChunkMetaData.getPrimitiveType(),
+ columnChunkMetaData.getStatistics().genericGetMin()),
+ convertToNativeJavaType(
+ columnChunkMetaData.getPrimitiveType(),
+ columnChunkMetaData.getStatistics().genericGetMax()),
+ columnChunkMetaData.getStatistics().getNumNulls(),
+ columnChunkMetaData.getPrimitiveType().stringifier()))
+ )
+ .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());
}
- private HoodieColumnRangeMetadata<Comparable> getColumnRangeInFile(final List<HoodieColumnRangeMetadata<Comparable>> blockRanges) {
+ private <T extends Comparable<T>> HoodieColumnRangeMetadata<T> getColumnRangeInFile(
+ @Nonnull List<HoodieColumnRangeMetadata<T>> blockRanges
+ ) {
if (blockRanges.size() == 1) {
// only one block in parquet file. we can just return that range.
return blockRanges.get(0);
- } else {
- // there are multiple blocks. Compute min(block_mins) and max(block_maxs)
- return blockRanges.stream().reduce((b1, b2) -> combineRanges(b1, b2)).get();
}
+
+ // there are multiple blocks. Compute min(block_mins) and max(block_maxs)
+ return blockRanges.stream()
+ .sequential()
+ .reduce(this::combineRanges).get();
}
- private HoodieColumnRangeMetadata<Comparable> combineRanges(HoodieColumnRangeMetadata<Comparable> range1,
- HoodieColumnRangeMetadata<Comparable> range2) {
- final Comparable minValue;
- final Comparable maxValue;
- if (range1.getMinValue() != null && range2.getMinValue() != null) {
- minValue = range1.getMinValue().compareTo(range2.getMinValue()) < 0 ? range1.getMinValue() : range2.getMinValue();
- } else if (range1.getMinValue() == null) {
- minValue = range2.getMinValue();
+ private <T extends Comparable<T>> HoodieColumnRangeMetadata<T> combineRanges(
+ HoodieColumnRangeMetadata<T> one,
+ HoodieColumnRangeMetadata<T> another
+ ) {
+ final T minValue;
+ final T maxValue;
+ if (one.getMinValue() != null && another.getMinValue() != null) {
+ minValue = one.getMinValue().compareTo(another.getMinValue()) < 0 ? one.getMinValue() : another.getMinValue();
+ } else if (one.getMinValue() == null) {
+ minValue = another.getMinValue();
} else {
- minValue = range1.getMinValue();
+ minValue = one.getMinValue();
}
- if (range1.getMaxValue() != null && range2.getMaxValue() != null) {
- maxValue = range1.getMaxValue().compareTo(range2.getMaxValue()) < 0 ? range2.getMaxValue() : range1.getMaxValue();
- } else if (range1.getMaxValue() == null) {
- maxValue = range2.getMaxValue();
+ if (one.getMaxValue() != null && another.getMaxValue() != null) {
+ maxValue = one.getMaxValue().compareTo(another.getMaxValue()) < 0 ? another.getMaxValue() : one.getMaxValue();
+ } else if (one.getMaxValue() == null) {
+ maxValue = another.getMaxValue();
} else {
- maxValue = range1.getMaxValue();
+ maxValue = one.getMaxValue();
+ }
+
+ return new HoodieColumnRangeMetadata<T>(
+ one.getFilePath(),
+ one.getColumnName(), minValue, maxValue, one.getNumNulls() + another.getNumNulls(), one.getStringifier());
+ }
+
+ private static Comparable<?> convertToNativeJavaType(PrimitiveType primitiveType, Comparable val) {
+ if (primitiveType.getOriginalType() == OriginalType.DECIMAL) {
Review comment:
when we read statistic info from paquet, we can use Statistics.minAsString and Statistics.maxAsString to deal with DecimalType and dateType correctly. and avoids additional conversion operations, as well as synchronization operations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org