Repository: polarsignals/frostdb (go)

Issue:
Flaky unit test during CI
```
=== RUN   TestAggregationProjection
panic: runtime error: index out of range [2] with length 2

goroutine 286 [running]:
github.com/apache/arrow/go/v8/arrow/array.(*Binary).Value(...)
	/home/runner/go/pkg/mod/github.com/apache/arrow/go/v8@v8.0.1/arrow/array/binary.go:50
github.com/polarsignals/frostdb/query/physicalplan.hashBinaryArray(0xc0003fd3e0)
	/home/runner/work/frostdb/frostdb/query/physicalplan/aggregate.go:207 +0x475
github.com/polarsignals/frostdb/query/physicalplan.hashArray({0x150fec0?, 0xc0003fd3e0?})
	/home/runner/work/frostdb/frostdb/query/physicalplan/aggregate.go:193 +0x127
github.com/polarsignals/frostdb/query/physicalplan.(*HashAggregate).Callback(0xc0000dcfc0, {0x4aad25?, 0xde40b4?}, {0x1510b58, 0xc000152450})
	/home/runner/work/frostdb/frostdb/query/physicalplan/aggregate.go:290 +0xf70
github.com/polarsignals/frostdb/query/physicalplan.(*Synchronizer).Callback(0xc0003ed020, {0x15082e0, 0xc0001522d0}, {0x1510b58, 0xc000152450})
	/home/runner/work/frostdb/frostdb/query/physicalplan/synchronize.go:34 +0x10d
github.com/polarsignals/frostdb/query/physicalplan.(*HashAggregate).Finish(0xc0000dca20, {0x1508238, 0xc0002f4b80})
	/home/runner/work/frostdb/frostdb/query/physicalplan/aggregate.go:411 +0x924
github.com/polarsignals/frostdb/query/physicalplan.(*TableScan).Execute.func2()
	/home/runner/work/frostdb/frostdb/query/physicalplan/physicalplan.go:142 +0x62
golang.org/x/sync/errgroup.(*Group).Go.func1()
	/home/runner/go/pkg/mod/golang.org/x/sync@v0.0.0-20220601150217-0de741cfad7f/errgroup/errgroup.go:75 +0x87
created by golang.org/x/sync/errgroup.(*Group).Go
	/home/runner/go/pkg/mod/golang.org/x/sync@v0.0.0-20220601150217-0de741cfad7f/errgroup/errgroup.go:72 +0x12f
FAIL	github.com/polarsignals/frostdb	0.127s
```

Pull request:
builder: append offset when appending NULL in OptBinaryBuilder
Fixes #270 

This PR also adds the optimized builders back into the hash aggregate and distinct given they were removed until the issue was resolved.

Patch:
diff --git a/dynparquet/schema.go b/dynparquet/schema.go
index 17aae7f..841a81e 100644
--- a/dynparquet/schema.go
+++ b/dynparquet/schema.go
@@ -852,6 +852,10 @@ func (s *Schema) SerializeBuffer(w io.Writer, buffer *Buffer) error {
 	return nil
 }
 
