- commit
- 470224b1e63d2a3172639aeb89416fc6c8efdce7
- parent
- 35de47e13dfaab8a66c9939d6cb6a37d2fa60f3e
- author
- Dami <dami.etoile@gmail.com>
- kind
- feat risk score 2.3 of 3, Jev's estimate
| M | internal/ | 4 | + |
| M | internal/ | 10 | +- |
| M | internal/ | 19 | +++ |
| M | internal/ | 15 | ++- |
| M | internal/ | 3 | + |
| A | internal/ | 210 | ++++++++++++++++++++++++++++ |
| A | internal/ | 225 | ++++++++++++++++++++++++++++++ |
@@ -152,6 +152,8 @@ func eval(n Node, get lookup) Value {
152152 var v Value153153 if get.lift > 0 && f.arrays != takesArrays {154154 v = liftCall(f, n.Args, get)155+ } else if sv, ok := get.streamed(f, n.Args); ok {156+ v = sv155157 } else {156158 want := get.wantArr157159 get.wantArr = f.arrays != liftScalar@@ -183,6 +185,8 @@ func evalOther(n Node, get lookup) Value {
183185 return get.invoke(n)184186 case liftArg:185187 return n.one(get)188+ case constArg:189+ return n.v186190 }187191 return value.ErrValue188192 }@@ -227,8 +227,14 @@ func xlookup(args []Node, get lookup) Value {
227227 return value.ErrValue228228 }229229 q := lineSeq(look, get, look.cols == 1)230- order := q.order(search < 0)231- best := -1230+ best, streamed := -1, false231+ if get.stream && q.sparse {232+ best, streamed = q.seekStream(key, mode, search < 0)233+ }234+ var order []int235+ if !streamed {236+ best, order = -1, q.order(search < 0)237+ }232238 for _, i := range order {233239 v := look.at(i)234240 if lookupEqual(key, v, mode == 2) {@@ -191,6 +191,9 @@ func sumIfTerms(args []Node, get lookup, add func(float64)) *Value {
191191 return &value.ErrValue192192 }193193 }194+ if streamable(get, rng, sum) {195+ return sumStream(get, sum, []matrix{rng}, []criterion{newCriterion(cv)}, add)196+ }194197 return sumMasked(sum, applyCriteria(get, []matrix{rng}, []criterion{newCriterion(cv)}, sum), add)195198 }196199 @@ -204,6 +207,8 @@ func sumIfsTerms(args []Node, get lookup, add func(float64)) *Value {
204207 return err205208 case len(ms) == 0 || ms[0].rows != sum.rows || ms[0].cols != sum.cols:206209 return &value.ErrValue210+ case streamable(get, append(ms, sum)...):211+ return sumStream(get, sum, ms, cs, add)207212 }208213 return sumMasked(sum, applyCriteria(get, ms, cs, sum), add)209214 }@@ -218,6 +223,9 @@ func averageIf(args []Node, get lookup) Value {
218223 if len(args) > 2 {219224 avg = matrixArg(args[2], get).resized(rng.rows, rng.cols, get)220225 }226+ if streamable(get, rng, avg) && avg.rows == rng.rows && avg.cols == rng.cols {227+ return averageStream(get, avg, []matrix{rng}, []criterion{newCriterion(cv)})228+ }221229 return averageMasked(avg, applyCriteria(get, []matrix{rng}, []criterion{newCriterion(cv)}, avg))222230 }223231 @@ -229,6 +237,8 @@ func averageIfs(args []Node, get lookup) Value {
229237 return *err230238 case len(ms) == 0 || ms[0].rows != avg.rows || ms[0].cols != avg.cols:231239 return value.ErrValue240+ case streamable(get, append(ms, avg)...):241+ return averageStream(get, avg, ms, cs)232242 }233243 return averageMasked(avg, applyCriteria(get, ms, cs, avg))234244 }@@ -238,6 +248,9 @@ func countIfs(args []Node, get lookup) Value {
238248 if err != nil {239249 return *err240250 }251+ if streamable(get, ms...) {252+ return countStream(get, ms, cs)253+ }241254 k := applyCriteria(get, ms, cs)242255 n := len(k.pass)243256 if k.tailPass {@@ -248,6 +261,9 @@ func countIfs(args []Node, get lookup) Value {
248261 249262 func countBlank(args []Node, get lookup) Value {250263 m := matrixArg(args[0], get)264+ if streamable(get, m) {265+ return blankStream(get, m)266+ }251267 isBlank := func(v Value) bool { return v.Kind == value.Empty || v.Kind == value.Text && v.Str == "" }252268 cells := cellsOf(get, m)253269 n := 0@@ -274,6 +290,9 @@ func sumProductTerms(args []Node, get lookup, term func(fs []float64)) *Value {
274290 return &value.ErrValue275291 }276292 }293+ if streamable(get, ms...) {294+ return productStream(get, ms, term)295+ }277296 cells := cellsOf(get, ms...)278297 fs := make([]float64, len(ms))279298 for _, pc := range cells {@@ -40,6 +40,12 @@ type Book interface {
4040 // value is an error when there is none to give: Pending while it is4141 // on its way, ErrNoRemote when nothing answers them.4242 Ask(call RemoteCall) (RemoteAnswer, Value)43+ // Paged reports whether the sheet is a paged source (stream.go),44+ // whose ranges a function is given through Stream.45+ Paged(sheet string) bool46+ // Stream answers a call over paged sources, with an array when it47+ // computes several values: Pending while the answer is on its way.48+ Stream(call StreamCall) (Value, *Array)4349 }4450 4551 // Reader is what a function is given to read the cells its arguments@@ -49,8 +55,13 @@ type Reader struct {
4955 book Book5056 depth *int // how deeply evaluation is nested; see the engine's evaluate.go5157 dense bool // read every address of a range; see NewReader52- bufs []*scanBuf53- level int // how many of bufs are in use by range reads in progress58+ // stream is set on a Reader computing a StreamCall, over a Book that59+ // streams paged sources: functions over aligned ranges read them a60+ // window of rows at a time (streamed.go) rather than gathering the61+ // positions of their cells.62+ stream bool63+ bufs []*scanBuf64+ level int // how many of bufs are in use by range reads in progress5465 5566 evalState5667 // arena holds the arrays and LAMBDAs of the formulas being evaluated@@ -61,6 +61,9 @@ func (b *fakeBook) Bounds(string, Rect) (Rect, bool, bool) { return Rect{},
6161 func (b *fakeBook) RangeAgg(string, Rect) (Agg, *Value, bool) { return Agg{}, nil, false }6262 func (b *fakeBook) Fold(_ string, _ Rect, s Agg) (Agg, *Value) { return s, nil }6363 func (b *fakeBook) Ask(RemoteCall) (RemoteAnswer, Value) { return RemoteAnswer{}, ErrNoRemote }64+func (b *fakeBook) Paged(string) bool { return false }65+func (b *fakeBook) Stream(StreamCall) (Value, *Array) { return value.ErrValue, nil }66+6467 func newFakeReader(cells map[Addr]Value) (*Reader, *fakeBook) {6568 b := &fakeBook{cells: cells}6669 depth := 0@@ -0,0 +1,210 @@
1+package functions2+3+import (4+ "fmt"5+ "strings"6+7+ "github.com/FelineStateMachine/012/internal/formula"8+ "github.com/FelineStateMachine/012/internal/value"9+)10+11+// Calls over paged sources. A linked source (the engine's paged sheets:12+// a Parquet file or a SQLite table read in place) is too big to hold,13+// so a function given one of its ranges isn't evaluated where the14+// formula is: the Reader describes the call, its source ranges and the15+// values of its other arguments, as a StreamCall and asks the Book for16+// the answer, which is worked out in the background by EvalStream over17+// a Book that streams the source, and kept until the file changes.18+// Until it is known the cell shows Loading…, as a JEV question does.19+//20+// 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.24+25+// StreamCall is a call of a function some of whose arguments are ranges26+// of paged sheets. It isn't comparable: Key identifies it by content.27+type StreamCall struct {28+ Fn string29+ // Dec is set when the call is computed in decimal (decimal.go).30+ Dec bool31+ Args []StreamArg32+}33+34+// StreamArg is an argument of a StreamCall: a range of a paged sheet35+// (Sheet set, as the engine names the source), or a value.36+type StreamArg struct {37+ Sheet string38+ R Rect39+ V Value40+}41+42+// StreamAnswer is what a StreamCall computes: a value, or an array of43+// them, and for an error that needs one, why.44+type StreamAnswer struct {45+ V Value46+ A *Array47+ Why string48+}49+50+// Key identifies the call by its content.51+func (c StreamCall) Key() string {52+ var b strings.Builder53+ b.WriteString(c.Fn)54+ if c.Dec {55+ b.WriteString("#dec")56+ }57+ for _, a := range c.Args {58+ b.WriteByte('\x1f')59+ if a.Sheet != "" {60+ fmt.Fprintf(&b, "r%s\x1e%d:%d:%d:%d", a.Sheet, a.R.From.Col, a.R.From.Row, a.R.To.Col, a.R.To.Row)61+ continue62+ }63+ fmt.Fprintf(&b, "v%d\x1e%v\x1e%s", a.V.Kind, a.V.Num, a.V.Str)64+ }65+ return b.String()66+}67+68+// streaming are the functions that read a source range of any size in69+// one pass, holding only what they compute: aggregates, criteria70+// functions and lookups.71+var streaming = map[string]bool{72+ "SUM": true, "AVERAGE": true, "COUNT": true, "COUNTA": true, "MIN": true, "MAX": true, "PRODUCT": true,73+ "COUNTIF": true, "COUNTIFS": true, "SUMIF": true, "SUMIFS": true, "AVERAGEIF": true, "AVERAGEIFS": true,74+ "COUNTBLANK": true, "SUMPRODUCT": true,75+ "MATCH": true, "XLOOKUP": true, "VLOOKUP": true, "HLOOKUP": true, "INDEX": true, "ROWS": true, "COLUMNS": true,76+}77+78+// 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+82+// StreamingFuncs lists the functions that read a source of any size, in83+// alphabetical order, for the docs and messages.84+func StreamingFuncs() []string {85+ var out []string86+ for _, f := range Funcs() {87+ if streaming[f.Name] {88+ out = append(out, f.Name)89+ }90+ }91+ return out92+}93+94+// streamed asks the Book for the call of f with args when one of its95+// arguments is a range of a paged sheet, reporting whether it asked. A96+// call whose other arguments aren't single values (a range of another97+// sheet, an array) is evaluated as usual, reading the source through98+// the Book's cells, which it holds only up to the budget.99+func (rd *Reader) streamed(f *FuncDef, args []Node) (Value, bool) {100+ if rd.stream || f.remote != nil || f.binds != formula.BindNone || !rd.pagedArg(args) {101+ return Value{}, false102+ }103+ call := StreamCall{Fn: f.Name, Dec: funcs[f.Name] != f, Args: make([]StreamArg, len(args))}104+ for i, a := range args {105+ switch r := rd.refOf(a).(type) {106+ case formula.Range:107+ if r.Sheet == "" || !rd.book.Paged(r.Sheet) {108+ return Value{}, false109+ }110+ call.Args[i] = StreamArg{Sheet: r.Sheet, R: r.Rect}111+ continue112+ }113+ v := eval1(a, rd)114+ if v.Kind == value.Array {115+ return Value{}, false116+ }117+ call.Args[i] = StreamArg{V: v}118+ }119+ v, a := rd.book.Stream(call)120+ if a != nil {121+ return rd.arrayValue(a), true122+ }123+ return v, true124+}125+126+// pagedArg reports whether an argument is a range of a paged sheet.127+func (rd *Reader) pagedArg(args []Node) bool {128+ for _, a := range args {129+ if r, ok := rd.refOf(a).(formula.Range); ok && r.Sheet != "" && rd.book.Paged(r.Sheet) {130+ return true131+ }132+ }133+ return false134+}135+136+// constArg is a StreamCall's value argument, as a node.137+type constArg struct{ v Value }138+139+// EvalStream computes c over book, a Book that reads the paged sheets140+// its ranges name by streaming them: in the background, where reading141+// takes as long as the source is big. A function not in streaming is142+// given the source's ranges only up to budget cells.143+func EvalStream(c StreamCall, book Book, budget int) StreamAnswer {144+ f, ok := LookupFunc(c.Fn)145+ if !ok {146+ return StreamAnswer{V: value.ErrName}147+ }148+ if twin, ok := decFuncs()[c.Fn]; ok && c.Dec {149+ f = twin150+ }151+ if why := overBudget(c, book, budget); why != "" {152+ return StreamAnswer{V: value.ErrValue, Why: why}153+ }154+ args := make([]Node, len(c.Args))155+ const fixed = formula.AbsCol | formula.AbsRow156+ for i, a := range c.Args {157+ if a.Sheet != "" {158+ args[i] = formula.Range{Rect: a.R, Sheet: a.Sheet, Abs: [2]formula.Abs{fixed, fixed}}159+ } else {160+ args[i] = constArg{a.V}161+ }162+ }163+ depth := 0164+ rd := NewReader(book, &depth, false)165+ rd.stream = true166+ v := EvalCell(formula.Call{Fn: f, Args: args}, rd, Addr{})167+ return StreamAnswer{V: v, A: rd.Spilled()}168+}169+170+// overBudget says why c can't be computed, when it names a function171+// that doesn't stream over more cells of its sources than budget.172+func overBudget(c StreamCall, book Book, budget int) string {173+ if streaming[c.Fn] {174+ return ""175+ }176+ cells, most := 0, 0177+ biggest := ""178+ for _, a := range c.Args {179+ if a.Sheet == "" {180+ continue181+ }182+ b, any, _ := book.Bounds(a.Sheet, a.R)183+ if !any {184+ continue185+ }186+ n := (b.To.Row - b.From.Row + 1) * (b.To.Col - b.From.Col + 1)187+ if cells += n; n > most {188+ most, biggest = n, a.Sheet189+ }190+ }191+ if cells <= budget {192+ return ""193+ }194+ return fmt.Sprintf("%s holds what it reads: %s cells of %s, more than max-cells (%s). SUM, COUNTIFS, XLOOKUP and the like read a source of any size",195+ c.Fn, grouped(cells), biggest, grouped(budget))196+}197+198+// grouped is n with thousands separators: 12,000,000.199+func grouped(n int) string {200+ s := fmt.Sprint(n)201+ for i := len(s) - 3; i > 0 && s[i-1] != '-'; i -= 3 {202+ s = s[:i] + "," + s[i:]203+ }204+ return s205+}206+207+// Const is a node that evaluates to v: what the engine binds a name to208+// while what it names can't be read yet (a source being opened shows209+// Loading…).210+func Const(v Value) Node { return constArg{v} }@@ -0,0 +1,225 @@
1+package functions2+3+import "github.com/FelineStateMachine/012/internal/value"4+5+// Functions over aligned ranges computing a StreamCall (stream.go). On6+// a sheet they gather the positions of the cells their ranges hold and7+// visit those (masked.go), which over a source of ten million rows8+// would hold ten million positions. Computing a StreamCall they read9+// their ranges instead a window of rows at a time, every cell of the10+// window's rows, blank or not, testing and adding as they go, so they11+// hold one window whatever the source's size. Blank cells meet12+// criteria or add nothing as they would when accounted for at once, so13+// the results are the same.14+15+// streamRows is how many rows of each range a window holds.16+const streamRows = 204817+18+// streamable reports whether same-sized matrices are read a window at19+// a time: computing a StreamCall, every one a range of a sheet that20+// exists.21+func streamable(get lookup, ms ...matrix) bool {22+ if !get.stream {23+ return false24+ }25+ for _, m := range ms {26+ if !m.ref || m.blank.Kind == value.Error {27+ return false28+ }29+ }30+ return true31+}32+33+// windows calls fn with each window of the rows of same-sized ranges34+// that may hold data, the values of each range's window row by row (the35+// slices are reused), until fn returns false. It returns how many cells36+// of each range the windows cover; the rest are blank in all of them.37+func windows(get lookup, ms []matrix, fn func(vals [][]Value) bool) int {38+ rows, cols := 0, ms[0].cols39+ for _, m := range ms {40+ rows = max(rows, min(m.dataRows, m.rows))41+ }42+ bufs := make([][]Value, len(ms))43+ for r0 := 0; r0 < rows; r0 += streamRows {44+ n := min(streamRows, rows-r0)45+ for i, m := range ms {46+ bufs[i] = m.window(get, r0, n, bufs[i])47+ }48+ if !fn(bufs) {49+ break50+ }51+ }52+ return rows * cols53+}54+55+// window reads n rows of the range m from its row r0 into buf, every56+// cell of them, blanks as blank values.57+func (m matrix) window(get lookup, r0, n int, buf []Value) []Value {58+ size := n * m.cols59+ if cap(buf) < size {60+ buf = make([]Value, size)61+ }62+ buf = buf[:size]63+ clear(buf)64+ top := m.origin.Row + r065+ r := Rect{From: Addr{Col: m.origin.Col, Row: top}, To: Addr{Col: m.origin.Col + m.cols - 1, Row: top + n - 1}}66+ get.cells(m.sheet, r, func(a Addr, v Value) bool {67+ buf[(a.Row-top)*m.cols+a.Col-m.origin.Col] = v68+ return true69+ })70+ return buf71+}72+73+// passAll reports whether the cell at i of each window meets its74+// criterion.75+func passAll(vals [][]Value, cs []criterion, i int) bool {76+ for j, c := range cs {77+ if !c.test(vals[j][i]) {78+ return false79+ }80+ }81+ return true82+}83+84+// blanksPass reports whether blank cells meet every criterion.85+func blanksPass(cs []criterion) bool {86+ for _, c := range cs {87+ if !c.test(Value{}) {88+ return false89+ }90+ }91+ return true92+}93+94+// countStream is COUNTIFS over windows.95+func countStream(get lookup, ms []matrix, cs []criterion) Value {96+ n := 097+ covered := windows(get, ms, func(vals [][]Value) bool {98+ for i := range vals[0] {99+ if passAll(vals, cs, i) {100+ n++101+ }102+ }103+ return true104+ })105+ if blanksPass(cs) {106+ n += ms[0].size() - covered107+ }108+ return num(float64(n))109+}110+111+// sumStream calls add with the numbers of val where the criteria on ms112+// pass, over windows, or returns the first error there.113+func sumStream(get lookup, val matrix, ms []matrix, cs []criterion, add func(float64)) *Value {114+ var err *Value115+ windows(get, append([]matrix{val}, ms...), func(vals [][]Value) bool {116+ for i, v := range vals[0] {117+ if !passAll(vals[1:], cs, i) {118+ continue119+ }120+ switch v.Kind {121+ case value.Error:122+ err = &v123+ return false124+ case value.Number:125+ add(v.Num)126+ }127+ }128+ return true129+ })130+ return err131+}132+133+// averageStream is AVERAGEIFS over windows.134+func averageStream(get lookup, val matrix, ms []matrix, cs []criterion) Value {135+ sum, n := 0.0, 0136+ if e := sumStream(get, val, ms, cs, func(f float64) { sum += f; n++ }); e != nil {137+ return *e138+ }139+ if n == 0 {140+ return value.ErrDiv0141+ }142+ return num(sum / float64(n))143+}144+145+// blankStream is COUNTBLANK over windows.146+func blankStream(get lookup, m matrix) Value {147+ n := 0148+ covered := windows(get, []matrix{m}, func(vals [][]Value) bool {149+ for _, v := range vals[0] {150+ if v.Kind == value.Empty || v.Kind == value.Text && v.Str == "" {151+ n++152+ }153+ }154+ return true155+ })156+ return num(float64(n + m.size() - covered))157+}158+159+// productStream calls term with the factors of each entry of160+// same-sized ranges, over windows, as sumProductTerms does.161+func productStream(get lookup, ms []matrix, term func(fs []float64)) *Value {162+ var err *Value163+ fs := make([]float64, len(ms))164+ windows(get, ms, func(vals [][]Value) bool {165+ for i := range vals[0] {166+ for j := range ms {167+ switch v := vals[j][i]; v.Kind {168+ case value.Error:169+ err = &v170+ return false171+ case value.Number, value.Bool:172+ fs[j] = v.Num173+ default:174+ fs[j] = 0175+ }176+ }177+ term(fs)178+ }179+ return true180+ })181+ return err182+}183+184+// seekStream is XLOOKUP's search of a sparse line computing a185+// StreamCall: one pass over the cells it holds, forwards whichever way186+// the search goes, keeping what the search in order would find (the187+// first exact match, or the last searching backwards; else the closest188+// value, the first or last of equals) without listing the entries. ok189+// is false when a blank would match the key, which the search in order190+// handles.191+func (q *seq) seekStream(key Value, mode int, reverse bool) (int, bool) {192+ wild := mode == 2193+ if lookupEqual(key, q.blank, wild) {194+ return 0, false195+ }196+ exact, best := -1, -1197+ var bestV Value198+ c := q.walk()199+ defer c.close()200+ for a, ok := c.next(); ok; a, ok = c.next() {201+ i, v := q.index(a), q.value(a)202+ switch {203+ case lookupEqual(key, v, wild):204+ if !reverse {205+ return i, true206+ }207+ exact = i208+ case exact >= 0 || mode == 0 || wild || v.Kind != key.Kind:209+ case compare(v, key)*mode > 0 && (best < 0 || closer(v, bestV, mode, reverse)):210+ best, bestV = i, v211+ }212+ }213+ if exact >= 0 {214+ return exact, true215+ }216+ return best, true217+}218+219+// closer reports whether v is closer to the key than the best so far,220+// in XLOOKUP's next smaller (-1) or next larger (1) mode: equals keep221+// the first met, or the last searching backwards.222+func closer(v, best Value, mode int, reverse bool) bool {223+ d := compare(v, best) * mode224+ return d < 0 || reverse && d == 0225+}