PARQUET-2249: Write IEEE 754 total order by default for floating-point columns

parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java

@@ -225,6 +225,12 @@ private static void addPrimitiveType(
       check(t, ")", "logical type ended by )", st);
       t = st.nextToken();
     }
+    if (t.equalsIgnoreCase(PrimitiveType.COLUMN_ORDER_KEYWORD)) {
+      check(st.nextToken(), "(", "column order followed by (", st);
+      childBuilder.columnOrder(parseColumnOrder(st.nextToken(), st));
+      check(st.nextToken(), ")", "column order ended by )", st);
+      t = st.nextToken();
+    }
     if (t.equals("=")) {
       childBuilder.id(Integer.parseInt(st.nextToken()));
       t = st.nextToken();
@@ -240,6 +246,26 @@ private static void addPrimitiveType(
     }
   }
 
+  private static ColumnOrder parseColumnOrder(String t, Tokenizer st) {
+    ColumnOrder.ColumnOrderName name;
+    try {
+      name = ColumnOrder.ColumnOrderName.valueOf(t.toUpperCase(Locale.ENGLISH));
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException("Unknown column order: " + t + " at " + st.getLocationString(), e);
+    }
+    switch (name) {
+      case UNDEFINED:
+        return ColumnOrder.undefined();
+      case TYPE_DEFINED_ORDER:
+        return ColumnOrder.typeDefined();
+      case IEEE_754_TOTAL_ORDER:
+        return ColumnOrder.ieee754TotalOrder();
+      default:
+        throw new IllegalArgumentException(
+            "Unsupported column order: " + name + " at " + st.getLocationString());
+    }
+  }
+
   private static boolean isLogicalType(String t) {
     return Arrays.stream(LogicalTypeAnnotation.LogicalTypeToken.values())
         .anyMatch((type) -> type.name().equals(t));

parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java

@@ -492,6 +492,9 @@ public PrimitiveComparator<?> comparator(LogicalTypeAnnotation logicalType) {
     }
   }
 
+  // Keyword used to render/parse a non-default column order in the text schema representation.
+  static final String COLUMN_ORDER_KEYWORD = "columnorder";
+
   private final PrimitiveTypeName primitive;
   private final int length;
   private final DecimalMetadata decimalMeta;
