Repository: polarsignals/frostdb (go)

Issue:
memory leak for PredicateFilter
I tested this test case, I found PredicateFilter has a memory leak:

TableScan [concurrent] - Projection (2 * 3 as va) - PredicateFilter (va > 6) - Synchronizer

<details><summary>Details</summary>
<p>

        exec [0 args]
        select 2 * 3 as va having va > 6
        ----
    checked_allocator.go:186: LEAK of 192 bytes FROM
        	github.com/apache/arrow/go/v14/arrow/array.(*Int64Builder).newData+2f
        	github.com/asubiotto/arrow/go/v14@v14.0.0-20231129090719-b321865d34e9/arrow/array/numericbuilder.gen.go:181
    checked_allocator.go:186: LEAK of 64 bytes FROM
        	github.com/apache/arrow/go/v14/arrow/array.(*builder).init+af
github.com/asubiotto/arrow/go/v14@v14.0.0-20231129090719-b321865d34e9/arrow/array/builder.go:142
    logic_test.go:140: invalid memory size exp=0, got=256

</p>
</details> 

The cause of the problem is the same as #695

physicalplan/binaryscalarexpr.go
```go
func ArrayScalarCompute(funcName string, left arrow.Array, right scalar.Scalar) (*Bitmap, error) {
	equalsResult, err := compute.CallFunction(context.TODO(), funcName, nil, compute.NewDatum(left), compute.NewDatum(right))
	if err != nil {
		if errors.Unwrap(err).Error() == "not implemented" {
			return nil, ErrUnsupportedBinaryOperation
		}
		return nil, fmt.Errorf("error calling equal function: %w", err)
	}
``` 
 compute.NewDatum(left), compute.NewDatum(right), increases ref count,not release.


Discussion:
@jicanghaixb thanks for reporting and awesome investigation work

You are right, there is no need owning `left` and `right`.

It will be nice to just submit a PR.

- add the logic test that trigger the memory leak 
- use  `compute.NewDatumWithoutOwning` instead of  `compute.NewDatum` in the snippet you shared.

I will help reviewing it. Just mention me on the PR message so I can get notified when it is ready.

Just let me know if you need help with anything( to get you up to speed with your PR).

Patch:
diff --git a/query/physicalplan/binaryscalarexpr.go b/query/physicalplan/binaryscalarexpr.go
--- a/query/physicalplan/binaryscalarexpr.go
+++ b/query/physicalplan/binaryscalarexpr.go
@@ -104,7 +104,11 @@ func BinaryScalarOperation(left arrow.Array, right scalar.Scalar, operator logic
 }
 
 func ArrayScalarCompute(funcName string, left arrow.Array, right scalar.Scalar) (*Bitmap, error) {
-	equalsResult, err := compute.CallFunction(context.TODO(), funcName, nil, compute.NewDatum(left), compute.NewDatum(right))
+	leftData := compute.NewDatum(left)
+	defer leftData.Release()
+	rightData := compute.NewDatum(right)
+	defer rightData.Release()
+	equalsResult, err := compute.CallFunction(context.TODO(), funcName, nil, leftData, rightData)
 	if err != nil {
 		if errors.Unwrap(err).Error() == "not implemented" {
 			return nil, ErrUnsupportedBinaryOperation


Test patch:
diff --git a/query/physicalplan/binaryscalarexpr_test.go b/query/physicalplan/binaryscalarexpr_test.go
--- a/query/physicalplan/binaryscalarexpr_test.go
+++ b/query/physicalplan/binaryscalarexpr_test.go
@@ -51,3 +51,19 @@ func TestBinaryScalarOperationNotImplemented(t *testing.T) {
 	_, err := BinaryScalarOperation(arr, s, logicalplan.OpAnd)
 	require.Equal(t, err, ErrUnsupportedBinaryOperation)
 }
+
+func Test_ArrayScalarCompute_Leak(t *testing.T) {
+	allocator := memory.NewCheckedAllocator(memory.NewGoAllocator())
+	defer allocator.AssertSize(t, 0)
+
+	ab := array.NewInt64Builder(allocator)
+	defer ab.Release()
+
+	ab.AppendValues([]int64{1, 2, 3}, nil)
+	arr := ab.NewInt64Array()
+	defer arr.Release()
+
+	s := scalar.NewInt64Scalar(4)
+	_, err := ArrayScalarCompute("equal", arr, s)
+	require.NoError(t, err)
+}