- commit
- 34eef0930d55c840e2e495b444a9822e8c65abdd
- parent
- 0173b14dbd70762991374ced6c1eaf73ea2c06c8
- author
- Dami <dami.etoile@gmail.com>
- kind
- feat risk score 2.2 of 3, Jev's estimate
| M | internal/ | 3 | + |
| M | internal/ | 62 | +++++- |
| A | internal/ | 330 | +++++++++++++++++++++++++++ |
| M | internal/ | 3 | + |
| A | internal/ | 288 | ++++++++++++++++++++++++ |
| A | internal/ | 191 | ++++++++++++++++ |
| A | internal/ | 364 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 209 | +++++++++++++++++ |
| A | internal/ | 329 | +++++++++++++++++++++++++++ |
| A | internal/ | 204 | +++++++++++++++++ |
| A | internal/ | 216 | ++++++++++++++++++ |
| M | internal/ | 3 | + |
@@ -20,6 +20,9 @@ import (
2020 type parquetColumn struct {2121 logical format.LogicalTypeValue2222 kind parquet.Kind23+ // repeated is set on a list's column, whose rows hold several24+ // values; sources read it a row at a time (sourceparquet.go).25+ repeated bool2326 }2427 2528 // importParquet reads every leaf column, named by its path, into a@@ -1,8 +1,12 @@
11 package fileio22 33 import (4+ "cmp"45 "context"56 "errors"7+ "fmt"8+ "os"9+ "slices"610 711 "github.com/FelineStateMachine/012/internal/sheet"812 )@@ -98,7 +102,63 @@ type SourceView interface {
98102 // Parquet nor SQLite.99103 var ErrNotSource = errors.New("only Parquet files and SQLite tables or queries can be linked as sources")100104 105+// SourceKind is the format spec reads, Parquet or SQLite, by its Format106+// or else the file's extension.107+func SourceKind(spec SourceSpec) (Kind, error) {108+ k, ok := KindNamed(spec.Format)109+ if spec.Format == "" {110+ k, ok = KindOf(spec.Path)111+ }112+ if !ok || k != Parquet && k != SQLite {113+ return 0, ErrNotSource114+ }115+ return k, nil116+}117+101118 // OpenSource opens what spec names.102119 func OpenSource(ctx context.Context, spec SourceSpec) (Source, error) {103- return nil, ErrNotSource120+ k, err := SourceKind(spec)121+ if err != nil {122+ return nil, err123+ }124+ if spec.TempDir == "" {125+ spec.TempDir = os.TempDir()126+ }127+ if k == Parquet {128+ return openParquetSource(spec)129+ }130+ return openSQLiteSource(ctx, spec)131+}132+133+// allCols is cols, or every one of n columns when cols is nil.134+func allCols(cols []int, n int) []int {135+ if cols != nil {136+ return cols137+ }138+ out := make([]int, n)139+ for i := range out {140+ out[i] = i141+ }142+ return out143+}144+145+// checkCols reports a column past the source's n.146+func checkCols(cols []int, n int) error {147+ for _, c := range cols {148+ if c < 0 || c >= n {149+ return fmt.Errorf("the source has no column %d", c+1)150+ }151+ }152+ return nil153+}154+155+// fetchOrder is the positions of rows sorted by row number, so a fetch156+// reads forward, the answers going back to the order asked.157+func fetchOrder(rows []int64) []int {158+ idx := make([]int, len(rows))159+ for i := range idx {160+ idx[i] = i161+ }162+ slices.SortFunc(idx, func(a, b int) int { return cmp.Compare(rows[a], rows[b]) })163+ return idx104164 }@@ -0,0 +1,330 @@
1+package fileio2+3+import (4+ "context"5+ "fmt"6+ "os"7+ "path/filepath"8+ "slices"9+ "strings"10+ "sync"11+ "testing"12+ "time"13+14+ "github.com/parquet-go/parquet-go"15+16+ "github.com/FelineStateMachine/012/internal/sheet"17+)18+19+// sale is a row of the Parquet sources tests read.20+type sale struct {21+ ID int64 `parquet:"id"`22+ Cat *string `parquet:"cat,optional"`23+ Amount *float64 `parquet:"amount,optional"`24+ Day int32 `parquet:"day,date"`25+ Tags []string `parquet:"tags,list"`26+}27+28+var saleCats = []string{"North", "south", "East", "west", "Éire", "", "north"}29+30+// writeSales writes n rows in row groups of group rows.31+func writeSales(t testing.TB, dir string, n, group int) string {32+ t.Helper()33+ name := filepath.Join(dir, "sales.parquet")34+ f, err := os.Create(name)35+ if err != nil {36+ t.Fatal(err)37+ }38+ w := parquet.NewGenericWriter[sale](f, parquet.MaxRowsPerRowGroup(int64(group)), parquet.PageBufferSize(512))39+ day := int32(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).Unix() / 86400)40+ rows := make([]sale, n)41+ for i := range rows {42+ r := &rows[i]43+ r.ID, r.Day = int64(i), day+int32(i%40)44+ if i%9 != 4 {45+ c := saleCats[i%len(saleCats)]46+ r.Cat = &c47+ }48+ if i%11 != 7 {49+ a := float64((i*37)%101) - 20.550+ r.Amount = &a51+ }52+ if i%5 == 0 {53+ r.Tags = []string{"a", fmt.Sprint(i)}54+ }55+ }56+ if _, err := w.Write(rows); err != nil {57+ t.Fatal(err)58+ }59+ if err := w.Close(); err != nil {60+ t.Fatal(err)61+ }62+ f.Close()63+ return name64+}65+66+// cellsText is a row as its cells show, separated by |.67+func cellsText(row []sheet.LiveCell) string {68+ parts := make([]string, len(row))69+ for i, c := range row {70+ parts[i] = sheet.FormatText(c.V, c.F)71+ }72+ return strings.Join(parts, "|")73+}74+75+// imported is a file imported as a sheet, which sources must read as76+// the import does, and the rows under its header.77+func imported(t *testing.T, name string, opt Options) (*sheet.Sheet, sheet.Rect) {78+ t.Helper()79+ res, err := Import(context.Background(), name, opt)80+ if err != nil {81+ t.Fatal(err)82+ }83+ used, _ := res.Sheet.UsedRange()84+ used.From.Row = 185+ return res.Sheet, used86+}87+88+// sheetRows are the rows of r a sheet shows, the filter's hidden ones89+// left out.90+func sheetRows(s *sheet.Sheet, r sheet.Rect) []string {91+ var out []string92+ for row := r.From.Row; row <= r.To.Row; row++ {93+ if s.RowHidden(row) {94+ continue95+ }96+ parts := make([]string, 0, r.To.Col+1)97+ for c := r.From.Col; c <= r.To.Col; c++ {98+ parts = append(parts, shown(s, sheet.Addr{Col: c, Row: row}))99+ }100+ out = append(out, strings.Join(parts, "|"))101+ }102+ return out103+}104+105+// viewRows pages through a whole view, n rows at a time.106+func viewRows(t *testing.T, v SourceView, n int) ([]int64, []string) {107+ t.Helper()108+ var nums []int64109+ var out []string110+ for from := int64(0); from < v.Rows(); from += int64(n) {111+ ns, rows, err := v.Page(context.Background(), from, n)112+ if err != nil {113+ t.Fatal(err)114+ }115+ if len(ns) != min(n, int(v.Rows()-from)) {116+ t.Fatalf("page at %d has %d rows", from, len(ns))117+ }118+ for i, r := range rows {119+ nums, out = append(nums, ns[i]), append(out, cellsText(r))120+ }121+ }122+ return nums, out123+}124+125+// orders are the sorts and filters every source is checked with,126+// against a sheet sorting and filtering the same rows.127+var orders = []SourceOrder{128+ {},129+ {Sort: []SourceSort{{Col: 1}}},130+ {Sort: []SourceSort{{Col: 1, Desc: true}}},131+ {Sort: []SourceSort{{Col: 2, Desc: true}, {Col: 1}}},132+ {Filter: []SourceFilter{{Col: 2, Cond: sheet.Condition{Op: sheet.CondGreater, Arg: "30"}}}},133+ {Filter: []SourceFilter{{Col: 1, Cond: sheet.Condition{Op: sheet.CondContains, Arg: "OR"}}}},134+ {Filter: []SourceFilter{{Col: 1, Cond: sheet.Condition{Op: sheet.CondEmpty}}}},135+ {Filter: []SourceFilter{{Col: 1, Cond: sheet.Condition{Op: sheet.CondNotEqual, Arg: "north"}}}},136+ {Filter: []SourceFilter{{Col: 1, Cond: sheet.Condition{Op: sheet.CondStartsWith, Arg: "e"}}}},137+ {Filter: []SourceFilter{{Col: 1, Cond: sheet.Condition{Op: sheet.CondEndsWith, Arg: "TH"}}}},138+ {Filter: []SourceFilter{{Col: 1, Cond: sheet.Condition{Op: sheet.CondExactly, Arg: "EAST"}}}},139+ {Filter: []SourceFilter{{Col: 2, Cond: sheet.Condition{Op: sheet.CondLessEq, Arg: "-10"}}},140+ Sort: []SourceSort{{Col: 2}}},141+ {Filter: []SourceFilter{{Col: 1, Cond: sheet.Condition{Op: sheet.CondNotEmpty}},142+ {Col: 2, Cond: sheet.Condition{Op: sheet.CondGreaterEq, Arg: "0"}}},143+ Sort: []SourceSort{{Col: 1, Desc: true}}},144+}145+146+// checkOrders checks each order's view against the sheet the file147+// imports as, sorted and filtered the same way.148+func checkOrders(t *testing.T, src Source, name string, opt Options, list []SourceOrder) {149+ t.Helper()150+ for _, o := range list {151+ s, r := imported(t, name, opt)152+ if len(o.Sort) > 0 {153+ var keys []sheet.SortKey154+ for _, k := range o.Sort {155+ keys = append(keys, sheet.SortKey{Col: k.Col, Desc: k.Desc})156+ }157+ s.SortRange(r, keys)158+ }159+ if len(o.Filter) > 0 {160+ s.CreateFilter(sheet.Rect{To: r.To})161+ for _, f := range o.Filter {162+ s.FilterColumn(f.Col, sheet.Criteria{Cond: f.Cond})163+ }164+ }165+ want := sheetRows(s, r)166+ v, err := src.View(context.Background(), o)167+ if err != nil {168+ t.Fatalf("%+v: %v", o, err)169+ }170+ _, got := viewRows(t, v, 37)171+ if !slices.Equal(got, want) {172+ i := 0173+ for i < min(len(got), len(want)) && got[i] == want[i] {174+ i++175+ }176+ t.Errorf("%+v: %d rows, want %d; from row %d\n got %q\nwant %q", o, len(got), len(want), i, head(got[i:]), head(want[i:]))177+ }178+ if err := v.Close(); err != nil {179+ t.Error(err)180+ }181+ }182+}183+184+func head(rows []string) []string { return rows[:min(len(rows), 12)] }185+186+func TestParquetSource(t *testing.T) {187+ dir := t.TempDir()188+ name := writeSales(t, dir, 1000, 128)189+ tmp := t.TempDir()190+ src, err := OpenSource(context.Background(), SourceSpec{Path: name, TempDir: tmp})191+ if err != nil {192+ t.Fatal(err)193+ }194+ defer src.Close()195+ if src.Rows() != 1000 {196+ t.Fatalf("%d rows", src.Rows())197+ }198+ var cols []string199+ for _, c := range src.Columns() {200+ cols = append(cols, fmt.Sprintf("%s:%v:%v", c.Name, c.Numeric, c.Format.Kind == sheet.FmtDate))201+ }202+ if got := strings.Join(cols, " "); got != "id:true:false cat:false:false amount:true:false day:true:true tags.list.element:false:false" {203+ t.Errorf("columns %s", got)204+ }205+ s, r := imported(t, name, Options{})206+ want := sheetRows(s, r)207+208+ // Scan from part way, two columns in another order.209+ var got []string210+ err = src.Scan(context.Background(), 500, []int{2, 0}, func(row int64, vals []sheet.LiveCell) bool {211+ got = append(got, fmt.Sprintf("%d:%s", row, cellsText(vals)))212+ return row < 510213+ })214+ if err != nil {215+ t.Fatal(err)216+ }217+ for i, g := range got {218+ parts := strings.Split(want[500+i], "|")219+ if w := fmt.Sprintf("%d:%s|%s", 500+i, parts[2], parts[0]); g != w {220+ t.Errorf("scan %q, want %q", g, w)221+ }222+ }223+ if len(got) != 11 {224+ t.Errorf("scan read %d rows", len(got))225+ }226+227+ // Fetch rows in any order, across row groups.228+ nums := []int64{999, 3, 500, 128, 127, 3}229+ rows, err := src.Fetch(context.Background(), nums, nil)230+ if err != nil {231+ t.Fatal(err)232+ }233+ for i, n := range nums {234+ if g := cellsText(rows[i]); g != want[n] {235+ t.Errorf("row %d is %q, want %q", n, g, want[n])236+ }237+ }238+ checkOrders(t, src, name, Options{}, orders)239+ if left, _ := os.ReadDir(tmp); len(left) != 0 {240+ t.Errorf("views left %d files", len(left))241+ }242+}243+244+// TestSourceSortSpills sorts in runs spilled to files and merged.245+func TestSourceSortSpills(t *testing.T) {246+ defer func(n int) { sortRunBytes = n }(sortRunBytes)247+ sortRunBytes = 4096248+ name := writeSales(t, t.TempDir(), 3000, 1000)249+ tmp := t.TempDir()250+ src, err := OpenSource(context.Background(), SourceSpec{Path: name, TempDir: tmp})251+ if err != nil {252+ t.Fatal(err)253+ }254+ checkOrders(t, src, name, Options{}, orders[1:4])255+ v, err := src.View(context.Background(), orders[2])256+ if err != nil {257+ t.Fatal(err)258+ }259+ if left, _ := os.ReadDir(tmp); len(left) != 1 {260+ t.Errorf("a view keeps %d files, want its row order", len(left))261+ }262+ if err := src.Close(); err != nil {263+ t.Fatal(err)264+ }265+ if left, _ := os.ReadDir(tmp); len(left) != 0 {266+ t.Errorf("closing left %d files", len(left))267+ }268+ if err := v.Close(); err != nil {269+ t.Errorf("closing a view after its source: %v", err)270+ }271+}272+273+// TestSourceConcurrent reads a source from several goroutines at once.274+func TestSourceConcurrent(t *testing.T) {275+ dir := t.TempDir()276+ for _, name := range []string{writeSales(t, dir, 2000, 300), writeSalesDB(t, dir, 2000, false)} {277+ src, err := OpenSource(context.Background(), SourceSpec{Path: name, Table: "sales", TempDir: dir})278+ if err != nil {279+ t.Fatal(err)280+ }281+ sum := func() float64 {282+ total := 0.0283+ err := src.Scan(context.Background(), 0, []int{2}, func(_ int64, v []sheet.LiveCell) bool {284+ total += v[0].V.Num285+ return true286+ })287+ if err != nil {288+ t.Error(err)289+ }290+ return total291+ }292+ want := sum()293+ var wg sync.WaitGroup294+ for i := range 8 {295+ wg.Go(func() {296+ if got := sum(); got != want {297+ t.Errorf("%s: sum %v, want %v", name, got, want)298+ }299+ if _, err := src.Fetch(context.Background(), []int64{int64(i * 200), 1999}, nil); err != nil {300+ t.Error(err)301+ }302+ v, err := src.View(context.Background(), SourceOrder{Sort: []SourceSort{{Col: 2}}})303+ if err != nil {304+ t.Error(err)305+ return306+ }307+ if _, _, err := v.Page(context.Background(), 1500, 60); err != nil {308+ t.Error(err)309+ }310+ v.Close()311+ })312+ }313+ wg.Wait()314+ src.Close()315+ }316+}317+318+func TestSourceKind(t *testing.T) {319+ for spec, ok := range map[SourceSpec]bool{320+ {Path: "a.parquet"}: true, {Path: "a.sqlite"}: true, {Path: "a.db"}: true,321+ {Path: "a.csv"}: false, {Path: "a.bin", Format: "Parquet"}: true, {Path: "a.parquet", Format: "CSV"}: false,322+ } {323+ if _, err := SourceKind(spec); (err == nil) != ok {324+ t.Errorf("%+v: %v", spec, err)325+ }326+ }327+ if _, err := OpenSource(context.Background(), SourceSpec{Path: filepath.Join(t.TempDir(), "none.parquet")}); err == nil {328+ t.Error("opened a file that isn't there")329+ }330+}@@ -76,6 +76,9 @@ func (c fileCell) typed() sheet.LiveCell {
7676 return c.LiveCell7777 }7878 s := flatten(c.V.Str)79+ if s == "" {80+ return sheet.LiveCell{} // blank, as an import leaves it81+ }7982 if n, f, ok := sheet.ParseValue(s); ok {8083 switch f.Kind {8184 case sheet.FmtDate, sheet.FmtTime, sheet.FmtDateTime:@@ -0,0 +1,288 @@
1+package fileio2+3+import (4+ "context"5+ "errors"6+ "io"7+ "os"8+ "slices"9+ "strings"10+ "sync"11+12+ "github.com/parquet-go/parquet-go"13+14+ "github.com/FelineStateMachine/012/internal/sheet"15+)16+17+// parquetSource is a Parquet file read in place. Each read opens its18+// own readers of the column chunks it needs (the file is read with19+// ReadAt), so reads on several goroutines don't meet.20+type parquetSource struct {21+ f *os.File22+ pf *parquet.File23+ groups []parquet.RowGroup24+ starts []int64 // each row group's first row, and the rows in all last25+ cols []parquetColumn26+ meta []SourceColumn27+ dir string28+29+ mu sync.Mutex30+ views []*rowFile // what views wrote, removed by Close31+}32+33+func openParquetSource(spec SourceSpec) (*parquetSource, error) {34+ f, err := os.Open(spec.Path)35+ if err != nil {36+ return nil, err37+ }38+ st, err := f.Stat()39+ if err != nil {40+ f.Close()41+ return nil, err42+ }43+ pf, err := parquet.OpenFile(f, st.Size())44+ if err != nil {45+ f.Close()46+ return nil, err47+ }48+ s := &parquetSource{f: f, pf: pf, groups: pf.RowGroups(), dir: spec.TempDir}49+ s.starts = make([]int64, len(s.groups)+1)50+ for i, g := range s.groups {51+ s.starts[i+1] = s.starts[i] + g.NumRows()52+ }53+ schema := pf.Schema()54+ for _, path := range schema.Columns() {55+ var c parquetColumn56+ if leaf, ok := schema.Lookup(path...); ok && leaf.Node != nil {57+ t := leaf.Node.Type()58+ c.kind = t.Kind()59+ c.repeated = leaf.MaxRepetitionLevel > 060+ if lt := t.LogicalType(); lt != nil {61+ c.logical = lt.Value62+ }63+ }64+ f, num := parquetFormat(c)65+ s.cols = append(s.cols, c)66+ s.meta = append(s.meta, SourceColumn{Name: strings.Join(path, "."), Format: f, Numeric: num})67+ }68+ return s, nil69+}70+71+func (s *parquetSource) Columns() []SourceColumn { return slices.Clone(s.meta) }72+73+func (s *parquetSource) Rows() int64 { return s.starts[len(s.starts)-1] }74+75+func (s *parquetSource) Close() error {76+ s.mu.Lock()77+ views := s.views78+ s.views = nil79+ s.mu.Unlock()80+ var errs []error81+ for _, v := range views {82+ errs = append(errs, v.close())83+ }84+ return errors.Join(append(errs, s.f.Close())...)85+}86+87+// group is the row group holding row.88+func (s *parquetSource) group(row int64) int {89+ i, found := slices.BinarySearch(s.starts, row)90+ if !found {91+ i--92+ }93+ return i94+}95+96+// Scan streams rows from row from on.97+func (s *parquetSource) Scan(ctx context.Context, from int64, cols []int, fn func(int64, []sheet.LiveCell) bool) error {98+ cols = allCols(cols, len(s.cols))99+ if err := checkCols(cols, len(s.cols)); err != nil {100+ return err101+ }102+ rs := s.newRows(cols)103+ defer rs.close()104+ vals := make([]sheet.LiveCell, len(cols))105+ for row := max(from, 0); row < s.Rows(); row++ {106+ if row%1024 == 0 {107+ if err := ctx.Err(); err != nil {108+ return err109+ }110+ }111+ if err := rs.read(row, vals); err != nil {112+ return err113+ }114+ if !fn(row, vals) {115+ return nil116+ }117+ }118+ return nil119+}120+121+// Fetch reads the rows numbered rows.122+func (s *parquetSource) Fetch(ctx context.Context, rows []int64, cols []int) ([][]sheet.LiveCell, error) {123+ cols = allCols(cols, len(s.cols))124+ if err := checkCols(cols, len(s.cols)); err != nil {125+ return nil, err126+ }127+ rs := s.newRows(cols)128+ defer rs.close()129+ out := make([][]sheet.LiveCell, len(rows))130+ for _, i := range fetchOrder(rows) {131+ if err := ctx.Err(); err != nil {132+ return nil, err133+ }134+ row := rows[i]135+ if row < 0 || row >= s.Rows() {136+ return nil, errors.New("no such row in the source")137+ }138+ vals := make([]sheet.LiveCell, len(cols))139+ if err := rs.read(row, vals); err != nil {140+ return nil, err141+ }142+ out[i] = vals143+ }144+ return out, nil145+}146+147+// pqRows reads rows of some columns, each through a cursor of its own.148+type pqRows struct {149+ s *parquetSource150+ curs []*pqCursor151+ buf []parquet.Value152+}153+154+func (s *parquetSource) newRows(cols []int) *pqRows {155+ rs := &pqRows{s: s}156+ for _, c := range cols {157+ rs.curs = append(rs.curs, &pqCursor{s: s, col: c, repeated: s.cols[c].repeated, group: -1})158+ }159+ return rs160+}161+162+// read reads row's values into vals.163+func (rs *pqRows) read(row int64, vals []sheet.LiveCell) error {164+ for i, c := range rs.curs {165+ var err error166+ if rs.buf, err = c.row(row, rs.buf[:0]); err != nil {167+ return err168+ }169+ cell, _ := parquetCells(rs.buf, rs.s.cols[c.col])170+ vals[i] = cell.typed()171+ }172+ return nil173+}174+175+func (rs *pqRows) close() {176+ for _, c := range rs.curs {177+ c.close()178+ }179+}180+181+// pqCursor reads one column's values a row at a time: from the page182+// readers of its chunk in the row group being read, sought to a row183+// when a read isn't the next one.184+type pqCursor struct {185+ s *parquetSource186+ col int187+ repeated bool188+ group int // the row group being read, -1 before any189+ next int64 // the row the next value starts190+ pages parquet.Pages191+ vals parquet.ValueReader192+ buf []parquet.Value193+ i, n int194+}195+196+// skipRows is how far ahead a cursor reads its way rather than seeking.197+const skipRows = 4096198+199+// row appends the values of row to out: one, or a repeated column's200+// several (none for an empty list, read as null).201+func (c *pqCursor) row(row int64, out []parquet.Value) ([]parquet.Value, error) {202+ if err := c.seek(row); err != nil {203+ return out, err204+ }205+ for first := true; ; first = false {206+ if c.i == c.n {207+ if err := c.fill(); err != nil {208+ if errors.Is(err, io.EOF) && !first {209+ break210+ }211+ return out, err212+ }213+ }214+ v := c.buf[c.i]215+ if !first && v.RepetitionLevel() == 0 {216+ break217+ }218+ if c.repeated {219+ v = v.Clone() // the next fill may reuse what it points at220+ }221+ out = append(out, v)222+ c.i++223+ }224+ c.next = row + 1225+ return out, nil226+}227+228+// seek places the cursor at row: reading forward to it when it is a229+// little ahead in the same group, else seeking the group's pages.230+func (c *pqCursor) seek(row int64) error {231+ if row == c.next && c.group >= 0 && row < c.s.starts[c.group+1] {232+ return nil // the next row, as a scan reads them233+ }234+ g := c.s.group(row)235+ if g == c.group && row >= c.next && row-c.next <= skipRows {236+ var skip []parquet.Value237+ for c.next < row {238+ var err error239+ if skip, err = c.row(c.next, skip[:0]); err != nil {240+ return err241+ }242+ }243+ return nil244+ }245+ c.close()246+ pages := c.s.groups[g].ColumnChunks()[c.col].Pages()247+ if err := pages.SeekToRow(row - c.s.starts[g]); err != nil {248+ pages.Close()249+ return err250+ }251+ c.pages, c.group, c.next, c.vals, c.i, c.n = pages, g, row, nil, 0, 0252+ return nil253+}254+255+// fill reads the next values of the group's pages, io.EOF past them.256+func (c *pqCursor) fill() error {257+ if c.buf == nil {258+ c.buf = make([]parquet.Value, 1024)259+ }260+ for {261+ if c.vals != nil {262+ n, err := c.vals.ReadValues(c.buf)263+ if n > 0 {264+ c.i, c.n = 0, n265+ return nil266+ }267+ if err != nil && !errors.Is(err, io.EOF) {268+ return err269+ }270+ c.vals = nil271+ }272+ if c.pages == nil {273+ return io.EOF274+ }275+ p, err := c.pages.ReadPage()276+ if err != nil {277+ return err278+ }279+ c.vals = p.Values()280+ }281+}282+283+func (c *pqCursor) close() {284+ if c.pages != nil {285+ c.pages.Close()286+ }287+ c.pages, c.vals, c.group, c.i, c.n = nil, nil, -1, 0, 0288+}@@ -0,0 +1,191 @@
1+package fileio2+3+import (4+ "bufio"5+ "context"6+ "errors"7+ "os"8+ "slices"9+10+ "github.com/FelineStateMachine/012/internal/sheet"11+)12+13+// Views of a Parquet file stream: building one reads the columns it14+// sorts and filters by once, writing the rows it keeps, in its order,15+// to a file of row numbers (sourcesort.go); a page reads its row16+// numbers there and fetches those rows.17+18+// View orders the file's rows as o says.19+func (s *parquetSource) View(ctx context.Context, o SourceOrder) (SourceView, error) {20+ if o.IsZero() {21+ return pqPlainView{s}, nil22+ }23+ plan, err := newViewPlan(o, len(s.cols))24+ if err != nil {25+ return nil, err26+ }27+ f, err := newRowFile(s.dir)28+ if err != nil {29+ return nil, err30+ }31+ rf := &rowFile{f: f}32+ if rf.n, err = plan.write(ctx, s, f); err != nil {33+ rf.close()34+ return nil, err35+ }36+ s.mu.Lock()37+ s.views = append(s.views, rf)38+ s.mu.Unlock()39+ return &pqView{s: s, rf: rf}, nil40+}41+42+// viewPlan is what building a streamed view reads: the columns it sorts43+// and filters by, once each, and where each key and test finds its44+// column among them.45+type viewPlan struct {46+ cols []int47+ keys []int // positions in cols of the sort's columns48+ desc []bool49+ tests []viewTest50+}51+52+type viewTest struct {53+ at int // the column's position in cols54+ pass func(sheet.Value, string) bool55+}56+57+func newViewPlan(o SourceOrder, ncols int) (*viewPlan, error) {58+ p := &viewPlan{}59+ at := func(col int) (int, error) {60+ if col < 0 || col >= ncols {61+ return 0, errors.New("the source has no such column")62+ }63+ if i := slices.Index(p.cols, col); i >= 0 {64+ return i, nil65+ }66+ p.cols = append(p.cols, col)67+ return len(p.cols) - 1, nil68+ }69+ for _, k := range o.Sort {70+ i, err := at(k.Col)71+ if err != nil {72+ return nil, err73+ }74+ p.keys, p.desc = append(p.keys, i), append(p.desc, k.Desc)75+ }76+ for _, f := range o.Filter {77+ i, err := at(f.Col)78+ if err != nil {79+ return nil, err80+ }81+ p.tests = append(p.tests, viewTest{at: i, pass: f.Cond.Test()})82+ }83+ return p, nil84+}85+86+// passes reports whether a row's values pass every test, each on the87+// value and the text it shows, as a sheet's filter tests them.88+func (p *viewPlan) passes(vals []sheet.LiveCell) bool {89+ for _, t := range p.tests {90+ c := vals[t.at]91+ if !t.pass(c.V, sheet.FormatText(c.V, c.F)) {92+ return false93+ }94+ }95+ return true96+}97+98+// write streams the source, writing the numbers of the rows the view99+// keeps to f in its order, and returns how many.100+func (p *viewPlan) write(ctx context.Context, s *parquetSource, f *os.File) (int64, error) {101+ w := bufio.NewWriterSize(f, 1<<20)102+ var n int64103+ var sorter *extSorter104+ if len(p.keys) > 0 {105+ sorter = newExtSorter(s.dir, p.desc)106+ defer sorter.remove()107+ }108+ keys := make([]sheet.LiveCell, len(p.keys))109+ var ferr error110+ err := s.Scan(ctx, 0, p.cols, func(row int64, vals []sheet.LiveCell) bool {111+ if !p.passes(vals) {112+ return true113+ }114+ if sorter == nil {115+ ferr = writeRow(w, row)116+ n++117+ return ferr == nil118+ }119+ for i, k := range p.keys {120+ keys[i] = vals[k]121+ }122+ ferr = sorter.add(row, keys)123+ return ferr == nil124+ })125+ if err = errors.Join(err, ferr); err != nil {126+ return 0, err127+ }128+ if sorter != nil {129+ if n, err = sorter.finish(w); err != nil {130+ return 0, err131+ }132+ }133+ return n, w.Flush()134+}135+136+// pqView is a sorted or filtered view of a Parquet file.137+type pqView struct {138+ s *parquetSource139+ rf *rowFile140+}141+142+func (v *pqView) Rows() int64 { return v.rf.n }143+144+func (v *pqView) Page(ctx context.Context, from int64, n int) ([]int64, [][]sheet.LiveCell, error) {145+ nums, err := v.rf.read(from, n)146+ if err != nil || len(nums) == 0 {147+ return nil, nil, err148+ }149+ rows, err := v.s.Fetch(ctx, nums, nil)150+ if err != nil {151+ return nil, nil, err152+ }153+ return nums, rows, nil154+}155+156+func (v *pqView) Close() error {157+ v.s.mu.Lock()158+ i := slices.Index(v.s.views, v.rf)159+ if i < 0 {160+ v.s.mu.Unlock()161+ return nil // closed with the source162+ }163+ v.s.views = slices.Delete(v.s.views, i, i+1)164+ v.s.mu.Unlock()165+ return v.rf.close()166+}167+168+// pqPlainView is the file's rows in its own order.169+type pqPlainView struct{ s *parquetSource }170+171+func (v pqPlainView) Rows() int64 { return v.s.Rows() }172+173+func (v pqPlainView) Page(ctx context.Context, from int64, n int) ([]int64, [][]sheet.LiveCell, error) {174+ return scanPage(ctx, v.s, from, n)175+}176+177+func (pqPlainView) Close() error { return nil }178+179+// scanPage reads n rows from row from of a source in its own order.180+func scanPage(ctx context.Context, s Source, from int64, n int) ([]int64, [][]sheet.LiveCell, error) {181+ if from < 0 || from >= s.Rows() || n <= 0 {182+ return nil, nil, nil183+ }184+ var nums []int64185+ var rows [][]sheet.LiveCell186+ err := s.Scan(ctx, from, nil, func(row int64, vals []sheet.LiveCell) bool {187+ nums, rows = append(nums, row), append(rows, slices.Clone(vals))188+ return len(nums) < n189+ })190+ return nums, rows, err191+}@@ -0,0 +1,364 @@
1+package fileio2+3+import (4+ "bufio"5+ "cmp"6+ "container/heap"7+ "encoding/binary"8+ "errors"9+ "io"10+ "math"11+ "os"12+ "slices"13+ "strings"14+15+ "github.com/FelineStateMachine/012/internal/sheet"16+)17+18+// A Parquet view's rows are a file of row numbers in the view's order,19+// 8 bytes each, which a page reads with ReadAt: a filter writes the rows20+// it lets through as it streams them, and a sort is an external merge21+// sort, runs of at most sortRunBytes sorted in memory and spilled to22+// files, then merged. Memory stays bounded whatever the row count; the23+// disk holds the runs and the order, about 8 bytes a row plus the keys.24+25+// sortRunBytes bounds the keys a sort holds in memory at once.26+var sortRunBytes = 64 << 2027+28+// sortVal is one key of a row as a sort compares it: its kind's rank29+// (numbers, text, booleans, errors, then blanks), and its number or its30+// text in lower case.31+type sortVal struct {32+ rank uint833+ num float6434+ str string35+}36+37+// blankRank is a blank's, last in either direction.38+const blankRank = 439+40+func sortValOf(v sheet.Value) sortVal {41+ switch v.Kind {42+ case sheet.Number:43+ return sortVal{rank: 0, num: v.Num}44+ case sheet.Text:45+ return sortVal{rank: 1, str: strings.ToLower(v.Str)}46+ case sheet.Bool:47+ return sortVal{rank: 2, num: v.Num}48+ case sheet.Error:49+ return sortVal{rank: 3, str: strings.ToLower(v.Str)}50+ }51+ return sortVal{rank: blankRank}52+}53+54+// sortRec is a row being sorted: its number and its keys.55+type sortRec struct {56+ row int6457+ keys []sortVal58+}59+60+// size is about what the record holds in memory.61+func (r *sortRec) size() int {62+ n := 48 + 32*len(r.keys)63+ for _, k := range r.keys {64+ n += len(k.str)65+ }66+ return n67+}68+69+// compareRecs orders rows as a sheet's sort does (sheet.compareRows):70+// key by key, blanks last in either direction, Z to A reversing the71+// rest, and ties in the source's order.72+func compareRecs(a, b *sortRec, desc []bool) int {73+ for i := range a.keys {74+ x, y := &a.keys[i], &b.keys[i]75+ if (x.rank == blankRank) != (y.rank == blankRank) {76+ if x.rank == blankRank {77+ return 178+ }79+ return -180+ }81+ d := cmp.Compare(x.rank, y.rank)82+ if d == 0 {83+ switch x.rank {84+ case 0, 2:85+ d = cmp.Compare(x.num, y.num)86+ case 1, 3:87+ d = strings.Compare(x.str, y.str)88+ }89+ }90+ if desc[i] {91+ d = -d92+ }93+ if d != 0 {94+ return d95+ }96+ }97+ return cmp.Compare(a.row, b.row)98+}99+100+// extSorter sorts rows by their keys in bounded memory.101+type extSorter struct {102+ dir string103+ desc []bool104+ recs []sortRec105+ bytes int106+ runs []string107+ limit int108+}109+110+func newExtSorter(dir string, desc []bool) *extSorter {111+ return &extSorter{dir: dir, desc: desc, limit: sortRunBytes}112+}113+114+// add takes a row's keys, spilling a run when memory is full.115+func (s *extSorter) add(row int64, keys []sheet.LiveCell) error {116+ r := sortRec{row: row, keys: make([]sortVal, len(keys))}117+ for i, k := range keys {118+ r.keys[i] = sortValOf(k.V)119+ }120+ s.recs = append(s.recs, r)121+ s.bytes += r.size()122+ if s.bytes >= s.limit {123+ return s.spill()124+ }125+ return nil126+}127+128+func (s *extSorter) sort() {129+ slices.SortFunc(s.recs, func(a, b sortRec) int { return compareRecs(&a, &b, s.desc) })130+}131+132+// spill sorts the rows in memory and writes them as a run.133+func (s *extSorter) spill() error {134+ s.sort()135+ f, err := os.CreateTemp(s.dir, "012-sort-*.run")136+ if err != nil {137+ return err138+ }139+ s.runs = append(s.runs, f.Name())140+ w := bufio.NewWriterSize(f, 1<<20)141+ for i := range s.recs {142+ writeRec(w, &s.recs[i])143+ }144+ err = errors.Join(w.Flush(), f.Close())145+ clear(s.recs)146+ s.recs, s.bytes = s.recs[:0], 0147+ return err148+}149+150+// finish writes the sorted rows' numbers to out and removes the runs.151+func (s *extSorter) finish(out io.Writer) (n int64, err error) {152+ defer s.remove()153+ if len(s.runs) == 0 {154+ s.sort()155+ for i := range s.recs {156+ if err := writeRow(out, s.recs[i].row); err != nil {157+ return n, err158+ }159+ n++160+ }161+ return n, nil162+ }163+ if len(s.recs) > 0 {164+ if err := s.spill(); err != nil {165+ return 0, err166+ }167+ }168+ return s.merge(out)169+}170+171+// remove deletes the runs.172+func (s *extSorter) remove() {173+ for _, name := range s.runs {174+ os.Remove(name)175+ }176+ s.runs = nil177+}178+179+// merge merges the runs into out.180+func (s *extSorter) merge(out io.Writer) (int64, error) {181+ h := &runHeap{desc: s.desc}182+ for _, name := range s.runs {183+ f, err := os.Open(name)184+ if err != nil {185+ h.close()186+ return 0, err187+ }188+ r := &runReader{f: f, r: bufio.NewReaderSize(f, 256<<10), keys: len(s.desc)}189+ if r.next() {190+ h.runs = append(h.runs, r)191+ } else {192+ f.Close()193+ }194+ }195+ defer h.close()196+ heap.Init(h)197+ var n int64198+ for h.Len() > 0 {199+ r := h.runs[0]200+ if err := writeRow(out, r.rec.row); err != nil {201+ return n, err202+ }203+ n++204+ if r.next() {205+ heap.Fix(h, 0)206+ } else {207+ r.f.Close()208+ heap.Pop(h)209+ }210+ }211+ return n, h.err()212+}213+214+func writeRow(w io.Writer, row int64) error {215+ var b [8]byte216+ binary.LittleEndian.PutUint64(b[:], uint64(row))217+ _, err := w.Write(b[:])218+ return err219+}220+221+// writeRec writes a record to a run: its row, then each key's rank and222+// its number or text.223+func writeRec(w *bufio.Writer, r *sortRec) {224+ var b [binary.MaxVarintLen64]byte225+ w.Write(b[:binary.PutUvarint(b[:], uint64(r.row))])226+ for _, k := range r.keys {227+ w.WriteByte(k.rank)228+ switch k.rank {229+ case 0, 2:230+ binary.LittleEndian.PutUint64(b[:8], math.Float64bits(k.num))231+ w.Write(b[:8])232+ case 1, 3:233+ w.Write(b[:binary.PutUvarint(b[:], uint64(len(k.str)))])234+ w.WriteString(k.str)235+ }236+ }237+}238+239+// runReader reads a run's records in order.240+type runReader struct {241+ f *os.File242+ r *bufio.Reader243+ keys int244+ rec sortRec245+ fail error246+}247+248+// next reads the next record, false at the end or on an error.249+func (r *runReader) next() bool {250+ row, err := binary.ReadUvarint(r.r)251+ if err != nil {252+ if !errors.Is(err, io.EOF) {253+ r.fail = err254+ }255+ return false256+ }257+ r.rec.row = int64(row)258+ r.rec.keys = r.rec.keys[:0]259+ for range r.keys {260+ k, err := r.key()261+ if err != nil {262+ r.fail = err263+ return false264+ }265+ r.rec.keys = append(r.rec.keys, k)266+ }267+ return true268+}269+270+func (r *runReader) key() (sortVal, error) {271+ rank, err := r.r.ReadByte()272+ if err != nil {273+ return sortVal{}, err274+ }275+ k := sortVal{rank: rank}276+ switch rank {277+ case 0, 2:278+ var b [8]byte279+ if _, err := io.ReadFull(r.r, b[:]); err != nil {280+ return k, err281+ }282+ k.num = math.Float64frombits(binary.LittleEndian.Uint64(b[:]))283+ case 1, 3:284+ n, err := binary.ReadUvarint(r.r)285+ if err != nil {286+ return k, err287+ }288+ b := make([]byte, n)289+ if _, err := io.ReadFull(r.r, b); err != nil {290+ return k, err291+ }292+ k.str = string(b)293+ }294+ return k, nil295+}296+297+// runHeap merges runs by their next record.298+type runHeap struct {299+ runs []*runReader300+ desc []bool301+ done []*runReader302+}303+304+func (h *runHeap) Len() int { return len(h.runs) }305+func (h *runHeap) Less(i, j int) bool {306+ return compareRecs(&h.runs[i].rec, &h.runs[j].rec, h.desc) < 0307+}308+func (h *runHeap) Swap(i, j int) { h.runs[i], h.runs[j] = h.runs[j], h.runs[i] }309+func (h *runHeap) Push(x any) { h.runs = append(h.runs, x.(*runReader)) }310+func (h *runHeap) Pop() any {311+ r := h.runs[len(h.runs)-1]312+ h.runs = h.runs[:len(h.runs)-1]313+ h.done = append(h.done, r)314+ return r315+}316+317+// err is the first error a run met.318+func (h *runHeap) err() error {319+ for _, r := range append(h.done, h.runs...) {320+ if r.fail != nil {321+ return r.fail322+ }323+ }324+ return nil325+}326+327+func (h *runHeap) close() {328+ for _, r := range h.runs {329+ r.f.Close()330+ }331+}332+333+// rowFile is a view's rows as a file of row numbers in its order.334+type rowFile struct {335+ f *os.File336+ n int64337+}338+339+// newRowFile makes the file rows are written to, in dir.340+func newRowFile(dir string) (*os.File, error) { return os.CreateTemp(dir, "012-view-*.rows") }341+342+// read returns the row numbers at positions from to from+n, as many as343+// there are.344+func (rf *rowFile) read(from int64, n int) ([]int64, error) {345+ if from < 0 || from >= rf.n || n <= 0 {346+ return nil, nil347+ }348+ n = int(min(int64(n), rf.n-from))349+ b := make([]byte, 8*n)350+ if _, err := rf.f.ReadAt(b, 8*from); err != nil {351+ return nil, err352+ }353+ out := make([]int64, n)354+ for i := range out {355+ out[i] = int64(binary.LittleEndian.Uint64(b[8*i:]))356+ }357+ return out, nil358+}359+360+// close closes and removes the file.361+func (rf *rowFile) close() error {362+ name := rf.f.Name()363+ return errors.Join(rf.f.Close(), os.Remove(name))364+}@@ -0,0 +1,209 @@
1+package fileio2+3+import (4+ "context"5+ "database/sql"6+ "fmt"7+ "os"8+ "path/filepath"9+ "slices"10+ "testing"11+12+ "github.com/FelineStateMachine/012/internal/sheet"13+)14+15+// writeSalesDB writes a database whose table sales has n rows (fewer,16+// with rowids missing, when sparse), a view of it, a WITHOUT ROWID17+// table, and a column of mixed kinds with no declared type.18+func writeSalesDB(t testing.TB, dir string, n int, sparse bool) string {19+ t.Helper()20+ name := filepath.Join(dir, fmt.Sprintf("sales-%d-%v.sqlite", n, sparse))21+ db, err := sql.Open("sqlite", name)22+ if err != nil {23+ t.Fatal(err)24+ }25+ defer db.Close()26+ tx, err := db.Begin()27+ if err != nil {28+ t.Fatal(err)29+ }30+ for _, q := range []string{31+ "CREATE TABLE sales (id INTEGER, cat TEXT, amount REAL, day TEXT, mixed)",32+ "CREATE VIEW big AS SELECT * FROM sales WHERE amount > 50",33+ "CREATE TABLE keyed (k TEXT PRIMARY KEY, v REAL) WITHOUT ROWID",34+ "CREATE TABLE other (x)",35+ } {36+ if _, err := tx.Exec(q); err != nil {37+ t.Fatal(err)38+ }39+ }40+ ins, err := tx.Prepare("INSERT INTO sales VALUES (?, ?, ?, ?, ?)")41+ if err != nil {42+ t.Fatal(err)43+ }44+ for i := range n {45+ var cat, amount, mixed any46+ if i%9 != 4 {47+ cat = saleCats[i%len(saleCats)]48+ }49+ if i%11 != 7 {50+ amount = float64((i*37)%101) - 20.551+ }52+ switch i % 6 {53+ case 0:54+ mixed = int64(i % 13)55+ case 1:56+ mixed = fmt.Sprintf("t%d", i%7)57+ case 2:58+ mixed = 2.559+ case 3:60+ mixed = "12"61+ case 4:62+ mixed = ""63+ }64+ day := fmt.Sprintf("2026-01-%02d", 1+i%28)65+ if _, err := ins.Exec(i, cat, amount, day, mixed); err != nil {66+ t.Fatal(err)67+ }68+ if _, err := tx.Exec("INSERT INTO keyed VALUES (?, ?)", fmt.Sprintf("k%05d", n-i), i); err != nil {69+ t.Fatal(err)70+ }71+ }72+ if sparse {73+ if _, err := tx.Exec("DELETE FROM sales WHERE id % 7 = 3"); err != nil {74+ t.Fatal(err)75+ }76+ }77+ if err := tx.Commit(); err != nil {78+ t.Fatal(err)79+ }80+ return name81+}82+83+// sqlOrders are orders only a SQLite source's table has columns for.84+var sqlOrders = []SourceOrder{85+ {Sort: []SourceSort{{Col: 4}}},86+ {Sort: []SourceSort{{Col: 4, Desc: true}}},87+ {Filter: []SourceFilter{{Col: 4, Cond: sheet.Condition{Op: sheet.CondGreater, Arg: "2"}}}},88+ {Filter: []SourceFilter{{Col: 4, Cond: sheet.Condition{Op: sheet.CondLess, Arg: "T3"}}}},89+ {Filter: []SourceFilter{{Col: 4, Cond: sheet.Condition{Op: sheet.CondEqual, Arg: "12"}}}},90+}91+92+func TestSQLiteSource(t *testing.T) {93+ dir := t.TempDir()94+ for _, sparse := range []bool{false, true} {95+ t.Run(fmt.Sprint("sparse=", sparse), func(t *testing.T) { checkSQLiteTable(t, writeSalesDB(t, dir, 700, sparse)) })96+ }97+}98+99+// checkSQLiteTable reads the table sales of name every way a source100+// reads, as the sheet its import makes shows it.101+func checkSQLiteTable(t *testing.T, name string) {102+ tmp := t.TempDir()103+ src, err := OpenSource(context.Background(), SourceSpec{Path: name, Table: "sales", TempDir: tmp})104+ if err != nil {105+ t.Fatal(err)106+ }107+ s, r := imported(t, name, Options{Table: "sales"})108+ want := sheetRows(s, r)109+ if src.Rows() != int64(len(want)) {110+ t.Fatalf("%d rows, want %d", src.Rows(), len(want))111+ }112+ var cols []string113+ for _, c := range src.Columns() {114+ cols = append(cols, fmt.Sprintf("%s:%v", c.Name, c.Numeric))115+ }116+ if !slices.Equal(cols, []string{"id:true", "cat:false", "amount:true", "day:false", "mixed:false"}) {117+ t.Errorf("columns %q", cols)118+ }119+ var got []string120+ err = src.Scan(context.Background(), 300, nil, func(row int64, v []sheet.LiveCell) bool {121+ got = append(got, cellsText(v))122+ return true123+ })124+ if err != nil || !slices.Equal(got, want[300:]) {125+ t.Errorf("scan from 300: %v %q", err, head(got))126+ }127+ rows, err := src.Fetch(context.Background(), []int64{int64(len(want) - 1), 0, 350}, nil)128+ if err != nil {129+ t.Fatal(err)130+ }131+ for i, n := range []int{len(want) - 1, 0, 350} {132+ if g := cellsText(rows[i]); g != want[n] {133+ t.Errorf("row %d is %q, want %q", n, g, want[n])134+ }135+ }136+ checkOrders(t, src, name, Options{Table: "sales"}, append(slices.Clone(orders), sqlOrders...))137+ if err := src.Close(); err != nil {138+ t.Fatal(err)139+ }140+ if left, _ := os.ReadDir(tmp); len(left) != 0 {141+ t.Errorf("closing left %d files", len(left))142+ }143+}144+145+// TestSQLiteSourceKinds reads a view, a WITHOUT ROWID table and a146+// query, which are paged through a copy in the work database.147+func TestSQLiteSourceKinds(t *testing.T) {148+ dir := t.TempDir()149+ name := writeSalesDB(t, dir, 300, false)150+ for _, spec := range []SourceSpec{151+ {Path: name, Table: "big"},152+ {Path: name, Table: "keyed"},153+ {Path: name, Query: "SELECT cat, sum(amount) AS total, count(*) AS n FROM sales GROUP BY cat"},154+ {Path: name, Query: "SELECT id, id FROM sales WHERE id < 40"},155+ } {156+ spec.TempDir = t.TempDir()157+ src, err := OpenSource(context.Background(), spec)158+ if err != nil {159+ t.Fatalf("%+v: %v", spec, err)160+ }161+ s, r := imported(t, name, Options{Table: spec.Table, Query: spec.Query})162+ want := sheetRows(s, r)163+ v, err := src.View(context.Background(), SourceOrder{})164+ if err != nil {165+ t.Fatal(err)166+ }167+ if _, got := viewRows(t, v, 50); !slices.Equal(got, want) {168+ t.Errorf("%+v: %q, want %q", spec, head(got), head(want))169+ }170+ sorted := []SourceOrder{{Sort: []SourceSort{{Col: 1, Desc: true}}}}171+ checkOrders(t, src, name, Options{Table: spec.Table, Query: spec.Query}, sorted)172+ src.Close()173+ if left, _ := os.ReadDir(spec.TempDir); len(left) != 0 {174+ t.Errorf("%+v: closing left %d files", spec, len(left))175+ }176+ }177+ if _, err := OpenSource(context.Background(), SourceSpec{Path: name}); err == nil {178+ t.Error("a database of several tables opened without naming one")179+ }180+ if _, err := OpenSource(context.Background(), SourceSpec{Path: name, Query: "SELECT nope FROM sales"}); err == nil {181+ t.Error("a query that fails opened")182+ }183+}184+185+// TestSQLiteSourceReadOnly checks the database file is never written.186+func TestSQLiteSourceReadOnly(t *testing.T) {187+ name := writeSalesDB(t, t.TempDir(), 200, true)188+ before, err := os.ReadFile(name)189+ if err != nil {190+ t.Fatal(err)191+ }192+ src, err := OpenSource(context.Background(), SourceSpec{Path: name, Table: "sales", TempDir: t.TempDir()})193+ if err != nil {194+ t.Fatal(err)195+ }196+ v, err := src.View(context.Background(), SourceOrder{Sort: []SourceSort{{Col: 2}}})197+ if err != nil {198+ t.Fatal(err)199+ }200+ v.Page(context.Background(), 100, 20)201+ src.Close()202+ after, _ := os.ReadFile(name)203+ if !slices.Equal(before, after) {204+ t.Error("the database changed")205+ }206+ if _, err := os.Stat(name + "-journal"); err == nil {207+ t.Error("the database has a journal")208+ }209+}@@ -0,0 +1,329 @@
1+package fileio2+3+import (4+ "context"5+ "database/sql"6+ "errors"7+ "fmt"8+ "net/url"9+ "os"10+ "path/filepath"11+ "slices"12+ "strconv"13+ "strings"14+ "sync"15+16+ "github.com/FelineStateMachine/012/internal/sheet"17+)18+19+// sqliteSource is a SQLite table, view or query read in place, through20+// connections that can only read (mode=ro, query_only), so the database21+// is never written. Its rows are in rowid order for a table, and as the22+// query or view gives them otherwise.23+//24+// A row's number is found cheaply only in a table whose rowids are 1 to25+// its row count (dense), the common case of a table only ever appended26+// to: row n is rowid n+1. Any other source is copied, when something27+// first asks for rows by number, into a table of a work database in28+// the temporary directory (sourcesqlwork.go), where it is dense; views'29+// row orders are tables there too.30+type sqliteSource struct {31+ spec SourceSpec32+ db *sql.DB33+ query string // the SELECT giving the source's rows, in order34+ table string // the table, quoted, when the source is one with rowids35+ names []string36+ meta []SourceColumn37+ rows int6438+ dense bool39+40+ mu sync.Mutex41+ work *sqlWork // made when first needed42+ base *sqlBase // where rows are found by number, once known43+}44+45+// sqliteURI is the URI opening a database file to read only.46+func sqliteURI(path string, params ...string) (string, error) {47+ abs, err := filepath.Abs(path)48+ if err != nil {49+ return "", err50+ }51+ u := url.URL{Scheme: "file", Path: filepath.ToSlash(abs), RawQuery: strings.Join(params, "&")}52+ return u.String(), nil53+}54+55+func openSQLiteSource(ctx context.Context, spec SourceSpec) (*sqliteSource, error) {56+ if _, err := os.Stat(spec.Path); err != nil {57+ return nil, err58+ }59+ uri, err := sqliteURI(spec.Path, "mode=ro", "_pragma=query_only(1)", "_pragma=busy_timeout(10000)")60+ if err != nil {61+ return nil, err62+ }63+ db, err := sql.Open("sqlite", uri)64+ if err != nil {65+ return nil, err66+ }67+ s := &sqliteSource{spec: spec, db: db}68+ if err := s.open(ctx); err != nil {69+ db.Close()70+ return nil, err71+ }72+ return s, nil73+}74+75+// open finds what the source reads, its columns and its row count.76+func (s *sqliteSource) open(ctx context.Context) error {77+ s.query = s.spec.Query78+ if s.query == "" {79+ table := s.spec.Table80+ if table == "" {81+ ts, err := tables(ctx, s.db)82+ switch {83+ case err != nil:84+ return err85+ case len(ts) == 0:86+ return errors.New("the database has no tables")87+ case len(ts) > 1:88+ return &ErrNeedTable{Tables: ts}89+ }90+ table = ts[0].Name91+ }92+ s.query = "SELECT * FROM " + quoteIdent(table)93+ if s.hasRowid(ctx, table) {94+ s.table = quoteIdent(table)95+ s.query += " ORDER BY rowid"96+ }97+ }98+ if err := s.columns(ctx); err != nil {99+ return err100+ }101+ return s.count(ctx)102+}103+104+// hasRowid reports whether table is a table with rowids (not a view or105+// a WITHOUT ROWID table).106+func (s *sqliteSource) hasRowid(ctx context.Context, table string) bool {107+ rows, err := s.db.QueryContext(ctx, "SELECT rowid FROM "+quoteIdent(table)+" LIMIT 0")108+ if err != nil {109+ return false110+ }111+ rows.Close()112+ var typ string113+ err = s.db.QueryRowContext(ctx, "SELECT type FROM sqlite_master WHERE name = ?", table).Scan(&typ)114+ return err == nil && typ == "table"115+}116+117+// count counts the rows, and for a table whether its rowids are dense.118+func (s *sqliteSource) count(ctx context.Context) error {119+ if s.table == "" {120+ err := s.db.QueryRowContext(ctx, "SELECT count(*) FROM ("+s.query+")").Scan(&s.rows)121+ return sqliteErr(err)122+ }123+ var lo, hi sql.NullInt64124+ err := s.db.QueryRowContext(ctx, "SELECT count(*), min(rowid), max(rowid) FROM "+s.table).Scan(&s.rows, &lo, &hi)125+ if err != nil {126+ return sqliteErr(err)127+ }128+ s.dense = s.rows == 0 || lo.Int64 == 1 && hi.Int64 == s.rows129+ return nil130+}131+132+// columns reads the columns' names and types: a declared type says133+// whether one holds numbers or dates; an undeclared one (a query's134+// expressions) is judged by its first rows.135+func (s *sqliteSource) columns(ctx context.Context) error {136+ rows, err := s.db.QueryContext(ctx, "SELECT * FROM ("+s.query+") LIMIT 0")137+ if err != nil {138+ return sqliteErr(err)139+ }140+ types, err := rows.ColumnTypes()141+ rows.Close()142+ if err != nil {143+ return sqliteErr(err)144+ }145+ var untyped []int146+ for i, t := range types {147+ c := SourceColumn{Name: t.Name()}148+ decl := strings.ToUpper(t.DatabaseTypeName())149+ switch {150+ case decl == "":151+ untyped = append(untyped, i)152+ case strings.Contains(decl, "DATETIME"), strings.Contains(decl, "TIMESTAMP"):153+ c.Format, c.Numeric = sheet.Format{Kind: sheet.FmtDateTime}, true154+ case strings.Contains(decl, "DATE"):155+ c.Format, c.Numeric = sheet.Format{Kind: sheet.FmtDate}, true156+ default:157+ c.Numeric = numericAffinity(decl)158+ }159+ s.names = append(s.names, t.Name())160+ s.meta = append(s.meta, c)161+ }162+ if len(untyped) > 0 {163+ return s.sample(ctx, untyped)164+ }165+ return nil166+}167+168+// numericAffinity reports whether a declared type gives a column169+// SQLite's INTEGER, REAL or NUMERIC affinity.170+func numericAffinity(decl string) bool {171+ switch {172+ case strings.Contains(decl, "CHAR"), strings.Contains(decl, "CLOB"), strings.Contains(decl, "TEXT"),173+ strings.Contains(decl, "BLOB"):174+ return false175+ }176+ return true177+}178+179+// sampleRows is how many rows judge an undeclared column's type.180+const sampleRows = 100181+182+// sample marks the columns cols numeric when their first rows hold183+// numbers and nothing else.184+func (s *sqliteSource) sample(ctx context.Context, cols []int) error {185+ num := make([]bool, len(s.meta))186+ seen := make([]bool, len(s.meta))187+ for _, c := range cols {188+ num[c] = true189+ }190+ err := s.scanSQL(ctx, "SELECT * FROM ("+s.query+") LIMIT "+strconv.Itoa(sampleRows), nil, func(vals []any) bool {191+ for _, c := range cols {192+ switch vals[c].(type) {193+ case nil:194+ case int64, float64:195+ seen[c] = true196+ default:197+ num[c] = false198+ }199+ }200+ return true201+ })202+ for _, c := range cols {203+ s.meta[c].Numeric = num[c] && seen[c]204+ }205+ return err206+}207+208+func (s *sqliteSource) Columns() []SourceColumn { return slices.Clone(s.meta) }209+210+func (s *sqliteSource) Rows() int64 { return s.rows }211+212+// selectCols is the select list reading cols, and whether the columns213+// come back as asked: a table's by name, anything else whole, picked214+// from.215+func (s *sqliteSource) selectCols(cols []int) (string, bool) {216+ if s.table == "" {217+ return "*", false218+ }219+ names := make([]string, len(cols))220+ for i, c := range cols {221+ names[i] = quoteIdent(s.names[c])222+ }223+ return strings.Join(names, ", "), true224+}225+226+// Scan streams the rows from row from on.227+func (s *sqliteSource) Scan(ctx context.Context, from int64, cols []int, fn func(int64, []sheet.LiveCell) bool) error {228+ cols = allCols(cols, len(s.names))229+ if err := checkCols(cols, len(s.names)); err != nil {230+ return err231+ }232+ list, picked := s.selectCols(cols)233+ var q string234+ var args []any235+ switch {236+ case s.table != "" && s.dense:237+ q, args = "SELECT "+list+" FROM "+s.table+" WHERE rowid > ? ORDER BY rowid", []any{from}238+ case s.table != "":239+ q, args = "SELECT "+list+" FROM "+s.table+" ORDER BY rowid LIMIT -1 OFFSET ?", []any{from}240+ default:241+ q, args = "SELECT * FROM ("+s.query+") LIMIT -1 OFFSET ?", []any{from}242+ }243+ out := make([]sheet.LiveCell, len(cols))244+ row := max(from, 0)245+ return s.scanSQL(ctx, q, args, func(vals []any) bool {246+ for i, c := range cols {247+ v := vals[i]248+ if !picked {249+ v = vals[c]250+ }251+ cell, _ := sqliteCell(v)252+ out[i] = cell.typed()253+ }254+ row++255+ return fn(row-1, out)256+ })257+}258+259+// scanSQL runs a query on the source, calling fn with each row's values260+// until it returns false.261+func (s *sqliteSource) scanSQL(ctx context.Context, q string, args []any, fn func([]any) bool) error {262+ return queryRows(ctx, s.db, q, args, fn)263+}264+265+// queryRows runs a query, calling fn with each row's values (reused)266+// until it returns false.267+func queryRows(ctx context.Context, db interface {268+ QueryContext(context.Context, string, ...any) (*sql.Rows, error)269+}, q string, args []any, fn func([]any) bool) error {270+ rows, err := db.QueryContext(ctx, q, args...)271+ if err != nil {272+ return sqliteErr(err)273+ }274+ defer rows.Close()275+ cols, err := rows.Columns()276+ if err != nil {277+ return sqliteErr(err)278+ }279+ vals := make([]any, len(cols))280+ ptrs := make([]any, len(cols))281+ for i := range vals {282+ ptrs[i] = &vals[i]283+ }284+ for rows.Next() {285+ if err := rows.Scan(ptrs...); err != nil {286+ return sqliteErr(err)287+ }288+ if !fn(vals) {289+ return nil290+ }291+ }292+ if err := rows.Err(); err != nil {293+ if ctx.Err() != nil {294+ return ctx.Err()295+ }296+ return sqliteErr(err)297+ }298+ return nil299+}300+301+// Fetch reads the rows numbered rows.302+func (s *sqliteSource) Fetch(ctx context.Context, rows []int64, cols []int) ([][]sheet.LiveCell, error) {303+ cols = allCols(cols, len(s.names))304+ if err := checkCols(cols, len(s.names)); err != nil {305+ return nil, err306+ }307+ for _, r := range rows {308+ if r < 0 || r >= s.rows {309+ return nil, fmt.Errorf("the source has no row %d", r+1)310+ }311+ }312+ b, err := s.byNumber(ctx)313+ if err != nil {314+ return nil, err315+ }316+ return b.fetch(ctx, rows, cols)317+}318+319+func (s *sqliteSource) Close() error {320+ s.mu.Lock()321+ w := s.work322+ s.work, s.base = nil, nil323+ s.mu.Unlock()324+ var err error325+ if w != nil {326+ err = w.close()327+ }328+ return errors.Join(err, s.db.Close())329+}@@ -0,0 +1,204 @@
1+package fileio2+3+import (4+ "context"5+ "fmt"6+ "strings"7+8+ "github.com/FelineStateMachine/012/internal/sheet"9+)10+11+// SQLite views push their sort and filter down to SQL: building one is12+// one INSERT ... SELECT rowid ... WHERE ... ORDER BY into a table of the13+// work database (sourcesqlwork.go), and a page reads its rowids there,14+// by position, then the rows.15+//16+// The SQL follows a sheet's sort and filter, with these differences:17+// SQLite folds case in ASCII only; text holding a date (2026-09-26),18+// which the source shows as that date, is compared and sorted as text;19+// and the text conditions (contains, starts with ...) test a number's20+// text as SQLite writes it (1e+20), where a sheet tests it as shown.21+22+// sqlStmt is a statement and its arguments.23+type sqlStmt struct {24+ q string25+ args []any26+}27+28+// View orders the rows as o says.29+func (s *sqliteSource) View(ctx context.Context, o SourceOrder) (SourceView, error) {30+ b, err := s.byNumber(ctx)31+ if err != nil {32+ return nil, err33+ }34+ if o.IsZero() {35+ return sqlPlainView{s: s, b: b}, nil36+ }37+ where, args, order, err := viewSQL(o, b.cols)38+ if err != nil {39+ return nil, err40+ }41+ s.mu.Lock()42+ w, err := s.workDB()43+ s.mu.Unlock()44+ if err != nil {45+ return nil, err46+ }47+ name := fmt.Sprintf("_012_v%d", w.n.Add(1))48+ from := b.table49+ if !b.work {50+ from = "src." + b.table51+ }52+ n, err := w.build(ctx, s.spec.Path,53+ sqlStmt{q: "CREATE TABLE " + name + " (r INTEGER)"},54+ sqlStmt{q: "INSERT INTO " + name + " (r) SELECT rowid FROM " + from + where + " ORDER BY " + order, args: args})55+ if err != nil {56+ w.db.ExecContext(context.WithoutCancel(ctx), "DROP TABLE IF EXISTS "+name)57+ return nil, err58+ }59+ return &sqlView{s: s, b: b, w: w, name: name, rows: n}, nil60+}61+62+// viewSQL is a view's WHERE clause (with its leading space), its63+// arguments and its ORDER BY, over the columns cols.64+func viewSQL(o SourceOrder, cols []string) (string, []any, string, error) {65+ var conds []string66+ var args []any67+ for _, f := range o.Filter {68+ if f.Col < 0 || f.Col >= len(cols) {69+ return "", nil, "", fmt.Errorf("the source has no column %d", f.Col+1)70+ }71+ c, a := condSQL(cols[f.Col], f.Cond)72+ conds, args = append(conds, c), append(args, a...)73+ }74+ var keys []string75+ for _, k := range o.Sort {76+ if k.Col < 0 || k.Col >= len(cols) {77+ return "", nil, "", fmt.Errorf("the source has no column %d", k.Col+1)78+ }79+ keys = append(keys, orderSQL(cols[k.Col], k.Desc))80+ }81+ where := ""82+ if len(conds) > 0 {83+ where = " WHERE " + strings.Join(conds, " AND ")84+ }85+ return where, args, strings.Join(append(keys, "rowid"), ", "), nil86+}87+88+// orderSQL orders column x as a sheet sorts: numbers, text ignoring89+// case, then the rest (booleans are numbers in SQLite, blobs last), and90+// blanks last in either direction.91+func orderSQL(x string, desc bool) string {92+ dir := ""93+ if desc {94+ dir = " DESC"95+ }96+ v := "nullif(" + x + ", '')" // a blank, NULL or '', is NULL97+ return "(" + v + " IS NULL), " +98+ "CASE typeof(" + v + ") WHEN 'null' THEN 0 WHEN 'integer' THEN 0 WHEN 'real' THEN 0 WHEN 'text' THEN 1 ELSE 2 END" + dir + ", " +99+ v + " COLLATE NOCASE" + dir100+}101+102+// condSQL is a filter's condition on column x as SQL, as103+// sheet.Condition tests a cell: text conditions on the text a value104+// shows, blank for NULL; comparisons with a number (or a date, $1,200,105+// 12%) only of numbers, with text only of text ignoring case; neither106+// matching a blank.107+func condSQL(x string, c sheet.Condition) (string, []any) {108+ shown := "lower(coalesce(CAST(" + x + " AS TEXT), ''))"109+ arg := strings.ToLower(c.Arg)110+ switch c.Op {111+ case sheet.CondEmpty:112+ return "(" + x + " IS NULL OR " + x + " = '')", nil113+ case sheet.CondNotEmpty:114+ return "(" + x + " IS NOT NULL AND " + x + " <> '')", nil115+ case sheet.CondContains:116+ return "instr(" + shown + ", ?) > 0", []any{arg}117+ case sheet.CondNotContains:118+ return "instr(" + shown + ", ?) = 0", []any{arg}119+ case sheet.CondStartsWith:120+ return "substr(" + shown + ", 1, length(?)) = ?", []any{arg, arg}121+ case sheet.CondEndsWith:122+ return "(length(?) = 0 OR substr(" + shown + ", -length(?)) = ?)", []any{arg, arg, arg}123+ case sheet.CondExactly:124+ return shown + " = ?", []any{arg}125+ case sheet.CondGreater, sheet.CondGreaterEq, sheet.CondLess, sheet.CondLessEq, sheet.CondEqual:126+ return compareSQL(x, c)127+ case sheet.CondNotEqual:128+ c.Op = sheet.CondEqual129+ q, args := compareSQL(x, c)130+ return "NOT " + q, args131+ }132+ return "1", nil133+}134+135+var compareOps = map[sheet.CondOp]string{136+ sheet.CondGreater: ">", sheet.CondGreaterEq: ">=", sheet.CondLess: "<", sheet.CondLessEq: "<=", sheet.CondEqual: "=",137+}138+139+// compareSQL compares column x with a condition's value as the sheet's140+// filter does: a number with numbers, text with text.141+func compareSQL(x string, c sheet.Condition) (string, []any) {142+ op := compareOps[c.Op]143+ if n, _, ok := sheet.ParseValue(strings.TrimSpace(c.Arg)); ok {144+ return "(typeof(" + x + ") IN ('integer', 'real') AND " + x + " " + op + " ?)", []any{n}145+ }146+ return "(typeof(" + x + ") = 'text' AND " + x + " <> '' AND upper(" + x + ") " + op + " upper(?))", []any{c.Arg}147+}148+149+// sqlView is a sorted or filtered view: its rows' rowids in a table of150+// the work database.151+type sqlView struct {152+ s *sqliteSource153+ b *sqlBase154+ w *sqlWork155+ name string156+ rows int64157+}158+159+func (v *sqlView) Rows() int64 { return v.rows }160+161+func (v *sqlView) Page(ctx context.Context, from int64, n int) ([]int64, [][]sheet.LiveCell, error) {162+ if from < 0 || from >= v.rows || n <= 0 {163+ return nil, nil, nil164+ }165+ var nums []int64166+ err := queryRows(ctx, v.w.db, "SELECT r FROM "+v.name+" WHERE rowid BETWEEN ? AND ? ORDER BY rowid",167+ []any{from + 1, from + int64(n)}, func(x []any) bool {168+ r, _ := x[0].(int64)169+ nums = append(nums, r-1)170+ return true171+ })172+ if err != nil {173+ return nil, nil, err174+ }175+ rows, err := v.b.fetch(ctx, nums, allCols(nil, len(v.b.cols)))176+ if err != nil {177+ return nil, nil, err178+ }179+ return nums, rows, nil180+}181+182+func (v *sqlView) Close() error {183+ v.s.mu.Lock()184+ defer v.s.mu.Unlock()185+ if v.s.work != v.w {186+ return nil // closed with the source187+ }188+ _, err := v.w.db.Exec("DROP TABLE IF EXISTS " + v.name)189+ return sqliteErr(err)190+}191+192+// sqlPlainView is the rows in the source's own order.193+type sqlPlainView struct {194+ s *sqliteSource195+ b *sqlBase196+}197+198+func (v sqlPlainView) Rows() int64 { return v.s.rows }199+200+func (v sqlPlainView) Page(ctx context.Context, from int64, n int) ([]int64, [][]sheet.LiveCell, error) {201+ return v.b.page(ctx, from, n, v.s.rows)202+}203+204+func (sqlPlainView) Close() error { return nil }@@ -0,0 +1,216 @@
1+package fileio2+3+import (4+ "context"5+ "database/sql"6+ "errors"7+ "fmt"8+ "os"9+ "strings"10+ "sync/atomic"11+12+ "github.com/FelineStateMachine/012/internal/sheet"13+)14+15+// A SQLite source's work database: a file of its own in the temporary16+// directory, never the source's, holding what reading rows by number17+// needs: a copy of a source whose rows can't be found by number18+// otherwise (_012_rows), and each view's rows in its order as a table of19+// rowids (_012_v1, ...), paged by that table's own rowid, so a page20+// deep in a view costs what a page at its top does. Tables are built21+// in SQLite, the source attached read-only as src; SQLite sorts in22+// temporary files, so memory stays bounded. The file goes when the23+// source closes.24+25+type sqlWork struct {26+ path string27+ db *sql.DB28+ n atomic.Int64 // views made, naming their tables29+}30+31+func newSQLWork(dir string) (*sqlWork, error) {32+ f, err := os.CreateTemp(dir, "012-source-*.db")33+ if err != nil {34+ return nil, err35+ }36+ f.Close()37+ uri, err := sqliteURI(f.Name(), "_pragma=journal_mode(OFF)", "_pragma=synchronous(OFF)",38+ "_pragma=busy_timeout(60000)", "_pragma=temp_store(1)")39+ if err != nil {40+ os.Remove(f.Name())41+ return nil, err42+ }43+ db, err := sql.Open("sqlite", uri)44+ if err != nil {45+ os.Remove(f.Name())46+ return nil, err47+ }48+ return &sqlWork{path: f.Name(), db: db}, nil49+}50+51+func (w *sqlWork) close() error {52+ err := w.db.Close()53+ for _, suffix := range []string{"", "-journal", "-wal", "-shm"} {54+ if rerr := os.Remove(w.path + suffix); rerr != nil && !errors.Is(rerr, os.ErrNotExist) {55+ err = errors.Join(err, rerr)56+ }57+ }58+ return err59+}60+61+// build runs statements on a connection of the work database with the62+// source attached as src.63+func (w *sqlWork) build(ctx context.Context, source string, stmts ...sqlStmt) (int64, error) {64+ conn, err := w.db.Conn(ctx)65+ if err != nil {66+ return 0, err67+ }68+ defer conn.Close()69+ uri, err := sqliteURI(source, "mode=ro")70+ if err != nil {71+ return 0, err72+ }73+ if _, err := conn.ExecContext(ctx, "ATTACH DATABASE ? AS src", uri); err != nil {74+ return 0, sqliteErr(err)75+ }76+ defer conn.ExecContext(context.WithoutCancel(ctx), "DETACH DATABASE src")77+ var n int6478+ for _, q := range stmts {79+ res, err := conn.ExecContext(ctx, q.q, q.args...)80+ if err != nil {81+ return 0, sqliteErr(err)82+ }83+ n, _ = res.RowsAffected()84+ }85+ return n, nil86+}87+88+// workDB is the source's work database, made when first needed. It is89+// called with s.mu held.90+func (s *sqliteSource) workDB() (*sqlWork, error) {91+ if s.work == nil {92+ w, err := newSQLWork(s.spec.TempDir)93+ if err != nil {94+ return nil, err95+ }96+ s.work = w97+ }98+ return s.work, nil99+}100+101+// sqlBase is a table whose rowids are its rows' numbers plus one: a102+// dense source table, or the copy in the work database.103+type sqlBase struct {104+ db *sql.DB105+ table string // quoted106+ cols []string // its columns, quoted107+ work bool // in the work database108+}109+110+// byNumber is where rows are found by number, copying the source into111+// the work database the first time when it must.112+func (s *sqliteSource) byNumber(ctx context.Context) (*sqlBase, error) {113+ s.mu.Lock()114+ defer s.mu.Unlock()115+ if s.base != nil {116+ return s.base, nil117+ }118+ if s.table != "" && s.dense {119+ s.base = &sqlBase{db: s.db, table: s.table, cols: quoteAll(s.names)}120+ return s.base, nil121+ }122+ w, err := s.workDB()123+ if err != nil {124+ return nil, err125+ }126+ if _, err := w.build(ctx, s.spec.Path, sqlStmt{q: "CREATE TABLE _012_rows AS " + s.query}); err != nil {127+ w.db.ExecContext(context.WithoutCancel(ctx), "DROP TABLE IF EXISTS _012_rows")128+ return nil, err129+ }130+ var names []string131+ err = queryRows(ctx, w.db, "SELECT name FROM pragma_table_info('_012_rows') ORDER BY cid", nil, func(v []any) bool {132+ names = append(names, fmt.Sprint(v[0]))133+ return true134+ })135+ if err != nil {136+ return nil, err137+ }138+ s.base = &sqlBase{db: w.db, table: "_012_rows", cols: quoteAll(names), work: true}139+ return s.base, nil140+}141+142+func quoteAll(names []string) []string {143+ out := make([]string, len(names))144+ for i, n := range names {145+ out[i] = quoteIdent(n)146+ }147+ return out148+}149+150+// list is the select list of cols after the rowid.151+func (b *sqlBase) list(cols []int) string {152+ parts := make([]string, 0, len(cols)+1)153+ parts = append(parts, "rowid")154+ for _, c := range cols {155+ parts = append(parts, b.cols[c])156+ }157+ return strings.Join(parts, ", ")158+}159+160+// fetchChunk is how many rowids one query asks for.161+const fetchChunk = 500162+163+// fetch reads the rows numbered rows, in the order given.164+func (b *sqlBase) fetch(ctx context.Context, rows []int64, cols []int) ([][]sheet.LiveCell, error) {165+ got := make(map[int64][]sheet.LiveCell, len(rows))166+ for i := 0; i < len(rows); i += fetchChunk {167+ part := rows[i:min(i+fetchChunk, len(rows))]168+ marks := strings.Repeat("?, ", len(part))169+ args := make([]any, len(part))170+ for j, r := range part {171+ args[j] = r + 1172+ }173+ q := "SELECT " + b.list(cols) + " FROM " + b.table + " WHERE rowid IN (" + marks[:len(marks)-2] + ")"174+ if err := b.collect(ctx, q, args, got); err != nil {175+ return nil, err176+ }177+ }178+ out := make([][]sheet.LiveCell, len(rows))179+ for i, r := range rows {180+ out[i] = got[r]181+ }182+ return out, nil183+}184+185+// page reads n rows from row from on.186+func (b *sqlBase) page(ctx context.Context, from int64, n int, rows int64) ([]int64, [][]sheet.LiveCell, error) {187+ if from < 0 || from >= rows || n <= 0 {188+ return nil, nil, nil189+ }190+ got := map[int64][]sheet.LiveCell{}191+ q := "SELECT " + b.list(allCols(nil, len(b.cols))) + " FROM " + b.table + " WHERE rowid BETWEEN ? AND ?"192+ if err := b.collect(ctx, q, []any{from + 1, from + int64(n)}, got); err != nil {193+ return nil, nil, err194+ }195+ var nums []int64196+ var out [][]sheet.LiveCell197+ for r := from; r < min(from+int64(n), rows); r++ {198+ nums, out = append(nums, r), append(out, got[r])199+ }200+ return nums, out, nil201+}202+203+// collect runs a query of a rowid and values, keeping each row's204+// values by its number.205+func (b *sqlBase) collect(ctx context.Context, q string, args []any, got map[int64][]sheet.LiveCell) error {206+ return queryRows(ctx, b.db, q, args, func(v []any) bool {207+ cells := make([]sheet.LiveCell, len(v)-1)208+ for i, x := range v[1:] {209+ c, _ := sqliteCell(x)210+ cells[i] = c.typed()211+ }212+ id, _ := v[0].(int64)213+ got[id-1] = cells214+ return true215+ })216+}@@ -44,6 +44,9 @@ func openSQLite(ctx context.Context, name string) (*sql.DB, error) {
4444 // sqliteErr shortens the driver's errors, e.g. "file is not a database4545 // (26)" for a file that isn't SQLite.4646 func sqliteErr(err error) error {47+ if err == nil {48+ return nil49+ }4750 msg := err.Error()4851 if i := strings.Index(msg, ": "); i >= 0 && strings.HasPrefix(msg, "SQL logic error") {4952 msg = msg[i+2:]