GH-3696: Cache ParsedVersion in FileMetaData and use it in fromParquetMetadata

parquet-column/src/main/java/org/apache/parquet/CorruptStatistics.java

@@ -70,37 +70,67 @@ public static boolean shouldIgnoreStatistics(String createdBy, PrimitiveTypeName
 
     try {
       ParsedVersion version = VersionParser.parse(createdBy);
+      return shouldIgnoreStatistics(version, createdBy, columnType);
+    } catch (RuntimeException | VersionParseException e) {
+      // couldn't parse the created_by field, log what went wrong, don't trust the
+      // stats, but don't make this fatal.
+      warnParseErrorOnce(createdBy, e);
+      return true;
+    }
+  }
 
-      if (!"parquet-mr".equals(version.application)) {
-        // assume other applications don't have this bug
-        return false;
-      }
+  /**
+   * Decides if the statistics from a file should be ignored because they are potentially corrupt.
+   * Use this when the writer version has already been parsed to avoid redundant parsing.
+   *
+   * @param writerVersion the pre-parsed writer version, or {@code null} if unknown/unparseable
+   * @param createdBy     the original created-by string from the file footer (used for logging)
+   * @param columnType    the type of the column that this is checking
+   * @return true if the statistics may be invalid and should be ignored, false otherwise
+   */
+  public static boolean shouldIgnoreStatistics(
+      ParsedVersion writerVersion, String createdBy, PrimitiveTypeName columnType) {
 
-      if (Strings.isNullOrEmpty(version.version)) {
-        warnOnce("Ignoring statistics because created_by did not contain a semver (see PARQUET-251): "
-            + createdBy);
-        return true;
-      }
+    if (columnType != PrimitiveTypeName.BINARY && columnType != PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) {
+      return false;
+    }
+
+    if (writerVersion == null) {
+      warnOnce("Ignoring statistics because created_by is null or empty! See PARQUET-251 and PARQUET-297");
+      return true;
+    }
 
-      SemanticVersion semver = SemanticVersion.parse(version.version);
+    if (!"parquet-mr".equals(writerVersion.application)) {
+      return false;
+    }
+
+    if (Strings.isNullOrEmpty(writerVersion.version)) {
+      warnOnce("Ignoring statistics because created_by did not contain a semver (see PARQUET-251): " + createdBy);
+      return true;
+    }
 
-      if (semver.compareTo(PARQUET_251_FIXED_VERSION) < 0
-          && !(semver.compareTo(CDH_5_PARQUET_251_FIXED_START) >= 0
-              && semver.compareTo(CDH_5_PARQUET_251_FIXED_END) < 0)) {
-        warnOnce("Ignoring statistics because this file was created prior to "
-            + PARQUET_251_FIXED_VERSION
-            + ", see PARQUET-251");
-        return true;
+    if (!writerVersion.hasSemanticVersion()) {
+      try {
+        SemanticVersion.parse(writerVersion.version);
+      } catch (SemanticVersionParseException e) {
+        warnParseErrorOnce(createdBy, e);
       }
+      return true;
+    }
 
-      // this file was created after the fix
-      return false;
-    } catch (RuntimeException | SemanticVersionParseException | VersionParseException e) {
-      // couldn't parse the created_by field, log what went wrong, don't trust the stats,
-      // but don't make this fatal.
-      warnParseErrorOnce(createdBy, e);
+    SemanticVersion semver = writerVersion.getSemanticVersion();
+
+    if (semver.compareTo(PARQUET_251_FIXED_VERSION) < 0
+        && !(semver.compareTo(CDH_5_PARQUET_251_FIXED_START) >= 0
+            && semver.compareTo(CDH_5_PARQUET_251_FIXED_END) < 0)) {
+      warnOnce("Ignoring statistics because this file was created prior to "
+          + PARQUET_251_FIXED_VERSION
+          + ", see PARQUET-251");
       return true;
     }
+
+    // this file was created after the fix
+    return false;
   }
 
   private static void warnParseErrorOnce(String createdBy, Throwable e) {

parquet-column/src/test/java/org/apache/parquet/CorruptStatisticsTest.java

@@ -20,6 +20,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import org.apache.parquet.VersionParser.ParsedVersion;
 import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
 import org.junit.jupiter.api.Test;
 
@@ -129,6 +130,49 @@ public void testCorruptStatistics() {
         .isFalse();
   }
 
+  @Test
+  public void testShouldIgnoreStatisticsWithParsedVersion() throws Exception {
+    String createdBy = "parquet-mr version 1.6.0 (build abc)";
+
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(null, null, PrimitiveTypeName.BINARY))
+        .isTrue();
+
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(null, null, PrimitiveTypeName.INT32))
+        .isFalse();
+
+    ParsedVersion impala = VersionParser.parse("impala version 1.2.0 (build abc)");
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(
+            impala, "impala version 1.2.0 (build abc)", PrimitiveTypeName.BINARY))
+        .isFalse();
+
+    ParsedVersion corrupt = VersionParser.parse(createdBy);
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(corrupt, createdBy, PrimitiveTypeName.BINARY))
+        .isTrue();
+
+    ParsedVersion fixed = VersionParser.parse("parquet-mr version 1.8.0 (build abc)");
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(
+            fixed, "parquet-mr version 1.8.0 (build abc)", PrimitiveTypeName.BINARY))
+        .isFalse();
+
+    ParsedVersion newer = VersionParser.parse("parquet-mr version 1.12.0 (build abc)");
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(
+            newer, "parquet-mr version 1.12.0 (build abc)", PrimitiveTypeName.BINARY))
+        .isFalse();
+
+    // version field present but not a valid semantic version
+    ParsedVersion invalidSemver = new ParsedVersion("parquet-mr", "not-a-semver", "abc");
+    assertThat(invalidSemver.hasSemanticVersion()).isFalse();
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(
+            invalidSemver, "parquet-mr version not-a-semver (build abc)", PrimitiveTypeName.BINARY))
+        .isTrue();
+
+    // empty version field
+    ParsedVersion emptyVersion = new ParsedVersion("parquet-mr", "", "abc");
+    assertThat(CorruptStatistics.shouldIgnoreStatistics(
+            emptyVersion, "parquet-mr version (build abc)", PrimitiveTypeName.BINARY))
+        .isTrue();
+  }
+
   @Test
   public void testDistributionCorruptStatistics() {
     assertThat(CorruptStatistics.shouldIgnoreStatistics(

parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java

@@ -47,6 +47,8 @@
 import org.apache.parquet.CorruptStatistics;
 import org.apache.parquet.ParquetReadOptions;
 import org.apache.parquet.Preconditions;
+import org.apache.parquet.VersionParser.ParsedVersion;
+import org.apache.parquet.VersionParser.VersionParseException;
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.column.EncodingStats;
 import org.apache.parquet.column.ParquetProperties;
@@ -945,7 +947,6 @@ public static org.apache.parquet.column.statistics.Statistics fromParquetStatist
   // Visible for testing
   static org.apache.parquet.column.statistics.Statistics fromParquetStatisticsInternal(
       String createdBy, Statistics formatStats, PrimitiveType type, SortOrder typeSortOrder) {
-    // create stats object based on the column type
     org.apache.parquet.column.statistics.Statistics.Builder statsBuilder =
         org.apache.parquet.column.statistics.Statistics.getBuilderForReading(type);
 
@@ -986,12 +987,65 @@ static org.apache.parquet.column.statistics.Statistics fromParquetStatisticsInte
     return statsBuilder.build();
   }
 
+  // Visible for testing
+  static org.apache.parquet.column.statistics.Statistics fromParquetStatisticsInternal(
+      ParsedVersion writerVersion,
+      String createdBy,
+      Statistics formatStats,
+      PrimitiveType type,
+      SortOrder typeSortOrder) {
+    org.apache.parquet.column.statistics.Statistics.Builder statsBuilder =
+        org.apache.parquet.column.statistics.Statistics.getBuilderForReading(type);
+
+    if (formatStats != null) {
+      // Use the new V2 min-max statistics over the former one if it is filled
+      if (formatStats.isSetMin_value() && formatStats.isSetMax_value()) {
+        byte[] min = formatStats.min_value.array();
+        byte[] max = formatStats.max_value.array();
+        if (isMinMaxStatsSupported(type) || Arrays.equals(min, max)) {
+          statsBuilder.withMin(min);
+          statsBuilder.withMax(max);
+        }
+      } else {
+        boolean isSet = formatStats.isSetMax() && formatStats.isSetMin();
+        boolean maxEqualsMin = isSet ? Arrays.equals(formatStats.getMin(), formatStats.getMax()) : false;
+        boolean sortOrdersMatch = SortOrder.SIGNED == typeSortOrder;
+        // NOTE: See docs in CorruptStatistics for explanation of why this check is needed
+        // The sort order is checked to avoid returning min/max stats that are not
+        // valid with the type's sort order. In previous releases, all stats were
+        // aggregated using a signed byte-wise ordering, which isn't valid for all the
+        // types (e.g. strings, decimals etc.).
+        if (!CorruptStatistics.shouldIgnoreStatistics(writerVersion, createdBy, type.getPrimitiveTypeName())
+            && (sortOrdersMatch || maxEqualsMin)) {
+          if (isSet) {
+            statsBuilder.withMin(formatStats.min.array());
+            statsBuilder.withMax(formatStats.max.array());
+          }
+        }
+      }
+
+      if (formatStats.isSetNull_count()) {
+        statsBuilder.withNumNulls(formatStats.null_count);
+      }
+      if (formatStats.isSetNan_count()) {
+        statsBuilder.withNanCount(formatStats.getNan_count());
+      }
+    }
+    return statsBuilder.build();
+  }
+
   public org.apache.parquet.column.statistics.Statistics fromParquetStatistics(
       String createdBy, Statistics statistics, PrimitiveType type) {
     SortOrder expectedOrder = overrideSortOrderToSigned(type) ? SortOrder.SIGNED : sortOrder(type);
     return fromParquetStatisticsInternal(createdBy, statistics, type, expectedOrder);
   }
 
+  public org.apache.parquet.column.statistics.Statistics fromParquetStatistics(
+      ParsedVersion writerVersion, String createdBy, Statistics statistics, PrimitiveType type) {
+    SortOrder expectedOrder = overrideSortOrderToSigned(type) ? SortOrder.SIGNED : sortOrder(type);
+    return fromParquetStatisticsInternal(writerVersion, createdBy, statistics, type, expectedOrder);
+  }
+
   GeospatialStatistics toParquetGeospatialStatistics(
       org.apache.parquet.column.statistics.geospatial.GeospatialStatistics geospatialStatistics) {
     if (geospatialStatistics == null) {
@@ -1837,6 +1891,28 @@ public ColumnChunkMetaData buildColumnChunkMetaData(
         fromParquetStatistics(metaData.geospatial_statistics, type));
   }
 
+  public ColumnChunkMetaData buildColumnChunkMetaData(
+      ColumnMetaData metaData,
+      ColumnPath columnPath,
+      PrimitiveType type,
+      ParsedVersion writerVersion,
+      String createdBy) {
+    return ColumnChunkMetaData.get(
+        columnPath,
+        type,
+        fromFormatCodec(metaData.codec),
+        convertEncodingStats(metaData.getEncoding_stats()),
+        fromFormatEncodings(metaData.encodings),
+        fromParquetStatistics(writerVersion, createdBy, metaData.statistics, type),
+        metaData.data_page_offset,
+        metaData.dictionary_page_offset,
+        metaData.num_values,
+        metaData.total_compressed_size,
+        metaData.total_uncompressed_size,
+        fromParquetSizeStatistics(metaData.size_statistics, type),
+        fromParquetStatistics(metaData.geospatial_statistics, type));
+  }
+
   public ParquetMetadata fromParquetMetadata(FileMetaData parquetMetadata) throws IOException {
     return fromParquetMetadata(parquetMetadata, null, false);
   }
@@ -1854,6 +1930,17 @@ public ParquetMetadata fromParquetMetadata(
       Map<RowGroup, Long> rowGroupToRowIndexOffsetMap)
       throws IOException {
     MessageType messageType = fromParquetSchema(parquetMetadata.getSchema(), parquetMetadata.getColumn_orders());
+    org.apache.parquet.hadoop.metadata.FileMetaData fileMetaData =
+        buildFileMetaData(parquetMetadata, messageType, encryptedFooter, fileDecryptor);
+    String createdBy = fileMetaData.getCreatedBy();
+    ParsedVersion writerVersion = null;
+    boolean useWriterVersion = false;
+    try {
+      writerVersion = fileMetaData.getWriterVersion();
+      useWriterVersion = true;
+    } catch (VersionParseException e) {
+      // Fall back to String-based path which logs the parse error with full context
+    }
     List<BlockMetaData> blocks = new ArrayList<BlockMetaData>();
     List<RowGroup> row_groups = parquetMetadata.getRow_groups();
 
@@ -1930,13 +2017,13 @@ public ParquetMetadata fromParquetMetadata(
             }
           }
 
-          String createdBy = parquetMetadata.getCreated_by();
           if (!lazyMetadataDecryption) { // full column metadata (with stats) is available
-            column = buildColumnChunkMetaData(
-                metaData,
-                columnPath,
-                messageType.getType(columnPath.toArray()).asPrimitiveType(),
-                createdBy);
+            PrimitiveType primitiveType =
+                messageType.getType(columnPath.toArray()).asPrimitiveType();
+            column = useWriterVersion
+                ? buildColumnChunkMetaData(
+                    metaData, columnPath, primitiveType, writerVersion, createdBy)
+                : buildColumnChunkMetaData(metaData, columnPath, primitiveType, createdBy);
             column.setRowGroupOrdinal(rowGroup.getOrdinal());
             if (metaData.isSetBloom_filter_offset()) {
               column.setBloomFilterOffset(metaData.getBloom_filter_offset());
@@ -1975,6 +2062,15 @@ public ParquetMetadata fromParquetMetadata(
         blocks.add(blockMetaData);
       }
     }
+    return new ParquetMetadata(fileMetaData, blocks);
+  }
+
+  private static org.apache.parquet.hadoop.metadata.FileMetaData buildFileMetaData(
+      FileMetaData parquetMetadata,
+      MessageType messageType,
+      boolean encryptedFooter,
+      InternalFileDecryptor fileDecryptor) {
+    String createdBy = parquetMetadata.getCreated_by();
     Map<String, String> keyValueMetaData = new HashMap<String, String>();
     List<KeyValue> key_value_metadata = parquetMetadata.getKey_value_metadata();
     if (key_value_metadata != null) {
@@ -1990,10 +2086,8 @@ public ParquetMetadata fromParquetMetadata(
     } else {
       encryptionType = EncryptionType.UNENCRYPTED;
     }
-    return new ParquetMetadata(
-        new org.apache.parquet.hadoop.metadata.FileMetaData(
-            messageType, keyValueMetaData, parquetMetadata.getCreated_by(), encryptionType, fileDecryptor),
-        blocks);
+    return new org.apache.parquet.hadoop.metadata.FileMetaData(
+        messageType, keyValueMetaData, createdBy, encryptionType, fileDecryptor);
   }
 
   private static IndexReference toColumnIndexReference(ColumnChunk columnChunk) {