- commit
- 1fe5ba08f54df22585fe9c44b973b21124c986aa
- parent
- d19fab918bc23c50e27b8ec48d89f2e1888ec9f2
- author
- Dami <dami.etoile@gmail.com>
- kind
- feat risk score 2.2 of 3, Jev's estimate
| A | internal/ | 306 | +++++++++++++++++++++++++++++ |
| A | internal/ | 88 | ++++++++ |
| A | internal/ | 317 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 269 | +++++++++++++++++++++++++ |
| A | internal/ | 113 | +++++++++++ |
| A | internal/ | 50 | +++++ |
| M | internal/ | 4 | +- |
@@ -0,0 +1,306 @@
1+package paged2+3+import (4+ "cmp"5+ "context"6+ "iter"7+ "slices"8+9+ "github.com/FelineStateMachine/012/internal/functions"10+ "github.com/FelineStateMachine/012/internal/sheet"11+)12+13+// book is the functions.Book a job computes a call through: the sources14+// the call reads, each as a sheet named by the source's name, whose15+// first row is its columns' names and the rest its rows. Ranges are16+// streamed, a chunk at a time, through a cursor that goes on where the17+// last chunk ended, so a function reading a column once reads the file18+// once; a cell is fetched with its row, and the last chunk's cells are19+// at hand, for lookups that walk a column and read what they pass.20+type book struct {21+ ctx context.Context22+ srcs map[string]snapshot23+ err error // the first a source gave: the answer is then that24+25+ cursors map[cursorKey]*cursor26+ chunk []sheet.Addr // the last chunk Scan read, row by row27+ chunkV []sheet.Value28+ chunkOf string29+ rows map[rowKey][]sheet.LiveCell // rows fetched, a few30+}31+32+type cursorKey struct {33+ src string34+ c0, c1 int35+}36+37+type rowKey struct {38+ src string39+ row int40+}41+42+// maxRows bounds the rows fetched a book keeps.43+const maxRows = 25644+45+func (b *book) close() {46+ for _, c := range b.cursors {47+ c.stop()48+ }49+}50+51+func (b *book) src(name string) (snapshot, bool) {52+ s, ok := b.srcs[key(name)]53+ return s, ok54+}55+56+// table is the source's table as a sheet holds it, header included.57+func (s snapshot) table() sheet.Rect {58+ return sheet.Rect{To: sheet.Addr{Col: len(s.shape.Cols) - 1, Row: s.shape.Rows}}59+}60+61+// clip is the part of r the source's table holds.62+func (s snapshot) clip(r sheet.Rect) (sheet.Rect, bool) {63+ t := s.table()64+ if len(s.shape.Cols) == 0 {65+ return sheet.Rect{}, false66+ }67+ out := sheet.Rect{68+ From: sheet.Addr{Col: max(r.From.Col, t.From.Col), Row: max(r.From.Row, t.From.Row)},69+ To: sheet.Addr{Col: min(r.To.Col, t.To.Col), Row: min(r.To.Row, t.To.Row)},70+ }71+ return out, out.From.Col <= out.To.Col && out.From.Row <= out.To.Row72+}73+74+func (s snapshot) header(col int) sheet.Value {75+ return sheet.Value{Kind: sheet.Text, Str: s.shape.Cols[col]}76+}77+78+// Cell is a cell's value: a column's name, a value of its row, or blank79+// outside the table.80+func (b *book) Cell(name string, a sheet.Addr) sheet.Value {81+ s, ok := b.src(name)82+ if !ok {83+ return sheet.ErrRef84+ }85+ if !s.table().Contains(a) {86+ return sheet.Value{}87+ }88+ if a.Row == 0 {89+ return s.header(a.Col)90+ }91+ if b.chunkOf == key(name) {92+ if i, found := slices.BinarySearchFunc(b.chunk, a, addrOrder); found {93+ return b.chunkV[i]94+ }95+ }96+ row, ok := b.fetch(s, a.Row-1)97+ if !ok || a.Col >= len(row) {98+ return sheet.Value{}99+ }100+ return row[a.Col].V101+}102+103+func addrOrder(x, y sheet.Addr) int {104+ return cmp.Or(cmp.Compare(x.Row, y.Row), cmp.Compare(x.Col, y.Col))105+}106+107+// fetch is the source's row numbered row, every column.108+func (b *book) fetch(s snapshot, row int) ([]sheet.LiveCell, bool) {109+ k := rowKey{key(s.name), row}110+ if r, ok := b.rows[k]; ok {111+ return r, true112+ }113+ got, err := s.h.Fetch(b.ctx, []int64{int64(row)}, nil)114+ if err != nil {115+ b.fail(err)116+ return nil, false117+ }118+ if b.rows == nil || len(b.rows) >= maxRows {119+ b.rows = map[rowKey][]sheet.LiveCell{}120+ }121+ if len(got) == 0 {122+ return nil, false123+ }124+ b.rows[k] = got[0]125+ return got[0], true126+}127+128+func (b *book) fail(err error) {129+ if b.err == nil && b.ctx.Err() == nil {130+ b.err = err131+ }132+}133+134+// Scan reads the cells of r holding something from from on, as135+// functions.Book says: a chunk of a cursor streaming the rows.136+func (b *book) Scan(name string, r sheet.Rect, from sheet.Addr, addrs []sheet.Addr, vals []sheet.Value) int {137+ s, ok := b.src(name)138+ if !ok {139+ return -1140+ }141+ part, ok := s.clip(r)142+ if !ok || from.Row > part.To.Row {143+ return 0144+ }145+ if from.Row < part.From.Row || from.Row == part.From.Row && from.Col < part.From.Col {146+ from = part.From147+ }148+ n := 0149+ b.chunk, b.chunkV, b.chunkOf = b.chunk[:0], b.chunkV[:0], key(name)150+ emit := func(a sheet.Addr, v sheet.Value) bool {151+ if v.Kind == sheet.Empty {152+ return true153+ }154+ addrs[n] = a155+ if vals != nil {156+ vals[n] = v157+ }158+ b.chunk, b.chunkV = append(b.chunk, a), append(b.chunkV, v)159+ n++160+ return n < len(addrs) && !(vals != nil && v.Kind == sheet.Error)161+ }162+ if from.Row == 0 {163+ for c := from.Col; c <= part.To.Col; c++ {164+ if !emit(sheet.Addr{Col: c, Row: 0}, s.header(c)) {165+ return n166+ }167+ }168+ from = sheet.Addr{Col: part.From.Col, Row: 1}169+ }170+ if from.Row > part.To.Row {171+ return n172+ }173+ b.cursor(s, part, from).read(part, emit)174+ return n175+}176+177+// cursor is the cursor over the columns of part from from on: the one178+// the last chunk left there, or a new one.179+func (b *book) cursor(s snapshot, part sheet.Rect, from sheet.Addr) *cursor {180+ k := cursorKey{key(s.name), part.From.Col, part.To.Col}181+ if c := b.cursors[k]; c != nil {182+ if c.at == from {183+ return c184+ }185+ c.stop()186+ }187+ cols := make([]int, 0, part.To.Col-part.From.Col+1)188+ for c := part.From.Col; c <= part.To.Col; c++ {189+ cols = append(cols, c)190+ }191+ c := &cursor{b: b, at: from}192+ seq := func(yield func(int64, []sheet.LiveCell) bool) {193+ if err := s.h.Scan(b.ctx, int64(from.Row-1), cols, yield); err != nil {194+ b.fail(err)195+ }196+ }197+ c.next, c.stop = iter.Pull2(seq)198+ if b.cursors == nil {199+ b.cursors = map[cursorKey]*cursor{}200+ }201+ b.cursors[k] = c202+ return c203+}204+205+// cursor streams a range's rows for Scan, a chunk at a time.206+type cursor struct {207+ b *book208+ at sheet.Addr // where the next chunk starts209+ next func() (int64, []sheet.LiveCell, bool)210+ stop func()211+ row []sheet.LiveCell // the row being read, from the range's first column212+ have bool213+}214+215+// read gives emit the cells of part from the cursor on, until emit216+// returns false or the part ends.217+func (c *cursor) read(part sheet.Rect, emit func(sheet.Addr, sheet.Value) bool) {218+ for c.at.Row <= part.To.Row {219+ if !c.have {220+ _, row, ok := c.next()221+ if !ok {222+ c.at.Row = part.To.Row + 1223+ return224+ }225+ c.row, c.have = append(c.row[:0], row...), true226+ }227+ for c.at.Col <= part.To.Col {228+ v := c.row[c.at.Col-part.From.Col].V229+ a := c.at230+ c.at.Col++231+ if !emit(a, v) {232+ return233+ }234+ }235+ c.at, c.have = sheet.Addr{Col: part.From.Col, Row: c.at.Row + 1}, false236+ }237+}238+239+// Bounds is the part of r the source's table holds.240+func (b *book) Bounds(name string, r sheet.Rect) (sheet.Rect, bool, bool) {241+ s, ok := b.src(name)242+ if !ok {243+ return sheet.Rect{}, false, false244+ }245+ part, any := s.clip(r)246+ return part, any, true247+}248+249+// RangeAgg shares nothing: each call reads its ranges once.250+func (b *book) RangeAgg(string, sheet.Rect) (functions.Agg, *sheet.Value, bool) {251+ return functions.Agg{}, nil, false252+}253+254+// Fold adds the cells of r to agg, streaming the source's rows.255+func (b *book) Fold(name string, r sheet.Rect, agg functions.Agg) (functions.Agg, *sheet.Value) {256+ s, ok := b.src(name)257+ if !ok {258+ return agg, agg.Add(sheet.ErrRef, false)259+ }260+ part, ok := s.clip(r)261+ if !ok {262+ return agg, nil263+ }264+ var e *sheet.Value265+ if part.From.Row == 0 {266+ for c := part.From.Col; c <= part.To.Col && e == nil; c++ {267+ e = agg.Add(s.header(c), false)268+ }269+ part.From.Row = 1270+ }271+ if e != nil || part.From.Row > part.To.Row {272+ return agg, e273+ }274+ cols := make([]int, 0, part.To.Col-part.From.Col+1)275+ for c := part.From.Col; c <= part.To.Col; c++ {276+ cols = append(cols, c)277+ }278+ last := int64(part.To.Row - 1)279+ err := s.h.Scan(b.ctx, int64(part.From.Row-1), cols, func(row int64, vals []sheet.LiveCell) bool {280+ for _, v := range vals {281+ if v.V.Kind != sheet.Empty {282+ if e = agg.Add(v.V, false); e != nil {283+ return false284+ }285+ }286+ }287+ return row < last288+ })289+ if err != nil {290+ b.fail(err)291+ }292+ return agg, e293+}294+295+// Ask has no answers: sources don't ask JEV.296+func (b *book) Ask(functions.RemoteCall) (functions.RemoteAnswer, sheet.Value) {297+ return functions.RemoteAnswer{}, functions.ErrNoRemote298+}299+300+// Paged is false: a job reads its sources as sheets.301+func (b *book) Paged(string) bool { return false }302+303+// Stream isn't asked: nothing is paged here.304+func (b *book) Stream(functions.StreamCall) (sheet.Value, *functions.Array) {305+ return sheet.ErrValue, nil306+}@@ -0,0 +1,88 @@
1+package paged2+3+import (4+ "fmt"5+6+ "github.com/FelineStateMachine/012/internal/functions"7+ "github.com/FelineStateMachine/012/internal/sheet"8+)9+10+// rangeValues is a range of a source read whole, as an array: what an11+// operator or a function reading it some other way is given, no bigger12+// than budget cells.13+func (b *book) rangeValues(name string, r sheet.Rect, budget int) sheet.SourceAnswer {14+ s, ok := b.src(name)15+ if !ok {16+ return sheet.SourceAnswer{V: sheet.ErrRef}17+ }18+ part, ok := s.clip(r)19+ if !ok {20+ return sheet.SourceAnswer{A: &functions.Array{Rows: 1, Cols: 1}}21+ }22+ rows, cols := part.To.Row-part.From.Row+1, part.To.Col-part.From.Col+123+ if rows > budget/cols {24+ return sheet.SourceAnswer{V: sheet.ErrValue, Why: fmt.Sprintf(25+ "Reading %s whole takes %s cells, more than max-cells (%s): SUM, COUNTIFS, XLOOKUP and the like read a source of any size",26+ s.name, grouped(rows*cols), grouped(budget))}27+ }28+ a := functions.NewArray(rows, cols)29+ addrs, vals := make([]sheet.Addr, 1024), make([]sheet.Value, 1024)30+ for from := part.From; ; {31+ n := b.Scan(name, part, from, addrs, vals)32+ for i := range n {33+ a.Set(addrs[i].Row-part.From.Row, addrs[i].Col-part.From.Col, vals[i])34+ }35+ if n < len(addrs) && (n == 0 || vals[n-1].Kind != sheet.Error) {36+ break37+ }38+ from = after(part, addrs[n-1])39+ }40+ return sheet.SourceAnswer{A: a}41+}42+43+// after is the cell after a in r, row by row.44+func after(r sheet.Rect, a sheet.Addr) sheet.Addr {45+ if a.Col < r.To.Col {46+ return sheet.Addr{Col: a.Col + 1, Row: a.Row}47+ }48+ return sheet.Addr{Col: r.From.Col, Row: a.Row + 1}49+}50+51+// pivot gathers a pivot table's groups over the source, streaming the52+// columns it reads.53+func (b *book) pivot(name string, p sheet.Pivot) sheet.SourceAnswer {54+ s, ok := b.src(name)55+ if !ok {56+ return sheet.SourceAnswer{V: sheet.ErrRef}57+ }58+ cols := sheet.PivotCols(p)59+ width := len(s.shape.Cols)60+ for _, c := range cols {61+ if c >= width {62+ return sheet.SourceAnswer{V: sheet.ErrRef, Why: fmt.Sprintf("%s has no column %s", s.name, sheet.ColName(c))}63+ }64+ }65+ row := make([]sheet.LiveCell, width)66+ g, err := sheet.GatherPivot(p, s.shape, func(each func(int, []sheet.LiveCell) bool) error {67+ return s.h.Scan(b.ctx, 0, cols, func(n int64, vals []sheet.LiveCell) bool {68+ for i, c := range cols {69+ row[c] = vals[i]70+ }71+ return each(int(n), row)72+ })73+ })74+ if err != nil {75+ b.fail(err)76+ return sheet.SourceAnswer{V: sheet.ErrRef}77+ }78+ return sheet.SourceAnswer{Pivot: g}79+}80+81+// grouped is n with thousands separators: 12,000,000.82+func grouped(n int) string {83+ s := fmt.Sprint(n)84+ for i := len(s) - 3; i > 0; i -= 3 {85+ s = s[:i] + "," + s[i:]86+ }87+ return s88+}@@ -0,0 +1,317 @@
1+// Package paged reads linked sources in place: Parquet files and SQLite2+// tables or queries too big for any grid, linked as tables on tabs of3+// their own (the engine's source.go). Its Host answers what a4+// workbook's formulas and pivot tables ask of them (sheet.SourceHost)5+// by streaming the sources in the background, keeping the answers until6+// a source's file changes; its Pages are the rows a source's tab shows,7+// read a page at a time as it scrolls.8+//9+// The Host and Pages belong to the goroutine that owns the workbook (the10+// UI's). What reads files runs as Jobs on other goroutines, over what11+// the Job took with it, and hands its result back to be stored, so the12+// owner never waits on a file.13+package paged14+15+import (16+ "context"17+ "os"18+ "strings"19+ "time"20+21+ "github.com/FelineStateMachine/012/internal/fileio"22+ "github.com/FelineStateMachine/012/internal/sheet"23+)24+25+// Linked is a source as the workbook links it, its file resolved.26+type Linked struct {27+ Name string28+ Spec fileio.SourceSpec29+}30+31+// Opener opens what a spec names: fileio.OpenSource, or a stand-in.32+type Opener func(ctx context.Context, spec fileio.SourceSpec) (fileio.Source, error)33+34+// Host answers a workbook's questions about its linked sources.35+type Host struct {36+ // Budget is how many cells a function that holds what it reads may37+ // read of a source, and a range read whole may have: max-cells.38+ Budget int39+ // Open opens a source; fileio.OpenSource unless replaced.40+ Open Opener41+ // Debounce is how long a changed file must hold still before it's42+ // read again.43+ Debounce time.Duration44+ // Now is the clock; tests replace it.45+ Now func() time.Time46+47+ srcs map[string]*source // by name key48+ answers map[string]answer // by question key49+ asked map[string]bool // questions queued or running50+ fresh []*Job // jobs not yet handed out51+}52+53+// source is a linked source as the host has it.54+type source struct {55+ name string56+ spec fileio.SourceSpec57+ h fileio.Source // nil until opened58+ shape sheet.SourceShape59+ err string60+ gen int // counts openings; answers of earlier ones are stale61+ opening bool // a job is opening it62+ read stamp // the file when opened63+ seen stamp // the file when last polled, different from read64+ seenAt time.Time65+}66+67+// stamp is what tells a file changed.68+type stamp struct {69+ size int6470+ mod time.Time71+}72+73+// answer is a stored answer, with the generation of each source it read.74+type answer struct {75+ a sheet.SourceAnswer76+ gens map[string]int77+}78+79+// NewHost is a host with budget max-cells.80+func NewHost(budget int) *Host {81+ return &Host{Budget: budget, Open: fileio.OpenSource, Debounce: 300 * time.Millisecond, Now: time.Now,82+ srcs: map[string]*source{}, answers: map[string]answer{}, asked: map[string]bool{}}83+}84+85+func key(name string) string { return strings.ToLower(name) }86+87+// Link makes the host's sources the workbook's: new ones and those88+// reading something else are opened, those gone are closed. It returns89+// whether anything changed.90+func (h *Host) Link(list []Linked) bool {91+ changed := false92+ keep := map[string]bool{}93+ for _, l := range list {94+ k := key(l.Name)95+ keep[k] = true96+ s := h.srcs[k]97+ if s != nil && s.spec == l.Spec {98+ continue99+ }100+ if s != nil && s.h != nil {101+ s.h.Close()102+ }103+ gen := 0104+ if s != nil {105+ gen = s.gen + 1106+ }107+ s = &source{name: l.Name, spec: l.Spec, gen: gen}108+ h.srcs[k] = s109+ h.reopen(s)110+ changed = true111+ }112+ for k, s := range h.srcs {113+ if !keep[k] {114+ if s.h != nil {115+ s.h.Close()116+ }117+ delete(h.srcs, k)118+ changed = true119+ }120+ }121+ return changed122+}123+124+// reopen queues a job opening s.125+func (h *Host) reopen(s *source) {126+ s.opening = true127+ h.fresh = append(h.fresh, &Job{kind: jobOpen, name: s.name, spec: s.spec, gen: s.gen, open: h.Open})128+}129+130+// Close lets every source go.131+func (h *Host) Close() {132+ for _, s := range h.srcs {133+ if s.h != nil {134+ s.h.Close()135+ }136+ }137+ clear(h.srcs)138+}139+140+// Shape is what the host found of the source named name: its shape, or141+// why it can't be read, and whether it has been opened.142+func (h *Host) Shape(name string) (sheet.SourceShape, string, bool) {143+ s := h.srcs[key(name)]144+ if s == nil || s.h == nil && s.err == "" {145+ return sheet.SourceShape{}, "", false146+ }147+ return s.shape, s.err, true148+}149+150+// Source is the opened source named name, for its tab's pages.151+func (h *Host) Source(name string) (fileio.Source, int, bool) {152+ s := h.srcs[key(name)]153+ if s == nil || s.h == nil {154+ return nil, 0, false155+ }156+ return s.h, s.gen, true157+}158+159+// Answer returns the answer to q, or false while it isn't known, queuing160+// q to be worked out (sheet.SourceHost).161+func (h *Host) Answer(q sheet.SourceQuestion) (sheet.SourceAnswer, bool) {162+ k := q.Key()163+ if a, ok := h.answers[k]; ok && h.current(a.gens) {164+ return a.a, true165+ }166+ if h.asked[k] {167+ return sheet.SourceAnswer{}, false168+ }169+ j := &Job{kind: jobAnswer, q: q, key: k, budget: h.Budget, srcs: map[string]snapshot{}}170+ for _, name := range q.Sources() {171+ s := h.srcs[key(name)]172+ switch {173+ case s == nil:174+ return sheet.SourceAnswer{V: sheet.ErrRef, Why: "There's no linked source named " + name}, true175+ case s.err != "":176+ return sheet.SourceAnswer{V: sheet.ErrRef, Why: s.name + ": " + s.err}, true177+ case s.h == nil:178+ return sheet.SourceAnswer{}, false // opening: asked again once it's open179+ }180+ j.srcs[key(name)] = snapshot{name: s.name, h: s.h, shape: s.shape, gen: s.gen}181+ }182+ h.asked[k] = true183+ h.fresh = append(h.fresh, j)184+ return sheet.SourceAnswer{}, false185+}186+187+// current reports whether answers read with the sources' generations188+// gens are still the sources'.189+func (h *Host) current(gens map[string]int) bool {190+ for k, g := range gens {191+ if s := h.srcs[k]; s == nil || s.gen != g {192+ return false193+ }194+ }195+ return true196+}197+198+// Jobs hands out the jobs queued since the last call, to run in the199+// background, each once.200+func (h *Host) Jobs() []*Job {201+ out := h.fresh202+ h.fresh = nil203+ return out204+}205+206+// Pending reports whether jobs are queued or running.207+func (h *Host) Pending() bool { return len(h.fresh) > 0 || len(h.asked) > 0 || h.opening() }208+209+func (h *Host) opening() bool {210+ for _, s := range h.srcs {211+ if s.opening {212+ return true213+ }214+ }215+ return false216+}217+218+// Done is what storing a job's result tells the workbook.219+type Done struct {220+ // Keys are the questions answered.221+ Keys []string222+ // Opened names a source opened, or failing to, with its shape.223+ Opened string224+ Shape sheet.SourceShape225+ Err string226+}227+228+// Store keeps what the job j found, run and back on the owner's229+// goroutine, and says what the workbook is to be told.230+func (h *Host) Store(j *Job) Done {231+ if j.kind == jobOpen {232+ return h.storeOpen(j)233+ }234+ delete(h.asked, j.key)235+ gens := map[string]int{}236+ for k, snap := range j.srcs {237+ gens[k] = snap.gen238+ }239+ if !h.current(gens) {240+ return Done{} // a source changed while it ran: asked again241+ }242+ h.answers[j.key] = answer{a: j.ans, gens: gens}243+ return Done{Keys: []string{j.key}}244+}245+246+func (h *Host) storeOpen(j *Job) Done {247+ s := h.srcs[key(j.name)]248+ if s == nil || s.gen != j.gen {249+ if j.h != nil {250+ j.h.Close()251+ }252+ return Done{}253+ }254+ s.opening = false255+ if s.h != nil && s.h != j.h {256+ s.h.Close()257+ }258+ s.h, s.shape, s.err, s.read = j.h, j.shape, "", j.stamp259+ if j.err != nil {260+ s.h, s.shape, s.err = nil, sheet.SourceShape{}, describe(j.err)261+ }262+ h.forget(key(s.name))263+ return Done{Opened: s.name, Shape: s.shape, Err: s.err}264+}265+266+// forget drops the answers that read the source with key k.267+func (h *Host) forget(k string) {268+ for qk, a := range h.answers {269+ if _, ok := a.gens[k]; ok {270+ delete(h.answers, qk)271+ }272+ }273+}274+275+// describe is an error as the context line says it.276+func describe(err error) string {277+ if os.IsNotExist(err) {278+ return "the file isn't there"279+ }280+ return err.Error()281+}282+283+// Poll looks at each source's file, once they're open, and queues the284+// ones that changed and have held still for Debounce to be read again,285+// reporting whether it queued any. A file that goes away is noticed the286+// same way.287+func (h *Host) Poll() bool {288+ now, queued := h.Now(), false289+ for _, s := range h.srcs {290+ if s.opening {291+ continue292+ }293+ st, err := os.Stat(s.spec.Path)294+ cur := stamp{}295+ if err == nil {296+ cur = stamp{st.Size(), st.ModTime()}297+ }298+ switch {299+ case cur == s.read:300+ s.seen = stamp{}301+ continue302+ case cur != s.seen:303+ s.seen, s.seenAt = cur, now304+ continue305+ case now.Sub(s.seenAt) < h.Debounce:306+ continue307+ }308+ if s.h != nil {309+ s.h.Close() // jobs reading it fail, and their answers are dropped310+ s.h = nil311+ }312+ s.gen++313+ h.reopen(s)314+ queued = true315+ }316+ return queued317+}@@ -0,0 +1,269 @@
1+package paged2+3+import (4+ "context"5+ "os"6+ "path/filepath"7+ "strings"8+ "testing"9+ "time"10+11+ "github.com/parquet-go/parquet-go"12+13+ "github.com/FelineStateMachine/012/internal/fileio"14+ "github.com/FelineStateMachine/012/internal/sheet"15+)16+17+// sale is a row of the Parquet files the tests link.18+type sale struct {19+ ID int64 `parquet:"id"`20+ Cat *string `parquet:"cat,optional"`21+ Amount *float64 `parquet:"amount,optional"`22+ Day int32 `parquet:"day,date"`23+ Tags []string `parquet:"tags,list"`24+}25+26+var cats = []string{"North", "south", "East", "west", "", "north"}27+28+// writeSales writes n rows to name, amounts shifted by shift, in row29+// groups of 64 rows.30+func writeSales(t testing.TB, name string, n int, shift float64) {31+ t.Helper()32+ f, err := os.Create(name)33+ if err != nil {34+ t.Fatal(err)35+ }36+ w := parquet.NewGenericWriter[sale](f, parquet.MaxRowsPerRowGroup(64))37+ day := int32(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).Unix() / 86400)38+ rows := make([]sale, n)39+ for i := range rows {40+ r := &rows[i]41+ r.ID, r.Day = int64(i), day+int32(i%40)42+ if i%9 != 4 {43+ c := cats[i%len(cats)]44+ r.Cat = &c45+ }46+ if i%11 != 7 {47+ a := float64((i*37)%101) - 20.5 + shift48+ r.Amount = &a49+ }50+ if i%5 == 0 {51+ r.Tags = []string{"a", "b"}52+ }53+ }54+ if _, err := w.Write(rows); err != nil {55+ t.Fatal(err)56+ }57+ if err := w.Close(); err != nil {58+ t.Fatal(err)59+ }60+ if err := f.Close(); err != nil {61+ t.Fatal(err)62+ }63+}64+65+// linked is a workbook holding the file at name imported on Sheet1, as66+// the table ref, and linked as the source sales, settled.67+func linked(t *testing.T, name string, budget int) (*sheet.Workbook, *Host) {68+ t.Helper()69+ res, err := fileio.Import(context.Background(), name, fileio.Options{})70+ if err != nil {71+ t.Fatal(err)72+ }73+ w := res.Sheet.Book()74+ s := w.Sheet(0)75+ if err := w.RenameSheet(s, "Data"); err != nil {76+ t.Fatal(err)77+ }78+ b, _ := s.UsedRange()79+ if err := s.CreateTable("ref", b); err != nil {80+ t.Fatal(err)81+ }82+ if _, err := w.AddSource("sales", sheet.LinkSource{Path: name}, s); err != nil {83+ t.Fatal(err)84+ }85+ h := NewHost(budget)86+ h.Settle(context.Background(), w, func(p string) string { return p })87+ return w, h88+}89+90+// sameFormulas are formulas that read the source as {T}: each gives the91+// same over the source as over the table ref holding the same rows.92+var sameFormulas = []string{93+ "=SUM({T}[amount])", "=AVERAGE({T}[amount])", "=COUNT({T}[amount])", "=COUNTA({T}[cat])",94+ "=MIN({T}[amount])", "=MAX({T}[amount])", "=COUNTA({T}[tags])",95+ `=COUNTIF({T}[cat],"north")`, `=COUNTIF({T}[amount],">10")`, `=COUNTIFS({T}[cat],"North",{T}[amount],"<0")`,96+ `=SUMIF({T}[cat],"East",{T}[amount])`, `=SUMIFS({T}[amount],{T}[cat],"west",{T}[id],">100")`,97+ `=AVERAGEIF({T}[cat],"south",{T}[amount])`, `=AVERAGEIFS({T}[amount],{T}[cat],"<>East")`,98+ "=COUNTBLANK({T}[cat])", "=SUMPRODUCT({T}[amount],{T}[id])", `=COUNTIF({T}[day],">="&DATE(2026,1,20))`,99+ "=MATCH(17,{T}[id],0)", "=MATCH(50.5,{T}[amount],0)", "=XLOOKUP(42,{T}[id],{T}[amount])",100+ `=XLOOKUP("east",{T}[cat],{T}[id],"none",0,-1)`, "=XLOOKUP(10,{T}[amount],{T}[id],,1)",101+ "=XLOOKUP(10,{T}[amount],{T}[id],,-1,-1)", `=XLOOKUP("nowhere",{T}[cat],{T}[id],"none")`,102+ "=VLOOKUP(7,{T},3,FALSE)", "=INDEX({T}[amount],10)", "=ROWS({T})", "=COLUMNS({T}[#All])",103+ "=MEDIAN({T}[amount])", "=SUM({T}[amount]*2)", "=ROUND(SUM({T}[amount]),2)",104+}105+106+func TestFormulasOverASource(t *testing.T) {107+ dir := t.TempDir()108+ name := filepath.Join(dir, "sales.parquet")109+ writeSales(t, name, 5000, 0)110+ w, h := linked(t, name, 100000)111+ s := w.Sheet(0)112+ for i, f := range sameFormulas {113+ at := sheet.Addr{Col: 10, Row: i}114+ ref := sheet.Addr{Col: 11, Row: i}115+ if err := s.Set(at, strings.ReplaceAll(f, "{T}", "sales")); err != nil {116+ t.Fatalf("%s: %v", f, err)117+ }118+ if err := s.Set(ref, strings.ReplaceAll(f, "{T}", "ref")); err != nil {119+ t.Fatalf("%s: %v", f, err)120+ }121+ }122+ h.Settle(context.Background(), w, func(p string) string { return p })123+ for i, f := range sameFormulas {124+ got, want := s.Value(sheet.Addr{Col: 10, Row: i}), s.Value(sheet.Addr{Col: 11, Row: i})125+ if testing.Verbose() {126+ t.Logf("%s: %v", f, got)127+ }128+ if got.Kind == sheet.Error && !strings.Contains(f, "nowhere") {129+ t.Errorf("%s: %v", f, got)130+ }131+ if got != want {132+ t.Errorf("%s: %v over the source, %v over the sheet", f, got, want)133+ }134+ }135+}136+137+func TestPendingUntilAnswered(t *testing.T) {138+ dir := t.TempDir()139+ name := filepath.Join(dir, "sales.parquet")140+ writeSales(t, name, 100, 0)141+ w, h := linked(t, name, 1000)142+ s := w.Sheet(0)143+ a := sheet.Addr{Col: 10}144+ if err := s.Set(a, "=SUM(sales[amount])"); err != nil {145+ t.Fatal(err)146+ }147+ if v := s.Value(a); !sheet.IsPending(v) {148+ t.Fatalf("before the answer: %v, want Loading…", v)149+ }150+ jobs := h.Jobs()151+ if len(jobs) != 1 {152+ t.Fatalf("%d jobs queued, want 1", len(jobs))153+ }154+ jobs[0].Run(context.Background())155+ Apply(w, h.Store(jobs[0]))156+ if v := s.Value(a); v.Kind != sheet.Number {157+ t.Fatalf("after the answer: %v", v)158+ }159+ // Asked again, the answer is kept: no job.160+ if err := s.Set(sheet.Addr{Col: 11}, "=SUM(sales[amount])+1"); err != nil {161+ t.Fatal(err)162+ }163+ if jobs := h.Jobs(); len(jobs) != 0 {164+ t.Errorf("%d jobs for a kept answer", len(jobs))165+ }166+}167+168+func TestPastTheBudget(t *testing.T) {169+ dir := t.TempDir()170+ name := filepath.Join(dir, "sales.parquet")171+ writeSales(t, name, 300, 0)172+ w, h := linked(t, name, 100)173+ s := w.Sheet(0)174+ med, sum := sheet.Addr{Col: 10}, sheet.Addr{Col: 10, Row: 1}175+ s.Set(med, "=MEDIAN(sales[amount])")176+ s.Set(sum, "=SUM(sales[amount])")177+ h.Settle(context.Background(), w, func(p string) string { return p })178+ if v := s.Value(med); v != sheet.ErrValue {179+ t.Errorf("MEDIAN past the budget: %v, want #VALUE!", v)180+ }181+ if why := s.ExplainError(med); !strings.Contains(why, "max-cells") {182+ t.Errorf("MEDIAN past the budget explained as %q", why)183+ }184+ if v := s.Value(sum); v.Kind != sheet.Number {185+ t.Errorf("SUM streams past the budget: %v", v)186+ }187+}188+189+func TestPivotOverASource(t *testing.T) {190+ dir := t.TempDir()191+ name := filepath.Join(dir, "sales.parquet")192+ writeSales(t, name, 500, 0)193+ w, h := linked(t, name, 1000)194+ ref := w.Sheet(0)195+ src, _ := w.LookupSource("sales")196+ p := sheet.Pivot{197+ Rows: []sheet.PivotGroup{{Col: 1}},198+ Columns: []sheet.PivotGroup{{Col: 3}},199+ Values: []sheet.PivotValue{{Col: 2, Summarize: sheet.SumBy}, {Col: 2, Summarize: sheet.CountUniqueBy}},200+ RowTotals: true, ColumnTotals: true,201+ }202+ b, _ := ref.UsedRange()203+ want, err := w.CreatePivot(ref, b, "want", p)204+ if err != nil {205+ t.Fatal(err)206+ }207+ got, err := w.CreatePivot(src.Sheet, sheet.Rect{To: sheet.Addr{Col: 4, Row: sheet.MaxRows - 1}}, "got", p)208+ if err != nil {209+ t.Fatal(err)210+ }211+ h.Settle(context.Background(), w, func(p string) string { return p })212+ wb, _ := want.UsedRange()213+ gb, _ := got.UsedRange()214+ if wb != gb {215+ t.Fatalf("pivot over the source covers %v, over the sheet %v", gb, wb)216+ }217+ for row := wb.From.Row; row <= wb.To.Row; row++ {218+ for col := wb.From.Col; col <= wb.To.Col; col++ {219+ a := sheet.Addr{Col: col, Row: row}220+ if g, x := got.Value(a), want.Value(a); g != x && !(col == 0 && row == 0) {221+ t.Errorf("%s: %v over the source, %v over the sheet", a, g, x)222+ }223+ }224+ }225+}226+227+func TestSourceChanges(t *testing.T) {228+ dir := t.TempDir()229+ name := filepath.Join(dir, "sales.parquet")230+ writeSales(t, name, 100, 0)231+ w, h := linked(t, name, 1000)232+ s := w.Sheet(0)233+ a := sheet.Addr{Col: 10}234+ s.Set(a, "=SUM(sales[amount])")235+ h.Settle(context.Background(), w, func(p string) string { return p })236+ before := s.Value(a)237+ now := time.Now()238+ h.Now = func() time.Time { return now }239+ writeSales(t, name, 100, 1)240+ if h.Poll() {241+ t.Fatal("read again before the file held still")242+ }243+ now = now.Add(time.Second)244+ if !h.Poll() {245+ t.Fatal("a changed file isn't read again")246+ }247+ h.Settle(context.Background(), w, func(p string) string { return p })248+ after := s.Value(a)249+ if after.Kind != sheet.Number || after.Num-before.Num < 80 {250+ t.Errorf("SUM %v before the file changed, %v after", before, after)251+ }252+}253+254+func TestMissingSource(t *testing.T) {255+ w := sheet.NewBook()256+ s := w.Sheet(0)257+ if _, err := w.AddSource("gone", sheet.LinkSource{Path: filepath.Join(t.TempDir(), "gone.parquet")}, s); err != nil {258+ t.Fatal(err)259+ }260+ a := sheet.Addr{}261+ s.Set(a, "=SUM(gone[x])")262+ NewHost(1000).Settle(context.Background(), w, func(p string) string { return p })263+ if v := s.Value(a); v != sheet.ErrRef {264+ t.Errorf("a missing source: %v, want #REF!", v)265+ }266+ if why := s.ExplainError(a); !strings.Contains(why, "isn't there") {267+ t.Errorf("explained as %q", why)268+ }269+}@@ -0,0 +1,113 @@
1+package paged2+3+import (4+ "context"5+ "fmt"6+ "os"7+8+ "github.com/FelineStateMachine/012/internal/fileio"9+ "github.com/FelineStateMachine/012/internal/functions"10+ "github.com/FelineStateMachine/012/internal/sheet"11+)12+13+type jobKind uint814+15+const (16+ jobOpen jobKind = iota17+ jobAnswer18+)19+20+// A Job is work the host queued: opening a source, or answering a21+// question by reading the sources it names. Run it on any goroutine,22+// once, then give it back to Host.Store on the owner's.23+type Job struct {24+ kind jobKind25+26+ // Opening a source.27+ name string28+ spec fileio.SourceSpec29+ gen int30+ open Opener31+ h fileio.Source32+ shape sheet.SourceShape33+ stamp stamp34+ err error35+36+ // Answering a question.37+ q sheet.SourceQuestion38+ key string39+ srcs map[string]snapshot // the sources it reads, by name key40+ budget int41+ ans sheet.SourceAnswer42+}43+44+// snapshot is a source as a job reads it: opened, with what it is.45+type snapshot struct {46+ name string47+ h fileio.Source48+ shape sheet.SourceShape49+ gen int50+}51+52+// String says what the job does, for telemetry.53+func (j *Job) String() string {54+ if j.kind == jobOpen {55+ return "open " + j.name56+ }57+ return j.key58+}59+60+// Run does the job.61+func (j *Job) Run(ctx context.Context) {62+ if j.kind == jobOpen {63+ j.runOpen(ctx)64+ return65+ }66+ j.ans = j.answer(ctx)67+}68+69+func (j *Job) runOpen(ctx context.Context) {70+ if st, err := os.Stat(j.spec.Path); err == nil {71+ j.stamp = stamp{st.Size(), st.ModTime()}72+ }73+ h, err := j.open(ctx, j.spec)74+ if err != nil {75+ j.err = err76+ return77+ }78+ j.h, j.shape = h, ShapeOf(h)79+}80+81+// ShapeOf is what the engine keeps of a source.82+func ShapeOf(h fileio.Source) sheet.SourceShape {83+ cols := h.Columns()84+ sh := sheet.SourceShape{Rows: int(h.Rows())}85+ for _, c := range cols {86+ sh.Cols = append(sh.Cols, c.Name)87+ sh.Formats = append(sh.Formats, c.Format)88+ sh.Numeric = append(sh.Numeric, c.Numeric)89+ }90+ return sh91+}92+93+// answer works out the job's question.94+func (j *Job) answer(ctx context.Context) sheet.SourceAnswer {95+ b := &book{ctx: ctx, srcs: j.srcs}96+ defer b.close()97+ var ans sheet.SourceAnswer98+ switch j.q.Kind {99+ case sheet.AskCell:100+ ans.V = b.Cell(j.q.Source, j.q.At)101+ case sheet.AskRange:102+ ans = b.rangeValues(j.q.Source, j.q.R, j.budget)103+ case sheet.AskCall:104+ sa := functions.EvalStream(j.q.Call, b, j.budget)105+ ans = sheet.SourceAnswer{V: sa.V, A: sa.A, Why: sa.Why}106+ case sheet.AskPivot:107+ ans = b.pivot(j.q.Source, j.q.Pivot)108+ }109+ if b.err != nil {110+ return sheet.SourceAnswer{V: sheet.ErrRef, Why: fmt.Sprintf("%s: %s", j.q.Source, describe(b.err))}111+ }112+ return ans113+}@@ -0,0 +1,50 @@
1+package paged2+3+import (4+ "context"5+6+ "github.com/FelineStateMachine/012/internal/fileio"7+ "github.com/FelineStateMachine/012/internal/sheet"8+)9+10+// Links are the workbook's linked sources as the host reads them: each11+// one's file through abs (which resolves it from the workbook's folder),12+// what they build kept in tempDir ("" for the system's).13+func Links(w *sheet.Workbook, abs func(string) string, tempDir string) []Linked {14+ var out []Linked15+ for _, info := range w.Sources() {16+ src := info.Source17+ out = append(out, Linked{Name: info.Name, Spec: fileio.SourceSpec{18+ Path: abs(src.Path), Format: src.Format, Table: src.Table, Query: src.Query, TempDir: tempDir,19+ }})20+ }21+ return out22+}23+24+// Apply tells the workbook what storing a job found: a source opened,25+// or questions answered.26+func Apply(w *sheet.Workbook, d Done) {27+ if d.Opened != "" {28+ w.SetSourceShape(d.Opened, d.Shape, d.Err)29+ }30+ if len(d.Keys) > 0 {31+ w.RecalcSourced(d.Keys)32+ }33+}34+35+// Settle links the workbook's sources and runs every job the host36+// queues, one after another, applying each, until none is left: for37+// what has no screen to keep live while they run (012 get, tests).38+func (h *Host) Settle(ctx context.Context, w *sheet.Workbook, abs func(string) string) {39+ h.Link(Links(w, abs, ""))40+ w.SetSources(h)41+ for jobs := h.Jobs(); len(jobs) > 0; jobs = h.Jobs() {42+ for _, j := range jobs {43+ j.Run(ctx)44+ Apply(w, h.Store(j))45+ }46+ if ctx.Err() != nil {47+ return48+ }49+ }50+}@@ -83,8 +83,8 @@ func (rd *reader) pagedRange(t *Sheet, reg Region, r Rect) (Rect, *functions.Arr
8383 func (rd *reader) pagedScan(t *Sheet, reg Region, r Rect, from Addr, addrs []Addr, vals []Value) int {8484 part, arr, err := rd.pagedRange(t, reg, r)8585 if err != nil {86- if len(addrs) == 0 {87- return 086+ if len(addrs) == 0 || from != r.From {87+ return 0 // the error was the range's first cell8888 }8989 addrs[0] = r.From9090 if vals != nil {