dbshard.go

// Copyright 2022 Molecula Corp. (DBA FeatureBase).
// SPDX-License-Identifier: Apache-2.0
package pilosa

import (
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"

	rbfcfg "github.com/featurebasedb/featurebase/v3/rbf/cfg"
	txkey "github.com/featurebasedb/featurebase/v3/short_txkey"
	"github.com/featurebasedb/featurebase/v3/storage"
	"github.com/pkg/errors"

	"github.com/featurebasedb/featurebase/v3/vprint"
)

var _ = sort.Sort

const (
	// backendsDir is the default backends directory used to store the
	// data for each backend.
	backendsDir = "backends"
)

// types to support a database file per shard

type DBHolder struct {
	Index map[string]*DBIndex
}

func NewDBHolder() *DBHolder {
	return &DBHolder{
		Index: make(map[string]*DBIndex),
	}
}

type DBIndex struct {
	Shard map[uint64]*DBShard
}

type DBWrapper interface {
	NewTx(write bool, initialIndexName string, o Txo) (tx Tx, err error)
	Close() error
	DeleteFragment(index, field, view string, shard uint64, frag interface{}) error
	DeleteField(index, field, fieldPath string) error
	OpenListString() string
	Path() string
	HasData() (has bool, err error)
	SetHolder(h *Holder)
	//needed for restore
	CloseDB() error
	OpenDB() error
}

type DBRegistry interface {
	OpenDBWrapper(path string, doAllocZero bool, cfg *storage.Config) (DBWrapper, error)
}

type DBShard struct {
	HolderPath string

	Index string
	Shard uint64
	Open  bool

	typ  txtype
	styp string

	W             DBWrapper
	ParentDBIndex *DBIndex

	idx *Index
	per *DBPerShard

	closed bool
}

func (dbs *DBShard) DeleteFragment(index, field, view string, shard uint64, frag interface{}) (err error) {
	if index != dbs.Index {
		return fmt.Errorf("DeleteFragment called on DBShard for %q with index %q", dbs.Index, index)
	}
	if shard != dbs.Shard {
		return fmt.Errorf("DeleteFragment called on DBShard for %d with shard %d", dbs.Shard, shard)
	}
	return dbs.W.DeleteFragment(index, field, view, shard, frag)
}

func (dbs *DBShard) DeleteFieldFromStore(index, field, fieldPath string) (err error) {
	if index != dbs.Index {
		return fmt.Errorf("DeleteFieldFromStore called on DBShard for %q with index %q", dbs.Index, index)
	}
	return dbs.W.DeleteField(index, field, fieldPath)
}

func (dbs *DBShard) Close() (err error) {
	dbs.closed = true
	return dbs.W.Close()
}

func (dbs *DBShard) NewTx(write bool, initialIndexName string, o Txo) (tx Tx, err error) {
	if initialIndexName != dbs.Index {
		return nil, fmt.Errorf("NewTx called on DBShard for %q with index %q", dbs.Index, initialIndexName)
	}
	if o.dbs != dbs {
		return nil, fmt.Errorf("dbs mismatch: TxFactory.NewTx() should have set o.dbs(%p) to equal dbs(%p)", o.dbs, dbs)
	}
	if o.Shard != dbs.Shard {
		return nil, fmt.Errorf("shard disagreement: o.Shard='%v' but dbs.Shard='%v'", int(o.Shard), int(dbs.Shard))
	}
	return dbs.W.NewTx(write, initialIndexName, o)
}

type flatkey struct {
	index string
	shard uint64
}

type DBPerShard struct {
	Mu sync.Mutex

	HolderDir string

	dbh *DBHolder

	// just flat, not buried within the Node heirarchy.
	// Easily see how many we have.
	Flatmap map[flatkey]*DBShard

	typ txtype

	txf    *TxFactory
	holder *Holder

	// cache the shards per index to avoid excessive
	// directory scans of the index directory.
	// Keep it up-to-date as we add shards to avoid doing
	// a filesystem rescan on new shard creation.
	//
	// index -> *shardSet
	index2shards map[string]*shardSet

	StorageConfig *storage.Config
	RBFConfig     *rbfcfg.Config
}

