[Parquet]Fix RemoveIds change parquet repeatedElement name

parquet/src/main/java/org/apache/iceberg/parquet/RemoveIds.java

@@ -48,16 +48,19 @@ public Type struct(GroupType struct, List<Type> fields) {
 
   @Override
   public Type list(GroupType array, Type item) {
-    return Types.list(array.getRepetition())
-        .element(item)
+    GroupType repeatedElement = array.getType(0).asGroupType();
+    return Types.buildGroup(array.getRepetition())
+        .as(array.getLogicalTypeAnnotation())
+        .addField(repeatedElement.withNewFields(item))
         .named(array.getName());
   }
 
   @Override
   public Type map(GroupType map, Type key, Type value) {
-    return Types.map(map.getRepetition())
-        .key(key)
-        .value(value)
+    GroupType repeatedKeyValue = map.getType(0).asGroupType();
+    return Types.buildGroup(map.getRepetition())
+        .as(map.getLogicalTypeAnnotation())
+        .addField(repeatedKeyValue.withNewFields(key, value))
         .named(map.getName());
   }
 

parquet/src/test/java/org/apache/iceberg/parquet/TestParquetTypeVisitor.java

@@ -0,0 +1,177 @@
+/*
+ * 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.iceberg.parquet;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types.DoubleType;
+import org.apache.iceberg.types.Types.ListType;
+import org.apache.iceberg.types.Types.MapType;
+import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.types.Types.StringType;
+import org.apache.iceberg.types.Types.StructType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestParquetTypeVisitor {
+  MessageType mapSchema = Types.buildMessage()
+      .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+          .addField(Types.buildGroup(Type.Repetition.REPEATED)
+              .addField(Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED)
+                  .as(LogicalTypeAnnotation.stringType())
+                  .id(2)
+                  .named("key"))
+              .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(4).named("x"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(5).named("y"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(6).named("z"))
+                  .id(3)
+                  .named("value"))
+              .named("custom_key_value_name"))
+          .as(LogicalTypeAnnotation.mapType())
+          .id(1)
+          .named("m"))
+      .named("table");
+
+  MessageType mapSchemaWithoutIds = Types.buildMessage()
+      .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+          .addField(Types.buildGroup(Type.Repetition.REPEATED)
+              .addField(Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED)
+                  .as(LogicalTypeAnnotation.stringType())
+                  .named("key"))
+              .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).named("x"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).named("y"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).named("z"))
+                  .named("value"))
+              .named("custom_key_value_name"))
+          .as(LogicalTypeAnnotation.mapType())
+          .named("m"))
+      .named("table");
+
+  MessageType listSchema = Types.buildMessage()
+      .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+          .addField(Types.buildGroup(Type.Repetition.REPEATED)
+              .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(4).named("x"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(5).named("y"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(6).named("z"))
+                  .id(3)
+                  .named("custom_element_name"))
+              .named("custom_repeated_name"))
+          .as(LogicalTypeAnnotation.listType())
+          .id(1)
+          .named("m"))
+      .named("table");
+
+  MessageType listSchemaWithoutIds = Types.buildMessage()
+      .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+          .addField(Types.buildGroup(Type.Repetition.REPEATED)
+              .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).named("x"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).named("y"))
+                  .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).named("z"))
+                  .named("custom_element_name"))
+              .named("custom_repeated_name"))
+          .as(LogicalTypeAnnotation.listType())
+          .named("m"))
+      .named("table");
+
+  @Test
+  public void testRemoveIdsMap() {
+    MessageType actual = RemoveIds.removeIds(mapSchema);
+    Assert.assertEquals("RemoveIds should not rename repeated struct", mapSchemaWithoutIds, actual);
+  }
+
+  @Test
+  public void testRemoveIdsList() {
+    MessageType actual = RemoveIds.removeIds(listSchema);
+    Assert.assertEquals("RemoveIds should not rename repeated struct", listSchemaWithoutIds, actual);
+  }
+
+  @Test
+  public void testPruneColumnsMap() {
+    // project map.value.x and map.value.y
+    Schema projection = new Schema(
+        NestedField.optional(1, "m", MapType.ofOptional(2, 3,
+            StringType.get(),
+            StructType.of(
+                NestedField.required(4, "x", DoubleType.get()),
+                NestedField.required(5, "y", DoubleType.get())
+            )
+        ))
+    );
+
+    MessageType expected = Types.buildMessage()
+        .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+            .addField(Types.buildGroup(Type.Repetition.REPEATED)
+                .addField(Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED)
+                    .as(LogicalTypeAnnotation.stringType())
+                    .id(2)
+                    .named("key"))
+                .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+                    .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(4).named("x"))
+                    .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(5).named("y"))
+                    .id(3)
+                    .named("value"))
+                .named("custom_key_value_name"))
+            .as(LogicalTypeAnnotation.mapType())
+            .id(1)
+            .named("m"))
+        .named("table");
+
+    MessageType actual = ParquetSchemaUtil.pruneColumns(mapSchema, projection);
+    Assert.assertEquals("Pruned schema should not rename repeated struct", expected, actual);
+  }
+
+  @Test
+  public void testPruneColumnsList() {
+    // project map.value.x and map.value.y
+    Schema projection = new Schema(
+        NestedField.optional(1, "m", ListType.ofOptional(3,
+            StructType.of(
+                NestedField.required(4, "x", DoubleType.get()),
+                NestedField.required(5, "y", DoubleType.get())
+            )
+        ))
+    );
+
+    MessageType expected = Types.buildMessage()
+        .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+            .addField(Types.buildGroup(Type.Repetition.REPEATED)
+                .addField(Types.buildGroup(Type.Repetition.OPTIONAL)
+                    .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(4).named("x"))
+                    .addField(Types.primitive(PrimitiveTypeName.DOUBLE, Type.Repetition.REQUIRED).id(5).named("y"))
+                    .id(3)
+                    .named("custom_element_name"))
+                .named("custom_repeated_name"))
+            .as(LogicalTypeAnnotation.listType())
+            .id(1)
+            .named("m"))
+        .named("table");
+
+    MessageType actual = ParquetSchemaUtil.pruneColumns(listSchema, projection);
+    Assert.assertEquals("Pruned schema should not rename repeated struct", expected, actual);
+  }
+}