apache / apache/parquet-java

Buffer int overflow in CapacityByteArrayOutputStream, SnappyCompressor

未关闭
#2,528 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
Component: Java Component: Parquet Priority: Major Type: bug
主要语言
Java
星标
3.1k
派生
1.6k
平均合并
3 天 12 小时
30 天内合并 PR
33

描述

During an attempted write operation, a buffer position integer overflow is resulting in a IllegalArgumentException: Negative capacity: -2147336621 exception.

 
```

20/10/06 15:30:39 INFO HadoopRDD: Input split: s3a:///part-00015-96362e5d-d047-4f31-812b-38ff79f6919c-c000.txt.bz2:268435456+33554432
20/10/06 17:23:37 ERROR Utils: Aborting task
java.lang.IllegalArgumentException: Negative capacity: -2147336621
at java.nio.Buffer.(Buffer.java:199)
at java.nio.ByteBuffer.(ByteBuffer.java:281)
at java.nio.ByteBuffer.(ByteBuffer.java:289)
at java.nio.MappedByteBuffer.(MappedByteBuffer.java:89)
at java.nio.DirectByteBuffer.(DirectByteBuffer.java:119)
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:311)
at org.apache.parquet.hadoop.codec.SnappyCompressor.setInput(SnappyCompressor.java:97)
at org.apache.parquet.hadoop.codec.NonBlockedCompressorStream.write(NonBlockedCompressorStream.java:48)
at org.apache.parquet.bytes.CapacityByteArrayOutputStream.writeToOutput(CapacityByteArrayOutputStream.java:227)
at org.apache.parquet.bytes.CapacityByteArrayOutputStream.writeTo(CapacityByteArrayOutputStream.java:247)
at org.apache.parquet.bytes.BytesInput$CapacityBAOSBytesInput.writeAllTo(BytesInput.java:405)
at org.apache.parquet.bytes.BytesInput$SequenceBytesIn.writeAllTo(BytesInput.java:296)
at org.apache.parquet.hadoop.CodecFactory$HeapBytesCompressor.compress(CodecFactory.java:164)
at org.apache.parquet.hadoop.ColumnChunkPageWriteStore$ColumnChunkPageWriter.writePage(ColumnChunkPageWriteStore.java:95)
at org.apache.parquet.column.impl.ColumnWriterV1.writePage(ColumnWriterV1.java:147)
at org.apache.parquet.column.impl.ColumnWriterV1.flush(ColumnWriterV1.java:235)
at org.apache.parquet.column.impl.ColumnWriteStoreV1.flush(ColumnWriteStoreV1.java:122)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.flushRowGroupToStore(InternalParquetRecordWriter.java:172)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.checkBlockSizeReached(InternalParquetRecordWriter.java:148)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.write(InternalParquetRecordWriter.java:130)
at org.apache.parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:182)
at org.apache.parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:44)
at org.apache.spark.sql.execution.datasources.parquet.ParquetOutputWriter.write(ParquetOutputWriter.scala:40)
at org.apache.spark.sql.execution.datasources.SingleDirectoryDataWriter.write(FileFormatDataWriter.scala:137)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:242)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:239)
at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1394)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:245)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:169)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:168)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
```
2,147,483,647 (max int) - 2147336621 (negative capacity) = 147026.

The input bz2 files are all roughly 900 MiB in size. The target parquet part files are 1.7 GiB in size.

Increasing the partition count from 64 to 1024 fixes the issue. The output parquet part files drop to 100MiB in size.

However, it's unclear to me what the root cause is and why increasing partition count helps. Was it an unlucky row grouping that bumped the buffer size over by 147KB, i.e, any change up or down in parittion count would have helped? Is it approaching the parquet part file size limit?

This issue seems related to PARQUET-1632 but it's not using the ConcatenatingByteArrayCollector, which potentially means a distinct root cause. The input dataset does have large string columns (up to 10MB) but nothing close to the signed int max of 2.4G that was produced in PARQUET-1632.

 

 

 

 

**Reporter**: [Artem Shnayder](https://issues.apache.org/jira/secure/ViewProfile.jspa?name=shndr)

**Note**: *This issue was originally created as [PARQUET-1919](https://issues.apache.org/jira/browse/PARQUET-1919). Please see the [migration documentation](https://issues.apache.org/jira/browse/PARQUET-2502) for further details.*

贡献指南

这个仓库没有索引到贡献指南

调研方向

从报告的堆栈跟踪中的 CapacityByteArrayOutputStream.writeToOutput、SnappyCompressor.setInput 和 NonBlockedCompressorStream.write 开始,然后跟踪大型行组如何到达压缩器。使用大型输入重现问题并比较分区大小;完成的标准是确定溢出边界,并添加一个防止负容量失败的回归检查。

由索引模型根据 Issue 内容生成。

评估

技术栈
java
领域
data-engineering
Issue 类型
缺陷
难度
4/5
预计耗时
3-5 天
活跃度
停滞
描述清晰度
需要澄清
新手友好度
35/100

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。