change database load to partition ingest into multiple objects

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 {

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

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}

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")
 }
 

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")

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
 	}

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
 }

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"

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

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
 	}

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

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

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,
 	}
 }
 

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
 	}