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..d77ef8b6ab 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,100 @@ 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}). 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: + *
    + *
  • {@code uri} (STRING): a URI-reference (RFC 3986) that identifies an external file, for + * example {@code s3://bucket/file.jpg}. 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. Must not be negative.
    • + *
  • {@code size} (INT64): byte length of the referenced data. Must be set whenever + * {@code offset} is set or {@code uri} 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. Must + * not be negative.
    • + *
  • {@code content_type} (STRING): the media (MIME) type (RFC 2046) of the resolved bytes; + * when not set, {@code application/octet-stream} is assumed.
    • + *
  • {@code checksum} (STRING): a self-describing integrity token for the resolved bytes, of + * the form {@code :}.
    • + *
  • {@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 uri}, or {@code offset} (a value resolves to bytes only via one of + * these; a group declaring none of them — even if it declares {@code size} — can never produce a + * resolvable value). A group that declares {@code offset} but not {@code uri} permits only + * self-references (a value with {@code uri} unset that locates bytes within the current file) and + * must therefore also declare {@code inline}: the {@code inline} column chunk of the same row + * group is the reference point whose compression and encryption a self-reference inherits. A + * group that declares {@code uri} is treated as an external-reference schema and is not required + * to declare {@code inline}. Each declared field must also match its required physical type. + * Per-value + * rules that depend on the data in each row — {@code offset} being set for a self-reference + * (unset {@code uri}), {@code size} being set whenever {@code offset} is set, 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 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() {} + + @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 +1538,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..0bd8fd480b 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,123 @@ 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 hasUri = 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); + validateFileTypeFieldPhysicalType(name, field.asPrimitiveType()); + if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.URI_FIELD.equals(fieldName)) { + hasUri = 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); + // Per the spec resolution table, a value resolves to bytes only if `inline`, `uri`, or + // `offset` is set; `size` on its own never resolves. A group that declares none of `inline`, + // `uri`, or `offset` can therefore never produce a resolvable value, so reject it at + // schema-build time. + Preconditions.checkArgument( + hasInline || hasUri || hasOffset, + "FILE type group '%s' must declare at least one of 'inline', 'uri', or 'offset'; a value " + + "resolves to bytes only via one of these, so a group declaring none of them can " + + "never produce a valid value", + name); + // A schema that permits self-references must declare `inline`. A self-reference (`uri` not + // set) always sets `offset`, and the `inline` column chunk of the same row group is the + // reference point whose compression and encryption a self-reference inherits. A group that + // declares `offset` but not `uri` can only produce self-references (an offset-based read with + // no `uri` is a self-reference), so it must also declare `inline`. A group that declares + // `uri` is not required to declare `inline`: `offset`/`size` there describe an external + // ranged reference, and although the per-value `uri` could be left unset in some rows, the + // schema is treated as an external-reference schema and the `inline` requirement is not + // imposed. + Preconditions.checkArgument( + !(hasOffset && !hasUri) || hasInline, + "FILE type group '%s' declares field 'offset' but neither 'uri' nor 'inline'; a schema " + + "that permits self-references (offset without uri) must declare 'inline' as the " + + "reference point for storage inheritance", + 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: a + // self-reference (unset `uri`) must set `offset`, `size` must be set whenever `offset` is + // set, and `offset`/`size` must be non-negative. Those are the responsibility of writers and + // consumers of FILE values. + } + + /** + * Validates that a declared FILE field uses the physical type required by the spec: + * {@code uri}, {@code content_type}, and {@code checksum} are STRING (BINARY), {@code offset} + * and {@code size} are INT64, and {@code inline} is BYTE_ARRAY (BINARY). + */ + private static void validateFileTypeFieldPhysicalType(String name, PrimitiveType field) { + String fieldName = field.getName(); + PrimitiveType.PrimitiveTypeName physicalType = field.getPrimitiveTypeName(); + switch (fieldName) { + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.URI_FIELD: + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CONTENT_TYPE_FIELD: + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CHECKSUM_FIELD: + Preconditions.checkArgument( + physicalType == PrimitiveType.PrimitiveTypeName.BINARY + && field.getLogicalTypeAnnotation() + instanceof LogicalTypeAnnotation.StringLogicalTypeAnnotation, + "FILE type field '%s' must be a STRING (BINARY annotated as STRING) in group '%s'", + fieldName, + name); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.OFFSET_FIELD: + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.SIZE_FIELD: + Preconditions.checkArgument( + physicalType == PrimitiveType.PrimitiveTypeName.INT64, + "FILE type field '%s' must be an INT64 in group '%s'", + fieldName, + name); + break; + case LogicalTypeAnnotation.FileLogicalTypeAnnotation.INLINE_FIELD: + Preconditions.checkArgument( + physicalType == PrimitiveType.PrimitiveTypeName.BINARY, + "FILE type field '%s' must be a BYTE_ARRAY (BINARY) in group '%s'", + fieldName, + name); + break; + default: + // Unreachable: field names are validated against FIELD_NAMES before this call. + break; + } + } + 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 0d7791a19b..a02b6bd610 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 @@ -549,4 +549,241 @@ 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 testFileLogicalTypeSelfReference() { + // A self-reference omits 'uri' and locates bytes within the current file via offset/size. + // A schema that permits self-references must declare 'inline' as the storage-inheritance + // reference point. + GroupType file = Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64).named("offset") + .optional(INT64).named("size") + .optional(BINARY).named("inline") + .named("self_ref_file"); + + assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(file.getFieldCount()).isEqualTo(3); + } + + @Test + public void testFileLogicalTypeOffsetRequiresInline() { + // A schema that permits self-references (declares 'offset' but not 'uri') must declare 'inline' + // as the reference point for storage inheritance. A group declaring 'offset'/'size' with + // neither 'uri' nor 'inline' is rejected at build time. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64).named("offset") + .optional(INT64).named("size") + .named("self_ref_without_inline")) + .isInstanceOf(IllegalArgumentException.class); + } + + @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 testFileLogicalTypeMetadataOnlyRejected() { + // Per the spec resolution table, a value resolves to bytes only via 'inline', 'uri', or + // 'offset'. A group declaring only metadata fields can never produce a resolvable value. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("content_type") + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("checksum") + .named("file_metadata_only")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testFileLogicalTypeSizeOnlyRejected() { + // 'size' alone never resolves to bytes (spec resolution table), so a size-only group is + // rejected: it declares no locator ('inline', 'uri', or 'offset'). + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(INT64).named("size") + .named("file_size_only")) + .isInstanceOf(IllegalArgumentException.class); + } + + @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. + assertThatThrownBy(() -> Types.requiredGroup() + .as(LogicalTypeAnnotation.fileType()) + .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri") + .optional(INT64).named("offset") + .named("file_offset_without_size")) + .isInstanceOf(IllegalArgumentException.class); + } + + @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); + } } 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/crypto/AesCipher.java b/parquet-hadoop/src/main/java/org/apache/parquet/crypto/AesCipher.java index 2386412b65..e70e8658b9 100755 --- a/parquet-hadoop/src/main/java/org/apache/parquet/crypto/AesCipher.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/crypto/AesCipher.java @@ -120,6 +120,53 @@ public static byte[] createFooterAAD(byte[] aadPrefixBytes) { return createModuleAAD(aadPrefixBytes, ModuleType.Footer, -1, -1, -1); } + /** + * Builds the module AAD for a self-reference (FILE self-reference payload). Unlike pages, which + * are identified by a 2-byte page ordinal, a self-reference is identified by an 8-byte + * little-endian self-reference ordinal that follows the row group and column ordinals. The column + * ordinal is that of the {@code inline} column whose encryption the self-reference inherits. + * + * @param fileAAD the file AAD (AAD prefix concatenated with the AAD file-unique bytes) + * @param rowGroupOrdinal the row group ordinal of the self-reference + * @param columnOrdinal the ordinal of the {@code inline} column the self-reference inherits from + * @param selfReferenceOrdinal the zero-based self-reference ordinal within the column chunk + * @return the module AAD bytes + */ + public static byte[] createSelfReferenceAAD( + byte[] fileAAD, int rowGroupOrdinal, int columnOrdinal, long selfReferenceOrdinal) { + + byte[] typeOrdinalBytes = new byte[1]; + typeOrdinalBytes[0] = ModuleType.SelfReference.getValue(); + + if (rowGroupOrdinal < 0) { + throw new IllegalArgumentException("Wrong row group ordinal: " + rowGroupOrdinal); + } + short shortRGOrdinal = (short) rowGroupOrdinal; + if (shortRGOrdinal != rowGroupOrdinal) { + throw new ParquetCryptoRuntimeException("Encrypted parquet files can't have " + "more than " + + Short.MAX_VALUE + " row groups: " + rowGroupOrdinal); + } + byte[] rowGroupOrdinalBytes = shortToBytesLE(shortRGOrdinal); + + if (columnOrdinal < 0) { + throw new IllegalArgumentException("Wrong column ordinal: " + columnOrdinal); + } + short shortColumnOrdinal = (short) columnOrdinal; + if (shortColumnOrdinal != columnOrdinal) { + throw new ParquetCryptoRuntimeException("Encrypted parquet files can't have " + "more than " + + Short.MAX_VALUE + " columns: " + columnOrdinal); + } + byte[] columnOrdinalBytes = shortToBytesLE(shortColumnOrdinal); + + if (selfReferenceOrdinal < 0) { + throw new IllegalArgumentException("Wrong self-reference ordinal: " + selfReferenceOrdinal); + } + byte[] selfReferenceOrdinalBytes = longToBytesLE(selfReferenceOrdinal); + + return concatByteArrays( + fileAAD, typeOrdinalBytes, rowGroupOrdinalBytes, columnOrdinalBytes, selfReferenceOrdinalBytes); + } + // Update last two bytes with new page ordinal (instead of creating new page AAD from scratch) public static void quickUpdatePageAAD(byte[] pageAAD, int newPageOrdinal) { java.util.Objects.requireNonNull(pageAAD); @@ -159,4 +206,13 @@ private static byte[] shortToBytesLE(short input) { return output; } + + private static byte[] longToBytesLE(long input) { + byte[] output = new byte[8]; + for (int i = 0; i < 8; i++) { + output[i] = (byte) (0xff & (input >> (8 * i))); + } + + return output; + } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/crypto/ModuleCipherFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/crypto/ModuleCipherFactory.java index 9d258e2825..94c9c68097 100755 --- a/parquet-hadoop/src/main/java/org/apache/parquet/crypto/ModuleCipherFactory.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/crypto/ModuleCipherFactory.java @@ -34,7 +34,8 @@ public enum ModuleType { ColumnIndex((byte) 6), OffsetIndex((byte) 7), BloomFilterHeader((byte) 8), - BloomFilterBitset((byte) 9); + BloomFilterBitset((byte) 9), + SelfReference((byte) 10); private final byte value; 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 465516e48f..30cac68e28 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( @@ -1389,6 +1395,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/main/java/org/apache/parquet/hadoop/CodecFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java index c9391201f4..71ca455e6e 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java @@ -281,6 +281,48 @@ public BytesCompressor getCompressor(CompressionCodecName codecName, int level) return comp; } + /** + * Decompresses a complete compression block whose decompressed size is not known in advance, + * draining the codec stream to end-of-input. This is used to resolve FILE self-references, whose + * stored representation records only the size of the (compressed) stored block and not the size + * of the resolved bytes. Each self-reference is an independent compression block, so the entire + * {@code compressed} range is supplied to the codec in one shot. + * + * @param codecName the {@link CompressionCodecName} of the {@code inline} column chunk the + * self-reference inherits from; {@link CompressionCodecName#UNCOMPRESSED} returns the bytes + * unchanged + * @param compressed the complete compressed block + * @return the decompressed (resolved) bytes + * @throws IOException if decompression fails + */ + public BytesInput decompressUnknownSize(CompressionCodecName codecName, BytesInput compressed) + throws IOException { + CompressionCodec codec = getCodec(codecName); + if (codec == null) { + // UNCOMPRESSED: the stored bytes are the resolved bytes. + return compressed; + } + Decompressor decompressor = CodecPool.getDecompressor(codec); + try { + if (decompressor != null) { + decompressor.reset(); + } + try (InputStream is = codec.createInputStream(compressed.toInputStream(), decompressor); + ByteArrayOutputStream out = new ByteArrayOutputStream()) { + byte[] buffer = new byte[8192]; + int read; + while ((read = is.read(buffer)) != -1) { + out.write(buffer, 0, read); + } + return BytesInput.from(out.toByteArray()); + } + } finally { + if (decompressor != null) { + CodecPool.returnDecompressor(decompressor); + } + } + } + @Override public BytesDecompressor getDecompressor(CompressionCodecName codecName) { BytesDecompressor decomp = decompressors.get(codecName); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index 9af4b4ac60..5cfaae4ef2 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -78,6 +78,7 @@ import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.column.values.bloomfilter.BlockSplitBloomFilter; import org.apache.parquet.column.values.bloomfilter.BloomFilter; +import org.apache.parquet.compression.CompressionCodecFactory; import org.apache.parquet.compression.CompressionCodecFactory.BytesInputDecompressor; import org.apache.parquet.crypto.AesCipher; import org.apache.parquet.crypto.FileDecryptionProperties; @@ -1070,6 +1071,65 @@ public String getFile() { return file.toString(); } + /** + * Resolves a {@code FILE} self-reference to its logical bytes. A self-reference (a {@code FILE} + * value with {@code uri} unset) records the {@code offset} and {@code size} of a stored + * representation within this file; the stored bytes inherit the compression and encryption of the + * {@code inline} column chunk in the same row group. This method reads the stored bytes, decrypts + * them when the {@code inline} column chunk is encrypted, and decompresses them with the column + * chunk's codec, returning the resolved bytes. See {@link SelfReferenceStorage} and the Parquet + * format's "FILE" logical type specification. + * + * @param inlineColumn the {@link ColumnChunkMetaData} of the {@code inline} column chunk whose + * compression and encryption the self-reference inherits + * @param offset the self-reference {@code offset} field (start of the stored representation) + * @param size the self-reference {@code size} field (byte length of the stored representation) + * @param selfReferenceOrdinal the zero-based self-reference ordinal within the column chunk + * @return the resolved (logical) bytes of the self-reference + * @throws IOException if reading or resolving fails + */ + public BytesInput resolveSelfReference( + ColumnChunkMetaData inlineColumn, long offset, long size, long selfReferenceOrdinal) throws IOException { + if (offset < 0) { + throw new IllegalArgumentException("Self-reference offset must not be negative: " + offset); + } + if (size < 0) { + throw new IllegalArgumentException("Self-reference size must not be negative: " + size); + } + + byte[] stored = new byte[Math.toIntExact(size)]; + f.seek(offset); + f.readFully(stored); + + BlockCipher.Decryptor pageDecryptor = null; + byte[] fileAAD = null; + int columnOrdinal = -1; + if (null != fileDecryptor && !fileDecryptor.plaintextFile()) { + InternalColumnDecryptionSetup columnDecryptionSetup = fileDecryptor.getColumnSetup(inlineColumn.getPath()); + if (columnDecryptionSetup.isEncrypted()) { + pageDecryptor = columnDecryptionSetup.getDataDecryptor(); + fileAAD = fileDecryptor.getFileAAD(); + columnOrdinal = columnDecryptionSetup.getOrdinal(); + } + } + + CompressionCodecFactory codecFactory = options.getCodecFactory(); + if (!(codecFactory instanceof CodecFactory)) { + throw new IllegalStateException("Resolving FILE self-references requires a CodecFactory-based " + + "codec factory but found: " + codecFactory.getClass().getName()); + } + + return SelfReferenceStorage.resolve( + BytesInput.from(stored), + inlineColumn.getCodec(), + (CodecFactory) codecFactory, + pageDecryptor, + fileAAD, + inlineColumn.getRowGroupOrdinal(), + columnOrdinal, + selfReferenceOrdinal); + } + private List filterRowGroups(List blocks) throws IOException { FilterCompat.Filter recordFilter = options.getRecordFilter(); if (FilterCompat.isFilteringRequired(recordFilter)) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java index 82f4577b83..fba8c34733 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java @@ -609,6 +609,54 @@ public InternalFileEncryptor getEncryptor() { return fileEncryptor; } + /** + * Writes a {@code FILE} self-reference payload into the file body, inheriting the compression and + * encryption of the {@code inline} column chunk, and returns the {@code offset} and {@code size} a + * writer records in the self-reference's {@code offset} and {@code size} fields. See + * {@link SelfReferenceStorage} for the layout and the Parquet format's "FILE" logical type + * specification for the storage-inheritance semantics. + * + *

