diff --git a/parquet-column/src/main/java/org/apache/parquet/file/FileConverters.java b/parquet-column/src/main/java/org/apache/parquet/file/FileConverters.java new file mode 100644 index 0000000000..afec69273c --- /dev/null +++ b/parquet-column/src/main/java/org/apache/parquet/file/FileConverters.java @@ -0,0 +1,150 @@ +/* + * 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.file; + +import java.util.Objects; +import java.util.function.Consumer; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.io.api.Converter; +import org.apache.parquet.io.api.GroupConverter; +import org.apache.parquet.io.api.PrimitiveConverter; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.Type; + +/** Converters for assembling values represented by a Parquet {@code FILE}-annotated group. */ +public final class FileConverters { + private FileConverters() {} + + /** + * Creates a converter for a possibly projected FILE group. + * + *

FILE values are atomic and cannot be projected by child field. The projected schema may + * contain fields absent from this particular file for schema-evolution purposes, but it must + * contain every FILE field stored in the file schema. The consumer receives {@code null} when an + * invalid FILE reference is encountered, as permitted by the FILE logical-type specification. + * + * @param schema schema presented to the converter + * @param fileSchema FILE group stored in the file + * @param consumer consumer for converted values + * @return a converter for the complete FILE value + * @throws IllegalArgumentException if the projected schema omits a stored FILE field + */ + public static GroupConverter newFileConverter( + GroupType schema, GroupType fileSchema, Consumer consumer) { + Objects.requireNonNull(schema, "schema cannot be null"); + Objects.requireNonNull(fileSchema, "fileSchema cannot be null"); + Objects.requireNonNull(consumer, "consumer cannot be null"); + FileValueValidator.requireFileSchema(schema); + FileValueValidator.requireFileSchema(fileSchema); + for (Type field : fileSchema.getFields()) { + if (!schema.containsField(field.getName())) { + throw new IllegalArgumentException( + "FILE group '" + schema.getName() + "' cannot project stored field '" + field.getName() + "'"); + } + } + return new FileGroupConverter(schema, consumer); + } + + private static final class FileGroupConverter extends GroupConverter { + private final Converter[] converters; + private final Consumer consumer; + private FileValue.Builder builder; + + private FileGroupConverter(GroupType schema, Consumer consumer) { + this.converters = new Converter[schema.getFieldCount()]; + this.consumer = consumer; + for (int index = 0; index < schema.getFieldCount(); index++) { + String fieldName = schema.getFieldName(index); + switch (fieldName) { + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.URI_FIELD: + converters[index] = stringConverter(value -> builder.withUri(value)); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.OFFSET_FIELD: + converters[index] = longConverter(value -> builder.withOffset(value)); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.SIZE_FIELD: + converters[index] = longConverter(value -> builder.withSize(value)); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CONTENT_TYPE_FIELD: + converters[index] = stringConverter(value -> builder.withContentType(value)); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CHECKSUM_FIELD: + converters[index] = stringConverter(value -> builder.withChecksum(value)); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.INLINE_FIELD: + converters[index] = binaryConverter(value -> builder.withInline(value)); + break; + default: + throw new IllegalArgumentException("Unrecognized FILE field: " + fieldName); + } + } + } + + @Override + public Converter getConverter(int fieldIndex) { + return converters[fieldIndex]; + } + + @Override + public void start() { + builder = FileValue.builder(); + } + + @Override + public void end() { + FileValue value = builder.build(); + boolean valid = true; + try { + FileValueValidator.validateValue(value); + } catch (IllegalArgumentException e) { + valid = false; + } + builder = null; + consumer.accept(valid ? value : null); + } + } + + private static PrimitiveConverter stringConverter(Consumer consumer) { + return new PrimitiveConverter() { + @Override + public void addBinary(Binary value) { + consumer.accept(value.toStringUsingUTF8()); + } + }; + } + + private static PrimitiveConverter binaryConverter(Consumer consumer) { + return new PrimitiveConverter() { + @Override + public void addBinary(Binary value) { + consumer.accept(value.copy()); + } + }; + } + + private static PrimitiveConverter longConverter(Consumer consumer) { + return new PrimitiveConverter() { + @Override + public void addLong(long value) { + consumer.accept(value); + } + }; + } +} diff --git a/parquet-column/src/main/java/org/apache/parquet/file/FileValue.java b/parquet-column/src/main/java/org/apache/parquet/file/FileValue.java new file mode 100644 index 0000000000..e746e61d2c --- /dev/null +++ b/parquet-column/src/main/java/org/apache/parquet/file/FileValue.java @@ -0,0 +1,114 @@ +/* + * 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.file; + +import org.apache.parquet.io.api.Binary; + +/** A value represented by a Parquet {@code FILE}-annotated group. */ +public final class FileValue { + private final String uri; + private final Long offset; + private final Long size; + private final String contentType; + private final String checksum; + private final Binary inline; + + private FileValue(Builder builder) { + this.uri = builder.uri; + this.offset = builder.offset; + this.size = builder.size; + this.contentType = builder.contentType; + this.checksum = builder.checksum; + this.inline = builder.inline; + } + + public static Builder builder() { + return new Builder(); + } + + public String getUri() { + return uri; + } + + public Long getOffset() { + return offset; + } + + public Long getSize() { + return size; + } + + public String getContentType() { + return contentType; + } + + public String getChecksum() { + return checksum; + } + + public Binary getInline() { + return inline; + } + + /** Builder for {@link FileValue}. */ + public static final class Builder { + private String uri; + private Long offset; + private Long size; + private String contentType; + private String checksum; + private Binary inline; + + private Builder() {} + + public Builder withUri(String uri) { + this.uri = uri; + return this; + } + + public Builder withOffset(long offset) { + this.offset = offset; + return this; + } + + public Builder withSize(long size) { + this.size = size; + return this; + } + + public Builder withContentType(String contentType) { + this.contentType = contentType; + return this; + } + + public Builder withChecksum(String checksum) { + this.checksum = checksum; + return this; + } + + public Builder withInline(Binary inline) { + this.inline = inline; + return this; + } + + public FileValue build() { + return new FileValue(this); + } + } +} diff --git a/parquet-column/src/main/java/org/apache/parquet/file/FileValueValidator.java b/parquet-column/src/main/java/org/apache/parquet/file/FileValueValidator.java new file mode 100644 index 0000000000..4bd4775b61 --- /dev/null +++ b/parquet-column/src/main/java/org/apache/parquet/file/FileValueValidator.java @@ -0,0 +1,58 @@ +/* + * 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.file; + +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; + +/** Per-value checks; FILE schema fields are validated when the GroupType is constructed. */ +final class FileValueValidator { + private FileValueValidator() {} + + static void requireFileSchema(GroupType schema) { + if (!(schema.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation)) { + throw new IllegalArgumentException( + "Cannot use a FILE value with a group without the FILE logical type: " + schema.getName()); + } + } + + static void validateValue(FileValue value) { + boolean hasUri = value.getUri() != null && !value.getUri().isEmpty(); + boolean hasInline = value.getInline() != null; + + if (!hasUri && !hasInline) { + throw new IllegalArgumentException("FILE value must set at least one of 'inline' or non-empty 'uri'"); + } + if (value.getOffset() != null && !hasUri) { + throw new IllegalArgumentException("FILE value field 'offset' may only be set together with 'uri'"); + } + if (value.getSize() != null && !hasUri) { + throw new IllegalArgumentException("FILE value field 'size' may only be set together with 'uri'"); + } + if (value.getOffset() != null && value.getSize() == null) { + throw new IllegalArgumentException("FILE value field 'size' must be set whenever 'offset' is set"); + } + if (value.getOffset() != null && value.getOffset() < 0) { + throw new IllegalArgumentException("FILE value field 'offset' must not be negative: " + value.getOffset()); + } + if (value.getSize() != null && value.getSize() < 0) { + throw new IllegalArgumentException("FILE value field 'size' must not be negative: " + value.getSize()); + } + } +} diff --git a/parquet-column/src/main/java/org/apache/parquet/file/FileValueWriter.java b/parquet-column/src/main/java/org/apache/parquet/file/FileValueWriter.java new file mode 100644 index 0000000000..a577d7cab9 --- /dev/null +++ b/parquet-column/src/main/java/org/apache/parquet/file/FileValueWriter.java @@ -0,0 +1,114 @@ +/* + * 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.file; + +import java.util.Objects; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; + +/** Writes and validates values represented by a Parquet {@code FILE}-annotated group. */ +public final class FileValueWriter { + private FileValueWriter() {} + + /** + * Writes one FILE value. The caller is responsible for starting and ending the field containing + * the FILE group. + */ + public static void write(RecordConsumer recordConsumer, GroupType schema, FileValue value) { + Objects.requireNonNull(recordConsumer, "recordConsumer cannot be null"); + Objects.requireNonNull(schema, "schema cannot be null"); + Objects.requireNonNull(value, "value cannot be null"); + + FileValueValidator.requireFileSchema(schema); + FileValueValidator.validateValue(value); + validateSchemaContainsSetFields(schema, value); + + recordConsumer.startGroup(); + for (int index = 0; index < schema.getFieldCount(); index++) { + String fieldName = schema.getFieldName(index); + switch (fieldName) { + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.URI_FIELD: + writeString(recordConsumer, fieldName, index, value.getUri()); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.OFFSET_FIELD: + writeLong(recordConsumer, fieldName, index, value.getOffset()); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.SIZE_FIELD: + writeLong(recordConsumer, fieldName, index, value.getSize()); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CONTENT_TYPE_FIELD: + writeString(recordConsumer, fieldName, index, value.getContentType()); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CHECKSUM_FIELD: + writeString(recordConsumer, fieldName, index, value.getChecksum()); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.INLINE_FIELD: + writeBinary(recordConsumer, fieldName, index, value.getInline()); + break; + default: + throw new IllegalArgumentException( + "Unrecognized field '" + fieldName + "' in FILE group '" + schema.getName() + "'"); + } + } + recordConsumer.endGroup(); + } + + private static void validateSchemaContainsSetFields(GroupType schema, FileValue value) { + requireField(schema, LogicalTypeAnnotation.FileLogicalTypeAnnotation.URI_FIELD, value.getUri()); + requireField(schema, LogicalTypeAnnotation.FileLogicalTypeAnnotation.OFFSET_FIELD, value.getOffset()); + requireField(schema, LogicalTypeAnnotation.FileLogicalTypeAnnotation.SIZE_FIELD, value.getSize()); + requireField( + schema, LogicalTypeAnnotation.FileLogicalTypeAnnotation.CONTENT_TYPE_FIELD, value.getContentType()); + requireField(schema, LogicalTypeAnnotation.FileLogicalTypeAnnotation.CHECKSUM_FIELD, value.getChecksum()); + requireField(schema, LogicalTypeAnnotation.FileLogicalTypeAnnotation.INLINE_FIELD, value.getInline()); + } + + private static void requireField(GroupType schema, String fieldName, Object value) { + if (value != null && !schema.containsField(fieldName)) { + throw new IllegalArgumentException( + "FILE value sets field '" + fieldName + "' which is absent from group '" + schema.getName() + "'"); + } + } + + private static void writeString(RecordConsumer consumer, String name, int index, String value) { + if (value != null) { + consumer.startField(name, index); + consumer.addBinary(Binary.fromString(value)); + consumer.endField(name, index); + } + } + + private static void writeLong(RecordConsumer consumer, String name, int index, Long value) { + if (value != null) { + consumer.startField(name, index); + consumer.addLong(value); + consumer.endField(name, index); + } + } + + private static void writeBinary(RecordConsumer consumer, String name, int index, Binary value) { + if (value != null) { + consumer.startField(name, index); + consumer.addBinary(value); + consumer.endField(name, index); + } + } +} 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..a2f9632575 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 @@ -22,6 +22,7 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Locale; @@ -116,10 +117,16 @@ public GroupType(Repetition repetition, String name, OriginalType originalType, GroupType( Repetition repetition, String name, LogicalTypeAnnotation logicalTypeAnnotation, List fields, ID id) { super(name, repetition, logicalTypeAnnotation, id); - this.fields = fields; + if (logicalTypeAnnotation instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation) { + // Keep the validated schema immutable so readers and writers need not revalidate every value. + this.fields = Collections.unmodifiableList(new ArrayList<>(fields)); + ((LogicalTypeAnnotation.FileLogicalTypeAnnotation) logicalTypeAnnotation).validateFields(name, this.fields); + } else { + this.fields = fields; + } this.indexByName = new HashMap<>(); - for (int i = 0; i < fields.size(); i++) { - indexByName.put(fields.get(i).getName(), i); + for (int i = 0; i < this.fields.size(); i++) { + indexByName.put(this.fields.get(i).getName(), i); } } @@ -135,6 +142,7 @@ public GroupType withId(int id) { /** * @param newFields a list of types to use as fields in a copy of this group * @return a group with the same attributes and new fields. + * @throws IllegalArgumentException if the new fields do not conform to this group's FILE annotation */ public GroupType withNewFields(List newFields) { return new GroupType(getRepetition(), getName(), getLogicalTypeAnnotation(), newFields, getId()); diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java b/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java index 625e9fd9d3..fc1de0c807 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java @@ -33,6 +33,7 @@ import static org.apache.parquet.schema.PrimitiveStringifier.TIME_STRINGIFIER; import static org.apache.parquet.schema.PrimitiveStringifier.TIME_UTC_STRINGIFIER; +import java.util.HashSet; import java.util.List; import java.util.Objects; import java.util.Optional; @@ -188,6 +189,12 @@ protected LogicalTypeAnnotation fromString(List params) { protected LogicalTypeAnnotation fromString(List params) { return unknownType(); } + }, + FILE { + @Override + protected LogicalTypeAnnotation fromString(List params) { + return fileType(); + } }; protected abstract LogicalTypeAnnotation fromString(List params); @@ -378,6 +385,10 @@ public static UnknownLogicalTypeAnnotation unknownType() { return UnknownLogicalTypeAnnotation.INSTANCE; } + public static FileLogicalTypeAnnotation fileType() { + return FileLogicalTypeAnnotation.INSTANCE; + } + public static class StringLogicalTypeAnnotation extends LogicalTypeAnnotation { private static final StringLogicalTypeAnnotation INSTANCE = new StringLogicalTypeAnnotation(); @@ -1229,6 +1240,146 @@ public boolean equals(Object obj) { } } + /** + * File logical type annotation. Annotates a group (struct) that represents a reference to a + * range of bytes, which may be stored inline in the value or in an external file. Every field is + * optional, both in the schema (a writer may omit any field from the group definition) and in the + * data (any field that is present has a field repetition type of {@code OPTIONAL}). Fields are + * identified by name (case sensitively), not by field order. A group need only define the fields + * it uses. The group may contain the following fields: + *

+ * No fields with names other than the above are permitted, and field names must be unique. Each + * declared field must match its required physical and logical types. Rules involving which fields + * are set and their values are enforced per value rather than by the schema. + */ + public static class FileLogicalTypeAnnotation extends LogicalTypeAnnotation { + private static final FileLogicalTypeAnnotation INSTANCE = new FileLogicalTypeAnnotation(); + + /** Field name holding the URI-reference of an external file. */ + public static final String URI_FIELD = "uri"; + + /** Field name holding the start of the byte range. */ + public static final String OFFSET_FIELD = "offset"; + + /** Field name holding the byte length of the referenced data. */ + public static final String SIZE_FIELD = "size"; + + /** Field name holding the media (MIME) type of the resolved bytes. */ + public static final String CONTENT_TYPE_FIELD = "content_type"; + + /** Field name holding the integrity token for the resolved bytes. */ + public static final String CHECKSUM_FIELD = "checksum"; + + /** Field name holding the referenced bytes stored inline. */ + public static final String INLINE_FIELD = "inline"; + + /** All recognized field names in a FILE-annotated group. All fields are optional. */ + public static final Set FIELD_NAMES = + Set.of(URI_FIELD, OFFSET_FIELD, SIZE_FIELD, CONTENT_TYPE_FIELD, CHECKSUM_FIELD, INLINE_FIELD); + + private FileLogicalTypeAnnotation() {} + + void validateFields(String groupName, List fields) { + Set fieldNames = new HashSet<>(); + for (Type field : fields) { + String fieldName = field.getName(); + Preconditions.checkArgument( + fieldNames.add(fieldName), + "FILE type group '%s' contains duplicate field '%s'", + groupName, + fieldName); + Preconditions.checkArgument( + FIELD_NAMES.contains(fieldName), + "FILE type group '%s' contains unrecognized field '%s'. Valid fields are: %s", + groupName, + fieldName, + String.join(", ", FIELD_NAMES)); + Preconditions.checkArgument( + field.isPrimitive() && field.getRepetition() == Type.Repetition.OPTIONAL, + "FILE type field '%s' must be an optional primitive in group '%s'", + fieldName, + groupName); + validateField(groupName, field.asPrimitiveType()); + } + } + + private void validateField(String groupName, PrimitiveType field) { + String fieldName = field.getName(); + PrimitiveType.PrimitiveTypeName physicalType = field.getPrimitiveTypeName(); + switch (fieldName) { + case URI_FIELD: + case CONTENT_TYPE_FIELD: + case CHECKSUM_FIELD: + Preconditions.checkArgument( + physicalType == PrimitiveType.PrimitiveTypeName.BINARY + && field.getLogicalTypeAnnotation() instanceof StringLogicalTypeAnnotation, + "FILE type field '%s' must be a STRING (BINARY annotated as STRING) in group '%s'", + fieldName, + groupName); + break; + case OFFSET_FIELD: + case SIZE_FIELD: + Preconditions.checkArgument( + physicalType == PrimitiveType.PrimitiveTypeName.INT64 + && field.getLogicalTypeAnnotation() == null, + "FILE type field '%s' must be an unannotated INT64 in group '%s'", + fieldName, + groupName); + break; + case INLINE_FIELD: + Preconditions.checkArgument( + physicalType == PrimitiveType.PrimitiveTypeName.BINARY + && field.getLogicalTypeAnnotation() == null, + "FILE type field '%s' must be an unannotated BYTE_ARRAY (BINARY) in group '%s'", + fieldName, + groupName); + break; + default: + throw new IllegalArgumentException("Unrecognized FILE field: " + fieldName); + } + } + + @Override + public OriginalType toOriginalType() { + return null; + } + + @Override + public Optional accept(LogicalTypeAnnotationVisitor logicalTypeAnnotationVisitor) { + return logicalTypeAnnotationVisitor.visit(this); + } + + @Override + LogicalTypeToken getType() { + return LogicalTypeToken.FILE; + } + + @Override + public boolean equals(Object obj) { + return obj instanceof FileLogicalTypeAnnotation; + } + + @Override + public int hashCode() { + return getClass().hashCode(); + } + } + public static class GeometryLogicalTypeAnnotation extends LogicalTypeAnnotation { private final String crs; @@ -1434,5 +1585,9 @@ default Optional visit(GeographyLogicalTypeAnnotation geographyLogicalType) { default Optional visit(UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) { return empty(); } + + default Optional visit(FileLogicalTypeAnnotation fileLogicalType) { + return empty(); + } } } diff --git a/parquet-column/src/test/java/org/apache/parquet/file/TestFileConverters.java b/parquet-column/src/test/java/org/apache/parquet/file/TestFileConverters.java new file mode 100644 index 0000000000..33028df6ae --- /dev/null +++ b/parquet-column/src/test/java/org/apache/parquet/file/TestFileConverters.java @@ -0,0 +1,224 @@ +/* + * 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.file; + +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.ArrayList; +import java.util.List; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.io.api.GroupConverter; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +public class TestFileConverters { + private static final GroupType SCHEMA = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("checksum") + .optional(BINARY) + .named("inline") + .named("file"); + + @Test + public void testConvertsFileValue() { + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(SCHEMA, SCHEMA, values::add); + + converter.start(); + converter.getConverter(0).asPrimitiveConverter().addBinary(Binary.fromString("s3://bucket/file")); + converter.getConverter(1).asPrimitiveConverter().addLong(10); + converter.getConverter(2).asPrimitiveConverter().addLong(20); + converter.getConverter(3).asPrimitiveConverter().addBinary(Binary.fromString("image/png")); + converter.getConverter(4).asPrimitiveConverter().addBinary(Binary.fromString("ETAG:abc")); + converter.getConverter(5).asPrimitiveConverter().addBinary(Binary.fromString("bytes")); + converter.end(); + + assertThat(values).hasSize(1); + FileValue value = values.get(0); + assertThat(value.getUri()).isEqualTo("s3://bucket/file"); + assertThat(value.getOffset()).isEqualTo(10L); + assertThat(value.getSize()).isEqualTo(20L); + assertThat(value.getContentType()).isEqualTo("image/png"); + assertThat(value.getChecksum()).isEqualTo("ETAG:abc"); + assertThat(value.getInline().toStringUsingUTF8()).isEqualTo("bytes"); + } + + @Test + public void testConvertsInvalidValueToNull() { + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(SCHEMA, SCHEMA, values::add); + + converter.start(); + converter.getConverter(1).asPrimitiveConverter().addLong(10); + converter.getConverter(2).asPrimitiveConverter().addLong(20); + converter.end(); + + assertThat(values).containsExactly((FileValue) null); + } + + @Test + public void testResetsBetweenValues() { + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(SCHEMA, SCHEMA, values::add); + + converter.start(); + converter.getConverter(0).asPrimitiveConverter().addBinary(Binary.fromString("first")); + converter.end(); + converter.start(); + converter.getConverter(5).asPrimitiveConverter().addBinary(Binary.fromString("second")); + converter.end(); + + assertThat(values).hasSize(2); + assertThat(values.get(0).getUri()).isEqualTo("first"); + assertThat(values.get(0).getInline()).isNull(); + assertThat(values.get(1).getUri()).isNull(); + assertThat(values.get(1).getInline().toStringUsingUTF8()).isEqualTo("second"); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testSizeWithoutUriConvertsToNull(boolean emptyUri) { + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(SCHEMA, SCHEMA, values::add); + + converter.start(); + if (emptyUri) { + converter.getConverter(0).asPrimitiveConverter().addBinary(Binary.fromString("")); + } + converter.getConverter(2).asPrimitiveConverter().addLong(2); + converter.getConverter(5).asPrimitiveConverter().addBinary(Binary.fromString("bytes")); + converter.end(); + + assertThat(values).containsExactly((FileValue) null); + } + + @Test + public void testResetsAfterInvalidValue() { + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(SCHEMA, SCHEMA, values::add); + + converter.start(); + converter.getConverter(0).asPrimitiveConverter().addBinary(Binary.fromString("first")); + converter.getConverter(1).asPrimitiveConverter().addLong(1); + converter.getConverter(2).asPrimitiveConverter().addLong(2); + converter.getConverter(3).asPrimitiveConverter().addBinary(Binary.fromString("image/png")); + converter.getConverter(4).asPrimitiveConverter().addBinary(Binary.fromString("ETAG:abc")); + converter.end(); + converter.start(); + converter.getConverter(3).asPrimitiveConverter().addBinary(Binary.fromString("image/jpeg")); + converter.end(); + converter.start(); + converter.getConverter(5).asPrimitiveConverter().addBinary(Binary.fromConstantByteArray(new byte[0])); + converter.end(); + + assertThat(values).hasSize(3); + assertThat(values.get(0).getUri()).isEqualTo("first"); + assertThat(values.get(1)).isNull(); + FileValue last = values.get(2); + assertThat(last.getInline().getBytes()).isEmpty(); + assertThat(last.getUri()).isNull(); + assertThat(last.getOffset()).isNull(); + assertThat(last.getSize()).isNull(); + assertThat(last.getContentType()).isNull(); + assertThat(last.getChecksum()).isNull(); + } + + @Test + public void testCopiesReusedInlineBuffer() { + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(SCHEMA, SCHEMA, values::add); + byte[] bytes = {1, 2, 3}; + Binary binary = Binary.fromReusedByteArray(bytes); + + converter.start(); + converter.getConverter(5).asPrimitiveConverter().addBinary(binary); + converter.end(); + bytes[0] = 4; + converter.start(); + converter.getConverter(5).asPrimitiveConverter().addBinary(binary); + converter.end(); + + assertThat(values).hasSize(2); + assertThat(values.get(0).getInline().getBytes()).containsExactly(1, 2, 3); + assertThat(values.get(1).getInline().getBytes()).containsExactly(4, 2, 3); + } + + @Test + public void testRetainsConstantInlineBuffer() { + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(SCHEMA, SCHEMA, values::add); + Binary binary = Binary.fromConstantByteArray(new byte[] {1, 2, 3}); + + converter.start(); + converter.getConverter(5).asPrimitiveConverter().addBinary(binary); + converter.end(); + + assertThat(values).hasSize(1); + assertThat(values.get(0).getInline()).isSameAs(binary); + } + + @Test + public void testAllowsAddedFieldsAndReorderedProjection() { + GroupType storedSchema = SCHEMA.withNewFields(SCHEMA.getType("uri")); + GroupType requestedSchema = SCHEMA.withNewFields(SCHEMA.getType("inline"), SCHEMA.getType("uri")); + List values = new ArrayList<>(); + GroupConverter converter = FileConverters.newFileConverter(requestedSchema, storedSchema, values::add); + + converter.start(); + converter.getConverter(1).asPrimitiveConverter().addBinary(Binary.fromString("s3://bucket/file")); + converter.end(); + + assertThat(values).hasSize(1); + assertThat(values.get(0).getUri()).isEqualTo("s3://bucket/file"); + assertThat(values.get(0).getInline()).isNull(); + } + + @Test + public void testRejectsPartialFileProjection() { + GroupType projectedSchema = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .named("file"); + + assertThatThrownBy(() -> FileConverters.newFileConverter(projectedSchema, SCHEMA, value -> {})) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE group 'file' cannot project stored field 'uri'"); + } +} diff --git a/parquet-column/src/test/java/org/apache/parquet/file/TestFileValueWriter.java b/parquet-column/src/test/java/org/apache/parquet/file/TestFileValueWriter.java new file mode 100644 index 0000000000..e1ec857b0c --- /dev/null +++ b/parquet-column/src/test/java/org/apache/parquet/file/TestFileValueWriter.java @@ -0,0 +1,231 @@ +/* + * 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.file; + +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Deque; +import org.apache.parquet.io.ExpectationValidatingRecordConsumer; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.Types; +import org.junit.jupiter.api.Test; + +public class TestFileValueWriter { + private static final GroupType ALL_FIELDS_SCHEMA = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("checksum") + .optional(BINARY) + .named("inline") + .named("file"); + + @Test + public void testWritesSetFieldsInSchemaOrder() { + FileValue value = FileValue.builder() + .withChecksum("ETAG:abc") + .withSize(20) + .withInline(Binary.fromString("bytes")) + .withContentType("image/png") + .withUri("s3://bucket/file") + .withOffset(10) + .build(); + GroupType schema = ALL_FIELDS_SCHEMA.withNewFields( + ALL_FIELDS_SCHEMA.getType("inline"), + ALL_FIELDS_SCHEMA.getType("size"), + ALL_FIELDS_SCHEMA.getType("uri"), + ALL_FIELDS_SCHEMA.getType("checksum"), + ALL_FIELDS_SCHEMA.getType("offset"), + ALL_FIELDS_SCHEMA.getType("content_type")); + Deque expectations = new ArrayDeque<>(Arrays.asList( + "startGroup()", + "startField(inline, 0)", + "addBinary(bytes)", + "endField(inline, 0)", + "startField(size, 1)", + "addLong(20)", + "endField(size, 1)", + "startField(uri, 2)", + "addBinary(s3://bucket/file)", + "endField(uri, 2)", + "startField(checksum, 3)", + "addBinary(ETAG:abc)", + "endField(checksum, 3)", + "startField(offset, 4)", + "addLong(10)", + "endField(offset, 4)", + "startField(content_type, 5)", + "addBinary(image/png)", + "endField(content_type, 5)", + "endGroup()")); + + FileValueWriter.write(new ExpectationValidatingRecordConsumer(expectations), schema, value); + + assertThat(expectations).isEmpty(); + } + + @Test + public void testWritesInlineOnlyValue() { + GroupType schema = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .named("inline") + .named("file"); + FileValue value = FileValue.builder() + .withInline(Binary.fromConstantByteArray(new byte[0])) + .build(); + Deque expectations = new ArrayDeque<>(Arrays.asList( + "startGroup()", "startField(inline, 0)", "addBinary()", "endField(inline, 0)", "endGroup()")); + + FileValueWriter.write(new ExpectationValidatingRecordConsumer(expectations), schema, value); + + assertThat(expectations).isEmpty(); + } + + @Test + public void testRejectsUnresolvableValue() { + FileValue value = FileValue.builder().withContentType("image/png").build(); + + assertThatThrownBy(() -> FileValueWriter.write(noopConsumer(), ALL_FIELDS_SCHEMA, value)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE value must set at least one of 'inline' or non-empty 'uri'"); + } + + @Test + public void testRejectsOffsetWithoutUri() { + FileValue value = FileValue.builder() + .withOffset(1) + .withSize(2) + .withInline(Binary.fromString("bytes")) + .build(); + + assertThatThrownBy(() -> FileValueWriter.write(noopConsumer(), ALL_FIELDS_SCHEMA, value)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE value field 'offset' may only be set together with 'uri'"); + } + + @Test + public void testRejectsOffsetWithoutSize() { + FileValue value = FileValue.builder().withUri("file").withOffset(1).build(); + + assertThatThrownBy(() -> FileValueWriter.write(noopConsumer(), ALL_FIELDS_SCHEMA, value)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE value field 'size' must be set whenever 'offset' is set"); + } + + @Test + public void testRejectsSizeWithoutUri() { + FileValue value = FileValue.builder() + .withSize(2) + .withInline(Binary.fromString("bytes")) + .build(); + + assertThatThrownBy(() -> FileValueWriter.write(noopConsumer(), ALL_FIELDS_SCHEMA, value)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE value field 'size' may only be set together with 'uri'"); + } + + @Test + public void testRejectsNegativeOffsetAndSize() { + FileValue negativeOffset = + FileValue.builder().withUri("file").withOffset(-1).withSize(1).build(); + FileValue negativeSize = + FileValue.builder().withUri("file").withSize(-1).build(); + + assertThatThrownBy(() -> FileValueWriter.write(noopConsumer(), ALL_FIELDS_SCHEMA, negativeOffset)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE value field 'offset' must not be negative: -1"); + assertThatThrownBy(() -> FileValueWriter.write(noopConsumer(), ALL_FIELDS_SCHEMA, negativeSize)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE value field 'size' must not be negative: -1"); + } + + @Test + public void testRejectsSetFieldMissingFromSchema() { + GroupType schema = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .named("file"); + FileValue value = FileValue.builder().withUri("file").withSize(10).build(); + + assertThatThrownBy(() -> FileValueWriter.write(noopConsumer(), schema, value)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE value sets field 'size' which is absent from group 'file'"); + } + + private static RecordConsumer noopConsumer() { + return new RecordConsumer() { + @Override + public void startMessage() {} + + @Override + public void endMessage() {} + + @Override + public void startField(String field, int index) {} + + @Override + public void endField(String field, int index) {} + + @Override + public void startGroup() {} + + @Override + public void endGroup() {} + + @Override + public void addInteger(int value) {} + + @Override + public void addLong(long value) {} + + @Override + public void addBoolean(boolean value) {} + + @Override + public void addBinary(Binary value) {} + + @Override + public void addFloat(float value) {} + + @Override + public void addDouble(double value) {} + }; + } +} diff --git a/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java b/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java index 9a4c0eeecd..ca12479323 100644 --- a/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java +++ b/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java @@ -52,12 +52,14 @@ import org.apache.parquet.io.api.RecordConsumer; import org.apache.parquet.io.api.RecordMaterializer; import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.MessageTypeParser; import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type; import org.apache.parquet.schema.Type.Repetition; +import org.apache.parquet.schema.Types; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -243,6 +245,45 @@ public void testReadUsingProjectedSchema(boolean useDictionary) { } } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testColumnIOAllowsProjectionWithinFileGroup(boolean useDictionary) { + GroupType fileGroup = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .named("file"); + GroupType projectedFileGroup = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .named("file"); + + MessageColumnIO columnIO = new ColumnIOFactory() + .getColumnIO(new MessageType("schema", projectedFileGroup), new MessageType("schema", fileGroup)); + + assertThat(columnIO.getLeaves()).hasSize(1); + + MessageType fileSchema = new MessageType("schema", fileGroup); + MessageType requestedSchema = new MessageType("schema", projectedFileGroup); + Group record = new SimpleGroupFactory(fileSchema).newGroup(); + record.addGroup("file").append("uri", "s3://bucket/file").append("content_type", "image/png"); + MemPageStore pages = new MemPageStore(1); + writeGroups(fileSchema, pages, useDictionary, record); + + List records = readGroups(pages, fileSchema, requestedSchema, 1); + + assertThat(records).hasSize(1); + Group projected = records.get(0).getGroup("file", 0); + assertThat(projected.getType()).isEqualTo(projectedFileGroup); + assertThat(projected.getString("content_type", 0)).isEqualTo("image/png"); + } + private void validateGroups(List groups1, Object[][] e1) { Iterator i1 = groups1.iterator(); for (int i = 0; i < e1.length; i++) { diff --git a/parquet-column/src/test/java/org/apache/parquet/parser/TestParquetParser.java b/parquet-column/src/test/java/org/apache/parquet/parser/TestParquetParser.java index fc5b98ff18..5c9f85d7b8 100644 --- a/parquet-column/src/test/java/org/apache/parquet/parser/TestParquetParser.java +++ b/parquet-column/src/test/java/org/apache/parquet/parser/TestParquetParser.java @@ -66,6 +66,28 @@ import org.junit.jupiter.api.Test; public class TestParquetParser { + @Test + public void testFILEAnnotation() { + String schema = "message FileMessage {\n" + + " optional group file (FILE) {\n" + + " optional binary uri (STRING);\n" + + " }\n" + + "}\n"; + MessageType expected = Types.buildMessage() + .optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(stringType()) + .named("uri") + .named("file") + .named("FileMessage"); + + MessageType parsed = parseMessageType(schema); + + assertThat(parsed).isEqualTo(expected); + assertThat(parseMessageType(parsed.toString())).isEqualTo(expected); + } + @Test public void testPaperExample() { String example = "message Document {\n" + " required int64 DocId;\n" diff --git a/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java b/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java index fbc79cc81e..ae2f32477c 100644 --- a/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java +++ b/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java @@ -43,9 +43,14 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type.Repetition; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; public class TestTypeBuildersWithLogicalTypes { @Test @@ -587,4 +592,374 @@ public void testVariantLogicalTypeWithShredded() { assertThat(((LogicalTypeAnnotation.VariantLogicalTypeAnnotation) annotation).getSpecVersion()) .isEqualTo(specVersion); } + + @Test + public void testFileLogicalTypeUriOnly() { + String name = "file_field"; + GroupType file = new GroupType( + REQUIRED, + name, + LogicalTypeAnnotation.fileType(), + Types.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")); + + assertThat(file.toString()) + .isEqualTo("required group file_field (FILE) {\n" + " optional binary uri (STRING);\n" + "}"); + + LogicalTypeAnnotation annotation = file.getLogicalTypeAnnotation(); + assertThat(annotation.getType()).isEqualTo(LogicalTypeAnnotation.LogicalTypeToken.FILE); + assertThat(annotation.toOriginalType()).isNull(); + assertThat(annotation).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + } + + @Test + public void testFileLogicalTypeAllFields() { + String name = "file_field"; + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("checksum") + .optional(BINARY) + .named("inline") + .named(name); + + LogicalTypeAnnotation annotation = file.getLogicalTypeAnnotation(); + assertThat(annotation).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(file.getFieldCount()).isEqualTo(6); + assertThat(file.getType("uri").getName()).isEqualTo("uri"); + assertThat(file.getType("offset").getName()).isEqualTo("offset"); + assertThat(file.getType("size").getName()).isEqualTo("size"); + assertThat(file.getType("content_type").getName()).isEqualTo("content_type"); + assertThat(file.getType("checksum").getName()).isEqualTo("checksum"); + assertThat(file.getType("inline").getName()).isEqualTo("inline"); + } + + @Test + public void testFileLogicalTypeInlineOnly() { + // Every field is optional, so an inline-only group is valid (spec inline case). + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .named("inline") + .named("inline_file"); + + assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(file.getFieldCount()).isEqualTo(1); + assertThat(file.getType("inline").getName()).isEqualTo("inline"); + } + + @Test + public void testFileLogicalTypeAllowsOffsetWithoutUri() { + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .optional(BINARY) + .named("inline") + .named("file_offset_without_uri"); + + assertThat(file.getFieldCount()).isEqualTo(3); + } + + @Test + public void testFileLogicalTypeExternalRangedReferenceWithoutInline() { + // An external ranged reference declares 'uri' + 'offset' + 'size' to point at a byte range of + // an external file. Because 'uri' is declared, the schema is treated as an external-reference + // schema and is not required to declare 'inline', even though it declares 'offset'. + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .named("external_ranged_file"); + + assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(file.getFieldCount()).isEqualTo(3); + } + + @Test + public void testFileLogicalTypeAllowsMetadataOnlySchema() { + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("checksum") + .named("file_metadata_only"); + + assertThat(file.getFieldCount()).isEqualTo(2); + } + + @Test + public void testFileLogicalTypeAllowsSizeOnlySchema() { + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64) + .named("size") + .named("file_size_only"); + + assertThat(file.getFieldCount()).isEqualTo(1); + } + + @Test + public void testFileLogicalTypeAllowsOffsetWithoutSize() { + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") + .named("file_offset_without_size"); + + assertThat(file.getFieldCount()).isEqualTo(2); + } + + @Test + public void testFileLogicalTypeOffsetWithSize() { + // 'offset' accompanied by 'size' is valid. + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .named("file_offset_with_size"); + + assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(file.getFieldCount()).isEqualTo(3); + } + + @Test + public void testFileLogicalTypeSizeWithoutOffset() { + // 'uri' + 'size' (without 'offset') is valid: an external reference to '[0, size)'. + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("size") + .named("file_size_without_offset"); + + assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(file.getFieldCount()).isEqualTo(2); + } + + @Test + public void testFileLogicalTypeRejectsUnrecognizedField() { + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(BINARY) + .named("unknown_field") + .named("file_with_bad_field")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testFileLogicalTypeRejectsRequiredField() { + // All FILE fields must have OPTIONAL repetition under the current spec. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .required(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .named("file_with_required_uri")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testFileLogicalTypeRejectsGroupField() { + // FILE fields must be primitives, not nested groups. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optionalGroup() + .optional(BINARY) + .named("nested") + .named("uri") + .named("file_with_group_field")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testFileLogicalTypeRejectsWrongStringPhysicalType() { + // 'uri' must be a STRING (BINARY annotated as STRING); an INT64 is rejected. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64) + .named("uri") + .named("file_uri_wrong_type")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testFileLogicalTypeRejectsUnannotatedStringField() { + // A STRING field must carry the STRING logical annotation; plain BINARY is rejected. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .named("uri") + .named("file_uri_unannotated")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testFileLogicalTypeRejectsWrongInt64PhysicalType() { + // 'offset' and 'size' must be INT64; an INT32 is rejected. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT32) + .named("size") + .named("file_size_wrong_type")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testFileLogicalTypeRejectsWrongInlinePhysicalType() { + // 'inline' must be a BYTE_ARRAY (BINARY); an INT64 is rejected. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64) + .named("inline") + .named("file_inline_wrong_type")) + .isInstanceOf(IllegalArgumentException.class); + } + + @ParameterizedTest + @ValueSource(strings = {"offset", "size"}) + public void testFileLogicalTypeRejectsAnnotatedRangeField(String fieldName) { + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64) + .as(timestampType(true, MICROS)) + .named(fieldName) + .named("file_with_annotated_range")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE type field '" + fieldName + + "' must be an unannotated INT64 in group 'file_with_annotated_range'"); + } + + @Test + public void testFileLogicalTypeRejectsAnnotatedInline() { + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(stringType()) + .named("inline") + .named("file_with_annotated_inline")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE type field 'inline' must be an unannotated BYTE_ARRAY (BINARY) in group " + + "'file_with_annotated_inline'"); + } + + @ParameterizedTest + @ValueSource(strings = {"uri", "offset", "size", "content_type", "checksum", "inline"}) + public void testFileLogicalTypeRejectsDuplicateFields(String fieldName) { + PrimitiveType field; + if (fieldName.equals("offset") || fieldName.equals("size")) { + field = Types.optional(INT64).named(fieldName); + } else if (fieldName.equals("inline")) { + field = Types.optional(BINARY).named(fieldName); + } else { + field = Types.optional(BINARY).as(stringType()).named(fieldName); + } + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .addFields(field, field) + .named("file_with_duplicate_field")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE type group 'file_with_duplicate_field' contains duplicate field '" + fieldName + "'"); + } + + @Test + public void testFileLogicalTypeConstructorRejectsInvalidFields() { + PrimitiveType invalidUri = Types.optional(INT32).named("uri"); + + assertThatThrownBy( + () -> new GroupType(Repetition.OPTIONAL, "file", LogicalTypeAnnotation.fileType(), invalidUri)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE type field 'uri' must be a STRING (BINARY annotated as STRING) in group 'file'"); + } + + @Test + public void testFileLogicalTypeFieldsCannotChangeAfterValidation() { + PrimitiveType uri = Types.optional(BINARY).as(stringType()).named("uri"); + List fields = new ArrayList<>(Arrays.asList(uri)); + GroupType file = new GroupType(Repetition.OPTIONAL, "file", LogicalTypeAnnotation.fileType(), fields); + + fields.add(Types.optional(INT32).named("unknown")); + + assertThat(file.getFields()).containsExactly(uri); + assertThat(file.getFieldIndex("uri")).isZero(); + assertThat(file.containsField("unknown")).isFalse(); + // Collections.unmodifiableList throws without a message. + assertThatThrownBy(() -> file.getFields().clear()).isInstanceOf(UnsupportedOperationException.class); + } + + @Test + public void testFileLogicalTypeWithNewFieldsRejectsInvalidFields() { + GroupType file = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(stringType()) + .named("uri") + .named("file"); + PrimitiveType invalidUri = Types.optional(INT32).named("uri"); + + assertThatThrownBy(() -> file.withNewFields(invalidUri)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE type field 'uri' must be a STRING (BINARY annotated as STRING) in group 'file'"); + assertThatThrownBy(() -> file.withNewFields(Arrays.asList(file.getType("uri"), file.getType("uri")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE type group 'file' contains duplicate field 'uri'"); + } + + @Test + public void testFileLogicalTypeWithNewFieldsPreservesAttributes() { + GroupType file = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .id(42) + .optional(BINARY) + .as(stringType()) + .named("uri") + .optional(BINARY) + .named("inline") + .named("file"); + + GroupType projected = file.withNewFields(file.getType("inline")); + + assertThat(projected.getFields()).containsExactly(file.getType("inline")); + assertThat(projected.getLogicalTypeAnnotation()).isEqualTo(LogicalTypeAnnotation.fileType()); + assertThat(projected.getName()).isEqualTo("file"); + assertThat(projected.getRepetition()).isEqualTo(Repetition.OPTIONAL); + assertThat(projected.getId()).isEqualTo(file.getId()); + assertThat(file.getFields()).hasSize(2); + } } diff --git a/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java b/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java index 8956d3944e..8aa21e0ae3 100644 --- a/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java +++ b/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java @@ -60,4 +60,5 @@ public static LogicalType VARIANT(byte specificationVersion) { public static final LogicalType BSON = LogicalType.BSON(new BsonType()); public static final LogicalType FLOAT16 = LogicalType.FLOAT16(new Float16Type()); public static final LogicalType UUID = LogicalType.UUID(new UUIDType()); + public static final LogicalType FILE = LogicalType.FILE(new FileType()); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java index 37c71d888b..f3813f836f 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java @@ -606,6 +606,11 @@ public Optional visit(LogicalTypeAnnotation.GeographyLogicalTypeAnn geographyType.setAlgorithm(fromParquetEdgeInterpolationAlgorithm(geographyLogicalType.getAlgorithm())); return of(LogicalType.GEOGRAPHY(geographyType)); } + + @Override + public Optional visit(LogicalTypeAnnotation.FileLogicalTypeAnnotation fileLogicalType) { + return of(LogicalTypes.FILE); + } } private void addRowGroup( @@ -1431,9 +1436,7 @@ LogicalTypeAnnotation getLogicalTypeAnnotation(LogicalType type) { VariantType variant = type.getVARIANT(); return LogicalTypeAnnotation.variantType(variant.getSpecification_version()); case FILE: - // Present in the format but not mapped to a LogicalTypeAnnotation yet. Ignore it to - // preserve the physical type, as an unrecognised logical type would be. - return null; + return LogicalTypeAnnotation.fileType(); default: throw new RuntimeException("Unknown logical type " + type); } diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/file/TestFileValueReadWrite.java b/parquet-hadoop/src/test/java/org/apache/parquet/file/TestFileValueReadWrite.java new file mode 100644 index 0000000000..fda8515fbb --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/file/TestFileValueReadWrite.java @@ -0,0 +1,293 @@ +/* + * 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.file; + +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.util.Collections; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.DirectWriterTest; +import org.apache.parquet.conf.ParquetConfiguration; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.api.ReadSupport; +import org.apache.parquet.hadoop.api.WriteSupport; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.io.api.Converter; +import org.apache.parquet.io.api.GroupConverter; +import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.io.api.RecordMaterializer; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +public class TestFileValueReadWrite extends DirectWriterTest { + private static final GroupType FILE_SCHEMA = Types.optionalGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("checksum") + .optional(BINARY) + .named("inline") + .named("file"); + private static final MessageType SCHEMA = new MessageType("message", FILE_SCHEMA); + + @TempDir + private java.nio.file.Path tempDir; + + @Test + public void testFileValueRoundTrip() throws IOException { + FileValue[] expectedValues = { + FileValue.builder() + .withUri("s3://bucket/file") + .withOffset(10) + .withSize(20) + .withContentType("image/png") + .withChecksum("ETAG:abc") + .withInline(Binary.fromString("bytes")) + .build(), + FileValue.builder().withInline(Binary.fromString("inline only")).build(), + null, + FileValue.builder().withUri("relative/path").build(), + FileValue.builder().withUri("file").withOffset(0).withSize(0).build(), + FileValue.builder() + .withInline(Binary.fromConstantByteArray(new byte[0])) + .build() + }; + Path path = new Path(tempDir.resolve("files.parquet").toUri()); + try (ParquetWriter writer = new ParquetWriter<>(path, new FileWriteSupport())) { + for (FileValue value : expectedValues) { + writer.write(value); + } + } + + try (ParquetReader reader = + ParquetReader.builder(new FileReadSupport(), path).build()) { + for (FileValue expected : expectedValues) { + FileRecord record = reader.read(); + assertThat(record).isNotNull(); + if (expected == null) { + assertThat(record.value).isNull(); + } else { + FileValue actual = record.value; + assertThat(actual).isNotNull(); + assertThat(actual.getUri()).isEqualTo(expected.getUri()); + assertThat(actual.getOffset()).isEqualTo(expected.getOffset()); + assertThat(actual.getSize()).isEqualTo(expected.getSize()); + assertThat(actual.getContentType()).isEqualTo(expected.getContentType()); + assertThat(actual.getChecksum()).isEqualTo(expected.getChecksum()); + assertThat(actual.getInline()).isEqualTo(expected.getInline()); + } + } + assertThat(reader.read()).isNull(); + } + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testInvalidStoredFileValueConvertsToNull(boolean sizeWithoutUri) throws IOException { + Path path = writeDirect(SCHEMA, consumer -> { + consumer.startMessage(); + consumer.startField("file", 0); + consumer.startGroup(); + if (sizeWithoutUri) { + consumer.startField("size", FILE_SCHEMA.getFieldIndex("size")); + consumer.addLong(2); + consumer.endField("size", FILE_SCHEMA.getFieldIndex("size")); + } + consumer.startField("content_type", FILE_SCHEMA.getFieldIndex("content_type")); + consumer.addBinary(Binary.fromString("image/png")); + consumer.endField("content_type", FILE_SCHEMA.getFieldIndex("content_type")); + if (sizeWithoutUri) { + consumer.startField("inline", FILE_SCHEMA.getFieldIndex("inline")); + consumer.addBinary(Binary.fromString("bytes")); + consumer.endField("inline", FILE_SCHEMA.getFieldIndex("inline")); + } + consumer.endGroup(); + consumer.endField("file", 0); + consumer.endMessage(); + }); + + try (ParquetReader reader = + ParquetReader.builder(new FileReadSupport(), path).build()) { + assertThat(reader.read().value).isNull(); + assertThat(reader.read()).isNull(); + } + } + + @Test + public void testFileConverterRejectsPartialProjectionDuringReaderAssembly() throws IOException { + Path path = writeDirect(SCHEMA, consumer -> { + consumer.startMessage(); + consumer.startField("file", 0); + FileValueWriter.write( + consumer, FILE_SCHEMA, FileValue.builder().withUri("file").build()); + consumer.endField("file", 0); + consumer.endMessage(); + }); + MessageType requestedSchema = new MessageType("message", FILE_SCHEMA.withNewFields(FILE_SCHEMA.getType("uri"))); + + try (ParquetReader reader = ParquetReader.builder(new FileReadSupport(requestedSchema), path) + .build()) { + assertThatThrownBy(reader::read) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("FILE group 'file' cannot project stored field 'offset'"); + } + } + + @Test + public void testFileConverterReadsReorderedSchemaWithAddedFields() throws IOException { + GroupType storedFile = FILE_SCHEMA.withNewFields(FILE_SCHEMA.getType("uri")); + Path path = writeDirect(new MessageType("message", storedFile), consumer -> { + consumer.startMessage(); + consumer.startField("file", 0); + FileValueWriter.write( + consumer, storedFile, FileValue.builder().withUri("file").build()); + consumer.endField("file", 0); + consumer.endMessage(); + }); + MessageType requestedSchema = new MessageType( + "message", FILE_SCHEMA.withNewFields(FILE_SCHEMA.getType("inline"), FILE_SCHEMA.getType("uri"))); + + try (ParquetReader reader = ParquetReader.builder(new FileReadSupport(requestedSchema), path) + .build()) { + FileValue actual = reader.read().value; + assertThat(actual.getUri()).isEqualTo("file"); + assertThat(actual.getInline()).isNull(); + assertThat(reader.read()).isNull(); + } + } + + private static final class FileRecord { + private final FileValue value; + + private FileRecord(FileValue value) { + this.value = value; + } + } + + private static final class FileReadSupport extends ReadSupport { + private final MessageType requestedSchema; + + private FileReadSupport() { + this(null); + } + + private FileReadSupport(MessageType requestedSchema) { + this.requestedSchema = requestedSchema; + } + + @Override + public ReadContext init( + ParquetConfiguration configuration, Map metadata, MessageType fileSchema) { + return new ReadContext(requestedSchema == null ? fileSchema : requestedSchema); + } + + @Override + public RecordMaterializer prepareForRead( + Configuration configuration, + Map metadata, + MessageType fileSchema, + ReadContext readContext) { + return new FileRecordMaterializer(fileSchema, readContext.getRequestedSchema()); + } + } + + private static final class FileWriteSupport extends WriteSupport { + private RecordConsumer consumer; + + @Override + public WriteContext init(Configuration configuration) { + return new WriteContext(SCHEMA, Collections.emptyMap()); + } + + @Override + public void prepareForWrite(RecordConsumer recordConsumer) { + consumer = recordConsumer; + } + + @Override + public void write(FileValue value) { + consumer.startMessage(); + if (value != null) { + consumer.startField("file", 0); + FileValueWriter.write(consumer, FILE_SCHEMA, value); + consumer.endField("file", 0); + } + consumer.endMessage(); + } + } + + private static final class FileRecordMaterializer extends RecordMaterializer { + private final GroupConverter root; + private FileRecord current; + + private FileRecordMaterializer(MessageType fileSchema, MessageType requestedSchema) { + GroupType requestedFile = requestedSchema.getType("file").asGroupType(); + GroupType storedFile = fileSchema.getType("file").asGroupType(); + GroupConverter fileConverter = FileConverters.newFileConverter( + requestedFile, storedFile, value -> current = new FileRecord(value)); + root = new GroupConverter() { + @Override + public Converter getConverter(int fieldIndex) { + return fileConverter; + } + + @Override + public void start() { + current = new FileRecord(null); + } + + @Override + public void end() {} + }; + } + + @Override + public FileRecord getCurrentRecord() { + return current; + } + + @Override + public GroupConverter getRootConverter() { + return root; + } + } +} diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java b/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java index a47166ae93..baade1a1ac 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java @@ -68,6 +68,7 @@ import java.util.Random; import java.util.Set; import java.util.TreeSet; +import java.util.stream.Stream; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.parquet.FixedBinaryTestUtils; @@ -144,6 +145,8 @@ import org.apache.parquet.schema.Types; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; public class TestParquetMetadataConverter { private static SecureRandom random = new SecureRandom(); @@ -574,12 +577,10 @@ public void testLogicalToConvertedTypeConversion() { } @Test - public void testFileLogicalTypeIsIgnoredRatherThanFailing() { + public void testFileLogicalTypeIsConverted() { ParquetMetadataConverter converter = new ParquetMetadataConverter(); - // FILE has no LogicalTypeAnnotation yet, so it must degrade to the physical type the way an - // unrecognised logical type does, rather than throwing. assertThat(converter.getLogicalTypeAnnotation(LogicalType.FILE(new FileType()))) - .isNull(); + .isEqualTo(LogicalTypeAnnotation.fileType()); } @Test @@ -2449,6 +2450,125 @@ public void testColumnIndexNanCountsRoundTrip() { assertThat(roundTrip.getNanCounts()).containsExactly(1L, 0L, 0L); } + @Test + public void testFileLogicalType() { + ParquetMetadataConverter parquetMetadataConverter = new ParquetMetadataConverter(); + + MessageType expected = Types.buildMessage() + .requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(PrimitiveTypeName.INT64) + .named("offset") + .optional(PrimitiveTypeName.INT64) + .named("size") + .optional(PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .optional(PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("checksum") + .optional(PrimitiveTypeName.BINARY) + .named("inline") + .named("f") + .named("example"); + + List parquetSchema = parquetMetadataConverter.toParquetSchema(expected); + MessageType schema = parquetMetadataConverter.fromParquetSchema(parquetSchema, null); + assertThat(schema).isEqualTo(expected); + LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); + assertThat(logicalType).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(logicalType).isEqualTo(LogicalTypeAnnotation.fileType()); + } + + @Test + public void testFileLogicalTypeRoundTripUriOnly() { + ParquetMetadataConverter parquetMetadataConverter = new ParquetMetadataConverter(); + + MessageType expected = Types.buildMessage() + .requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .named("f") + .named("example"); + + List parquetSchema = parquetMetadataConverter.toParquetSchema(expected); + MessageType schema = parquetMetadataConverter.fromParquetSchema(parquetSchema, null); + assertThat(schema).isEqualTo(expected); + LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); + assertThat(logicalType).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + } + + @ParameterizedTest + @MethodSource("malformedFileFields") + public void testMalformedFileLogicalTypeFailsOnRead(List fields) + throws IOException { + ParquetMetadataConverter parquetMetadataConverter = new ParquetMetadataConverter(); + MessageType physicalSchema = Types.buildMessage() + .requiredGroup() + .id(17) + .addFields(fields.toArray(new org.apache.parquet.schema.Type[0])) + .named("f") + .required(PrimitiveTypeName.INT32) + .named("sibling") + .named("example"); + List parquetSchema = parquetMetadataConverter.toParquetSchema(physicalSchema); + parquetSchema.get(1).setLogicalType(LogicalType.FILE(new FileType())); + FileMetaData footer = new FileMetaData(1, parquetSchema, 0, emptyList()); + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + Util.writeFileMetaData(footer, bytes); + + FileMetaData storedFooter = Util.readFileMetaData(new ByteArrayInputStream(bytes.toByteArray())); + + assertThatThrownBy(() -> parquetMetadataConverter.fromParquetMetadata(storedFooter)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageStartingWith("FILE type "); + } + + @Test + public void testFileLogicalTypeOnPrimitiveIsIgnored() { + ParquetMetadataConverter converter = new ParquetMetadataConverter(); + MessageType physicalSchema = Types.buildMessage() + .optional(PrimitiveTypeName.BINARY) + .columnOrder(org.apache.parquet.schema.ColumnOrder.undefined()) + .named("file") + .named("example"); + List parquetSchema = converter.toParquetSchema(physicalSchema); + parquetSchema.get(1).setLogicalType(LogicalType.FILE(new FileType())); + + assertThat(converter.fromParquetSchema(parquetSchema, null)).isEqualTo(physicalSchema); + } + + private static Stream> malformedFileFields() { + PrimitiveType uri = + Types.optional(PrimitiveTypeName.BINARY).as(stringType()).named("uri"); + return Stream.of( + List.of(Types.optional(PrimitiveTypeName.BINARY).named("uri")), + List.of(Types.optional(PrimitiveTypeName.INT32).named("uri")), + List.of(Types.required(PrimitiveTypeName.BINARY) + .as(stringType()) + .named("uri")), + List.of(Types.optionalGroup() + .optional(PrimitiveTypeName.BINARY) + .named("nested") + .named("uri")), + List.of(Types.optional(PrimitiveTypeName.BINARY).named("unknown")), + List.of(Types.optional(PrimitiveTypeName.INT64) + .as(timestampType(true, MICROS)) + .named("offset")), + List.of(Types.optional(PrimitiveTypeName.INT64) + .as(intType(64, false)) + .named("size")), + List.of(Types.optional(PrimitiveTypeName.BINARY) + .as(stringType()) + .named("inline")), + List.of(uri, uri)); + } + @Test public void testV2StatsDoNotTriggerCorruptStatisticsCheck() { // Regression test: when V2 stats (min_value/max_value) are present,