apache / apache/parquet-java

Buffer int overflow in CapacityByteArrayOutputStream, SnappyCompressor

Aperta
#2,528 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub
Component: Java Component: Parquet Priority: Major Type: bug
Lingua principale
Java
Stelle
3.1k
Fork
1.6k
Merge medio
3g 12h
PR unite (30g)
33

Descrizione

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.*

Guida per i contributori

Nessuna guida per i contributori indicizzata per questo repository

Direzione di ricerca

Inizia da CapacityByteArrayOutputStream.writeToOutput, SnappyCompressor.setInput e NonBlockedCompressorStream.write dello stack trace riportato, quindi traccia il percorso con cui i gruppi di righe di grandi dimensioni raggiungono il compressore. Riproduci il problema con un input di grandi dimensioni e confronta le dimensioni delle partizioni; il lavoro è completato quando viene identificato il limite di overflow e viene aggiunto un controllo di regressione che impedisca il fallimento dovuto a una capacità negativa.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Valutazione

Stack tecnologico
java
Ambito
data-engineering
Tipo di issue
Bug
Difficoltà
4/5
Tempo stimato
3-5 giorni
Stato di attività
Ferma
Chiarezza
Da chiarire
Idoneità per principianti
35/100

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.