[BUG] Task may fail when reading parquet files that contains MapType from s3a file system

[wjxiz1992]

Describe the bug
When reading parquet file that contains MapType, this error is observed:

Job aborted due to stage failure: Task 1335 in stage 7.0 failed 4 times, most recent failure: Lost task 1335.3 in stage 7.0 (TID 2515, 10.148.44.12, executor 23): ai.rapids.cudf.CudfException: cuDF failure at: /ansible-managed/jenkins-slave/slave4/workspace/spark/cudf18_nightly/cpp/src/structs/structs_column_factories.cu:94: Child columns must have the same number of rows as the Struct column.
	at ai.rapids.cudf.Table.readParquet(Native Method)
	at ai.rapids.cudf.Table.readParquet(Table.java:683)
	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.$anonfun$readBufferToTable$2(GpuParquetScan.scala:1533)
	at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:28)
	at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:26)
	at com.nvidia.spark.rapids.FileParquetPartitionReaderBase.withResource(GpuParquetScan.scala:477)
	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.$anonfun$readBufferToTable$1(GpuParquetScan.scala:1532)
	at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:28)
	at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:26)
	at com.nvidia.spark.rapids.FileParquetPartitionReaderBase.withResource(GpuParquetScan.scala:477)
	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.readBufferToTable(GpuParquetScan.scala:1520)
	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.readBatch(GpuParquetScan.scala:1399)
	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.$anonfun$next$2(GpuParquetScan.scala:1449)
	at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:28)
	at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:26)
	at com.nvidia.spark.rapids.FileParquetPartitionReaderBase.withResource(GpuParquetScan.scala:477)
	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.next(GpuParquetScan.scala:1422)
	at com.nvidia.spark.rapids.PartitionIterator.hasNext(GpuDataSourceRDD.scala:59)
	at com.nvidia.spark.rapids.MetricsBatchIterator.hasNext(GpuDataSourceRDD.scala:76)

2 SQL charts from 1 successful run and a failed run:

Steps/Code to reproduce bug

It's one of our LHA queries that run on NGC.
It's not 100% reproduced during several runs.
Also, this error will not 100% make the job fail, sometimes a retry task will succeed, and the whole application will be finished successfully.

Environment details (please complete the following information)

• Environment location: Nvidia NGC
• Spark configuration settings related to the issue :

--executor-memory 32G \
--executor-cores 8 \
--driver-memory 16G \
--conf spark.cores.max=224 \
--conf spark.local.dir=${SPARK_LOCAL_DIR} \
--conf spark.eventLog.enabled=true \
--conf spark.driver.host=${SPARK_MASTER_HOST} \
--conf spark.driver.extraClassPath=${SPARK_CUDF_JAR}:${SPARK_RAPIDS_PLUGIN_JAR} \
--conf spark.executor.extraClassPath=/usr/lib:/usr/lib/ucx:${SPARK_CUDF_JAR}:${SPARK_RAPIDS_PLUGIN_JAR} \
--conf spark.executor.extraJavaOptions=-Djava.io.tmpdir=${SPARK_LOCAL_DIR}\ -Dai.rapids.cudf.prefer-pinned=true \
--conf spark.sql.broadcastTimeout=7200 \
--conf spark.sql.adaptive.enabled=false \
--conf spark.rapids.sql.explain=true \
--conf spark.sql.files.maxPartitionBytes=2G \
--conf spark.network.timeout=3600s \
--conf spark.executor.resource.gpu.vendor=nvidia.com \
--conf spark.executor.resource.gpu.amount=1 \
--conf spark.task.resource.gpu.amount=0.1 \
--conf spark.rapids.sql.concurrentGpuTasks=2 \
--conf spark.task.cpus=1 \
--conf spark.locality.wait=0 \
--conf spark.shuffle.manager=com.nvidia.spark.rapids.spark301.RapidsShuffleManager \
--conf spark.shuffle.service.enabled=false \
--conf spark.rapids.shuffle.maxMetadataSize=1MB \
--conf spark.rapids.shuffle.transport.enabled=true \
--conf spark.rapids.shuffle.compression.codec=none \
--conf spark.executorEnv.UCX_TLS=cuda_copy,cuda_ipc,rc \
--conf spark.executorEnv.UCX_ERROR_SIGNALS= \
--conf spark.executorEnv.UCX_MAX_RNDV_RAILS=1 \
--conf spark.executorEnv.UCX_CUDA_IPC_CACHE=y \
--conf spark.executorEnv.UCX_MEMTYPE_CACHE=n \
--conf spark.executorEnv.LD_LIBRARY_PATH=$LD_LIBRARY_PATH \
--conf spark.rapids.shuffle.ucx.bounceBuffers.size=4MB \
--conf spark.rapids.shuffle.ucx.bounceBuffers.device.count=32 \
--conf spark.rapids.shuffle.ucx.bounceBuffers.host.count=32 \
--conf spark.sql.shuffle.partitions=200 \
--conf spark.plugins=com.nvidia.spark.SQLPlugin \
--conf spark.rapids.memory.pinnedPool.size=4G \
--conf spark.rapids.memory.host.spillStorageSize=12G \
--conf spark.rapids.memory.gpu.pool=ARENA \
--conf spark.rapids.memory.gpu.allocFraction=0.1 \
--conf spark.rapids.memory.gpu.maxAllocFraction=1 \
--conf spark.rapids.sql.castFloatToString.enabled=true \
--conf spark.rapids.sql.castStringToFloat.enabled=true \
--conf spark.rapids.sql.castStringToInteger.enabled=true \
--conf spark.rapids.sql.variableFloatAgg.enabled=true \
--conf spark.rapids.sql.incompatibleOps.enabled=true \
--conf spark.hadoop.fs.s3a.access.key=$AWS_ACCESS_KEY_ID \
--conf spark.hadoop.fs.s3a.secret.key=$AWS_SECRET_ACCESS_KEY \
--conf spark.hadoop.fs.s3a.endpoint=swiftstack-maglev.ngc.nvidia.com \
--conf spark.hadoop.fs.s3a.path.style.access=true \
--conf spark.hadoop.fs.s3a.connection.maximum=256 \
...

