This is an automated email from the ASF dual-hosted git repository. jamesbognar pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/juneau.git
commit ac39eb933888f3145d748b01311ed3d22b46fef4 Author: James Bognar <[email protected]> AuthorDate: Sun Aug 16 13:40:28 2026 -0400 READY-386: Add aggregate decompression budget to the Parquet parser The page loop capped per-page size and page count but never summed decompressed bytes, so many valid pages could still trigger unbounded aggregate decompression. Charge each page against a per-parse budget (new maxDecompressedBytes, default 64 MiB). --- .../juneau/marshall/parquet/ParquetConfig.java | 13 ++++ .../marshall/parquet/ParquetConfigAnnotation.java | 1 + .../juneau/marshall/parquet/ParquetParser.java | 37 +++++++++++- .../marshall/parquet/ParquetParserSession.java | 66 +++++++++++++++++--- .../parquet/ParquetParser_MaxLength_Test.java | 70 +++++++++++++++++++++- 5 files changed, 175 insertions(+), 12 deletions(-) diff --git a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfig.java b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfig.java index 29ef57c46a..ce600a1028 100644 --- a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfig.java +++ b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfig.java @@ -84,6 +84,19 @@ public @interface ParquetConfig { */ String maxInputLength() default ""; + /** + * Maximum allowed aggregate decompressed-byte total across every Parquet page read during a single + * parse. + * + * <p> + * Guards against a decompression bomb assembled from many pages that each individually pass the + * per-page {@link #maxLength()} cap. Default is <js>"67108864"</js> (64 MiB). A value of + * <js>"0"</js> or less disables the cap. + * + * @return The annotation value. + */ + String maxDecompressedBytes() default ""; + /** Rank for application order. */ int rank() default 0; } diff --git a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfigAnnotation.java b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfigAnnotation.java index bfe72d6d32..fa89863616 100644 --- a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfigAnnotation.java +++ b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetConfigAnnotation.java @@ -85,6 +85,7 @@ public class ParquetConfigAnnotation { integer(a.maxLength(), "maxLength").ifPresent(b::maxLength); integer(a.maxCount(), "maxCount").ifPresent(b::maxCount); integer(a.maxInputLength(), "maxInputLength").ifPresent(b::maxInputLength); + integer(a.maxDecompressedBytes(), "maxDecompressedBytes").ifPresent(b::maxDecompressedBytes); } } diff --git a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParser.java b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParser.java index 10a90e1db9..e9f1133392 100644 --- a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParser.java +++ b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParser.java @@ -51,6 +51,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi private static final int DEFAULT_MAX_LENGTH = ParquetParserSession.DEFAULT_MAX_LENGTH; private static final int DEFAULT_MAX_COUNT = ParquetParserSession.DEFAULT_MAX_COUNT; private static final int DEFAULT_MAX_INPUT_LENGTH = ParquetParserSession.DEFAULT_MAX_INPUT_LENGTH; + private static final int DEFAULT_MAX_DECOMPRESSED_BYTES = ParquetParserSession.DEFAULT_MAX_DECOMPRESSED_BYTES; private final Map<BeanPropertyMeta,ParquetBeanPropertyMeta> parquetBeanPropertyMetas = new ConcurrentHashMap<>(); private final Map<ClassMeta<?>,ParquetClassMeta> parquetClassMetas = new ConcurrentHashMap<>(); @@ -69,6 +70,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi private int maxLength = DEFAULT_MAX_LENGTH; private int maxCount = DEFAULT_MAX_COUNT; private int maxInputLength = DEFAULT_MAX_INPUT_LENGTH; + private int maxDecompressedBytes = DEFAULT_MAX_DECOMPRESSED_BYTES; /** * Constructor, default settings. @@ -79,6 +81,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi maxLength = env("ParquetParser.maxLength", DEFAULT_MAX_LENGTH); maxCount = env("ParquetParser.maxCount", DEFAULT_MAX_COUNT); maxInputLength = env("ParquetParser.maxInputLength", DEFAULT_MAX_INPUT_LENGTH); + maxDecompressedBytes = env("ParquetParser.maxDecompressedBytes", DEFAULT_MAX_DECOMPRESSED_BYTES); } protected Builder(Builder copyFrom) { @@ -87,6 +90,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi maxLength = copyFrom.maxLength; maxCount = copyFrom.maxCount; maxInputLength = copyFrom.maxInputLength; + maxDecompressedBytes = copyFrom.maxDecompressedBytes; } protected Builder(ParquetParser copyFrom) { @@ -95,6 +99,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi maxLength = copyFrom.maxLength; maxCount = copyFrom.maxCount; maxInputLength = copyFrom.maxInputLength; + maxDecompressedBytes = copyFrom.maxDecompressedBytes; } /** @@ -161,6 +166,24 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi return this; } + /** + * The maximum allowed aggregate decompressed-byte total across every Parquet page read during a + * single parse. + * + * <p> + * Guards against a decompression bomb assembled from many pages that each individually pass the + * per-page {@link #maxLength(int)} cap: this ceiling charges every page's declared uncompressed + * size against a running per-parse total, so a chunk (or a whole file) whose pages sum past the + * ceiling is rejected with a clean parse error instead of buffering the excess. + * + * @param value The maximum aggregate decompressed-byte total. Default is 64 MiB. Values ≤ 0 disable the cap. + * @return This object. + */ + public Builder maxDecompressedBytes(int value) { + maxDecompressedBytes = value; + return this; + } + @Override /* InputStreamParser.Builder<?> */ public Builder copy() { return new Builder(this); @@ -173,7 +196,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi @Override public HashKey hashKey() { - return HashKey.of(super.hashKey(), nullKeyString, maxLength, maxCount, maxInputLength); + return HashKey.of(super.hashKey(), nullKeyString, maxLength, maxCount, maxInputLength, maxDecompressedBytes); } } @@ -190,6 +213,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi private final int maxLength; private final int maxCount; private final int maxInputLength; + private final int maxDecompressedBytes; /** * Constructor. @@ -202,6 +226,7 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi maxLength = builder.maxLength; maxCount = builder.maxCount; maxInputLength = builder.maxInputLength; + maxDecompressedBytes = builder.maxDecompressedBytes; } /** @@ -231,6 +256,16 @@ public class ParquetParser extends InputStreamParser implements ParquetMetaProvi return maxInputLength; } + /** + * Returns the maximum allowed aggregate decompressed-byte total across every page read during a + * single parse. + * + * @return The maximum aggregate decompressed-byte total. + */ + public int getMaxDecompressedBytes() { + return maxDecompressedBytes; + } + @Override /* Overridden from Context */ public Builder copy() { return new Builder(this); diff --git a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParserSession.java b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParserSession.java index bf032bf61d..b3f349500f 100644 --- a/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParserSession.java +++ b/juneau-core/juneau-marshall/src/main/java/org/apache/juneau/marshall/parquet/ParquetParserSession.java @@ -150,6 +150,17 @@ public class ParquetParserSession extends InputStreamParserSession implements Re return v <= 0 ? Long.MAX_VALUE : v; } + /** + * Resolves the configured {@link ParquetParser#getMaxDecompressedBytes()} to an effective bound, + * translating the {@code <= 0} "disabled" sentinel to {@link Long#MAX_VALUE}. + * + * @return The effective aggregate decompressed-byte budget. + */ + private long effectiveMaxDecompressedBytes() { + var v = ctx.getMaxDecompressedBytes(); + return v <= 0 ? Long.MAX_VALUE : v; + } + /** * Opens a whole-value pull-parser cursor over a Parquet document, bound to this live session. * {@link RecordReader#read(Class) read(...)} delegates to the polymorphic @@ -372,6 +383,13 @@ public class ParquetParserSession extends InputStreamParserSession implements Re */ static final int DEFAULT_MAX_INPUT_LENGTH = 256 * 1024 * 1024; + /** + * Default cap (64 MiB) on the aggregate decompressed-byte total across every page read during a + * single parse, used when {@link ParquetParser#getMaxDecompressedBytes()} has not been explicitly + * configured. + */ + static final int DEFAULT_MAX_DECOMPRESSED_BYTES = 64 * 1024 * 1024; + /** * Clamps an untrusted row-count to a sane initial-capacity ceiling before it is handed to an * {@link ArrayList} constructor. Defense in depth: the count is already bounded upstream (see @@ -414,6 +432,32 @@ public class ParquetParserSession extends InputStreamParserSession implements Re /** Sentinel for a null intermediate OPTIONAL group at the given def level (GAP-14 multi-level nesting). */ private record GroupNull(int defLevel) {} + /** + * Mutable per-parse aggregate decompressed-byte budget, threaded through every {@code codec.decompress} + * call site (plain, list, and map column-chunk page reads) so many pages that each individually pass + * the per-page {@code maxLength} cap cannot together accumulate into an unbounded decompression bomb. + */ + private static final class DecompressionBudget { + + private long remaining; + + DecompressionBudget(long remaining) { + this.remaining = remaining; + } + + /** + * Charges {@code uncompressedSize} bytes against the running budget before the page is actually + * decompressed, so an over-budget page fails fast without performing the decompression work. + * + * @throws ParseException If the charge would drive the remaining budget negative. + */ + void charge(long uncompressedSize, String columnPath) throws ParseException { + remaining -= uncompressedSize; + if (remaining < 0) + throw new ParseException("Column '%s' exceeds the aggregate decompressed-byte budget", columnPath); + } + } + private record FileMeta(long numRows, List<RowGroupMeta> rowGroups, Map<String,Integer> schemaRepetition, Set<String> rawByteArrayPaths, Set<String> uuidPaths, Map<String,ColumnLogical> columnLogical) {} private record RowGroupMeta(long numRows, List<ColumnChunkMeta> columns) {} private record ColumnChunkMeta(int type, List<String> pathInSchema, int codec, long numValues, long dataPageOffset, long totalCompressedSize) {} @@ -776,6 +820,9 @@ public class ParquetParserSession extends InputStreamParserSession implements Re // per-group count wasn't recorded (older Juneau-written files). var maxCount = effectiveMaxCount(); var allRows = new ArrayList<Object>((int)Math.min(meta.numRows(), maxCount)); + // Shared across every row group/column in this parse (see DecompressionBudget): per-page maxLength + // alone can't stop many individually-valid pages from summing to a decompression bomb. + var budget = new DecompressionBudget(effectiveMaxDecompressedBytes()); // Each per-row-group num_rows is untrusted and drives ArrayList pre-allocation and reassembly loops // downstream. Bound every group count (and the running total) against the same configured ceiling // used for the file-level total above, before it is used to size anything, so a tiny footer @@ -788,7 +835,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re if (totalGroupRows > maxCount) throw new ParseException("Total row group numRows %s exceeds maximum allowed %s", totalGroupRows, maxCount); int groupRows = (int)group.numRows(); - allRows.addAll(readRowGroupRows(fileBytes, group, groupRows, elementType, schemaRepetition, rawByteArrayPaths, uuidPaths, columnLogical)); + allRows.addAll(readRowGroupRows(fileBytes, group, groupRows, elementType, schemaRepetition, rawByteArrayPaths, uuidPaths, columnLogical, budget)); } return allRows; } @@ -796,7 +843,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re @SuppressWarnings({ "java:S107" // Parser-internal method threads decode state (column paths, schema repetition, logical types); parameter count is intentional. }) - private List<?> readRowGroupRows(byte[] fileBytes, RowGroupMeta group, int numRows, ClassMeta<?> elementType, Map<String,Integer> schemaRepetition, Set<String> rawByteArrayPaths, Set<String> uuidPaths, Map<String,ColumnLogical> columnLogical) throws ParseException { + private List<?> readRowGroupRows(byte[] fileBytes, RowGroupMeta group, int numRows, ClassMeta<?> elementType, Map<String,Integer> schemaRepetition, Set<String> rawByteArrayPaths, Set<String> uuidPaths, Map<String,ColumnLogical> columnLogical, DecompressionBudget budget) throws ParseException { var columnData = new LinkedHashMap<String,List<Object>>(); var maxLength = effectiveMaxLength(); var maxCount = effectiveMaxCount(); @@ -805,11 +852,11 @@ public class ParquetParserSession extends InputStreamParserSession implements Re List<Object> values; var trim = isTrimStrings(); if (isListColumnPath(path)) - values = readListColumnChunk(fileBytes, cc, numRows, trim, maxLength, maxCount); + values = readListColumnChunk(fileBytes, cc, numRows, trim, maxLength, maxCount, budget); else if (isMapKeyValueColumnPath(path)) - values = readMapKeyValueColumnChunk(fileBytes, cc, numRows, trim, maxLength, maxCount); + values = readMapKeyValueColumnChunk(fileBytes, cc, numRows, trim, maxLength, maxCount, budget); else - values = readColumnChunk(fileBytes, cc, numRows, schemaRepetition, rawByteArrayPaths, uuidPaths, columnLogical, trim, maxLength, maxCount); + values = readColumnChunk(fileBytes, cc, numRows, schemaRepetition, rawByteArrayPaths, uuidPaths, columnLogical, trim, maxLength, maxCount, budget); columnData.put(path, values); } if (parquetDebug()) { @@ -852,7 +899,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re return idx < 0 ? listColumnPath : listColumnPath.substring(0, idx); } - private static List<Object> readListColumnChunk(byte[] fileBytes, ColumnChunkMeta cc, int numRows, boolean trimStrings, int maxLength, long maxCount) throws ParseException { + private static List<Object> readListColumnChunk(byte[] fileBytes, ColumnChunkMeta cc, int numRows, boolean trimStrings, int maxLength, long maxCount, DecompressionBudget budget) throws ParseException { try { var path = String.join(".", cc.pathInSchema()); var rowRelPath = rowRelativePath(path); @@ -872,6 +919,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re var chunkPath = String.join(".", cc.pathInSchema()); int dataPageOff = skipToDataPage(fileBytes, (int)cc.dataPageOffset(), chunkPath, maxLength, maxCount); var ph = readPageHeader(fileBytes, dataPageOff, chunkPath, maxLength, maxCount); + budget.charge(ph.uncompressedSize(), chunkPath); var compressedData = Arrays.copyOfRange(fileBytes, ph.bodyStart(), ph.bodyStart() + ph.compressedSize()); var decompressed = codec.decompress(compressedData, ph.uncompressedSize()); @@ -1021,7 +1069,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re private static final int MAP_MAX_DEF = 2; private static final int MAP_MAX_REP = 1; - private static List<Object> readMapKeyValueColumnChunk(byte[] fileBytes, ColumnChunkMeta cc, int numRows, boolean trimStrings, int maxLength, long maxCount) throws ParseException { + private static List<Object> readMapKeyValueColumnChunk(byte[] fileBytes, ColumnChunkMeta cc, int numRows, boolean trimStrings, int maxLength, long maxCount, DecompressionBudget budget) throws ParseException { try { int maxDef = MAP_MAX_DEF; int maxRep = MAP_MAX_REP; @@ -1038,6 +1086,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re var chunkPath = String.join(".", cc.pathInSchema()); int dataPageOff = skipToDataPage(fileBytes, (int)cc.dataPageOffset(), chunkPath, maxLength, maxCount); var ph = readPageHeader(fileBytes, dataPageOff, chunkPath, maxLength, maxCount); + budget.charge(ph.uncompressedSize(), chunkPath); var compressedData = Arrays.copyOfRange(fileBytes, ph.bodyStart(), ph.bodyStart() + ph.compressedSize()); var decompressed = codec.decompress(compressedData, ph.uncompressedSize()); @@ -1111,7 +1160,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re @SuppressWarnings({ "java:S107" // Parser-internal method threads decode state (column paths, schema repetition, logical types); parameter count is intentional. }) - private static List<Object> readColumnChunk(byte[] fileBytes, ColumnChunkMeta cc, int numRows, Map<String,Integer> schemaRepetition, Set<String> rawByteArrayPaths, Set<String> uuidPaths, Map<String,ColumnLogical> columnLogical, boolean trimStrings, int maxLength, long maxCount) throws ParseException { + private static List<Object> readColumnChunk(byte[] fileBytes, ColumnChunkMeta cc, int numRows, Map<String,Integer> schemaRepetition, Set<String> rawByteArrayPaths, Set<String> uuidPaths, Map<String,ColumnLogical> columnLogical, boolean trimStrings, int maxLength, long maxCount, DecompressionBudget budget) throws ParseException { try { if (cc.numValues() < 0 || cc.numValues() > maxCount) throw new ParseException("Invalid numValues for column '%s': %s", String.join(".", cc.pathInSchema()), cc.numValues()); @@ -1147,6 +1196,7 @@ public class ParquetParserSession extends InputStreamParserSession implements Re if (++pageGuard > MAX_PAGES_PER_CHUNK) throw new ParseException("Column '%s' exceeds the maximum of %s pages per chunk", path, MAX_PAGES_PER_CHUNK); var ph = readPageHeader(fileBytes, pageOff, path, maxLength, maxCount); + budget.charge(ph.uncompressedSize(), path); var compressedData = Arrays.copyOfRange(fileBytes, ph.bodyStart(), ph.bodyStart() + ph.compressedSize()); var decompressed = codec.decompress(compressedData, ph.uncompressedSize()); if (ph.pageType() == PAGE_DICTIONARY) { diff --git a/juneau-core/juneau-marshall/src/test/java/org/apache/juneau/marshall/parquet/ParquetParser_MaxLength_Test.java b/juneau-core/juneau-marshall/src/test/java/org/apache/juneau/marshall/parquet/ParquetParser_MaxLength_Test.java index e6c146c184..ba58c53a38 100644 --- a/juneau-core/juneau-marshall/src/test/java/org/apache/juneau/marshall/parquet/ParquetParser_MaxLength_Test.java +++ b/juneau-core/juneau-marshall/src/test/java/org/apache/juneau/marshall/parquet/ParquetParser_MaxLength_Test.java @@ -42,9 +42,12 @@ import org.junit.jupiter.api.*; * A full Parquet file embeds these values deep inside a Thrift-compact footer/page, so hand-crafting a * complete malformed file to exercise the end-to-end path is impractical for the low-level helpers; those * are covered directly (oversized rejected, in-range accepted, and the boundary). The configurable - * {@code maxLength}/{@code maxCount} knobs on {@link ParquetParser} themselves are covered end-to-end via - * real serialize/parse round trips (category <b>c</b>/<b>d</b> below), mirroring - * {@code MsgPackParser_MaxLength_Test}. + * {@code maxLength}/{@code maxCount}/{@code maxInputLength}/{@code maxDecompressedBytes} knobs on + * {@link ParquetParser} themselves are covered end-to-end via real serialize/parse round trips (category + * <b>c</b>/<b>d</b>/<b>e</b>/<b>f</b> below), mirroring {@code MsgPackParser_MaxLength_Test}. Category + * <b>f</b> additionally proves the <i>aggregate</i> decompressed-byte budget (as opposed to the per-page + * {@code maxLength} bound above): many pages that each individually pass {@code maxLength} must still be + * rejected once their decompressed sizes sum past the configured ceiling. */ @SuppressWarnings("unchecked") class ParquetParser_MaxLength_Test extends TestBase { @@ -254,4 +257,65 @@ class ParquetParser_MaxLength_Test extends TestBase { var parsed = (List<SimpleBean>) p.read(bytes, List.class, SimpleBean.class); assertEquals(3, parsed.size()); } + + // ================================================================ + // f. Configurable maxDecompressedBytes (aggregate decompressed-byte budget) + // ================================================================ + + /** + * Serializes {@code n} rows with a small {@code pageSize}, forcing the writer to split each column + * chunk into many small data pages (32 rows/page at the 1024-byte pageSize floor) instead of the + * single page a default-sized write produces. Every individual page stays far under the (huge + * default) {@code maxLength} bound; only their aggregate decompressed size is at risk. + */ + private static byte[] manyPagesBytes(int n) throws Exception { + return ParquetSerializer.create().pageSize(1024).build().write(beans(n)); + } + + @Test + void f01_configurableAggregateBudgetEnforcedEndToEnd() throws Exception { + var bytes = manyPagesBytes(2_000); + + // A generous cap parses fine even though the file has dozens of pages per column. + var lenient = ParquetParser.create().maxDecompressedBytes(1_000_000).build(); + var parsed = (List<SimpleBean>) lenient.read(bytes, List.class, SimpleBean.class); + assertEquals(2_000, parsed.size()); + + // Every individual page is tiny (well under the default 256 MiB maxLength), but a cap that only + // a handful of pages can satisfy on their own is still exceeded once their sizes accumulate. + var strict = ParquetParser.create().maxDecompressedBytes(2_000).build(); + assertThrowsWithMessage(ParseException.class, "aggregate decompressed-byte budget", + () -> strict.read(bytes, List.class, SimpleBean.class)); + } + + @Test + void f02_singlePageWithinBothCapsStillParses() throws Exception { + // A single-page file (default pageSize) comfortably under both the per-page and aggregate caps. + var bytes = ParquetSerializer.DEFAULT.write(beans(3)); + var p = ParquetParser.create().maxDecompressedBytes(1_024).build(); + var parsed = (List<SimpleBean>) p.read(bytes, List.class, SimpleBean.class); + assertEquals(3, parsed.size()); + } + + @Test + void f03_maxDecompressedBytesAffectsCacheKey() { + // Different maxDecompressedBytes values must NOT collide in the parser cache (hashKey wiring). + var p1 = ParquetParser.create().maxDecompressedBytes(100).build(); + var p2 = ParquetParser.create().maxDecompressedBytes(200).build(); + var p3 = ParquetParser.create().maxDecompressedBytes(100).build(); + assertNotSame(p1, p2); + assertSame(p1, p3); + assertEquals(100, p1.getMaxDecompressedBytes()); + assertEquals(200, p2.getMaxDecompressedBytes()); + } + + @Test + void f04_maxDecompressedBytesNonPositiveDisablesCap() throws Exception { + // A non-positive cap disables the aggregate-budget check, even for the many-small-pages file that + // a small positive cap (f01) would reject. + var bytes = manyPagesBytes(2_000); + var p = ParquetParser.create().maxDecompressedBytes(0).build(); + var parsed = (List<SimpleBean>) p.read(bytes, List.class, SimpleBean.class); + assertEquals(2_000, parsed.size()); + } }
