global arrow allocator

arrow.go

@@ -443,21 +443,21 @@ type arrowCache struct {
 func (e *executor) getDataTable(ctx context.Context, fname string) (arrow.Table, memory.Allocator, error) {
 	cache, ok := e.arrowCache[fname]
 	if ok {
-		return cache.table, cache.pool, nil
+		return cache.table, e.pool, nil
 	}
 	// ignoring the passed in allocatorsince where caching
 	mem := memory.NewGoAllocator()
 	if e.typeIsParquet() {
 		table, err := readTableParquetCtx(ctx, fname, mem)
 		e.arrowCache[fname] = &arrowCache{table: table, pool: mem}
-		return table, mem, err
+		return table, e.pool, err
 	}
 	table, err := readTableArrow(fname, mem)
 	if err != nil {
 		return nil, nil, err
 	}
 	e.arrowCache[fname] = &arrowCache{table: table, pool: mem}
-	return table, mem, nil
+	return table, e.pool, nil
 }
 
 func (e *executor) typeIsParquet() bool {

executor.go

@@ -17,6 +17,7 @@ import (
 	"time"
 	"unsafe"
 
+	"github.com/apache/arrow/go/v10/arrow/memory"
 	"github.com/featurebasedb/featurebase/v3/dax"
 	"github.com/featurebasedb/featurebase/v3/disco"
 	"github.com/featurebasedb/featurebase/v3/pql"
@@ -81,6 +82,7 @@ type executor struct {
 	dataframeEnabled   bool
 	datafameUseParquet bool
 	arrowCache         map[string]*arrowCache
+	pool               memory.Allocator
 }
 
 // executorOption is a functional option type for pilosa.executor
@@ -145,6 +147,7 @@ func newExecutor(opts ...executorOption) *executor {
 	_ = testhook.Opened(NewAuditor(), e, nil)
 	e.workers = task.NewPool(e.workerPoolSize, e.doOneJob, e)
 	e.arrowCache = make(map[string]*arrowCache)
+	e.pool = memory.NewGoAllocator()
 	return e
 }