use Reader instead of Scanner to bypass buffer limits

Author: James Cor

.gitattributes

@@ -7,4 +7,5 @@ enginetest/testdata/test5.txt binary
 enginetest/testdata/test6.csv binary
 enginetest/testdata/test7.txt binary
 enginetest/testdata/test8.txt binary
-enginetest/testdata/test9.txt binary
+enginetest/testdata/test9.txt binary
+enginetest/testdata/test10.txt binary

enginetest/queries/load_queries.go

@@ -539,6 +539,23 @@ var LoadDataScripts = []ScriptTest{
 			},
 		},
 	},
+	{
+		Name: "LOAD DATA with column data larger than 64KB",
+		SetUpScript: []string{
+			"create table t(id int primary key, lt longtext);",
+			"load data infile './testdata/test10.txt' into table t fields terminated by ',' lines terminated by '\n';",
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query: "select id, length(lt) from t order by id",
+				Expected: []sql.Row{
+					{1, 65536},
+					{2, 100000},
+					{3, 1000000},
+				},
+			},
+		},
+	},
 }
 
 var LoadDataErrorScripts = []ScriptTest{

enginetest/testdata/test10.txt

@@ -93,20 +93,6 @@ func (b *BaseBuilder) buildLoadData(ctx *sql.Context, n *plan.LoadData, row sql.
 		reader = file
 	}
 
-	scanner := bufio.NewScanner(reader)
-	scanner.Split(n.SplitLines)
-
-	// Skip through the lines that need to be ignored.
-	for n.IgnoreNum > 0 && scanner.Scan() {
-		scanner.Text()
-		n.IgnoreNum--
-	}
-
-	if scanner.Err() != nil {
-		reader.Close()
-		return nil, scanner.Err()
-	}
-
 	sch := n.Schema()
 	source := sch[0].Source // Schema will always have at least one column
 	colNames := n.ColNames
@@ -130,7 +116,7 @@ func (b *BaseBuilder) buildLoadData(ctx *sql.Context, n *plan.LoadData, row sql.
 	return &loadDataIter{
 		destSch:       n.DestSch,
 		reader:        reader,
-		scanner:       scanner,
+		scanner:       bufio.NewReader(reader),
 		colCount:      len(n.ColNames), // Needs to be the original column count
 		fieldToColMap: fieldToColMap,
 		setExprs:      n.SetExprs,

sql/rowexec/ddl.go

@@ -16,6 +16,7 @@ package rowexec
 
 import (
 	"bufio"
+	"bytes"
 	"fmt"
 	"io"
 	"strings"
@@ -38,15 +39,18 @@ import (
 )
 
 type loadDataIter struct {
-	scanner *bufio.Scanner
+	scanner *bufio.Reader
 	reader  io.ReadCloser
+	buffer  bytes.Buffer
 
 	destSch       sql.Schema
 	colCount      int
 	fieldToColMap []int
 	setExprs      []sql.Expression
 	userVars      []sql.Expression
 
+	ignoreNum int64
+
 	fieldsTerminatedBy  string
 	fieldsEnclosedBy    string
 	fieldsEnclosedByOpt bool
@@ -56,28 +60,51 @@ type loadDataIter struct {
 	linesTerminatedBy string
 }
 
+// readLine reads a line from the scanner.
+func (l loadDataIter) readLine() ([]byte, error) {
+	// last byte of the line terminator
+	delim := []byte(l.linesTerminatedBy)
+	delimDelim := delim[len(delim)-1]
+	l.buffer.Reset()
+	for {
+		chunk, err := l.scanner.ReadBytes(delimDelim)
+		if err != nil {
+			return nil, err
+		}
+		l.buffer.Write(chunk)
+		buf := l.buffer.Bytes()
+		if bytes.HasSuffix(buf, delim) {
+			return buf, nil
+		}
+	}
+}
+
 func (l loadDataIter) Next(ctx *sql.Context) (returnRow sql.Row, returnErr error) {
-	var exprs []sql.Expression
-	var err error
+	// skip first ignoreNum lines
+	for ; l.ignoreNum > 0; l.ignoreNum-- {
+		_, err := l.readLine()
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	// If exprs is nil then this is a skipped line (see test cases). Keep skipping
 	// until exprs != nil
+	var exprs []sql.Expression
 	for exprs == nil {
-		if keepGoing := l.scanner.Scan(); !keepGoing {
-			if l.scanner.Err() != nil {
-				return nil, l.scanner.Err()
-			}
-			return nil, io.EOF
+		line, err := l.readLine()
+		if err != nil {
+			 return nil, err
 		}
-
-		line := l.scanner.Text()
-		exprs, err = l.parseFields(ctx, line)
+		exprs, err = l.parseFields(ctx, string(line))
 		if err != nil {
 			return nil, err
 		}
 	}
 
 	row := make(sql.Row, len(exprs))
 	var secondPass []int
+	var err error
 	for i, expr := range exprs {
 		if expr != nil {
 			// Non-literal default values may reference other columns, so we need to evaluate them in a second pass.