- commit
- 9373b4dd607bfd4f0e389ecc164aa6f07d37c4a6
- parent
- 3f36d964b8fa781fa6b3a6d687bde9e96eb16594
- author
- Dami <dami.etoile@gmail.com>
- kind
- feat risk score 2.1 of 3, Jev's estimate
| M | internal/ | 120 | ++--------------------- |
| D | internal/ | 8 | -- |
| M | internal/ | 71 | +------------ |
| A | internal/ | 160 | ++++++++++++++++++++++++++++++ |
| M | internal/ | 11 | +- |
| A | internal/ | 122 | +++++++++++++++++++++++ |
| A | internal/ | 45 | ++++++++ |
| A | internal/ | 13 | ++ |
| R | internal/ | 34 | ++++--- |
| A | internal/ | 125 | +++++++++++++++++++++++ |
@@ -10,131 +10,29 @@ package fileio
1010 1111 import (1212 "context"13- "database/sql"14- "fmt"1513 "math/rand/v2"16- "os"17- "path/filepath"1814 "testing"19- "time"20-21- "github.com/parquet-go/parquet-go"2215 2316 "github.com/FelineStateMachine/012/internal/sheet"17+ "github.com/FelineStateMachine/012/internal/stress"2418 )2519 26-// bigRows is the rows of the generated sources.20+// bigRows is the rows of the generated sources (stress.SalesParquet).2721 const bigRows = 10_000_0002822 29-// bigRow is a row of them: an id, one of eight categories, an amount30-// and a date.31-type bigRow struct {32- ID int64 `parquet:"id"`33- Cat string `parquet:"cat,dict"`34- Amount float64 `parquet:"amount"`35- Day int32 `parquet:"day,date"`36-}37-38-var bigCats = []string{"north", "south", "east", "west", "central", "coast", "hills", "islands"}39-40-func bigRowAt(i int) bigRow {41- r := rand.New(rand.NewPCG(uint64(i), 12))42- return bigRow{ID: int64(i), Cat: bigCats[r.IntN(len(bigCats))], Amount: float64(r.IntN(1_000_000)) / 100,43- Day: int32(time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC).Unix()/86400) + int32(r.IntN(2000))}44-}45-46-// generated is the directory the sources are kept in between runs.47-func generated(b *testing.B) string {48- if d := os.Getenv("STRESS_DIR"); d != "" {49- dir := filepath.Join(d, "generated")50- if err := os.MkdirAll(dir, 0o755); err == nil {51- return dir52- }53- }54- return b.TempDir()55-}56-57-// bigParquet writes the Parquet source unless it's there.58-func bigParquet(b *testing.B, dir string) string {59- name := filepath.Join(dir, fmt.Sprintf("sales-%d.parquet", bigRows))60- if _, err := os.Stat(name); err == nil {61- return name62- }63- tmp := name + ".part"64- f, err := os.Create(tmp)65- if err != nil {66- b.Fatal(err)67- }68- w := parquet.NewGenericWriter[bigRow](f, parquet.MaxRowsPerRowGroup(1<<20))69- batch := make([]bigRow, 1<<16)70- for i := 0; i < bigRows; i += len(batch) {71- n := min(len(batch), bigRows-i)72- for j := range n {73- batch[j] = bigRowAt(i + j)74- }75- if _, err := w.Write(batch[:n]); err != nil {76- b.Fatal(err)77- }78- }79- if err := w.Close(); err != nil {80- b.Fatal(err)81- }82- f.Close()83- if err := os.Rename(tmp, name); err != nil {84- b.Fatal(err)85- }86- return name87-}88-89-// bigSQLite writes the SQLite source unless it's there.90-func bigSQLite(b *testing.B, dir string) string {91- name := filepath.Join(dir, fmt.Sprintf("sales-%d.sqlite", bigRows))92- if _, err := os.Stat(name); err == nil {93- return name94- }95- tmp := name + ".part"96- os.Remove(tmp)97- db, err := sql.Open("sqlite", tmp+"?_pragma=journal_mode(OFF)&_pragma=synchronous(OFF)")98- if err != nil {99- b.Fatal(err)100- }101- defer db.Close()102- tx, err := db.Begin()23+// BenchmarkSource reads ten million rows of each kind of source.24+func BenchmarkSource(b *testing.B) {25+ dir := stress.GeneratedDir(b.TempDir())26+ pq, err := stress.SalesParquet(dir, bigRows)10327 if err != nil {10428 b.Fatal(err)10529 }106- if _, err := tx.Exec("CREATE TABLE sales (id INTEGER, cat TEXT, amount REAL, day DATE)"); err != nil {107- b.Fatal(err)108- }109- ins, err := tx.Prepare("INSERT INTO sales VALUES (?, ?, ?, ?)")30+ db, err := stress.SalesSQLite(dir, bigRows)11031 if err != nil {11132 b.Fatal(err)11233 }113- for i := range bigRows {114- r := bigRowAt(i)115- day := time.Unix(int64(r.Day)*86400, 0).UTC().Format(time.DateOnly)116- if _, err := ins.Exec(r.ID, r.Cat, r.Amount, day); err != nil {117- b.Fatal(err)118- }119- }120- if err := tx.Commit(); err != nil {121- b.Fatal(err)122- }123- db.Close()124- if err := os.Rename(tmp, name); err != nil {125- b.Fatal(err)126- }127- return name128-}129-130-// BenchmarkSource reads ten million rows of each kind of source.131-func BenchmarkSource(b *testing.B) {132- dir := generated(b)133- files := map[string]SourceSpec{134- "parquet": {Path: bigParquet(b, dir)},135- "sqlite": {Path: bigSQLite(b, dir)},136- }137- defer benchLock(b)()34+ files := map[string]SourceSpec{"parquet": {Path: pq}, "sqlite": {Path: db}}35+ defer stress.BenchLock(b)()13836 for _, kind := range []string{"parquet", "sqlite"} {13937 spec := files[kind]14038 spec.TempDir = b.TempDir()@@ -3,7 +3,6 @@ package paged
33 import (44 "cmp"55 "context"6- "iter"76 "slices"87 98 "github.com/FelineStateMachine/012/internal/functions"@@ -44,7 +43,7 @@ const maxRows = 256
4443 4544 func (b *book) close() {4645 for _, c := range b.cursors {47- c.stop()46+ b.stopCursor(c)4847 }4948 }5049 @@ -170,70 +169,12 @@ func (b *book) Scan(name string, r sheet.Rect, from sheet.Addr, addrs []sheet.Ad
170169 if from.Row > part.To.Row {171170 return n172171 }173- b.cursor(s, part, from).read(part, emit)174- return n175-}176-177-// cursor is the cursor over the columns of part from from on: the one178-// the last chunk left there, or a new one.179-func (b *book) cursor(s snapshot, part sheet.Rect, from sheet.Addr) *cursor {180- k := cursorKey{key(s.name), part.From.Col, part.To.Col}181- if c := b.cursors[k]; c != nil {182- if c.at == from {183- return c184- }185- c.stop()186- }187- cols := make([]int, 0, part.To.Col-part.From.Col+1)188- for c := part.From.Col; c <= part.To.Col; c++ {189- cols = append(cols, c)190- }191- c := &cursor{b: b, at: from}192- seq := func(yield func(int64, []sheet.LiveCell) bool) {193- if err := s.h.Scan(b.ctx, int64(from.Row-1), cols, yield); err != nil {194- b.fail(err)195- }196- }197- c.next, c.stop = iter.Pull2(seq)198- if b.cursors == nil {199- b.cursors = map[cursorKey]*cursor{}200- }201- b.cursors[k] = c202- return c203-}204-205-// cursor streams a range's rows for Scan, a chunk at a time.206-type cursor struct {207- b *book208- at sheet.Addr // where the next chunk starts209- next func() (int64, []sheet.LiveCell, bool)210- stop func()211- row []sheet.LiveCell // the row being read, from the range's first column212- have bool213-}214-215-// read gives emit the cells of part from the cursor on, until emit216-// returns false or the part ends.217-func (c *cursor) read(part sheet.Rect, emit func(sheet.Addr, sheet.Value) bool) {218- for c.at.Row <= part.To.Row {219- if !c.have {220- _, row, ok := c.next()221- if !ok {222- c.at.Row = part.To.Row + 1223- return224- }225- c.row, c.have = append(c.row[:0], row...), true226- }227- for c.at.Col <= part.To.Col {228- v := c.row[c.at.Col-part.From.Col].V229- a := c.at230- c.at.Col++231- if !emit(a, v) {232- return233- }234- }235- c.at, c.have = sheet.Addr{Col: part.From.Col, Row: c.at.Row + 1}, false172+ 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 failing236176 }177+ return n237178 }238179 239180 // Bounds is the part of r the source's table holds.@@ -0,0 +1,160 @@
1+package paged2+3+import (4+ "context"5+6+ "github.com/FelineStateMachine/012/internal/sheet"7+)8+9+// A cursor streams a range's rows for the book's Scan, a chunk at a10+// time: a goroutine of its own scans the source and hands the rows over11+// in batches, so the source's scan runs on while functions read what12+// it read, and taking a row costs a slice index.13+14+// batchRows is how many rows a batch holds.15+const batchRows = 409616+17+// cursor is a range's rows from where the last chunk ended.18+type cursor struct {19+ at sheet.Addr // where the next chunk starts20+ width int21+ batches chan []sheet.LiveCell22+ free chan []sheet.LiveCell23+ cancel context.CancelFunc24+ err error // what the scan failed with, read once batches is closed25+ cur []sheet.LiveCell26+ i int // the next row of cur27+ row []sheet.LiveCell // the row being read28+ have bool29+}30+31+// cursor is the cursor over the columns of part from from on: the one32+// the last chunk left there, or a new one.33+func (b *book) cursor(s snapshot, part sheet.Rect, from sheet.Addr) *cursor {34+ k := cursorKey{key(s.name), part.From.Col, part.To.Col}35+ if c := b.cursors[k]; c != nil {36+ if c.at == from {37+ return c38+ }39+ b.stopCursor(c)40+ }41+ cols := make([]int, 0, part.To.Col-part.From.Col+1)42+ for c := part.From.Col; c <= part.To.Col; c++ {43+ cols = append(cols, c)44+ }45+ c := startCursor(b.ctx, s, from, cols)46+ if b.cursors == nil {47+ b.cursors = map[cursorKey]*cursor{}48+ }49+ b.cursors[k] = c50+ return c51+}52+53+// startCursor scans the source s's columns cols from the sheet row of54+// from on.55+func startCursor(ctx context.Context, s snapshot, from sheet.Addr, cols []int) *cursor {56+ ctx, cancel := context.WithCancel(ctx)57+ c := &cursor{at: from, width: len(cols), batches: make(chan []sheet.LiveCell, 2), free: make(chan []sheet.LiveCell, 3), cancel: cancel}58+ go func() {59+ defer close(c.batches)60+ batch := c.take()61+ err := s.h.Scan(ctx, int64(from.Row-1), cols, func(_ int64, vals []sheet.LiveCell) bool {62+ if batch = append(batch, vals...); len(batch) < batchRows*c.width {63+ return true64+ }65+ if !c.send(ctx, batch) {66+ return false67+ }68+ batch = c.take()69+ return true70+ })71+ if len(batch) > 0 {72+ c.send(ctx, batch)73+ }74+ if ctx.Err() == nil {75+ c.err = err76+ }77+ }()78+ return c79+}80+81+// take is a batch to fill: one read already, or a new one.82+func (c *cursor) take() []sheet.LiveCell {83+ select {84+ case b := <-c.free:85+ return b[:0]86+ default:87+ return make([]sheet.LiveCell, 0, batchRows*c.width)88+ }89+}90+91+// send hands a batch over, reporting false once the cursor is stopped.92+func (c *cursor) send(ctx context.Context, batch []sheet.LiveCell) bool {93+ select {94+ case c.batches <- batch:95+ return true96+ case <-ctx.Done():97+ return false98+ }99+}100+101+// next is the next row, with false after the last.102+func (c *cursor) next() ([]sheet.LiveCell, bool) {103+ if c.i*c.width >= len(c.cur) {104+ if c.cur != nil {105+ select {106+ case c.free <- c.cur:107+ default:108+ }109+ }110+ b, ok := <-c.batches111+ if !ok {112+ c.cur = nil113+ return nil, false114+ }115+ c.cur, c.i = b, 0116+ }117+ row := c.cur[c.i*c.width : (c.i+1)*c.width]118+ c.i++119+ return row, true120+}121+122+// stopCursor stops c's scan, noting what it failed with.123+func (b *book) stopCursor(c *cursor) {124+ c.cancel()125+ for range c.batches {126+ }127+ if c.err != nil {128+ b.fail(c.err)129+ }130+}131+132+// read gives emit the cells of part from the cursor on, until emit133+// returns false or the part ends.134+func (c *cursor) read(part sheet.Rect, emit func(sheet.Addr, sheet.Value) bool) {135+ for c.at.Row <= part.To.Row {136+ if !c.have {137+ row, ok := c.next()138+ if !ok {139+ c.at.Row = part.To.Row + 1140+ return141+ }142+ c.row, c.have = row, true143+ }144+ more := true145+ for more && c.at.Col <= part.To.Col {146+ v := c.row[c.at.Col-part.From.Col].V147+ a := c.at148+ c.at.Col++149+ more = emit(a, v)150+ }151+ if c.at.Col > part.To.Col {152+ // The row is read: the next chunk starts at the next one's153+ // first cell, as the Reader asks for it.154+ c.at, c.have = sheet.Addr{Col: part.From.Col, Row: c.at.Row + 1}, false155+ }156+ if !more {157+ return158+ }159+ }160+}@@ -32,11 +32,14 @@ func Apply(w *sheet.Workbook, d Done) {
3232 }3333 }3434 35-// Settle links the sources links names (Links) and runs every job the36-// host queues, one after another, applying each, until none is left: for37-// what has no screen to keep live while they run (012 get, tests).35+// Settle links the sources links names (Links), unless links is nil,36+// and runs every job the host queues, one after another, applying each,37+// until none is left: for what has no screen to keep live while they38+// run (012 get, tests).3839 func (h *Host) Settle(ctx context.Context, w *sheet.Workbook, links []Linked) {39- h.Link(links)40+ if links != nil {41+ h.Link(links)42+ }4043 w.SetSources(h)4144 for jobs := h.Jobs(); len(jobs) > 0; jobs = h.Jobs() {4245 for _, j := range jobs {@@ -0,0 +1,122 @@
1+//go:build stress2+3+// Stress benchmarks of formulas and pivot tables over linked sources of4+// ten million rows (stress.SalesParquet, stress.SalesSQLite): each5+// question answered from scratch by a job, as the screen's host runs it6+// in the background, with the peak heap it took. Timed under the bench7+// lock.8+package paged9+10+import (11+ "context"12+ "testing"13+14+ "github.com/FelineStateMachine/012/internal/sheet"15+ "github.com/FelineStateMachine/012/internal/stress"16+)17+18+// bigRows is the rows of the generated sources.19+const bigRows = 10_000_00020+21+// bigFormulas are the formulas measured, reading the source as sales:22+// A id, B cat, C amount, D day.23+var bigFormulas = []struct{ name, formula string }{24+ {"SUM", "=SUM(sales[amount])"},25+ {"AVERAGE", "=AVERAGE(sales[amount])"},26+ {"COUNTIFS", `=COUNTIFS(sales[cat],"north",sales[amount],">5000")`},27+ {"SUMIFS", `=SUMIFS(sales[amount],sales[cat],"east")`},28+ {"SUMPRODUCT", "=SUMPRODUCT(sales[amount],sales[id])"},29+ {"XLOOKUP-last", "=XLOOKUP(9999999,sales[id],sales[amount])"},30+ {"MATCH-middle", "=MATCH(5000000,sales[id],0)"},31+ {"MEDIAN-held", "=MEDIAN(sales[amount])"},32+}33+34+func BenchmarkSourceFormulas(b *testing.B) {35+ dir := stress.GeneratedDir(b.TempDir())36+ pq, err := stress.SalesParquet(dir, bigRows)37+ if err != nil {38+ b.Fatal(err)39+ }40+ db, err := stress.SalesSQLite(dir, bigRows)41+ if err != nil {42+ b.Fatal(err)43+ }44+ defer stress.BenchLock(b)()45+ for _, src := range []struct{ kind, path string }{{"parquet", pq}, {"sqlite", db}} {46+ b.Run(src.kind, func(b *testing.B) {47+ for _, f := range bigFormulas {48+ b.Run(f.name, func(b *testing.B) { benchFormula(b, src.path, f.formula) })49+ }50+ b.Run("pivot-cat", func(b *testing.B) { benchPivot(b, src.path) })51+ })52+ }53+}54+55+// bigBook is a workbook linking the source at path as sales, opened by56+// a host of its own.57+func bigBook(b *testing.B, path string) (*sheet.Workbook, *Host) {58+ w := sheet.NewBook()59+ if _, err := w.AddSource("sales", sheet.LinkSource{Path: path}, w.Sheet(0)); err != nil {60+ b.Fatal(err)61+ }62+ h := NewHost(sheet.DefaultMaxCells)63+ h.Settle(context.Background(), w, Links(w, same, b.TempDir()))64+ if info, _ := w.LookupSource("sales"); info.Shape.Rows != bigRows {65+ b.Fatalf("the source has %d rows: %s", info.Shape.Rows, info.Err)66+ }67+ return w, h68+}69+70+// benchFormula times the job answering formula over the source.71+func benchFormula(b *testing.B, path, formula string) {72+ w, h := bigBook(b, path)73+ defer h.Close()74+ s := w.Sheet(0)75+ var v sheet.Value76+ measure(b, func() {77+ h.forget("sales")78+ s.Set(sheet.Addr{}, "")79+ s.Set(sheet.Addr{}, formula)80+ h.Settle(context.Background(), w, nil)81+ v = s.Value(sheet.Addr{})82+ })83+ if sheet.IsPending(v) {84+ b.Fatalf("%s never answered", formula)85+ }86+ b.ReportMetric(float64(bigRows)*float64(b.N)/b.Elapsed().Seconds(), "rows/s")87+}88+89+// benchPivot times a pivot table of the amounts by category.90+func benchPivot(b *testing.B, path string) {91+ w, h := bigBook(b, path)92+ defer h.Close()93+ info, _ := w.LookupSource("sales")94+ r := sheet.Rect{To: sheet.Addr{Col: 3, Row: sheet.MaxRows - 1}}95+ p := sheet.NewPivot(info.Sheet, r)96+ p.Rows = []sheet.PivotGroup{{Col: 1}}97+ p.Values = []sheet.PivotValue{{Col: 2, Summarize: sheet.SumBy}, {Col: 2, Summarize: sheet.CountBy}}98+ measure(b, func() {99+ h.forget("sales")100+ pt, err := w.CreatePivot(info.Sheet, r, "", p)101+ if err != nil {102+ b.Fatal(err)103+ }104+ h.Settle(context.Background(), w, nil)105+ if v := pt.Value(sheet.Addr{Row: 9}); v.Kind != sheet.Text {106+ b.Fatalf("no Grand Total row: %v", v)107+ }108+ w.DeleteSheet(pt)109+ })110+}111+112+// measure runs fn in the benchmark's loop, reporting the peak heap it113+// took above what was live before.114+func measure(b *testing.B, fn func()) {115+ before := stress.LiveHeap()116+ stop := stress.SamplePeak()117+ for b.Loop() {118+ fn()119+ }120+ peak := stop()121+ b.ReportMetric(float64(peak-min(peak, before))/(1<<20), "peak-MB")122+}@@ -0,0 +1,45 @@
1+package stress2+3+import (4+ "runtime"5+ "runtime/metrics"6+ "time"7+)8+9+// SamplePeak samples the heap's live objects every 100 us until the10+// function it returns is called, which returns the most seen.11+func SamplePeak() func() uint64 {12+ sample := []metrics.Sample{{Name: "/memory/classes/heap/objects:bytes"}}13+ read := func() uint64 {14+ metrics.Read(sample)15+ return sample[0].Value.Uint64()16+ }17+ done := make(chan struct{})18+ peak := make(chan uint64)19+ go func() {20+ hi := read()21+ t := time.NewTicker(100 * time.Microsecond)22+ defer t.Stop()23+ for {24+ select {25+ case <-done:26+ peak <- max(hi, read())27+ return28+ case <-t.C:29+ hi = max(hi, read())30+ }31+ }32+ }()33+ return func() uint64 {34+ close(done)35+ return <-peak36+ }37+}38+39+// LiveHeap is the heap in use after a collection.40+func LiveHeap() uint64 {41+ runtime.GC()42+ var ms runtime.MemStats43+ runtime.ReadMemStats(&ms)44+ return ms.HeapAlloc45+}@@ -0,0 +1,13 @@
1+//go:build !unix2+3+package stress4+5+// Logger is what BenchLock says why it times without the lock through:6+// a testing.TB.7+type Logger interface {8+ Logf(format string, args ...any)9+ Fatalf(format string, args ...any)10+}11+12+// BenchLock times without the lock where there is no flock(2).13+func BenchLock(Logger) func() { return func() {} }@@ -1,32 +1,38 @@
1-//go:build stress && unix1+//go:build unix22 3-package fileio3+package stress44 55 import (66 "errors"77 "io/fs"88 "os"99 "syscall"10- "testing"1110 "time"1211 )1312 14-// benchLockPath is the lock timed runs on this machine share, as the13+// BenchLockPath is the lock timed runs on this machine share, as the1514 // speed gate takes it (internal/ui/speedlock_unix_test.go): a file1615 // locked with flock(2), or a directory made and removed after where1716 // there is no flock command (macOS).18-const benchLockPath = "/tmp/012-bench.lock"17+const BenchLockPath = "/tmp/012-bench.lock"1918 20-// benchLock waits up to ten minutes for the lock and returns its19+// Logger is what BenchLock says why it times without the lock through:20+// a testing.TB.21+type Logger interface {22+ Logf(format string, args ...any)23+ Fatalf(format string, args ...any)24+}25+26+// BenchLock waits up to ten minutes for the lock and returns its2127 // release; past that, or when it can't be taken, it times without it.22-func benchLock(tb testing.TB) func() {28+func BenchLock(tb Logger) func() {2329 deadline := time.Now().Add(10 * time.Minute)2430 for {25- err := os.Mkdir(benchLockPath, 0o755)31+ err := os.Mkdir(BenchLockPath, 0o755)2632 if err == nil {27- return func() { os.Remove(benchLockPath) }33+ return func() { os.Remove(BenchLockPath) }2834 }29- info, serr := os.Stat(benchLockPath)35+ info, serr := os.Stat(BenchLockPath)3036 switch {3137 case !errors.Is(err, fs.ErrExist) || serr != nil:3238 tb.Logf("timing without the lock: %v", err)@@ -34,7 +40,7 @@ func benchLock(tb testing.TB) func() {
3440 case info.Mode().IsRegular():3541 return flockBench(tb)3642 case time.Now().After(deadline):37- tb.Logf("timing without the lock: %s held too long", benchLockPath)43+ tb.Logf("timing without the lock: %s held too long", BenchLockPath)3844 return func() {}3945 }4046 time.Sleep(200 * time.Millisecond)@@ -42,15 +48,15 @@ func benchLock(tb testing.TB) func() {
4248 }4349 4450 // flockBench takes the lock as a file, with flock(2).45-func flockBench(tb testing.TB) func() {46- f, err := os.OpenFile(benchLockPath, os.O_RDWR, 0)51+func flockBench(tb Logger) func() {52+ f, err := os.OpenFile(BenchLockPath, os.O_RDWR, 0)4753 if err != nil {4854 tb.Logf("timing without the lock: %v", err)4955 return func() {}5056 }5157 if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {5258 f.Close()53- tb.Fatalf("locking %s: %v", benchLockPath, err)59+ tb.Fatalf("locking %s: %v", BenchLockPath, err)5460 }5561 return func() {5662 syscall.Flock(int(f.Fd()), syscall.LOCK_UN)@@ -0,0 +1,125 @@
1+package stress2+3+import (4+ "database/sql"5+ "fmt"6+ "math/rand/v2"7+ "os"8+ "path/filepath"9+ "time"10+11+ "github.com/parquet-go/parquet-go"12+ _ "modernc.org/sqlite" // registers the "sqlite" driver13+)14+15+// Sources for the benchmarks of linked sources: a Parquet file and a16+// SQLite table of sales, rows rows each (ten million for make stress),17+// written once into $STRESS_DIR/generated, or a directory the caller18+// gives, and kept between runs, as generating them takes a while.19+20+// SaleRow is a row of the generated sources: an id, one of eight21+// categories, an amount and a date. Its columns are A to D of the22+// sources' tabs.23+type SaleRow struct {24+ ID int64 `parquet:"id"`25+ Cat string `parquet:"cat,dict"`26+ Amount float64 `parquet:"amount"`27+ Day int32 `parquet:"day,date"`28+}29+30+// SaleCats are the categories, in the order rows draw them from.31+var SaleCats = []string{"north", "south", "east", "west", "central", "coast", "hills", "islands"}32+33+// Sale is row i of the sources, the same whenever it's asked for.34+func Sale(i int) SaleRow {35+ r := rand.New(rand.NewPCG(uint64(i), 12))36+ return SaleRow{ID: int64(i), Cat: SaleCats[r.IntN(len(SaleCats))], Amount: float64(r.IntN(1_000_000)) / 100,37+ Day: int32(time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC).Unix()/86400) + int32(r.IntN(2000))}38+}39+40+// GeneratedDir is where generated sources are kept: $STRESS_DIR's41+// generated folder, or fallback.42+func GeneratedDir(fallback string) string {43+ if d := os.Getenv("STRESS_DIR"); d != "" {44+ dir := filepath.Join(d, "generated")45+ if err := os.MkdirAll(dir, 0o755); err == nil {46+ return dir47+ }48+ }49+ return fallback50+}51+52+// SalesParquet is the Parquet source of rows rows in dir, written unless53+// it's there.54+func SalesParquet(dir string, rows int) (string, error) {55+ name := filepath.Join(dir, fmt.Sprintf("sales-%d.parquet", rows))56+ if _, err := os.Stat(name); err == nil {57+ return name, nil58+ }59+ tmp := name + ".part"60+ f, err := os.Create(tmp)61+ if err != nil {62+ return "", err63+ }64+ defer f.Close()65+ w := parquet.NewGenericWriter[SaleRow](f, parquet.MaxRowsPerRowGroup(1<<20))66+ batch := make([]SaleRow, 1<<16)67+ for i := 0; i < rows; i += len(batch) {68+ n := min(len(batch), rows-i)69+ for j := range n {70+ batch[j] = Sale(i + j)71+ }72+ if _, err := w.Write(batch[:n]); err != nil {73+ return "", err74+ }75+ }76+ if err := w.Close(); err != nil {77+ return "", err78+ }79+ if err := f.Close(); err != nil {80+ return "", err81+ }82+ return name, os.Rename(tmp, name)83+}84+85+// SalesSQLite is the SQLite source of rows rows in dir, a table named86+// sales, written unless it's there.87+func SalesSQLite(dir string, rows int) (string, error) {88+ name := filepath.Join(dir, fmt.Sprintf("sales-%d.sqlite", rows))89+ if _, err := os.Stat(name); err == nil {90+ return name, nil91+ }92+ tmp := name + ".part"93+ os.Remove(tmp)94+ db, err := sql.Open("sqlite", tmp+"?_pragma=journal_mode(OFF)&_pragma=synchronous(OFF)")95+ if err != nil {96+ return "", err97+ }98+ defer db.Close()99+ tx, err := db.Begin()100+ if err != nil {101+ return "", err102+ }103+ defer tx.Rollback()104+ if _, err := tx.Exec("CREATE TABLE sales (id INTEGER, cat TEXT, amount REAL, day DATE)"); err != nil {105+ return "", err106+ }107+ ins, err := tx.Prepare("INSERT INTO sales VALUES (?, ?, ?, ?)")108+ if err != nil {109+ return "", err110+ }111+ for i := range rows {112+ r := Sale(i)113+ day := time.Unix(int64(r.Day)*86400, 0).UTC().Format(time.DateOnly)114+ if _, err := ins.Exec(r.ID, r.Cat, r.Amount, day); err != nil {115+ return "", err116+ }117+ }118+ if err := tx.Commit(); err != nil {119+ return "", err120+ }121+ if err := db.Close(); err != nil {122+ return "", err123+ }124+ return name, os.Rename(tmp, name)125+}