func newIndex2Shards() (r map[string]*shardSet) {
	r = make(map[string]*shardSet)
	return
}

type shardSet struct {
	shardsMap map[uint64]struct{}
	shardsVer int64 // increment with each change.

	// give out readonly to repeated consumers if
	// readonlyVer == shardsVer
	readonly    map[uint64]struct{}
	readonlyVer int64
}

func (a *shardSet) unionInPlace(b *shardSet) {
	shards := b.CloneMaybe()
	for shard := range shards {
		a.add(shard)
	}
}

func (a *shardSet) equals(b *shardSet) bool {
	if len(a.shardsMap) != len(b.shardsMap) {
		return false
	}
	for shardInA := range a.shardsMap {
		_, ok := b.shardsMap[shardInA]
		if !ok {
			return false
		}
	}
	return true

}

func (a *shardSet) shards() []uint64 {
	s := make([]uint64, 0, len(a.shardsMap))
	for si := range a.shardsMap {
		s = append(s, si)
	}
	return s
}

func (ss *shardSet) String() (r string) {
	r = "["
	for k := range ss.shardsMap {
		r += fmt.Sprintf("%v, ", k)
	}
	r += "]"
	return
}

func (ss *shardSet) add(shard uint64) {
	_, already := ss.shardsMap[shard]
	if !already {
		ss.shardsMap[shard] = struct{}{}
		ss.shardsVer++
	}
}

// CloneMaybe maintains a re-usable readonly version
// ss.shards that can be returned to multiple goroutine
// reads as it will never change. A copy is only made
// once for each change in the shard set.
func (ss *shardSet) CloneMaybe() map[uint64]struct{} {

	if ss.readonlyVer == ss.shardsVer {
		return ss.readonly
	}

	// readonlyVer is out of date.
	// readonly needs update. We cannot
	// modify the readonly map in place;
	// must make a fully new copy here.
	ss.readonly = make(map[uint64]struct{})

	for k := range ss.shardsMap {
		ss.readonly[k] = struct{}{}
	}
	ss.readonlyVer = ss.shardsVer
	return ss.readonly
}

func newShardSet() *shardSet {
	return &shardSet{
		shardsMap: make(map[uint64]struct{}),
	}
}

func (per *DBPerShard) LoadExistingDBs() (err error) {
	idxs := per.holder.Indexes()

	for _, idx := range idxs {

		shardset, err := per.txf.GetShardsForIndex(idx, "", true)
		if err != nil {
			return err
		}
		for shard := range shardset {
			_, err := per.GetDBShard(idx.name, shard, idx)
			if err != nil {
				return errors.Wrap(err, "DBPerShard.LoadExistingDBs GetDBShard()")
			}
		}
	}
	return
}

func (txf *TxFactory) NewDBPerShard(typ txtype, holderDir string, holder *Holder) (d *DBPerShard) {
	if holder.cfg == nil || holder.cfg.RBFConfig == nil || holder.cfg.StorageConfig == nil {
		vprint.PanicOn("must have holder.cfg.RBFConfig and holder.cfg.StorageConfig set here")
	}

	d = &DBPerShard{
		typ:           typ,
		HolderDir:     holderDir,
		holder:        holder,
		dbh:           NewDBHolder(),
		Flatmap:       make(map[flatkey]*DBShard),
		txf:           txf,
		index2shards:  newIndex2Shards(),
		StorageConfig: holder.cfg.StorageConfig,
		RBFConfig:     holder.cfg.RBFConfig,
	}
	return
}

func (per *DBPerShard) DeleteIndex(index string) (err error) {

	per.Mu.Lock()
	defer per.Mu.Unlock()

	dbi, ok := per.dbh.Index[index]
	if !ok {
		// since we lazily make indexes upon use by a Tx now, we won't
		// have an index for server/ TestQuerySQLUnary/test-20 to delete.
		// Don't freak out. Just return nil.
		return nil
	}
	for _, dbs := range dbi.Shard {
		err = dbs.Close()
		if err != nil {
			return errors.Wrap(err, "DBPerShard.DeleteIndex dbs.Close()")
		}
		path := dbs.pathForType(per.typ)
		err = os.RemoveAll(path)
		if err != nil {
			return errors.Wrap(err, fmt.Sprintf("DBPerShard.DeleteIndex os.RemoveAll('%v')", path))
		}
		delete(per.index2shards, index)
	}

	// allow the index to be created again anew.
	delete(per.dbh.Index, index)

	return
}