+// bloomFilterBitsPerValue is the number of bits used by the bloom filter. 10
+// was the default value used by parquet before it was made configurable.
+const bloomFilterBitsPerValue = 10
+
 // NewWriter returns a new parquet writer with a concrete parquet schema
 // generated using the given concrete dynamic column names.
 func (s *Schema) NewWriter(w io.Writer, dynamicColumns map[string][]string) (*parquet.Writer, error) {
@@ -873,7 +877,9 @@ func (s *Schema) NewWriter(w io.Writer, dynamicColumns map[string][]string) (*pa
 			continue
 		}
 
-		bloomFilterColumns = append(bloomFilterColumns, parquet.SplitBlockFilter(col.Path()...))
+		bloomFilterColumns = append(
+			bloomFilterColumns, parquet.SplitBlockFilter(bloomFilterBitsPerValue, col.Path()...),
+		)
 	}
 
 	return parquet.NewWriter(w,
diff --git a/go.mod b/go.mod
index 17c0518..6970058 100644
--- a/go.mod
+++ b/go.mod
@@ -16,7 +16,7 @@ require (
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/pingcap/tidb/parser v0.0.0-20220921115303-5aab87679fde
 	github.com/prometheus/client_golang v1.12.2
-	github.com/segmentio/parquet-go v0.0.0-20221207043206-db0f5ce13cf1
+	github.com/segmentio/parquet-go v0.0.0-20221209161419-3f277a904e0e
 	github.com/stretchr/testify v1.7.2
 	github.com/thanos-io/objstore v0.0.0-20220715165016-ce338803bc1e
 	github.com/tidwall/wal v1.1.7
diff --git a/go.sum b/go.sum
index be3b450..2adecaa 100644
--- a/go.sum
+++ b/go.sum
@@ -445,8 +445,8 @@ github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg
 github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg=
 github.com/segmentio/encoding v0.3.5 h1:UZEiaZ55nlXGDL92scoVuw00RmiRCazIEmvPSbSvt8Y=
 github.com/segmentio/encoding v0.3.5/go.mod h1:n0JeuIqEQrQoPDGsjo8UNd1iA0U8d8+oHAA4E3G3OxM=
-github.com/segmentio/parquet-go v0.0.0-20221207043206-db0f5ce13cf1 h1:MajB8YDwcYNncLcHaw59hh8upwuMDRsXVZxJH2Kn/UQ=
-github.com/segmentio/parquet-go v0.0.0-20221207043206-db0f5ce13cf1/go.mod h1:SclLlCfB7c7CH0YerV+OtYmZExyK5rhVOd6UT90erVw=
+github.com/segmentio/parquet-go v0.0.0-20221209161419-3f277a904e0e h1:oB6VqjFrx6HC1G736iZIe3xII1Zj/QCWInKdUw98AtU=
+github.com/segmentio/parquet-go v0.0.0-20221209161419-3f277a904e0e/go.mod h1:SclLlCfB7c7CH0YerV+OtYmZExyK5rhVOd6UT90erVw=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
diff --git a/pqarrow/builder/optbuilders.go b/pqarrow/builder/optbuilders.go
index bbcc124..e10d201 100644
--- a/pqarrow/builder/optbuilders.go
+++ b/pqarrow/builder/optbuilders.go
@@ -121,6 +121,7 @@ func (b *OptBinaryBuilder) Release() {
 // AppendNull adds a new null value to the array being built. This is slow,
 // don't use it.
 func (b *OptBinaryBuilder) AppendNull() {
+	b.offsets = append(b.offsets, uint32(len(b.data)))
 	b.builderBase.AppendNulls(1)
 }
 
diff --git a/query/physicalplan/aggregate.go b/query/physicalplan/aggregate.go
index 4a77335..3bad188 100644
--- a/query/physicalplan/aggregate.go
+++ b/query/physicalplan/aggregate.go
@@ -111,9 +111,9 @@ type HashAggregate struct {
 	pool                  memory.Allocator
 	tracer                trace.Tracer
 	resultColumnName      string
-	groupByCols           map[string]array.Builder
+	groupByCols           map[string]builder.ColumnBuilder
 	colOrdering           []string
-	arraysToAggregate     []array.Builder
+	arraysToAggregate     []builder.ColumnBuilder
 	hashToAggregate       map[uint64]int
 	groupByColumnMatchers []logicalplan.Expr
 	columnToAggregate     logicalplan.Expr
@@ -143,9 +143,9 @@ func NewHashAggregate(
 		pool:              pool,
 		tracer:            tracer,
 		resultColumnName:  resultColumnName,
-		groupByCols:       map[string]array.Builder{},
+		groupByCols:       map[string]builder.ColumnBuilder{},
 		colOrdering:       []string{},
-		arraysToAggregate: make([]array.Builder, 0),
+		arraysToAggregate: make([]builder.ColumnBuilder, 0),
 		hashToAggregate:   map[uint64]int{},
 		columnToAggregate: columnToAggregate,
 		// TODO: Matchers can be optimized to be something like a radix tree or just a fast-lookup datastructure for exact matches or prefix matches.
@@ -347,7 +347,7 @@ func (a *HashAggregate) Callback(ctx context.Context, r arrow.Record) error {
 
 		k, ok := a.hashToAggregate[hash]
 		if !ok {
-			agg := array.NewBuilder(a.pool, columnToAggregate.DataType())
+			agg := builder.NewBuilder(a.pool, columnToAggregate.DataType())
 			a.arraysToAggregate = append(a.arraysToAggregate, agg)
 			k = len(a.arraysToAggregate) - 1
 			a.hashToAggregate[hash] = k
@@ -358,7 +358,7 @@ func (a *HashAggregate) Callback(ctx context.Context, r arrow.Record) error {
 
 				groupByCol, found := a.groupByCols[fieldName]
 				if !found {
-					groupByCol = array.NewBuilder(a.pool, groupByFields[j].Type)
+					groupByCol = builder.NewBuilder(a.pool, groupByFields[j].Type)
 					a.groupByCols[fieldName] = groupByCol
 					a.colOrdering = append(a.colOrdering, fieldName)
 				}
diff --git a/query/physicalplan/distinct.go b/query/physicalplan/distinct.go
index cd717c0..fe51abd 100644
--- a/query/physicalplan/distinct.go
+++ b/query/physicalplan/distinct.go
@@ -81,9 +81,9 @@ func (d *Distinction) Callback(ctx context.Context, r arrow.Record) error {
 		}
 	}
 
-	resBuilders := make([]array.Builder, 0, len(distinctArrays))
+	resBuilders := make([]builder.ColumnBuilder, 0, len(distinctArrays))
 	for _, arr := range distinctArrays {
-		resBuilders = append(resBuilders, array.NewBuilder(d.pool, arr.DataType()))
+		resBuilders = append(resBuilders, builder.NewBuilder(d.pool, arr.DataType()))
 	}
 	rows := int64(0)
 


Test patch:
diff --git a/pqarrow/builder/optbuilders_test.go b/pqarrow/builder/optbuilders_test.go
new file mode 100644
index 0000000..62ba211
--- /dev/null
+++ b/pqarrow/builder/optbuilders_test.go
@@ -0,0 +1,25 @@
+package builder_test
+
+import (
+	"testing"
+
+	"github.com/apache/arrow/go/v8/arrow"
+	"github.com/apache/arrow/go/v8/arrow/array"
+	"github.com/stretchr/testify/require"
+
+	"github.com/polarsignals/frostdb/pqarrow/builder"
+)
+
+// https://github.com/polarsignals/frostdb/issues/270
+func TestIssue270(t *testing.T) {
+	b := builder.NewOptBinaryBuilder(arrow.BinaryTypes.Binary)
+	b.AppendNull()
+	const expString = "hello"
+	b.Append([]byte(expString))
+	require.Equal(t, b.Len(), 2)
+
+	a := b.NewArray().(*array.Binary)
+	require.Equal(t, a.Len(), 2)
+	require.True(t, a.IsNull(0))
+	require.Equal(t, string(a.Value(1)), expString)
+}