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..286a20b0dd 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 @@ -188,6 +188,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 +384,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 +1239,89 @@ 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, elsewhere within the current file, + * 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}). The group may contain the following fields, identified + * by name: + *
    + *
  • {@code path} (STRING): an opaque path that identifies an external file, for example a + * URI such as s3://bucket/key. If not set, the value refers to the current file (a + * self-reference).
    • + *
  • {@code offset} (INT64): start of the byte range within the referenced data; if not set, + * treated as 0.
    • + *
  • {@code size} (INT64): byte length of the referenced data. Must be set whenever + * {@code offset} is set or {@code path} is not set; may be omitted only for a whole-file + * external reference, in which case the range runs to the end of the referenced file.
    • + *
  • {@code content_type} (STRING): the media (MIME) type of the resolved bytes.
    • + *
  • {@code checksum} (STRING): an algorithm-tagged integrity token for the resolved bytes, + * of the form {@code :base64()}.
    • + *
  • {@code inline} (BYTE_ARRAY): the referenced bytes stored inline in the value.
    • + *
+ * No fields with names other than the above are permitted. The schema builder additionally + * rejects group definitions that could never produce a valid value: a group that declares + * {@code offset} must also declare {@code size}, and a group must declare at least one of + * {@code inline}, {@code path}, or {@code size} (a group without {@code path} or {@code inline} + * holds only self-references, which require {@code size}). Per-value rules that depend on the + * data in each row — {@code size} being present for a self-reference (null {@code path}) and + * {@code offset}/{@code size} being non-negative — cannot be enforced here and are the + * responsibility of writers and consumers. + */ + public static class FileLogicalTypeAnnotation extends LogicalTypeAnnotation { + private static final FileLogicalTypeAnnotation INSTANCE = new FileLogicalTypeAnnotation(); + + /** Field name holding the path/URI of an external file. */ + public static final String PATH_FIELD = "path"; + + /** 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( + PATH_FIELD, OFFSET_FIELD, SIZE_FIELD, CONTENT_TYPE_FIELD, CHECKSUM_FIELD, INLINE_FIELD); + + private FileLogicalTypeAnnotation() {} + + @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 +1527,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/main/java/org/apache/parquet/schema/Types.java b/parquet-column/src/main/java/org/apache/parquet/schema/Types.java index 2f12991ab0..95e3a0163c 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/Types.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/Types.java @@ -821,12 +821,65 @@ public THIS addFields(Type... types) { @Override protected GroupType build(String name) { if (newLogicalTypeSet) { + if (logicalTypeAnnotation instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation) { + validateFileTypeFields(name, fields); + } return new GroupType(repetition, name, logicalTypeAnnotation, fields, id); } else { return new GroupType(repetition, name, getOriginalType(), fields, id); } } + private static void validateFileTypeFields(String name, List fields) { + boolean hasPath = false; + boolean hasOffset = false; + boolean hasSize = false; + boolean hasInline = false; + for (Type field : fields) { + String fieldName = field.getName(); + if (!LogicalTypeAnnotation.FileLogicalTypeAnnotation.FIELD_NAMES.contains(fieldName)) { + throw new IllegalArgumentException("FILE type group '" + name + "' contains unrecognized field '" + + fieldName + "'. Valid fields are: " + + String.join(", ", LogicalTypeAnnotation.FileLogicalTypeAnnotation.FIELD_NAMES)); + } + Preconditions.checkArgument( + field.isPrimitive() && field.getRepetition() == Type.Repetition.OPTIONAL, + "FILE type field '%s' must be an optional primitive in group '%s'", + fieldName, + name); + if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.PATH_FIELD.equals(fieldName)) { + hasPath = true; + } else if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.OFFSET_FIELD.equals(fieldName)) { + hasOffset = true; + } else if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.SIZE_FIELD.equals(fieldName)) { + hasSize = true; + } else if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.INLINE_FIELD.equals(fieldName)) { + hasInline = true; + } + } + // The spec requires `size` to be set whenever `offset` is set. A group that declares + // `offset` but not `size` can never produce a valid value, so reject it at schema-build + // time. + Preconditions.checkArgument( + !hasOffset || hasSize, + "FILE type group '%s' declares field 'offset' but not 'size'; 'size' is required whenever 'offset' is set", + name); + // The spec requires `size` to be set whenever `path` is not set (a self-reference). A group + // that declares neither `path` nor `inline` can only hold self-references, so it must + // declare `size`. More generally, a value can only resolve to bytes via `inline`, `path`, + // or `size`, so a group that declares none of these can never produce a valid value. + Preconditions.checkArgument( + hasInline || hasPath || hasSize, + "FILE type group '%s' must declare at least one of 'inline', 'path', or 'size'; a group " + + "without 'path' or 'inline' holds only self-references, which require 'size'", + name); + // The remaining spec rules are per-value constraints that the schema builder cannot verify + // because it sees only which fields are declared, not their values in each row: when `path` + // is null in a row that value is a self-reference and must carry a non-null `size`, and + // `offset`/`size` must be non-negative. Those are the responsibility of writers and + // consumers of FILE values. + } + public MapBuilder map(Type.Repetition repetition) { return new MapBuilder<>(self()).repetition(repetition); } 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 61fe3065e1..ca41f4aef2 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 @@ -528,6 +528,182 @@ public void testVariantLogicalTypeWithShredded() { assertEquals(specVersion, ((LogicalTypeAnnotation.VariantLogicalTypeAnnotation) annotation).getSpecVersion()); } + @Test + public void testFileLogicalTypePathOnly() { + String name = "file_field"; + GroupType file = new GroupType( + REQUIRED, + name, + LogicalTypeAnnotation.fileType(), + Types.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path")); + + assertEquals( + "required group file_field (FILE) {\n" + + " optional binary path (STRING);\n" + + "}", + file.toString()); + + LogicalTypeAnnotation annotation = file.getLogicalTypeAnnotation(); + assertEquals(LogicalTypeAnnotation.LogicalTypeToken.FILE, annotation.getType()); + assertNull(annotation.toOriginalType()); + assertTrue(annotation instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + } + + @Test + public void testFileLogicalTypeAllFields() { + String name = "file_field"; + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path") + .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(); + assertTrue(annotation instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(6, file.getFieldCount()); + assertEquals("path", file.getType("path").getName()); + assertEquals("offset", file.getType("offset").getName()); + assertEquals("size", file.getType("size").getName()); + assertEquals("content_type", file.getType("content_type").getName()); + assertEquals("checksum", file.getType("checksum").getName()); + assertEquals("inline", file.getType("inline").getName()); + } + + @Test + public void testFileLogicalTypeInlineOnly() { + // Every field is optional, so an inline-only group is valid (spec self-reference / inline case). + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).named("inline") + .named("inline_file"); + + assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(1, file.getFieldCount()); + assertEquals("inline", file.getType("inline").getName()); + } + + @Test + public void testFileLogicalTypeSelfReference() { + // A self-reference omits 'path' and locates bytes within the current file via offset/size. + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64).named("offset") + .optional(INT64).named("size") + .named("self_ref_file"); + + assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(2, file.getFieldCount()); + } + + @Test + public void testFileLogicalTypeSelfReferenceRequiresSize() { + // A group without 'path' or 'inline' can only hold self-references, which require 'size'. + // Declaring only metadata fields leaves no way to resolve or size the referenced bytes. + assertThrows( + "FILE type group without 'path'/'inline' must declare 'size'", + IllegalArgumentException.class, + () -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("content_type") + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("checksum") + .named("file_metadata_only")); + } + + @Test + public void testFileLogicalTypeSelfReferenceWithSize() { + // A self-reference (no 'path') that declares 'size' is valid. + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64).named("size") + .named("self_ref_with_size"); + + assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(1, file.getFieldCount()); + } + + @Test + public void testFileLogicalTypeOffsetRequiresSize() { + // The spec requires 'size' whenever 'offset' is set, so a group declaring 'offset' + // without 'size' can never produce a valid value and is rejected at build time. + assertThrows( + "FILE type group with 'offset' must also declare 'size'", + IllegalArgumentException.class, + () -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path") + .optional(INT64).named("offset") + .named("file_offset_without_size")); + } + + @Test + public void testFileLogicalTypeOffsetWithSize() { + // 'offset' accompanied by 'size' is valid. + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path") + .optional(INT64).named("offset") + .optional(INT64).named("size") + .named("file_offset_with_size"); + + assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(3, file.getFieldCount()); + } + + @Test + public void testFileLogicalTypeSizeWithoutOffset() { + // 'size' without 'offset' is valid (e.g. a whole-file range starting at 0). + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path") + .optional(INT64).named("size") + .named("file_size_without_offset"); + + assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(2, file.getFieldCount()); + } + + @Test + public void testFileLogicalTypeRejectsUnrecognizedField() { + assertThrows( + "FILE type group must not contain unrecognized field names", + IllegalArgumentException.class, + () -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path") + .optional(BINARY).named("unknown_field") + .named("file_with_bad_field")); + } + + @Test + public void testFileLogicalTypeRejectsRequiredField() { + // All FILE fields must have OPTIONAL repetition under the current spec. + assertThrows( + "FILE type field must be optional", + IllegalArgumentException.class, + () -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .required(BINARY).as(LogicalTypeAnnotation.stringType()).named("path") + .named("file_with_required_path")); + } + + @Test + public void testFileLogicalTypeRejectsGroupField() { + // FILE fields must be primitives, not nested groups. + assertThrows( + "FILE type field must be primitive", + IllegalArgumentException.class, + () -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optionalGroup() + .optional(BINARY).named("nested") + .named("path") + .named("file_with_group_field")); + } + /** * A convenience method to avoid a large number of @Test(expected=...) tests * 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 50c2e344e2..4ef3b0576f 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 @@ -111,6 +111,7 @@ import org.apache.parquet.format.Type; import org.apache.parquet.format.TypeDefinedOrder; import org.apache.parquet.format.Uncompressed; +import org.apache.parquet.format.FileType; import org.apache.parquet.format.VariantType; import org.apache.parquet.format.XxHash; import org.apache.parquet.hadoop.metadata.BlockMetaData; @@ -591,6 +592,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( @@ -1386,6 +1392,8 @@ LogicalTypeAnnotation getLogicalTypeAnnotation(LogicalType type) { case VARIANT: VariantType variant = type.getVARIANT(); return LogicalTypeAnnotation.variantType(variant.getSpecification_version()); + case FILE: + return LogicalTypeAnnotation.fileType(); default: throw new RuntimeException("Unknown logical type " + type); } 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 8d778f7b91..9a8540b57c 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 @@ -2159,4 +2159,57 @@ public void testColumnIndexNanCountsRoundTrip() { assertNotNull(roundTrip); assertEquals(List.of(1L, 0L, 0L), roundTrip.getNanCounts()); } + + @Test + public void testFileLogicalType() { + ParquetMetadataConverter parquetMetadataConverter = new ParquetMetadataConverter(); + + MessageType expected = Types.buildMessage() + .requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("path") + .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); + assertEquals(expected, schema); + LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); + assertTrue(logicalType instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(LogicalTypeAnnotation.fileType(), logicalType); + } + + @Test + public void testFileLogicalTypeRoundTripPathOnly() { + ParquetMetadataConverter parquetMetadataConverter = new ParquetMetadataConverter(); + + MessageType expected = Types.buildMessage() + .requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("path") + .named("f") + .named("example"); + + List parquetSchema = parquetMetadataConverter.toParquetSchema(expected); + MessageType schema = parquetMetadataConverter.fromParquetSchema(parquetSchema, null); + assertEquals(expected, schema); + LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); + assertTrue(logicalType instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + } }