bug in arrowtable accounting

apply.go

@@ -58,6 +58,7 @@ func (e *executor) IvyReduce(reduceCode string, opCode string, opt *ExecOptions)
 		if v == nil {
 			return prev
 		}
+
 		if accumulator == nil {
 			switch val := v.(type) {
 			case *dataframe.DataFrame:
@@ -228,7 +229,6 @@ func (e *executor) executeApplyShard(ctx context.Context, qcx *Qcx, index string
 	if err != nil {
 		return nil, err
 	}
-	defer table.Release()
 	df, err := dataframe.NewDataFrameFromTable(e.pool, table)
 	if err != nil {
 		return nil, err
@@ -250,6 +250,7 @@ func (e *executor) executeApplyShard(ctx context.Context, qcx *Qcx, index string
 	if err != nil {
 		return nil, fmt.Errorf("ivy map error: %w", err)
 	}
+
 	return context.Global("_"), nil
 }
 
@@ -407,7 +408,11 @@ func (sf *ShardFile) Process(cs *ChangesetRequest) error {
 	}
 	fname := sf.dest + sf.executor.TableExtension()
 	sf.executor.arrowmu.Lock()
-	delete(sf.executor.arrowCache, fname)
+	t, ok := sf.executor.arrowCache[fname]
+	if ok {
+		t.Release()
+		delete(sf.executor.arrowCache, fname)
+	}
 	sf.executor.arrowmu.Unlock()
 	return os.Rename(rtemp+sf.executor.TableExtension(), fname)
 }

arrow.go

@@ -435,14 +435,23 @@ func (e *executor) dataFrameExists(fname string) bool {
 	return true
 }
 
+func (e *executor) dumpCache(msg string) {
+	e.arrowmu.Lock()
+	defer e.arrowmu.Unlock()
+	vprint.VV("dump:%v", msg)
+	for k, v := range e.arrowCache {
+		table := v
+		fname := k
+		vprint.VV(" cache name:%v numcols:%v numrows:%v", fname, table.NumCols(), table.NumRows())
+
+	}
+}
+
 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())
-
-		table.Retain()
 		return table, nil
 	}
 	// ignoring the passed in allocatorsince where caching
@@ -458,7 +467,6 @@ func (e *executor) getDataTable(ctx context.Context, fname string) (arrow.Table,
 	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

@@ -6587,7 +6587,6 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
 	}
 
 	ch := make(chan mapResponse, len(shards))
-
 	expected := 0
 shardLoop:
 	for _, shard := range shards {