1package fileio23import (4 "bufio"5 "cmp"6 "encoding/binary"7 "errors"8 "io"9 "math"10 "os"11 "strings"1213 "src.rest/cat/012/internal/sheet"14)1516// A Parquet view's rows are a file of row numbers in the view's order,17// 8 bytes each, which a page reads with ReadAt: a filter writes the rows18// it lets through as it streams them, and a sort is an external merge19// 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.2223// sortVal is one key of a row as a sort compares it: its kind's rank24// (numbers, text, booleans, errors, then blanks), and its number or its25// text in lower case.26type sortVal struct {27 rank uint828 num float6429 str string30}3132// blankRank is a blank's, last in either direction.33const blankRank = 43435func sortValOf(v sheet.Value) sortVal {36 switch v.Kind {37 case sheet.Number:38 return sortVal{rank: 0, num: v.Num}39 case sheet.Text:40 return sortVal{rank: 1, str: strings.ToLower(v.Str)}41 case sheet.Bool:42 return sortVal{rank: 2, num: v.Num}43 case sheet.Error:44 return sortVal{rank: 3, str: strings.ToLower(v.Str)}45 }46 return sortVal{rank: blankRank}47}4849// sortRec is a row being sorted: its number and its keys.50type sortRec struct {51 row int6452 keys []sortVal53}5455// size is about what the record holds in memory.56func (r *sortRec) size() int {57 n := 48 + 32*len(r.keys)58 for _, k := range r.keys {59 n += len(k.str)60 }61 return n62}6364// compareRecs orders rows as a sheet's sort does (sheet.compareRows):65// key by key, blanks last in either direction, Z to A reversing the66// rest, and ties in the source's order.67func compareRecs(a, b *sortRec, desc []bool) int {68 for i := range a.keys {69 x, y := &a.keys[i], &b.keys[i]70 if (x.rank == blankRank) != (y.rank == blankRank) {71 if x.rank == blankRank {72 return 173 }74 return -175 }76 d := cmp.Compare(x.rank, y.rank)77 if d == 0 {78 switch x.rank {79 case 0, 2:80 d = cmp.Compare(x.num, y.num)81 case 1, 3:82 d = strings.Compare(x.str, y.str)83 }84 }85 if desc[i] {86 d = -d87 }88 if d != 0 {89 return d90 }91 }92 return cmp.Compare(a.row, b.row)93}9495// extSorter sorts a view's rows by their keys in bounded memory.96type extSorter struct{ *extSort[sortRec] }9798func newExtSorter(dir string, desc []bool) *extSorter {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 })}106}107108// add takes a row's keys.109func (s *extSorter) add(row int64, keys []sheet.LiveCell) error {110 r := sortRec{row: row, keys: make([]sortVal, len(keys))}111 for i, k := range keys {112 r.keys[i] = sortValOf(k.V)113 }114 return s.extSort.add(r)115}116117// finish writes the sorted rows' numbers to out and removes the runs.118func (s *extSorter) finish(out io.Writer) (n int64, err error) {119 o, err := s.sorted()120 if err != nil {121 return 0, err122 }123 defer o.close()124 for r := o.next(); r != nil; r = o.next() {125 if err := writeRow(out, r.row); err != nil {126 return n, err127 }128 n++129 }130 return n, o.err()131}132133func writeRow(w io.Writer, row int64) error {134 var b [8]byte135 binary.LittleEndian.PutUint64(b[:], uint64(row))136 _, err := w.Write(b[:])137 return err138}139140// writeRec writes a record to a run: its row, then each key's rank and141// its number or text.142func writeRec(w *bufio.Writer, r *sortRec) {143 var b [binary.MaxVarintLen64]byte144 w.Write(b[:binary.PutUvarint(b[:], uint64(r.row))])145 for _, k := range r.keys {146 w.WriteByte(k.rank)147 switch k.rank {148 case 0, 2:149 binary.LittleEndian.PutUint64(b[:8], math.Float64bits(k.num))150 w.Write(b[:8])151 case 1, 3:152 w.Write(b[:binary.PutUvarint(b[:], uint64(len(k.str)))])153 w.WriteString(k.str)154 }155 }156}157158// readRec reads a record of keys keys from a run, io.EOF at its end.159func readRec(r *bufio.Reader, rec *sortRec, keys int) error {160 row, err := binary.ReadUvarint(r)161 if err != nil {162 return err163 }164 rec.row = int64(row)165 rec.keys = rec.keys[:0]166 for range keys {167 k, err := readKey(r)168 if err != nil {169 if errors.Is(err, io.EOF) {170 err = io.ErrUnexpectedEOF171 }172 return err173 }174 rec.keys = append(rec.keys, k)175 }176 return nil177}178179func readKey(r *bufio.Reader) (sortVal, error) {180 rank, err := r.ReadByte()181 if err != nil {182 return sortVal{}, err183 }184 k := sortVal{rank: rank}185 switch rank {186 case 0, 2:187 var b [8]byte188 if _, err := io.ReadFull(r, b[:]); err != nil {189 return k, err190 }191 k.num = math.Float64frombits(binary.LittleEndian.Uint64(b[:]))192 case 1, 3:193 n, err := binary.ReadUvarint(r)194 if err != nil {195 return k, err196 }197 b := make([]byte, n)198 if _, err := io.ReadFull(r, b); err != nil {199 return k, err200 }201 k.str = string(b)202 }203 return k, nil204}205206// rowFile is a view's rows as a file of row numbers in its order.207type rowFile struct {208 f *os.File209 n int64210}211212// newRowFile makes the file rows are written to, in dir.213func newRowFile(dir string) (*os.File, error) { return os.CreateTemp(dir, "012-view-*.rows") }214215// read returns the row numbers at positions from to from+n, as many as216// there are.217func (rf *rowFile) read(from int64, n int) ([]int64, error) {218 if from < 0 || from >= rf.n || n <= 0 {219 return nil, nil220 }221 n = int(min(int64(n), rf.n-from))222 b := make([]byte, 8*n)223 if _, err := rf.f.ReadAt(b, 8*from); err != nil {224 return nil, err225 }226 out := make([]int64, n)227 for i := range out {228 out[i] = int64(binary.LittleEndian.Uint64(b[8*i:]))229 }230 return out, nil231}232233// close closes and removes the file.234func (rf *rowFile) close() error {235 name := rf.f.Name()236 return errors.Join(rf.f.Close(), os.Remove(name))237}