[
https://issues.apache.org/jira/browse/FLINK-40245?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Nikita Singhvi updated FLINK-40245:
-----------------------------------
Description:
*Problem:*
FlinkParquetBuilder.createWriter() correctly reads parquet.page.size and
parquet.block.size from conf, but never calls
withMinRowCountForPageSizeCheck() / withMaxRowCountForPageSizeCheck().
This means parquet.page.size.row.check.min and parquet.page.size.row.check.max
are silently ignored regardless of what values are set.
Without frequent page size checks, a single row with a large binary field can
accumulate beyond Integer.MAX_VALUE bytes in the Parquet page buffer, causing:
{code:java}
java.lang.OutOfMemoryError: Size of data exceeded Integer.MAX_VALUE at
org.apache.parquet.bytes.CapacityByteArrayOutputStream.addSlab at
org.apache.flink.formats.parquet.row.ParquetRowDataWriter$BinaryWriter.write
{code}
*Fix:*
Add two missing method calls in FlinkParquetBuilder.createWriter():
{code:java}
.withMinRowCountForPageSizeCheck(getMinRowCountForPageSizeCheck(conf))
.withMaxRowCountForPageSizeCheck(getMaxRowCountForPageSizeCheck(conf)) {code}
*Steps to Reproduce:*
1. Configure a Flink job writing large binary fields to Parquet
2. Set parquet.page.size.row.check.min=1 and parquet.page.size.row.check.max=1
in conf
3. Observe that the settings have no effect — Parquet still uses defaults (100,
10000)
4. With large enough binary fields, OOM occurs
{code:java}
{code}
was:
*Problem:*
FlinkParquetBuilder.createWriter() correctly reads parquet.page.size and
parquet.block.size from conf, but never calls
withMinRowCountForPageSizeCheck() / withMaxRowCountForPageSizeCheck().
This means parquet.page.size.row.check.min and parquet.page.size.row.check.max
are silently ignored regardless of what values are set.
Without frequent page size checks, a single row with a large binary field can
accumulate beyond Integer.MAX_VALUE bytes in the Parquet page buffer, causing:
{code:java}
java.lang.OutOfMemoryError: Size of data exceeded Integer.MAX_VALUE at
org.apache.parquet.bytes.CapacityByteArrayOutputStream.addSlab at
org.apache.flink.formats.parquet.row.ParquetRowDataWriter$BinaryWriter.write
{code}
*Fix:*
Add two missing method calls in FlinkParquetBuilder.createWriter():
{code:java}
.withMinRowCountForPageSizeCheck(getMinRowCountForPageSizeCheck(conf))
.withMaxRowCountForPageSizeCheck(getMaxRowCountForPageSizeCheck(conf)) {code}
*Steps to Reproduce:*
1. Configure a Flink job writing large binary fields to Parquet
2. Set parquet.page.size.row.check.min=1 and parquet.page.size.row.check.max=1
in conf
3. Observe that the settings have no effect — Parquet still uses defaults (100,
10000)
4. With large enough binary fields, OOM occurs
{code:java}
{code}
> Wire parquet.page.size.row.check.min/max conf keys in FlinkParquetBuilder
> -------------------------------------------------------------------------
>
> Key: FLINK-40245
> URL: https://issues.apache.org/jira/browse/FLINK-40245
> Project: Flink
> Issue Type: Bug
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
> Reporter: Nikita Singhvi
> Priority: Major
>
> *Problem:*
> FlinkParquetBuilder.createWriter() correctly reads parquet.page.size and
> parquet.block.size from conf, but never calls
> withMinRowCountForPageSizeCheck() / withMaxRowCountForPageSizeCheck().
> This means parquet.page.size.row.check.min and
> parquet.page.size.row.check.max
> are silently ignored regardless of what values are set.
> Without frequent page size checks, a single row with a large binary field can
> accumulate beyond Integer.MAX_VALUE bytes in the Parquet page buffer,
> causing:
> {code:java}
> java.lang.OutOfMemoryError: Size of data exceeded Integer.MAX_VALUE at
> org.apache.parquet.bytes.CapacityByteArrayOutputStream.addSlab at
> org.apache.flink.formats.parquet.row.ParquetRowDataWriter$BinaryWriter.write
> {code}
>
> *Fix:*
> Add two missing method calls in FlinkParquetBuilder.createWriter():
> {code:java}
> .withMinRowCountForPageSizeCheck(getMinRowCountForPageSizeCheck(conf))
> .withMaxRowCountForPageSizeCheck(getMaxRowCountForPageSizeCheck(conf)) {code}
>
> *Steps to Reproduce:*
> 1. Configure a Flink job writing large binary fields to Parquet
> 2. Set parquet.page.size.row.check.min=1 and
> parquet.page.size.row.check.max=1 in conf
> 3. Observe that the settings have no effect — Parquet still uses defaults
> (100, 10000)
> 4. With large enough binary fields, OOM occurs
>
> {code:java}
> {code}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
