An order statistic of one range of a linked source no longer holds the column: the job streams its numbers into the external merge sort a Parquet view's sort uses, made generic over its records, in runs of a million numbers spilled to temporary files, and reads them back in order once. RANK counts as it reads. Both work alike for Parquet and SQLite, where ORDER BY with OFFSET took 19 s on ten million rows. On ten million rows MEDIAN takes 1.7 s (Parquet) or 2.8 s (SQLite) in 50 MB, against 248 MB held; answers are kept as other questions'.
- commit
- 1bbd3fd89ec9c7b6903714b8e90cfcb383680ec2
- parent
- 7920678c606cb21f8a298c68c3b5886776e35af6
- author
- Dami <dami.etoile@gmail.com>
- kind
- feat risk score 1.9 of 3, Jev's estimate
| M | ROADMAP.md | 2 | +- |
| M | docs/ | 23 | ++- |
| A | internal/ | 233 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 113 | +++++++++++++++ |
| A | internal/ | 50 | ++++++ |
| M | internal/ | 207 | +++++---------------------- |
| M | internal/ | 25 | +++ |
| M | internal/ | 25 | ++- |
| M | internal/ | 69 | +++++++++ |
| M | internal/ | 2 | +- |
| M | internal/ | 6 | +- |
| M | internal/ | 1 | + |
| A | internal/ | 97 | ++++++++++++ |
| M | internal/ | 7 | +- |
@@ -117,7 +117,7 @@ needs cgo.
117117 - Commands for scripts without the screen: `012 get` (text, CSV, TSV, JSON or NUON), `012 set`, `012 recalc` and `012 export`, running notebooks and JEV only behind flags: [Scripts](docs/files/scripts.md)118118 - `012 diff` cell by cell, as git's diff command or textconv, and `012 merge-driver` merging cell by cell with conflicts noted on the cells: [Diff and merge in git](docs/files/git.md)119119 - Following files: a linked region follows a CSV, TSV, JSON lines or NUON file as it grows and any importable file as it's rewritten, keeping every row or the last ones, its rows arriving as the change stream: [Following files](docs/files/following.md)120-- Linked sources: a Parquet file or a SQLite table or query read in place on a tab of its own, scrolled, sorted and filtered however many rows it has, formulas and pivot tables streaming over it: [Linked sources](docs/files/sources.md)120+- Linked sources: a Parquet file or a SQLite table or query read in place on a tab of its own, scrolled, sorted and filtered however many rows it has, formulas (order statistics such as MEDIAN included, sorted on disk) and pivot tables streaming over it, results in their columns' formats: [Linked sources](docs/files/sources.md)121121 - The `.012` format written down with a version, and each release's workbooks kept as fixtures that every later build opens and saves unchanged: [The .012 format](docs/files/format.md#versions), [Releasing](docs/contributing/releasing.md#file-fixtures)122122 - Web pages in 012's look: a sheet, a range or a chart as one self-contained `.html` file, charts as SVG, from File > Download and `012 export`: [Files](docs/files/README.md#web-pages)123123 - A crash keeps unsaved work for recovery, restores the terminal and writes a report, locally and in `012 serve`: [Saving](docs/files/saving.md#if-012-crashes)@@ -39,7 +39,7 @@ it lags, and past a second it stalls.
3939 | Find, filter, sort, fill | Filter or sort 8191 rows: 1 to 32 ms; find over 213 k cells: 38 ms; fill 8192 rows: 4 ms | Replace all over 213 k cells: 280 ms | | Per-cell string conversion and regexp |4040 | SSH sessions (`012 serve`) | 50 sessions typing at once: frames within one frame interval (p95 8.7 ms), 1.3 MiB per session plus its sheets (upper bound) | | More than `--max-sessions` (8 by default): turned away | Bubble Tea's 120 fps pacing; per session, the terminal's cell buffers |4141 | Pivot tables | A pivot over 8191 rows recomputes after an edit to its source in 0.8 to 2.5 ms | | Results past the grid (the pivot shows #REF!) | Grouping the source rows |42-| Linked sources | Ten million rows of Parquet or SQLite: opened in 0.4 ms or 0.65 s, a frame scrolling them 0.9 ms at 200 x 60, a page anywhere 0.2 to 12 ms; SUM of a column 0.42 s (Parquet) or 1.4 s (SQLite), COUNTIFS 1.2 or 2.3 s, a pivot table 2.5 or 3.9 s, each in a few MB | Sorting ten million rows: 3.5 s (Parquet) or 19 s (SQLite), in the background | Functions that hold what they read, over more than `max-cells` cells: `#VALUE!`, saying why | Reading the column from the file; sorting |42+| Linked sources | Ten million rows of Parquet or SQLite: opened in 0.4 ms or 0.65 s, a frame scrolling them 0.9 ms at 200 x 60, a page anywhere 0.2 to 12 ms; SUM of a column 0.42 s (Parquet) or 1.4 s (SQLite), COUNTIFS 1.2 or 2.3 s, a pivot table 2.5 or 3.9 s, each in a few MB; MEDIAN 1.7 or 2.8 s in 50 MB | Sorting ten million rows: 3.5 s (Parquet) or 19 s (SQLite), in the background | Functions that hold what they read, over more than `max-cells` cells: `#VALUE!`, saying why | Reading the column from the file; sorting |4343 | Following files | A log growing by 10,000 rows a second, formulas over it: each update (about 2,500 rows, four times a second) applied in 7 ms keeping every row, 1.7 to 2 ms keeping the last 1000, 14 ms the last 10,000; frames 2.5 to 3.6 ms at 200 x 60 | | Rows past `max-cells`: left out, with a note | Writing the rows and recalculating what reads them; a window rewrites its rows on every update |4444 | Notebook streams (`TestStreamGridAtFrameSpeed`) | 10,000 rows arriving in batches of 250, the output's grid on screen: 2.3 ms a batch through to its frame at 200 x 60 (3.6 ms at worst) | Past 20,000 rows in one run: the grid is read again from the 10,000 the output keeps, about 60 ms once every 10,000 rows | | Reading the batch's rows into the grid and measuring their text; the output's NUON copied once a batch |4545 | Arrays and spills | FILTER, SORT or UNIQUE over 8192 rows: about 0.1 ms each per edit; 1000 of them spilling 516 k cells, an edit recomputing 500: 52 ms | Full recalculation of those 1000: 106 ms | An array past `maxArray` values (2,097,152) stored: `#VALUE!`; a spill past the sheet's edge or `max-cells` cells: `#REF!` | Computing each array's values; writing only the spilled cells that changed |@@ -403,16 +403,27 @@ run each, the peak heap above what was live before:
403403 | `SUMPRODUCT` of two columns | 0.44 s, 9 MB | 1.56 s, 7.7 MB |404404 | `XLOOKUP` of the last id | 0.62 s, 5.9 MB | 1.28 s, 4.3 MB |405405 | `MATCH` of the middle id | 0.31 s, 5.9 MB | 0.63 s, 3.8 MB |406-| `MEDIAN`, which holds every value: ten million cells are within `max-cells` | 1.15 s, 248 MB | 2.15 s, 247 MB |406+| `MEDIAN`, sorting the column on disk | 1.71 s, 50 MB | 2.75 s, 49 MB |407+| `PERCENTILE`, `LARGE`, `MODE`, the same sort | 1.86 to 1.93 s, 50 MB | 2.93 to 2.97 s, 49 MB |408+| `RANK`, counting as it reads | 0.40 s, 5.7 MB | 1.39 s, 4.8 MB |409+| `STDEV`, which holds every value: ten million cells are within `max-cells` | 0.43 s, 248 MB | 1.42 s, 247 MB |407410 | A pivot table: SUM and COUNT of the amounts by category | 2.45 s, 5.9 MB | 3.9 s, 2.7 MB |408411 409412 SQLite's own memory, through modernc's libc, isn't in the Go heap.410413 Functions over aligned ranges (the criteria functions, `SUMPRODUCT`)411414 read their ranges a window of 2048 rows at a time, each column412-streamed by a goroutine of its own in batches of 4096 rows. A function413-that holds what it reads is refused past `max-cells` cells before it414-reads any, so `MEDIAN` of a hundred million values says so at once415-rather than holding them.415+streamed by a goroutine of its own in batches of 4096 rows. The order416+statistics (`MEDIAN`, `PERCENTILE`, `QUARTILE`, `LARGE`, `SMALL`,417+`MODE`) of one column stream its numbers into the external merge sort a418+Parquet view's sort uses, in runs of a million numbers (16 MB) spilled419+beside the system's temporary files, 160 MB of them for ten million420+rows, then read back in order once, picking what they need as they421+pass; the runs are removed as the answer is found. SQLite's own422+`ORDER BY amount LIMIT 2 OFFSET 4999999` takes 19 s on the same table,423+so a SQLite source is read and sorted the same way. A function that424+holds what it reads is refused past `max-cells` cells before it reads425+any, so `STDEV` of a hundred million values says so at once rather426+than holding them.416427 417428 Scrolling (`BenchmarkSourceScroll`, `internal/ui`) draws a frame in418429 0.26 ms at 80 x 24 and 0.9 ms at 200 x 60 within the pages read, as a@@ -0,0 +1,233 @@
1+package fileio2+3+import (4+ "bufio"5+ "container/heap"6+ "errors"7+ "io"8+ "os"9+ "slices"10+)11+12+// An external merge sort of records of any kind in bounded memory: runs13+// of at most sortRunBytes sorted in memory and spilled to files, then14+// merged, read back one record at a time. A Parquet view's sort orders15+// its rows with it (sourcesort.go), and the order statistics of a16+// source's column its numbers (numsort.go).17+18+// sortRunBytes bounds the records a sort holds in memory at once.19+var sortRunBytes = 64 << 2020+21+// recCodec is how a sort compares, sizes and stores its records.22+type recCodec[T any] struct {23+ cmp func(a, b T) int24+ size func(*T) int // about what a record holds in memory25+ // fixed is every record's size when they're all one size, so the26+ // sort can grow its run to what one holds and no further.27+ fixed int28+ put func(*bufio.Writer, *T)29+ // get reads a record into the one given, io.EOF at a run's end.30+ get func(*bufio.Reader, *T) error31+}32+33+// extSort sorts records in bounded memory.34+type extSort[T any] struct {35+ dir string36+ codec recCodec[T]37+ recs []T38+ bytes int39+ runs []string40+ limit int41+ n int6442+}43+44+func newExtSort[T any](dir string, codec recCodec[T]) *extSort[T] {45+ return &extSort[T]{dir: dir, codec: codec, limit: sortRunBytes}46+}47+48+// add takes a record, spilling a run when memory is full.49+func (s *extSort[T]) add(r T) error {50+ if f := s.codec.fixed; f > 0 && len(s.recs) == cap(s.recs) {51+ // Grow no further than a run holds: doubling past it would52+ // hold twice the run while copying.53+ most := max(s.limit/f, 1)54+ grown := make([]T, len(s.recs), len(s.recs)+max(min(max(cap(s.recs), 1024), most-len(s.recs)), 1))55+ copy(grown, s.recs)56+ s.recs = grown57+ }58+ s.recs = append(s.recs, r)59+ s.bytes += s.codec.size(&s.recs[len(s.recs)-1])60+ s.n++61+ if s.bytes >= s.limit {62+ return s.spill()63+ }64+ return nil65+}66+67+func (s *extSort[T]) sort() {68+ slices.SortFunc(s.recs, s.codec.cmp)69+}70+71+// spill sorts the records in memory and writes them as a run.72+func (s *extSort[T]) spill() error {73+ s.sort()74+ f, err := os.CreateTemp(s.dir, "012-sort-*.run")75+ if err != nil {76+ return err77+ }78+ s.runs = append(s.runs, f.Name())79+ w := bufio.NewWriterSize(f, 1<<20)80+ for i := range s.recs {81+ s.codec.put(w, &s.recs[i])82+ }83+ err = errors.Join(w.Flush(), f.Close())84+ clear(s.recs)85+ s.recs, s.bytes = s.recs[:0], 086+ return err87+}88+89+// sorted is the records in order, read once: those held when none were90+// spilled, else the runs merged. Closing it removes the runs.91+func (s *extSort[T]) sorted() (*sortedRecs[T], error) {92+ out := &sortedRecs[T]{s: s}93+ if len(s.runs) == 0 {94+ s.sort()95+ return out, nil96+ }97+ if len(s.recs) > 0 {98+ if err := s.spill(); err != nil {99+ s.remove()100+ return nil, err101+ }102+ }103+ out.h = &runHeap[T]{cmp: s.codec.cmp}104+ for _, name := range s.runs {105+ f, err := os.Open(name)106+ if err != nil {107+ out.close()108+ return nil, err109+ }110+ r := &runReader[T]{f: f, r: bufio.NewReaderSize(f, 256<<10), get: s.codec.get}111+ if r.next() {112+ out.h.runs = append(out.h.runs, r)113+ } else {114+ f.Close()115+ out.h.done = append(out.h.done, r)116+ }117+ }118+ heap.Init(out.h)119+ return out, nil120+}121+122+// remove deletes the runs.123+func (s *extSort[T]) remove() {124+ for _, name := range s.runs {125+ os.Remove(name)126+ }127+ s.runs = nil128+}129+130+// sortedRecs reads a sort's records in order.131+type sortedRecs[T any] struct {132+ s *extSort[T]133+ h *runHeap[T] // nil when the records are held134+ i int135+ last *runReader[T] // the run the record handed out last came from136+}137+138+// next is the next record, valid until the next call; nil at the end139+// or on an error, which err says.140+func (o *sortedRecs[T]) next() *T {141+ if o.h == nil {142+ if o.i >= len(o.s.recs) {143+ return nil144+ }145+ o.i++146+ return &o.s.recs[o.i-1]147+ }148+ if o.last != nil { // move past the record handed out last149+ if o.last.next() {150+ heap.Fix(o.h, 0)151+ } else {152+ o.last.f.Close()153+ heap.Pop(o.h)154+ }155+ o.last = nil156+ }157+ if o.h.Len() == 0 {158+ return nil159+ }160+ o.last = o.h.runs[0]161+ return &o.last.rec162+}163+164+// err is the first error a run met.165+func (o *sortedRecs[T]) err() error {166+ if o.h == nil {167+ return nil168+ }169+ return o.h.err()170+}171+172+// close lets go of the runs.173+func (o *sortedRecs[T]) close() {174+ if o.h != nil {175+ o.h.close()176+ }177+ o.s.remove()178+}179+180+// runReader reads a run's records in order.181+type runReader[T any] struct {182+ f *os.File183+ r *bufio.Reader184+ get func(*bufio.Reader, *T) error185+ rec T186+ fail error187+}188+189+// next reads the next record, false at the end or on an error.190+func (r *runReader[T]) next() bool {191+ if err := r.get(r.r, &r.rec); err != nil {192+ if !errors.Is(err, io.EOF) {193+ r.fail = err194+ }195+ return false196+ }197+ return true198+}199+200+// runHeap merges runs by their next record.201+type runHeap[T any] struct {202+ runs []*runReader[T]203+ cmp func(a, b T) int204+ done []*runReader[T]205+}206+207+func (h *runHeap[T]) Len() int { return len(h.runs) }208+func (h *runHeap[T]) Less(i, j int) bool { return h.cmp(h.runs[i].rec, h.runs[j].rec) < 0 }209+func (h *runHeap[T]) Swap(i, j int) { h.runs[i], h.runs[j] = h.runs[j], h.runs[i] }210+func (h *runHeap[T]) Push(x any) { h.runs = append(h.runs, x.(*runReader[T])) }211+func (h *runHeap[T]) Pop() any {212+ r := h.runs[len(h.runs)-1]213+ h.runs = h.runs[:len(h.runs)-1]214+ h.done = append(h.done, r)215+ return r216+}217+218+// err is the first error a run met.219+func (h *runHeap[T]) err() error {220+ for _, r := range append(h.done, h.runs...) {221+ if r.fail != nil {222+ return r.fail223+ }224+ }225+ return nil226+}227+228+func (h *runHeap[T]) close() {229+ for _, r := range h.runs {230+ r.f.Close()231+ }232+ h.runs = nil233+}@@ -0,0 +1,113 @@
1+package fileio2+3+import (4+ "bufio"5+ "cmp"6+ "encoding/binary"7+ "io"8+ "math"9+)10+11+// NumberSort sorts numbers in bounded memory, the external merge sort12+// of a view's rows (extsort.go) over 16-byte records: each number and13+// its position in the order it was read. The order statistics of a14+// source's column (MEDIAN, PERCENTILE, MODE) read them back in15+// ascending order, ties in that order, holding at most a run of them,16+// sortRunBytes, while the rest wait on disk.17+type NumberSort struct {18+ s *extSort[numRec]19+}20+21+type numRec struct {22+ v float6423+ pos int6424+}25+26+var numCodec = recCodec[numRec]{27+ cmp: func(a, b numRec) int {28+ switch {29+ case a.v < b.v:30+ return -131+ case a.v > b.v:32+ return 133+ }34+ return cmp.Compare(a.pos, b.pos)35+ },36+ size: func(*numRec) int { return 16 },37+ fixed: 16,38+ put: func(w *bufio.Writer, r *numRec) {39+ var b [16]byte40+ binary.LittleEndian.PutUint64(b[:], math.Float64bits(r.v))41+ binary.LittleEndian.PutUint64(b[8:], uint64(r.pos))42+ w.Write(b[:])43+ },44+ get: func(r *bufio.Reader, rec *numRec) error {45+ var b [16]byte46+ if _, err := io.ReadFull(r, b[:]); err != nil {47+ return err // io.EOF only where a record would start48+ }49+ rec.v = math.Float64frombits(binary.LittleEndian.Uint64(b[:]))50+ rec.pos = int64(binary.LittleEndian.Uint64(b[8:]))51+ return nil52+ },53+}54+55+// NewNumberSort is a sort spilling its runs to files in dir, "" for the56+// system's temporary directory.57+func NewNumberSort(dir string) *NumberSort {58+ s := newExtSort(dir, numCodec)59+ s.limit = max(sortRunBytes/numRunShare, 16)60+ return &NumberSort{s: s}61+}62+63+// numRunShare is the share of sortRunBytes a run of numbers holds: a64+// million numbers, sorted faster in runs of their own and merged, and65+// a few times that while the run grows and the collector catches up.66+const numRunShare = 467+68+// Add takes a number read at position pos.69+func (n *NumberSort) Add(v float64, pos int64) error { return n.s.add(numRec{v, pos}) }70+71+// Len counts the numbers added.72+func (n *NumberSort) Len() int { return int(n.s.n) }73+74+// Close removes what the sort spilled, when Sorted wasn't asked for.75+func (n *NumberSort) Close() { n.s.remove() }76+77+// Sorted reads the numbers back in ascending order, once. Closing it78+// removes the runs.79+func (n *NumberSort) Sorted() (*SortedNumbers, error) {80+ o, err := n.s.sorted()81+ if err != nil {82+ return nil, err83+ }84+ return &SortedNumbers{o: o, n: n.Len()}, nil85+}86+87+// SortedNumbers is a NumberSort's numbers in ascending order.88+type SortedNumbers struct {89+ o *sortedRecs[numRec]90+ n int91+}92+93+// Len counts the numbers.94+func (s *SortedNumbers) Len() int { return s.n }95+96+// Next is the next number and its position as read; false at the end,97+// or on an error, which Err says.98+func (s *SortedNumbers) Next() (float64, int64, bool) {99+ r := s.o.next()100+ if r == nil {101+ return 0, 0, false102+ }103+ return r.v, r.pos, true104+}105+106+// Err is the error reading the runs met, if any.107+func (s *SortedNumbers) Err() error { return s.o.err() }108+109+// Close removes the runs.110+func (s *SortedNumbers) Close() error {111+ s.o.close()112+ return nil113+}@@ -0,0 +1,50 @@
1+package fileio2+3+import (4+ "cmp"5+ "math/rand/v2"6+ "os"7+ "slices"8+ "testing"9+)10+11+// TestNumberSort sorts numbers held and spilled, against sorting them12+// in memory, ties in the order they were added, and leaves no files.13+func TestNumberSort(t *testing.T) {14+ defer func(n int) { sortRunBytes = n }(sortRunBytes)15+ for _, run := range []int{1 << 20, 4096} {16+ sortRunBytes = run17+ dir := t.TempDir()18+ r := rand.New(rand.NewPCG(1, uint64(run)))19+ s := NewNumberSort(dir)20+ var want []numRec21+ for i := range 5000 {22+ v := float64(r.IntN(700)) - 350.523+ want = append(want, numRec{v, int64(i)})24+ if err := s.Add(v, int64(i)); err != nil {25+ t.Fatal(err)26+ }27+ }28+ slices.SortStableFunc(want, func(a, b numRec) int { return cmp.Compare(a.v, b.v) })29+ got, err := s.Sorted()30+ if err != nil {31+ t.Fatal(err)32+ }33+ if got.Len() != len(want) {34+ t.Errorf("run %d: %d numbers, want %d", run, got.Len(), len(want))35+ }36+ for i, w := range want {37+ v, pos, ok := got.Next()38+ if !ok || v != w.v || pos != w.pos {39+ t.Fatalf("run %d: #%d is %v at %d (%v), want %v at %d", run, i, v, pos, ok, w.v, w.pos)40+ }41+ }42+ if _, _, ok := got.Next(); ok || got.Err() != nil {43+ t.Errorf("run %d: more past the end, or %v", run, got.Err())44+ }45+ got.Close()46+ if left, _ := os.ReadDir(dir); len(left) != 0 {47+ t.Errorf("run %d: %d files left", run, len(left))48+ }49+ }50+}@@ -3,13 +3,11 @@ package fileio
33 import (44 "bufio"55 "cmp"6- "container/heap"76 "encoding/binary"87 "errors"98 "io"109 "math"1110 "os"12- "slices"1311 "strings"1412 1513 "github.com/FelineStateMachine/012/internal/sheet"@@ -18,12 +16,9 @@ import (
1816 // A Parquet view's rows are a file of row numbers in the view's order,1917 // 8 bytes each, which a page reads with ReadAt: a filter writes the rows2018 // 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 << 2019+// sort (extsort.go) of each row's number and keys. Memory stays bounded20+// whatever the row count; the disk holds the runs and the order, about21+// 8 bytes a row plus the keys.2722 2823 // sortVal is one key of a row as a sort compares it: its kind's rank2924 // (numbers, text, booleans, errors, then blanks), and its number or its@@ -97,118 +92,42 @@ func compareRecs(a, b *sortRec, desc []bool) int {
9792 return cmp.Compare(a.row, b.row)9893 }9994 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-}95+// extSorter sorts a view's rows by their keys in bounded memory.96+type extSorter struct{ *extSort[sortRec] }10997 11098 func newExtSorter(dir string, desc []bool) *extSorter {111- return &extSorter{dir: dir, desc: desc, limit: sortRunBytes}99+ keys := len(desc)100+ return &extSorter{newExtSort(dir, recCodec[sortRec]{101+ cmp: func(a, b sortRec) int { return compareRecs(&a, &b, desc) },102+ size: (*sortRec).size,103+ put: writeRec,104+ get: func(r *bufio.Reader, rec *sortRec) error { return readRec(r, rec, keys) },105+ })}112106 }113107 114-// add takes a row's keys, spilling a run when memory is full.108+// add takes a row's keys.115109 func (s *extSorter) add(row int64, keys []sheet.LiveCell) error {116110 r := sortRec{row: row, keys: make([]sortVal, len(keys))}117111 for i, k := range keys {118112 r.keys[i] = sortValOf(k.V)119113 }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 err114+ return s.extSort.add(r)148115 }149116 150117 // finish writes the sorted rows' numbers to out and removes the runs.151118 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- }119+ o, err := s.sorted()120+ if err != nil {121+ return 0, err194122 }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 {123+ defer o.close()124+ for r := o.next(); r != nil; r = o.next() {125+ if err := writeRow(out, r.row); err != nil {201126 return n, err202127 }203128 n++204- if r.next() {205- heap.Fix(h, 0)206- } else {207- r.f.Close()208- heap.Pop(h)209- }210129 }211- return n, h.err()130+ return n, o.err()212131 }213132 214133 func writeRow(w io.Writer, row int64) error {@@ -236,39 +155,29 @@ func writeRec(w *bufio.Writer, r *sortRec) {
236155 }237156 }238157 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)158+// readRec reads a record of keys keys from a run, io.EOF at its end.159+func readRec(r *bufio.Reader, rec *sortRec, keys int) error {160+ row, err := binary.ReadUvarint(r)251161 if err != nil {252- if !errors.Is(err, io.EOF) {253- r.fail = err254- }255- return false162+ return err256163 }257- r.rec.row = int64(row)258- r.rec.keys = r.rec.keys[:0]259- for range r.keys {260- k, err := r.key()164+ rec.row = int64(row)165+ rec.keys = rec.keys[:0]166+ for range keys {167+ k, err := readKey(r)261168 if err != nil {262- r.fail = err263- return false169+ if errors.Is(err, io.EOF) {170+ err = io.ErrUnexpectedEOF171+ }172+ return err264173 }265- r.rec.keys = append(r.rec.keys, k)174+ rec.keys = append(rec.keys, k)266175 }267- return true176+ return nil268177 }269178 270-func (r *runReader) key() (sortVal, error) {271- rank, err := r.r.ReadByte()179+func readKey(r *bufio.Reader) (sortVal, error) {180+ rank, err := r.ReadByte()272181 if err != nil {273182 return sortVal{}, err274183 }@@ -276,17 +185,17 @@ func (r *runReader) key() (sortVal, error) {
276185 switch rank {277186 case 0, 2:278187 var b [8]byte279- if _, err := io.ReadFull(r.r, b[:]); err != nil {188+ if _, err := io.ReadFull(r, b[:]); err != nil {280189 return k, err281190 }282191 k.num = math.Float64frombits(binary.LittleEndian.Uint64(b[:]))283192 case 1, 3:284- n, err := binary.ReadUvarint(r.r)193+ n, err := binary.ReadUvarint(r)285194 if err != nil {286195 return k, err287196 }288197 b := make([]byte, n)289- if _, err := io.ReadFull(r.r, b); err != nil {198+ if _, err := io.ReadFull(r, b); err != nil {290199 return k, err291200 }292201 k.str = string(b)@@ -294,42 +203,6 @@ func (r *runReader) key() (sortVal, error) {
294203 return k, nil295204 }296205 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-333206 // rowFile is a view's rows as a file of row numbers in its order.334207 type rowFile struct {335208 f *os.File@@ -37,6 +37,31 @@ type Sorter interface {
3737 SortedNums(sheet string, r Rect) (Sorted, *Value)3838 }3939 40+// orderFuncs are the functions that read their numbers in order: from41+// every argument (true: MEDIAN, MODE) or from the first, the rest being42+// what to pick.43+var orderFuncs = map[string]bool{44+ "MEDIAN": true, "MODE": true, "MODE.SNGL": true, "LARGE": false, "SMALL": false,45+ "PERCENTILE": false, "PERCENTILE.INC": false, "PERCENTILE.EXC": false,46+ "QUARTILE": false, "QUARTILE.INC": false, "QUARTILE.EXC": false,47+}48+49+// sortsSource reports whether c is an order statistic of one range of a50+// source, its other arguments single values: the Book sorts that range51+// on disk, so it streams.52+func sortsSource(c StreamCall) bool {53+ all, ok := orderFuncs[c.Fn]54+ if !ok || len(c.Args) == 0 || c.Args[0].Sheet == "" || all && len(c.Args) > 1 {55+ return false56+ }57+ for _, a := range c.Args[1:] {58+ if a.Sheet != "" {59+ return false60+ }61+ }62+ return true63+}64+4065 // held is numbers held in memory, sorted.4166 type held struct {4267 x []posNum@@ -18,9 +18,11 @@ import (
1818 // Until it is known the cell shows Loading…, as a JEV question does.1919 //2020 // The functions in streaming read any size in one pass and hold only21-// what they compute; any other reads what its ranges hold, so it is22-// given a source range only while that holds no more cells than the23-// budget (max-cells), and past it answers #VALUE! saying why.21+// what they compute, and an order statistic of one range (MEDIAN,22+// PERCENTILE; order.go) has the Book sort its numbers on disk; any other23+// reads what its ranges hold, so it is given a source range only while24+// that holds no more cells than the budget (max-cells), and past it25+// answers #VALUE! saying why.2426 2527 // StreamCall is a call of a function some of whose arguments are ranges2628 // of paged sheets. It isn't comparable: Key identifies it by content.@@ -67,24 +69,27 @@ func (c StreamCall) Key() string {
6769 6870 // streaming are the functions that read a source range of any size in6971 // one pass, holding only what they compute: aggregates, criteria70-// functions and lookups.72+// functions, RANK and lookups.7173 var streaming = map[string]bool{7274 "SUM": true, "AVERAGE": true, "COUNT": true, "COUNTA": true, "MIN": true, "MAX": true, "PRODUCT": true,7375 "COUNTIF": true, "COUNTIFS": true, "SUMIF": true, "SUMIFS": true, "AVERAGEIF": true, "AVERAGEIFS": true,74- "COUNTBLANK": true, "SUMPRODUCT": true,76+ "COUNTBLANK": true, "SUMPRODUCT": true, "RANK": true, "RANK.EQ": true,7577 "MATCH": true, "XLOOKUP": true, "VLOOKUP": true, "HLOOKUP": true, "INDEX": true, "ROWS": true, "COLUMNS": true,7678 }7779 7880 // Streams reports whether the function named name reads a source of79-// any size (see streaming).80-func Streams(name string) bool { return streaming[name] }81+// any size (see streaming), or one range of it (orderFuncs).82+func Streams(name string) bool {83+ _, sorts := orderFuncs[name]84+ return streaming[name] || sorts85+}8186 8287 // StreamingFuncs lists the functions that read a source of any size, in8388 // alphabetical order, for the docs and messages.8489 func StreamingFuncs() []string {8590 var out []string8691 for _, f := range Funcs() {87- if streaming[f.Name] {92+ if Streams(f.Name) {8893 out = append(out, f.Name)8994 }9095 }@@ -170,7 +175,7 @@ func EvalStream(c StreamCall, book Book, budget int) StreamAnswer {
170175 // overBudget says why c can't be computed, when it names a function171176 // that doesn't stream over more cells of its sources than budget.172177 func overBudget(c StreamCall, book Book, budget int) string {173- if streaming[c.Fn] {178+ if streaming[c.Fn] || sortsSource(c) {174179 return ""175180 }176181 cells, most := 0, 0@@ -191,7 +196,7 @@ func overBudget(c StreamCall, book Book, budget int) string {
191196 if cells <= budget {192197 return ""193198 }194- return fmt.Sprintf("Raise max-cells to %s to %s %s. It is %s, and %s holds what it reads; SUMIFS, MEDIAN, XLOOKUP and the like stream a source of any size.",199+ return fmt.Sprintf("Raise max-cells to %s for %s over %s. It is %s, and %s holds what it reads; SUMIFS, MEDIAN, XLOOKUP and the like stream a source of any size.",195200 grouped(cells), c.Fn, biggest, grouped(budget), c.Fn)196201 }197202 @@ -1,8 +1,10 @@
11 package paged22 33 import (4+ "cmp"45 "fmt"56 7+ "github.com/FelineStateMachine/012/internal/fileio"68 "github.com/FelineStateMachine/012/internal/functions"79 "github.com/FelineStateMachine/012/internal/sheet"810 )@@ -40,6 +42,73 @@ func (b *book) rangeValues(name string, r sheet.Rect, budget int) sheet.SourceAn
4042 return sheet.SourceAnswer{A: a}4143 }4244 45+// SortedNums sorts the numbers of r on disk (functions.Sorter),46+// streaming the source's rows: what isn't a number is skipped, as SUM47+// skips it in a range, and the first error is the answer.48+func (b *book) SortedNums(name string, r sheet.Rect) (functions.Sorted, *sheet.Value) {49+ s, ok := b.src(name)50+ if !ok {51+ return nil, refErr()52+ }53+ ns := fileio.NewNumberSort(s.dir)54+ part, ok := s.clip(r)55+ if ok && part.From.Row == 0 { // the header: names, not numbers56+ part.From.Row = 157+ }58+ if ok && part.From.Row <= part.To.Row {59+ if e := b.sortRows(s, part, ns); e != nil {60+ ns.Close()61+ return nil, e62+ }63+ }64+ sorted, err := ns.Sorted()65+ if err != nil {66+ b.fail(err)67+ return nil, refErr()68+ }69+ return sorted, nil70+}71+72+// sortRows adds the numbers of the rows of part to ns, numbered in the73+// order they're read, returning the first error among them, or #REF!74+// when reading or spilling failed.75+func (b *book) sortRows(s snapshot, part sheet.Rect, ns *fileio.NumberSort) *sheet.Value {76+ cols := make([]int, 0, part.To.Col-part.From.Col+1)77+ for c := part.From.Col; c <= part.To.Col; c++ {78+ cols = append(cols, c)79+ }80+ last := int64(part.To.Row - 1)81+ var e *sheet.Value82+ var pos int6483+ var failed error84+ err := s.h.Scan(b.ctx, int64(part.From.Row-1), cols, func(row int64, vals []sheet.LiveCell) bool {85+ for _, v := range vals {86+ switch v.V.Kind {87+ case sheet.Error:88+ e = &v.V89+ return false90+ case sheet.Number:91+ if failed = ns.Add(v.V.Num, pos); failed != nil {92+ return false93+ }94+ }95+ pos++96+ }97+ return row < last98+ })99+ if err = cmp.Or(err, failed); err != nil {100+ b.fail(err)101+ return refErr()102+ }103+ return e104+}105+106+// refErr is a #REF! of its own, for an answer to point at.107+func refErr() *sheet.Value {108+ v := sheet.ErrRef109+ return &v110+}111+43112 // after is the cell after a in r, row by row.44113 func after(r sheet.Rect, a sheet.Addr) sheet.Addr {45114 if a.Col < r.To.Col {@@ -192,7 +192,7 @@ func (h *Host) Answer(q sheet.SourceQuestion) (sheet.SourceAnswer, bool) {
192192 case s.h == nil:193193 return sheet.SourceAnswer{}, false // opening: asked again once it's open194194 }195- j.srcs[key(name)] = snapshot{name: s.name, h: s.h, shape: s.shape, gen: s.gen}195+ j.srcs[key(name)] = snapshot{name: s.name, h: s.h, shape: s.shape, gen: s.gen, dir: s.spec.TempDir}196196 }197197 h.asked[k] = true198198 h.fresh = append(h.fresh, j)@@ -172,14 +172,14 @@ func TestPastTheBudget(t *testing.T) {
172172 w, h := linked(t, name, 100)173173 s := w.Sheet(0)174174 med, sum := sheet.Addr{Col: 10}, sheet.Addr{Col: 10, Row: 1}175- s.Set(med, "=MEDIAN(sales[amount])")175+ s.Set(med, "=STDEV(sales[amount])")176176 s.Set(sum, "=SUM(sales[amount])")177177 h.Settle(context.Background(), w, Links(w, same, ""))178178 if v := s.Value(med); v != sheet.ErrValue {179- t.Errorf("MEDIAN past the budget: %v, want #VALUE!", v)179+ t.Errorf("STDEV past the budget: %v, want #VALUE!", v)180180 }181181 if why := s.ExplainError(med); !strings.Contains(why, "max-cells") {182- t.Errorf("MEDIAN past the budget explained as %q", why)182+ t.Errorf("STDEV past the budget explained as %q", why)183183 }184184 if v := s.Value(sum); v.Kind != sheet.Number {185185 t.Errorf("SUM streams past the budget: %v", v)@@ -0,0 +1,97 @@
1+package paged2+3+import (4+ "context"5+ "database/sql"6+ "path/filepath"7+ "strings"8+ "testing"9+10+ "github.com/FelineStateMachine/012/internal/sheet"11+)12+13+// orderFormulas are order statistics over the source as {T}, each the14+// same as over the table ref holding the same rows, though the source15+// holds more cells than the budget: its column is sorted on disk.16+var orderFormulas = []string{17+ "=MEDIAN({T}[amount])", "=MEDIAN({T}[id])", "=MEDIAN({T}[day])", "=MEDIAN({T}[cat])",18+ "=MODE({T}[amount])", "=MODE.SNGL({T}[day])", "=MODE({T}[id])",19+ "=PERCENTILE({T}[amount],0.3)", "=PERCENTILE.INC({T}[amount],0.999)", "=PERCENTILE({T}[amount],2)",20+ "=PERCENTILE.EXC({T}[amount],0.1)", "=PERCENTILE.EXC({T}[amount],0.0001)",21+ "=QUARTILE({T}[amount],1)", "=QUARTILE({T}[amount],3)", "=QUARTILE.INC({T}[id],4)",22+ "=QUARTILE.EXC({T}[amount],2)", "=QUARTILE.EXC({T}[amount],4)",23+ "=LARGE({T}[amount],10)", "=SMALL({T}[amount],1)", "=LARGE({T}[id],1)", "=SMALL({T}[amount],100000)",24+ "=RANK(50.5,{T}[amount])", "=RANK.EQ(17,{T}[amount],1)", "=RANK(0.25,{T}[amount])",25+ "=MEDIAN({T}[[id]:[amount]])",26+}27+28+// checkSame sets each of formulas over the source sales and over the29+// table ref, and checks the two agree.30+func checkSame(t *testing.T, w *sheet.Workbook, h *Host, formulas []string) {31+ t.Helper()32+ s := w.Sheet(0)33+ for i, f := range formulas {34+ s.Set(sheet.Addr{Col: 10, Row: i}, strings.ReplaceAll(f, "{T}", "sales"))35+ s.Set(sheet.Addr{Col: 11, Row: i}, strings.ReplaceAll(f, "{T}", "ref"))36+ }37+ h.Settle(context.Background(), w, nil)38+ for i, f := range formulas {39+ got, want := s.Value(sheet.Addr{Col: 10, Row: i}), s.Value(sheet.Addr{Col: 11, Row: i})40+ if testing.Verbose() {41+ t.Logf("%s: %v", f, got)42+ }43+ if got != want {44+ t.Errorf("%s: %v over the source, %v over the sheet (%s)", f, got, want, s.ExplainError(sheet.Addr{Col: 10, Row: i}))45+ }46+ }47+}48+49+func TestOrderOverAParquetSource(t *testing.T) {50+ name := filepath.Join(t.TempDir(), "sales.parquet")51+ writeSales(t, name, 5000, 0)52+ w, h := linked(t, name, 1000)53+ checkSame(t, w, h, orderFormulas)54+}55+56+func TestOrderOverASQLiteSource(t *testing.T) {57+ name := filepath.Join(t.TempDir(), "sales.sqlite")58+ db, err := sql.Open("sqlite", name)59+ if err != nil {60+ t.Fatal(err)61+ }62+ if _, err := db.Exec("CREATE TABLE sales (id INTEGER, cat TEXT, amount REAL, day DATE)"); err != nil {63+ t.Fatal(err)64+ }65+ for i := range 3000 {66+ var amount any67+ if i%11 != 7 {68+ amount = float64((i*37)%101) - 20.569+ }70+ _, err := db.Exec("INSERT INTO sales VALUES (?, ?, ?, ?)", i, cats[i%len(cats)], amount,71+ "2026-01-"+string(rune('1'+i%9))+"0")72+ if err != nil {73+ t.Fatal(err)74+ }75+ }76+ db.Close()77+ w, h := linked(t, name, 1000)78+ checkSame(t, w, h, orderFormulas)79+}80+81+// TestOrderAnswersKept checks an order statistic is answered once, then82+// kept as the other answers are.83+func TestOrderAnswersKept(t *testing.T) {84+ name := filepath.Join(t.TempDir(), "sales.parquet")85+ writeSales(t, name, 500, 0)86+ w, h := linked(t, name, 100)87+ s := w.Sheet(0)88+ s.Set(sheet.Addr{Col: 10}, "=MEDIAN(sales[amount])")89+ h.Settle(context.Background(), w, nil)90+ s.Set(sheet.Addr{Col: 10, Row: 1}, "=MEDIAN(sales[amount])+1")91+ if jobs := h.Jobs(); len(jobs) != 0 {92+ t.Errorf("asked again: %v", jobs)93+ }94+ if v := s.Value(sheet.Addr{Col: 10, Row: 1}); v.Kind != sheet.Number {95+ t.Errorf("the kept answer: %v", v)96+ }97+}@@ -28,7 +28,12 @@ var bigFormulas = []struct{ name, formula string }{
2828 {"SUMPRODUCT", "=SUMPRODUCT(sales[amount],sales[id])"},2929 {"XLOOKUP-last", "=XLOOKUP(9999999,sales[id],sales[amount])"},3030 {"MATCH-middle", "=MATCH(5000000,sales[id],0)"},31- {"MEDIAN-held", "=MEDIAN(sales[amount])"},31+ {"MEDIAN", "=MEDIAN(sales[amount])"},32+ {"PERCENTILE", "=PERCENTILE(sales[amount],0.9)"},33+ {"MODE", "=MODE(sales[amount])"},34+ {"LARGE", "=LARGE(sales[amount],1000)"},35+ {"RANK", "=RANK(5000,sales[amount])"},36+ {"STDEV-held", "=STDEV(sales[amount])"},3237 }3338 3439 func BenchmarkSourceFormulas(b *testing.B) {