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()));
+      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,20 @@ private static void addPrimitiveType(
     }
   }
 
+  private static ColumnOrder parseColumnOrder(String t) {
+    // An unrecognized order degrades to UNDEFINED rather than failing, matching
+    // ParquetMetadataConverter.fromParquetColumnOrder, so a schema written by a newer API with an
+    // order this version does not know stays parseable. Statistics under an unknown order are
+    // ignored by readers anyway.
+    if (t.equalsIgnoreCase(ColumnOrder.ColumnOrderName.TYPE_DEFINED_ORDER.name())) {
+      return ColumnOrder.typeDefined();
+    }
+    if (t.equalsIgnoreCase(ColumnOrder.ColumnOrderName.IEEE_754_TOTAL_ORDER.name())) {
+      return ColumnOrder.ieee754TotalOrder();
+    }
+    return ColumnOrder.undefined();
+  }
+
   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());
     }
@@ -857,17 +883,13 @@ private void reportSchemaMergeError(Type toMerge) {
     throw new IncompatibleSchemaModificationException("can not merge type " + toMerge + " into " + this);
   }
 
-  private void reportSchemaMergeErrorWithColumnOrder(Type toMerge) {
-    throw new IncompatibleSchemaModificationException("can not merge type " + toMerge + " with column order "
-        + toMerge.asPrimitiveType().columnOrder() + " into " + this + " with column order " + columnOrder());
-  }
-
   @Override
   protected Type union(Type toMerge, boolean strict) {
     if (!toMerge.isPrimitive()) {
       reportSchemaMergeError(toMerge);
     }
 
+    ColumnOrder mergedColumnOrder = columnOrder();
     if (strict) {
       // Can't merge primitive fields of different type names or different original types
       if (!primitive.equals(toMerge.asPrimitiveType().getPrimitiveTypeName())
@@ -881,9 +903,14 @@ protected Type union(Type toMerge, boolean strict) {
         reportSchemaMergeError(toMerge);
       }
 
-      // Can't merge primitive fields with different column orders
+      // A column-order difference is the only remaining difference here (type, logical type and
+      // length already match). Reconcile to UNDEFINED instead of failing the merge: it lets an
+      // aggregation over otherwise-identical files with different column orders succeed -- e.g. a
+      // pre-upgrade float footer read as TYPE_DEFINED_ORDER merged with a post-upgrade one written
+      // as IEEE_754_TOTAL_ORDER. Per-file statistics are still read under each file's own column
+      // order, so this only drops the merged schema's (now ambiguous) ordering claim.
       if (!columnOrder().equals(toMerge.asPrimitiveType().columnOrder())) {
-        reportSchemaMergeErrorWithColumnOrder(toMerge);
+        mergedColumnOrder = ColumnOrder.undefined();
       }
     }
 
@@ -894,7 +921,9 @@ protected Type union(Type toMerge, boolean strict) {
       builder.length(length);
     }
 
-    return builder.as(getLogicalTypeAnnotation()).columnOrder(columnOrder()).named(getName());
+    return builder.as(getLogicalTypeAnnotation())
+        .columnOrder(mergedColumnOrder)
+        .named(getName());
   }
 
   /**

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

@@ -199,10 +199,76 @@ public void testMergeSchemaWithColumnOrder() {
                 Types.optional(INT96).named("b"),
                 Types.optional(BINARY).named("c"))
             .named("root"));
-    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");
+    // Merging columns that differ only in column order reconciles to UNDEFINED rather than failing:
+    // m1's "a" is UNDEFINED and m3's "a" is TYPE_DEFINED_ORDER, so the merged "a" stays UNDEFINED
+    // (schema equality includes column order, so equality with m1 asserts the reconciled order).
+    assertThat(m1.union(m3)).isEqualTo(m1);
+  }
+
+  @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
+  public void testUnknownColumnOrderParsesAsUndefined() {
+    // A column order this version does not recognize (e.g. written by a newer API) degrades to
+    // UNDEFINED rather than failing the whole parse.
+    MessageType schema =
+        MessageTypeParser.parseMessageType("message msg { required binary a columnorder(SOME_FUTURE_ORDER); }");
+    assertThat(schema.getType("a").asPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.undefined());
+  }
+
+  @Test
+  public void testMergeMixedFloatingColumnOrder() {
+    // A float column written post-upgrade (IEEE 754 total order) and the same column read from a
+    // legacy footer (type-defined) must merge rather than throw -- e.g. when aggregating footers
+    // over a directory that spans the upgrade. The reconciled order is UNDEFINED; per-file
+    // statistics are still read under each file's own column order.
+    MessageType newFile = Types.buildMessage()
+        .required(PrimitiveTypeName.FLOAT)
+        .columnOrder(ColumnOrder.ieee754TotalOrder())
+        .named("f")
+        .named("root");
+    MessageType legacyFile = Types.buildMessage()
+        .required(PrimitiveTypeName.FLOAT)
+        .columnOrder(ColumnOrder.typeDefined())
+        .named("f")
+        .named("root");
+
+    MessageType merged = newFile.union(legacyFile);
+    assertThat(merged.getType("f").asPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.undefined());
+    // Merge is symmetric.
+    assertThat(legacyFile.union(newFile).getType("f").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.undefined());
+    // Same order on both sides is preserved (no spurious downgrade to UNDEFINED).
+    assertThat(newFile.union(newFile).getType("f").asPrimitiveType().columnOrder())
+        .isEqualTo(ColumnOrder.ieee754TotalOrder());
   }
 
   @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")));
   }