[SPARK-23415][SQL][TEST] Make behavior of BufferHolderSparkSubmitSuite correct and stable

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/BufferHolder.java

@@ -35,6 +35,7 @@ final class BufferHolder {
 
   private static final int ARRAY_MAX = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH;
 
+  // buffer is guarantee to be word-aligned since UnsafeRow assumes each field is word-aligned.
   private byte[] buffer;
   private int cursor = Platform.BYTE_ARRAY_OFFSET;
   private final UnsafeRow row;
@@ -52,7 +53,8 @@ final class BufferHolder {
           "too many fields (number of fields: " + row.numFields() + ")");
     }
     this.fixedSize = bitsetWidthInBytes + 8 * row.numFields();
-    this.buffer = new byte[fixedSize + initialSize];
+    int roundedSize = ByteArrayMethods.roundNumberOfBytesToNearestWord(fixedSize + initialSize);
+    this.buffer = new byte[roundedSize];
     this.row = row;
     this.row.pointTo(buffer, buffer.length);
   }
@@ -61,6 +63,8 @@ final class BufferHolder {
    * Grows the buffer by at least neededSize and points the row to the buffer.
    */
   void grow(int neededSize) {
+    assert neededSize >= 0 :
+      "Cannot grow BufferHolder by size " + neededSize + " because the size is negative";
     if (neededSize > ARRAY_MAX - totalSize()) {
       throw new UnsupportedOperationException(
         "Cannot grow BufferHolder by size " + neededSize + " because the size after growing " +
@@ -70,7 +74,8 @@ void grow(int neededSize) {
     if (buffer.length < length) {
       // This will not happen frequently, because the buffer is re-used.
       int newLength = length < ARRAY_MAX / 2 ? length * 2 : ARRAY_MAX;
-      final byte[] tmp = new byte[newLength];
+      int roundedSize = ByteArrayMethods.roundNumberOfBytesToNearestWord(newLength);
+      final byte[] tmp = new byte[roundedSize];
       Platform.copyMemory(
         buffer,
         Platform.BYTE_ARRAY_OFFSET,

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/BufferHolderSparkSubmitSuite.scala

@@ -39,8 +39,8 @@ class BufferHolderSparkSubmitSuite
     val argsForSparkSubmit = Seq(
       "--class", BufferHolderSparkSubmitSuite.getClass.getName.stripSuffix("$"),
       "--name", "SPARK-22222",
-      "--master", "local-cluster[2,1,1024]",
-      "--driver-memory", "4g",
+      "--master", "local-cluster[1,1,7168]",
+      "--driver-memory", "7g",
       "--conf", "spark.ui.enabled=false",
       "--conf", "spark.master.rest.enabled=false",
       "--conf", "spark.driver.extraJavaOptions=-ea",
@@ -55,22 +55,30 @@ object BufferHolderSparkSubmitSuite {
 
     val ARRAY_MAX = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
 
-    val holder = new BufferHolder(new UnsafeRow(1000))
+    val unsafeRow = new UnsafeRow(1000)
+    val holder = new BufferHolder(unsafeRow)
 
     holder.reset()
-    holder.grow(roundToWord(ARRAY_MAX / 2))
 
-    holder.reset()
-    holder.grow(roundToWord(ARRAY_MAX / 2 + 8))
+    // while to reuse a buffer may happen, this test checks whether the buffer can be grown
+    holder.grow(ARRAY_MAX / 2)
+    assert(unsafeRow.getSizeInBytes % 8 == 0)
 
-    holder.reset()
-    holder.grow(roundToWord(Integer.MAX_VALUE / 2))
+    holder.grow(ARRAY_MAX / 2 + 7)
+    assert(unsafeRow.getSizeInBytes % 8 == 0)
 
-    holder.reset()
-    holder.grow(roundToWord(Integer.MAX_VALUE))
-  }
+    holder.grow(Integer.MAX_VALUE / 2)
+    assert(unsafeRow.getSizeInBytes % 8 == 0)
+
+    holder.grow(ARRAY_MAX - holder.totalSize())
+    assert(unsafeRow.getSizeInBytes % 8 == 0)
 
-  private def roundToWord(len: Int): Int = {
-    ByteArrayMethods.roundNumberOfBytesToNearestWord(len)
+    try {
+      holder.grow(ARRAY_MAX + 1 - holder.totalSize())
+      assert(false)
+    } catch {
+        case _: UnsupportedOperationException => assert(true)
+        case _: Throwable => assert(false)
+    }
   }
 }