func (per *DBPerShard) DeleteFieldFromStore(index, field, fieldPath string) (err error) {
	per.Mu.Lock()
	defer func() {
		if fieldPath != "" {
			_ = os.RemoveAll(fieldPath)
		}
		per.Mu.Unlock()
	}()

	dbi, ok := per.dbh.Index[index]
	if !ok {
		// TestIndex_Existence_Delete in index_internal_test.go
		// will call us without having ever created a Tx or DB,
		// so we can't complain here.
		return nil
	}
	for _, dbs := range dbi.Shard {
		if e := dbs.W.DeleteField(index, field, fieldPath); e != nil && err == nil {
			err = errors.Wrap(e, "DeleteFieldFromStore()")
		}
	}
	return err
}

func (per *DBPerShard) DeleteFragment(index, field, view string, shard uint64, frag *fragment) error {

	idx := per.txf.holder.Index(index)
	dbs, err := per.GetDBShard(index, shard, idx)
	if err != nil {
		return err
	}
	return dbs.DeleteFragment(index, field, view, shard, frag)
}

// if you know the shard, you can use this
// pathForType and prefixForType must be kept in sync!
func (dbs *DBShard) pathForType(ty txtype) string {
	// top level paths will end in "@@"

	// what here for roaring? well, roaringRegistrar.OpenDBWrapper()
	// is a no-op anyhow. so doesn't need to be correct atm.

	path := dbs.HolderPath + sep + dbs.Index + sep + backendsDir + sep + ty.DirectoryName() + sep + fmt.Sprintf("shard.%04v", dbs.Shard)
	return path
}

// if you don't know the shard, you have to use this.
// prefixForType and pathForType must be kept in sync!
func (per *DBPerShard) prefixForType(idx *Index, ty txtype) string {
	// top level paths will end in "@@"
	return per.HolderDir + sep + idx.name + sep + backendsDir + sep + ty.DirectoryName() + sep
}

var ErrNoData = fmt.Errorf("no data")

// keep our cache of shards up-to-date in memory; after the initial
// directory scan, this is all we should we need. Prevents us from
// doing additional, expensive, directory scans.
//
// Caller must hold per.Mu.Lock() already.
func (per *DBPerShard) updateIndex2ShardCacheWithNewShard(dbs *DBShard) {
	shardset, ok := per.index2shards[dbs.Index]
	if !ok {
		shardset = newShardSet()
		per.index2shards[dbs.Index] = shardset
	}
	// INVAR: shardset is present, not nil; a map that can be added to.
	shardset.add(dbs.Shard)
}

func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *DBShard, err error) {
	per.Mu.Lock()
	defer per.Mu.Unlock()
	return per.unprotectedGetDBShard(index, shard, idx)
}

func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *Index) (dbs *DBShard, err error) {

	dbi, ok := per.dbh.Index[index]
	if !ok {
		dbi = &DBIndex{
			Shard: make(map[uint64]*DBShard),
		}
		per.dbh.Index[index] = dbi
	}
	dbs, ok = dbi.Shard[shard]
	if dbs != nil && dbs.closed {
		vprint.PanicOn(fmt.Sprintf("cannot retain closed dbs across holder ReOpen dbs='%p'; per.typ='%v'", dbs, per.typ))
	}
	if !ok {
		dbs = &DBShard{
			typ:           per.typ,
			ParentDBIndex: dbi,
			Index:         index,
			Shard:         shard,
			HolderPath:    per.HolderDir,
			idx:           idx,
			per:           per,
		}
		dbs.styp = per.typ.String()
		dbi.Shard[shard] = dbs
		per.updateIndex2ShardCacheWithNewShard(dbs)
	}
	if !dbs.Open {
		var registry DBRegistry
		switch dbs.typ {
		case rbfTxn:
			registry = globalRbfDBReg
			registry.(*rbfDBRegistrar).SetRBFConfig(per.RBFConfig)
		default:
			vprint.PanicOn(fmt.Sprintf("unknown txtyp: '%v'", dbs.typ))
		}
		path := dbs.pathForType(dbs.typ)
		w, err := registry.OpenDBWrapper(path, DetectMemAccessPastTx, per.StorageConfig)
		vprint.PanicOn(err)
		h := idx.Holder()
		w.SetHolder(h)
		dbs.Open = true
		per.Flatmap[flatkey{index: index, shard: shard}] = dbs
		dbs.W = w
	}
	return dbs, nil
}

