scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-08 03:47:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

cache the arrow table

Browse source
This commit is contained in:
Todd Gruben 2023-02-28 08:45:29 -06:00
parent d7c6258f16
commit 96e83c950e
3 changed files with 16 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
apply.go
View file

@ -407,7 +407,9 @@ func (sf *ShardFile) Process(cs *ChangesetRequest) error {
if err != nil {
return err
}
return os.Rename(rtemp+sf.executor.TableExtension(), sf.dest+sf.executor.TableExtension())
key := sf.dest + sf.executor.TableExtension()
delete(sf.executor.frameCache, key)
return os.Rename(rtemp+sf.executor.TableExtension(), key)
}
func (sf *ShardFile) LoadBlobs() error {

12
arrow.go
View file

@ -436,11 +436,21 @@ func (e *executor) dataFrameExists(fname string) bool {
}
func (e *executor) getDataTable(ctx context.Context, fname string, mem memory.Allocator) (arrow.Table, error) {
table, ok := e.frameCache[fname]
if ok {
return table, nil
}
if e.typeIsParquet() {
table, err := readTableParquetCtx(ctx, fname, mem)
e.frameCache[fname] = table
return table, err
}
return readTableArrow(fname, mem)
table, err := readTableArrow(fname, mem)
if err != nil {
return nil, err
}
e.frameCache[fname] = table
return table, err
}
func (e *executor) typeIsParquet() bool {

2
executor.go
View file

@ -17,6 +17,7 @@ import (
"time"
"unsafe"
"github.com/apache/arrow/go/v10/arrow"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/disco"
"github.com/featurebasedb/featurebase/v3/pql"
@ -80,6 +81,7 @@ type executor struct {
// Temporary flag to be removed when stablized
dataframeEnabled bool
datafameUseParquet bool
frameCache map[string]arrow.Table
}
// executorOption is a functional option type for pilosa.executor