Additional context
This is only a tracking issue for this error. There are some internal links and I will send a mail with them and more details.

[kuhushukla]

Looks like we might be dropping data and the row counts are going out of sync due to data loss.

[kuhushukla]

I am almost always seeing them with an associated error:

13:04:41  20/12/30 18:46:30 WARN TaskSetManager: Lost task 1623.0 in stage 7.0 (TID 2354, 10.148.108.165, executor 15): ai.rapids.cudf.CudfException: cuDF failure at: /ansible-managed/jenkins-slave/slave4/workspace/spark/cudf18_nightly/cpp/src/strings/strings_column_factories.cu:58: total size of strings is too large for cudf column
13:04:41  	at ai.rapids.cudf.Table.readParquet(Native Method)
13:04:41  	at ai.rapids.cudf.Table.readParquet(Table.java:683)
13:04:41  	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.$anonfun$readBufferToTable$2(GpuParquetScan.scala:1533)
13:04:41  	at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:28)
13:04:41  	at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:26)
13:04:41  	at com.nvidia.spark.rapids.FileParquetPartitionReaderBase.withResource(GpuParquetScan.scala:477)
13:04:41  	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.$anonfun$readBufferToTable$1(GpuParquetScan.scala:1532)
13:04:41  	at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:28)
13:04:41  	at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:26)
13:04:41  	at com.nvidia.spark.rapids.FileParquetPartitionReaderBase.withResource(GpuParquetScan.scala:477)
13:04:41  	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.readBufferToTable(GpuParquetScan.scala:1520)
13:04:41  	at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.readBatch(GpuParquetScan.scala:1399)

[wjxiz1992]

I just found the allocFraction was set to 0.1 in our NGC query jobs( this field is copied from an NGC job template). So I'm changing it back to 0.9 to see if it's related.

did the following tests:

• plugin 0.3 with 4 TB on NGC, spark.rapids.memory.gpu.allocFraction=0.1, => ai.rapids.cudf.CudfException: parallel_for failed: cudaErrorMemoryAllocation: out of memory
• plugin 0.3 with 4 TB on NGC, spark.rapids.memory.gpu.allocFraction=0.9, => no error
• plugin 0.4 with 4 TB on NGC, spark.rapids.memory.gpu.allocFraction=0.1, => about 25% possibility child column error
• plugin 0.4 with 4 TB on NGC, spark.rapids.memory.gpu.allocFraction=0.9 => didn't see errors till now (3 tries)

Not able to reproduce any error on Yarn with 0.3 or 0.4 plugin jar.

[revans2]

