diff --git a/apply.go b/apply.go index a8e140fec..3b0ef0160 100644 --- a/apply.go +++ b/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) } diff --git a/arrow.go b/arrow.go index 3ad33c0b1..6eebd7482 100644 --- a/arrow.go +++ b/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 } diff --git a/executor.go b/executor.go index 308ed62a6..732603ac0 100644 --- a/executor.go +++ b/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 {