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/README.md b/README.md index 75d91a5..8380f4b 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,27 @@ 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. +## User query grammar + +Square brackets (`[]`) mark optional clauses. Braces (`{}`) require one choice. +A vertical bar (`|`) separates choices. An ellipsis (`...`) marks a repeated item. + +``` +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 ``` @@ -50,13 +71,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/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 35c0451..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,63 +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) - } - } - for _, loc := range tbl.DataLocs { - 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, ok := <-sp.Out - if !ok { - w.Flush() - break - } - - out := make([]string, len(colind)) - for i, ci := range colind { - out[i] = row[ci] - } - - w.Write(out) - } + 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 36022dd..5f8191b 100644 --- a/detail.go +++ b/detail.go @@ -17,9 +17,19 @@ func NewInsertDetailParser() *InsertDetailParser { } 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 { @@ -43,9 +53,11 @@ func (d *InsertDetailParser) ParseStart(p *Parser) parseState { v, ok := p.scan() if !ok { p.errorUnexpectedEOF() + return nil } switch v.Type { + case TInsertValues: case TString: val, err := stringValue(v) if err != nil { @@ -61,11 +73,15 @@ func (d *InsertDetailParser) ParseStart(p *Parser) parseState { case TNull: row = append(row, "\u0000") // investigate case TComma: - d.Out <- row + if !yield(row) { + return nil + } row = []string{} case TSemi: - d.Out <- row - close(d.Out) + yield(row) + return nil + default: + p.errorUnexpectedLex(v, TInsertValues, TString, TNumber, THexLiteral, TNull, TComma, TSemi) return nil } } 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/interpreter.go b/interpreter.go index fccce03..c69883f 100644 --- a/interpreter.go +++ b/interpreter.go @@ -201,8 +201,20 @@ func (intp *Intp) selectFromIntpState(l *lexer) state { l.accept(whitespace) l.start = l.pos - 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 + } + } + if c == byte('i') || c == byte('I') { if l.hasPrefixI("NTO") { l.pos += 3 @@ -219,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 } @@ -244,6 +251,68 @@ func (intp *Intp) selectFromIntoOutfileIntpState(l *lexer) state { } l.emit(TString) + l.accept(whitespace) + l.start = l.pos + 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 + } + } + + l.start = l.pos + l.emit(TIllegal) + 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..45ea125 100644 --- a/query.go +++ b/query.go @@ -1,11 +1,20 @@ package sqlread -import "strings" +import ( + "strconv" + "strings" +) + +type QueryLimit struct { + Offset uint64 + Count uint64 +} type Query struct { Columns []string Table string Outfile *string + Limit *QueryLimit } type QueryTree struct { @@ -214,7 +223,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 +263,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..18adb3b --- /dev/null +++ b/query_test.go @@ -0,0 +1,125 @@ +package sqlread + +import ( + "io" + "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;", + "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) + if err == nil { + t.Fatal("parseUserQuery() returned nil error") + } + }) + } +} + +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 parseUserQuery(input string) (QueryTree, error) { + return parseUserQueryReader(strings.NewReader(input)) +} + +func parseUserQueryReader(input io.ReaderAt) (QueryTree, error) { + l, items := Lex(input) + intp := &Intp{} + 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) + 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 +} diff --git a/scanner.go b/scanner.go index cb416c5..ed2523a 100644 --- a/scanner.go +++ b/scanner.go @@ -4,6 +4,7 @@ import ( "io" "slices" "strings" + "sync" "unicode/utf8" ) @@ -51,6 +52,7 @@ const ( TIntpStar TIntpFrom TIntpIntoOutfile + TIntpLimit TIntpShowTables TIntpShowColumns @@ -74,6 +76,9 @@ type lexer struct { pos int64 // width int items chan LexItem + done chan struct{} + stop sync.Once + err error } const ( @@ -112,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 @@ -150,8 +159,17 @@ var ( func (l *lexer) Run(start state) { for state := start; state != nil; { + 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) } @@ -246,6 +264,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 @@ -254,15 +273,23 @@ 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{}), } return l, l.items } 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, @@ -270,11 +297,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) } 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');