I can see why having the pool start out too small is a problem for memory allocation, but I don't see how it could result in a problem like we are seeing unless the parquet reader some how ignores a memory allocation error and just returns less data instead.

[jlowe]

Totally agree with @revans2. The alloc fraction sets the initial size of the GPU pool, not the total size, so setting it to a low value shouldn't cause more allocation failures relative to the 0.9 setting.

[kuhushukla]

Agree with the above comments.

plugin 0.4 with 4 TB on NGC, spark.rapids.memory.gpu.allocFraction=0.1, => about 25% possibility child column error

Do we ever see any indication of memory allocation failing for this case?

[revans2]

Agree with the above comments.

plugin 0.4 with 4 TB on NGC, spark.rapids.memory.gpu.allocFraction=0.1, => about 25% possibility child column error

Do we ever see any indication of memory allocation failing for this case?

Yes in 0.3 we get a cudaErrorMemoryAllocation: out of memory in 0.4 we get the corruption error. it looks like some change that went into 0.4 or cudf-0.18 that is causing this to happen.

[kuhushukla]

Thanks for clarifying @revans2 , maybe I was not clear. My question was mostly to confirm if we saw any allocation failure specifically besides the corruption in 0.4. Like you said, it might be that cudf is ignoring it altogether and showing up only as a corruption failure.

[wjxiz1992]

I went through the executor logs and found something related to s3a reading:

20/12/30 17:45:36 INFO GpuParquetMultiFilePartitionReaderFactory: Using the multi-threaded multi-file parquet reader, files: s3a://spark-data/LHA/meituan/q4/t2/dt=2020-08-14/hour=1/part-00016-bbad31bf-1b26-42fd-af92-cedde8d83359.c000.snappy.parquet task attemptid: 2448
.....
.....
20/12/30 17:47:57 INFO S3AInputStream: Got exception while trying to read from stream s3a://spark-data/LHA/meituan/q4/t2/dt=2020-08-14/hour=1/part-00016-bbad31bf-1b26-42fd-af92-cedde8d83359.c000.snappy.parquet trying to recover: com.amazonaws.thirdparty.apache.http.ConnectionClosedException: Premature end of Content-Length delimited message body (expected: 4342704529; received: 655360 
.....
.....
20/12/30 18:43:34 INFO S3AInputStream: Got exception while trying to read from stream s3a://spark-data/LHA/meituan/q4/t2/dt=2020-08-14/hour=1/part-00016-bbad31bf-1b26-42fd-af92-cedde8d83359.c000.snappy.parquet trying to recover: com.amazonaws.thirdparty.apache.http.ConnectionClosedException: Premature end of Content-Length delimited message body (expected: 385598675; received: 393216
.....
ai.rapids.cudf.CudfException: cuDF failure at: /ansible-managed/jenkins-slave/slave4/workspace/spark/cudf18_nightly/cpp/src/strings/strings_column_factories.cu:58: total size of strings is too large for cudf column
        at ai.rapids.cudf.Table.readParquet(Native Method)
        at ai.rapids.cudf.Table.readParquet(Table.java:683)
        at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.$anonfun$readBufferToTable$2(GpuParquetScan.scala:1533)
        at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:28)
        at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:26)
        at com.nvidia.spark.rapids.FileParquetPartitionReaderBase.withResource(GpuParquetScan.scala:477)
        at com.nvidia.spark.rapids.MultiFileCloudParquetPartitionReader.$anonfun$readBufferToTable$1(GpuParquetScan.scala:1532)
        at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:28)
        at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:26)
.....

I searched about this s3a issue, and found an answer here:https://stackoverflow.com/questions/9952815/s3-java-client-fails-a-lot-with-premature-end-of-content-length-delimited-messa. and fix for it: apache/hadoop#2497
According to this answer, this error happens when s3a fetcher client is created to many and causes GC on s3a side.
But during my dump, the parquet reader process is heavily slowed down by the "dump" action(it's NGC so I have to write them back to s3a). So I guess, maybe I cannot reproduce the child column error if the root cause is really the "s3a".

But anyway I'm now still trying to dump parquet files to find the broken one but that takes time. Only files that dumped from the failed run is useful. I'm still running the query.

[jlowe]

If we can't reproduce the issue by dumping the files beforehand, we could build a custom plugin that tries to dump the parquet buffer to disk only if the Table.readParquet call threw an exception. Then we won't be slowing down the main code path before the error occurs.