title: Spark Requested array size exceeds VM limit from BufferHolder.grow
body: <p>I'm getting this error on Spark 2.1 running on Hadoop cluster, on a mixed scala-python application (similar to Zeppelin):</p>
  
  <pre><code>18/04/09 08:19:34 ERROR Utils: Uncaught exception in thread stdout writer for /x/python/miniconda/bin/python
  java.lang.OutOfMemoryError: Requested array size exceeds VM limit
      at org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder.grow(BufferHolder.java:73)
      at org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter.write(UnsafeRowWriter.java:214)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply6_4$(Unknown Source)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply7_16$(Unknown Source)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
      at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateResultProjection$1.apply(AggregationIterator.scala:232)
      at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateResultProjection$1.apply(AggregationIterator.scala:221)
      at org.apache.spark.sql.execution.aggregate.SortBasedAggregationIterator.next(SortBasedAggregationIterator.scala:159)
      at org.apache.spark.sql.execution.aggregate.SortBasedAggregationIterator.next(SortBasedAggregationIterator.scala:29)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at scala.collection.Iterator$GroupedIterator.takeDestructively(Iterator.scala:1076)
      at scala.collection.Iterator$GroupedIterator.go(Iterator.scala:1091)
      at scala.collection.Iterator$GroupedIterator.fill(Iterator.scala:1129)
      at scala.collection.Iterator$GroupedIterator.hasNext(Iterator.scala:1132)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:504)
      at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3.apply(PythonRDD.scala:328)
      at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1951)
      at org.apache.spark.api.python.PythonRunner$WriterThread.run(PythonRDD.scala:269)
  </code></pre>
  
  <p>It seems strange that this kind of error is thrown from <code>BufferHolder.grow</code> as it contains explicit check:</p>
  
  <pre><code>if (neededSize &gt; Integer.MAX_VALUE - totalSize()) {
    throw new UnsupportedOperationException(
      "Cannot grow BufferHolder by size " + neededSize + " because the size after growing " +
        "exceeds size limitation " + Integer.MAX_VALUE);
  }
  </code></pre>
  
  <p>However on runtime it gets through this assertion to initialize an array with size greater than Integer.MAX_VALUE (line 73). This error does not seem to be related to configuration tuning (please correct me if I'm wrong), so I'd skip the specifications of the application/cluster, except for - 150 executors, 2 cores each. <code>spark.sql.shuffle.partitions</code> set to 8000 as an attempt to eliminate shuffle skew.</p>
  
  <p>The parent RDD of PythonRDD is actually a DataFrame which is the result of a shuffle, which has ~30 columns, one of them is a very big String type (up to 100MB, but avg 150KB). I'm mentioning this because from the stack trace it looks like the error is raised somewhere between the shuffle read and the PythonRDD. Also, <strong>this always happens at the last 10% of the partitions</strong> (the input data is static), and the first 90% finish without errors.</p>
  
  <p>Has anyone experienced this issue? Or can shed some light on it?</p>