Allow for maxCharsPerCol to be set

Currently, the maxCharactersPerColumn value is hardcoded to
100,000. I (sadly) have CSV fields with more than 100k characters so
need to be able to configure this.

This allows the value to be passed the same way as other paramters while
keeping a backward compatible default of 100k

Author: Addison Higham <ahigham@instructure.com>

Closes #307 from addisonj/master.

Author: Addison Higham

src/main/scala/com/databricks/spark/csv/CsvParser.scala

@@ -42,6 +42,7 @@ class CsvParser extends Serializable {
   private var codec: String = null
   private var nullValue: String = ""
   private var dateFormat: String = null
+  private var maxCharsPerCol: Int = 100000
 
   def withUseHeader(flag: Boolean): CsvParser = {
     this.useHeader = flag
@@ -123,6 +124,11 @@ class CsvParser extends Serializable {
     this
   }
 
+  def withMaxCharsPerCol(maxCharsPerCol: Int): CsvParser = {
+    this.maxCharsPerCol = maxCharsPerCol
+    this
+  }
+
   /** Returns a Schema RDD for the given CSV path. */
   @throws[RuntimeException]
   def csvFile(sqlContext: SQLContext, path: String): DataFrame = {
@@ -143,7 +149,8 @@ class CsvParser extends Serializable {
       inferSchema,
       codec,
       nullValue,
-      dateFormat)(sqlContext)
+      dateFormat,
+      maxCharsPerCol)(sqlContext)
     sqlContext.baseRelationToDataFrame(relation)
   }
 
@@ -165,7 +172,8 @@ class CsvParser extends Serializable {
       inferSchema,
       codec,
       nullValue,
-      dateFormat)(sqlContext)
+      dateFormat,
+      maxCharsPerCol)(sqlContext)
     sqlContext.baseRelationToDataFrame(relation)
   }
 }

src/main/scala/com/databricks/spark/csv/CsvRelation.scala

@@ -49,7 +49,8 @@ case class CsvRelation protected[spark] (
     inferCsvSchema: Boolean,
     codec: String = null,
     nullValue: String = "",
-    dateFormat: String = null)(@transient val sqlContext: SQLContext)
+    dateFormat: String = null,
+    maxCharsPerCol: Int = 100000)(@transient val sqlContext: SQLContext)
   extends BaseRelation with TableScan with PrunedScan with InsertableRelation {
 
   // Share date format object as it is expensive to parse date pattern.
@@ -287,7 +288,8 @@ case class CsvRelation protected[spark] (
 
         new BulkCsvReader(iter, split,
           headers = header, fieldSep = delimiter,
-          quote = quoteChar, escape = escapeVal, commentMarker = commentChar)
+          quote = quoteChar, escape = escapeVal,
+          commentMarker = commentChar, maxCharsPerCol = maxCharsPerCol)
       }
     }, true)
 

src/main/scala/com/databricks/spark/csv/DefaultSource.scala

@@ -142,6 +142,14 @@ class DefaultSource
 
     val codec = parameters.getOrElse("codec", null)
 
