Add vectorized JSON reader

Author: mattnibs

compiler/semantic/op.go

@@ -284,7 +284,7 @@ func (t *translator) file(n ast.Node, name string, args []ast.OpArg) sem.Op {
 	if format == "" {
 		format = sio.FormatFromPath(name)
 	}
-	if format == "csup" || format == "parquet" {
+	if format == "csup" || format == "parquet" || format == "json" {
 		t.hasVectorizedInput = true
 	}
 	typ, err := t.fileType(name, format)

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de
 	github.com/aws/aws-sdk-go v1.36.17
 	github.com/axiomhq/hyperloglog v0.2.5
+	github.com/bytedance/sonic v1.15.0
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/goccy/go-yaml v1.19.0
 	github.com/golang-jwt/jwt/v4 v4.4.3
@@ -49,7 +50,10 @@ require (
 	github.com/apache/thrift v0.22.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.12.0 // indirect
+	github.com/bytedance/gopkg v0.1.3 // indirect
+	github.com/bytedance/sonic/loader v0.5.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
@@ -75,9 +79,11 @@ require (
 	github.com/prometheus/common v0.37.0 // indirect
 	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
+	golang.org/x/arch v0.0.0-20210923205945-b76863e36670 // indirect
 	golang.org/x/mod v0.32.0 // indirect
 	golang.org/x/net v0.49.0 // indirect
 	golang.org/x/telemetry v0.0.0-20260109210033-bd525da824e2 // indirect

go.sum

@@ -66,6 +66,12 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bits-and-blooms/bitset v1.12.0 h1:U/q1fAF7xXRhFCrhROzIfffYnu+dlS38vCZtmFVPHmA=
 github.com/bits-and-blooms/bitset v1.12.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
+github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M=
+github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM=
+github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE=
+github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k=
+github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE=
+github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -75,6 +81,8 @@ github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWR
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M=
+github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -326,16 +334,24 @@ github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6Mwd
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/teamortix/golang-wasm/wasm v0.0.0-20230719150929-5d000994c833 h1:PE/ebx5HZAsK42Bs/syRaSWBInfZpj9RifI/sEhGHvo=
 github.com/teamortix/golang-wasm/wasm v0.0.0-20230719150929-5d000994c833/go.mod h1:nskvTyoGIaAsC+664SkRitVI1ft6dm1xerCr50YZsnY=
+github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
+github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
 github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
@@ -378,6 +394,8 @@ go.uber.org/multierr v1.8.0 h1:dg6GjLku4EH+249NNmoIciG9N/jURbDG+pFlTkhzIC8=
 go.uber.org/multierr v1.8.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak=
 go.uber.org/zap v1.23.0 h1:OjGQ5KQDEUawVHxNwQgPpiypGHOxo2mNZsOqTak4fFY=
 go.uber.org/zap v1.23.0/go.mod h1:D+nX8jyLsMHMYrln8A0rJjFt/T/9/bGgIhAqxv5URuY=
+golang.org/x/arch v0.0.0-20210923205945-b76863e36670 h1:18EFjUmQOcUvxNYSkA6jO9VAiXCnxFY6NyDX0bHDmkU=
+golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=

runtime/exec/environment.go

@@ -16,6 +16,7 @@ import (
 	"github.com/brimdata/super/sbuf"
 	"github.com/brimdata/super/sio/anyio"
 	"github.com/brimdata/super/sio/csupio"
+	"github.com/brimdata/super/sio/fjsonio"
 	"github.com/brimdata/super/sio/parquetio"
 	"github.com/brimdata/super/vector"
 	"github.com/brimdata/super/vector/vio"
@@ -188,6 +189,8 @@ func (e *Environment) VectorOpen(ctx context.Context, sctx *super.Context, path,
 		puller, err = csupio.NewVectorReader(ctx, sctx, reader, p, concurrentReaders)
 	case "parquet":
 		puller, err = parquetio.NewVectorReader(ctx, sctx, reader, p, concurrentReaders)
+	case "fjson":
+		puller = fjsonio.NewVectorReader(ctx, sctx, reader, p, concurrentReaders)
 	default:
 		var sbufPuller sbuf.Puller
 		sbufPuller, err = e.Open(ctx, sctx, path, format, p)

sio/anyio/lookup.go

@@ -1,15 +1,18 @@
 package anyio
 
 import (
+	"context"
 	"fmt"
 	"io"
 
 	"github.com/brimdata/super"
+	"github.com/brimdata/super/sbuf"
 	"github.com/brimdata/super/sio"
 	"github.com/brimdata/super/sio/arrowio"
 	"github.com/brimdata/super/sio/bsupio"
 	"github.com/brimdata/super/sio/csupio"
 	"github.com/brimdata/super/sio/csvio"
+	"github.com/brimdata/super/sio/fjsonio"
 	"github.com/brimdata/super/sio/jsonio"
 	"github.com/brimdata/super/sio/jsupio"
 	"github.com/brimdata/super/sio/lineio"
@@ -34,6 +37,10 @@ func lookupReader(sctx *super.Context, r io.Reader, opts ReaderOpts) (sio.ReadCl
 		return sio.NopReadCloser(csvio.NewReader(sctx, r, opts.CSV)), nil
 	case "line":
 		return sio.NopReadCloser(lineio.NewReader(r)), nil
+	case "fjson":
+		v := fjsonio.NewVectorReader(context.Background(), sctx, r, nil, 1)
+		puller := sbuf.NewMaterializer(v)
+		return sio.NopReadCloser(sbuf.PullerReader(puller)), nil
 	case "json":
 		return sio.NopReadCloser(jsonio.NewReader(sctx, r)), nil
 	case "parquet":

sio/fjsonio/jsonvec/array.go

@@ -0,0 +1,40 @@
+package jsonvec
+
+import "github.com/brimdata/super/vector"
+
+var _ Value = (*Array)(nil)
+
+type Array struct {
+	Offsets []uint32
+	Inner   Value
+}
+
+func NewArray() *Array {
+	return &Array{
+		Offsets: []uint32{0},
+		Inner:   Unknown{},
+	}
+}
+
+func (a *Array) BeginArray() Value { return a }
+func (a *Array) EnterArray() Value { return a.Inner }
+
+func (a *Array) EndArray(inner Value) {
+	a.Inner = inner
+	n := a.Inner.Len()
+	if n == 0 {
+		a.Inner = new(Null)
+	}
+	a.Offsets = append(a.Offsets, n)
+}
+
+func (a *Array) OnNull() Value           { return ToUnion(a).OnNull() }
+func (a *Array) OnBool(v bool) Value     { return ToUnion(a).OnBool(v) }
+func (a *Array) OnInt(v int64) Value     { return ToUnion(a).OnInt(v) }
+func (a *Array) OnFloat(v float64) Value { return ToUnion(a).OnFloat(v) }
+func (a *Array) OnString(v string) Value { return ToUnion(a).OnString(v) }
+func (a *Array) BeginRecord() Value      { return ToUnion(a).BeginRecord() }
+func (a *Array) Field(v string) Value    { panic("system error") }
+func (a *Array) EndRecord()              { panic("system error") }
+func (a *Array) Kind() vector.Kind       { return vector.KindNull }
+func (a *Array) Len() uint32             { return uint32(len(a.Offsets)) - 1 }

sio/fjsonio/jsonvec/bool.go

@@ -0,0 +1,30 @@
+package jsonvec
+
+import (
+	"github.com/brimdata/super/vector"
+	"github.com/brimdata/super/vector/bitvec"
+)
+
+type Bool struct {
+	Value bitvec.Bits
+}
+
+var _ Value = (*Bool)(nil)
+
+func (b *Bool) OnBool(v bool) Value {
+	b.Value.Append(v)
+	return b
+}
+
+func (b *Bool) OnNull() Value           { return ToUnion(b).OnNull() }
+func (b *Bool) OnString(v string) Value { return ToUnion(b).OnString(v) }
+func (b *Bool) OnInt(v int64) Value     { return ToUnion(b).OnInt(v) }
+func (b *Bool) OnFloat(v float64) Value { return ToUnion(b).OnFloat(v) }
+func (b *Bool) BeginRecord() Value      { return ToUnion(b).BeginRecord() }
+func (b *Bool) Field(k string) Value    { panic("system error") }
+func (b *Bool) EndRecord()              { panic("system error") }
+func (b *Bool) BeginArray() Value       { return ToUnion(b).BeginArray() }
+func (b *Bool) EnterArray() Value       { panic("system error") }
+func (b *Bool) EndArray(Value)          { panic("system error") }
+func (b *Bool) Kind() vector.Kind       { return vector.KindBool }
+func (b *Bool) Len() uint32             { return b.Value.Len() }

sio/fjsonio/jsonvec/builder.go

@@ -0,0 +1,92 @@
+package jsonvec
+
+import (
+	"encoding/json"
+
+	"github.com/bytedance/sonic/ast"
+)
+
+var _ ast.Visitor = (*Builder)(nil)
+
+type Builder struct {
+	stack []Value
+}
+
+func NewBuilder() *Builder {
+	return &Builder{
+		stack: []Value{Unknown{}},
+	}
+}
+
+func (b *Builder) Reset() {
+	b.stack = b.stack[:1]
+}
+
+func (b *Builder) OnNull() error {
+	b.stack[len(b.stack)-1] = b.tos().OnNull()
+	return nil
+}
+
+func (b *Builder) OnBool(v bool) error {
+	b.stack[len(b.stack)-1] = b.tos().OnBool(v)
+	return nil
+}
+
+func (b *Builder) OnInt64(v int64, _ json.Number) error {
+	b.stack[len(b.stack)-1] = b.tos().OnInt(v)
+	return nil
+}
+
+func (b *Builder) OnFloat64(v float64, _ json.Number) error {
+	b.stack[len(b.stack)-1] = b.tos().OnFloat(v)
+	return nil
+}
+
+func (b *Builder) OnString(v string) error {
+	b.stack[len(b.stack)-1] = b.tos().OnString(v)
+	return nil
+}
+
+func (b *Builder) OnObjectBegin(capacity int) error {
+	b.stack[len(b.stack)-1] = b.tos().BeginRecord()
+	b.push(nil)
+	return nil
+}
+
+func (b *Builder) OnObjectKey(name string) error {
+	b.pop()
+	b.push(b.tos().Field(name))
+	return nil
+}
+
+func (b *Builder) OnObjectEnd() error {
+	b.pop()
+	b.tos().EndRecord()
+	return nil
+}
+
+func (b *Builder) OnArrayBegin(capacity int) error {
+	b.stack[len(b.stack)-1] = b.tos().BeginArray()
+	b.push(b.tos().EnterArray())
+	return nil
+}
+
+func (b *Builder) OnArrayEnd() error {
+	inner := b.pop()
+	b.tos().EndArray(inner)
+	return nil
+}
+
+func (b *Builder) tos() Value {
+	return b.stack[len(b.stack)-1]
+}
+
+func (b *Builder) push(val Value) {
+	b.stack = append(b.stack, val)
+}
+
+func (b *Builder) pop() Value {
+	v := b.tos()
+	b.stack = b.stack[:len(b.stack)-1]
+	return v
+}

sio/fjsonio/jsonvec/float.go

@@ -0,0 +1,34 @@
+package jsonvec
+
+import (
+	"github.com/brimdata/super"
+	"github.com/brimdata/super/vector"
+)
+
+var _ Value = (*Float)(nil)
+
+type Float struct {
+	Value *vector.Float
+}
+
+func NewFloat() *Float {
+	return &Float{Value: vector.NewFloatEmpty(super.TypeFloat64, 0)}
+}
+
+func (f *Float) OnFloat(v float64) Value {
+	f.Value.Append(v)
+	return f
+}
+
+func (f *Float) OnNull() Value           { return ToUnion(f).OnNull() }
+func (f *Float) OnBool(v bool) Value     { return ToUnion(f).OnBool(v) }
+func (f *Float) OnInt(v int64) Value     { return ToUnion(f).OnInt(v) }
+func (f *Float) OnString(v string) Value { return ToUnion(f).OnString(v) }
+func (f *Float) BeginRecord() Value      { return ToUnion(f).BeginRecord() }
+func (f *Float) Field(key string) Value  { panic("system error") }
+func (f *Float) EndRecord()              { panic("system error") }
+func (f *Float) BeginArray() Value       { return ToUnion(f).BeginArray() }
+func (f *Float) EnterArray() Value       { panic("system error") }
+func (f *Float) EndArray(Value)          { panic("system error") }
+func (f *Float) Kind() vector.Kind       { return vector.KindFloat }
+func (f *Float) Len() uint32             { return f.Value.Len() }

sio/fjsonio/jsonvec/int.go

@@ -0,0 +1,34 @@
+package jsonvec
+
+import (
+	"github.com/brimdata/super"
+	"github.com/brimdata/super/vector"
+)
+
+var _ Value = (*Int)(nil)
+
+type Int struct {
+	Value *vector.Int
+}
+
+func NewInt() *Int {
+	return &Int{Value: vector.NewIntEmpty(super.TypeInt64, 0)}
+}
+
+func (i *Int) OnInt(v int64) Value {
+	i.Value.Append(v)
+	return i
+}
+
+func (i *Int) OnNull() Value           { return ToUnion(i).OnNull() }
+func (i *Int) OnBool(v bool) Value     { return ToUnion(i).OnBool(v) }
+func (i *Int) OnFloat(v float64) Value { return ToUnion(i).OnFloat(v) }
+func (i *Int) OnString(v string) Value { return ToUnion(i).OnString(v) }
+func (i *Int) BeginRecord() Value      { return ToUnion(i).BeginRecord() }
+func (i *Int) Field(key string) Value  { panic("system error") }
+func (i *Int) EndRecord()              { panic("system error") }
+func (i *Int) BeginArray() Value       { return ToUnion(i).BeginArray() }
+func (i *Int) EnterArray() Value       { panic("system error") }
+func (i *Int) EndArray(Value)          { panic("system error") }
+func (i *Int) Kind() vector.Kind       { return vector.KindInt }
+func (i *Int) Len() uint32             { return i.Value.Len() }