diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/GroupType.java b/parquet-column/src/main/java/org/apache/parquet/schema/GroupType.java index 2763b8553b..4c6a26589a 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/GroupType.java +++ b/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 diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/MessageType.java b/parquet-column/src/main/java/org/apache/parquet/schema/MessageType.java index 1aba0cc6af..da6885ad3b 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/MessageType.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/MessageType.java @@ -44,6 +44,15 @@ public MessageType(String name, List 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} */ diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java b/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java index f94ede0c91..7cf32b65b0 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java +++ b/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 diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/Type.java b/parquet-column/src/main/java/org/apache/parquet/schema/Type.java index bd682f92fd..068ab22bed 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/Type.java +++ b/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 diff --git a/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeWithName.java b/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeWithName.java new file mode 100644 index 0000000000..149ee9b406 --- /dev/null +++ b/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 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"); + } +} diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java index fce8866850..16c416a078 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java +++ b/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 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( diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/RewriteOptions.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/RewriteOptions.java index d6c9aeb1eb..67cb737967 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/RewriteOptions.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/RewriteOptions.java @@ -445,6 +445,7 @@ public Builder mask(Map maskColumns) { /** * Set the columns to be renamed. + * Renaming preserves all other schema properties. *

* Note that nested columns can't be renamed, in case of GroupType column only top level column can be renamed. * diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/TestParquetRewriterRenameColumns.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/TestParquetRewriterRenameColumns.java new file mode 100644 index 0000000000..358b67b7fd --- /dev/null +++ b/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 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 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(); + } + } +}