diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java index 8fe45e01ef..0df47bcde7 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java @@ -68,6 +68,13 @@ public class ParquetProperties { public static final boolean DEFAULT_STATISTICS_ENABLED = true; public static final boolean DEFAULT_SIZE_STATISTICS_ENABLED = true; + /** + * Payload size at or below which a {@code FILE} value is stored inline rather than as a + * self-reference. Defaults to the page size: a payload that would fill a page on its own is + * better kept out of the column chunk. + */ + public static final int DEFAULT_FILE_SELF_REFERENCE_THRESHOLD = DEFAULT_PAGE_SIZE; + public static final boolean DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED = true; /** @@ -138,6 +145,7 @@ public static WriterVersion fromString(String name) { private final ColumnProperty sizeStatistics; private final ColumnProperty columnCodecs; private final ColumnProperty columnCompressionLevels; + private final int fileSelfReferenceThreshold; private ParquetProperties(Builder builder) { this.pageSizeThreshold = builder.pageSize; @@ -172,6 +180,7 @@ private ParquetProperties(Builder builder) { this.sizeStatistics = builder.sizeStatistics.build(); this.columnCodecs = builder.columnCodecs.build(); this.columnCompressionLevels = builder.columnCompressionLevels.build(); + this.fileSelfReferenceThreshold = builder.fileSelfReferenceThreshold; } public static Builder builder() { @@ -345,6 +354,14 @@ public int getMaxBloomFilterBytes() { return maxBloomFilterBytes; } + /** + * @return the payload size at or below which a {@code FILE} value is stored inline rather than as + * a self-reference + */ + public int getFileSelfReferenceThreshold() { + return fileSelfReferenceThreshold; + } + public boolean getAdaptiveBloomFilterEnabled(ColumnDescriptor column) { return adaptiveBloomFilterEnabled.getValue(column); } @@ -415,7 +432,8 @@ public String toString() { + "Page row count limit to " + getPageRowCountLimit() + '\n' + "Writing page checksums is: " + (getPageWriteChecksumEnabled() ? "on" : "off") + '\n' + "Statistics enabled: " + statisticsEnabled + '\n' - + "Size statistics enabled: " + sizeStatisticsEnabled; + + "Size statistics enabled: " + sizeStatisticsEnabled + '\n' + + "FILE self-reference threshold is: " + getFileSelfReferenceThreshold(); String perColumn = ""; if (!columnCodecs.toString().equals(Objects.toString(columnCodecs.getDefaultValue()))) { perColumn = "Per-column codecs: " + columnCodecs; @@ -460,6 +478,7 @@ public static class Builder { private final ColumnProperty.Builder sizeStatistics; private final ColumnProperty.Builder columnCodecs; private final ColumnProperty.Builder columnCompressionLevels; + private int fileSelfReferenceThreshold = DEFAULT_FILE_SELF_REFERENCE_THRESHOLD; private Builder() { enableDict = ColumnProperty.builder().withDefaultValue(DEFAULT_IS_DICTIONARY_ENABLED); @@ -511,6 +530,7 @@ private Builder(ParquetProperties toCopy) { this.sizeStatisticsEnabled = toCopy.sizeStatisticsEnabled; this.columnCodecs = ColumnProperty.builder(toCopy.columnCodecs); this.columnCompressionLevels = ColumnProperty.builder(toCopy.columnCompressionLevels); + this.fileSelfReferenceThreshold = toCopy.fileSelfReferenceThreshold; } /** @@ -657,6 +677,27 @@ public Builder withStatisticsTruncateLength(int length) { return this; } + /** + * Set the payload size at or below which a {@code FILE} value is stored inline rather than as a + * self-reference. + * + *

Small payloads are cheaper to keep in the column chunk, where they are read as part of the + * ordinary page stream. Large ones are better stored out of line as self-references, so that + * reading the surrounding columns does not pull the payload bytes along with them. Set to 0 to + * store every payload as a self-reference, or to {@link Integer#MAX_VALUE} to always inline. + * + * @param fileSelfReferenceThreshold the inline size limit in bytes; must not be negative + * @return this builder for method chaining + */ + public Builder withFileSelfReferenceThreshold(int fileSelfReferenceThreshold) { + Preconditions.checkArgument( + fileSelfReferenceThreshold >= 0, + "Invalid FILE self-reference threshold (negative): %s", + fileSelfReferenceThreshold); + this.fileSelfReferenceThreshold = fileSelfReferenceThreshold; + return this; + } + /** * Set max Bloom filter bytes for related columns. * 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 d77ef8b6ab..9601e3eb84 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 @@ -1302,8 +1302,8 @@ public static class FileLogicalTypeAnnotation extends LogicalTypeAnnotation { 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); + public static final Set FIELD_NAMES = + Set.of(URI_FIELD, OFFSET_FIELD, SIZE_FIELD, CONTENT_TYPE_FIELD, CHECKSUM_FIELD, INLINE_FIELD); private FileLogicalTypeAnnotation() {} 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 0bd8fd480b..5c40556d16 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 @@ -883,7 +883,9 @@ private static void validateFileTypeFields(String name, List fields) { // `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. + // imposed. A writer must therefore not emit a self-reference under such a schema, since there + // would be no `inline` column chunk to inherit compression and encryption from; that is + // enforced on the write path rather than here. Preconditions.checkArgument( !(hasOffset && !hasUri) || hasInline, "FILE type group '%s' declares field 'offset' but neither 'uri' nor 'inline'; a schema " 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 a02b6bd610..1a619ce9b8 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 @@ -560,9 +560,7 @@ public void testFileLogicalTypeUriOnly() { Types.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")); assertThat(file.toString()) - .isEqualTo("required group file_field (FILE) {\n" - + " optional binary uri (STRING);\n" - + "}"); + .isEqualTo("required group file_field (FILE) {\n" + " optional binary uri (STRING);\n" + "}"); LogicalTypeAnnotation annotation = file.getLogicalTypeAnnotation(); assertThat(annotation.getType()).isEqualTo(LogicalTypeAnnotation.LogicalTypeToken.FILE); @@ -575,12 +573,21 @@ 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") + .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(); @@ -599,7 +606,8 @@ 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") + .optional(BINARY) + .named("inline") .named("inline_file"); assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); @@ -614,9 +622,12 @@ public void testFileLogicalTypeSelfReference() { // reference point. GroupType file = Types.requiredGroup() .as(LogicalTypeAnnotation.fileType()) - .optional(INT64).named("offset") - .optional(INT64).named("size") - .optional(BINARY).named("inline") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") + .optional(BINARY) + .named("inline") .named("self_ref_file"); assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); @@ -630,8 +641,10 @@ public void testFileLogicalTypeOffsetRequiresInline() { // neither 'uri' nor 'inline' is rejected at build time. assertThatThrownBy(() -> Types.requiredGroup() .as(LogicalTypeAnnotation.fileType()) - .optional(INT64).named("offset") - .optional(INT64).named("size") + .optional(INT64) + .named("offset") + .optional(INT64) + .named("size") .named("self_ref_without_inline")) .isInstanceOf(IllegalArgumentException.class); } @@ -643,9 +656,13 @@ public void testFileLogicalTypeExternalRangedReferenceWithoutInline() { // 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") + .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); @@ -658,8 +675,12 @@ public void testFileLogicalTypeMetadataOnlyRejected() { // '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") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("content_type") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("checksum") .named("file_metadata_only")) .isInstanceOf(IllegalArgumentException.class); } @@ -670,7 +691,8 @@ public void testFileLogicalTypeSizeOnlyRejected() { // rejected: it declares no locator ('inline', 'uri', or 'offset'). assertThatThrownBy(() -> Types.requiredGroup() .as(LogicalTypeAnnotation.fileType()) - .optional(INT64).named("size") + .optional(INT64) + .named("size") .named("file_size_only")) .isInstanceOf(IllegalArgumentException.class); } @@ -681,8 +703,11 @@ public void testFileLogicalTypeOffsetRequiresSize() { // 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") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("offset") .named("file_offset_without_size")) .isInstanceOf(IllegalArgumentException.class); } @@ -692,9 +717,13 @@ 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") + .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); @@ -706,8 +735,11 @@ 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") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT64) + .named("size") .named("file_size_without_offset"); assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); @@ -718,8 +750,11 @@ public void testFileLogicalTypeSizeWithoutOffset() { public void testFileLogicalTypeRejectsUnrecognizedField() { assertThatThrownBy(() -> Types.requiredGroup() .as(LogicalTypeAnnotation.fileType()) - .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri") - .optional(BINARY).named("unknown_field") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(BINARY) + .named("unknown_field") .named("file_with_bad_field")) .isInstanceOf(IllegalArgumentException.class); } @@ -729,7 +764,9 @@ 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") + .required(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") .named("file_with_required_uri")) .isInstanceOf(IllegalArgumentException.class); } @@ -740,7 +777,8 @@ public void testFileLogicalTypeRejectsGroupField() { assertThatThrownBy(() -> Types.requiredGroup() .as(LogicalTypeAnnotation.fileType()) .optionalGroup() - .optional(BINARY).named("nested") + .optional(BINARY) + .named("nested") .named("uri") .named("file_with_group_field")) .isInstanceOf(IllegalArgumentException.class); @@ -751,7 +789,8 @@ 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") + .optional(INT64) + .named("uri") .named("file_uri_wrong_type")) .isInstanceOf(IllegalArgumentException.class); } @@ -761,7 +800,8 @@ 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") + .optional(BINARY) + .named("uri") .named("file_uri_unannotated")) .isInstanceOf(IllegalArgumentException.class); } @@ -771,8 +811,11 @@ 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") + .optional(BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named("uri") + .optional(INT32) + .named("size") .named("file_size_wrong_type")) .isInstanceOf(IllegalArgumentException.class); } @@ -782,7 +825,8 @@ public void testFileLogicalTypeRejectsWrongInlinePhysicalType() { // 'inline' must be a BYTE_ARRAY (BINARY); an INT64 is rejected. assertThatThrownBy(() -> Types.requiredGroup() .as(LogicalTypeAnnotation.fileType()) - .optional(INT64).named("inline") + .optional(INT64) + .named("inline") .named("file_inline_wrong_type")) .isInstanceOf(IllegalArgumentException.class); } 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 e70e8658b9..8b4fb77577 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 @@ -122,18 +122,25 @@ public static byte[] createFooterAAD(byte[] aadPrefixBytes) { /** * 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. + * are identified by a 2-byte page ordinal, a self-reference is identified by the 8-byte + * little-endian offset of its stored representation within the file, following the row group and + * column ordinals. The column ordinal is that of the {@code inline} column whose encryption the + * self-reference inherits. + * + *

The offset is the value the writer records in the {@code offset} field of the {@code FILE} + * group. Because it is carried in the data, a reader can rebuild this AAD from the value alone, + * without counting the self-references that precede it and therefore without decoding the pages + * it skips. * * @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 + * @param selfReferenceOffset the offset of the stored representation within the file, i.e. the + * value of the self-reference's {@code offset} field * @return the module AAD bytes */ public static byte[] createSelfReferenceAAD( - byte[] fileAAD, int rowGroupOrdinal, int columnOrdinal, long selfReferenceOrdinal) { + byte[] fileAAD, int rowGroupOrdinal, int columnOrdinal, long selfReferenceOffset) { byte[] typeOrdinalBytes = new byte[1]; typeOrdinalBytes[0] = ModuleType.SelfReference.getValue(); @@ -158,13 +165,13 @@ public static byte[] createSelfReferenceAAD( } byte[] columnOrdinalBytes = shortToBytesLE(shortColumnOrdinal); - if (selfReferenceOrdinal < 0) { - throw new IllegalArgumentException("Wrong self-reference ordinal: " + selfReferenceOrdinal); + if (selfReferenceOffset < 0) { + throw new IllegalArgumentException("Wrong self-reference offset: " + selfReferenceOffset); } - byte[] selfReferenceOrdinalBytes = longToBytesLE(selfReferenceOrdinal); + byte[] selfReferenceOffsetBytes = longToBytesLE(selfReferenceOffset); return concatByteArrays( - fileAAD, typeOrdinalBytes, rowGroupOrdinalBytes, columnOrdinalBytes, selfReferenceOrdinalBytes); + fileAAD, typeOrdinalBytes, rowGroupOrdinalBytes, columnOrdinalBytes, selfReferenceOffsetBytes); } // Update last two bytes with new page ordinal (instead of creating new page AAD from scratch) 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 30cac68e28..0df057a53d 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,7 +111,6 @@ 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; 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 71ca455e6e..de06bbcfbb 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,12 +281,33 @@ public BytesCompressor getCompressor(CompressionCodecName codecName, int level) return comp; } + /** Smallest output buffer tried by {@link #decompressUnknownSize}. */ + private static final int MIN_UNKNOWN_SIZE_BUFFER = 8 * 1024; + + /** + * Largest output buffer tried by {@link #decompressUnknownSize}. Java arrays are indexed by int, + * and some JVMs reserve a few header words, so this stays just below {@link Integer#MAX_VALUE}. + */ + private static final int MAX_UNKNOWN_SIZE_BUFFER = Integer.MAX_VALUE - 8; + /** - * 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. + * Decompresses a complete compression block whose decompressed size is not known in advance. 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. + * + *

All codecs are supported, including those that record no decompressed size of their own. The + * format spec allows a reader to "decompress into a dynamically sized buffer", which is what this + * does: it guesses an output size, and whenever the codec fills the buffer exactly — the signal + * that the output may have been cut off — it doubles the guess and retries. Retries are bounded by + * the 2 GiB ceiling on a Java array. + * + *

The {@link Decompressor} is driven directly rather than through + * {@link BytesDecompressor#decompress(BytesInput, int)} or a stream-drain loop. The former reads + * back exactly the requested number of bytes and so cannot report a short read, and Parquet's + * codec streams are deliberately unframed ({@code NonBlockedDecompressorStream}), signalling a + * fully consumed block by throwing rather than by returning end-of-input. * * @param codecName the {@link CompressionCodecName} of the {@code inline} column chunk the * self-reference inherits from; {@link CompressionCodecName#UNCOMPRESSED} returns the bytes @@ -295,31 +316,73 @@ public BytesCompressor getCompressor(CompressionCodecName codecName, int level) * @return the decompressed (resolved) bytes * @throws IOException if decompression fails */ - public BytesInput decompressUnknownSize(CompressionCodecName codecName, BytesInput compressed) - throws IOException { + 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; } + + byte[] compressedBytes = compressed.toByteArray(); + if (compressedBytes.length == 0) { + // An empty payload compresses to nothing and resolves back to nothing. + return BytesInput.empty(); + } + 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]; + if (decompressor == null) { + // Some codecs (ZSTD) expose no Decompressor and decompress only through their stream, which + // is framed and so reports end-of-input properly. Drain it. + try (InputStream is = codec.createInputStream(compressed.toInputStream(), null); + ByteArrayOutputStream out = new ByteArrayOutputStream(compressedBytes.length * 2)) { + byte[] buffer = new byte[MIN_UNKNOWN_SIZE_BUFFER]; 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); + } + try { + // Compression rarely achieves better than 2x on the payloads worth storing out of line, so + // the first attempt usually suffices. + long attemptSize = Math.max((long) compressedBytes.length * 2, MIN_UNKNOWN_SIZE_BUFFER); + while (true) { + int candidate = (int) Math.min(attemptSize, MAX_UNKNOWN_SIZE_BUFFER); + boolean lastAttempt = candidate == MAX_UNKNOWN_SIZE_BUFFER; + byte[] output = new byte[candidate]; + int total = 0; + boolean undersized = false; + + decompressor.reset(); + decompressor.setInput(compressedBytes, 0, compressedBytes.length); + try { + while (total < candidate && !decompressor.finished()) { + int written = decompressor.decompress(output, total, candidate - total); + if (written <= 0) { + break; + } + total += written; + } + } catch (IOException | RuntimeException e) { + // Codecs with no length information (e.g. raw LZ4) fail outright when the output buffer is + // too small rather than filling it, so treat a failure as a signal to grow. On the last + // attempt there is nothing left to try, so let it surface. + if (lastAttempt) { + throw e; + } + undersized = true; + } + + // Filling the buffer exactly is also ambiguous: the payload may be complete, or the codec may + // have had more to write. Grow and retry unless the decompressor confirmed it finished. + if (!undersized && (total < candidate || decompressor.finished() || lastAttempt)) { + return BytesInput.from(output, 0, total); + } + attemptSize = (long) candidate * 2; } + } finally { + CodecPool.returnDecompressor(decompressor); } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/FileValueWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/FileValueWriter.java new file mode 100644 index 0000000000..59d4d75abd --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/FileValueWriter.java @@ -0,0 +1,197 @@ +/* + * 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.format.BlockCipher; +import org.apache.parquet.io.api.Binary; + +/** + * Decides how a {@code FILE} value's payload is stored: inline in the value, or out of line as a + * self-reference within the same Parquet file. + * + *

Object models hand over the resolved (logical) bytes and receive back a {@link Placement} + * describing which fields of the {@code FILE} group to write. Callers do not choose between the two + * forms themselves; the choice follows the configured threshold, so the same writing code produces + * either form: + * + * 

{@code
+ * FileValueWriter.Placement placement = fileValueWriter.write(payload);
+ * if (placement.isInline()) {
+ *   group.add("inline", placement.getInlineBytes());
+ * } else {
+ *   group.add("offset", placement.getOffset());
+ *   group.add("size", placement.getSize());
+ * }
+ * }
+ * + *

Both forms describe the same logical bytes, so {@code content_type} and {@code checksum} are + * written identically either way — they describe the resolved bytes, not the storage. Consumers see + * no difference beyond which fields are set. + * + *

A self-reference payload is written immediately, while the record is being written and before + * the row group's column chunks are flushed. It therefore lands in a contiguous run ahead of those + * chunks, leaving each column chunk contiguous on disk. Writing eagerly is what makes the offset + * knowable in time: {@code offset} and {@code size} are ordinary column values, and once a value has + * been handed to a column writer it is encoded into a buffered page and cannot be revised, so a + * placeholder could never be patched up later. + * + * @see SelfReferenceStorage + */ +public class FileValueWriter { + + /** + * Where a {@code FILE} value's payload was placed, and therefore which fields of the {@code FILE} + * group the caller should write. Either the payload is inline, or it is a self-reference located by + * {@code offset} and {@code size}. + */ + public static final class Placement { + private final Binary inlineBytes; + private final long offset; + private final long size; + + private Placement(Binary inlineBytes, long offset, long size) { + this.inlineBytes = inlineBytes; + this.offset = offset; + this.size = size; + } + + static Placement inline(Binary inlineBytes) { + return new Placement(inlineBytes, -1, -1); + } + + static Placement selfReference(SelfReferenceStorage.StoredRange range) { + return new Placement(null, range.getOffset(), range.getSize()); + } + + /** Whether the payload is stored inline, i.e. whether the {@code inline} field should be set. */ + public boolean isInline() { + return inlineBytes != null; + } + + /** + * The bytes to write to the {@code inline} field. + * + * @throws IllegalStateException if the payload was stored as a self-reference + */ + public Binary getInlineBytes() { + if (!isInline()) { + throw new IllegalStateException("Payload was stored as a self-reference, not inline"); + } + return inlineBytes; + } + + /** + * The value to write to the {@code offset} field. + * + * @throws IllegalStateException if the payload was stored inline + */ + public long getOffset() { + if (isInline()) { + throw new IllegalStateException("Payload was stored inline; it has no offset"); + } + return offset; + } + + /** + * The value to write to the {@code size} field. This is the size of the stored representation + * after compression and encryption, not the size of the resolved bytes. + * + * @throws IllegalStateException if the payload was stored inline + */ + public long getSize() { + if (isInline()) { + throw new IllegalStateException("Payload was stored inline; it has no size"); + } + return size; + } + } + + private final ParquetFileWriter fileWriter; + private final CodecFactory.BytesCompressor inlineColumnCompressor; + private final BlockCipher.Encryptor inlineColumnEncryptor; + private final int inlineColumnOrdinal; + private final int selfReferenceThreshold; + + /** + * @param fileWriter the writer for the file being written; a block must be open when + * {@link #write} is called + * @param inlineColumnCompressor the compressor for the {@code inline} column chunk's codec, whose + * compression a self-reference inherits + * @param inlineColumnEncryptor the data-module encryptor of the {@code inline} column chunk, or + * {@code null} if that column chunk is not encrypted + * @param inlineColumnOrdinal the ordinal of the {@code inline} column within the schema. The + * schema must declare {@code inline}: it is the reference point whose compression and + * encryption a self-reference inherits, so a schema without it can only store payloads inline + * or as external references. Note that the schema builder does not require {@code inline} for + * groups that declare {@code uri}, so an external-reference schema may reach here; pair such a + * schema with a threshold of {@link Integer#MAX_VALUE} so nothing is stored out of line. + * @param selfReferenceThreshold payloads of at most this many bytes are stored inline; larger ones + * become self-references. See + * {@code ParquetProperties.Builder#withFileSelfReferenceThreshold(int)}. + */ + public FileValueWriter( + ParquetFileWriter fileWriter, + CodecFactory.BytesCompressor inlineColumnCompressor, + BlockCipher.Encryptor inlineColumnEncryptor, + int inlineColumnOrdinal, + int selfReferenceThreshold) { + if (selfReferenceThreshold < 0) { + throw new IllegalArgumentException( + "Self-reference threshold must not be negative: " + selfReferenceThreshold); + } + if (inlineColumnOrdinal < 0) { + throw new IllegalArgumentException("Invalid inline column ordinal: " + inlineColumnOrdinal); + } + this.fileWriter = fileWriter; + this.inlineColumnCompressor = inlineColumnCompressor; + this.inlineColumnEncryptor = inlineColumnEncryptor; + this.inlineColumnOrdinal = inlineColumnOrdinal; + this.selfReferenceThreshold = selfReferenceThreshold; + } + + /** + * Stores {@code payload} and returns which {@code FILE} group fields to write for it. Payloads at + * or below the configured threshold are returned for inline storage; larger ones are written to the + * file body immediately as self-references. + * + *

Must be called while a block is open on the underlying writer, and before that block's column + * chunks are flushed. + * + * @param payload the resolved (logical) bytes of the value + * @return the placement describing which fields to write + * @throws IOException if writing the self-reference payload fails + */ + public Placement write(Binary payload) throws IOException { + if (payload == null) { + throw new IllegalArgumentException("FILE payload must not be null"); + } + if (payload.length() <= selfReferenceThreshold) { + return Placement.inline(payload); + } + SelfReferenceStorage.StoredRange range = fileWriter.writeSelfReference( + BytesInput.from(payload.toByteBuffer()), + inlineColumnCompressor, + inlineColumnEncryptor, + inlineColumnOrdinal); + return Placement.selfReference(range); + } +} 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 5cfaae4ef2..6e70445336 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 @@ -1080,24 +1080,30 @@ public String getFile() { * chunk's codec, returning the resolved bytes. See {@link SelfReferenceStorage} and the Parquet * format's "FILE" logical type specification. * + *

Everything needed to resolve the value comes from the value itself plus the {@code inline} + * column chunk's metadata, so a self-reference can be read without decoding the pages that + * precede it. + * * @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 { + public BytesInput resolveSelfReference(ColumnChunkMetaData inlineColumn, long offset, long size) + 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); } + if (size > SelfReferenceStorage.MAX_ENCRYPTED_MODULE_SIZE) { + throw new IllegalArgumentException("Self-reference size exceeds the maximum readable range: " + size); + } - byte[] stored = new byte[Math.toIntExact(size)]; + byte[] stored = new byte[(int) size]; f.seek(offset); f.readFully(stored); @@ -1127,7 +1133,7 @@ public BytesInput resolveSelfReference( fileAAD, inlineColumn.getRowGroupOrdinal(), columnOrdinal, - selfReferenceOrdinal); + offset); } private List filterRowGroups(List blocks) throws IOException { 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 fba8c34733..eb4de8d28a 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 @@ -622,6 +622,11 @@ public InternalFileEncryptor getEncryptor() { * {@code Self-Reference} module type. The row group ordinal is that of the block currently being * written. * + *

Payloads are written while a block is open but before its column chunks are flushed, so they + * land in a contiguous run ahead of the row group's chunks. This keeps each column chunk + * contiguous on disk, which the read path relies on when coalescing adjacent chunks into a single + * range read. + * *

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. * @@ -630,7 +635,6 @@ public InternalFileEncryptor getEncryptor() { * @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 */ @@ -638,22 +642,14 @@ public SelfReferenceStorage.StoredRange writeSelfReference( BytesInput resolvedBytes, CodecFactory.BytesCompressor compressor, BlockCipher.Encryptor pageBlockEncryptor, - int columnOrdinal, - long selfReferenceOrdinal) + int columnOrdinal) 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); + resolvedBytes, compressor, pageBlockEncryptor, fileAAD, rowGroupOrdinal, columnOrdinal, out); }); } 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 index c534157864..0660b05ab5 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SelfReferenceStorage.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SelfReferenceStorage.java @@ -48,15 +48,33 @@ *

  • 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}. + * complete encrypted module. The AAD uses the {@code Self-Reference} module type (10) with the + * 8-byte file offset of the stored representation; see + * {@link AesCipher#createSelfReferenceAAD}. * * + *

    Because the AAD is keyed on the file offset — a value the {@code FILE} group already carries in + * its {@code offset} field — a reader can resolve a self-reference directly from the value, without + * decoding the pages preceding it. An encrypted stored representation is therefore bound to one + * column chunk at one offset and must not be shared between column chunks. + * *

    Compression is always applied before encryption on write; decryption is applied before * decompression on read. */ public final class SelfReferenceStorage { + /** + * The largest encrypted module a writer can serialize: the 4-byte little-endian length field is + * read back as a signed int, so the buffer it describes cannot exceed 2 GiB. + */ + public static final long MAX_ENCRYPTED_MODULE_SIZE = Integer.MAX_VALUE; + + /** + * Bytes an encrypted module adds around the compressed block: the 4-byte length, the 12-byte + * nonce, and the 16-byte GCM tag. AES_GCM_CTR_V1 omits the tag, so this is an upper bound. + */ + private static final long MAX_ENCRYPTION_OVERHEAD = 4 + 12 + 16; + private SelfReferenceStorage() {} /** @@ -97,7 +115,6 @@ public long getSize() { * @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 @@ -109,7 +126,6 @@ public static StoredRange write( byte[] fileAAD, int rowGroupOrdinal, int columnOrdinal, - long selfReferenceOrdinal, org.apache.parquet.io.PositionOutputStream out) throws IOException { @@ -117,16 +133,29 @@ public static StoredRange write( // the bytes unchanged (the NO_OP_COMPRESSOR returns its input). BytesInput stored = compressor.compress(resolvedBytes); + // The offset of the stored representation is the current stream position, and it is also the + // AAD's self-reference identity, so it must be read before anything is written. + long offset = out.getPos(); + // 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. + // independent module keyed on that offset. 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); + long plaintextSize = stored.size(); + // The 4-byte length field of an encrypted module caps the buffer at 2 GiB. Check before + // encrypting so an oversized value fails with a diagnostic instead of a corrupt length. + long encryptedSize = plaintextSize + MAX_ENCRYPTION_OVERHEAD; + if (encryptedSize > MAX_ENCRYPTED_MODULE_SIZE) { + throw new IllegalArgumentException("Self-reference is too large to encrypt: " + plaintextSize + + " compressed bytes exceed the " + MAX_ENCRYPTED_MODULE_SIZE + + "-byte limit imposed by the 4-byte length field of an encrypted module. " + + "Store this value as an external reference (uri) instead."); + } + byte[] selfReferenceAAD = AesCipher.createSelfReferenceAAD(fileAAD, rowGroupOrdinal, columnOrdinal, offset); stored = BytesInput.from(pageBlockEncryptor.encrypt(stored.toByteArray(), selfReferenceAAD)); } - long offset = out.getPos(); long size = stored.size(); stored.writeAllTo(out); return new StoredRange(offset, size); @@ -146,7 +175,8 @@ public static StoredRange write( * @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 + * @param selfReferenceOffset the value of the self-reference's {@code offset} field, which is both + * where {@code storedBytes} was read from and the self-reference's AAD identity * @return the resolved (logical) bytes * @throws IOException if decompression fails */ @@ -158,20 +188,22 @@ public static BytesInput resolve( byte[] fileAAD, int rowGroupOrdinal, int columnOrdinal, - long selfReferenceOrdinal) + long selfReferenceOffset) 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. + // length, nonce, ciphertext, and GCM tag and returns the compressed block. The AAD is rebuilt + // from the offset alone, so no state from preceding values is needed. if (pageBlockDecryptor != null) { byte[] selfReferenceAAD = - AesCipher.createSelfReferenceAAD(fileAAD, rowGroupOrdinal, columnOrdinal, selfReferenceOrdinal); + AesCipher.createSelfReferenceAAD(fileAAD, rowGroupOrdinal, columnOrdinal, selfReferenceOffset); 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. + // Step 2: decompress. The resolved size is not stored, so the codec decompresses into a + // dynamically sized buffer. 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 index 2e6772e8bd..718798c157 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestSelfReferenceAAD.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestSelfReferenceAAD.java @@ -45,13 +45,12 @@ public void testSelfReferenceModuleTypeValue() { public void testSelfReferenceAADLayout() { int rowGroupOrdinal = 3; int columnOrdinal = 7; - long selfReferenceOrdinal = 0x0102030405060708L; + long selfReferenceOffset = 0x0102030405060708L; - byte[] aad = - AesCipher.createSelfReferenceAAD(FILE_AAD, rowGroupOrdinal, columnOrdinal, selfReferenceOrdinal); + byte[] aad = AesCipher.createSelfReferenceAAD(FILE_AAD, rowGroupOrdinal, columnOrdinal, selfReferenceOffset); // Layout: fileAAD | moduleType(1) | rowGroupOrdinal(2 LE) | columnOrdinal(2 LE) | - // selfReferenceOrdinal(8 LE) + // selfReferenceOffset(8 LE) assertThat(aad.length).isEqualTo(FILE_AAD.length + 1 + 2 + 2 + 8); ByteBuffer buf = ByteBuffer.wrap(aad).order(ByteOrder.LITTLE_ENDIAN); @@ -61,21 +60,30 @@ public void testSelfReferenceAADLayout() { 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); + // The self-reference is identified by the 8-byte file offset of its stored representation, + // unlike the 2-byte page ordinal. + assertThat(buf.getLong()).isEqualTo(selfReferenceOffset); } @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); + public void testSelfReferenceAADSupportsLargeOffset() { + // File offsets routinely exceed the 2-byte page-ordinal range, so the field must be 8 bytes. + long largeOffset = 5L * 1024 * 1024 * 1024; + byte[] aad = AesCipher.createSelfReferenceAAD(FILE_AAD, 0, 0, largeOffset); ByteBuffer buf = ByteBuffer.wrap(aad, FILE_AAD.length + 1 + 2 + 2, 8).order(ByteOrder.LITTLE_ENDIAN); - assertThat(buf.getLong()).isEqualTo(largeOrdinal); + assertThat(buf.getLong()).isEqualTo(largeOffset); } @Test - public void testSelfReferenceAADRejectsNegativeOrdinals() { + public void testDistinctOffsetsProduceDistinctAADs() { + // Two self-references in the same column chunk are distinguished solely by their offsets. + byte[] first = AesCipher.createSelfReferenceAAD(FILE_AAD, 1, 2, 1000L); + byte[] second = AesCipher.createSelfReferenceAAD(FILE_AAD, 1, 2, 1064L); + assertThat(first).isNotEqualTo(second); + } + + @Test + public void testSelfReferenceAADRejectsNegativeValues() { assertThatThrownBy(() -> AesCipher.createSelfReferenceAAD(FILE_AAD, -1, 0, 0)) .isInstanceOf(IllegalArgumentException.class); assertThatThrownBy(() -> AesCipher.createSelfReferenceAAD(FILE_AAD, 0, -1, 0)) @@ -87,7 +95,7 @@ public void testSelfReferenceAADRejectsNegativeOrdinals() { @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). + // type byte differs (and the trailing field differs in width and meaning). 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 f554157c6d..31d6a7380e 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 @@ -2307,10 +2307,10 @@ public void testFileLogicalType() { List parquetSchema = parquetMetadataConverter.toParquetSchema(expected); MessageType schema = parquetMetadataConverter.fromParquetSchema(parquetSchema, null); - assertEquals(expected, schema); + assertThat(schema).isEqualTo(expected); LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); - assertTrue(logicalType instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); - assertEquals(LogicalTypeAnnotation.fileType(), logicalType); + assertThat(logicalType).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); + assertThat(logicalType).isEqualTo(LogicalTypeAnnotation.fileType()); } @Test @@ -2328,8 +2328,8 @@ public void testFileLogicalTypeRoundTripUriOnly() { List parquetSchema = parquetMetadataConverter.toParquetSchema(expected); MessageType schema = parquetMetadataConverter.fromParquetSchema(parquetSchema, null); - assertEquals(expected, schema); + assertThat(schema).isEqualTo(expected); LogicalTypeAnnotation logicalType = schema.getType("f").getLogicalTypeAnnotation(); - assertTrue(logicalType instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation); + assertThat(logicalType).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class); } } diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFileValueWriter.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFileValueWriter.java new file mode 100644 index 0000000000..5110b1b5bb --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFileValueWriter.java @@ -0,0 +1,274 @@ +/* + * 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.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 static org.assertj.core.api.Assertions.assertThatThrownBy; + +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.ParquetReadOptions; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.ParquetProperties; +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.ColumnPath; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.hadoop.util.HadoopOutputFile; +import org.apache.parquet.io.InputFile; +import org.apache.parquet.io.api.Binary; +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; + +/** + * Tests that {@link FileValueWriter} routes a {@code FILE} payload to inline storage or to a + * self-reference according to the configured threshold, and that both forms describe the same logical + * bytes. + */ +public class TestFileValueWriter { + + private static final MessageType SCHEMA = MessageTypeParser.parseMessageType("message m {" + + " optional group file (FILE) {" + + " optional int64 offset;" + + " optional int64 size;" + + " optional binary inline;" + + " }" + + "}"); + + 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.BINARY).named("inline")) + .build(); + + @TempDir + java.nio.file.Path tempDir; + + @Test + public void testDefaultThresholdIsPageSize() { + assertThat(ParquetProperties.builder().build().getFileSelfReferenceThreshold()) + .isEqualTo(ParquetProperties.DEFAULT_PAGE_SIZE); + } + + @Test + public void testThresholdMustNotBeNegative() { + assertThatThrownBy(() -> ParquetProperties.builder().withFileSelfReferenceThreshold(-1)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testPayloadAtThresholdIsInlinedAndAboveIsSelfReference() throws IOException { + Configuration conf = new Configuration(); + Path path = new Path(tempDir.resolve("routing.parquet").toUri()); + CodecFactory codecFactory = new CodecFactory(conf, DEFAULT_BLOCK_SIZE); + + int threshold = 64; + Binary atThreshold = Binary.fromConstantByteArray(payload(threshold)); + Binary aboveThreshold = Binary.fromConstantByteArray(payload(threshold + 1)); + + ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, conf), SCHEMA, CREATE, DEFAULT_BLOCK_SIZE, MAX_PADDING_SIZE_DEFAULT); + writer.start(); + writer.startBlock(2); + + FileValueWriter valueWriter = new FileValueWriter( + writer, codecFactory.getCompressor(CODEC), null, columnOrdinalOf(INLINE_COLUMN), threshold); + + FileValueWriter.Placement inlined = valueWriter.write(atThreshold); + FileValueWriter.Placement outOfLine = valueWriter.write(aboveThreshold); + + // A payload exactly at the threshold stays inline; one byte more goes out of line. + assertThat(inlined.isInline()).isTrue(); + assertThat(inlined.getInlineBytes()).isEqualTo(atThreshold); + assertThatThrownBy(inlined::getOffset).isInstanceOf(IllegalStateException.class); + + assertThat(outOfLine.isInline()).isFalse(); + assertThat(outOfLine.getSize()).isGreaterThan(0L); + assertThatThrownBy(outOfLine::getInlineBytes).isInstanceOf(IllegalStateException.class); + + // Write the inline column chunk so the reader has metadata carrying the inherited codec. + writer.startColumn(INLINE_COLUMN, 1, CODEC); + writer.writeDataPage( + 1, + (int) inlined.getInlineBytes().length(), + codecFactory + .getCompressor(CODEC) + .compress(BytesInput.from(inlined.getInlineBytes().toByteBuffer())), + EMPTY_STATS, + Encoding.BIT_PACKED, + Encoding.BIT_PACKED, + Encoding.PLAIN); + writer.endColumn(); + writer.endBlock(); + writer.end(new java.util.HashMap<>()); + + // The out-of-line payload resolves back to the original bytes. + InputFile inputFile = HadoopInputFile.fromPath(path, conf); + try (ParquetFileReader reader = + ParquetFileReader.open(inputFile, ParquetReadOptions.builder().build())) { + BlockMetaData block = reader.getFooter().getBlocks().get(0); + ColumnChunkMetaData inlineMeta = findColumn(block, INLINE_COLUMN); + BytesInput resolved = reader.resolveSelfReference(inlineMeta, outOfLine.getOffset(), outOfLine.getSize()); + assertThat(resolved.toByteArray()).isEqualTo(aboveThreshold.getBytes()); + } + codecFactory.release(); + } + + @Test + public void testZeroThresholdAlwaysUsesSelfReferences() throws IOException { + Configuration conf = new Configuration(); + Path path = new Path(tempDir.resolve("always_out_of_line.parquet").toUri()); + CodecFactory codecFactory = new CodecFactory(conf, DEFAULT_BLOCK_SIZE); + + byte[][] payloads = {payload(1), payload(1000)}; + + ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, conf), SCHEMA, CREATE, DEFAULT_BLOCK_SIZE, MAX_PADDING_SIZE_DEFAULT); + writer.start(); + writer.startBlock(payloads.length); + + FileValueWriter valueWriter = + new FileValueWriter(writer, codecFactory.getCompressor(CODEC), null, columnOrdinalOf(INLINE_COLUMN), 0); + + List placements = new ArrayList<>(); + for (byte[] p : payloads) { + placements.add(valueWriter.write(Binary.fromConstantByteArray(p))); + } + // Every payload went out of line, including the single-byte one. An empty payload would still be + // inlined, since its length is not greater than the threshold. + assertThat(placements).allMatch(p -> !p.isInline()); + + writer.startColumn(INLINE_COLUMN, 0, CODEC); + writer.writeDataPage( + 0, + 0, + codecFactory.getCompressor(CODEC).compress(BytesInput.empty()), + EMPTY_STATS, + Encoding.BIT_PACKED, + Encoding.BIT_PACKED, + Encoding.PLAIN); + writer.endColumn(); + writer.endBlock(); + writer.end(new java.util.HashMap<>()); + + InputFile inputFile = HadoopInputFile.fromPath(path, conf); + try (ParquetFileReader reader = + ParquetFileReader.open(inputFile, ParquetReadOptions.builder().build())) { + BlockMetaData block = reader.getFooter().getBlocks().get(0); + ColumnChunkMetaData inlineMeta = findColumn(block, INLINE_COLUMN); + for (int i = 0; i < payloads.length; i++) { + FileValueWriter.Placement placement = placements.get(i); + BytesInput resolved = + reader.resolveSelfReference(inlineMeta, placement.getOffset(), placement.getSize()); + assertThat(resolved.toByteArray()).isEqualTo(payloads[i]); + } + } + codecFactory.release(); + } + + @Test + public void testMaxThresholdAlwaysInlines() throws IOException { + Configuration conf = new Configuration(); + Path path = new Path(tempDir.resolve("always_inline.parquet").toUri()); + CodecFactory codecFactory = new CodecFactory(conf, DEFAULT_BLOCK_SIZE); + + ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, conf), SCHEMA, CREATE, DEFAULT_BLOCK_SIZE, MAX_PADDING_SIZE_DEFAULT); + writer.start(); + writer.startBlock(1); + + FileValueWriter valueWriter = new FileValueWriter( + writer, codecFactory.getCompressor(CODEC), null, columnOrdinalOf(INLINE_COLUMN), Integer.MAX_VALUE); + + long posBefore = writer.getPos(); + FileValueWriter.Placement placement = valueWriter.write(Binary.fromConstantByteArray(payload(1 << 20))); + + assertThat(placement.isInline()).isTrue(); + // Nothing was written to the file body, because the payload is carried by the value itself. + assertThat(writer.getPos()).isEqualTo(posBefore); + + writer.abort(); + codecFactory.release(); + } + + @Test + public void testNullPayloadIsRejected() throws IOException { + Configuration conf = new Configuration(); + Path path = new Path(tempDir.resolve("null_payload.parquet").toUri()); + CodecFactory codecFactory = new CodecFactory(conf, DEFAULT_BLOCK_SIZE); + + ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, conf), SCHEMA, CREATE, DEFAULT_BLOCK_SIZE, MAX_PADDING_SIZE_DEFAULT); + writer.start(); + writer.startBlock(1); + + FileValueWriter valueWriter = new FileValueWriter( + writer, codecFactory.getCompressor(CODEC), null, columnOrdinalOf(INLINE_COLUMN), 64); + assertThatThrownBy(() -> valueWriter.write(null)).isInstanceOf(IllegalArgumentException.class); + + writer.abort(); + 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) { + ColumnPath target = 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[] payload(int length) { + StringBuilder sb = new StringBuilder(); + while (sb.length() < length) { + sb.append("file-payload-"); + } + return sb.substring(0, length).getBytes(StandardCharsets.UTF_8); + } +} 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 index 9966b657ec..c1ebcdb581 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceFileWrite.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceFileWrite.java @@ -32,6 +32,7 @@ import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.parquet.ParquetReadOptions; import org.apache.parquet.bytes.BytesInput; import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.column.page.DataPage; @@ -77,8 +78,7 @@ public class TestSelfReferenceFileWrite { 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 ColumnDescriptor INLINE_COLUMN = SCHEMA.getColumnDescription(new String[] {"file", "inline"}); private static final CompressionCodecName CODEC = CompressionCodecName.SNAPPY; @@ -122,8 +122,7 @@ public void testWriteDataAlongsideSelfReferences() throws IOException { BytesInput.from(payloads[i]), codecFactory.getCompressor(CODEC), null, // unencrypted file - inlineColumnOrdinal, - i)); + inlineColumnOrdinal)); } // Write a normal data page for the id column in the same block. @@ -167,11 +166,18 @@ public void testWriteDataAlongsideSelfReferences() throws IOException { } // Each self-reference resolves back to its original payload, inheriting the inline column's - // codec. + // codec. Only the offset and size recorded in the value are needed -- no per-value counter, so + // resolution does not depend on having read the preceding values. for (int i = 0; i < payloads.length; i++) { SelfReferenceStorage.StoredRange range = ranges.get(i); - BytesInput resolved = - reader.resolveSelfReference(inlineMeta, range.getOffset(), range.getSize(), i); + BytesInput resolved = reader.resolveSelfReference(inlineMeta, range.getOffset(), range.getSize()); + assertThat(resolved.toByteArray()).isEqualTo(payloads[i]); + } + + // Resolution order is irrelevant, which is the point of keying on the offset. + for (int i = payloads.length - 1; i >= 0; i--) { + SelfReferenceStorage.StoredRange range = ranges.get(i); + BytesInput resolved = reader.resolveSelfReference(inlineMeta, range.getOffset(), range.getSize()); assertThat(resolved.toByteArray()).isEqualTo(payloads[i]); } } 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 index dca5753539..a17daffee0 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceStorage.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSelfReferenceStorage.java @@ -20,6 +20,7 @@ package org.apache.parquet.hadoop; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -29,6 +30,7 @@ import org.apache.parquet.crypto.AesCipher; import org.apache.parquet.crypto.AesMode; import org.apache.parquet.crypto.ModuleCipherFactory; +import org.apache.parquet.crypto.ParquetCryptoRuntimeException; import org.apache.parquet.format.BlockCipher; import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.apache.parquet.io.PositionOutputStream; @@ -76,7 +78,7 @@ byte[] toByteArray() { @ParameterizedTest @EnumSource( value = CompressionCodecName.class, - names = {"UNCOMPRESSED", "SNAPPY", "GZIP"}) + names = {"UNCOMPRESSED", "SNAPPY", "GZIP", "ZSTD", "LZ4_RAW"}) public void testRoundTripUnencrypted(CompressionCodecName codec) throws IOException { CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); byte[] resolved = highlyCompressiblePayload(4096); @@ -86,14 +88,7 @@ public void testRoundTripUnencrypted(CompressionCodecName codec) throws IOExcept 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); + BytesInput.from(resolved), codecFactory.getCompressor(codec), null, null, 0, 0, out); byte[] fileBytes = out.toByteArray(); assertThat(range.getOffset()).isEqualTo(1L); @@ -105,9 +100,65 @@ public void testRoundTripUnencrypted(CompressionCodecName codec) throws IOExcept 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); + BytesInput resolvedBack = + SelfReferenceStorage.resolve(BytesInput.from(stored), codec, codecFactory, null, null, 0, 0, 0L); + + assertThat(resolvedBack.toByteArray()).isEqualTo(resolved); + codecFactory.release(); + } + + /** + * The decompressed size of a self-reference is not stored, so the reader grows its output buffer + * until the payload fits. This exercises payload sizes spanning several doublings, including sizes + * that are exact powers of two, where a full output buffer is ambiguous between "complete" and + * "truncated". + */ + @ParameterizedTest + @EnumSource( + value = CompressionCodecName.class, + names = {"SNAPPY", "GZIP", "ZSTD", "LZ4_RAW"}) + public void testRoundTripAcrossBufferGrowth(CompressionCodecName codec) throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + int[] sizes = {1, 8192, 8193, 16384, 100_000, 1 << 20}; + for (int size : sizes) { + byte[] resolved = highlyCompressiblePayload(size); + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + SelfReferenceStorage.StoredRange range = SelfReferenceStorage.write( + BytesInput.from(resolved), codecFactory.getCompressor(codec), null, null, 0, 0, out); + + byte[] stored = out.toByteArray(); + assertThat(range.getSize()).isEqualTo((long) stored.length); + BytesInput resolvedBack = + SelfReferenceStorage.resolve(BytesInput.from(stored), codec, codecFactory, null, null, 0, 0, 0L); + assertThat(resolvedBack.toByteArray()) + .as("payload of %s bytes", size) + .isEqualTo(resolved); + } + codecFactory.release(); + } + + /** + * Incompressible data expands slightly under most codecs, so the initial guess of twice the + * compressed size is generous; this simply confirms such payloads round-trip too. + */ + @ParameterizedTest + @EnumSource( + value = CompressionCodecName.class, + names = {"SNAPPY", "GZIP", "ZSTD", "LZ4_RAW"}) + public void testRoundTripIncompressiblePayload(CompressionCodecName codec) throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + byte[] resolved = new byte[64 * 1024]; + new java.util.Random(42).nextBytes(resolved); + + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + SelfReferenceStorage.StoredRange range = SelfReferenceStorage.write( + BytesInput.from(resolved), codecFactory.getCompressor(codec), null, null, 0, 0, out); + + byte[] stored = out.toByteArray(); + assertThat(range.getSize()).isEqualTo((long) stored.length); + BytesInput resolvedBack = + SelfReferenceStorage.resolve(BytesInput.from(stored), codec, codecFactory, null, null, 0, 0, 0L); assertThat(resolvedBack.toByteArray()).isEqualTo(resolved); codecFactory.release(); } @@ -118,40 +169,94 @@ 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); + BytesInput.from(resolved), codecFactory.getCompressor(codec), encryptor, FILE_AAD, 1, 2, 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); + 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); + BytesInput.from(stored), codec, codecFactory, decryptor, FILE_AAD, 1, 2, range.getOffset()); assertThat(resolvedBack.toByteArray()).isEqualTo(resolved); codecFactory.release(); } + /** + * The AAD binds a stored representation to its offset, so resolving the same bytes as if they lived + * at a different offset must fail rather than silently return data. For GCM the tag check catches + * it; CTR has no tag, so it yields garbage instead -- either way the bytes must not come back + * intact. + */ + @Test + public void testResolveWithWrongOffsetDoesNotReturnPayload() throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + byte[] resolved = highlyCompressiblePayload(4096); + CompressionCodecName codec = CompressionCodecName.SNAPPY; + + BlockCipher.Encryptor encryptor = ModuleCipherFactory.getEncryptor(AesMode.GCM, COLUMN_KEY); + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + SelfReferenceStorage.StoredRange range = SelfReferenceStorage.write( + BytesInput.from(resolved), codecFactory.getCompressor(codec), encryptor, FILE_AAD, 1, 2, out); + + byte[] stored = out.toByteArray(); + BlockCipher.Decryptor decryptor = ModuleCipherFactory.getDecryptor(AesMode.GCM, COLUMN_KEY); + assertThatThrownBy(() -> SelfReferenceStorage.resolve( + BytesInput.from(stored), codec, codecFactory, decryptor, FILE_AAD, 1, 2, range.getOffset() + 1)) + .isInstanceOf(ParquetCryptoRuntimeException.class); + codecFactory.release(); + } + + /** + * Two self-references with identical payloads in the same column chunk sit at different offsets, so + * their AADs differ and their ciphertexts must not be interchangeable. + */ + @Test + public void testIdenticalPayloadsAtDifferentOffsetsAreNotInterchangeable() throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + byte[] resolved = highlyCompressiblePayload(1024); + CompressionCodecName codec = CompressionCodecName.SNAPPY; + + BlockCipher.Encryptor encryptor = ModuleCipherFactory.getEncryptor(AesMode.GCM, COLUMN_KEY); + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + SelfReferenceStorage.StoredRange first = SelfReferenceStorage.write( + BytesInput.from(resolved), codecFactory.getCompressor(codec), encryptor, FILE_AAD, 1, 2, out); + SelfReferenceStorage.StoredRange second = SelfReferenceStorage.write( + BytesInput.from(resolved), codecFactory.getCompressor(codec), encryptor, FILE_AAD, 1, 2, out); + + assertThat(second.getOffset()).isGreaterThan(first.getOffset()); + + byte[] fileBytes = out.toByteArray(); + byte[] firstStored = + Arrays.copyOfRange(fileBytes, (int) first.getOffset(), (int) (first.getOffset() + first.getSize())); + + // The first block's bytes cannot be resolved at the second block's offset. + BlockCipher.Decryptor decryptor = ModuleCipherFactory.getDecryptor(AesMode.GCM, COLUMN_KEY); + assertThatThrownBy(() -> SelfReferenceStorage.resolve( + BytesInput.from(firstStored), + codec, + codecFactory, + decryptor, + FILE_AAD, + 1, + 2, + second.getOffset())) + .isInstanceOf(ParquetCryptoRuntimeException.class); + codecFactory.release(); + } + @Test public void testEmptyPayloadRoundTrip() throws IOException { CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); @@ -165,7 +270,6 @@ public void testEmptyPayloadRoundTrip() throws IOException { null, 0, 0, - 0L, out); assertThat(range.getSize()).isEqualTo(0L); @@ -175,6 +279,25 @@ public void testEmptyPayloadRoundTrip() throws IOException { codecFactory.release(); } + /** An empty payload round-trips through a real codec too, not only UNCOMPRESSED. */ + @ParameterizedTest + @EnumSource( + value = CompressionCodecName.class, + names = {"SNAPPY", "GZIP", "ZSTD", "LZ4_RAW"}) + public void testEmptyPayloadRoundTripCompressed(CompressionCodecName codec) throws IOException { + CodecFactory codecFactory = new CodecFactory(new Configuration(), PAGE_SIZE); + InMemoryPositionOutputStream out = new InMemoryPositionOutputStream(); + SelfReferenceStorage.StoredRange range = SelfReferenceStorage.write( + BytesInput.from(new byte[0]), codecFactory.getCompressor(codec), null, null, 0, 0, out); + + byte[] stored = out.toByteArray(); + assertThat(range.getSize()).isEqualTo((long) stored.length); + BytesInput resolvedBack = + SelfReferenceStorage.resolve(BytesInput.from(stored), codec, codecFactory, null, null, 0, 0, 0L); + assertThat(resolvedBack.toByteArray()).isEmpty(); + codecFactory.release(); + } + private static byte[] highlyCompressiblePayload(int length) { byte[] payload = new byte[length]; for (int i = 0; i < length; i++) {