arrow cache needed a lock

apply.go

@@ -406,7 +406,9 @@ func (sf *ShardFile) Process(cs *ChangesetRequest) error {
 		return err
 	}
 	fname := sf.dest + sf.executor.TableExtension()
+	sf.executor.arrowmu.Lock()
 	delete(sf.executor.arrowCache, fname)
+	sf.executor.arrowmu.Unlock()
 	return os.Rename(rtemp+sf.executor.TableExtension(), fname)
 }
 

arrow.go

@@ -436,7 +436,9 @@ func (e *executor) dataFrameExists(fname string) bool {
 }
 
 func (e *executor) getDataTable(ctx context.Context, fname string) (arrow.Table, error) {
+	e.arrowmu.Lock()
 	table, ok := e.arrowCache[fname]
+	e.arrowmu.Unlock()
 	if ok {
 		vprint.VV("returning table from cache name:%v numcols:%v numrows:%v", fname, table.NumCols(), table.NumRows())
 
@@ -453,7 +455,9 @@ func (e *executor) getDataTable(ctx context.Context, fname string) (arrow.Table,
 	if err != nil {
 		return nil, err
 	}
+	e.arrowmu.Lock()
 	e.arrowCache[fname] = table
+	e.arrowmu.Unlock()
 	vprint.VV("returning new table and cacheing at cache name:%v numcols:%v numrows:%v", fname, table.NumCols(), table.NumRows())
 	return table, nil
 }

executor.go

@@ -84,6 +84,7 @@ type executor struct {
 	datafameUseParquet bool
 	arrowCache         map[string]arrow.Table
 	pool               memory.Allocator
+	arrowmu            sync.Mutex
 }
 
 // executorOption is a functional option type for pilosa.executor