The payload is compressed as an independent compression block using {@code compressor} (the + * compressor for the {@code inline} column chunk's {@link CompressionCodecName}) and, when + * {@code pageBlockEncryptor} is non-null, encrypted as an independent module with the + * {@code Self-Reference} module type. The row group ordinal is that of the block currently being + * written. + * + *

This must be called while a block is open (after {@link #startBlock(long)} and before + * {@link #endBlock()}) so that the returned offset falls within the file body. + * + * @param resolvedBytes the resolved (logical) bytes of the self-reference + * @param compressor the compressor for the {@code inline} column chunk's codec + * @param pageBlockEncryptor the data-module encryptor of the {@code inline} column chunk, or + * {@code null} if the column chunk is not encrypted + * @param columnOrdinal the ordinal of the {@code inline} column the self-reference inherits from + * @param selfReferenceOrdinal the zero-based self-reference ordinal within the column chunk + * @return the offset and size of the stored representation + * @throws IOException if writing or compression fails + */ + public SelfReferenceStorage.StoredRange writeSelfReference( + BytesInput resolvedBytes, + CodecFactory.BytesCompressor compressor, + BlockCipher.Encryptor pageBlockEncryptor, + int columnOrdinal, + long selfReferenceOrdinal) + throws IOException { + return withAbortOnFailure(() -> { + // The block currently being written will be assigned ordinal blocks.size() in endBlock(). + int rowGroupOrdinal = blocks.size(); + byte[] fileAAD = (null == fileEncryptor) ? null : fileEncryptor.getFileAAD(); + return SelfReferenceStorage.write( + resolvedBytes, + compressor, + pageBlockEncryptor, + fileAAD, + rowGroupOrdinal, + columnOrdinal, + selfReferenceOrdinal, + out); + }); + } + /** * start a block * diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SelfReferenceStorage.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SelfReferenceStorage.java new file mode 100644 index 0000000000..c534157864 --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SelfReferenceStorage.java @@ -0,0 +1,177 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.parquet.hadoop; + +import java.io.IOException; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.crypto.AesCipher; +import org.apache.parquet.format.BlockCipher; +import org.apache.parquet.hadoop.CodecFactory.BytesCompressor; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; + +/** + * Implements the storage-inheritance semantics for {@code FILE} self-references as specified in the + * Parquet format (see {@code LogicalTypes.md}, section "FILE"). A self-reference is a {@code FILE} + * value whose {@code uri} is not set and that locates a byte range within the same Parquet file via + * {@code offset} and {@code size}. + * + *

A self-reference does not point at the resolved (logical) bytes directly. Instead it points at + * a stored representation: the resolved bytes after being compressed and (optionally) + * encrypted, inheriting the {@link CompressionCodecName} and encryption settings of the + * {@code inline} column chunk in the same row group. Each self-reference is an independent + * compression block and (when encrypted) an independent encryption module; state is not shared with + * data pages or with other self-references. + * + *

Layout of a stored self-reference: + * + *

    + *
  • Unencrypted: the compressed block (or the raw bytes when the codec is + * {@link CompressionCodecName#UNCOMPRESSED}). {@code offset}/{@code size} cover exactly these + * bytes. + *
  • Encrypted: the modular-encryption serialization of the compressed block — a 4-byte + * little-endian length, a 12-byte nonce, the ciphertext, and (for AES_GCM_V1) a 16-byte GCM + * tag. {@code offset} points to the beginning of the 4-byte length and {@code size} covers the + * complete encrypted module. The AAD uses the {@code Self-Reference} module type (10) with an + * 8-byte self-reference ordinal; see {@link AesCipher#createSelfReferenceAAD}. + *
+ * + *

Compression is always applied before encryption on write; decryption is applied before + * decompression on read. + */ +public final class SelfReferenceStorage { + + private SelfReferenceStorage() {} + + /** + * The location of a stored self-reference within the Parquet file. The {@code offset} and + * {@code size} are exactly the values a writer records in the {@code offset} and {@code size} + * fields of the {@code FILE} group. + */ + public static final class StoredRange { + private final long offset; + private final long size; + + public StoredRange(long offset, long size) { + this.offset = offset; + this.size = size; + } + + /** The byte offset of the stored representation within the Parquet file. */ + public long getOffset() { + return offset; + } + + /** The byte length of the stored representation. */ + public long getSize() { + return size; + } + } + + /** + * Compresses (and optionally encrypts) {@code resolvedBytes} as an independent stored block and + * appends it to {@code out}, returning the {@link StoredRange} that a writer records in the + * {@code offset} and {@code size} fields of the self-reference. + * + * @param resolvedBytes the resolved (logical) bytes of the self-reference + * @param compressor the compressor for the {@code inline} column chunk's codec; must not be null + * (use the {@link CompressionCodecName#UNCOMPRESSED} compressor to store bytes uncompressed) + * @param pageBlockEncryptor the data-module encryptor of the {@code inline} column chunk, or + * {@code null} if the column chunk is not encrypted + * @param fileAAD the file AAD, required when {@code pageBlockEncryptor} is non-null + * @param rowGroupOrdinal the row group ordinal of the self-reference + * @param columnOrdinal the ordinal of the {@code inline} column the self-reference inherits from + * @param selfReferenceOrdinal the zero-based self-reference ordinal within the column chunk + * @param out the Parquet file output stream, positioned where the stored block should be written + * @return the offset and size of the stored representation + * @throws IOException if writing or compression fails + */ + public static StoredRange write( + BytesInput resolvedBytes, + BytesCompressor compressor, + BlockCipher.Encryptor pageBlockEncryptor, + byte[] fileAAD, + int rowGroupOrdinal, + int columnOrdinal, + long selfReferenceOrdinal, + org.apache.parquet.io.PositionOutputStream out) + throws IOException { + + // Step 1: compress the resolved bytes as an independent compression block. UNCOMPRESSED leaves + // the bytes unchanged (the NO_OP_COMPRESSOR returns its input). + BytesInput stored = compressor.compress(resolvedBytes); + + // Step 2: when the inline column chunk is encrypted, encrypt the compressed block as an + // independent module. The encryptor prepends the 4-byte length and the nonce and appends the + // GCM tag (for AES_GCM_V1); the returned byte array is the complete stored module. + if (pageBlockEncryptor != null) { + byte[] selfReferenceAAD = + AesCipher.createSelfReferenceAAD(fileAAD, rowGroupOrdinal, columnOrdinal, selfReferenceOrdinal); + stored = BytesInput.from(pageBlockEncryptor.encrypt(stored.toByteArray(), selfReferenceAAD)); + } + + long offset = out.getPos(); + long size = stored.size(); + stored.writeAllTo(out); + return new StoredRange(offset, size); + } + + /** + * Resolves a stored self-reference back to its logical bytes: decrypts the stored representation + * (when the {@code inline} column chunk is encrypted) and then decompresses it using the column + * chunk's codec. + * + * @param storedBytes the stored representation, i.e. the {@code [offset, offset + size)} range + * read from the Parquet file + * @param codecName the {@link CompressionCodecName} of the {@code inline} column chunk + * @param codecFactory the codec factory used to decompress the block + * @param pageBlockDecryptor the data-module decryptor of the {@code inline} column chunk, or + * {@code null} if the column chunk is not encrypted + * @param fileAAD the file AAD, required when {@code pageBlockDecryptor} is non-null + * @param rowGroupOrdinal the row group ordinal of the self-reference + * @param columnOrdinal the ordinal of the {@code inline} column the self-reference inherits from + * @param selfReferenceOrdinal the zero-based self-reference ordinal within the column chunk + * @return the resolved (logical) bytes + * @throws IOException if decompression fails + */ + public static BytesInput resolve( + BytesInput storedBytes, + CompressionCodecName codecName, + CodecFactory codecFactory, + BlockCipher.Decryptor pageBlockDecryptor, + byte[] fileAAD, + int rowGroupOrdinal, + int columnOrdinal, + long selfReferenceOrdinal) + throws IOException { + + BytesInput compressed = storedBytes; + + // Step 1: decrypt when the inline column chunk is encrypted. The decryptor consumes the 4-byte + // length, nonce, ciphertext, and GCM tag and returns the compressed block. + if (pageBlockDecryptor != null) { + byte[] selfReferenceAAD = + AesCipher.createSelfReferenceAAD(fileAAD, rowGroupOrdinal, columnOrdinal, selfReferenceOrdinal); + compressed = BytesInput.from(pageBlockDecryptor.decrypt(storedBytes.toByteArray(), selfReferenceAAD)); + } + + // Step 2: decompress. The resolved size is not stored, so the codec stream is drained to EOF. + return codecFactory.decompressUnknownSize(codecName, compressed); + } +} diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestSelfReferenceAAD.java b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestSelfReferenceAAD.java new file mode 100644 index 0000000000..2e6772e8bd --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestSelfReferenceAAD.java @@ -0,0 +1,95 @@ +/* + * 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.crypto; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import org.apache.parquet.crypto.ModuleCipherFactory.ModuleType; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for the {@code Self-Reference} module type and the {@link AesCipher#createSelfReferenceAAD} + * AAD construction defined for FILE self-references (parquet-format PR #603). + */ +public class TestSelfReferenceAAD { + + private static final byte[] FILE_AAD = new byte[] {1, 2, 3, 4, 5, 6, 7, 8}; + + @Test + public void testSelfReferenceModuleTypeValue() { + // The spec assigns module type 10 to Self-Reference. + assertThat(ModuleType.SelfReference.getValue()).isEqualTo((byte) 10); + } + + @Test + public void testSelfReferenceAADLayout() { + int rowGroupOrdinal = 3; + int columnOrdinal = 7; + long selfReferenceOrdinal = 0x0102030405060708L; + + byte[] aad = + AesCipher.createSelfReferenceAAD(FILE_AAD, rowGroupOrdinal, columnOrdinal, selfReferenceOrdinal); + + // Layout: fileAAD | moduleType(1) | rowGroupOrdinal(2 LE) | columnOrdinal(2 LE) | + // selfReferenceOrdinal(8 LE) + assertThat(aad.length).isEqualTo(FILE_AAD.length + 1 + 2 + 2 + 8); + + ByteBuffer buf = ByteBuffer.wrap(aad).order(ByteOrder.LITTLE_ENDIAN); + byte[] filePart = new byte[FILE_AAD.length]; + buf.get(filePart); + assertThat(filePart).isEqualTo(FILE_AAD); + assertThat(buf.get()).isEqualTo((byte) 10); // module type + assertThat(buf.getShort()).isEqualTo((short) rowGroupOrdinal); + assertThat(buf.getShort()).isEqualTo((short) columnOrdinal); + // The self-reference ordinal is an 8-byte little-endian integer, unlike the 2-byte page ordinal. + assertThat(buf.getLong()).isEqualTo(selfReferenceOrdinal); + } + + @Test + public void testSelfReferenceAADSupportsLargeOrdinal() { + // A self-reference ordinal can exceed the 2-byte page-ordinal range, so it must be 8 bytes. + long largeOrdinal = ((long) Short.MAX_VALUE) + 1000L; + byte[] aad = AesCipher.createSelfReferenceAAD(FILE_AAD, 0, 0, largeOrdinal); + ByteBuffer buf = ByteBuffer.wrap(aad, FILE_AAD.length + 1 + 2 + 2, 8).order(ByteOrder.LITTLE_ENDIAN); + assertThat(buf.getLong()).isEqualTo(largeOrdinal); + } + + @Test + public void testSelfReferenceAADRejectsNegativeOrdinals() { + assertThatThrownBy(() -> AesCipher.createSelfReferenceAAD(FILE_AAD, -1, 0, 0)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> AesCipher.createSelfReferenceAAD(FILE_AAD, 0, -1, 0)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> AesCipher.createSelfReferenceAAD(FILE_AAD, 0, 0, -1)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testSelfReferenceAADDiffersFromPageAAD() { + // A self-reference and a data page in the same column must not share an AAD, because the module + // type byte differs (and the ordinal width differs). + byte[] selfRefAAD = AesCipher.createSelfReferenceAAD(FILE_AAD, 1, 2, 0); + byte[] dataPageAAD = AesCipher.createModuleAAD(FILE_AAD, ModuleType.DataPage, 1, 2, 0); + assertThat(selfRefAAD).isNotEqualTo(dataPageAAD); + } +} 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 4d361d6aa0..f554157c6d 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 @@ -2279,4 +2279,57 @@ public void testColumnIndexNanCountsRoundTrip() { assertThat(roundTrip).isNotNull(); 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); + assertEquals(expected, schema); + LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); + assertTrue(logicalType instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertEquals(LogicalTypeAnnotation.fileType(), logicalType); + } + + @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); + assertEquals(expected, schema); + LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); + assertTrue(logicalType instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + } } diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceFileWrite.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceFileWrite.java new file mode 100644 index 0000000000..9966b657ec --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceFileWrite.java @@ -0,0 +1,210 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.parquet.hadoop; + +import static org.apache.parquet.column.Encoding.BIT_PACKED; +import static org.apache.parquet.column.Encoding.PLAIN; +import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.CREATE; +import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE; +import static org.apache.parquet.hadoop.ParquetWriter.MAX_PADDING_SIZE_DEFAULT; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.page.DataPage; +import org.apache.parquet.column.page.DataPageV1; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.column.page.PageReader; +import org.apache.parquet.column.statistics.Statistics; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.hadoop.util.HadoopOutputFile; +import org.apache.parquet.io.InputFile; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; +import org.apache.parquet.schema.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * End-to-end test that writes normal column data and FILE self-reference payloads into the + * same Parquet file with {@link ParquetFileWriter}, then reopens it and both reads the data pages + * back and resolves the self-references via {@link ParquetFileReader#resolveSelfReference}. This + * exercises the interaction between the storage-inheritance APIs and the ordinary file-write path: + * self-reference payloads are written into the file body while a block is open, and the + * {@code offset}/{@code size} they return would be recorded in the {@code offset} and {@code size} + * columns of a FILE group. + */ +public class TestSelfReferenceFileWrite { + + // A FILE group whose values are self-references: the inline column supplies the codec/encryption + // reference point, and offset/size locate the stored payload within this file. + private static final MessageType SCHEMA = MessageTypeParser.parseMessageType("message m {" + + " required int64 id;" + + " optional group file (FILE) {" + + " optional int64 offset;" + + " optional int64 size;" + + " optional binary inline;" + + " }" + + "}"); + + private static final ColumnDescriptor ID_COLUMN = SCHEMA.getColumnDescription(new String[] {"id"}); + // The inline column is the storage-inheritance reference point for the FILE group. + private static final ColumnDescriptor INLINE_COLUMN = + SCHEMA.getColumnDescription(new String[] {"file", "inline"}); + + private static final CompressionCodecName CODEC = CompressionCodecName.SNAPPY; + + private static final Statistics EMPTY_STATS = Statistics.getBuilderForReading( + Types.required(PrimitiveTypeName.INT64).named("id")) + .build(); + + @TempDir + java.nio.file.Path tempDir; + + @Test + public void testWriteDataAlongsideSelfReferences() throws IOException { + Configuration conf = new Configuration(); + Path path = new Path(tempDir.resolve("self_ref.parquet").toUri()); + + // Payloads that will be stored as self-references, inheriting the SNAPPY codec of the inline + // column. Made highly compressible so the stored size differs from the resolved size. + byte[][] payloads = { + repeat("hello self-reference ", 200), + repeat("second blob ", 400), + new byte[0], // empty payload is a valid self-reference + }; + + byte[] idPageBytes = {0, 1, 2, 3, 4, 5, 6, 7}; + + CodecFactory codecFactory = new CodecFactory(conf, DEFAULT_BLOCK_SIZE); + List ranges = new ArrayList<>(); + + ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, conf), SCHEMA, CREATE, DEFAULT_BLOCK_SIZE, MAX_PADDING_SIZE_DEFAULT); + + writer.start(); + writer.startBlock(payloads.length); + + // Self-references are written into the file body while the block is open. In a real writer + // these would be interleaved with the column data; the returned offset/size feed the FILE + // group's offset/size columns. + int inlineColumnOrdinal = columnOrdinalOf(INLINE_COLUMN); + for (int i = 0; i < payloads.length; i++) { + ranges.add(writer.writeSelfReference( + BytesInput.from(payloads[i]), + codecFactory.getCompressor(CODEC), + null, // unencrypted file + inlineColumnOrdinal, + i)); + } + + // Write a normal data page for the id column in the same block. + writer.startColumn(ID_COLUMN, 4, CompressionCodecName.UNCOMPRESSED); + writer.writeDataPage(4, idPageBytes.length, BytesInput.from(idPageBytes), EMPTY_STATS, PLAIN, PLAIN, PLAIN); + writer.endColumn(); + + // Write the inline column chunk so the reader has a ColumnChunkMetaData carrying the SNAPPY + // codec that the self-references inherit. (The inline values themselves are empty here because + // the payload lives in the self-reference blocks.) + writer.startColumn(INLINE_COLUMN, 0, CODEC); + BytesInput emptyInline = codecFactory.getCompressor(CODEC).compress(BytesInput.empty()); + writer.writeDataPage(0, 0, emptyInline, EMPTY_STATS, BIT_PACKED, BIT_PACKED, PLAIN); + writer.endColumn(); + + writer.endBlock(); + writer.end(new java.util.HashMap<>()); + + // The stored ranges are non-overlapping and ordered as written. + assertThat(ranges.get(0).getOffset()).isLessThan(ranges.get(1).getOffset()); + assertThat(ranges.get(0).getOffset() + ranges.get(0).getSize()) + .isLessThanOrEqualTo(ranges.get(1).getOffset()); + + // Reopen and verify both the data page and the self-references coexist and resolve correctly. + InputFile inputFile = HadoopInputFile.fromPath(path, conf); + ParquetReadOptions options = ParquetReadOptions.builder().build(); + try (ParquetFileReader reader = ParquetFileReader.open(inputFile, options)) { + ParquetMetadata footer = reader.getFooter(); + assertThat(footer.getBlocks()).hasSize(1); + BlockMetaData block = footer.getBlocks().get(0); + + // The normal id column reads back exactly as written. + ColumnChunkMetaData inlineMeta = findColumn(block, INLINE_COLUMN); + assertThat(inlineMeta.getCodec()).isEqualTo(CODEC); + + try (ParquetFileReader dataReader = ParquetFileReader.open(inputFile, options)) { + PageReadStore pages = dataReader.readNextRowGroup(); + PageReader idPages = pages.getPageReader(ID_COLUMN); + DataPage idPage = idPages.readPage(); + assertThat(((DataPageV1) idPage).getBytes().toByteArray()).isEqualTo(idPageBytes); + } + + // Each self-reference resolves back to its original payload, inheriting the inline column's + // codec. + for (int i = 0; i < payloads.length; i++) { + SelfReferenceStorage.StoredRange range = ranges.get(i); + BytesInput resolved = + reader.resolveSelfReference(inlineMeta, range.getOffset(), range.getSize(), i); + assertThat(resolved.toByteArray()).isEqualTo(payloads[i]); + } + } + + codecFactory.release(); + } + + private static int columnOrdinalOf(ColumnDescriptor column) { + List columns = SCHEMA.getColumns(); + for (int i = 0; i < columns.size(); i++) { + if (columns.get(i).equals(column)) { + return i; + } + } + throw new IllegalStateException("Column not found in schema: " + column); + } + + private static ColumnChunkMetaData findColumn(BlockMetaData block, ColumnDescriptor column) { + org.apache.parquet.hadoop.metadata.ColumnPath target = + org.apache.parquet.hadoop.metadata.ColumnPath.get(column.getPath()); + for (ColumnChunkMetaData meta : block.getColumns()) { + if (meta.getPath().equals(target)) { + return meta; + } + } + throw new IllegalStateException("Column chunk not found: " + target); + } + + private static byte[] repeat(String token, int times) { + StringBuilder sb = new StringBuilder(token.length() * times); + for (int i = 0; i < times; i++) { + sb.append(token); + } + return sb.toString().getBytes(StandardCharsets.UTF_8); + } +} diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceStorage.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceStorage.java new file mode 100644 index 0000000000..dca5753539 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceStorage.java @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.parquet.hadoop; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Arrays; +import org.apache.hadoop.conf.Configuration; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.crypto.AesCipher; +import org.apache.parquet.crypto.AesMode; +import org.apache.parquet.crypto.ModuleCipherFactory; +import org.apache.parquet.format.BlockCipher; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.io.PositionOutputStream; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +/** + * Round-trip tests for {@link SelfReferenceStorage}, the storage-inheritance engine for FILE + * self-references (parquet-format PR #603). Each test writes a stored representation and resolves it + * back, asserting the resolved bytes equal the original and that the recorded {@code offset}/ + * {@code size} cover exactly the stored bytes. + */ +public class TestSelfReferenceStorage { + + private static final int PAGE_SIZE = 64 * 1024; + // A 32-byte AES key. + private static final byte[] COLUMN_KEY = "0123456789012345".getBytes(); + private static final byte[] FILE_AAD = "unique-file-aad!".getBytes(); + + /** A simple in-memory {@link PositionOutputStream} for capturing written bytes. */ + private static final class InMemoryPositionOutputStream extends PositionOutputStream { + private final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + + @Override + public long getPos() { + return baos.size(); + } + + @Override + public void write(int b) { + baos.write(b); + } + + @Override + public void write(byte[] b, int off, int len) { + baos.write(b, off, len); + } + + byte[] toByteArray() { + return baos.toByteArray(); + } + } + + @ParameterizedTest + @EnumSource( + value = CompressionCodecName.class, + names = {"UNCOMPRESSED", "SNAPPY", "GZIP"}) + public void testRoundTripUnencrypted(CompressionCodecName codec) throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + byte[] resolved = highlyCompressiblePayload(4096); + + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + // Simulate a leading byte already in the file, so offset is non-zero. + out.write(new byte[] {(byte) 0xAB}, 0, 1); + + SelfReferenceStorage.StoredRange range = SelfReferenceStorage.write( + BytesInput.from(resolved), + codecFactory.getCompressor(codec), + null, + null, + 0, + 0, + 0L, + out); + + byte[] fileBytes = out.toByteArray(); + assertThat(range.getOffset()).isEqualTo(1L); + assertThat(range.getSize()).isEqualTo(fileBytes.length - 1L); + if (codec == CompressionCodecName.UNCOMPRESSED) { + // Uncompressed: the stored bytes are exactly the resolved bytes. + assertThat(range.getSize()).isEqualTo((long) resolved.length); + } + + byte[] stored = + Arrays.copyOfRange(fileBytes, (int) range.getOffset(), (int) (range.getOffset() + range.getSize())); + BytesInput resolvedBack = SelfReferenceStorage.resolve( + BytesInput.from(stored), codec, codecFactory, null, null, 0, 0, 0L); + + assertThat(resolvedBack.toByteArray()).isEqualTo(resolved); + codecFactory.release(); + } + + @ParameterizedTest + @EnumSource(value = AesMode.class) + public void testRoundTripEncrypted(AesMode mode) throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + byte[] resolved = highlyCompressiblePayload(4096); + CompressionCodecName codec = CompressionCodecName.SNAPPY; + long selfReferenceOrdinal = 42L; + + BlockCipher.Encryptor encryptor = ModuleCipherFactory.getEncryptor(mode, COLUMN_KEY); + + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + SelfReferenceStorage.StoredRange range = SelfReferenceStorage.write( + BytesInput.from(resolved), + codecFactory.getCompressor(codec), + encryptor, + FILE_AAD, + 1, + 2, + selfReferenceOrdinal, + out); + + byte[] fileBytes = out.toByteArray(); + assertThat(range.getOffset()).isEqualTo(0L); + assertThat(range.getSize()).isEqualTo((long) fileBytes.length); + // The stored module carries the 4-byte length prefix and 12-byte nonce (and a 16-byte GCM tag + // for GCM), so it is larger than the raw compressed payload. + int expectedOverhead = + AesCipher.NONCE_LENGTH + 4 + (mode == AesMode.GCM ? AesCipher.GCM_TAG_LENGTH : 0); + assertThat(range.getSize()).isGreaterThan((long) expectedOverhead); + + BlockCipher.Decryptor decryptor = ModuleCipherFactory.getDecryptor(mode, COLUMN_KEY); + byte[] stored = + Arrays.copyOfRange(fileBytes, (int) range.getOffset(), (int) (range.getOffset() + range.getSize())); + BytesInput resolvedBack = SelfReferenceStorage.resolve( + BytesInput.from(stored), codec, codecFactory, decryptor, FILE_AAD, 1, 2, selfReferenceOrdinal); + + assertThat(resolvedBack.toByteArray()).isEqualTo(resolved); + codecFactory.release(); + } + + @Test + public void testEmptyPayloadRoundTrip() throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + byte[] resolved = new byte[0]; + + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + SelfReferenceStorage.StoredRange range = SelfReferenceStorage.write( + BytesInput.from(resolved), + codecFactory.getCompressor(CompressionCodecName.UNCOMPRESSED), + null, + null, + 0, + 0, + 0L, + out); + + assertThat(range.getSize()).isEqualTo(0L); + BytesInput resolvedBack = SelfReferenceStorage.resolve( + BytesInput.from(new byte[0]), CompressionCodecName.UNCOMPRESSED, codecFactory, null, null, 0, 0, 0L); + assertThat(resolvedBack.toByteArray()).isEqualTo(resolved); + codecFactory.release(); + } + + private static byte[] highlyCompressiblePayload(int length) { + byte[] payload = new byte[length]; + for (int i = 0; i < length; i++) { + payload[i] = (byte) (i % 16); + } + return payload; + } +}