+    val maxCharsPerColStr = parameters.getOrElse("maxCharsPerCol", "100000")
+    val maxCharsPerCol = try {
+      maxCharsPerColStr.toInt
+    } catch {
+      case e: Exception => throw new Exception("maxCharsPerCol must be a valid integer")
+    }
+
+
     CsvRelation(
       () => TextFile.withCharset(sqlContext.sparkContext, path, charset),
       Some(path),
@@ -159,7 +167,8 @@ class DefaultSource
       inferSchemaFlag,
       codec,
       nullValue,
-      dateFormat)(sqlContext)
+      dateFormat,
+      maxCharsPerCol)(sqlContext)
   }
 
   override def createRelation(

src/main/scala/com/databricks/spark/csv/readers/readers.scala

@@ -44,7 +44,8 @@ private[readers] abstract class CsvReader(
     ignoreTrailingSpace: Boolean = true,
     headers: Seq[String],
     inputBufSize: Int = 128,
-    maxCols: Int = 20480) {
+    maxCols: Int = 20480,
+    maxCharsPerCol: Int = 100000) {
   protected lazy val parser: CsvParser = {
     val settings = new CsvParserSettings()
     val format = settings.getFormat
@@ -59,7 +60,7 @@ private[readers] abstract class CsvReader(
     settings.setInputBufferSize(inputBufSize)
     settings.setMaxColumns(maxCols)
     settings.setNullValue("")
-    settings.setMaxCharsPerColumn(100000)
+    settings.setMaxCharsPerColumn(maxCharsPerCol)
     if (headers != null) settings.setHeaders(headers: _*)
 
     new CsvParser(settings)
@@ -86,7 +87,8 @@ private[csv] class LineCsvReader(
     ignoreLeadingSpace: Boolean = true,
     ignoreTrailingSpace: Boolean = true,
     inputBufSize: Int = 128,
-    maxCols: Int = 20480)
+    maxCols: Int = 20480,
+    maxCharsPerCol: Int = 100000)
   extends CsvReader(
     fieldSep,
     lineSep,
@@ -97,7 +99,8 @@ private[csv] class LineCsvReader(
     ignoreTrailingSpace,
     null,
     inputBufSize,
-    maxCols) {
+    maxCols,
+    maxCharsPerCol) {
   /**
    * parse a line
    * @param line a String with no newline at the end
@@ -136,7 +139,8 @@ private[csv] class BulkCsvReader(
     ignoreTrailingSpace: Boolean = true,
     headers: Seq[String],
     inputBufSize: Int = 128,
-    maxCols: Int = 20480)
+    maxCols: Int = 20480,
+    maxCharsPerCol: Int = 100000)
   extends CsvReader(
     fieldSep,
     lineSep,
@@ -147,7 +151,8 @@ private[csv] class BulkCsvReader(
     ignoreTrailingSpace,
     headers,
     inputBufSize,
-    maxCols)
+    maxCols,
+    maxCharsPerCol)
   with Iterator[Array[String]] {
 
   private val reader = new StringIteratorReader(iter)

src/test/resources/long-cols.csv

@@ -0,0 +1,4 @@
+id,text
+1,bacon is yummy
+2,Bacon ipsum dolor amet dolore ipsum boudin jowl. Frankfurter flank turkey incididunt minim elit tongue id t-bone labore magna bresaola alcatra hamburger. Ribeye alcatra in shank, turkey ground round sunt short ribs reprehenderit labore tongue. Qui consectetur salami labore. Est sed venison sausage, kielbasa cupidatat porchetta ad beef. Ex capicola short loin drumstick shank, pork reprehenderit shankle beef ribs proident short ribs kevin sausage ribeye chicken. Sirloin kielbasa capicola sausage tenderloin, aliquip brisket short ribs ham duis filet mignon. Boudin turkey jerky in ullamco rump turducken bresaola kevin qui andouille tri-tip pastrami. Tenderloin ut frankfurter hamburger in pork belly cow proident short loin veniam shank filet mignon beef pastrami. Pariatur porchetta in swine officia spare ribs sunt. Landjaeger fatback nisi tenderloin kielbasa. Swine jowl filet mignon fatback. Turducken picanha reprehenderit, ham hock pancetta consectetur sirloin sed excepteur jowl qui commodo corned beef lorem. Non rump nostrud sunt alcatra aute. Nulla irure aliqua commodo laboris. Fugiat do ad occaecat in, ut porchetta commodo. Ham hock cillum in cow salami minim in. Beef pancetta reprehenderit, leberkas excepteur swine qui cow magna chicken exercitation esse short ribs.Nulla jowl esse aute, quis fugiat ipsum tail commodo ex anim kevin. Ribeye quis porchetta meatball, shankle pork loin jowl. Landjaeger andouille dolore jowl ut drumstick pariatur. Et pork chop id laborum landjaeger pancetta occaecat swine. Porchetta picanha flank consectetur shankle reprehenderit pork chop short ribs. Brisket dolore ribeye ea drumstick landjaeger. Dolore consectetur flank, tenderloin occaecat do reprehenderit.Pig elit in dolore, ullamco nostrud biltong kevin culpa venison labore swine tri-tip. Brisket laborum short ribs id capicola venison. Velit picanha nostrud strip steak exercitation, tongue pig duis id ad labore tempor. Veniam frankfurter sed hamburger short ribs kevin cupim sint lorem Bacon ipsum dolor amet dolore ipsum boudin jowl. Frankfurter flank turkey incididunt minim elit tongue id t-bone labore magna bresaola alcatra hamburger. Ribeye alcatra in shank, turkey ground round sunt short ribs reprehenderit labore tongue. Qui consectetur salami labore. Est sed venison sausage, kielbasa cupidatat porchetta ad beef. Ex capicola short loin drumstick shank, pork reprehenderit shankle beef ribs proident short ribs kevin sausage ribeye chicken. Sirloin kielbasa capicola sausage tenderloin, aliquip brisket short ribs ham duis filet mignon. Boudin turkey jerky in ullamco rump turducken bresaola kevin qui andouille tri-tip pastrami. Tenderloin ut frankfurter hamburger in pork belly cow proident short loin veniam shank filet mignon beef pastrami. Pariatur porchetta in swine officia spare ribs sunt. Landjaeger fatback nisi tenderloin kielbasa. Swine jowl filet mignon fatback. Turducken picanha reprehenderit, ham hock pancetta consectetur sirloin sed excepteur jowl qui commodo corned beef lorem. Non rump nostrud sunt alcatra aute. Nulla irure aliqua commodo laboris. Fugiat do ad occaecat in, ut porchetta commodo. Ham hock cillum in cow salami minim in. Beef pancetta reprehenderit, leberkas excepteur swine qui cow magna chicken exercitation esse short ribs.Nulla jowl esse aute, quis fugiat ipsum tail commodo ex anim kevin. Ribeye quis porchetta meatball, shankle pork loin jowl. Landjaeger andouille dolore jowl ut drumstick pariatur. Et pork chop id laborum landjaeger pancetta occaecat swine. Porchetta picanha flank consectetur shankle reprehenderit pork chop short ribs. Brisket dolore ribeye ea drumstick landjaeger. Dolore consectetur flank, tenderloin occaecat do reprehenderit.Pig elit in dolore, ullamco nostrud biltong kevin culpa venison labore swine tri-tip. Brisket laborum short ribs id capicola venison. Velit picanha nostrud strip steak exercitation, tongue pig duis id ad labore tempor. Veniam frankfurter sed hamburger short ribs kevin cupim sint lorem Bacon ipsum dolor amet dolore ipsum boudin jowl. Frankfurter flank turkey incididunt minim elit tongue id t-bone labore magna bresaola alcatra hamburger. Ribeye alcatra in shank, turkey ground round sunt short ribs reprehenderit labore tongue. Qui consectetur salami labore. Est sed venison sausage, kielbasa cupidatat porchetta ad beef. Ex capicola short loin drumstick shank, pork reprehenderit shankle beef ribs proident short ribs kevin sausage ribeye chicken. Sirloin kielbasa capicola sausage tenderloin, aliquip brisket short ribs ham duis filet mignon. Boudin turkey jerky in ullamco rump turducken bresaola kevin qui andouille tri-tip pastrami. Tenderloin ut frankfurter hamburger in pork belly cow proident short loin veniam shank filet mignon beef pastrami. Pariatur porchetta in swine officia spare ribs sunt. Landjaeger fatback nisi tenderloin kielbasa. Swine jowl filet mignon fatback. Turducken picanha reprehenderit, ham hock pancetta consectetur sirloin sed excepteur jowl qui commodo corned beef lorem. Non rump nostrud sunt alcatra aute. Nulla irure aliqua commodo laboris. Fugiat do ad occaecat in, ut porchetta commodo. Ham hock cillum in cow salami minim in. Beef pancetta reprehenderit, leberkas excepteur swine qui cow magna chicken exercitation esse short ribs.Nulla jowl esse aute, quis fugiat ipsum tail commodo ex anim kevin. Ribeye quis porchetta meatball, shankle pork loin jowl. Landjaeger andouille dolore jowl ut drumstick pariatur. Et pork chop id laborum landjaeger pancetta occaecat swine. Porchetta picanha flank consectetur shankle reprehenderit pork chop short ribs. Brisket dolore ribeye ea drumstick landjaeger. Dolore consectetur flank, tenderloin occaecat do reprehenderit.Pig elit in dolore, ullamco nostrud biltong kevin culpa venison labore swine tri-tip. Brisket laborum short ribs id capicola venison. Velit picanha nostrud strip steak exercitation, tongue pig duis id ad labore tempor. Veniam frankfurter sed hamburger short ribs kevin cupim sint lorem Bacon ipsum dolor amet dolore ipsum boudin jowl. Frankfurter flank turkey incididunt minim elit tongue id t-bone labore magna bresaola alcatra hamburger. Ribeye alcatra in shank, turkey ground round sunt short ribs reprehenderit labore tongue. Qui consectetur salami labore. Est sed venison sausage, kielbasa cupidatat porchetta ad beef. Ex capicola short loin drumstick shank, pork reprehenderit shankle beef ribs proident short ribs kevin sausage ribeye chicken. Sirloin kielbasa capicola sausage tenderloin, aliquip brisket short ribs ham duis filet mignon. Boudin turkey jerky in ullamco rump turducken bresaola kevin qui andouille tri-tip pastrami. Tenderloin ut frankfurter hamburger in pork belly cow proident short loin veniam shank filet mignon beef pastrami. Pariatur porchetta in swine officia spare ribs sunt. Landjaeger fatback nisi tenderloin kielbasa. Swine jowl filet mignon fatback. Turducken picanha reprehenderit, ham hock pancetta consectetur sirloin sed excepteur jowl qui commodo corned beef lorem. Non rump nostrud sunt alcatra aute. Nulla irure aliqua commodo laboris. Fugiat do ad occaecat in, ut porchetta commodo. Ham hock cillum in cow salami minim in. Beef pancetta reprehenderit, leberkas excepteur swine qui cow magna chicken exercitation esse short ribs.Nulla jowl esse aute, quis fugiat ipsum tail commodo ex anim kevin. Ribeye quis porchetta meatball, shankle pork loin jowl. Landjaeger andouille dolore jowl ut drumstick pariatur. Et pork chop id laborum landjaeger pancetta occaecat swine. Porchetta picanha flank consectetur shankle reprehenderit pork chop short ribs. Brisket dolore ribeye ea drumstick landjaeger. Dolore consectetur flank, tenderloin occaecat do reprehenderit.Pig elit in dolore, ullamco nostrud biltong kevin culpa venison labore swine tri-tip. Brisket laborum short ribs id capicola venison. Velit picanha nostrud strip steak exercitation, tongue pig duis id ad labore tempor. Veniam frankfurter sed hamburger short ribs kevin cupim sint lorem Bacon ipsum dolor amet dolore ipsum boudin jowl. Frankfurter flank turkey incididunt minim elit tongue id t-bone labore magna bresaola alcatra hamburger. Ribeye alcatra in shank, turkey ground round sunt short ribs reprehenderit labore tongue. Qui consectetur salami labore. Est sed venison sausage, kielbasa cupidatat porchetta ad beef. Ex capicola short loin drumstick shank, pork reprehenderit shankle beef ribs proident short ribs kevin sausage ribeye chicken. Sirloin kielbasa capicola sausage tenderloin, aliquip brisket short ribs ham duis filet mignon. Boudin turkey jerky in ullamco rump turducken bresaola kevin qui andouille tri-tip pastrami. Tenderloin ut frankfurter hamburger in pork belly cow proident short loin veniam shank filet mignon beef pastrami. Pariatur porchetta in swine officia spare ribs sunt. Landjaeger fatback nisi tenderloin kielbasa. Swine jowl filet mignon fatback. Turducken picanha reprehenderit, ham hock pancetta consectetur sirloin sed excepteur jowl qui commodo corned beef lorem. Non rump nostrud sunt alcatra aute. Nulla irure aliqua commodo laboris. Fugiat do ad occaecat in, ut porchetta commodo. Ham hock cillum in cow salami minim in. Beef pancetta reprehenderit, leberkas excepteur swine qui cow magna chicken exercitation esse short ribs.Nulla jowl esse aute, quis fugiat ipsum tail commodo ex anim kevin. Ribeye quis porchetta meatball, shankle pork loin jowl. Landjaeger andouille dolore jowl ut drumstick pariatur. Et pork chop id laborum landjaeger pancetta occaecat swine. Porchetta picanha flank consectetur shankle reprehenderit pork chop short ribs. Brisket dolore ribeye ea drumstick landjaeger. Dolore consectetur flank, tenderloin occaecat do reprehenderit.Pig elit in dolore, ullamco nostrud biltong kevin culpa venison labore swine tri-tip. Brisket laborum short ribs id capicola venison. Velit picanha nostrud strip steak exercitation, tongue pig duis id ad labore tempor. Veniam frankfurter sed hamburger short ribs kevin cupim sint lorem
+3,pork is the best

src/test/scala/com/databricks/spark/csv/CsvSuite.scala

@@ -47,6 +47,7 @@ abstract class AbstractCsvSuite extends FunSuite with BeforeAndAfterAll {
   val disableCommentsFile = "src/test/resources/disable_comments.csv"
   val boolFile = "src/test/resources/bool.csv"
   val datesFile = "src/test/resources/dates.csv"
+  val longColsFile = "src/test/resources/long-cols.csv"
   private val simpleDatasetFile = "src/test/resources/simple.csv"
 
   val numCars = 3
@@ -857,6 +858,38 @@ abstract class AbstractCsvSuite extends FunSuite with BeforeAndAfterAll {
     assert(results.toSeq.map(_.toSeq) === expected)
   }
 
+  test("DSL allows for setting maxColsPerChar and expect error") {
+    val parser = new CsvParser()
+      .withDelimiter(',')
+      .withUseHeader(true)
+      .withParseMode(ParseModes.FAIL_FAST_MODE)
+      .withMaxCharsPerCol(5000)
+
+    val exception = intercept[SparkException]{
+      parser.csvFile(sqlContext, longColsFile)
+        .select("text")
+        .collect()
+    }
+
+    assert(exception.getMessage.contains("Malformed line in FAILFAST mode: 2,Bacon ipsum dolor amet dolore"))
+  }
+
+  test("DSL allows for setting maxColsPerChar and succeeds") {
+    val parser = new CsvParser()
+      .withDelimiter(',')
+      .withUseHeader(true)
+      .withMaxCharsPerCol(15000)
+
+    val res = parser.csvFile(sqlContext, longColsFile)
+      .collect()
+
+    assert(res.size === 3)
+    assert(res(0).toSeq === Seq("1", "bacon is yummy"))
+    assert(res(1).getAs[String](0) === "2")
+    assert(res(1).getAs[String](1).startsWith("Bacon ipsum dolor amet dolore"))
+    assert(res(2).toSeq === Seq("3", "pork is the best"))
+  }
+
   test("DSL load csv from rdd") {
     val csvRdd = sqlContext.sparkContext.parallelize(Seq("age,height", "20,1.8", "16,1.7"))
     val df = new CsvParser()