@@ -578,9 +581,7 @@ public PrimitiveType(
     this.decimalMeta = decimalMeta;
 
     if (columnOrder == null) {
-      columnOrder = primitive == PrimitiveTypeName.INT96 || originalType == OriginalType.INTERVAL
-          ? ColumnOrder.undefined()
-          : ColumnOrder.typeDefined();
+      columnOrder = defaultColumnOrder(primitive, originalType, getLogicalTypeAnnotation());
     } else if (columnOrder.getColumnOrderName() == ColumnOrderName.IEEE_754_TOTAL_ORDER) {
       Preconditions.checkArgument(
           primitive == PrimitiveTypeName.FLOAT || primitive == PrimitiveTypeName.DOUBLE,
@@ -629,10 +630,7 @@ public PrimitiveType(
     }
 
     if (columnOrder == null) {
-      columnOrder = primitive == PrimitiveTypeName.INT96
-              || logicalTypeAnnotation instanceof LogicalTypeAnnotation.IntervalLogicalTypeAnnotation
-          ? ColumnOrder.undefined()
-          : ColumnOrder.typeDefined();
+      columnOrder = defaultColumnOrder(primitive, getOriginalType(), logicalTypeAnnotation);
     } else if (columnOrder.getColumnOrderName() == ColumnOrderName.IEEE_754_TOTAL_ORDER) {
       Preconditions.checkArgument(
           primitive == PrimitiveTypeName.FLOAT
@@ -648,6 +646,27 @@ public PrimitiveType(
     this.columnOrder = requireValidColumnOrder(columnOrder);
   }
 
+  /**
+   * The column order used when none is specified explicitly. INT96 and INTERVAL have no defined
+   * ordering, so they default to undefined. Floating-point types default to IEEE 754 total order so
+   * that NaN values and the sign of zero are ordered deterministically and nan_count statistics can
+   * be written; this is skipped when the logical type annotation does not accept IEEE 754 total
+   * order (e.g. an unknown annotation), leaving the type constructible with the type-defined order.
+   */
+  private static ColumnOrder defaultColumnOrder(
+      PrimitiveTypeName primitive, OriginalType originalType, LogicalTypeAnnotation logicalTypeAnnotation) {
+    if (primitive == PrimitiveTypeName.INT96 || originalType == OriginalType.INTERVAL) {
+      return ColumnOrder.undefined();
+    }
+    boolean isFloatingType = primitive == PrimitiveTypeName.FLOAT
+        || primitive == PrimitiveTypeName.DOUBLE
+        || (logicalTypeAnnotation != null
+            && logicalTypeAnnotation.getType() == LogicalTypeAnnotation.LogicalTypeToken.FLOAT16);
+    boolean acceptsIeee754 = logicalTypeAnnotation == null
+        || logicalTypeAnnotation.isValidColumnOrder(ColumnOrder.ieee754TotalOrder());
+    return isFloatingType && acceptsIeee754 ? ColumnOrder.ieee754TotalOrder() : ColumnOrder.typeDefined();
+  }
+
   private ColumnOrder requireValidColumnOrder(ColumnOrder columnOrder) {
     if (primitive == PrimitiveTypeName.INT96) {
       Preconditions.checkArgument(
@@ -748,6 +767,13 @@ public void writeToStringBuilder(StringBuilder sb, String indent) {
       // TODO: should we print decimal metadata too?
       sb.append(" (").append(getLogicalTypeAnnotation().toString()).append(")");
     }
+    // Only emit the column order when it differs from the default, so schemas that rely on the
+    // default stay textually unchanged.
+    if (!columnOrder.equals(defaultColumnOrder(primitive, getOriginalType(), getLogicalTypeAnnotation()))) {
+      sb.append(" ").append(COLUMN_ORDER_KEYWORD).append("(");
+      sb.append(columnOrder.getColumnOrderName().name());
+      sb.append(")");
+    }
     if (getId() != null) {
       sb.append(" = ").append(getId());
     }

parquet-column/src/main/java/org/apache/parquet/schema/Types.java

@@ -414,9 +414,11 @@ public THIS scale(int scale) {
     /**
      * Adds the column order for the primitive type.
      * <p>
-     * In case of not set the default column order is {@link ColumnOrderName#TYPE_DEFINED_ORDER} except the type
-     * {@link PrimitiveTypeName#INT96} and the types annotated by {@link OriginalType#INTERVAL} where the default column
-     * order is {@link ColumnOrderName#UNDEFINED}.
+     * In case of not set the default column order is {@link ColumnOrderName#TYPE_DEFINED_ORDER}, with the following
+     * exceptions: the floating-point types {@link PrimitiveTypeName#FLOAT}, {@link PrimitiveTypeName#DOUBLE} and the
+     * {@code FLOAT16} logical type default to {@link ColumnOrderName#IEEE_754_TOTAL_ORDER}; the type
+     * {@link PrimitiveTypeName#INT96} and the types annotated by {@link OriginalType#INTERVAL} default to
+     * {@link ColumnOrderName#UNDEFINED}.
      *
      * @param columnOrder the column order for the primitive type
      * @return this builder for method chaining

parquet-column/src/test/java/org/apache/parquet/column/statistics/TestStatistics.java

@@ -38,6 +38,7 @@
 import java.nio.ByteBuffer;
 import java.util.Locale;
 import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.ColumnOrder;
 import org.apache.parquet.schema.OriginalType;
 import org.apache.parquet.schema.PrimitiveType;
 import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
@@ -775,7 +776,8 @@ private void testBuilder(PrimitiveType type, Object min, byte[] minBytes, Object
 
   @Test
   public void testSpecBuilderForFloat() {
-    PrimitiveType type = Types.required(FLOAT).named("test_float");
+    PrimitiveType type =
+        Types.required(FLOAT).columnOrder(ColumnOrder.typeDefined()).named("test_float");
     Statistics.Builder builder = Statistics.getBuilderForReading(type);
     Statistics<?> stats = builder.withMin(intToBytes(floatToIntBits(Float.NaN)))
         .withMax(intToBytes(floatToIntBits(42.0f)))
@@ -839,7 +841,8 @@ public void testSpecBuilderForFloat() {
 
   @Test
   public void testSpecBuilderForDouble() {
-    PrimitiveType type = Types.required(DOUBLE).named("test_double");
+    PrimitiveType type =
+        Types.required(DOUBLE).columnOrder(ColumnOrder.typeDefined()).named("test_double");
     Statistics.Builder builder = Statistics.getBuilderForReading(type);
     Statistics<?> stats = builder.withMin(longToBytes(doubleToLongBits(Double.NaN)))
         .withMax(longToBytes(doubleToLongBits(42.0)))

parquet-column/src/test/java/org/apache/parquet/column/statistics/TestStatisticsNanCount.java

@@ -32,13 +32,16 @@
 
 public class TestStatisticsNanCount {
 
-  private static final PrimitiveType FLOAT_TYPE =
-      Types.optional(PrimitiveTypeName.FLOAT).named("test_float");
-  private static final PrimitiveType DOUBLE_TYPE =
-      Types.optional(PrimitiveTypeName.DOUBLE).named("test_double");
+  private static final PrimitiveType FLOAT_TYPE = Types.optional(PrimitiveTypeName.FLOAT)
+      .columnOrder(ColumnOrder.typeDefined())
+      .named("test_float");
+  private static final PrimitiveType DOUBLE_TYPE = Types.optional(PrimitiveTypeName.DOUBLE)
+      .columnOrder(ColumnOrder.typeDefined())
+      .named("test_double");
   private static final PrimitiveType FLOAT16_TYPE = Types.optional(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
       .length(2)
       .as(LogicalTypeAnnotation.float16Type())
+      .columnOrder(ColumnOrder.typeDefined())
       .named("test_float16");
 
   private static final PrimitiveType FLOAT_IEEE754_TYPE = Types.optional(PrimitiveTypeName.FLOAT)

parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestColumnIndexBuilder.java

@@ -69,6 +69,7 @@
 import org.apache.parquet.filter2.predicate.Operators.LongColumn;
 import org.apache.parquet.filter2.predicate.UserDefinedPredicate;
 import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.ColumnOrder;
 import org.apache.parquet.schema.PrimitiveType;
 import org.apache.parquet.schema.Types;
 import org.junit.jupiter.api.Test;
@@ -1030,7 +1031,8 @@ public void testBuildDouble() {
 
   @Test
   public void testBuildDoubleZeroNaN() {
-    PrimitiveType type = Types.required(DOUBLE).named("test_double");
+    PrimitiveType type =
+        Types.required(DOUBLE).columnOrder(ColumnOrder.typeDefined()).named("test_double");
     ColumnIndexBuilder builder = ColumnIndexBuilder.getBuilder(type, Integer.MAX_VALUE);
     StatsBuilder sb = new StatsBuilder();
     builder.add(sb.stats(type, -1.0, -0.0));
@@ -1183,7 +1185,8 @@ public void testBuildFloat() {
 
   @Test
   public void testBuildFloatZeroNaN() {
-    PrimitiveType type = Types.required(FLOAT).named("test_float");
+    PrimitiveType type =
+        Types.required(FLOAT).columnOrder(ColumnOrder.typeDefined()).named("test_float");
     ColumnIndexBuilder builder = ColumnIndexBuilder.getBuilder(type, Integer.MAX_VALUE);
     StatsBuilder sb = new StatsBuilder();
     builder.add(sb.stats(type, -1.0f, -0.0f));

parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestColumnIndexBuilderNaN.java

@@ -44,19 +44,22 @@
  */
 public class TestColumnIndexBuilderNaN {
 
-  private static final PrimitiveType FLOAT_TYPE =
-      Types.required(PrimitiveTypeName.FLOAT).named("test_float");
+  private static final PrimitiveType FLOAT_TYPE = Types.required(PrimitiveTypeName.FLOAT)
+      .columnOrder(ColumnOrder.typeDefined())
+      .named("test_float");
   private static final PrimitiveType FLOAT_IEEE754_TYPE = Types.required(PrimitiveTypeName.FLOAT)
       .columnOrder(ColumnOrder.ieee754TotalOrder())
       .named("test_float_ieee754");
-  private static final PrimitiveType DOUBLE_TYPE =
-      Types.required(PrimitiveTypeName.DOUBLE).named("test_double");
+  private static final PrimitiveType DOUBLE_TYPE = Types.required(PrimitiveTypeName.DOUBLE)
+      .columnOrder(ColumnOrder.typeDefined())
+      .named("test_double");
   private static final PrimitiveType DOUBLE_IEEE754_TYPE = Types.required(PrimitiveTypeName.DOUBLE)
       .columnOrder(ColumnOrder.ieee754TotalOrder())
       .named("test_double_ieee754");
   private static final PrimitiveType FLOAT16_TYPE = Types.required(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
       .length(2)
       .as(LogicalTypeAnnotation.float16Type())
+      .columnOrder(ColumnOrder.typeDefined())
       .named("test_float16");
   private static final PrimitiveType FLOAT16_IEEE754_TYPE = Types.required(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
       .length(2)

parquet-column/src/test/java/org/apache/parquet/schema/TestMessageType.java

@@ -202,7 +202,37 @@ public void testMergeSchemaWithColumnOrder() {
     assertThatThrownBy(() -> m1.union(m3))
         .isInstanceOf(IncompatibleSchemaModificationException.class)
         .hasMessage(
-            "can not merge type optional binary a with column order TYPE_DEFINED_ORDER into optional binary a with column order UNDEFINED");
+            "can not merge type optional binary a with column order TYPE_DEFINED_ORDER into optional binary a columnorder(UNDEFINED) with column order UNDEFINED");
+  }
+
+  @Test
+  public void testColumnOrderTextRoundTrip() {
+    // A non-default column order must survive toString() -> parseMessageType() so that schemas
+    // serialized through the text representation (e.g. by GroupWriteSupport) keep it.
+    MessageType schema = Types.buildMessage()
+        .required(PrimitiveTypeName.FLOAT)
+        .columnOrder(ColumnOrder.typeDefined())
+        .named("float_typedef")
+        .required(PrimitiveTypeName.DOUBLE)
+        .columnOrder(ColumnOrder.ieee754TotalOrder())
+        .named("double_ieee754")
+        .required(PrimitiveTypeName.INT32)
+        .named("int_default")
+        .named("msg");
+
+    assertThat(schema.getType("float_typedef").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.typeDefined());
+    MessageType roundTripped = MessageTypeParser.parseMessageType(schema.toString());
+    assertThat(roundTripped).isEqualTo(schema);
+    assertThat(roundTripped.getType("float_typedef").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.typeDefined());
+    assertThat(roundTripped.getType("double_ieee754").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.ieee754TotalOrder());
+    assertThat(roundTripped.getType("int_default").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.typeDefined());
+
+    // A column left at its default emits no columnorder(...) token.
+    assertThat(schema.toString()).doesNotContain("int_default columnorder");
   }
 
   @Test

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

@@ -2061,6 +2061,13 @@ private void buildChildren(
             columnOrder = org.apache.parquet.schema.ColumnOrder.undefined();
           }
           primitiveBuilder.columnOrder(columnOrder);
+        } else if (schemaElement.type == Type.FLOAT
+            || schemaElement.type == Type.DOUBLE
+            || (schemaElement.isSetLogicalType() && schemaElement.logicalType.isSetFLOAT16())) {
+          // A footer without column orders predates IEEE_754_TOTAL_ORDER, so a floating-point column
+          // here must not inherit the (IEEE 754 total order) construction-time default: its stats, if
+          // any, were written under the legacy type-defined order and must be read under it.
+          primitiveBuilder.columnOrder(org.apache.parquet.schema.ColumnOrder.typeDefined());
         }
         childBuilder = primitiveBuilder;
       } else {

parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java

@@ -706,15 +706,20 @@ private static List<ColumnChunkMetaData> nanColumns() {
     return List.of(
         nanColumn(
             "double_nan_field",
-            Types.required(PrimitiveTypeName.DOUBLE).named("double_nan_field")),
+            Types.required(PrimitiveTypeName.DOUBLE)
+                .columnOrder(ColumnOrder.typeDefined())
+                .named("double_nan_field")),
         nanColumn(
             "float_nan_field",
-            Types.required(PrimitiveTypeName.FLOAT).named("float_nan_field")),
+            Types.required(PrimitiveTypeName.FLOAT)
+                .columnOrder(ColumnOrder.typeDefined())
+                .named("float_nan_field")),
         nanColumn(
             "float16_nan_field",
             Types.required(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
                 .length(2)
                 .as(LogicalTypeAnnotation.float16Type())
+                .columnOrder(ColumnOrder.typeDefined())
                 .named("float16_nan_field")));
   }
 

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

@@ -2133,6 +2133,72 @@ public void testIEEE754TotalOrderColumnOrder() throws IOException {
         .isEqualTo(ColumnOrder.ieee754TotalOrder());
   }
 
+  @Test
+  public void testFloatingPointColumnsDefaultToIeee754TotalOrder() throws IOException {
+    MessageType schema = parseMessageType("message test {"
+        + "  required float float_col;"
+        + "  required double double_col;"
+        + "  required fixed_len_byte_array(2) float16_col (FLOAT16);"
+        + "  required int32 int_col;"
+        + "}");
+
+    org.apache.parquet.hadoop.metadata.FileMetaData fileMetaData =
+        new org.apache.parquet.hadoop.metadata.FileMetaData(schema, new HashMap<String, String>(), null);
+    ParquetMetadata metadata = new ParquetMetadata(fileMetaData, new ArrayList<BlockMetaData>());
+    ParquetMetadataConverter converter = new ParquetMetadataConverter();
+    FileMetaData formatMetadata = converter.toParquetMetadata(1, metadata);
+
+    // Floating-point columns serialize the new order; the int column keeps type-defined order.
+    List<org.apache.parquet.format.ColumnOrder> columnOrders = formatMetadata.getColumn_orders();
+    assertThat(columnOrders).hasSize(4);
+    assertThat(columnOrders.get(0).isSetIEEE_754_TOTAL_ORDER()).isTrue();
+    assertThat(columnOrders.get(1).isSetIEEE_754_TOTAL_ORDER()).isTrue();
+    assertThat(columnOrders.get(2).isSetIEEE_754_TOTAL_ORDER()).isTrue();
+    assertThat(columnOrders.get(3).isSetTYPE_ORDER()).isTrue();
+
+    MessageType resultSchema =
+        converter.fromParquetMetadata(formatMetadata).getFileMetaData().getSchema();
+    assertThat(resultSchema.getType("float_col").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.ieee754TotalOrder());
+    assertThat(resultSchema.getType("double_col").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.ieee754TotalOrder());
+    assertThat(resultSchema.getType("float16_col").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.ieee754TotalOrder());
+    assertThat(resultSchema.getType("int_col").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.typeDefined());
+  }
+
+  @Test
+  public void testUndefinedFloatingPointColumnOrderReadsAsTypeDefined() throws IOException {
+    // A footer without column orders predates IEEE_754_TOTAL_ORDER: floating-point columns must be
+    // read back as type-defined order (the pre-existing default) so that legacy stats are not
+    // reinterpreted under IEEE 754 total order.
+    MessageType schema = parseMessageType("message test {"
+        + "  required float float_col;"
+        + "  required double double_col;"
+        + "  required fixed_len_byte_array(2) float16_col (FLOAT16);"
+        + "}");
+
+    org.apache.parquet.hadoop.metadata.FileMetaData fileMetaData =
+        new org.apache.parquet.hadoop.metadata.FileMetaData(schema, new HashMap<String, String>(), null);
+    ParquetMetadata metadata = new ParquetMetadata(fileMetaData, new ArrayList<BlockMetaData>());
+    ParquetMetadataConverter converter = new ParquetMetadataConverter();
+    FileMetaData formatMetadata = converter.toParquetMetadata(1, metadata);
+
+    // Simulate a legacy footer that carries no column orders at all.
+    formatMetadata.unsetColumn_orders();
+    assertThat(formatMetadata.isSetColumn_orders()).isFalse();
+
+    MessageType resultSchema =
+        converter.fromParquetMetadata(formatMetadata).getFileMetaData().getSchema();
+    assertThat(resultSchema.getType("float_col").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.typeDefined());
+    assertThat(resultSchema.getType("double_col").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.typeDefined());
+    assertThat(resultSchema.getType("float16_col").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.typeDefined());
+  }
+
   @Test
   public void testNestedColumnOrdersUseLeafOrder() throws IOException {
     MessageType schema = Types.buildMessage()
@@ -2159,7 +2225,9 @@ public void testNestedColumnOrdersUseLeafOrder() throws IOException {
     List<ColumnDescriptor> columns = resultSchema.getColumns();
     assertThat(columns).hasSize(3);
     assertThat(columns.get(0).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.ieee754TotalOrder());
-    assertThat(columns.get(1).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.typeDefined());
+    // Column "b" is a DOUBLE built without an explicit column order, so it picks up the
+    // floating-point default of IEEE 754 total order.
+    assertThat(columns.get(1).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.ieee754TotalOrder());
     assertThat(columns.get(2).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.ieee754TotalOrder());
   }