From 2d4b87c792dc8ceb7276f11449bd3098f6eb4d26 Mon Sep 17 00:00:00 2001 From: Jesse Donat Date: Sun, 11 Oct 2026 05:26:30 -0500 Subject: [PATCH 1/6] feat: add query limit support --- README.md | 9 ++-- cmd/sqlread/query.go | 12 +++--- interpreter.go | 64 ++++++++++++++++++++++++++++ lexitemtype_string.go | 17 ++++---- query.go | 98 +++++++++++++++++++++++++++++++++++++++++-- query_test.go | 90 +++++++++++++++++++++++++++++++++++++++ scanner.go | 1 + 7 files changed, 270 insertions(+), 21 deletions(-) create mode 100644 query_test.go diff --git a/README.md b/README.md index 75d91a5..875d506 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,9 @@ Available on the [Releases](https://github.com/donatj/sqlread/releases) page cur ## Example usage +Interactive `SELECT` queries support `LIMIT count` and `LIMIT offset, count`. +Both values must be non-negative integers. + ``` $ sqlread buildings.sql 2017/12/22 12:23:52 starting initial pass @@ -50,13 +53,9 @@ zip_code,varchar deleted,tinyint 2017/12/22 12:24:08 restarting lexer -> SELECT building_id, title FROM buildings; +> SELECT building_id, title FROM buildings LIMIT 2; 2,Home Building 190,Test Building (demo) -192,Donat Building -194,Other Building -201,Sample Building (demo) -205,Johnson Building 2017/12/22 12:24:44 restarting lexer > SELECT `building_id`, `title` FROM `buildings` INTO OUTFILE "dump.csv"; diff --git a/cmd/sqlread/query.go b/cmd/sqlread/query.go index 35c0451..d0d1074 100644 --- a/cmd/sqlread/query.go +++ b/cmd/sqlread/query.go @@ -33,6 +33,7 @@ func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w return fmt.Errorf("column `%s` not found", col) } } + var rowsSeen uint64 for _, loc := range tbl.DataLocs { start := loc.Start.Pos end := loc.End.Pos @@ -52,12 +53,12 @@ func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w } }() - for { - row, ok := <-sp.Out - if !ok { - w.Flush() - break + for row := range sp.Out { + if qry.Limit != nil && !qry.Limit.Includes(rowsSeen) { + rowsSeen++ + continue } + rowsSeen++ out := make([]string, len(colind)) for i, ci := range colind { @@ -66,6 +67,7 @@ func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w w.Write(out) } + w.Flush() } return nil diff --git a/interpreter.go b/interpreter.go index fccce03..d264671 100644 --- a/interpreter.go +++ b/interpreter.go @@ -201,6 +201,11 @@ func (intp *Intp) selectFromIntpState(l *lexer) state { l.accept(whitespace) l.start = l.pos + if l.hasPrefixI("LIMIT") { + l.pos += 5 + l.emit(TIntpLimit) + return intp.selectLimitIntpState + } c := l.next() if c == byte('i') || c == byte('I') { @@ -244,6 +249,65 @@ func (intp *Intp) selectFromIntoOutfileIntpState(l *lexer) state { } l.emit(TString) + l.accept(whitespace) + l.start = l.pos + if l.hasPrefixI("LIMIT") { + l.pos += 5 + l.emit(TIntpLimit) + return intp.selectLimitIntpState + } + + c := l.next() + if c != semi { + l.emit(TIllegal) + return nil + } + + l.emit(TSemi) + return nil +} + +func (intp *Intp) selectLimitIntpState(l *lexer) state { + if l.accept(whitespace) < 1 { + l.emit(TIllegal) + return nil + } + + l.start = l.pos + if l.accept(numbers) == 0 { + l.emit(TIllegal) + return nil + } + l.emit(TNumber) + + l.accept(whitespace) + l.start = l.pos + + c := l.next() + if c == coma { + l.emit(TComma) + return intp.selectLimitCountIntpState + } + + if c != semi { + l.emit(TIllegal) + return nil + } + + l.emit(TSemi) + return nil +} + +func (intp *Intp) selectLimitCountIntpState(l *lexer) state { + l.accept(whitespace) + l.start = l.pos + + if l.accept(numbers) == 0 { + l.emit(TIllegal) + return nil + } + l.emit(TNumber) + l.accept(whitespace) l.start = l.pos diff --git a/lexitemtype_string.go b/lexitemtype_string.go index ee8f8c5..0e914af 100644 --- a/lexitemtype_string.go +++ b/lexitemtype_string.go @@ -38,17 +38,18 @@ func _() { _ = x[TIntpStar-27] _ = x[TIntpFrom-28] _ = x[TIntpIntoOutfile-29] - _ = x[TIntpShowTables-30] - _ = x[TIntpShowColumns-31] - _ = x[TIntpShowCreateTable-32] - _ = x[TIntpQuit-33] - _ = x[TBeginFullStmt-34] - _ = x[TCommitFullStmt-35] + _ = x[TIntpLimit-30] + _ = x[TIntpShowTables-31] + _ = x[TIntpShowColumns-32] + _ = x[TIntpShowCreateTable-33] + _ = x[TIntpQuit-34] + _ = x[TBeginFullStmt-35] + _ = x[TCommitFullStmt-36] } -const _lexItemType_name = "TIllegalTEofTSemiTCommaTCommentTDelimTNullTStringTNumberTHexLiteralTIdentifierTDropTableFullStmtTLockTableFullStmtTUnlockTablesFullStmtTSetFullStmtTLParenTRParenTCreateTableTCreateTableDetailTCreateTableExtraDetailTColumnTypeTColumnSizeTColumnEnumValTColumnDetailsTInsertIntoTInsertValuesTIntpSelectTIntpStarTIntpFromTIntpIntoOutfileTIntpShowTablesTIntpShowColumnsTIntpShowCreateTableTIntpQuitTBeginFullStmtTCommitFullStmt" +const _lexItemType_name = "TIllegalTEofTSemiTCommaTCommentTDelimTNullTStringTNumberTHexLiteralTIdentifierTDropTableFullStmtTLockTableFullStmtTUnlockTablesFullStmtTSetFullStmtTLParenTRParenTCreateTableTCreateTableDetailTCreateTableExtraDetailTColumnTypeTColumnSizeTColumnEnumValTColumnDetailsTInsertIntoTInsertValuesTIntpSelectTIntpStarTIntpFromTIntpIntoOutfileTIntpLimitTIntpShowTablesTIntpShowColumnsTIntpShowCreateTableTIntpQuitTBeginFullStmtTCommitFullStmt" -var _lexItemType_index = [...]uint16{0, 8, 12, 17, 23, 31, 37, 42, 49, 56, 67, 78, 96, 114, 135, 147, 154, 161, 173, 191, 214, 225, 236, 250, 264, 275, 288, 299, 308, 317, 333, 348, 364, 384, 393, 407, 422} +var _lexItemType_index = [...]uint16{0, 8, 12, 17, 23, 31, 37, 42, 49, 56, 67, 78, 96, 114, 135, 147, 154, 161, 173, 191, 214, 225, 236, 250, 264, 275, 288, 299, 308, 317, 333, 343, 358, 374, 394, 403, 417, 432} func (i lexItemType) String() string { if i >= lexItemType(len(_lexItemType_index)-1) { diff --git a/query.go b/query.go index 67405e6..393ae89 100644 --- a/query.go +++ b/query.go @@ -1,11 +1,24 @@ package sqlread -import "strings" +import ( + "strconv" + "strings" +) + +type QueryLimit struct { + Offset uint64 + Count uint64 +} + +func (l QueryLimit) Includes(row uint64) bool { + return row >= l.Offset && row-l.Offset < l.Count +} type Query struct { Columns []string Table string Outfile *string + Limit *QueryLimit } type QueryTree struct { @@ -214,7 +227,11 @@ func (q *QueryParser) parseSelectFromBuilder(qry *Query) parseState { return q.parseSelectIntoOutfileBuilder(qry) } - p.errorUnexpectedLex(d, TSemi, TIntpIntoOutfile) + if d.Type == TIntpLimit { + return q.parseSelectLimitBuilder(qry) + } + + p.errorUnexpectedLex(d, TSemi, TIntpIntoOutfile, TIntpLimit) return nil } } @@ -250,7 +267,82 @@ func (q *QueryParser) parseSelectIntoOutfileBuilder(qry *Query) parseState { return nil } - p.errorUnexpectedLex(d, TSemi) + if d.Type == TIntpLimit { + return q.parseSelectLimitBuilder(qry) + } + + p.errorUnexpectedLex(d, TSemi, TIntpLimit) + return nil + } +} + +func (q *QueryParser) parseSelectLimitBuilder(qry *Query) parseState { + return func(p *Parser) parseState { + c, ok := p.scan() + if !ok { + p.errorUnexpectedEOF() + return nil + } + if c.Type != TNumber { + p.errorUnexpectedLex(c, TNumber) + return nil + } + + first, err := strconv.ParseUint(c.Val, 10, 64) + if err != nil { + p.err = err + return nil + } + + d, ok := p.scan() + if !ok { + p.errorUnexpectedEOF() + return nil + } + if d.Type == TSemi { + qry.Limit = &QueryLimit{Count: first} + q.Tree.Queries = append(q.Tree.Queries, *qry) + return nil + } + if d.Type == TComma { + return q.parseSelectLimitCountBuilder(qry, first) + } + + p.errorUnexpectedLex(d, TSemi, TComma) + return nil + } +} + +func (q *QueryParser) parseSelectLimitCountBuilder(qry *Query, offset uint64) parseState { + return func(p *Parser) parseState { + c, ok := p.scan() + if !ok { + p.errorUnexpectedEOF() + return nil + } + if c.Type != TNumber { + p.errorUnexpectedLex(c, TNumber) + return nil + } + + count, err := strconv.ParseUint(c.Val, 10, 64) + if err != nil { + p.err = err + return nil + } + + d, ok := p.scan() + if !ok { + p.errorUnexpectedEOF() + return nil + } + if d.Type != TSemi { + p.errorUnexpectedLex(d, TSemi) + return nil + } + + qry.Limit = &QueryLimit{Offset: offset, Count: count} + q.Tree.Queries = append(q.Tree.Queries, *qry) return nil } } diff --git a/query_test.go b/query_test.go new file mode 100644 index 0000000..744ef8c --- /dev/null +++ b/query_test.go @@ -0,0 +1,90 @@ +package sqlread + +import ( + "reflect" + "strings" + "testing" +) + +func TestQueryParserLimit(t *testing.T) { + tests := []struct { + name string + input string + want QueryLimit + }{ + { + name: "count", + input: "SELECT * FROM records LIMIT 2;", + want: QueryLimit{Count: 2}, + }, + { + name: "offset and count", + input: "select name from records limit 1, 2;", + want: QueryLimit{Offset: 1, Count: 2}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + query, err := parseUserQuery(test.input) + if err != nil { + t.Fatal(err) + } + + if len(query.Queries) != 1 { + t.Fatalf("parsed queries = %d; want 1", len(query.Queries)) + } + if !reflect.DeepEqual(query.Queries[0].Limit, &test.want) { + t.Errorf("limit = %#v; want %#v", query.Queries[0].Limit, &test.want) + } + }) + } +} + +func TestQueryParserLimitRejectsNonIntegers(t *testing.T) { + for _, input := range []string{ + "SELECT * FROM records LIMIT -1;", + "SELECT * FROM records LIMIT 1.5;", + "SELECT * FROM records LIMIT 1, count;", + } { + t.Run(input, func(t *testing.T) { + _, err := parseUserQuery(input) + if err == nil { + t.Fatal("parseUserQuery() returned nil error") + } + }) + } +} + +func TestQueryLimitIncludes(t *testing.T) { + tests := []struct { + name string + limit QueryLimit + row uint64 + want bool + }{ + {name: "before offset", limit: QueryLimit{Offset: 1, Count: 2}, row: 0, want: false}, + {name: "at offset", limit: QueryLimit{Offset: 1, Count: 2}, row: 1, want: true}, + {name: "within count", limit: QueryLimit{Offset: 1, Count: 2}, row: 2, want: true}, + {name: "after count", limit: QueryLimit{Offset: 1, Count: 2}, row: 3, want: false}, + {name: "zero count", limit: QueryLimit{Count: 0}, row: 0, want: false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if actual := test.limit.Includes(test.row); actual != test.want { + t.Errorf("Includes(%d) = %t; want %t", test.row, actual, test.want) + } + }) + } +} + +func parseUserQuery(input string) (QueryTree, error) { + l, items := Lex(strings.NewReader(input)) + intp := &Intp{} + go l.Run(intp.StartIntpState) + + query := NewQueryParser() + err := Parse(items).Run(query.ParseStart) + return query.Tree, err +} diff --git a/scanner.go b/scanner.go index cb416c5..f7074b5 100644 --- a/scanner.go +++ b/scanner.go @@ -51,6 +51,7 @@ const ( TIntpStar TIntpFrom TIntpIntoOutfile + TIntpLimit TIntpShowTables TIntpShowColumns From 6e548685eee8a1d9a5ab1e4e5f94980129fb33da Mon Sep 17 00:00:00 2001 From: Jesse Donat Date: Sun, 11 Oct 2026 05:30:04 -0500 Subject: [PATCH 2/6] docs: describe user query grammar --- README.md | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 875d506..8380f4b 100644 --- a/README.md +++ b/README.md @@ -28,10 +28,28 @@ go install github.com/donatj/sqlread/cmd/sqlread@latest Available on the [Releases](https://github.com/donatj/sqlread/releases) page currently for Linux and macOS. -## Example usage +## User query grammar + +Square brackets (`[]`) mark optional clauses. Braces (`{}`) require one choice. +A vertical bar (`|`) separates choices. An ellipsis (`...`) marks a repeated item. -Interactive `SELECT` queries support `LIMIT count` and `LIMIT offset, count`. -Both values must be non-negative integers. +``` +SELECT { * | column_name } [, { * | column_name }] ... + FROM table_name + [INTO OUTFILE string] + [LIMIT { row_count | offset, row_count }]; + +SHOW TABLES; +SHOW COLUMNS FROM table_name; +SHOW CREATE TABLE table_name; +{ EXIT | QUIT }; +``` + +Keywords are case-insensitive. Table names and column names may use backticks. +`string` values may use single or double quotes. `offset` and `row_count` must be +non-negative integers. Expressions, `WHERE`, and `ORDER BY` are not supported. + +## Example usage ``` $ sqlread buildings.sql From f177f7fdb66e9363d9f26dc3889f28f1d07294a2 Mon Sep 17 00:00:00 2001 From: Jesse Donat Date: Sun, 11 Oct 2026 05:38:48 -0500 Subject: [PATCH 3/6] fix: stop parsing after reaching query limit --- cmd/sqlread/query.go | 9 +++++++++ detail.go | 46 ++++++++++++++++++++++++++++++++++++++++---- detail_test.go | 31 +++++++++++++++++++++++++++++ scanner.go | 28 ++++++++++++++++++++++++++- 4 files changed, 109 insertions(+), 5 deletions(-) diff --git a/cmd/sqlread/query.go b/cmd/sqlread/query.go index d0d1074..26448a2 100644 --- a/cmd/sqlread/query.go +++ b/cmd/sqlread/query.go @@ -35,6 +35,10 @@ func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w } var rowsSeen uint64 for _, loc := range tbl.DataLocs { + if qry.Limit != nil && !qry.Limit.Includes(rowsSeen) { + return nil + } + start := loc.Start.Pos end := loc.End.Pos @@ -66,6 +70,11 @@ func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w } w.Write(out) + + if qry.Limit != nil && !qry.Limit.Includes(rowsSeen) { + sp.Stop() + sl.Stop() + } } w.Flush() } diff --git a/detail.go b/detail.go index 36022dd..8cfc3bf 100644 --- a/detail.go +++ b/detail.go @@ -3,20 +3,50 @@ package sqlread import ( "errors" "strings" + "sync" ) type InsertDetailParser struct { Out chan []string ColInd []int + done chan struct{} + stop sync.Once } func NewInsertDetailParser() *InsertDetailParser { return &InsertDetailParser{ - Out: make(chan []string, 100), + Out: make(chan []string, 100), + done: make(chan struct{}), + } +} + +func (d *InsertDetailParser) Stop() { + d.stop.Do(func() { + close(d.done) + }) +} + +func (d *InsertDetailParser) stopped() bool { + select { + case <-d.done: + return true + default: + return false + } +} + +func (d *InsertDetailParser) writeRow(row []string) bool { + select { + case d.Out <- row: + return true + case <-d.done: + return false } } func (d *InsertDetailParser) ParseStart(p *Parser) parseState { + defer close(d.Out) + cii, ok := p.scan() if !ok { p.errorUnexpectedEOF() @@ -40,8 +70,15 @@ func (d *InsertDetailParser) ParseStart(p *Parser) parseState { row := []string{} for { + if d.stopped() { + return nil + } + v, ok := p.scan() if !ok { + if d.stopped() { + return nil + } p.errorUnexpectedEOF() } @@ -61,11 +98,12 @@ func (d *InsertDetailParser) ParseStart(p *Parser) parseState { case TNull: row = append(row, "\u0000") // investigate case TComma: - d.Out <- row + if !d.writeRow(row) { + return nil + } row = []string{} case TSemi: - d.Out <- row - close(d.Out) + d.writeRow(row) return nil } } diff --git a/detail_test.go b/detail_test.go index 0780f18..b66d402 100644 --- a/detail_test.go +++ b/detail_test.go @@ -4,6 +4,7 @@ import ( "reflect" "strings" "testing" + "time" ) func TestStringValue(t *testing.T) { @@ -96,3 +97,33 @@ func TestInsertDetailControlEscapes(t *testing.T) { }) } } + +func TestInsertDetailParserStop(t *testing.T) { + input := "INSERT INTO records VALUES " + strings.Repeat("(1),", 200) + "(1);" + l, items := Lex(strings.NewReader(input)) + go l.Run(StartState) + + detail := NewInsertDetailParser() + done := make(chan error, 1) + go func() { + done <- Parse(items).Run(detail.ParseStart) + }() + + for range 10 { + <-detail.Out + } + + detail.Stop() + l.Stop() + for range detail.Out { + } + + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("parser did not stop") + } +} diff --git a/scanner.go b/scanner.go index f7074b5..4298b23 100644 --- a/scanner.go +++ b/scanner.go @@ -4,6 +4,7 @@ import ( "io" "slices" "strings" + "sync" "unicode/utf8" ) @@ -75,6 +76,8 @@ type lexer struct { pos int64 // width int items chan LexItem + done chan struct{} + stop sync.Once } const ( @@ -151,6 +154,9 @@ var ( func (l *lexer) Run(start state) { for state := start; state != nil; { + if l.stopped() { + break + } state = state(l) } close(l.items) @@ -247,6 +253,7 @@ func Lex(input io.ReaderAt) (*lexer, chan LexItem) { l := &lexer{ input: input, items: make(chan LexItem, 2000000), + done: make(chan struct{}), } return l, l.items @@ -256,6 +263,7 @@ func LexSection(input io.ReaderAt, off, n int64) (*lexer, chan LexItem) { l := &lexer{ input: io.NewSectionReader(input, off, n), items: make(chan LexItem, 2000000), + done: make(chan struct{}), } return l, l.items @@ -271,11 +279,29 @@ func (l *lexer) emit(t lexItemType) LexItem { Pos: l.start, } - l.items <- li + select { + case l.items <- li: + case <-l.done: + } return li } +func (l *lexer) Stop() { + l.stop.Do(func() { + close(l.done) + }) +} + +func (l *lexer) stopped() bool { + select { + case <-l.done: + return true + default: + return false + } +} + func in(b byte, bs []byte) bool { return slices.Contains(bs, b) } From d53f89c3e0b67cf80412d4a1f01a6bf2c7d9d72f Mon Sep 17 00:00:00 2001 From: Jesse Donat Date: Sun, 11 Oct 2026 05:48:49 -0500 Subject: [PATCH 4/6] fix: execute semicolon-terminated queries --- interpreter.go | 45 +++++++++++++++++++++++++-------------------- query_test.go | 49 ++++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 73 insertions(+), 21 deletions(-) diff --git a/interpreter.go b/interpreter.go index d264671..c69883f 100644 --- a/interpreter.go +++ b/interpreter.go @@ -201,13 +201,20 @@ func (intp *Intp) selectFromIntpState(l *lexer) state { l.accept(whitespace) l.start = l.pos - if l.hasPrefixI("LIMIT") { - l.pos += 5 - l.emit(TIntpLimit) - return intp.selectLimitIntpState + c := l.next() + if c == semi { + l.emit(TSemi) + return nil + } + + if c == byte('l') || c == byte('L') { + if l.hasPrefixI("IMIT") { + l.pos += 4 + l.emit(TIntpLimit) + return intp.selectLimitIntpState + } } - c := l.next() if c == byte('i') || c == byte('I') { if l.hasPrefixI("NTO") { l.pos += 3 @@ -224,12 +231,7 @@ func (intp *Intp) selectFromIntpState(l *lexer) state { l.start = l.pos - if c != semi { - l.emit(TIllegal) - return nil - } - - l.emit(TSemi) + l.emit(TIllegal) return nil } @@ -251,19 +253,22 @@ func (intp *Intp) selectFromIntoOutfileIntpState(l *lexer) state { l.accept(whitespace) l.start = l.pos - if l.hasPrefixI("LIMIT") { - l.pos += 5 - l.emit(TIntpLimit) - return intp.selectLimitIntpState - } - c := l.next() - if c != semi { - l.emit(TIllegal) + if c == semi { + l.emit(TSemi) return nil } - l.emit(TSemi) + if c == byte('l') || c == byte('L') { + if l.hasPrefixI("IMIT") { + l.pos += 4 + l.emit(TIntpLimit) + return intp.selectLimitIntpState + } + } + + l.start = l.pos + l.emit(TIllegal) return nil } diff --git a/query_test.go b/query_test.go index 744ef8c..d9e58b6 100644 --- a/query_test.go +++ b/query_test.go @@ -1,6 +1,7 @@ package sqlread import ( + "io" "reflect" "strings" "testing" @@ -56,6 +57,26 @@ func TestQueryParserLimitRejectsNonIntegers(t *testing.T) { } } +func TestQueryParserDoesNotReadPastSemicolon(t *testing.T) { + for _, query := range []string{ + "SELECT * FROM records;", + "SELECT * FROM records INTO OUTFILE 'records.csv';", + } { + t.Run(query, func(t *testing.T) { + input := &readPastEndReader{input: query} + + _, err := parseUserQueryReader(input) + if err != nil { + t.Fatal(err) + } + + if input.readPastEnd { + t.Fatal("parser read past the statement terminator") + } + }) + } +} + func TestQueryLimitIncludes(t *testing.T) { tests := []struct { name string @@ -80,7 +101,11 @@ func TestQueryLimitIncludes(t *testing.T) { } func parseUserQuery(input string) (QueryTree, error) { - l, items := Lex(strings.NewReader(input)) + return parseUserQueryReader(strings.NewReader(input)) +} + +func parseUserQueryReader(input io.ReaderAt) (QueryTree, error) { + l, items := Lex(input) intp := &Intp{} go l.Run(intp.StartIntpState) @@ -88,3 +113,25 @@ func parseUserQuery(input string) (QueryTree, error) { err := Parse(items).Run(query.ParseStart) return query.Tree, err } + +type readPastEndReader struct { + input string + readPastEnd bool +} + +func (r *readPastEndReader) ReadAt(b []byte, off int64) (int, error) { + if off+int64(len(b)) > int64(len(r.input)) { + r.readPastEnd = true + } + + if off >= int64(len(r.input)) { + return 0, io.EOF + } + + n := copy(b, r.input[off:]) + if n != len(b) { + return n, io.EOF + } + + return n, nil +} From 2e8a98d07c60970579fa00cc8e7b8d806d63be5c Mon Sep 17 00:00:00 2001 From: Jesse Donat Date: Sun, 11 Oct 2026 06:18:05 -0500 Subject: [PATCH 5/6] fix: apply limit offsets across data blocks --- cmd/sqlread/query.go | 4 ++-- query.go | 4 ++++ query_test.go | 22 ++++++++++++++++++++++ 3 files changed, 28 insertions(+), 2 deletions(-) diff --git a/cmd/sqlread/query.go b/cmd/sqlread/query.go index 26448a2..ead2c82 100644 --- a/cmd/sqlread/query.go +++ b/cmd/sqlread/query.go @@ -35,7 +35,7 @@ func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w } var rowsSeen uint64 for _, loc := range tbl.DataLocs { - if qry.Limit != nil && !qry.Limit.Includes(rowsSeen) { + if qry.Limit != nil && qry.Limit.Exhausted(rowsSeen) { return nil } @@ -71,7 +71,7 @@ func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w w.Write(out) - if qry.Limit != nil && !qry.Limit.Includes(rowsSeen) { + if qry.Limit != nil && qry.Limit.Exhausted(rowsSeen) { sp.Stop() sl.Stop() } diff --git a/query.go b/query.go index 393ae89..29b8ae8 100644 --- a/query.go +++ b/query.go @@ -14,6 +14,10 @@ func (l QueryLimit) Includes(row uint64) bool { return row >= l.Offset && row-l.Offset < l.Count } +func (l QueryLimit) Exhausted(rowsSeen uint64) bool { + return l.Count == 0 || (rowsSeen >= l.Offset && rowsSeen-l.Offset >= l.Count) +} + type Query struct { Columns []string Table string diff --git a/query_test.go b/query_test.go index d9e58b6..dfd3b25 100644 --- a/query_test.go +++ b/query_test.go @@ -100,6 +100,28 @@ func TestQueryLimitIncludes(t *testing.T) { } } +func TestQueryLimitExhausted(t *testing.T) { + tests := []struct { + name string + limit QueryLimit + rowsSeen uint64 + want bool + }{ + {name: "before offset", limit: QueryLimit{Offset: 1, Count: 2}, rowsSeen: 0, want: false}, + {name: "within count", limit: QueryLimit{Offset: 1, Count: 2}, rowsSeen: 2, want: false}, + {name: "after count", limit: QueryLimit{Offset: 1, Count: 2}, rowsSeen: 3, want: true}, + {name: "zero count", limit: QueryLimit{Count: 0}, rowsSeen: 0, want: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if actual := test.limit.Exhausted(test.rowsSeen); actual != test.want { + t.Errorf("Exhausted(%d) = %t; want %t", test.rowsSeen, actual, test.want) + } + }) + } +} + func parseUserQuery(input string) (QueryTree, error) { return parseUserQueryReader(strings.NewReader(input)) } From aa64c9a5e21967671efb3290b16af7ba52b0cdb1 Mon Sep 17 00:00:00 2001 From: Jesse Donat Date: Sun, 11 Oct 2026 12:00:41 -0500 Subject: [PATCH 6/6] refactor: align query execution with architecture goals --- ARCHITECTURE.md | 10 +- cmd/sqlread/interactive_test.go | 77 ++++++++++ cmd/sqlread/query.go | 71 +-------- cmd/sqlread/query_bench_test.go | 40 ++++++ cmd/sqlread/query_test.go | 245 ++++++++++++++++++++++++++++++++ detail.go | 54 +++---- detail_test.go | 31 ---- execute.go | 93 ++++++++++++ query.go | 8 -- query_test.go | 58 ++------ scanner.go | 26 +++- testfixtures/queries.sql | 11 ++ 12 files changed, 528 insertions(+), 196 deletions(-) create mode 100644 cmd/sqlread/interactive_test.go create mode 100644 cmd/sqlread/query_bench_test.go create mode 100644 cmd/sqlread/query_test.go create mode 100644 execute.go create mode 100644 testfixtures/queries.sql diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 54c3f87..2222c61 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -243,8 +243,9 @@ The current code already uses state functions and file positions to support large dumps with limited memory. These areas merit measurement or refinement as the tool evolves: -- [scanner.go](scanner.go) allocates channels for 2,000,000 tokens. Token emission - copies source text into strings. +- [scanner.go](scanner.go) still allocates 2,000,000-token channels for full scans + and user-query parsing. Data-section scans use a 64-token queue. Token emission + still copies source text into strings. Queue byte limits need further work. - [cmd/sqlread/main.go](cmd/sqlread/main.go) configures a 100,000,000-byte reader buffer. Its capacity needs a measured justification against the memory budget. - [summary.go](summary.go) retains all data-region references in memory. @@ -253,8 +254,9 @@ as the tool evolves: values still need a bounded representation. - [cmd/sqlread/readlinewrap.go](cmd/sqlread/readlinewrap.go) connects parser reads directly to terminal input. -- [cmd/sqlread/query.go](cmd/sqlread/query.go) keeps execution in the command - package. Worker completion and output errors need a consistent owner. +- [execute.go](execute.go) streams query results outside the command package. + It owns the active lexer worker. It waits for completion on limits and errors. + Other scanning paths still need the same lifecycle review. Refinements must preserve the tool's ability to process very large dumps with limited memory. Keep the state-function foundation and the file as the backing diff --git a/cmd/sqlread/interactive_test.go b/cmd/sqlread/interactive_test.go new file mode 100644 index 0000000..ad9e6b2 --- /dev/null +++ b/cmd/sqlread/interactive_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "bufio" + "bytes" + "context" + "fmt" + "os" + "os/exec" + "strings" + "testing" + "time" +) + +func TestInteractiveQueries(t *testing.T) { + file := queryFixture(t) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestInteractiveProcess$", file.Name()) + cmd.Env = append(os.Environ(), "SQLREAD_TEST_INTERACTIVE=1") + var diagnostics bytes.Buffer + cmd.Stderr = &diagnostics + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + defer func() { + stdin.Close() + cancel() + cmd.Wait() + }() + + reader := bufio.NewReader(stdout) + for _, test := range []struct { + query string + rows []string + }{ + {"SELECT id FROM other;", []string{"99"}}, + {"SELECT id FROM records LIMIT 1, 2;", []string{"2", "3"}}, + {"SELECT id FROM records LIMIT 0;\nSELECT id FROM other;", []string{"99"}}, + } { + if _, err := fmt.Fprintln(stdin, test.query); err != nil { + t.Fatal(err) + } + // Keep stdin open. Each query must complete without another terminator or EOF. + for _, want := range test.rows { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatalf("query %q did not produce %q: %v", test.query, want, err) + } + if strings.TrimSpace(line) != want { + t.Fatalf("query %q: output = %q; want %q", test.query, line, want) + } + } + } + if _, err := fmt.Fprintln(stdin, "QUIT;"); err != nil { + t.Fatal(err) + } + if err := cmd.Wait(); err != nil { + t.Fatalf("interactive process: %v\n%s", err, diagnostics.String()) + } +} + +func TestInteractiveProcess(t *testing.T) { + if os.Getenv("SQLREAD_TEST_INTERACTIVE") != "1" { + return + } + main() + os.Exit(0) +} diff --git a/cmd/sqlread/query.go b/cmd/sqlread/query.go index ead2c82..6706ada 100644 --- a/cmd/sqlread/query.go +++ b/cmd/sqlread/query.go @@ -3,7 +3,6 @@ package main import ( "fmt" "io" - "log" "sort" "github.com/donatj/sqlread" @@ -12,74 +11,16 @@ import ( type DataWriter interface { Write(record []string) error Flush() + Error() error } func execQuery(tree sqlread.SummaryTree, qry sqlread.Query, buff io.ReaderAt, w DataWriter) error { - tbl, tok := tree[qry.Table] - if !tok { - return fmt.Errorf("table `%s` not found", qry.Table) - } - colind := []int{} - for _, col := range qry.Columns { - found := false - for tci, tcol := range tbl.Cols { - if col == "*" || col == tcol.Name { - found = true - colind = append(colind, tci) - } - } - - if !found { - return fmt.Errorf("column `%s` not found", col) - } - } - var rowsSeen uint64 - for _, loc := range tbl.DataLocs { - if qry.Limit != nil && qry.Limit.Exhausted(rowsSeen) { - return nil - } - - start := loc.Start.Pos - end := loc.End.Pos - - sl, sli := sqlread.LexSection(buff, start, end-start+1) - go func() { - sl.Run(sqlread.StartState) - }() - - sp := sqlread.NewInsertDetailParser() - - spr := sqlread.Parse(sli) - go func() { - err := spr.Run(sp.ParseStart) - if err != nil { - log.Fatal(err) - } - }() - - for row := range sp.Out { - if qry.Limit != nil && !qry.Limit.Includes(rowsSeen) { - rowsSeen++ - continue - } - rowsSeen++ - - out := make([]string, len(colind)) - for i, ci := range colind { - out[i] = row[ci] - } - - w.Write(out) - - if qry.Limit != nil && qry.Limit.Exhausted(rowsSeen) { - sp.Stop() - sl.Stop() - } - } - w.Flush() + err := sqlread.ExecuteQuery(tree, qry, buff, w.Write) + w.Flush() + if err != nil { + return err } - - return nil + return w.Error() } func showColumns(tree sqlread.SummaryTree, sctbl string, w DataWriter) error { diff --git a/cmd/sqlread/query_bench_test.go b/cmd/sqlread/query_bench_test.go new file mode 100644 index 0000000..cf9377a --- /dev/null +++ b/cmd/sqlread/query_bench_test.go @@ -0,0 +1,40 @@ +package main + +import ( + "encoding/csv" + "io" + "strings" + "testing" + + "github.com/donatj/sqlread" +) + +func BenchmarkExecQuery(b *testing.B) { + input := "CREATE TABLE `records` (`id` int);\nINSERT INTO `records` VALUES " + + strings.Repeat("(1),", 20000) + "(2);" + source := strings.NewReader(input) + l, items := sqlread.Lex(source) + done := make(chan struct{}) + go func() { defer close(done); l.Run(sqlread.StartState) }() + defer func() { l.Stop(); <-done }() + summary := sqlread.NewSummaryParser() + if err := sqlread.Parse(items).Run(summary.ParseStart); err != nil { + b.Fatal(err) + } + for _, test := range []struct { + name string + limit *sqlread.QueryLimit + }{{"all_rows", nil}, {"limit_10", &sqlread.QueryLimit{Count: 10}}} { + b.Run(test.name, func(b *testing.B) { + qry := sqlread.Query{Table: "records", Columns: []string{"*"}, Limit: test.limit} + writer := csv.NewWriter(io.Discard) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := execQuery(summary.Tree, qry, source, writer); err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/cmd/sqlread/query_test.go b/cmd/sqlread/query_test.go new file mode 100644 index 0000000..a94404e --- /dev/null +++ b/cmd/sqlread/query_test.go @@ -0,0 +1,245 @@ +package main + +import ( + "bytes" + "encoding/csv" + "errors" + "io" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + + "github.com/donatj/sqlread" + "github.com/donatj/sqlread/mapcache" +) + +func TestQueryWorkflow(t *testing.T) { + file := queryFixture(t) + tree := indexDump(t, file) + cache := mapcache.New(file) + if err := cache.Store(tree); err != nil { + t.Fatal(err) + } + cached, err := cache.Get() + if err != nil { + t.Fatal(err) + } + + tests := []struct { + query string + want string + }{ + {"SELECT id FROM records;", "1\n2\n3\n4\n5\n6\n"}, + {"SELECT * FROM records LIMIT 1;", "1,alpha\n"}, + {"SELECT * FROM records LIMIT 3;", "1,alpha\n2,\"two,parts\"\n3,\"three\nlines\"\n"}, + {"select id from records limit 1, 2;", "2\n3\n"}, + {"SELECT id FROM records LIMIT 2, 2;", "3\n4\n"}, + {"SELECT name, id FROM records LIMIT 3, 2;", "four,4\nfive,5\n"}, + {"SELECT id FROM records LIMIT 5, 10;", "6\n"}, + {"SELECT id FROM records LIMIT 6, 1;", ""}, + {"SELECT id FROM records LIMIT 0;", ""}, + {"SELECT id FROM records LIMIT 5, 0;", ""}, + {"SELECT id FROM records LIMIT 18446744073709551615, 2;", ""}, + {"SELECT id FROM records LIMIT 1, 18446744073709551615;", "2\n3\n4\n5\n6\n"}, + {"SELECT id FROM other LIMIT 2;", "99\n"}, + } + for _, indexed := range []struct { + name string + tree sqlread.SummaryTree + }{{"fresh", tree}, {"cached", cached}} { + t.Run(indexed.name, func(t *testing.T) { + for _, test := range tests { + t.Run(test.query, func(t *testing.T) { + qry := parseQuery(t, test.query) + var output bytes.Buffer + if err := execQuery(indexed.tree, qry, file, csv.NewWriter(&output)); err != nil { + t.Fatal(err) + } + if got := output.String(); got != test.want { + t.Errorf("output = %q; want %q", got, test.want) + } + }) + } + }) + } +} + +func TestQueryStopsReading(t *testing.T) { + // Many rows in one INSERT ensure a small LIMIT must stop inside a data block. + input := "CREATE TABLE `records` (`id` int);\nINSERT INTO `records` VALUES " + + strings.Repeat("(1),", 10000) + "(2);" + tree := indexDump(t, strings.NewReader(input)) + for _, query := range []string{ + "SELECT * FROM records LIMIT 0;", + "SELECT * FROM records LIMIT 500, 0;", + "SELECT * FROM records LIMIT 1;", + } { + t.Run(query, func(t *testing.T) { + source := &queryReader{ReaderAt: strings.NewReader(input)} + qry := parseQuery(t, query) + var output bytes.Buffer + if err := execQuery(tree, qry, source, csv.NewWriter(&output)); err != nil { + t.Fatal(err) + } + // This also detects a lexer worker still reading after execution returns. + reads := source.reads.Load() + if qry.Limit.Count == 0 { + if reads != 0 || output.Len() != 0 { + t.Fatalf("zero limit: reads = %d, output = %q", reads, output.String()) + } + } else if output.String() != "1\n" || source.farthest.Load() >= int64(len(input)/2) { + t.Fatalf("limit did not stop early: offset = %d, output = %q", source.farthest.Load(), output.String()) + } + source.closed.Store(true) + if reads != source.reads.Load() { + t.Fatal("query returned before its reads completed") + } + }) + } +} + +func TestQueryFailures(t *testing.T) { + file := queryFixture(t) + tree := indexDump(t, file) + for _, query := range []string{ + "SELECT missing FROM records LIMIT 1;", + "SELECT * FROM missing LIMIT 1;", + } { + t.Run(query, func(t *testing.T) { + if err := execQuery(tree, parseQuery(t, query), file, csv.NewWriter(io.Discard)); err == nil { + t.Fatal("expected a query error") + } + }) + } + + t.Run("source read", func(t *testing.T) { + want := errors.New("source read failed") + source := &queryReader{ReaderAt: file, err: want} + err := execQuery(tree, parseQuery(t, "SELECT * FROM records LIMIT 1;"), source, csv.NewWriter(io.Discard)) + if !errors.Is(err, want) { + t.Fatalf("error = %v; want %v", err, want) + } + }) + + t.Run("flush", func(t *testing.T) { + want := errors.New("output failed") + writer := csv.NewWriter(errorWriter{want}) + err := execQuery(tree, parseQuery(t, "SELECT * FROM records LIMIT 1;"), file, writer) + if !errors.Is(err, want) { + t.Fatalf("error = %v; want %v", err, want) + } + }) + + t.Run("write", func(t *testing.T) { + want := errors.New("row write failed") + source := &queryReader{ReaderAt: file} + writes := 0 + err := sqlread.ExecuteQuery(tree, parseQuery(t, "SELECT * FROM records;"), source, func([]string) error { + writes++ + return want + }) + source.closed.Store(true) + if !errors.Is(err, want) || writes != 1 { + t.Fatalf("error = %v, writes = %d; want %v and 1 write", err, writes, want) + } + }) + + for _, test := range []struct { + row string + want string + }{ + {"(1)", "row has 1 columns; expected 2"}, + {"(1,'bad\\x')", "invalid string"}, + {"(1,unsupported)", "TIllegal"}, + {"(1,'unterminated", "EOF"}, + } { + t.Run(test.row, func(t *testing.T) { + // Keep valid table metadata. Point its data span at the entire malformed INSERT. + input := "CREATE TABLE `records` (`id` int, `name` text);\n" + valid := input + "INSERT INTO `records` VALUES (1,'valid');" + indexed := indexDump(t, strings.NewReader(valid)) + bad := input + "INSERT INTO `records` VALUES " + test.row + ";" + indexed["records"].DataLocs[0].End.Pos = int64(len(bad) - 1) + err := execQuery(indexed, parseQuery(t, "SELECT * FROM records;"), strings.NewReader(bad), csv.NewWriter(io.Discard)) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("error = %v; want %q", err, test.want) + } + }) + } +} + +func queryFixture(t testing.TB) *os.File { + t.Helper() + data, err := os.ReadFile("../../testfixtures/queries.sql") + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "dump.sql") + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + file, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { file.Close() }) + return file +} + +func indexDump(t testing.TB, source io.ReaderAt) sqlread.SummaryTree { + t.Helper() + l, items := sqlread.Lex(source) + done := make(chan struct{}) + go func() { defer close(done); l.Run(sqlread.StartState) }() + defer func() { l.Stop(); <-done }() + summary := sqlread.NewSummaryParser() + if err := sqlread.Parse(items).Run(summary.ParseStart); err != nil { + t.Fatal(err) + } + return summary.Tree +} + +func parseQuery(t testing.TB, input string) sqlread.Query { + t.Helper() + l, items := sqlread.Lex(strings.NewReader(input)) + done := make(chan struct{}) + intp := &sqlread.Intp{} + go func() { defer close(done); l.Run(intp.StartIntpState) }() + defer func() { l.Stop(); <-done }() + parser := sqlread.NewQueryParser() + if err := sqlread.Parse(items).Run(parser.ParseStart); err != nil { + t.Fatal(err) + } + if len(parser.Tree.Queries) != 1 { + t.Fatalf("parsed %d queries; want 1", len(parser.Tree.Queries)) + } + return parser.Tree.Queries[0] +} + +type queryReader struct { + io.ReaderAt + reads atomic.Int64 + farthest atomic.Int64 + closed atomic.Bool + err error +} + +func (r *queryReader) ReadAt(b []byte, off int64) (int, error) { + if r.closed.Load() { + panic("source used after query returned") + } + r.reads.Add(1) + if off+int64(len(b)) > r.farthest.Load() { + r.farthest.Store(off + int64(len(b))) + } + if r.err != nil { + return 0, r.err + } + return r.ReaderAt.ReadAt(b, off) +} + +type errorWriter struct{ err error } + +func (w errorWriter) Write([]byte) (int, error) { return 0, w.err } diff --git a/detail.go b/detail.go index 8cfc3bf..5f8191b 100644 --- a/detail.go +++ b/detail.go @@ -3,53 +3,33 @@ package sqlread import ( "errors" "strings" - "sync" ) type InsertDetailParser struct { Out chan []string ColInd []int - done chan struct{} - stop sync.Once } func NewInsertDetailParser() *InsertDetailParser { return &InsertDetailParser{ - Out: make(chan []string, 100), - done: make(chan struct{}), - } -} - -func (d *InsertDetailParser) Stop() { - d.stop.Do(func() { - close(d.done) - }) -} - -func (d *InsertDetailParser) stopped() bool { - select { - case <-d.done: - return true - default: - return false - } -} - -func (d *InsertDetailParser) writeRow(row []string) bool { - select { - case d.Out <- row: - return true - case <-d.done: - return false + Out: make(chan []string, 100), } } func (d *InsertDetailParser) ParseStart(p *Parser) parseState { defer close(d.Out) + return parseInsertRows(p, func(row []string) bool { + d.Out <- row + return true + }) +} +// parseInsertRows stops when yield returns false. It does not retain completed rows. +func parseInsertRows(p *Parser, yield func([]string) bool) parseState { cii, ok := p.scan() if !ok { p.errorUnexpectedEOF() + return nil } if cii.Type != TInsertInto { @@ -70,19 +50,14 @@ func (d *InsertDetailParser) ParseStart(p *Parser) parseState { row := []string{} for { - if d.stopped() { - return nil - } - v, ok := p.scan() if !ok { - if d.stopped() { - return nil - } p.errorUnexpectedEOF() + return nil } switch v.Type { + case TInsertValues: case TString: val, err := stringValue(v) if err != nil { @@ -98,12 +73,15 @@ func (d *InsertDetailParser) ParseStart(p *Parser) parseState { case TNull: row = append(row, "\u0000") // investigate case TComma: - if !d.writeRow(row) { + if !yield(row) { return nil } row = []string{} case TSemi: - d.writeRow(row) + yield(row) + return nil + default: + p.errorUnexpectedLex(v, TInsertValues, TString, TNumber, THexLiteral, TNull, TComma, TSemi) return nil } } diff --git a/detail_test.go b/detail_test.go index b66d402..0780f18 100644 --- a/detail_test.go +++ b/detail_test.go @@ -4,7 +4,6 @@ import ( "reflect" "strings" "testing" - "time" ) func TestStringValue(t *testing.T) { @@ -97,33 +96,3 @@ func TestInsertDetailControlEscapes(t *testing.T) { }) } } - -func TestInsertDetailParserStop(t *testing.T) { - input := "INSERT INTO records VALUES " + strings.Repeat("(1),", 200) + "(1);" - l, items := Lex(strings.NewReader(input)) - go l.Run(StartState) - - detail := NewInsertDetailParser() - done := make(chan error, 1) - go func() { - done <- Parse(items).Run(detail.ParseStart) - }() - - for range 10 { - <-detail.Out - } - - detail.Stop() - l.Stop() - for range detail.Out { - } - - select { - case err := <-done: - if err != nil { - t.Fatal(err) - } - case <-time.After(time.Second): - t.Fatal("parser did not stop") - } -} diff --git a/execute.go b/execute.go new file mode 100644 index 0000000..478cf49 --- /dev/null +++ b/execute.go @@ -0,0 +1,93 @@ +package sqlread + +import ( + "fmt" + "io" +) + +// ExecuteQuery reads a SELECT's rows from the dump and passes them to writeRow. +// It stops after the requested limit or the first error. The caller owns the +// output destination and must flush it. Outfile is handled by the caller. +func ExecuteQuery(tree SummaryTree, qry Query, source io.ReaderAt, writeRow func([]string) error) error { + table, ok := tree[qry.Table] + if !ok { + return fmt.Errorf("table `%s` not found", qry.Table) + } + + var columns []int + for _, name := range qry.Columns { + found := false + for i, col := range table.Cols { + if name == "*" || name == col.Name { + columns = append(columns, i) + found = true + } + } + if !found { + return fmt.Errorf("column `%s` not found", name) + } + } + + var offset, remaining uint64 + if qry.Limit != nil { + offset, remaining = qry.Limit.Offset, qry.Limit.Count + } + for _, loc := range table.DataLocs { + if qry.Limit != nil && remaining == 0 { + break + } + + var rowErr error + err := readInsertRows(source, loc, func(row []string) bool { + if len(row) != len(table.Cols) { + rowErr = fmt.Errorf("table `%s`: row has %d columns; expected %d", qry.Table, len(row), len(table.Cols)) + return false + } + if offset > 0 { + offset-- + return true + } + + out := make([]string, len(columns)) + for i, col := range columns { + out[i] = row[col] + } + if rowErr = writeRow(out); rowErr != nil { + return false + } + if qry.Limit != nil { + remaining-- + return remaining > 0 + } + return true + }) + if rowErr != nil { + return rowErr + } + if err != nil { + return err + } + } + return nil +} + +// readInsertRows owns the lexer worker and waits for it on every return path. +func readInsertRows(source io.ReaderAt, loc SummaryDataLoc, yield func([]string) bool) (err error) { + l, items := LexSection(source, loc.Start.Pos, loc.End.Pos-loc.Start.Pos+1) + done := make(chan struct{}) + go func() { + defer close(done) + l.Run(StartState) + }() + defer func() { + l.Stop() + <-done + if l.err != nil { + err = l.err + } + }() + + return Parse(items).Run(func(p *Parser) parseState { + return parseInsertRows(p, yield) + }) +} diff --git a/query.go b/query.go index 29b8ae8..45ea125 100644 --- a/query.go +++ b/query.go @@ -10,14 +10,6 @@ type QueryLimit struct { Count uint64 } -func (l QueryLimit) Includes(row uint64) bool { - return row >= l.Offset && row-l.Offset < l.Count -} - -func (l QueryLimit) Exhausted(rowsSeen uint64) bool { - return l.Count == 0 || (rowsSeen >= l.Offset && rowsSeen-l.Offset >= l.Count) -} - type Query struct { Columns []string Table string diff --git a/query_test.go b/query_test.go index dfd3b25..18adb3b 100644 --- a/query_test.go +++ b/query_test.go @@ -47,6 +47,9 @@ func TestQueryParserLimitRejectsNonIntegers(t *testing.T) { "SELECT * FROM records LIMIT -1;", "SELECT * FROM records LIMIT 1.5;", "SELECT * FROM records LIMIT 1, count;", + "SELECT * FROM records LIMIT 18446744073709551616;", + "SELECT * FROM records LIMIT 1, 18446744073709551616;", + "SELECT * FROM records LIMIT 1 + 1;", } { t.Run(input, func(t *testing.T) { _, err := parseUserQuery(input) @@ -77,51 +80,6 @@ func TestQueryParserDoesNotReadPastSemicolon(t *testing.T) { } } -func TestQueryLimitIncludes(t *testing.T) { - tests := []struct { - name string - limit QueryLimit - row uint64 - want bool - }{ - {name: "before offset", limit: QueryLimit{Offset: 1, Count: 2}, row: 0, want: false}, - {name: "at offset", limit: QueryLimit{Offset: 1, Count: 2}, row: 1, want: true}, - {name: "within count", limit: QueryLimit{Offset: 1, Count: 2}, row: 2, want: true}, - {name: "after count", limit: QueryLimit{Offset: 1, Count: 2}, row: 3, want: false}, - {name: "zero count", limit: QueryLimit{Count: 0}, row: 0, want: false}, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - if actual := test.limit.Includes(test.row); actual != test.want { - t.Errorf("Includes(%d) = %t; want %t", test.row, actual, test.want) - } - }) - } -} - -func TestQueryLimitExhausted(t *testing.T) { - tests := []struct { - name string - limit QueryLimit - rowsSeen uint64 - want bool - }{ - {name: "before offset", limit: QueryLimit{Offset: 1, Count: 2}, rowsSeen: 0, want: false}, - {name: "within count", limit: QueryLimit{Offset: 1, Count: 2}, rowsSeen: 2, want: false}, - {name: "after count", limit: QueryLimit{Offset: 1, Count: 2}, rowsSeen: 3, want: true}, - {name: "zero count", limit: QueryLimit{Count: 0}, rowsSeen: 0, want: true}, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - if actual := test.limit.Exhausted(test.rowsSeen); actual != test.want { - t.Errorf("Exhausted(%d) = %t; want %t", test.rowsSeen, actual, test.want) - } - }) - } -} - func parseUserQuery(input string) (QueryTree, error) { return parseUserQueryReader(strings.NewReader(input)) } @@ -129,7 +87,15 @@ func parseUserQuery(input string) (QueryTree, error) { func parseUserQueryReader(input io.ReaderAt) (QueryTree, error) { l, items := Lex(input) intp := &Intp{} - go l.Run(intp.StartIntpState) + done := make(chan struct{}) + go func() { + defer close(done) + l.Run(intp.StartIntpState) + }() + defer func() { + l.Stop() + <-done + }() query := NewQueryParser() err := Parse(items).Run(query.ParseStart) diff --git a/scanner.go b/scanner.go index 4298b23..ed2523a 100644 --- a/scanner.go +++ b/scanner.go @@ -78,6 +78,7 @@ type lexer struct { items chan LexItem done chan struct{} stop sync.Once + err error } const ( @@ -116,12 +117,16 @@ func (l *lexer) rewind() { func (l *lexer) peek(s int) (int, []byte) { b := make([]byte, s) + if l.stopped() || l.err != nil { + return 0, b + } n, err := l.input.ReadAt(b, l.pos) if err == io.EOF { return n, b } else if err != nil { - panic(err) + l.err = err + return 0, b } return n, b @@ -154,11 +159,17 @@ var ( func (l *lexer) Run(start state) { for state := start; state != nil; { - if l.stopped() { + if l.stopped() || l.err != nil { break } state = state(l) } + if l.err != nil { + select { + case l.items <- LexItem{Type: TIllegal, Val: l.err.Error(), Pos: l.pos}: + case <-l.done: + } + } close(l.items) } @@ -262,7 +273,8 @@ func Lex(input io.ReaderAt) (*lexer, chan LexItem) { func LexSection(input io.ReaderAt, off, n int64) (*lexer, chan LexItem) { l := &lexer{ input: io.NewSectionReader(input, off, n), - items: make(chan LexItem, 2000000), + // Keep section lookahead small so short queries do not scan far ahead. + items: make(chan LexItem, 64), done: make(chan struct{}), } @@ -270,8 +282,14 @@ func LexSection(input io.ReaderAt, off, n int64) (*lexer, chan LexItem) { } func (l *lexer) emit(t lexItemType) LexItem { + if l.stopped() || l.err != nil { + return LexItem{Type: t, Pos: l.start} + } b := make([]byte, l.pos-l.start) - l.input.ReadAt(b, l.start) + if _, err := l.input.ReadAt(b, l.start); err != nil { + l.err = err + return LexItem{Type: t, Pos: l.start} + } li := LexItem{ Type: t, diff --git a/testfixtures/queries.sql b/testfixtures/queries.sql new file mode 100644 index 0000000..74828ae --- /dev/null +++ b/testfixtures/queries.sql @@ -0,0 +1,11 @@ +CREATE TABLE `records` ( + `id` int NOT NULL, + `name` varchar(255) NOT NULL +); +INSERT INTO `records` VALUES (1,'alpha'),(2,'two,parts'); +CREATE TABLE `other` ( + `id` int NOT NULL +); +INSERT INTO `other` VALUES (99); +INSERT INTO `records` VALUES (3,'three\nlines'),(4,'four'); +INSERT INTO `records` VALUES (5,'five'),(6,'six');