1package fileio23import (4 "bufio"5 "cmp"6 "encoding/binary"7 "io"8 "math"9)1011// 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.17type NumberSort struct {18 s *extSort[numRec]19}2021type numRec struct {22 v float6423 pos int6424}2526var 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}5455// NewNumberSort is a sort spilling its runs to files in dir, "" for the56// system's temporary directory.57func NewNumberSort(dir string) *NumberSort {58 s := newExtSort(dir, numCodec)59 s.limit = max(sortRunBytes/numRunShare, 16)60 return &NumberSort{s: s}61}6263// 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.66const numRunShare = 46768// Add takes a number read at position pos.69func (n *NumberSort) Add(v float64, pos int64) error { return n.s.add(numRec{v, pos}) }7071// Len counts the numbers added.72func (n *NumberSort) Len() int { return int(n.s.n) }7374// Close removes what the sort spilled, when Sorted wasn't asked for.75func (n *NumberSort) Close() { n.s.remove() }7677// Sorted reads the numbers back in ascending order, once. Closing it78// removes the runs.79func (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}8687// SortedNumbers is a NumberSort's numbers in ascending order.88type SortedNumbers struct {89 o *sortedRecs[numRec]90 n int91}9293// Len counts the numbers.94func (s *SortedNumbers) Len() int { return s.n }9596// Next is the next number and its position as read; false at the end,97// or on an error, which Err says.98func (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}105106// Err is the error reading the runs met, if any.107func (s *SortedNumbers) Err() error { return s.o.err() }108109// Close removes the runs.110func (s *SortedNumbers) Close() error {111 s.o.close()112 return nil113}