Have size limit cut off right at the correct number of characters.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/SizeLimitedWriter.scala

@@ -19,8 +19,8 @@ package org.apache.spark.sql.catalyst.util
 
 import java.io.Writer
 
-class WriterSizeException(val attemptedSize: Long, val charLimit: Long) extends Exception(
-  s"Attempted to write $attemptedSize characters to a writer that is limited to $charLimit")
+class WriterSizeException(val extraChars: Long, val charLimit: Long) extends Exception(
+  s"Writer reached limit of $charLimit characters.  $extraChars extra characters ignored.")
 
 /**
  * This class is used to control the size of generated writers.  Guarantees that the total number
@@ -31,15 +31,15 @@ class WriterSizeException(val attemptedSize: Long, val charLimit: Long) extends
  */
 class SizeLimitedWriter(underlying: Writer, charLimit: Long) extends Writer {
 
-  var charsWritten: Long = 0
+  private var charsWritten: Long = 0
 
   override def write(cbuf: Array[Char], off: Int, len: Int): Unit = {
-    val newLength = charsWritten + Math.min(cbuf.length - off, len)
-    if (newLength > charLimit) {
-      throw new WriterSizeException(newLength, charLimit)
+    val charsToWrite = Math.min(charLimit - charsWritten, len).toInt
+    underlying.write(cbuf, off, charsToWrite)
+    charsWritten += charsToWrite
+    if (charsToWrite < len) {
+      throw new WriterSizeException(len - charsToWrite, charLimit)
     }
-    charsWritten = newLength
-    underlying.write(cbuf, off, len)
   }
 
   override def flush(): Unit = underlying.flush()

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala

@@ -207,7 +207,8 @@ package object util extends Logging {
 
   def withSizeLimitedWriter[T](writer: Writer)(f: (Writer) => T): Option[T] = {
     try {
-      val limited = new SizeLimitedWriter(writer, SQLConf.get.maxPlanStringLength)
+	    // Subtract 3 from the string length to leave room for the "..."
+      val limited = new SizeLimitedWriter(writer, SQLConf.get.maxPlanStringLength - 3)
       Some(f(limited))
     }
     catch {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/trees/TreeNodeSuite.scala

@@ -597,13 +597,13 @@ class TreeNodeSuite extends SparkFunSuite {
     assert(result === expected)
   }
 
-  test("toString() tree depth") {
+  test("treeString limits plan length") {
     val ds = (1 until 100).foldLeft(Literal("TestLiteral"): Expression) { case (treeNode, x) =>
       Add(Literal(x), treeNode)
     }
 
     val planString = ds.treeString
     assert(planString.endsWith("..."))
-    assert(planString.length <= SQLConf.get.maxPlanStringLength)
+    assert(planString.length === SQLConf.get.maxPlanStringLength)
   }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/SizeLimitedWriterSuite.scala

@@ -33,13 +33,13 @@ class SizeLimitedWriterSuite extends SparkFunSuite {
     assert(writer.toString === "testtest")
   }
 
-  test("filter pattern") {
+  test("truncate at the limit") {
     val writer = new StringBuilderWriter()
     val limited = new SizeLimitedWriter(writer, 5)
     assertThrows[WriterSizeException] {
       limited.write("test")
       limited.write("test")
     }
-    assert(writer.toString === "test")
+    assert(writer.toString === "testt")
   }
 }