1package paged23import (4 "cmp"5 "context"6 "slices"78 "src.rest/cat/012/internal/functions"9 "src.rest/cat/012/internal/sheet"10)1112// book is the functions.Book a job computes a call through: the sources13// the call reads, each as a sheet named by the source's name, whose14// first row is its columns' names and the rest its rows. Ranges are15// streamed, a chunk at a time, through a cursor that goes on where the16// last chunk ended, so a function reading a column once reads the file17// once; a cell is fetched with its row, and the last chunk's cells are18// at hand, for lookups that walk a column and read what they pass.19type book struct {20 ctx context.Context21 srcs map[string]snapshot22 err error // the first a source gave: the answer is then that2324 cursors map[cursorKey]*cursor25 chunk []sheet.Addr // the last chunk Scan read, row by row26 chunkV []sheet.Value27 chunkOf string28 rows map[rowKey][]sheet.LiveCell // rows fetched, a few29}3031type cursorKey struct {32 src string33 c0, c1 int34}3536type rowKey struct {37 src string38 row int39}4041// maxRows bounds the rows fetched a book keeps.42const maxRows = 2564344func (b *book) close() {45 for _, c := range b.cursors {46 b.stopCursor(c)47 }48}4950func (b *book) src(name string) (snapshot, bool) {51 s, ok := b.srcs[key(name)]52 return s, ok53}5455// table is the source's table as a sheet holds it, header included.56func (s snapshot) table() sheet.Rect {57 return sheet.Rect{To: sheet.Addr{Col: len(s.shape.Cols) - 1, Row: s.shape.Rows}}58}5960// clip is the part of r the source's table holds.61func (s snapshot) clip(r sheet.Rect) (sheet.Rect, bool) {62 t := s.table()63 if len(s.shape.Cols) == 0 {64 return sheet.Rect{}, false65 }66 out := sheet.Rect{67 From: sheet.Addr{Col: max(r.From.Col, t.From.Col), Row: max(r.From.Row, t.From.Row)},68 To: sheet.Addr{Col: min(r.To.Col, t.To.Col), Row: min(r.To.Row, t.To.Row)},69 }70 return out, out.From.Col <= out.To.Col && out.From.Row <= out.To.Row71}7273func (s snapshot) header(col int) sheet.Value {74 return sheet.Value{Kind: sheet.Text, Str: s.shape.Cols[col]}75}7677// Cell is a cell's value: a column's name, a value of its row, or blank78// outside the table.79func (b *book) Cell(name string, a sheet.Addr) sheet.Value {80 s, ok := b.src(name)81 if !ok {82 return sheet.ErrRef83 }84 if !s.table().Contains(a) {85 return sheet.Value{}86 }87 if a.Row == 0 {88 return s.header(a.Col)89 }90 if b.chunkOf == key(name) {91 if i, found := slices.BinarySearchFunc(b.chunk, a, addrOrder); found {92 return b.chunkV[i]93 }94 }95 row, ok := b.fetch(s, a.Row-1)96 if !ok || a.Col >= len(row) {97 return sheet.Value{}98 }99 return row[a.Col].V100}101102func addrOrder(x, y sheet.Addr) int {103 return cmp.Or(cmp.Compare(x.Row, y.Row), cmp.Compare(x.Col, y.Col))104}105106// fetch is the source's row numbered row, every column.107func (b *book) fetch(s snapshot, row int) ([]sheet.LiveCell, bool) {108 k := rowKey{key(s.name), row}109 if r, ok := b.rows[k]; ok {110 return r, true111 }112 got, err := s.h.Fetch(b.ctx, []int64{int64(row)}, nil)113 if err != nil {114 b.fail(err)115 return nil, false116 }117 if b.rows == nil || len(b.rows) >= maxRows {118 b.rows = map[rowKey][]sheet.LiveCell{}119 }120 if len(got) == 0 {121 return nil, false122 }123 b.rows[k] = got[0]124 return got[0], true125}126127func (b *book) fail(err error) {128 if b.err == nil && b.ctx.Err() == nil {129 b.err = err130 }131}132133// Scan reads the cells of r holding something from from on, as134// functions.Book says: a chunk of a cursor streaming the rows.135func (b *book) Scan(name string, r sheet.Rect, from sheet.Addr, addrs []sheet.Addr, vals []sheet.Value) int {136 s, ok := b.src(name)137 if !ok {138 return -1139 }140 part, ok := s.clip(r)141 if !ok || from.Row > part.To.Row {142 return 0143 }144 if from.Row < part.From.Row || from.Row == part.From.Row && from.Col < part.From.Col {145 from = part.From146 }147 n := 0148 b.chunk, b.chunkV, b.chunkOf = b.chunk[:0], b.chunkV[:0], key(name)149 emit := func(a sheet.Addr, v sheet.Value) bool {150 if v.Kind == sheet.Empty {151 return true152 }153 addrs[n] = a154 if vals != nil {155 vals[n] = v156 }157 b.chunk, b.chunkV = append(b.chunk, a), append(b.chunkV, v)158 n++159 return n < len(addrs) && !(vals != nil && v.Kind == sheet.Error)160 }161 if from.Row == 0 {162 for c := from.Col; c <= part.To.Col; c++ {163 if !emit(sheet.Addr{Col: c, Row: 0}, s.header(c)) {164 return n165 }166 }167 from = sheet.Addr{Col: part.From.Col, Row: 1}168 }169 if from.Row > part.To.Row {170 return n171 }172 c := b.cursor(s, part, from)173 c.read(part, emit)174 if c.cur == nil && c.err != nil {175 b.fail(c.err) // the scan ended failing176 }177 return n178}179180// Bounds is the part of r the source's table holds.181func (b *book) Bounds(name string, r sheet.Rect) (sheet.Rect, bool, bool) {182 s, ok := b.src(name)183 if !ok {184 return sheet.Rect{}, false, false185 }186 part, any := s.clip(r)187 return part, any, true188}189190// RangeAgg shares nothing: each call reads its ranges once.191func (b *book) RangeAgg(string, sheet.Rect) (functions.Agg, *sheet.Value, bool) {192 return functions.Agg{}, nil, false193}194195// Fold adds the cells of r to agg, streaming the source's rows.196func (b *book) Fold(name string, r sheet.Rect, agg functions.Agg) (functions.Agg, *sheet.Value) {197 s, ok := b.src(name)198 if !ok {199 return agg, agg.Add(sheet.ErrRef, false)200 }201 part, ok := s.clip(r)202 if !ok {203 return agg, nil204 }205 var e *sheet.Value206 if part.From.Row == 0 {207 for c := part.From.Col; c <= part.To.Col && e == nil; c++ {208 e = agg.Add(s.header(c), false)209 }210 part.From.Row = 1211 }212 if e != nil || part.From.Row > part.To.Row {213 return agg, e214 }215 cols := make([]int, 0, part.To.Col-part.From.Col+1)216 for c := part.From.Col; c <= part.To.Col; c++ {217 cols = append(cols, c)218 }219 last := int64(part.To.Row - 1)220 err := s.h.Scan(b.ctx, int64(part.From.Row-1), cols, func(row int64, vals []sheet.LiveCell) bool {221 for _, v := range vals {222 if v.V.Kind != sheet.Empty {223 if e = agg.Add(v.V, false); e != nil {224 return false225 }226 }227 }228 return row < last229 })230 if err != nil {231 b.fail(err)232 }233 return agg, e234}235236// Ask has no answers: sources don't ask JEV.237func (b *book) Ask(functions.RemoteCall) (functions.RemoteAnswer, sheet.Value) {238 return functions.RemoteAnswer{}, functions.ErrNoRemote239}240241// Paged is false: a job reads its sources as sheets.242func (b *book) Paged(string) bool { return false }243244// Stream isn't asked: nothing is paged here.245func (b *book) Stream(functions.StreamCall) (sheet.Value, *functions.Array) {246 return sheet.ErrValue, nil247}