func (per *DBPerShard) Close() (err error) {
	per.Mu.Lock()
	defer per.Mu.Unlock()

	for _, dbi := range per.dbh.Index {
		for _, dbs := range dbi.Shard {
			err = dbs.Close()
			vprint.PanicOn(err)
		}
	}
	return
}

// DBPerShardGetShardsForIndex returns the shards for idx.
// If requireData, we open the database and see that it has a key, rather
// than assume that the database file presence is enough.
func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requireData bool) (map[uint64]struct{}, error) {
	return f.dbPerShard.TypedDBPerShardGetShardsForIndex(f.typ, idx, roaringViewPath, requireData)
}

// requireData means open the database file and verify that at least one key is set.
// The returned sliceOfShards should not be modified. We will cache it for subsequent
// queries.
//
// when a new DBShard is made, we will update the list of shards then. Thus
// the per.index2shard should always be up to date AFTER the first call here.
func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (shardMap map[uint64]struct{}, err error) {

	// use the cache, always
	per.Mu.Lock()
	defer per.Mu.Unlock()

	i2ss := per.index2shards

	ss, ok := i2ss[idx.name]
	if ok {
		return ss.CloneMaybe(), nil
	}
	// INVAR: cache miss, and index2shards[ty] exists.

	// gotta read shards from disk directory layout.
	setOfShards := newShardSet()
	per.index2shards[idx.name] = setOfShards

	// Upon return, cache the setOfShards value and reuse it next time

	path := per.prefixForType(idx, ty)

	ignoreEmpty := false
	includeRoot := true
	dbf, err := listDirUnderDir(path, includeRoot, ignoreEmpty)
	vprint.PanicOn(err)

	for _, nm := range dbf {
		base := filepath.Base(nm)

		// We're only interested in "shard.*" files, so skip everything else.
		const shardPrefix = "shard."
		const lenOfShardPrefix = len(shardPrefix)
		if !strings.HasPrefix(base, shardPrefix) {
			continue
		}

		// Parse filename into integer.
		shard, err := strconv.ParseUint(base[lenOfShardPrefix:], 10, 64)
		if err != nil {
			vprint.PanicOn(err)
			continue
		}

		// exclude those without data?
		hasData := false

		if requireData {
			hasData, err = per.unprotectedTypedIndexShardHasData(ty, idx, shard)
			if err != nil {
				return nil, err
			}
			if hasData {
				setOfShards.add(shard)
			}
		} else {
			// file presence is enough
			setOfShards.add(shard)
		}
	}
	return setOfShards.CloneMaybe(), nil
}

func (per *DBPerShard) unprotectedTypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) {
	if ty != per.typ {
		return
	}

	// make the dbs if it doesn't get exist
	dbs, err := per.unprotectedGetDBShard(idx.name, shard, idx)
	if err != nil {
		return false, errors.Wrap(err, fmt.Sprintf("DBPerShard.TypedIndexShardHasData() "+
			"per.GetDBShard(index='%v', shard='%v', ty='%v')", idx.name, shard, ty.String()))
	}

	return dbs.W.HasData()
}

func listDirUnderDir(root string, includeRoot bool, ignoreEmpty bool) (files []string, err error) {
	if !dirExists(root) {
		return
	}

	n := len(root) + 1
	if includeRoot {
		n = 0
	}
	err = filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if len(path) < n {
			// ignore
		} else {
			if info == nil {
				// re-opening an RBF database hit this, racing with a directory rename.
				// Don't freak out.
				return nil
			}
			if !info.IsDir() {
				// ignore files
			} else {
				if ignoreEmpty && info.Size() == 0 {
					return nil
				}
				files = append(files, path[n:])
			}
		}
		return nil
	})
	return
}

