Repository: polarsignals/frostdb (go)

Issue:
Fix Release() arrow callbacks
Our usage of Release of arrow records during callback is inconsistent. If operators are utilized they won't release the callback record, but will create a new one which the final callback can release. 

However if there is no operators and it's just straight iteration then a release on the final callback will cause too many releases to occur.

This is easy to reproduce by running unit tests with `-tags assert`

The final callback should always be able to safely release the given record.

We should detect this automatically but adding `-tags assert` to our github workflow.


Discussion:
Oh cool didn’t know about the assert tag!
I could take a look at it
It's all yours @albertlockett 
@thorfour I think the issue we're having is us calling HashAggregate's Close() method multiple times (as that's where it releases its arrow records).

For a query like this:
```golang
err = engine.ScanTable("simple_table").
Aggregate(
	[]logicalplan.Expr{
		logicalplan.Sum(logicalplan.Col("value")),
	},
	[]logicalplan.Expr{
		logicalplan.Col("names.first_name"),
	},
).
```

We end up w/ a physical plan like this:
```
     1                           2                                           3                 4
TableScan [concurrent] - HashAggregate (sum(value) by names.first_name) - Synchronizer - HashAggregate (sum(value) by names.first_name)
                       - HashAggregate (sum(value) by names.first_name) -
   multiple plans ->   ... for each TableScan.plan
                       - HashAggregate (sum(value) by names.first_name) -
 ```
 
 We end up calling the `Close()` method for each of tablescan's plans (2) 
 https://github.com/polarsignals/frostdb/blob/de942d0ce46e8ca34b3f08cc96f3ce61da96f26a/query/physicalplan/physicalplan.go#L125-L129
 
 This in turn will call Synchronizer's (3) `Close` multiple times, which will call HashAggregate's (4) `Close` method mulitple times.
 
 I think it's reasonable for a given step in the physical plan to expect to only have its `Close()` method called once, with the exception of possibly synchronizer. To that end, I think it's reasonable for us to add some logic into synchronizer to call `next.Close()` only once.
 
 What do you think of that plan? 
 
I think that makes a lot of sense. The synchronizer should have the logic to only call close once.

I think when I first wrote this issue there was also an issue with performing a query with no operators. So that may be something to look into as well.
You were thinking something like this?
```golang
err = engine.ScanTable("simple_table").
  Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
	  fmt.Printf("%v", r)
	  return nil
  })
```
That actually seems to work alright.

We have a number of tests that seem to release the record in the  table iterator callback, which I'll clean up as well
https://github.com/polarsignals/frostdb/blob/de942d0ce46e8ca34b3f08cc96f3ce61da96f26a/table_test.go#L176

Patch:
diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml
--- a/.github/workflows/go.yml
+++ b/.github/workflows/go.yml
@@ -19,4 +19,4 @@ jobs:
         go-version: '1.20'
 
     - name: Test
-      run: go test -short -race -tags debug -v ./...
+      run: go test -short -race -tags assert,debug -v ./...
diff --git a/Makefile b/Makefile
--- a/Makefile
+++ b/Makefile
@@ -1,5 +1,5 @@
 test:
-	go test -tags debug -race ./...
+	go test -tags assert,debug -race ./...
 
 .PHONY: gen/proto
 gen/proto:
diff --git a/query/physicalplan/synchronize.go b/query/physicalplan/synchronize.go
--- a/query/physicalplan/synchronize.go
+++ b/query/physicalplan/synchronize.go
@@ -17,12 +17,15 @@ type Synchronizer struct {
 	next    PhysicalPlan
 	nextMtx sync.Mutex
 	running *atomic.Int64
+	open    *atomic.Int64
 }
 
 func Synchronize(concurrency int) *Synchronizer {
 	running := &atomic.Int64{}
 	running.Add(int64(concurrency))
-	return &Synchronizer{running: running}
+	open := &atomic.Int64{}
+	open.Add(int64(concurrency))
+	return &Synchronizer{running: running, open: open}
 }
 
 func (m *Synchronizer) Callback(ctx context.Context, r arrow.Record) error {
@@ -62,5 +65,12 @@ func (m *Synchronizer) Draw() *Diagram {
 }
 
 func (m *Synchronizer) Close() {
+	open := m.open.Add(-1)
+	if open < 0 {
+		panic("too many Synchronizer Close calls")
+	}
+	if open > 0 {
+		return
+	}
 	m.next.Close()
 }


Test patch:
diff --git a/table_test.go b/table_test.go
--- a/table_test.go
+++ b/table_test.go
@@ -173,7 +173,6 @@ func TestTable(t *testing.T) {
 			pool,
 			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
 				t.Log(ar)
-				defer ar.Release()
 				return nil
 			}},
 		)
@@ -288,7 +287,6 @@ func Test_Table_Concurrency(t *testing.T) {
 					pool,
 					[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
 						totalrows += ar.NumRows()
-						defer ar.Release()
 
 						return nil
 					}},
@@ -570,7 +568,6 @@ func Test_Table_ReadIsolation(t *testing.T) {
 			pool,
 			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
 				rows += ar.NumRows()
-				defer ar.Release()
 
 				return nil
 			}},
@@ -593,7 +590,6 @@ func Test_Table_ReadIsolation(t *testing.T) {
 			pool,
 			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
 				rows += ar.NumRows()
-				defer ar.Release()
 
 				return nil
 			}},
@@ -841,7 +837,6 @@ func Test_Table_Bloomfilter(t *testing.T) {
 			tx,
 			pool,
 			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
-				defer ar.Release()
 				iterations++
 				return nil
 			}},
@@ -917,7 +912,6 @@ func Test_DoubleTable(t *testing.T) {
 			tx,
 			pool,
 			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
-				defer ar.Release()
 				require.Equal(t, value, ar.Column(1).(*array.Float64).Value(0))
 				return nil
 			}},