diff --git a/api/api.go b/api/api.go index 1ac6f35cbd..5114a28938 100644 --- a/api/api.go +++ b/api/api.go @@ -36,9 +36,10 @@ type VersionResponse struct { } type PoolPostRequest struct { - Name string `json:"name"` - SortKeys SortKeys `json:"layout"` - Thresh int64 `json:"thresh"` + Name string `json:"name"` + SortKeys SortKeys `json:"layout"` + ObjectCap uint64 `json:"objectcap"` + FrameCap uint64 `json:"framecap"` } type SortKeys struct { diff --git a/bsup/writer.go b/bsup/writer.go index 6d321318f7..b3e45eadd6 100644 --- a/bsup/writer.go +++ b/bsup/writer.go @@ -14,45 +14,52 @@ import ( "github.com/superdb/super/vector/vio" ) -var maxFrameSize uint32 = 120_000 - -// XXX a future PR will wire in compress / thresh options to command line. -// XXX Rows is the key flag we need for the rows writer. type WriterOpts struct { Compress bool - // FrameThresh is the minimum frame size in uncompressed bytes. - FrameThresh int - Rows bool + FrameCap uint64 + Rows bool } +const DefaultFrameCap = 20_000 + // ColumnWriter implements the vio.Pusher interface. A Pusher creates a vector // BSUP object from a stream of vector.Any. type ColumnWriter struct { - writer io.WriteCloser - dynamic *vbuild.DynamicBuilder - sctx *super.Context - fuser fuser - size uint64 - ctrl *RowWriter + writer io.WriteCloser + dynamic *vbuild.DynamicBuilder + sctx *super.Context + fuser fuser + size uint64 + ctrl *RowWriter + framecap uint64 } var _ vio.Pusher = (*ColumnWriter)(nil) func NewColumnWriter(w io.WriteCloser) *ColumnWriter { + return NewColumnWriterWithCap(w, DefaultFrameCap) +} + +func NewColumnWriterWithCap(w io.WriteCloser, frameCap uint64) *ColumnWriter { sctx := super.NewContext() return &ColumnWriter{ - writer: w, - dynamic: vbuild.NewDynamicBuilder(), - sctx: sctx, - fuser: newFuser(sctx), + writer: w, + dynamic: vbuild.NewDynamicBuilder(), + sctx: sctx, + fuser: newFuser(sctx), + framecap: frameCap, } } func NewWriterWithOpts(w io.WriteCloser, opt WriterOpts) vio.PushCloser { if opt.Rows { - return NewRowWriter(w) + writer := NewRowWriter(w) + writer.framecap = uint32(opt.FrameCap) + return writer } - return NewColumnWriter(w) + writer := NewColumnWriter(w) + writer.framecap = opt.FrameCap + return writer } func (c *ColumnWriter) Close() error { @@ -87,7 +94,7 @@ func (c *ColumnWriter) WriteSuperFrame(vec vector.Any) (uint64, error) { func (c *ColumnWriter) Push(vec vector.Any) error { if vec.Len() != 0 { c.dynamic.Write(vec) - if c.dynamic.Len() >= maxFrameSize { + if c.dynamic.Len() >= uint32(c.framecap) { return c.pushFrame() } } @@ -202,6 +209,7 @@ type RowWriter struct { size uint64 bytes []byte len uint32 + framecap uint32 } var _ vio.Pusher = (*RowWriter)(nil) @@ -215,6 +223,7 @@ func NewRowWriter(w io.WriteCloser) *RowWriter { sctx: sctx, fuser: newFuser(sctx), superfuser: newFuser(sctx), + framecap: DefaultFrameCap, } } @@ -237,7 +246,7 @@ func (r *RowWriter) Push(vec vector.Any) error { } } r.len += vec.Len() - if r.len >= maxFrameSize { + if r.len >= r.framecap { return r.pushFrame(false) } return nil @@ -253,7 +262,7 @@ func (r *RowWriter) write(val super.Value, ctrl bool) error { r.fuser.fuse(typ) r.superfuser.fuse(typ) r.len++ - if r.len >= maxFrameSize || ctrl { + if r.len >= r.framecap || ctrl { return r.pushFrame(ctrl) } return nil diff --git a/bsup/ztests/framecap.yaml b/bsup/ztests/framecap.yaml new file mode 100644 index 0000000000..30add89cc4 --- /dev/null +++ b/bsup/ztests/framecap.yaml @@ -0,0 +1,9 @@ +script: | + seq 100 | super -inputcap 5 -framecap 5 -o out.bsup - + super dev bsup out.bsup | super -s -c "count() by nameof(typeof(this)) | ? Header or Footer | sort count" - + +outputs: + - name: stdout + data: | + {nameof:"SuperFooter",count:1} + {nameof:"ColumnHeader",count:20} diff --git a/cli/inputflags/flags.go b/cli/inputflags/flags.go index f37100eb7f..203f83c8f6 100644 --- a/cli/inputflags/flags.go +++ b/cli/inputflags/flags.go @@ -26,6 +26,7 @@ func (f *Flags) SetFlags(fs *flag.FlagSet) { }) fs.BoolVar(&f.Dynamic, "dynamic", false, "disable static type checking of inputs") fs.StringVar(&opts.Format, "i", "auto", "format of input data [auto,arrows,bsup,csv,json,line,parquet,sup,tsv,zeek]") + fs.IntVar(&opts.InputCap, "inputcap", 0, "limit size of batched units of input") fs.BoolVar(&f.Static, "static", false, "force static type checking of inputs") } diff --git a/cli/outputflags/flags.go b/cli/outputflags/flags.go index 34eece7312..1dbc46b576 100644 --- a/cli/outputflags/flags.go +++ b/cli/outputflags/flags.go @@ -41,6 +41,7 @@ func (f *Flags) Options() anyio.WriterOpts { func (f *Flags) setFlags(fs *flag.FlagSet) { fs.BoolVar(&f.color, "color", true, "enable/disable color formatting for -S and db text output") + fs.Uint64Var(&f.BSUP.FrameCap, "framecap", 10000, "number of values per BSUP frame") fs.BoolVar(&f.CSV.NoHeader, "noheader", false, "omit header for CSV and TSV output") fs.IntVar(&f.pretty, "pretty", 2, "tab size to pretty print JSON and Super JSON output (0 for newline-delimited output") diff --git a/cmd/super/db/create/command.go b/cmd/super/db/create/command.go index 7ddedc958a..96b912e2b1 100644 --- a/cmd/super/db/create/command.go +++ b/cmd/super/db/create/command.go @@ -5,12 +5,12 @@ import ( "flag" "fmt" + "github.com/superdb/super/bsup" "github.com/superdb/super/cli/poolflags" "github.com/superdb/super/cmd/super/db" "github.com/superdb/super/db/data" "github.com/superdb/super/order" "github.com/superdb/super/pkg/charm" - "github.com/superdb/super/pkg/units" ) var spec = &charm.Spec{ @@ -25,9 +25,10 @@ See https://superdb.org/command/db.html#super-db-create type Command struct { *db.Command - sortKey string - thresh units.Bytes - use bool + sortKey string + frameCap uint64 + objectCap uint64 + use bool } func init() { @@ -36,8 +37,8 @@ func init() { func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) { c := &Command{Command: parent.(*db.Command)} - c.thresh = data.DefaultThreshold - f.Var(&c.thresh, "S", "target size of pool data objects, as '10MB' or '4GiB', etc.") + f.Uint64Var(&c.frameCap, "framecap", bsup.DefaultFrameCap, "target number of values BSUP frames") + f.Uint64Var(&c.objectCap, "objectcap", data.DefaultObjectCap, "target number of values in pool data objects") f.BoolVar(&c.use, "use", false, "set created pool as the current pool") f.StringVar(&c.sortKey, "orderby", "ts:desc", "pool key with optional :asc or :desc suffix to organize data in pool (cannot be changed)") return c, nil @@ -61,7 +62,7 @@ func (c *Command) Run(args []string) error { return err } poolName := args[0] - id, err := db.CreatePool(ctx, poolName, sortKey, int64(c.thresh)) + id, err := db.CreatePool(ctx, poolName, sortKey, c.objectCap, c.frameCap) if err != nil { return err } diff --git a/cmd/super/db/internal/dbmanage/scan.go b/cmd/super/db/internal/dbmanage/scan.go index 35b7221bd3..6c3e60ec57 100644 --- a/cmd/super/db/internal/dbmanage/scan.go +++ b/cmd/super/db/internal/dbmanage/scan.go @@ -48,7 +48,7 @@ func scan(ctx context.Context, it *objectIterator, pool *pools.Config, runCh cha if err != nil { return err } - if run.overlaps(o.Min, o.Max) || run.size+o.Size < pool.Threshold { + if run.overlaps(o.Min, o.Max) || run.len+o.Count < pool.ObjectCap { run.add(o) continue } @@ -119,7 +119,7 @@ type runBuilder struct { span extent.Span cmp expr.CompareFn objects []*object - size int64 + len uint64 } func newRunBuilder() *runBuilder { @@ -135,7 +135,7 @@ func (r *runBuilder) overlaps(first, last super.Value) bool { func (r *runBuilder) add(o *object) { r.objects = append(r.objects, o) - r.size += o.Size + r.len += o.Count if r.span == nil { r.span = extent.NewGeneric(o.Min, o.Max, r.cmp) return @@ -155,5 +155,5 @@ func (r *runBuilder) objectIDs() []uuid.UUID { func (r *runBuilder) reset() { r.span = nil r.objects = r.objects[:0] - r.size = 0 + r.len = 0 } diff --git a/cmd/super/ztests/terminal-output-format.yaml b/cmd/super/ztests/terminal-output-format.yaml index 36cda712d6..9ef1d077d6 100644 --- a/cmd/super/ztests/terminal-output-format.yaml +++ b/cmd/super/ztests/terminal-output-format.yaml @@ -15,4 +15,4 @@ outputs: - name: stdout data: "{a:1}\r\n\ {b:2}\r\n\ - p1 xxx key ts order desc\r\n" + p1 xxx key ts order desc objectcap 1048576 framecap 20000\r\n" diff --git a/db/api/api.go b/db/api/api.go index ed390641d2..3a952ff287 100644 --- a/db/api/api.go +++ b/db/api/api.go @@ -29,7 +29,7 @@ type Interface interface { Query(ctx context.Context, query []srcfiles.Input) (vio.Scanner, error) PoolID(ctx context.Context, poolName string) (uuid.UUID, error) CommitObject(ctx context.Context, poolID uuid.UUID, branchName string) (uuid.UUID, error) - CreatePool(context.Context, string, order.SortKeys, int64) (uuid.UUID, error) + CreatePool(context.Context, string, order.SortKeys, uint64, uint64) (uuid.UUID, error) RemovePool(context.Context, uuid.UUID) error RenamePool(context.Context, uuid.UUID, string) error CreateBranch(ctx context.Context, pool uuid.UUID, name string, parent uuid.UUID) error diff --git a/db/api/local.go b/db/api/local.go index 1c2ce6122b..ba956ef969 100644 --- a/db/api/local.go +++ b/db/api/local.go @@ -61,11 +61,11 @@ func (l *local) Root() *db.Root { return l.db } -func (l *local) CreatePool(ctx context.Context, name string, sortKeys order.SortKeys, thresh int64) (uuid.UUID, error) { +func (l *local) CreatePool(ctx context.Context, name string, sortKeys order.SortKeys, objectCap, frameCap uint64) (uuid.UUID, error) { if name == "" { return uuid.Nil(), errors.New("no pool name provided") } - pool, err := l.db.CreatePool(ctx, name, sortKeys, thresh) + pool, err := l.db.CreatePool(ctx, name, sortKeys, objectCap, frameCap) if err != nil { return uuid.Nil(), err } diff --git a/db/api/remote.go b/db/api/remote.go index 9c7e79d4b8..e00559a110 100644 --- a/db/api/remote.go +++ b/db/api/remote.go @@ -53,14 +53,15 @@ func (r *remote) CommitObject(ctx context.Context, poolID uuid.UUID, branchName return res.Commit, err } -func (r *remote) CreatePool(ctx context.Context, name string, sortKeys order.SortKeys, thresh int64) (uuid.UUID, error) { +func (r *remote) CreatePool(ctx context.Context, name string, sortKeys order.SortKeys, objectCap, frameCap uint64) (uuid.UUID, error) { res, err := r.conn.CreatePool(ctx, api.PoolPostRequest{ Name: name, SortKeys: api.SortKeys{ Order: sortKeys.Primary().Order, Keys: field.List{sortKeys.Primary().Path}, }, - Thresh: thresh, + ObjectCap: objectCap, + FrameCap: frameCap, }) if err != nil { return uuid.Nil(), err diff --git a/db/data/object.go b/db/data/object.go index b511d8095f..75ba56f289 100644 --- a/db/data/object.go +++ b/db/data/object.go @@ -15,7 +15,7 @@ import ( ) const ( - DefaultThreshold = 500 * 1024 * 1024 + DefaultObjectCap = 1024 * 1024 ) // A FileKind is the first part of a file name, used to differentiate files diff --git a/db/pools/config.go b/db/pools/config.go index 4a7180c803..fd492aa30d 100644 --- a/db/pools/config.go +++ b/db/pools/config.go @@ -3,6 +3,7 @@ package pools import ( "uuid" + "github.com/superdb/super/bsup" "github.com/superdb/super/db/data" "github.com/superdb/super/db/journal" "github.com/superdb/super/order" @@ -16,24 +17,29 @@ type Config struct { Name string `super:"name"` ID uuid.UUID `super:"id"` SortKeys order.SortKeys `super:"layout"` - Threshold int64 `super:"threshold"` + ObjectCap uint64 `super:"objectcap"` + FrameCap uint64 `super:"framecap"` } var _ journal.Entry = (*Config)(nil) -func NewConfig(name string, sortKeys order.SortKeys, thresh int64) *Config { +func NewConfig(name string, sortKeys order.SortKeys, objectCap, frameCap uint64) *Config { if sortKeys.IsNil() { sortKeys = order.SortKeys{order.NewSortKey(order.Desc, field.Dotted("ts"))} } - if thresh == 0 { - thresh = data.DefaultThreshold + if objectCap == 0 { + objectCap = data.DefaultObjectCap + } + if frameCap == 0 { + frameCap = bsup.DefaultFrameCap } return &Config{ Ts: nano.Now(), Name: name, ID: uuid.NewV7(), SortKeys: sortKeys, - Threshold: thresh, + ObjectCap: objectCap, + FrameCap: frameCap, } } diff --git a/db/root.go b/db/root.go index fa3f0ac0c3..f357869e08 100644 --- a/db/root.go +++ b/db/root.go @@ -10,6 +10,7 @@ import ( arc "github.com/hashicorp/golang-lru/arc/v2" "github.com/superdb/super" + "github.com/superdb/super/bsup" "github.com/superdb/super/bsupbytes" "github.com/superdb/super/compiler/dag" "github.com/superdb/super/db/branches" @@ -315,20 +316,23 @@ func (r *Root) RenamePool(ctx context.Context, id uuid.UUID, newName string) err return r.pools.Rename(ctx, id, newName) } -func (r *Root) CreatePool(ctx context.Context, name string, sortKeys order.SortKeys, thresh int64) (*Pool, error) { +func (r *Root) CreatePool(ctx context.Context, name string, sortKeys order.SortKeys, objectCap, frameCap uint64) (*Pool, error) { if name == "HEAD" { return nil, fmt.Errorf("pool cannot be named %q", name) } if r.pools.LookupByName(ctx, name) != nil { return nil, fmt.Errorf("%s: %w", name, pools.ErrExists) } - if thresh == 0 { - thresh = data.DefaultThreshold - } if len(sortKeys) > 1 { return nil, errors.New("multiple pool keys not supported") } - config := pools.NewConfig(name, sortKeys, thresh) + if objectCap == 0 { + objectCap = data.DefaultObjectCap + } + if frameCap == 0 { + frameCap = bsup.DefaultFrameCap + } + config := pools.NewConfig(name, sortKeys, objectCap, frameCap) if err := CreatePool(ctx, r.engine, r.logger, r.path, config); err != nil { return nil, err } diff --git a/db/writer.go b/db/writer.go index 8500443dd8..e716832cd5 100644 --- a/db/writer.go +++ b/db/writer.go @@ -50,14 +50,17 @@ func (w *Writer) newObject() *data.Object { } func (w *Writer) Push(vec vector.Any) error { - // XXX We need a threshold here (or in BSUP column writer) - // so we can write lots of smaller row-group-sized column frames - // per super frame. This will come in a subsequent PR. w.builder.Write(vec) + if uint64(w.builder.Len()) >= w.pool.ObjectCap { + return w.flush() + } return nil } func (w *Writer) flush() error { + // Each object is one big vector limited in length by ObjectCap. + // Inside of the superframe are Frames which are in turn limited + // in length by FrameCap. vec := w.builder.Build() if vec.Len() == 0 { return nil @@ -78,7 +81,7 @@ func (w *Writer) flush() error { object.Min = minVal object.Max = maxVal } - writer := bsup.NewColumnWriter(out) + writer := bsup.NewColumnWriterWithCap(out, w.pool.FrameCap) size, err := writer.WriteSuperFrame(vec) if err != nil { out.Close() diff --git a/db/ztests/aggregate-ts.yaml b/db/ztests/aggregate-ts.yaml index 527cf27e33..4d9f03dea8 100644 --- a/db/ztests/aggregate-ts.yaml +++ b/db/ztests/aggregate-ts.yaml @@ -3,9 +3,9 @@ script: | super db init -q # aggregate-ts.sup contains records over two days, mapping to two partitions, and # the small target size here causes each partition to have several segments. - super db create -q -S 256B logs + super db create -q -objectcap=50 -framecap=10 logs # causes parition to have several chunk files. - super db load -q -use logs aggregate-ts.sup + super db load -q -use logs -inputcap=10 aggregate-ts.sup echo === super db -s -c 'from logs | head 1' echo === diff --git a/db/ztests/delete-where.yaml b/db/ztests/delete-where.yaml index c770e1c74f..4db99f8244 100644 --- a/db/ztests/delete-where.yaml +++ b/db/ztests/delete-where.yaml @@ -3,8 +3,8 @@ script: | super db init -q for order in desc asc; do echo === $order === - super db create -use -q -S 1KB -orderby ts:$order test - seq 1000 | super -c '{ts:this-1,s:f"val{this-1}"}' - | super db load -q - + super db create -use -q -framecap 50 -objectcap 500 -orderby ts:$order test + seq 1000 | super -c '{ts:this-1,s:f"val{this-1}"}' - | super db load -q -inputcap=10 - super db delete -q -where 'ts > 400 and ts <= 500' super db -s -c 'from test | count:=count()' super db delete -q -where 's == "val1" or s == "val999"' diff --git a/db/ztests/deterministic-merge.yaml b/db/ztests/deterministic-merge.yaml index 29a70954e9..9b3c3453f0 100644 --- a/db/ztests/deterministic-merge.yaml +++ b/db/ztests/deterministic-merge.yaml @@ -1,8 +1,8 @@ script: | export SUPER_DB=test super db init -q - super db create -q -S 32B -orderby ts:asc logs - super db load -q -use logs in.sup + super db create -q -orderby ts:asc -objectcap 2 -framecap 2 logs + super db load -q -use logs -inputcap 2 in.sup super db -s -c "from logs" > 1.sup super db -s -c "from logs" > 2.sup diff --git a/db/ztests/ls.yaml b/db/ztests/ls.yaml index 88adb8fb07..3595880ce9 100644 --- a/db/ztests/ls.yaml +++ b/db/ztests/ls.yaml @@ -34,7 +34,8 @@ outputs: ] } ]::order.SortKeys, - threshold: 524288000 + objectcap: 1048576::uint64, + framecap: 20000::uint64 } === { diff --git a/db/ztests/merge-by-addr.yaml b/db/ztests/merge-by-addr.yaml index 3d2d64d336..8a5f5e6805 100644 --- a/db/ztests/merge-by-addr.yaml +++ b/db/ztests/merge-by-addr.yaml @@ -1,8 +1,8 @@ script: | export SUPER_DB=test super db init -q - super db create -q -S 75B -orderby addr:asc logs - super db load -q -use logs in.sup + super db create -q -framecap 5 -objectcap 5 -orderby addr:asc logs + super db load -q -use logs -inputcap 5 in.sup super db -s -c 'from logs | cut addr' inputs: diff --git a/db/ztests/meta.yaml b/db/ztests/meta.yaml index 139c1946e4..5401acd290 100644 --- a/db/ztests/meta.yaml +++ b/db/ztests/meta.yaml @@ -41,7 +41,8 @@ outputs: ] } ]::order.SortKeys, - threshold: 524288000 + objectcap: 1048576::uint64, + framecap: 20000::uint64 } { name: "poolB", @@ -53,7 +54,8 @@ outputs: ] } ]::order.SortKeys, - threshold: 524288000 + objectcap: 1048576::uint64, + framecap: 20000::uint64 } === { diff --git a/db/ztests/objectcap.yaml b/db/ztests/objectcap.yaml new file mode 100644 index 0000000000..c6afc53c7e --- /dev/null +++ b/db/ztests/objectcap.yaml @@ -0,0 +1,11 @@ +script: | + export SUPER_DB=test + super db init -q + super db create -use -q -objectcap 50 -framecap 5 test + seq 100 | super db load -q -inputcap 5 - + super db -s -c "from test:objects | count() by size:=count" + +outputs: + - name: stdout + data: | + {size:50::uint64,count:2} diff --git a/db/ztests/rename.yaml b/db/ztests/rename.yaml index 85e1623155..4a93f929d5 100644 --- a/db/ztests/rename.yaml +++ b/db/ztests/rename.yaml @@ -17,8 +17,8 @@ outputs: regexp: | pool [0-9a-f-]{36} renamed from p2 to p3 === - p1 [0-9a-f-]{36} key ts order desc - p3 [0-9a-f-]{36} key ts order desc + p1 [0-9a-f-]{36} key ts order desc objectcap 1048576 framecap 20000 + p3 [0-9a-f-]{36} key ts order desc objectcap 1048576 framecap 20000 === pool deleted: p1 === diff --git a/service/handlers.go b/service/handlers.go index 117ca7bca9..689f1caeea 100644 --- a/service/handlers.go +++ b/service/handlers.go @@ -259,7 +259,7 @@ func handlePoolPost(c *Core, w *ResponseWriter, r *Request) { if len(req.SortKeys.Keys) > 0 { sortKeys = append(sortKeys, order.NewSortKey(req.SortKeys.Order, req.SortKeys.Keys[0])) } - pool, err := c.root.CreatePool(r.Context(), req.Name, sortKeys, req.Thresh) + pool, err := c.root.CreatePool(r.Context(), req.Name, sortKeys, req.ObjectCap, req.FrameCap) if err != nil { w.Error(err) return diff --git a/service/ztests/curl-create.yaml b/service/ztests/curl-create.yaml index 44db00dd84..f2a8ed9709 100644 --- a/service/ztests/curl-create.yaml +++ b/service/ztests/curl-create.yaml @@ -25,7 +25,8 @@ outputs: ] } ], - threshold: 524288000 + objectcap: 1048576, + framecap: 20000 }, branch: { ts: 0, diff --git a/service/ztests/curl-pool-rename.yaml b/service/ztests/curl-pool-rename.yaml index 7d11c09df6..16c62ee9d2 100644 --- a/service/ztests/curl-pool-rename.yaml +++ b/service/ztests/curl-pool-rename.yaml @@ -31,5 +31,6 @@ outputs: ] } ], - threshold: 524288000 + objectcap: 1048576, + framecap: 20000 } diff --git a/service/ztests/delete-where.yaml b/service/ztests/delete-where.yaml index a2bda0720a..e70fe19082 100644 --- a/service/ztests/delete-where.yaml +++ b/service/ztests/delete-where.yaml @@ -2,8 +2,8 @@ script: | source service.sh for order in desc asc; do echo === $order === | tee >(cat >&2) - super db create -use -q -S 1KB -orderby ts:$order test - seq 1000 | super -c '{ts:this-1,s:f"val{this-1}"}' - | super db load -q - + super db create -use -q -objectcap 50 -framecap 10 -orderby ts:$order test + seq 1000 | super -c '{ts:this-1,s:f"val{this-1}"}' - | super db load -q -inputcap 10 - super db delete -q -where 'ts > 400 and ts <= 500' super db -s -c 'from test | count()' super db delete -q -where 's == "val1" or s == "val999"' diff --git a/service/ztests/ignore-trailing-slash.yaml b/service/ztests/ignore-trailing-slash.yaml index baffef54a4..98258a923b 100644 --- a/service/ztests/ignore-trailing-slash.yaml +++ b/service/ztests/ignore-trailing-slash.yaml @@ -10,4 +10,4 @@ inputs: outputs: - name: stdout regexp: | - test [0-9a-f-]{36} key ts order desc + test [0-9a-f-]{36} key ts order desc objectcap 1048576 framecap 20000 diff --git a/service/ztests/ls-pools.yaml b/service/ztests/ls-pools.yaml index 47feb96517..383c6c3f22 100644 --- a/service/ztests/ls-pools.yaml +++ b/service/ztests/ls-pools.yaml @@ -24,8 +24,8 @@ outputs: pool created: p1 [0-9a-f-]{36} pool created: p2 [0-9a-f-]{36} === - p1 [0-9a-f-]{36} key ts order desc - p2 [0-9a-f-]{36} key ts order desc + p1 [0-9a-f-]{36} key ts order desc objectcap 1048576 framecap 20000 + p2 [0-9a-f-]{36} key ts order desc objectcap 1048576 framecap 20000 === { name: "p1" diff --git a/service/ztests/offline-config-updates.yaml b/service/ztests/offline-config-updates.yaml index 79ede236c0..9cb39e56b9 100644 --- a/service/ztests/offline-config-updates.yaml +++ b/service/ztests/offline-config-updates.yaml @@ -13,7 +13,7 @@ inputs: outputs: - name: stdout regexp: | - test [0-9a-f-]{36} key ts order desc + test [0-9a-f-]{36} key ts order desc objectcap 1048576 framecap 20000 === - name: stderr data: "" diff --git a/sio/anyio/lookup.go b/sio/anyio/lookup.go index 2fb6734dd6..a8368ff93f 100644 --- a/sio/anyio/lookup.go +++ b/sio/anyio/lookup.go @@ -34,7 +34,7 @@ func lookupReader(ctx context.Context, sctx *super.Context, r io.Reader, opts Re case "line": return newVioPuller(sctx, lineio.NewReader(r)), nil case "json": - return jsonio.NewReader(context.Background(), sctx, r, opts.Pushdown, opts.ConcurrentReaders), nil + return jsonio.NewReader(context.Background(), sctx, r, opts.Pushdown, opts.ConcurrentReaders, opts.InputCap), nil case "parquet": return parquetio.NewReader(ctx, sctx, r, opts.Pushdown, opts.ConcurrentReaders) case "sup": diff --git a/sio/anyio/reader.go b/sio/anyio/reader.go index 40dd5b5ce5..486202d534 100644 --- a/sio/anyio/reader.go +++ b/sio/anyio/reader.go @@ -29,6 +29,7 @@ type ReaderOpts struct { Pushdown sbuf.Pushdown ConcurrentReaders int CSV csvio.ReaderOpts + InputCap int } func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, opts ReaderOpts) (vio.Puller, error) { @@ -38,7 +39,11 @@ func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, opts Reade if opts.Format != "" && opts.Format != "auto" { return lookupReader(ctx, sctx, r, opts) } - + if opts.InputCap != 0 { + // XXX There ought to be a better way to do this. Mostly this + // is for testing so not a big deal. + sbuf.PullerBatchValues = opts.InputCap + } track := NewTrack(r) bsupErr := bsup.Probe(track) @@ -74,7 +79,7 @@ func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, opts Reade // sake of tests. jsonErr := isJSONStream(track, 10) if jsonErr == nil { - return jsonio.NewReader(ctx, sctx, track.Reader(), opts.Pushdown, opts.ConcurrentReaders), nil + return jsonio.NewReader(ctx, sctx, track.Reader(), opts.Pushdown, opts.ConcurrentReaders, opts.InputCap), nil } jsonErr = fmt.Errorf("json: %w", jsonErr) track.Reset() @@ -152,7 +157,7 @@ func isCSVStream(track *Track, delim rune, name string) error { } func isJSONStream(track *Track, want int) error { - r := jsonio.NewValReader(track) + r := jsonio.NewValReader(track, 0) for range want { if _, err := r.Next(); err != nil { if errors.Is(err, io.EOF) { diff --git a/sio/dbio/writer.go b/sio/dbio/writer.go index 220162b658..a6fea5cdad 100644 --- a/sio/dbio/writer.go +++ b/sio/dbio/writer.go @@ -4,6 +4,7 @@ import ( "bytes" "fmt" "io" + "strconv" "strings" "uuid" @@ -117,6 +118,10 @@ func formatPoolConfig(b *bytes.Buffer, p *pools.Config) { b.WriteString(p.SortKeys.Primary().Path.String()) b.WriteString(" order ") b.WriteString(p.SortKeys.Primary().Order.String()) + b.WriteString(" objectcap ") + b.WriteString(strconv.FormatUint(p.ObjectCap, 10)) + b.WriteString(" framecap ") + b.WriteString(strconv.FormatUint(p.FrameCap, 10)) b.WriteByte('\n') } diff --git a/sio/jsonio/reader.go b/sio/jsonio/reader.go index 8732246fe1..964afc2a2f 100644 --- a/sio/jsonio/reader.go +++ b/sio/jsonio/reader.go @@ -17,7 +17,7 @@ import ( "github.com/superdb/super/vector" ) -var VecBatchSize uint32 = 10 * 1024 +const defaultInputCap = 10 * 1024 type Reader struct { sctx *super.Context @@ -28,7 +28,10 @@ type Reader struct { hasClosed atomic.Bool } -func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, p sbuf.Pushdown, concurrentReaders int) *Reader { +func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, p sbuf.Pushdown, concurrentReaders, inputCap int) *Reader { + if inputCap == 0 { + inputCap = defaultInputCap + } var fields field.List if p != nil { fields = p.Projection().Paths() @@ -36,7 +39,7 @@ func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, p sbuf.Pus return &Reader{ sctx: sctx, ctx: ctx, - stream: newStream(ctx, r, concurrentReaders), + stream: newStream(ctx, r, concurrentReaders, inputCap), projection: fields, } } diff --git a/sio/jsonio/stream.go b/sio/jsonio/stream.go index 4bb9bf0e9f..df156c75c6 100644 --- a/sio/jsonio/stream.go +++ b/sio/jsonio/stream.go @@ -10,19 +10,21 @@ import ( ) type stream struct { - r io.Reader - ch chan result - done chan struct{} - once sync.Once - ctx context.Context + r io.Reader + ch chan result + done chan struct{} + once sync.Once + ctx context.Context + inputCap int } -func newStream(ctx context.Context, r io.Reader, n int) *stream { +func newStream(ctx context.Context, r io.Reader, n int, inputCap int) *stream { return &stream{ - r: r, - ch: make(chan result, n), - ctx: ctx, - done: make(chan struct{}), + r: r, + ch: make(chan result, n), + ctx: ctx, + done: make(chan struct{}), + inputCap: inputCap, } } @@ -44,7 +46,7 @@ func (s *stream) next() (*vector.BytesTable, int, error) { func (s *stream) run() { defer close(s.ch) - r := NewValReader(s.r) + r := NewValReader(s.r, s.inputCap) for { batch, startLineNum, err := readBatch(r) select { @@ -66,20 +68,20 @@ func (s *stream) close() { var bytesTablePool sync.Pool -func newBytesTable() *vector.BytesTable { +func newBytesTable(inputCap int) *vector.BytesTable { b, ok := bytesTablePool.Get().(*vector.BytesTable) if !ok { - b = new(vector.NewBytesTableEmpty(VecBatchSize)) + b = new(vector.NewBytesTableEmpty(uint32(inputCap))) } b.Reset() return b } func readBatch(r *valReader) (*vector.BytesTable, int, error) { - t := newBytesTable() + t := newBytesTable(r.inputCap) start := r.lineNumber() var err error - for range VecBatchSize { + for range r.inputCap { var b []byte if b, err = r.Next(); err != nil { break diff --git a/sio/jsonio/valreader.go b/sio/jsonio/valreader.go index e094797ab1..848f3a0e2d 100644 --- a/sio/jsonio/valreader.go +++ b/sio/jsonio/valreader.go @@ -13,15 +13,16 @@ import ( const maxSize = 512 * 1024 * 1024 type valReader struct { - r io.Reader - buf []byte - cursor []byte - EOF bool - line int + r io.Reader + buf []byte + cursor []byte + EOF bool + line int + inputCap int } -func NewValReader(r io.Reader) *valReader { - return &valReader{r: r, buf: make([]byte, 512*1024)} +func NewValReader(r io.Reader, inputCap int) *valReader { + return &valReader{r: r, buf: make([]byte, 512*1024), inputCap: inputCap} } func (r *valReader) Next() ([]byte, error) {