Stream query execution from the library

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

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 {

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)
+				}
+			}
+		})
+	}
+}

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
 		}
 	}

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)
+	})
+}

execute_test.go

@@ -0,0 +1,96 @@
+package sqlread
+
+import (
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestExecuteQuery(t *testing.T) {
+	input := "CREATE TABLE `records` (\n" +
+		"  `id` int NOT NULL,\n" +
+		"  `name` varchar(255) NOT NULL\n" +
+		");\n" +
+		"INSERT INTO `records` VALUES (1,'one'),(2,'two');\n" +
+		"INSERT INTO `records` VALUES (3,'three'),(4,'four');"
+	source := strings.NewReader(input)
+	tree := indexExecuteTestDump(t, source)
+
+	var rows [][]string
+	err := ExecuteQuery(tree, Query{
+		Table:   "records",
+		Columns: []string{"name", "id"},
+		Limit:   &QueryLimit{Offset: 1, Count: 2},
+	}, source, func(row []string) error {
+		rows = append(rows, row)
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want := [][]string{{"two", "2"}, {"three", "3"}}; !reflect.DeepEqual(rows, want) {
+		t.Errorf("rows = %#v; want %#v", rows, want)
+	}
+}
+
+func TestExecuteQueryStopsOnOutputError(t *testing.T) {
+	input := "CREATE TABLE `records` (`id` int);\nINSERT INTO `records` VALUES (1),(2);"
+	source := strings.NewReader(input)
+	tree := indexExecuteTestDump(t, source)
+	want := errors.New("output failed")
+	calls := 0
+
+	err := ExecuteQuery(tree, Query{Table: "records", Columns: []string{"*"}}, source, func([]string) error {
+		calls++
+		return want
+	})
+	if !errors.Is(err, want) {
+		t.Fatalf("error = %v; want %v", err, want)
+	}
+	if calls != 1 {
+		t.Errorf("write calls = %d; want 1", calls)
+	}
+}
+
+func TestExecuteQueryZeroLimitReadsNoRows(t *testing.T) {
+	input := "CREATE TABLE `records` (`id` int);\nINSERT INTO `records` VALUES (1),(2);"
+	source := strings.NewReader(input)
+	tree := indexExecuteTestDump(t, source)
+	calls := 0
+
+	err := ExecuteQuery(tree, Query{
+		Table:   "records",
+		Columns: []string{"*"},
+		Limit:   &QueryLimit{},
+	}, source, func([]string) error {
+		calls++
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if calls != 0 {
+		t.Errorf("write calls = %d; want 0", calls)
+	}
+}
+
+func indexExecuteTestDump(t *testing.T, source *strings.Reader) SummaryTree {
+	t.Helper()
+	l, items := Lex(source)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		l.Run(StartState)
+	}()
+	defer func() {
+		l.Stop()
+		<-done
+	}()
+
+	summary := NewSummaryParser()
+	if err := Parse(items).Run(summary.ParseStart); err != nil {
+		t.Fatal(err)
+	}
+	return summary.Tree
+}