From 962f88a82696d013501201fcabc1f9439889f77c Mon Sep 17 00:00:00 2001 From: Nikita Singhvi Date: Mon, 20 Jul 2026 20:39:07 +0530 Subject: [PATCH] Wire parquet.page.size.row.check.min/max conf keys in FlinkParquetBuilder --- .../parquet/row/ParquetRowDataBuilder.java | 4 ++ .../parquet/row/ParquetRowDataWriterTest.java | 48 +++++++++++++++++++ 2 files changed, 52 insertions(+) diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/row/ParquetRowDataBuilder.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/row/ParquetRowDataBuilder.java index 5abb21d5a269c1..7b09ca00bbbe0d 100644 --- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/row/ParquetRowDataBuilder.java +++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/row/ParquetRowDataBuilder.java @@ -42,6 +42,8 @@ import static org.apache.parquet.hadoop.ParquetOutputFormat.getBlockSize; import static org.apache.parquet.hadoop.ParquetOutputFormat.getDictionaryPageSize; import static org.apache.parquet.hadoop.ParquetOutputFormat.getEnableDictionary; +import static org.apache.parquet.hadoop.ParquetOutputFormat.getMaxRowCountForPageSizeCheck; +import static org.apache.parquet.hadoop.ParquetOutputFormat.getMinRowCountForPageSizeCheck; import static org.apache.parquet.hadoop.ParquetOutputFormat.getPageSize; import static org.apache.parquet.hadoop.ParquetOutputFormat.getValidation; import static org.apache.parquet.hadoop.ParquetOutputFormat.getWriterVersion; @@ -143,6 +145,8 @@ public ParquetWriter createWriter(OutputFile out) throws IOException { .withDictionaryEncoding(getEnableDictionary(conf)) .withValidation(getValidation(conf)) .withWriterVersion(getWriterVersion(conf)) + .withMinRowCountForPageSizeCheck(getMinRowCountForPageSizeCheck(conf)) + .withMaxRowCountForPageSizeCheck(getMaxRowCountForPageSizeCheck(conf)) .withConf(conf) .build(); } diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/row/ParquetRowDataWriterTest.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/row/ParquetRowDataWriterTest.java index cd0bf0016672d7..f10803a4814bee 100644 --- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/row/ParquetRowDataWriterTest.java +++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/row/ParquetRowDataWriterTest.java @@ -483,4 +483,52 @@ private LocalDateTime toDateTime(Integer v) { v = (v > 0 ? v : -v) % 1000; return LocalDateTime.now().plusNanos(v).plusSeconds(v); } + + /** + * Verifies that {@code parquet.page.size.row.check.min} and {@code + * parquet.page.size.row.check.max} configuration keys are honoured by {@link + * ParquetRowDataBuilder}. + * + *

Before the fix, {@code FlinkParquetBuilder.createWriter()} never called {@code + * withMinRowCountForPageSizeCheck()} / {@code withMaxRowCountForPageSizeCheck()}, so these + * conf keys were silently ignored. A job writing rows with large binary fields could accumulate + * data across 100–10,000 rows (the defaults) before a page-size check, causing + * {@code java.lang.OutOfMemoryError: Size of data exceeded Integer.MAX_VALUE}. + */ + @Test + void testRowCountForPageSizeCheckConfIsHonoured(@TempDir java.nio.file.Path folder) + throws IOException { + // Use a tiny page size (4 KB) and check after every row so the writer + // flushes frequently — this would OOM with the defaults on large payloads. + Configuration conf = new Configuration(); + conf.setInt(ParquetOutputFormat.PAGE_SIZE, 4096); + conf.setInt(ParquetOutputFormat.MIN_ROW_COUNT_FOR_PAGE_SIZE_CHECK, 1); + conf.setInt(ParquetOutputFormat.MAX_ROW_COUNT_FOR_PAGE_SIZE_CHECK, 1); + + RowType rowType = + RowType.of(new VarBinaryType(VarBinaryType.MAX_LENGTH)); + + Path path = new Path(folder.toString(), UUID.randomUUID().toString()); + ParquetWriterFactory factory = + ParquetRowDataBuilder.createWriterFactory(rowType, conf, true); + BulkWriter writer = + factory.create(path.getFileSystem().create(path, FileSystem.WriteMode.OVERWRITE)); + + // Write rows with binary payloads larger than the page size. + // With row.check.min=1 the writer checks after every row and flushes before overflow. + byte[] largePayload = new byte[40_960]; // 40 KB — 10× the 4 KB page size + DataFormatConverters.DataFormatConverter converter = + (DataFormatConverters.DataFormatConverter) + DataFormatConverters.getConverterForDataType( + TypeConversions.fromLogicalToDataType(rowType)); + + // Write enough rows that the page buffer would overflow without frequent checks + RowData rowData = converter.toInternal(Row.of((Object) largePayload)); + for (int i = 0; i < 100; i++) { + writer.addElement(rowData); + } + // Should complete without OutOfMemoryError + writer.flush(); + writer.finish(); + } }