type FieldView2Shards struct {
	// field -> view -> *shardSet
	m map[string]map[string]*shardSet
}

func (vs *FieldView2Shards) getViewsForField(field string) map[string]*shardSet {
	return vs.m[field]
}

func (vs *FieldView2Shards) addViewShardSet(fv txkey.FieldView, ss *shardSet) {

	f, ok := vs.m[fv.Field]
	if !ok {
		f = make(map[string]*shardSet)
		vs.m[fv.Field] = f
	}
	// INVAR: f is ready to take ss.

	// existing stuff to merge with?
	prior, ok := f[fv.View]
	if !ok {
		f[fv.View] = ss
		return
	}
	// merge ss and prior. No need to put the union back into f[fv.View]
	// because prior is a pointer.
	prior.unionInPlace(ss)
}

func (a *FieldView2Shards) equals(b *FieldView2Shards) bool {
	if a == nil && b == nil {
		return true
	}
	if a == nil || b == nil {
		return false
	}
	if len(a.m) != len(b.m) {
		return false
	}
	for field, viewmapA := range a.m {
		viewmapB, ok := b.m[field]
		if !ok {
			return false
		}
		if len(viewmapB) != len(viewmapA) {
			return false
		}
		for k, va := range viewmapA {
			vb, ok := viewmapB[k]
			if !ok {
				return false
			}
			if !va.equals(vb) {
				return false
			}
		}
	}
	return true
}

func NewFieldView2Shards() *FieldView2Shards {
	return &FieldView2Shards{
		m: make(map[string]map[string]*shardSet), // expected response from GetView2ShardMapForIndex
	}
}

func (vs *FieldView2Shards) addShard(fv txkey.FieldView, shard uint64) {
	viewmap, ok := vs.m[fv.Field]
	if !ok {
		viewmap = make(map[string]*shardSet)
		vs.m[fv.Field] = viewmap
	}
	ss, ok := viewmap[fv.View]
	if !ok {
		ss = newShardSet()
		viewmap[fv.View] = ss
	}
	ss.add(shard)
}

func (vs *FieldView2Shards) String() (r string) {
	r = "\n"
	for field, viewmap := range vs.m {
		for view, shards := range viewmap {
			r += fmt.Sprintf("field '%v' view:'%v' shards:%v\n", field, view, shards)
		}
	}
	r += "\n"
	return
}

func (vs *FieldView2Shards) removeField(name string) {
	delete(vs.m, name)
}

func (per *DBPerShard) GetFieldView2ShardsMapForIndex(idx *Index) (vs *FieldView2Shards, err error) {
	ty := per.typ

	switch ty {
	default:
		vs = NewFieldView2Shards()

		shardMap, err := per.TypedDBPerShardGetShardsForIndex(ty, idx, "", true)
		if err != nil {
			return nil, err
		}

		for shard := range shardMap {
			dbs, err := per.GetDBShard(idx.name, shard, idx)
			if err != nil {
				return nil, errors.Wrap(err, "DBPerShard.GetFieldView2ShardsMapForIndex GetDBShard()")
			}
			fieldviews, err := dbs.AllFieldViews()
			if err != nil {
				return nil, errors.Wrap(err, "DBPerShard.GetFieldView2ShardsMapForIndex dbs.AllFieldViews()")
			}
			for _, fv := range fieldviews {
				vs.addShard(fv, shard)
			}
		}
	}

	return
}

func (dbs *DBShard) AllFieldViews() (fvs []txkey.FieldView, err error) {

	tx, err := dbs.NewTx(!writable, dbs.idx.name, Txo{Write: !writable, Shard: dbs.Shard, Index: dbs.idx, dbs: dbs})
	if err != nil {
		return nil, errors.Wrap(err, fmt.Sprintf("dbshard.NewTx for index '%v', shard %v", dbs.idx.name, dbs.Shard))
	}
	defer tx.Rollback()
	return tx.GetSortedFieldViewList(dbs.idx, dbs.Shard)
}