PARQUET-686: Address review comments.

Author: rdblue

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

@@ -85,11 +85,15 @@ public class ParquetMetadataConverter {
   private final boolean useSignedStringMinMax;
 
   public ParquetMetadataConverter() {
-    this.useSignedStringMinMax = false;
+    this(false);
   }
 
   public ParquetMetadataConverter(Configuration conf) {
-    this.useSignedStringMinMax = conf.getBoolean("parquet.strings.use-signed-order", false);
+    this(conf.getBoolean("parquet.strings.signed-min-max.enabled", false));
+  }
+
+  private ParquetMetadataConverter(boolean useSignedStringMinMax) {
+    this.useSignedStringMinMax = useSignedStringMinMax;
   }
 
   // NOTE: this cache is for memory savings, not cpu savings, and is used to de-duplicate
@@ -330,14 +334,14 @@ public static org.apache.parquet.column.statistics.Statistics fromParquetStatist
 
   // Visible for testing
   static org.apache.parquet.column.statistics.Statistics fromParquetStatisticsInternal
-      (String createdBy, Statistics statistics, PrimitiveTypeName type, SortOrder expectedOrder) {
+      (String createdBy, Statistics statistics, PrimitiveTypeName type, SortOrder typeSortOrder) {
     // create stats object based on the column type
     org.apache.parquet.column.statistics.Statistics stats = org.apache.parquet.column.statistics.Statistics.getStatsBasedOnType(type);
     // If there was no statistics written to the footer, create an empty Statistics object and return
 
     // NOTE: See docs in CorruptStatistics for explanation of why this check is needed
     if (statistics != null && !CorruptStatistics.shouldIgnoreStatistics(createdBy, type) &&
-        SortOrder.SIGNED == expectedOrder) {
+        SortOrder.SIGNED == typeSortOrder) {
       if (statistics.isSetMax() && statistics.isSetMin()) {
         stats.setMinMaxFromBytes(statistics.min.array(), statistics.max.array());
       }
@@ -348,7 +352,8 @@ public static org.apache.parquet.column.statistics.Statistics fromParquetStatist
 
   public org.apache.parquet.column.statistics.Statistics fromParquetStatistics(
       String createdBy, Statistics statistics, PrimitiveType type) {
-    SortOrder expectedOrder = isSignedOrderOkay(type) ? SortOrder.SIGNED : sortOrder(type);
+    SortOrder expectedOrder = overrideSortOrderToSigned(type) ?
+        SortOrder.SIGNED : sortOrder(type);
     return fromParquetStatisticsInternal(
         createdBy, statistics, type.getPrimitiveTypeName(), expectedOrder);
   }
@@ -359,45 +364,55 @@ enum SortOrder {
     UNKNOWN
   }
 
-  private boolean isSignedOrderOkay(PrimitiveType type) {
-    if (!useSignedStringMinMax) {
-      return false;
-    }
+  private static final Set<OriginalType> STRING_TYPES = Collections
+      .unmodifiableSet(new HashSet<>(Arrays.asList(
+          OriginalType.UTF8, OriginalType.ENUM, OriginalType.JSON
+      )));
 
+  /**
+   * Returns whether to use signed order min and max with a type. It is safe to
+   * use signed min and max when the type is a string type and contains only
+   * ASCII characters (where the sign bit was 0). This checks whether the type
+   * is a string type and uses {@code useSignedStringMinMax} to determine if
+   * only ASCII characters were written.
+   *
+   * @param type a primitive type with a logical type annotation
+   * @return true if signed order min/max can be used with this type
+   */
+  private boolean overrideSortOrderToSigned(PrimitiveType type) {
     // even if the override is set, only return stats for string-ish types
-    if (type.getPrimitiveTypeName() != PrimitiveTypeName.BINARY) {
-      return false;
-    }
-    if (type.getOriginalType() == null) {
-      return true; // plain binary is okay
-    }
-    switch (type.getOriginalType()) {
-      case UTF8:
-      case ENUM:
-      case BSON:
-      case JSON:
-        return true;
-      default: // includes decimal
-        return false;
-    }
+    // a null type annotation is considered string-ish because some writers
+    // failed to use the UTF8 annotation.
+    OriginalType annotation = type.getOriginalType();
+    return useSignedStringMinMax &&
+        PrimitiveTypeName.BINARY == type.getPrimitiveTypeName() &&
+        (annotation == null || STRING_TYPES.contains(annotation));
   }
 
+  /**
+   * @param primitive a primitive physical type
+   * @return the default sort order used when the logical type is not known
+   */
   private static SortOrder defaultSortOrder(PrimitiveTypeName primitive) {
     switch (primitive) {
       case BOOLEAN:
       case INT32:
       case INT64:
       case FLOAT:
       case DOUBLE:
-      case BINARY: // without a logical type, signed is okay
-      case FIXED_LEN_BYTE_ARRAY:
         return SortOrder.SIGNED;
+      case BINARY:
+      case FIXED_LEN_BYTE_ARRAY:
       case INT96: // only used for timestamp, which uses unsigned values
         return SortOrder.UNSIGNED;
     }
     return SortOrder.UNKNOWN;
   }
 
+  /**
+   * @param primitive a primitive type with a logical type annotation
+   * @return the "correct" sort order of the type that applications assume
+   */
   private static SortOrder sortOrder(PrimitiveType primitive) {
     OriginalType annotation = primitive.getOriginalType();
     if (annotation != null) {
@@ -416,12 +431,12 @@ private static SortOrder sortOrder(PrimitiveType primitive) {
         case UINT_16:
         case UINT_32:
         case UINT_64:
-        case DECIMAL:
         case ENUM:
         case UTF8:
         case BSON:
         case JSON:
           return SortOrder.UNSIGNED;
+        case DECIMAL:
         case LIST:
         case MAP:
         case MAP_KEY_VALUE:

parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java

@@ -541,7 +541,7 @@ public void testIgnoreStatsWithSignedSortOrder() {
   public void testUseStatsWithSignedSortOrder() {
     // override defaults and use stats that were accumulated using signed order
     Configuration conf = new Configuration();
-    conf.set("parquet.strings.use-signed-order", "true");
+    conf.setBoolean("parquet.strings.signed-min-max.enabled", true);
 
     ParquetMetadataConverter converter = new ParquetMetadataConverter(conf);
     BinaryStatistics stats = new BinaryStatistics();
@@ -554,7 +554,8 @@ public void testUseStatsWithSignedSortOrder() {
     Statistics convertedStats = converter.fromParquetStatistics(
         Version.FULL_VERSION,
         ParquetMetadataConverter.toParquetStatistics(stats),
-        Types.required(PrimitiveTypeName.BINARY).named("b"));
+        Types.required(PrimitiveTypeName.BINARY)
+            .as(OriginalType.UTF8).named("b"));
 
     Assert.assertFalse("Stats should not be empty", convertedStats.isEmpty());
     Assert.assertEquals("Should have 3 nulls", 3, convertedStats.getNumNulls());

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java

@@ -24,9 +24,7 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.parquet.CorruptStatistics;
 import org.apache.parquet.Version;
-import org.apache.parquet.VersionParser;
 import org.apache.parquet.bytes.BytesUtils;
 import org.apache.parquet.hadoop.ParquetOutputFormat.JobSummaryLevel;
 import org.junit.Assume;
@@ -452,9 +450,9 @@ public void testWriteReadStatistics() throws Exception {
 
     Path path = new Path(testFile.toURI());
     Configuration configuration = new Configuration();
-    configuration.set("parquet.strings.use-signed-order", "true");
+    configuration.setBoolean("parquet.strings.signed-min-max.enabled", true);
 
-    MessageType schema = MessageTypeParser.parseMessageType("message m { required group a {required binary b;} required group c { required int64 d; }}");
+    MessageType schema = MessageTypeParser.parseMessageType("message m { required group a {required binary b (UTF8);} required group c { required int64 d; }}");
     String[] path1 = {"a", "b"};
     ColumnDescriptor c1 = schema.getColumnDescription(path1);
     String[] path2 = {"c", "d"};
@@ -585,14 +583,14 @@ public void testWriteReadStatisticsAllNulls() throws Exception {
     testFile.delete();
 
     writeSchema = "message example {\n" +
-            "required binary content;\n" +
+            "required binary content (UTF8);\n" +
             "}";
 
     Path path = new Path(testFile.toURI());
 
     MessageType schema = MessageTypeParser.parseMessageType(writeSchema);
     Configuration configuration = new Configuration();
-    configuration.set("parquet.strings.use-signed-order", "true");
+    configuration.setBoolean("parquet.strings.signed-min-max.enabled", true);
     GroupWriteSupport.setSchema(schema, configuration);
 
     ParquetWriter<Group> writer = new ParquetWriter<Group>(path, configuration, new GroupWriteSupport());

parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestThriftToParquetFileWriter.java

@@ -114,7 +114,7 @@ public void testWriteStatistics() throws Exception {
                           new RequiredPrimitiveFixture(false, (byte)100, (short)100, 100, 287l, -9.0d, "world"),
                           new RequiredPrimitiveFixture(true, (byte)2, (short)2, 9, -17l, 9.63d, "hello"));
       final Configuration configuration = new Configuration();
-      configuration.set("parquet.strings.use-signed-order", "true");
+      configuration.setBoolean("parquet.strings.signed-min-max.enabled", true);
       final FileSystem fs = p.getFileSystem(configuration);
       FileStatus fileStatus = fs.getFileStatus(p);
       ParquetMetadata footer = ParquetFileReader.readFooter(configuration, p);
@@ -161,7 +161,7 @@ public void testWriteStatistics() throws Exception {
 
       // make new configuration and create file with new large stats
       final Configuration configuration_large = new Configuration();
-      configuration_large.set("parquet.strings.use-signed-order", "true");
+      configuration.setBoolean("parquet.strings.signed-min-max.enabled", true);
       final FileSystem fs_large = p_large.getFileSystem(configuration_large);
       FileStatus fileStatus_large = fs_large.getFileStatus(p_large);
       ParquetMetadata footer_large = ParquetFileReader.readFooter(configuration_large, p_large);