GH-3949: Preserve schema properties when renaming columns

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

@@ -123,6 +123,17 @@ public GroupType(Repetition repetition, String name, OriginalType originalType,
     }
   }
 
+  GroupType(GroupType toCopy, String newName) {
+    super(toCopy, newName);
+    this.fields = toCopy.fields;
+    this.indexByName = toCopy.indexByName;
+  }
+
+  @Override
+  public GroupType withName(String newName) {
+    return new GroupType(this, newName);
+  }
+
   /**
    * @param id the field id
    * @return a new GroupType with the same fields and a new id

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

@@ -44,6 +44,15 @@ public MessageType(String name, List<Type> fields) {
     super(Repetition.REPEATED, name, fields);
   }
 
+  private MessageType(MessageType toCopy, String newName) {
+    super(toCopy, newName);
+  }
+
+  @Override
+  public MessageType withName(String newName) {
+    return new MessageType(this, newName);
+  }
+
   /**
    * {@inheritDoc}
    */

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

@@ -654,6 +654,19 @@ public PrimitiveType(
     this.columnOrder = requireValidColumnOrder(columnOrder);
   }
 
+  private PrimitiveType(PrimitiveType toCopy, String newName) {
+    super(toCopy, newName);
+    this.primitive = toCopy.primitive;
+    this.length = toCopy.length;
+    this.decimalMeta = toCopy.decimalMeta;
+    this.columnOrder = toCopy.columnOrder;
+  }
+
+  @Override
+  public PrimitiveType withName(String newName) {
+    return new PrimitiveType(this, newName);
+  }
+
   /**
    * The column order used when none is specified explicitly. A plain INT96 is the deprecated
    * timestamp encoding and defaults to INT96 timestamp order so its min/max are chronological and

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

@@ -187,6 +187,28 @@ public Type(String name, Repetition repetition, OriginalType originalType) {
     this.id = id;
   }
 
+  Type(Type toCopy, String newName) {
+    this.name = Objects.requireNonNull(newName, "name cannot be null");
+    this.repetition = toCopy.repetition;
+    this.logicalTypeAnnotation = toCopy.logicalTypeAnnotation;
+    this.id = toCopy.id;
+  }
+
+  /**
+   * Returns a copy with a new name and all other attributes unchanged.
+   * Subclasses must override this method to support renaming.
+   *
+   * @param newName the name of the copy
+   * @return a copy of this type with the given name
+   * @throws NullPointerException if newName is null
+   * @throws UnsupportedOperationException if this type does not support renaming
+   */
+  public Type withName(String newName) {
+    Objects.requireNonNull(newName, "name cannot be null");
+    throw new UnsupportedOperationException(
+        "Renaming is not supported by " + getClass().getName());
+  }
+
   /**
    * @param id an integer id
    * @return the same type with the id field set

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

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.schema;
+
+import static org.apache.parquet.schema.LogicalTypeAnnotation.decimalType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.listType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.stringType;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.apache.parquet.schema.Type.Repetition.REPEATED;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.stream.Stream;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class TestTypeWithName {
+
+  @ParameterizedTest(name = "{0}")
+  @MethodSource("primitiveTypes")
+  void renamingPrimitivePreservesAllOtherAttributes(
+      String description, PrimitiveType original, PrimitiveType expected) {
+    PrimitiveType renamed = original.withName("renamed");
+
+    assertThat(renamed).isEqualTo(expected);
+    assertThat(original.getName()).isEqualTo("original");
+  }
+
+  @SuppressWarnings("deprecation")
+  private static Stream<Arguments> primitiveTypes() {
+    return Stream.of(
+        Arguments.of(
+            "annotated fixed-width primitive with explicit column order and field ID",
+            Types.optional(FIXED_LEN_BYTE_ARRAY)
+                .length(4)
+                .as(decimalType(2, 8))
+                .columnOrder(ColumnOrder.undefined())
+                .id(17)
+                .named("original"),
+            Types.optional(FIXED_LEN_BYTE_ARRAY)
+                .length(4)
+                .as(decimalType(2, 8))
+                .columnOrder(ColumnOrder.undefined())
+                .id(17)
+                .named("renamed")),
+        Arguments.of(
+            "legacy decimal annotation without decimal metadata",
+            new PrimitiveType(OPTIONAL, BINARY, "original", OriginalType.DECIMAL),
+            new PrimitiveType(OPTIONAL, BINARY, "renamed", OriginalType.DECIMAL)),
+        Arguments.of(
+            "legacy decimal metadata without a logical annotation",
+            new PrimitiveType(
+                REPEATED,
+                FIXED_LEN_BYTE_ARRAY,
+                4,
+                "original",
+                null,
+                new DecimalMetadata(8, 2),
+                new Type.ID(18)),
+            new PrimitiveType(
+                REPEATED,
+                FIXED_LEN_BYTE_ARRAY,
+                4,
+                "renamed",
+                null,
+                new DecimalMetadata(8, 2),
+                new Type.ID(18))));
+  }
+
+  @Test
+  void renamingGroupPreservesAnnotationFieldIdAndChildren() {
+    PrimitiveType element = Types.repeated(BINARY).as(stringType()).id(22).named("element");
+    GroupType original =
+        Types.optionalGroup().as(listType()).id(21).addField(element).named("original");
+    GroupType expected =
+        Types.optionalGroup().as(listType()).id(21).addField(element).named("renamed");
+
+    GroupType renamed = original.withName("renamed");
+
+    assertThat(renamed).isEqualTo(expected);
+    assertThat(original.getName()).isEqualTo("original");
+  }
+
+  @Test
+  void renamingMessagePreservesRootSemantics() {
+    MessageType original = new MessageType("original", Types.optional(INT32).named("value"));
+    MessageType expected = new MessageType("renamed", Types.optional(INT32).named("value"));
+
+    MessageType renamed = original.withName("renamed");
+
+    assertThat(renamed).isEqualTo(expected);
+    assertThat(renamed.getMaxDefinitionLevel("value")).isEqualTo(1);
+    assertThat(renamed.getMaxRepetitionLevel("value")).isZero();
+    assertThat(original.getName()).isEqualTo("original");
+  }
+
+  @Test
+  void renamingRejectsNullName() {
+    Type type = Types.optional(INT32).named("original");
+
+    assertThatThrownBy(() -> type.withName(null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("name cannot be null");
+  }
+}

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java

@@ -274,19 +274,11 @@ private MessageType getSchema() {
   private MessageType getSchemaWithRenamedColumns(MessageType schema) {
     List<Type> fields = schema.getFields().stream()
         .map(type -> {
-          if (!renamedColumns.containsKey(type.getName())) {
+          String newName = renamedColumns.get(type.getName());
+          if (newName == null) {
             return type;
-          } else if (type.isPrimitive()) {
-            return new PrimitiveType(
-                type.getRepetition(),
-                type.asPrimitiveType().getPrimitiveTypeName(),
-                renamedColumns.get(type.getName()));
-          } else {
-            return new GroupType(
-                type.getRepetition(),
-                renamedColumns.get(type.getName()),
-                type.asGroupType().getFields());
           }
+          return type.withName(newName);
         })
         .collect(Collectors.toList());
     return new MessageType(schema.getName(), fields);
@@ -482,14 +474,11 @@ private ColumnPath normalizeFieldsInPath(ColumnPath path) {
   }
 
   private PrimitiveType normalizeNameInType(PrimitiveType type) {
-    if (renamedColumns.isEmpty()) {
+    String newName = renamedColumns.get(type.getName());
+    if (newName == null) {
       return type;
-    } else {
-      return new PrimitiveType(
-          type.getRepetition(),
-          type.asPrimitiveType().getPrimitiveTypeName(),
-          renamedColumns.getOrDefault(type.getName(), type.getName()));
     }
+    return type.withName(newName);
   }
 
   private void processBlock(

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/RewriteOptions.java

@@ -445,6 +445,7 @@ public Builder mask(Map<String, MaskMode> maskColumns) {
 
     /**
      * Set the columns to be renamed.
+     * Renaming preserves all other schema properties.
      * <p>
      * Note that nested columns can't be renamed, in case of GroupType column only top level column can be renamed.
      *

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/TestParquetRewriterRenameColumns.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop.rewrite;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.example.GroupReadSupport;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+public class TestParquetRewriterRenameColumns {
+  @TempDir
+  private java.nio.file.Path tempDir;
+
+  private final Configuration conf = new Configuration();
+
+  @ParameterizedTest(name = "recompress={0}")
+  @ValueSource(booleans = {false, true})
+  public void renamingPreservesSchemaStatisticsAndValues(boolean recompress) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType("message event {"
+        + " optional fixed_len_byte_array(4) amount (DECIMAL(8,2)) = 1;"
+        + " optional group tags (LIST) = 2 {"
+        + "   repeated group list { optional binary element (STRING) = 3; }"
+        + " }"
+        + " optional float score columnorder(TYPE_DEFINED_ORDER) = 4;"
+        + "}");
+    MessageType expected = MessageTypeParser.parseMessageType("message event {"
+        + " optional fixed_len_byte_array(4) renamed_amount (DECIMAL(8,2)) = 1;"
+        + " optional group renamed_tags (LIST) = 2 {"
+        + "   repeated group list { optional binary element (STRING) = 3; }"
+        + " }"
+        + " optional float renamed_score columnorder(TYPE_DEFINED_ORDER) = 4;"
+        + "}");
+    Path input = new Path(tempDir.resolve("input.parquet").toUri());
+    Path output = new Path(tempDir.resolve("output.parquet").toUri());
+    Binary negative = Binary.fromConstantByteArray(new byte[] {-1, -1, -1, -1});
+    Binary positive = Binary.fromConstantByteArray(new byte[] {0, 0, 0, 1});
+    SimpleGroupFactory groups = new SimpleGroupFactory(schema);
+    Group first = groups.newGroup().append("amount", negative).append("score", 1.5f);
+    first.addGroup("tags").addGroup("list").append("element", "one");
+    Group second = groups.newGroup().append("amount", positive).append("score", 2.5f);
+    second.addGroup("tags").addGroup("list").append("element", "two");
+    // Separate pages make recompression merge statistics for negative and positive decimals.
+    try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(input)
+        .withConf(conf)
+        .withType(schema)
+        .withCompressionCodec(CompressionCodecName.UNCOMPRESSED)
+        .withPageRowCountLimit(1)
+        .withMinRowCountForPageSizeCheck(1)
+        .withMaxRowCountForPageSizeCheck(1)
+        .build()) {
+      writer.write(first);
+      writer.write(second);
+    }
+
+    RewriteOptions.Builder builder = new RewriteOptions.Builder(conf, input, output)
+        .renameColumns(Map.of(
+            "amount", "renamed_amount",
+            "tags", "renamed_tags",
+            "score", "renamed_score"));
+    if (recompress) {
+      builder.transform(CompressionCodecName.GZIP);
+    }
+    try (ParquetRewriter rewriter = new ParquetRewriter(builder.build())) {
+      rewriter.processBlocks();
+    }
+
+    try (ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(output, conf))) {
+      assertThat(reader.getFooter().getFileMetaData().getSchema()).isEqualTo(expected);
+      var columns = reader.getFooter().getBlocks().get(0).getColumns();
+      assertThat(columns)
+          .extracting(column -> column.getCodec())
+          .containsOnly(recompress ? CompressionCodecName.GZIP : CompressionCodecName.UNCOMPRESSED);
+      var amountStatistics = columns.get(0).getStatistics();
+      assertThat(amountStatistics.genericGetMin()).isEqualTo(negative);
+      assertThat(amountStatistics.genericGetMax()).isEqualTo(positive);
+    }
+    try (ParquetReader<Group> reader = ParquetReader.builder(new GroupReadSupport(), output)
+        .withConf(conf)
+        .build()) {
+      Group actual = reader.read();
+      assertThat(actual.getBinary("renamed_amount", 0)).isEqualTo(negative);
+      assertThat(actual.getGroup("renamed_tags", 0).getGroup("list", 0).getString("element", 0))
+          .isEqualTo("one");
+      assertThat(actual.getFloat("renamed_score", 0)).isEqualTo(1.5f);
+      actual = reader.read();
+      assertThat(actual.getBinary("renamed_amount", 0)).isEqualTo(positive);
+      assertThat(actual.getGroup("renamed_tags", 0).getGroup("list", 0).getString("element", 0))
+          .isEqualTo("two");
+      assertThat(actual.getFloat("renamed_score", 0)).isEqualTo(2.5f);
+      assertThat(reader.read()).isNull();
+    }
+  }
+}