mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
2b4d49e502
We distinguish between "TrackExistence option is set" and "we are actually doing existence tracking", to avoid mishaps like accidentally creating an "existence" view for a BSI field or something like that. This logic was being done probably-correctly in one place, and ignored or handwaved in some, so this is an attempt to just make it more consistent.
2594 lines
74 KiB
Go
2594 lines
74 KiB
Go
// Copyright 2022 Molecula Corp. (DBA FeatureBase).
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
package pilosa
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"math"
|
|
"math/bits"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"unicode"
|
|
|
|
"github.com/featurebasedb/featurebase/v3/pql"
|
|
"github.com/featurebasedb/featurebase/v3/roaring"
|
|
"github.com/featurebasedb/featurebase/v3/testhook"
|
|
"github.com/featurebasedb/featurebase/v3/tracing"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
// Default field settings.
|
|
const (
|
|
DefaultFieldType = FieldTypeSet
|
|
|
|
DefaultCacheType = CacheTypeRanked
|
|
|
|
// Default ranked field cache
|
|
DefaultCacheSize = 50000
|
|
|
|
bitsPerWord = 32 << (^uint(0) >> 63) // either 32 or 64
|
|
maxInt = 1<<(bitsPerWord-1) - 1 // either 1<<31 - 1 or 1<<63 - 1
|
|
|
|
)
|
|
|
|
// Field types.
|
|
const (
|
|
FieldTypeSet = "set"
|
|
FieldTypeInt = "int"
|
|
FieldTypeTime = "time"
|
|
FieldTypeMutex = "mutex"
|
|
FieldTypeBool = "bool"
|
|
FieldTypeDecimal = "decimal"
|
|
FieldTypeTimestamp = "timestamp"
|
|
)
|
|
|
|
type protected struct {
|
|
mu sync.Mutex
|
|
duration time.Duration
|
|
}
|
|
|
|
func (p *protected) Set(d time.Duration) {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
p.duration = d
|
|
}
|
|
func (p *protected) Get() time.Duration {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return p.duration
|
|
}
|
|
|
|
var availableShardFileFlushDuration = &protected{
|
|
duration: 5 * time.Second,
|
|
}
|
|
|
|
// Field represents a container for views.
|
|
type Field struct {
|
|
mu sync.RWMutex
|
|
createdAt int64
|
|
owner string
|
|
path string
|
|
index string
|
|
name string
|
|
qualifiedName string
|
|
|
|
idx *Index
|
|
|
|
viewMap map[string]*view
|
|
|
|
broadcaster broadcaster
|
|
serializer Serializer
|
|
|
|
// Field options.
|
|
options FieldOptions
|
|
|
|
bsiGroups []*bsiGroup
|
|
|
|
// Shards with data on any node in the cluster, according to this node.
|
|
remoteAvailableShardsMu sync.Mutex
|
|
remoteAvailableShards *roaring.Bitmap
|
|
|
|
translateStore TranslateStore
|
|
|
|
// Instantiates new translation stores
|
|
OpenTranslateStore OpenTranslateStoreFunc
|
|
|
|
// Used for looking up a foreign index.
|
|
holder *Holder
|
|
|
|
// Stores whether or not the field has keys enabled.
|
|
// This is most helpful for cases where the keys are
|
|
// based on a foreign index; this prevents having to
|
|
// call holder.index.Keys() every time.
|
|
usesKeys bool
|
|
|
|
// Synchronization primitives needed for async writing of
|
|
// the remoteAvailableShards
|
|
availableShardChan chan struct{}
|
|
wg sync.WaitGroup
|
|
|
|
// track whether we're shutting down
|
|
closing chan struct{}
|
|
}
|
|
|
|
// FieldOption is a functional option type for pilosa.fieldOptions.
|
|
type FieldOption func(fo *FieldOptions) error
|
|
|
|
// OptFieldKeys is a functional option on FieldOptions
|
|
// used to specify whether keys are used for this field.
|
|
func OptFieldKeys() FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
fo.Keys = true
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldForeignIndex marks this field as a foreign key to another
|
|
// index. That is, the values of this field should be interpreted as
|
|
// referencing records (Pilosa columns) in another index. TODO explain
|
|
// where/how this is used by Pilosa.
|
|
func OptFieldForeignIndex(index string) FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
fo.ForeignIndex = index
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTypeDefault is a functional option on FieldOptions
|
|
// used to set the field type and cache setting to the default values.
|
|
func OptFieldTypeDefault() FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("field type is already set to: %s", fo.Type)
|
|
}
|
|
fo.Type = FieldTypeSet
|
|
fo.CacheType = DefaultCacheType
|
|
fo.CacheSize = DefaultCacheSize
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTypeSet is a functional option on FieldOptions
|
|
// used to specify the field as being type `set` and to
|
|
// provide any respective configuration values.
|
|
func OptFieldTypeSet(cacheType string, cacheSize uint32) FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("field type is already set to: %s", fo.Type)
|
|
}
|
|
fo.Type = FieldTypeSet
|
|
fo.CacheType = cacheType
|
|
fo.CacheSize = cacheSize
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTypeInt is a functional option on FieldOptions
|
|
// used to specify the field as being type `int` and to
|
|
// provide any respective configuration values.
|
|
func OptFieldTypeInt(min, max int64) FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("field type is already set to: %s", fo.Type)
|
|
}
|
|
if min > max {
|
|
return errors.New("int field min cannot be greater than max")
|
|
}
|
|
fo.Type = FieldTypeInt
|
|
fo.Min = pql.NewDecimal(min, 0)
|
|
fo.Max = pql.NewDecimal(max, 0)
|
|
fo.Base = bsiBase(min, max)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTypeTimestamp is a functional option on FieldOptions
|
|
// used to specify the field as being type `timestamp` and to
|
|
// provide any respective configuration values.
|
|
func OptFieldTypeTimestamp(epoch time.Time, timeUnit string) FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("field type is already set to: %s", fo.Type)
|
|
}
|
|
|
|
minTime := MinTimestamp
|
|
maxTime := MaxTimestamp
|
|
|
|
var base, minInt, maxInt int64
|
|
switch timeUnit {
|
|
case TimeUnitSeconds:
|
|
base = epoch.Unix()
|
|
minInt = minTime.Unix() - base
|
|
maxInt = maxTime.Unix() - base
|
|
case TimeUnitMilliseconds:
|
|
base = epoch.UnixMilli()
|
|
minInt = minTime.UnixMilli() - base
|
|
maxInt = maxTime.UnixMilli() - base
|
|
case TimeUnitMicroseconds, TimeUnitUSeconds:
|
|
base = epoch.UnixMicro()
|
|
minInt = minTime.UnixMicro() - base
|
|
maxInt = maxTime.UnixMicro() - base
|
|
case TimeUnitNanoseconds:
|
|
// Note: For nano, the min and max values are also the min and max integer
|
|
// values we support. Also, keep in mind that MinNano is a negative
|
|
// number. So if base is positive and we do MinNano - base...it would increase minInt
|
|
// beyond what we support. This isn't an issue with larger granularities.
|
|
base = epoch.UnixNano()
|
|
if base > 0 {
|
|
maxInt = MaxTimestampNano.UnixNano() - base
|
|
minInt = MinTimestampNano.UnixNano()
|
|
} else {
|
|
maxInt = MaxTimestampNano.UnixNano()
|
|
minInt = MinTimestampNano.UnixNano() - base
|
|
}
|
|
minTime = MinTimestampNano
|
|
maxTime = MaxTimestampNano
|
|
default:
|
|
return errors.Errorf("invalid time unit: '%q'", fo.TimeUnit)
|
|
}
|
|
|
|
if err := CheckEpochOutOfRange(epoch, minTime, maxTime); err != nil {
|
|
return err
|
|
}
|
|
|
|
fo.Type = FieldTypeTimestamp
|
|
fo.TimeUnit = timeUnit
|
|
fo.Base = base
|
|
fo.Min = pql.NewDecimal(minInt, 0)
|
|
fo.Max = pql.NewDecimal(maxInt, 0)
|
|
|
|
return nil
|
|
}
|
|
|
|
}
|
|
|
|
// OptFieldTypeDecimal is a functional option for creating a `decimal` field.
|
|
// Unless we decide to expand the range of supported values, `scale` is
|
|
// restricted to the range [0,19]. This supports anything from:
|
|
//
|
|
// scale = 0:
|
|
// min: -9223372036854775808.
|
|
// max: 9223372036854775807.
|
|
//
|
|
// to:
|
|
//
|
|
// scale = 19:
|
|
// min: -0.9223372036854775808
|
|
// max: 0.9223372036854775807
|
|
//
|
|
// While it's possible to support scale values outside of this range,
|
|
// the coverage for those scales are no longer continuous. For example,
|
|
//
|
|
// scale = -2:
|
|
// min : [-922337203685477580800, -100]
|
|
// GAPs: [-99, -1], [-199, -101] ... [-922337203685477580799, -922337203685477580701]
|
|
//
|
|
// 0
|
|
//
|
|
// max : [100, 922337203685477580700]
|
|
// GAPs: [1, 99], [101, 199] ... [922337203685477580601, 922337203685477580699]
|
|
//
|
|
// An alternative to this gap strategy would be to scale the supported range
|
|
// to a continuous 64-bit space (which is not unreasonable using bsiGroup.Base).
|
|
// The issue with this approach is that we would need to know which direction
|
|
// to favor. For example, there are two possible ranges for `scale = -2`:
|
|
//
|
|
// min : [-922337203685477580800, -922337203685477580800+(2^64)]
|
|
// max : [922337203685477580700-(2^64), 922337203685477580700]
|
|
func OptFieldTypeDecimal(scale int64, minmax ...pql.Decimal) FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("can't set field type to 'decimal', already set to: %s", fo.Type)
|
|
}
|
|
if scale < 0 || scale > 19 {
|
|
return errors.Errorf("scale values outside the range [0,19] are not supported: %d", scale)
|
|
}
|
|
|
|
fo.Min, fo.Max = pql.MinMax(scale)
|
|
if len(minmax) == 2 {
|
|
min := minmax[0]
|
|
max := minmax[1]
|
|
if !min.IsValid() || !max.IsValid() {
|
|
return errors.Errorf("min/max range %s-%s is not supported", min, max)
|
|
} else if !min.SupportedByScale(scale) || !max.SupportedByScale(scale) {
|
|
return errors.Errorf("min/max range %s-%s is not supported by scale %d", min, max, scale)
|
|
} else if min.GreaterThan(max) {
|
|
return errors.Errorf("decimal field min cannot be greater than max, got %s, %s", min, max)
|
|
}
|
|
fo.Min = min
|
|
fo.Max = max
|
|
} else if len(minmax) > 2 {
|
|
return errors.Errorf("unknown extra parameters beyond min and max: %v", minmax)
|
|
} else if len(minmax) == 1 {
|
|
min := minmax[0]
|
|
if !min.IsValid() {
|
|
return errors.Errorf("min %s is not supported", min)
|
|
} else if !min.SupportedByScale(scale) {
|
|
return errors.Errorf("min %s is not supported by scale %d", min, scale)
|
|
}
|
|
fo.Min = min
|
|
}
|
|
fo.Type = FieldTypeDecimal
|
|
fo.Base = bsiBase(fo.Min.ToInt64(scale), fo.Max.ToInt64(scale))
|
|
fo.Scale = scale
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTypeTime is a functional option on FieldOptions
|
|
// used to specify the field as being type `time` and to
|
|
// provide any respective configuration values.
|
|
// Pass true to skip creation of the standard view.
|
|
func OptFieldTypeTime(timeQuantum TimeQuantum, ttl string, opt ...bool) FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("field type is already set to: %s", fo.Type)
|
|
}
|
|
if !timeQuantum.Valid() {
|
|
return ErrInvalidTimeQuantum
|
|
}
|
|
fo.Type = FieldTypeTime
|
|
fo.TimeQuantum = timeQuantum
|
|
ttlParsed, err := time.ParseDuration(ttl)
|
|
if err != nil {
|
|
return errors.Errorf("cannot parse ttl: %s", ttl)
|
|
}
|
|
if ttlParsed < 0 {
|
|
return errors.Errorf("ttl can't be negative: %s", ttl)
|
|
}
|
|
fo.TTL = ttlParsed
|
|
fo.NoStandardView = len(opt) >= 1 && opt[0]
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTypeMutex is a functional option on FieldOptions
|
|
// used to specify the field as being type `mutex` and to
|
|
// provide any respective configuration values.
|
|
func OptFieldTypeMutex(cacheType string, cacheSize uint32) FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("field type is already set to: %s", fo.Type)
|
|
}
|
|
fo.Type = FieldTypeMutex
|
|
fo.CacheType = cacheType
|
|
fo.CacheSize = cacheSize
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTypeBool is a functional option on FieldOptions
|
|
// used to specify the field as being type `bool` and to
|
|
// provide any respective configuration values.
|
|
func OptFieldTypeBool() FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
if fo.Type != "" {
|
|
return errors.Errorf("field type is already set to: %s", fo.Type)
|
|
}
|
|
fo.Type = FieldTypeBool
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// OptFieldTrackExistence exists mostly to allow the
|
|
// FieldFromFieldOptions/FieldOptionsFromField round-trip to work.
|
|
// If you are actually creating a field, via api.CreateField,
|
|
// it will be turned on unconditionally. You can't turn it
|
|
// off.
|
|
func OptFieldTrackExistence() FieldOption {
|
|
return func(fo *FieldOptions) error {
|
|
fo.TrackExistence = true
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// newField returns a new instance of field (without name validation).
|
|
func newField(holder *Holder, path, index, name string, opts ...FieldOption) (*Field, error) {
|
|
// Apply functional option.
|
|
fo := FieldOptions{}
|
|
for _, opt := range opts {
|
|
err := opt(&fo)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "applying option")
|
|
}
|
|
}
|
|
|
|
f := &Field{
|
|
path: path,
|
|
index: index,
|
|
name: name,
|
|
qualifiedName: FormatQualifiedFieldName(index, name),
|
|
|
|
viewMap: make(map[string]*view),
|
|
|
|
broadcaster: NopBroadcaster,
|
|
serializer: NopSerializer,
|
|
|
|
options: applyDefaultOptions(&fo),
|
|
|
|
remoteAvailableShards: roaring.NewBitmap(),
|
|
|
|
holder: holder,
|
|
|
|
OpenTranslateStore: OpenInMemTranslateStore,
|
|
}
|
|
|
|
return f, nil
|
|
}
|
|
|
|
// Name returns the name the field was initialized with.
|
|
func (f *Field) Name() string { return f.name }
|
|
|
|
// CreatedAt is an timestamp for a specific version of field.
|
|
func (f *Field) CreatedAt() int64 {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
|
|
return f.createdAt
|
|
}
|
|
|
|
// Index returns the index name the field was initialized with.
|
|
func (f *Field) Index() string { return f.index }
|
|
|
|
// Path returns the path the field was initialized with.
|
|
func (f *Field) Path() string { return f.path }
|
|
|
|
// TranslateStorePath returns the translation database path for the field.
|
|
func (f *Field) TranslateStorePath() string {
|
|
return filepath.Join(f.path, "keys")
|
|
}
|
|
|
|
// TranslateStore returns the field's translation store.
|
|
func (f *Field) TranslateStore() TranslateStore {
|
|
return f.translateStore
|
|
}
|
|
|
|
// AvailableShards returns a bitmap of shards that contain data.
|
|
func (f *Field) AvailableShards(localOnly bool) *roaring.Bitmap {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
f.remoteAvailableShardsMu.Lock()
|
|
defer f.remoteAvailableShardsMu.Unlock()
|
|
|
|
var b *roaring.Bitmap
|
|
if localOnly {
|
|
b = roaring.NewBitmap()
|
|
} else {
|
|
b = f.remoteAvailableShards.Clone()
|
|
}
|
|
for viewname, view := range f.viewMap {
|
|
availableShards := view.availableShards()
|
|
if availableShards == nil || availableShards.Containers == nil {
|
|
f.holder.Logger.Warnf("empty available shards for view: %s on field %s available shards: %v", viewname, f.name, availableShards)
|
|
continue
|
|
}
|
|
b.UnionInPlace(view.availableShards())
|
|
}
|
|
return b
|
|
}
|
|
|
|
// LocalAvailableShards returns a bitmap of shards that contain data, but
|
|
// only from the local node. This prevents txfactory from making
|
|
// db-per-shard for remote shards.
|
|
func (f *Field) LocalAvailableShards() *roaring.Bitmap {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
|
|
b := roaring.NewBitmap()
|
|
for _, view := range f.viewMap {
|
|
b.UnionInPlace(view.availableShards())
|
|
}
|
|
return b
|
|
}
|
|
|
|
// AddRemoteAvailableShards merges the set of available shards into the current known set
|
|
// and saves the set to a file.
|
|
func (f *Field) AddRemoteAvailableShards(b *roaring.Bitmap) error {
|
|
f.mergeRemoteAvailableShards(b)
|
|
// Save the updated bitmap to the data store.
|
|
return f.saveAvailableShards()
|
|
}
|
|
|
|
// mergeRemoteAvailableShards merges the set of available shards into the current known set.
|
|
func (f *Field) mergeRemoteAvailableShards(b *roaring.Bitmap) {
|
|
f.remoteAvailableShardsMu.Lock()
|
|
defer f.remoteAvailableShardsMu.Unlock()
|
|
f.remoteAvailableShards = f.remoteAvailableShards.Union(b)
|
|
}
|
|
|
|
// loadAvailableShards reads remoteAvailableShards data for the field, if any.
|
|
func (f *Field) loadAvailableShards() error {
|
|
shards, err := f.holder.sharder.Shards(context.Background(), f.index, f.name)
|
|
if err != nil {
|
|
return errors.Wrap(err, "loading available shards")
|
|
}
|
|
|
|
bm := roaring.NewBitmap()
|
|
for _, s := range shards {
|
|
b := roaring.NewBitmap()
|
|
if err = b.UnmarshalBinary(s); err != nil {
|
|
return errors.Wrap(err, "available shards corrupt")
|
|
}
|
|
bm.UnionInPlace(b)
|
|
}
|
|
// Merge bitmap from file into field.
|
|
f.mergeRemoteAvailableShards(bm)
|
|
|
|
return nil
|
|
}
|
|
|
|
// saveAvailableShards writes remoteAvailableShards data for the field.
|
|
func (f *Field) saveAvailableShards() error {
|
|
select {
|
|
case f.availableShardChan <- struct{}{}:
|
|
default:
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RemoveAvailableShard removes a shard from the bitmap cache.
|
|
//
|
|
// NOTE: This can be overridden on the next sync so all nodes should be updated.
|
|
func (f *Field) RemoveAvailableShard(v uint64) error {
|
|
f.remoteAvailableShardsMu.Lock()
|
|
defer f.remoteAvailableShardsMu.Unlock()
|
|
|
|
b := f.remoteAvailableShards.Clone()
|
|
if _, err := b.Remove(v); err != nil {
|
|
return err
|
|
}
|
|
f.remoteAvailableShards = b
|
|
|
|
return f.saveAvailableShards()
|
|
}
|
|
|
|
// Type returns the field type.
|
|
func (f *Field) Type() string {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
return f.options.Type
|
|
}
|
|
|
|
// CacheSize returns the ranked field cache size.
|
|
func (f *Field) CacheSize() uint32 {
|
|
f.mu.RLock()
|
|
v := f.options.CacheSize
|
|
f.mu.RUnlock()
|
|
return v
|
|
}
|
|
|
|
// Options returns all options for this field.
|
|
func (f *Field) Options() FieldOptions {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
return f.options
|
|
}
|
|
|
|
// Open opens and initializes the field.
|
|
func (f *Field) Open() error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if err := func() (err error) {
|
|
// Ensure the field's path exists.
|
|
f.holder.Logger.Debugf("ensure field path exists: %s", f.path)
|
|
if err := os.MkdirAll(f.path, 0750); err != nil {
|
|
return errors.Wrap(err, "creating field dir")
|
|
}
|
|
|
|
f.holder.Logger.Debugf("load available shards for index/field: %s/%s", f.index, f.name)
|
|
|
|
if err := f.loadAvailableShards(); err != nil {
|
|
return errors.Wrap(err, "loading available shards")
|
|
}
|
|
|
|
// Apply the field options loaded from etcd (or set via setOptions()).
|
|
f.holder.Logger.Debugf("apply options for index/field: %s/%s", f.index, f.name)
|
|
if err := f.applyOptions(f.options); err != nil {
|
|
return errors.Wrap(err, "applying options")
|
|
}
|
|
|
|
f.holder.Logger.Debugf("open views for index/field: %s/%s", f.index, f.name)
|
|
if err := f.openViews(); err != nil {
|
|
return errors.Wrap(err, "opening views")
|
|
}
|
|
|
|
// Apply the field-specific translateStore.
|
|
if err := f.applyTranslateStore(); err != nil {
|
|
return errors.Wrap(err, "applying translate store")
|
|
}
|
|
|
|
// If the field has a foreign index, make sure the index
|
|
// exists.
|
|
if f.options.ForeignIndex != "" {
|
|
if err := f.holder.checkForeignIndex(f); err != nil {
|
|
return errors.Wrap(err, "checking foreign index")
|
|
}
|
|
}
|
|
|
|
f.availableShardChan = make(chan struct{}, 1)
|
|
f.wg.Add(1)
|
|
go f.writeAvailableShards()
|
|
return nil
|
|
}(); err != nil {
|
|
f.unprotectedClose()
|
|
return err
|
|
}
|
|
f.closing = make(chan struct{})
|
|
|
|
_ = testhook.Opened(f.holder.Auditor, f, nil)
|
|
f.holder.Logger.Debugf("successfully opened field index/field: %s/%s", f.index, f.name)
|
|
return nil
|
|
}
|
|
|
|
func (f *Field) protectedRemoteAvailableShards() *roaring.Bitmap {
|
|
f.remoteAvailableShardsMu.Lock()
|
|
defer f.remoteAvailableShardsMu.Unlock()
|
|
|
|
f.remoteAvailableShards.Optimize()
|
|
return f.remoteAvailableShards.Clone()
|
|
}
|
|
|
|
func (f *Field) flushAvailableShards(ctx context.Context) {
|
|
shards := f.protectedRemoteAvailableShards()
|
|
var buf bytes.Buffer
|
|
if _, err := shards.WriteTo(&buf); err != nil {
|
|
f.holder.Logger.Errorf("writting available shards: %v", err)
|
|
return
|
|
}
|
|
|
|
if err := f.holder.sharder.SetShards(ctx, f.index, f.name, buf.Bytes()); err != nil {
|
|
f.holder.Logger.Errorf("setting available shards: %v", err)
|
|
}
|
|
}
|
|
|
|
func (f *Field) writeAvailableShards() {
|
|
defer f.wg.Done()
|
|
|
|
interval := availableShardFileFlushDuration.Get()
|
|
timer := time.NewTimer(interval)
|
|
defer timer.Stop()
|
|
|
|
for range f.availableShardChan {
|
|
// Available shards have been updated.
|
|
|
|
// Wait a bit so that we batch writes.
|
|
timerWait:
|
|
for {
|
|
select {
|
|
case _, ok := <-f.availableShardChan:
|
|
if !ok {
|
|
// The server is shutting down.
|
|
// Do the write immediately.
|
|
timer.Stop()
|
|
break timerWait
|
|
}
|
|
|
|
case <-timer.C:
|
|
// We have waited long enough.
|
|
break timerWait
|
|
}
|
|
}
|
|
|
|
// Set the timer for the next flush.
|
|
timer.Reset(interval)
|
|
|
|
// Actually write the shards.
|
|
f.flushAvailableShards(context.Background())
|
|
}
|
|
}
|
|
|
|
// applyTranslateStore opens the configured translate store.
|
|
func (f *Field) applyTranslateStore() error {
|
|
// Instantiate & open translation store.
|
|
var err error
|
|
f.translateStore, err = f.OpenTranslateStore(f.TranslateStorePath(), f.index, f.name, -1, -1, f.holder.cfg.StorageConfig.FsyncEnabled)
|
|
if err != nil {
|
|
return errors.Wrap(err, "opening field translate store")
|
|
}
|
|
f.usesKeys = f.options.Keys
|
|
|
|
// In the case where the field has a foreign index, set
|
|
// the usesKeys value accordingly.
|
|
if foreignIndexName := f.ForeignIndex(); foreignIndexName != "" {
|
|
if foreignIndex := f.holder.Index(foreignIndexName); foreignIndex != nil {
|
|
f.usesKeys = foreignIndex.Keys()
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// applyForeignIndex used to set the field's translateStore to
|
|
// that of the foreign index, but since moving to partitioned
|
|
// translate stores on indexes, that doesn't happen anymore.
|
|
// So now all this method does is check that the foreign index
|
|
// actually exists. If we decided this was unnecessary (which
|
|
// it kind of is), we could remove the field.holder and all
|
|
// the logic which does this check on holder open after all
|
|
// indexes have opened.
|
|
func (f *Field) applyForeignIndex() error {
|
|
foreignIndex := f.holder.Index(f.options.ForeignIndex)
|
|
if foreignIndex == nil {
|
|
return errors.Wrapf(ErrForeignIndexNotFound, "%s", f.options.ForeignIndex)
|
|
}
|
|
f.usesKeys = foreignIndex.Keys()
|
|
return nil
|
|
}
|
|
|
|
// ForeignIndex returns the foreign index name attached to the field.
|
|
// Returns blank string if no foreign index exists.
|
|
func (f *Field) ForeignIndex() string {
|
|
return f.options.ForeignIndex
|
|
}
|
|
|
|
// TTL returns the ttl of the field.
|
|
func (f *Field) TTL() time.Duration {
|
|
return f.options.TTL
|
|
}
|
|
|
|
func (f *Field) bitDepth() (uint64, error) {
|
|
var maxBitDepth uint64
|
|
|
|
view2shards := f.idx.fieldView2shard.getViewsForField(f.name)
|
|
for name, shardset := range view2shards {
|
|
view := f.view(name)
|
|
if view == nil {
|
|
continue
|
|
}
|
|
|
|
bd, err := view.bitDepth(shardset.shards())
|
|
if err != nil {
|
|
return 0, errors.Wrapf(err, "getting view(%s) bit depth", name)
|
|
}
|
|
if bd > maxBitDepth {
|
|
maxBitDepth = bd
|
|
}
|
|
}
|
|
|
|
return maxBitDepth, nil
|
|
}
|
|
|
|
// cacheBitDepth is used by Index.setFieldBitDepths() to updated the in-memory
|
|
// bitDepth values for each field and its bsiGroup.
|
|
func (f *Field) cacheBitDepth(bd uint64) error {
|
|
// Get the assocated bsiGroup so that its bitDepth can be updated as well.
|
|
bsig := f.bsiGroup(f.name)
|
|
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
|
|
if f.options.BitDepth < bd {
|
|
f.options.BitDepth = bd
|
|
}
|
|
|
|
if bsig != nil && bsig.BitDepth < bd {
|
|
bsig.BitDepth = bd
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// openViews opens and initializes the views inside the field.
|
|
func (f *Field) openViews() error {
|
|
view2shards := f.idx.fieldView2shard.getViewsForField(f.name)
|
|
if view2shards == nil {
|
|
// no data
|
|
return nil
|
|
}
|
|
|
|
for name, shardset := range view2shards {
|
|
view := f.newView(f.viewPath(name), name)
|
|
if err := view.openWithShardSet(shardset); err != nil {
|
|
return fmt.Errorf("opening view: view=%s, err=%s", view.name, err)
|
|
}
|
|
|
|
f.holder.Logger.Debugf("add index/field/view to field.viewMap: %s/%s/%s", f.index, f.name, view.name)
|
|
f.viewMap[view.name] = view
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// setOptions saves options for final application during Open().
|
|
func (f *Field) setOptions(opts *FieldOptions) {
|
|
f.options = applyDefaultOptions(opts)
|
|
}
|
|
|
|
// applyOptions configures the field based on opt.
|
|
func (f *Field) applyOptions(opt FieldOptions) error {
|
|
switch opt.Type {
|
|
case FieldTypeSet, FieldTypeMutex, "":
|
|
fldType := opt.Type
|
|
if fldType == "" {
|
|
fldType = FieldTypeSet
|
|
}
|
|
f.options.Type = fldType
|
|
if opt.CacheType != "" {
|
|
f.options.CacheType = opt.CacheType
|
|
}
|
|
if opt.CacheType == CacheTypeNone {
|
|
f.options.CacheSize = 0
|
|
} else if opt.CacheSize != 0 {
|
|
f.options.CacheSize = opt.CacheSize
|
|
}
|
|
f.options.Min = pql.Decimal{}
|
|
f.options.Max = pql.Decimal{}
|
|
f.options.Base = 0
|
|
f.options.BitDepth = 0
|
|
f.options.TimeQuantum = ""
|
|
f.options.TTL = 0
|
|
f.options.Keys = opt.Keys
|
|
f.options.ForeignIndex = opt.ForeignIndex
|
|
case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp:
|
|
f.options.Type = opt.Type
|
|
f.options.CacheType = CacheTypeNone
|
|
f.options.CacheSize = 0
|
|
f.options.Min = opt.Min
|
|
f.options.Max = opt.Max
|
|
f.options.Base = opt.Base
|
|
f.options.Scale = opt.Scale
|
|
f.options.BitDepth = opt.BitDepth
|
|
f.options.TimeUnit = opt.TimeUnit
|
|
f.options.TimeQuantum = ""
|
|
f.options.TTL = 0
|
|
f.options.Keys = opt.Keys
|
|
f.options.ForeignIndex = opt.ForeignIndex
|
|
|
|
// Create new bsiGroup.
|
|
bsig := &bsiGroup{
|
|
Name: f.name,
|
|
Type: bsiGroupTypeInt,
|
|
Min: opt.Min.ToInt64(opt.Scale),
|
|
Max: opt.Max.ToInt64(opt.Scale),
|
|
Base: opt.Base,
|
|
Scale: opt.Scale,
|
|
TimeUnit: opt.TimeUnit,
|
|
BitDepth: opt.BitDepth,
|
|
}
|
|
// Validate and create bsiGroup.
|
|
if err := f.createBSIGroup(bsig); err != nil {
|
|
return errors.Wrap(err, "creating bsigroup")
|
|
}
|
|
case FieldTypeTime:
|
|
f.options.Type = opt.Type
|
|
f.options.CacheType = CacheTypeNone
|
|
f.options.CacheSize = 0
|
|
f.options.Min = pql.Decimal{}
|
|
f.options.Max = pql.Decimal{}
|
|
f.options.Base = 0
|
|
f.options.BitDepth = 0
|
|
f.options.Keys = opt.Keys
|
|
f.options.NoStandardView = opt.NoStandardView
|
|
// Validate the time quantum.
|
|
if !opt.TimeQuantum.Valid() {
|
|
return ErrInvalidTimeQuantum
|
|
}
|
|
f.options.TimeQuantum = opt.TimeQuantum
|
|
f.options.TTL = opt.TTL
|
|
f.options.ForeignIndex = opt.ForeignIndex
|
|
case FieldTypeBool:
|
|
f.options.Type = FieldTypeBool
|
|
f.options.CacheType = CacheTypeNone
|
|
f.options.CacheSize = 0
|
|
f.options.Min = pql.Decimal{}
|
|
f.options.Max = pql.Decimal{}
|
|
f.options.Base = 0
|
|
f.options.BitDepth = 0
|
|
f.options.TimeQuantum = ""
|
|
f.options.TTL = 0
|
|
f.options.Keys = false
|
|
f.options.ForeignIndex = ""
|
|
default:
|
|
return errors.New("invalid field type")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Close closes the field and its views.
|
|
func (f *Field) Close() error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.unprotectedClose()
|
|
}
|
|
|
|
// unprotectedClose is the actual closing part of the operation, without the
|
|
// locking.
|
|
func (f *Field) unprotectedClose() error {
|
|
if f.closing != nil {
|
|
select {
|
|
case <-f.closing:
|
|
// already closed. prevent double-close
|
|
return errors.New("double close of field")
|
|
default:
|
|
}
|
|
close(f.closing)
|
|
}
|
|
defer func() {
|
|
_ = testhook.Closed(f.holder.Auditor, f, nil)
|
|
}()
|
|
// Shutdown the available shards writer
|
|
if f.availableShardChan != nil {
|
|
close(f.availableShardChan)
|
|
f.wg.Wait()
|
|
f.availableShardChan = nil
|
|
}
|
|
|
|
// Close field translation store.
|
|
if f.translateStore != nil {
|
|
if err := f.translateStore.Close(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Close all views.
|
|
for _, view := range f.viewMap {
|
|
if err := view.close(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
f.viewMap = make(map[string]*view)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (f *Field) flushCaches() {
|
|
// look up the close channel so if we somehow end up living until the
|
|
// field gets reopened, we don't have a data race, but correctly detect
|
|
// that the old one is closed.
|
|
f.mu.RLock()
|
|
closing := f.closing
|
|
f.mu.RUnlock()
|
|
for _, v := range f.views() {
|
|
select {
|
|
case <-closing:
|
|
return
|
|
default:
|
|
v.flushCaches()
|
|
}
|
|
}
|
|
}
|
|
|
|
// Keys returns true if the field uses string keys.
|
|
func (f *Field) Keys() bool {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
return f.usesKeys
|
|
}
|
|
|
|
// bsiGroup returns a bsiGroup by name.
|
|
func (f *Field) bsiGroup(name string) *bsiGroup {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
for _, bsig := range f.bsiGroups {
|
|
if bsig.Name == name {
|
|
return bsig
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// hasBSIGroup returns true if a bsiGroup exists on the field.
|
|
func (f *Field) hasBSIGroup(name string) bool {
|
|
for _, bsig := range f.bsiGroups {
|
|
if bsig.Name == name {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// cleanupViewName yields a "corrected" view name, handling some
|
|
// idioms we used elsewhere in code. Given an empty string,
|
|
// it yields a default view name (either "standard" or the BSI view
|
|
// for BSI fields). Given a string starting with numbers, it
|
|
// yields the corresponding time quantum view (prefixing "standard_").
|
|
// It yields an error if the view name given does not correspond
|
|
// to a view which should exist. For instance, the "standard" or
|
|
// "existence" views for a BSI field, or a time quantum view for
|
|
// a non-time field.
|
|
func (f *Field) cleanupViewName(viewName string) (string, error) {
|
|
if viewName == "" {
|
|
switch f.options.Type {
|
|
case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp:
|
|
return "bsig_" + f.name, nil
|
|
default:
|
|
return viewStandard, nil
|
|
}
|
|
}
|
|
switch f.options.Type {
|
|
case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp:
|
|
if viewName == "bsig_"+f.name {
|
|
return viewName, nil
|
|
}
|
|
return viewName, fmt.Errorf("BSI-type field view should be named bsig_[fieldname], got %q", viewName)
|
|
case FieldTypeTime:
|
|
switch {
|
|
case viewName == viewStandard, viewName == viewExistence:
|
|
return viewName, nil
|
|
case strings.HasPrefix(viewName, viewStandard):
|
|
return viewName, nil
|
|
case unicode.IsDigit(rune(viewName[0])):
|
|
return viewStandard + "_" + viewName, nil
|
|
default:
|
|
return viewName, fmt.Errorf("time field views are %q, %q, or %q_[digits], got %q", viewStandard, viewExistence, viewStandard, viewName)
|
|
}
|
|
default:
|
|
switch viewName {
|
|
case viewStandard, viewExistence:
|
|
return viewName, nil
|
|
default:
|
|
return viewName, fmt.Errorf("unexpected view name %q, expecting %q or %q", viewName, viewStandard, viewExistence)
|
|
}
|
|
}
|
|
}
|
|
|
|
// createBSIGroup creates a new bsiGroup on the field.
|
|
func (f *Field) createBSIGroup(bsig *bsiGroup) error {
|
|
// Append bsiGroup.
|
|
if err := bsig.validate(); err != nil {
|
|
return errors.Wrap(err, "validating bsigroup")
|
|
} else if f.hasBSIGroup(bsig.Name) {
|
|
return ErrBSIGroupExists
|
|
}
|
|
|
|
// Add bsiGroup to list.
|
|
f.bsiGroups = append(f.bsiGroups, bsig)
|
|
|
|
// Sort bsiGroups by name.
|
|
sort.Slice(f.bsiGroups, func(i, j int) bool {
|
|
return f.bsiGroups[i].Name < f.bsiGroups[j].Name
|
|
})
|
|
|
|
return nil
|
|
}
|
|
|
|
// TimeQuantum returns the time quantum for the field.
|
|
func (f *Field) TimeQuantum() TimeQuantum {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.options.TimeQuantum
|
|
}
|
|
|
|
// viewsByTimeRange is a wrapper on the non-method viewsByTimeRange,
|
|
// which computes views for a specific field for a given time
|
|
// range. The difference is that, it can return "standard" if from/to
|
|
// are not set and can automatically coerce from/to times to match the
|
|
// actual range present in the field.
|
|
func (f *Field) viewsByTimeRange(from, to time.Time) (views []string, err error) {
|
|
// if field is not a time field, return an error
|
|
// If we can't find time views at all, and standard view is available,
|
|
// yield standard view
|
|
// if we can't find time views, and standard view is disabled,
|
|
// yield union of all views
|
|
// yield "standard" if from and to were both not set and there is a
|
|
// standard view.
|
|
q := f.TimeQuantum()
|
|
if q == "" {
|
|
return nil, fmt.Errorf("field %s is not a time-field, 'from' and 'to' are not valid options for this field type", f.name)
|
|
}
|
|
|
|
if from.IsZero() && to.IsZero() && !f.options.NoStandardView {
|
|
return []string{viewStandard}, nil
|
|
}
|
|
|
|
// Get min/max based on existing views.
|
|
fv := f.views()
|
|
vs := make([]string, 0, len(fv))
|
|
for _, v := range fv {
|
|
vs = append(vs, v.name)
|
|
}
|
|
min, max := minMaxViews(vs, q)
|
|
|
|
// If min/max are empty, there were no time views.
|
|
if min == "" || max == "" {
|
|
return []string{}, nil
|
|
}
|
|
|
|
// Convert min/max from string to time.Time.
|
|
minTime, err := timeOfView(min, false)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "getting min time from view: %s", min)
|
|
}
|
|
if from.IsZero() || from.Before(minTime) {
|
|
from = minTime
|
|
}
|
|
|
|
maxTime, err := timeOfView(max, true)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "getting max time from view: %s", max)
|
|
}
|
|
if to.IsZero() || to.After(maxTime) {
|
|
to = maxTime
|
|
}
|
|
return viewsByTimeRange(viewStandard, from, to, q), nil
|
|
}
|
|
|
|
// RowTime gets the row at the particular time with the granularity specified by
|
|
// the quantum.
|
|
func (f *Field) RowTime(qcx *Qcx, rowID uint64, time time.Time, quantum string) (*Row, error) {
|
|
if !TimeQuantum(quantum).Valid() {
|
|
return nil, ErrInvalidTimeQuantum
|
|
}
|
|
viewname := viewByTimeUnit(viewStandard, time, rune(quantum[len(quantum)-1]))
|
|
view := f.view(viewname)
|
|
if view == nil {
|
|
return nil, errors.Errorf("view with quantum %v not found.", quantum)
|
|
}
|
|
|
|
return view.row(qcx, rowID)
|
|
}
|
|
|
|
// viewPath returns the path to a view in the field.
|
|
func (f *Field) viewPath(name string) string {
|
|
return filepath.Join(f.path, "views", name)
|
|
}
|
|
|
|
// view returns a view in the field by name.
|
|
func (f *Field) view(name string) *view {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
return f.unprotectedView(name)
|
|
}
|
|
|
|
func (f *Field) unprotectedView(name string) *view { return f.viewMap[name] }
|
|
|
|
// views returns a list of all views in the field.
|
|
func (f *Field) views() []*view {
|
|
f.mu.RLock()
|
|
defer f.mu.RUnlock()
|
|
|
|
other := make([]*view, 0, len(f.viewMap))
|
|
for _, view := range f.viewMap {
|
|
other = append(other, view)
|
|
}
|
|
return other
|
|
}
|
|
|
|
// recalculateCaches recalculates caches on every view in the field.
|
|
func (f *Field) recalculateCaches() {
|
|
for _, view := range f.views() {
|
|
view.recalculateCaches()
|
|
}
|
|
}
|
|
|
|
// createViewIfNotExists returns the named view, creating it if necessary.
|
|
// Additionally, a CreateViewMessage is sent to the cluster.
|
|
func (f *Field) createViewIfNotExists(name string) (*view, error) {
|
|
cvm := &CreateViewMessage{
|
|
Index: f.index,
|
|
Field: f.name,
|
|
View: name,
|
|
}
|
|
|
|
// call this base method to isolate the mu.Lock and ensure we aren't holding
|
|
// the lock while calling SendSync below.
|
|
view, created, err := f.createViewIfNotExistsBase(cvm)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if created {
|
|
// Broadcast view creation to the cluster.
|
|
err := f.holder.sendOrSpool(cvm)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "sending CreateView message")
|
|
}
|
|
}
|
|
|
|
return view, nil
|
|
}
|
|
|
|
// createViewIfNotExistsBase returns the named view, creating it if necessary.
|
|
// One purpose of isolating this method from createViewIfNotExists() is that we
|
|
// need to enforce the mu.Lock on everything in this method, but we can't be
|
|
// holding the lock when broadcasting the CreateViewMessage view
|
|
// broadcaster.SendSync(); calling that SendSync() while holding the lock can
|
|
// result in a deadlock waiting on the remote node to give up its lock obtained
|
|
// by performing the same action. The returned bool indicates whether the view
|
|
// was created or not.
|
|
func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
|
|
// If we already have this view, we can probably assume etcd already
|
|
// has it.
|
|
if view := f.viewMap[cvm.View]; view != nil && !view.isClosing() {
|
|
return view, false, nil
|
|
}
|
|
|
|
// Create the view in etcd as the system of record.
|
|
// Don't persist views related to the existence field.
|
|
if f.name != existenceFieldName {
|
|
if err := f.persistView(context.Background(), cvm); err != nil {
|
|
return nil, false, errors.Wrap(err, "persisting view")
|
|
}
|
|
}
|
|
view := f.newView(f.viewPath(cvm.View), cvm.View)
|
|
|
|
if err := view.openEmpty(); err != nil {
|
|
return nil, false, errors.Wrap(err, "opening view")
|
|
}
|
|
f.viewMap[view.name] = view
|
|
|
|
return view, true, nil
|
|
}
|
|
|
|
func (f *Field) newView(path, name string) *view {
|
|
view := newView(f.holder, path, f.index, f.name, name, f.options)
|
|
view.idx = f.idx
|
|
view.fld = f
|
|
view.broadcaster = f.broadcaster
|
|
return view
|
|
}
|
|
|
|
// deleteView removes the view from the field.
|
|
func (f *Field) deleteView(name string) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
view := f.viewMap[name]
|
|
if view == nil {
|
|
return ErrInvalidView
|
|
}
|
|
|
|
// Delete the view from etcd as the system of record.
|
|
if err := f.holder.Schemator.DeleteView(context.TODO(), f.index, f.name, name); err != nil {
|
|
return errors.Wrapf(err, "deleting view from etcd: %s/%s/%s", f.index, f.name, name)
|
|
}
|
|
|
|
// Close data files before deletion.
|
|
if err := view.close(); err != nil {
|
|
return errors.Wrap(err, "closing view")
|
|
}
|
|
|
|
// Delete view directory.
|
|
if err := os.RemoveAll(view.path); err != nil {
|
|
return errors.Wrap(err, "deleting directory")
|
|
}
|
|
|
|
delete(f.viewMap, name)
|
|
|
|
return nil
|
|
}
|
|
|
|
// Row returns a row of the standard view.
|
|
// It seems this method is only being used by the test
|
|
// package, and the fact that it's only allowed on
|
|
// `set`,`mutex`, and `bool` fields is odd. This may
|
|
// be considered for deprecation in a future version.
|
|
func (f *Field) Row(qcx *Qcx, rowID uint64) (*Row, error) {
|
|
switch f.Type() {
|
|
case FieldTypeSet, FieldTypeMutex, FieldTypeBool:
|
|
view := f.view(viewStandard)
|
|
if view == nil {
|
|
return nil, ErrInvalidView
|
|
}
|
|
return view.row(qcx, rowID)
|
|
default:
|
|
return nil, errors.Errorf("row method unsupported for field type: %s", f.Type())
|
|
}
|
|
}
|
|
|
|
// mutexCheck performs a sanity-check on the available fragments for a
|
|
// field. The return is map[column]map[shard][]values for collisions only.
|
|
func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
|
|
if f.Type() != FieldTypeMutex {
|
|
return nil, errors.New("mutex check only valid for mutex fields")
|
|
}
|
|
|
|
// Rather than deferring the unlock, we grab the standard view
|
|
// from the field's viewMap and unlock immediately. This avoids
|
|
// holding the rlock for a potentially long time which blocks any
|
|
// write lock, and pending write locks block other read locks.
|
|
f.mu.RLock()
|
|
standard := f.viewMap[viewStandard]
|
|
f.mu.RUnlock()
|
|
|
|
if standard == nil {
|
|
// no standard view present means we've never needed to create it,
|
|
// so it has no bits set, so it has no extra bits set.
|
|
return nil, nil
|
|
}
|
|
return standard.mutexCheck(ctx, qcx, details, limit)
|
|
}
|
|
|
|
// SetBit sets a bit on a view within the field.
|
|
func (f *Field) SetBit(qcx *Qcx, rowID, colID uint64, t *time.Time) (changed bool, err error) {
|
|
viewName := viewStandard
|
|
if !f.options.NoStandardView {
|
|
// Retrieve view. Exit if it doesn't exist.
|
|
view, err := f.createViewIfNotExists(viewName)
|
|
if err != nil {
|
|
return changed, errors.Wrap(err, "creating view")
|
|
}
|
|
|
|
// Set non-time bit.
|
|
if v, err := view.setBit(qcx, rowID, colID); err != nil {
|
|
return changed, errors.Wrap(err, "setting on view")
|
|
} else if v {
|
|
changed = v
|
|
}
|
|
if f.options.TrackExistence {
|
|
view, err := f.createViewIfNotExists(viewExistence)
|
|
if err != nil {
|
|
return changed, errors.Wrap(err, "creating existence view")
|
|
}
|
|
if _, err := view.setBit(qcx, bsiExistsBit, colID); err != nil {
|
|
return changed, errors.Wrap(err, "setting existence on view")
|
|
}
|
|
}
|
|
}
|
|
|
|
// Exit early if no timestamp is specified.
|
|
if t == nil {
|
|
return changed, nil
|
|
}
|
|
|
|
// If a timestamp is specified then set bits across all views for the quantum.
|
|
for _, subname := range viewsByTime(viewName, *t, f.TimeQuantum()) {
|
|
view, err := f.createViewIfNotExists(subname)
|
|
if err != nil {
|
|
return changed, errors.Wrapf(err, "creating view %s", subname)
|
|
}
|
|
|
|
if c, err := view.setBit(qcx, rowID, colID); err != nil {
|
|
return changed, errors.Wrapf(err, "setting on view %s", subname)
|
|
} else if c {
|
|
changed = true
|
|
}
|
|
}
|
|
|
|
return changed, nil
|
|
}
|
|
|
|
// ClearBit clears a bit within the field.
|
|
//
|
|
// This does not, for now, create existence bits for the field, because it
|
|
// doesn't create them for the index.
|
|
func (f *Field) ClearBit(qcx *Qcx, rowID, colID uint64) (changed bool, err error) {
|
|
viewName := viewStandard
|
|
|
|
// Retrieve view. Exit if it doesn't exist.
|
|
view, present := f.viewMap[viewName]
|
|
if !present {
|
|
return false, errors.Wrap(err, "clearing missing view")
|
|
}
|
|
|
|
// Clear non-time bit.
|
|
if v, err := view.clearBit(qcx, rowID, colID); err != nil {
|
|
return false, errors.Wrap(err, "clearing on view")
|
|
} else if v {
|
|
changed = changed || v
|
|
if changed && f.options.TrackExistence && f.options.Type == FieldTypeMutex {
|
|
// we also want to try to clear any existence bit
|
|
existView, ok := f.viewMap[viewExistence]
|
|
if ok {
|
|
_, err := existView.clearBit(qcx, 0, colID)
|
|
if err != nil {
|
|
return false, errors.Wrap(err, "clearing existence bit")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// We used to check length of view map here. Now that we might
|
|
// have an existence view, that won't work.
|
|
if f.options.Type != FieldTypeTime { // assuming no time views
|
|
return changed, nil
|
|
}
|
|
lastViewNameSize := 0
|
|
level := 0
|
|
skipAbove := maxInt
|
|
for _, view := range f.allTimeViewsSortedByQuantum() {
|
|
if lastViewNameSize < len(view.name) {
|
|
level++
|
|
} else if lastViewNameSize > len(view.name) {
|
|
level--
|
|
}
|
|
if level < skipAbove {
|
|
cleared, err := view.clearBit(qcx, rowID, colID)
|
|
changed = changed || cleared
|
|
if err != nil {
|
|
return changed, errors.Wrapf(err, "clearing on view %s", view.name)
|
|
}
|
|
if !cleared {
|
|
skipAbove = level + 1
|
|
} else {
|
|
skipAbove = maxInt
|
|
}
|
|
}
|
|
lastViewNameSize = len(view.name)
|
|
}
|
|
|
|
return changed, nil
|
|
}
|
|
|
|
func groupCompare(a, b string, offset int) (lt, eq bool) {
|
|
if len(a) > offset {
|
|
a = a[:offset]
|
|
}
|
|
if len(b) > offset {
|
|
b = b[:offset]
|
|
}
|
|
v := strings.Compare(a, b)
|
|
return v < 0, v == 0
|
|
}
|
|
|
|
func (f *Field) allTimeViewsSortedByQuantum() (me []*view) {
|
|
me = make([]*view, len(f.viewMap))
|
|
prefix := viewStandard + "_"
|
|
offset := len(viewStandard) + 1
|
|
i := 0
|
|
for _, v := range f.viewMap {
|
|
if len(v.name) > offset && strings.Compare(v.name[:offset], prefix) == 0 { // skip non-time views
|
|
me[i] = v
|
|
i++
|
|
}
|
|
}
|
|
// return the empty list if there weren't any. this could happen
|
|
// if we got called because this is a time field, but in fact
|
|
// no time views have been created.
|
|
if i == 0 {
|
|
return me[:0]
|
|
}
|
|
me = me[:i]
|
|
year := strings.Index(me[0].name, "_") + 4
|
|
month := year + 2
|
|
day := month + 2
|
|
sort.Slice(me, func(i, j int) (lt bool) {
|
|
var eq bool
|
|
// group by quantum from year to hour
|
|
if lt, eq = groupCompare(me[i].name, me[j].name, year); eq {
|
|
if lt, eq = groupCompare(me[i].name, me[j].name, month); eq {
|
|
if lt, eq = groupCompare(me[i].name, me[j].name, day); eq {
|
|
lt = strings.Compare(me[i].name, me[j].name) > 0
|
|
}
|
|
}
|
|
}
|
|
return lt
|
|
})
|
|
return me
|
|
}
|
|
|
|
// StringValue reads an integer field value for a column, and converts
|
|
// it to a string based on a foreign index string key.
|
|
func (f *Field) StringValue(qcx *Qcx, columnID uint64) (value string, exists bool, err error) {
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return value, false, ErrBSIGroupNotFound
|
|
}
|
|
|
|
val, exists, err := f.Value(qcx, columnID)
|
|
if exists {
|
|
value, err = f.translateStore.TranslateID(uint64(val))
|
|
}
|
|
return value, exists, err
|
|
}
|
|
|
|
// Value reads a field value for a column.
|
|
func (f *Field) Value(qcx *Qcx, columnID uint64) (value int64, exists bool, err error) {
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return 0, false, ErrBSIGroupNotFound
|
|
}
|
|
|
|
// Fetch target view.
|
|
view := f.view(viewBSIGroupPrefix + f.name)
|
|
if view == nil {
|
|
return 0, false, nil
|
|
}
|
|
|
|
v, exists, err := view.value(qcx, columnID, bsig.BitDepth)
|
|
if err != nil {
|
|
return 0, false, err
|
|
} else if !exists {
|
|
return 0, false, nil
|
|
}
|
|
return int64(v) + bsig.Base, true, nil
|
|
}
|
|
|
|
// SetValue sets a field value for a column.
|
|
func (f *Field) SetValue(qcx *Qcx, columnID uint64, value int64) (changed bool, err error) {
|
|
// Fetch bsiGroup & validate min/max.
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return false, ErrBSIGroupNotFound
|
|
}
|
|
|
|
// Determine base value to store.
|
|
baseValue := int64(value - bsig.Base)
|
|
//Timestamp expects incoming value to already be relative to epoch
|
|
if f.Type() == FieldTypeTimestamp {
|
|
value = baseValue
|
|
}
|
|
if value < bsig.Min {
|
|
return false, errors.Wrapf(ErrBSIGroupValueTooLow, "index = %v, field = %v, column ID = %v, value %v is smaller than min allowed %v", f.index, f.name, columnID, value, bsig.Min)
|
|
} else if value > bsig.Max {
|
|
return false, errors.Wrapf(ErrBSIGroupValueTooHigh, "index = %v, field = %v, column ID = %v, value %v is larger than max allowed %v", f.index, f.name, columnID, value, bsig.Max)
|
|
}
|
|
|
|
requiredBitDepth := bitDepthInt64(baseValue)
|
|
|
|
// Increase bit depth value if the unsigned value is greater.
|
|
if requiredBitDepth > bsig.BitDepth {
|
|
uvalue := uint64(baseValue)
|
|
if value < 0 {
|
|
uvalue = uint64(-baseValue)
|
|
}
|
|
bitDepth := bitDepth(uvalue)
|
|
|
|
f.mu.Lock()
|
|
bsig.BitDepth = bitDepth
|
|
f.options.BitDepth = bitDepth
|
|
f.mu.Unlock()
|
|
}
|
|
|
|
// Fetch target view.
|
|
view, err := f.createViewIfNotExists(viewBSIGroupPrefix + f.name)
|
|
if err != nil {
|
|
return false, errors.Wrap(err, "creating view")
|
|
}
|
|
if view.holder == nil {
|
|
panic("view.holder should not be nil")
|
|
}
|
|
if view.idx == nil {
|
|
panic("view.idx should not be nil")
|
|
}
|
|
view.holder.addIndex(view.idx)
|
|
|
|
return view.setValue(qcx, columnID, bsig.BitDepth, baseValue)
|
|
}
|
|
|
|
// ClearValue removes a field value for a column.
|
|
func (f *Field) ClearValue(qcx *Qcx, columnID uint64) (changed bool, err error) {
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return false, ErrBSIGroupNotFound
|
|
}
|
|
// Fetch target view.
|
|
view := f.view(viewBSIGroupPrefix + f.name)
|
|
if view == nil {
|
|
return false, nil
|
|
}
|
|
value, exists, err := view.value(qcx, columnID, bsig.BitDepth)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if exists {
|
|
return view.clearValue(qcx, columnID, bsig.BitDepth, value)
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
func (f *Field) MaxForShard(qcx *Qcx, shard uint64, filter *Row) (ValCount, error) {
|
|
tx, finisher, err := qcx.GetTx(Txo{Write: false, Index: f.idx, Shard: shard})
|
|
defer finisher(&err)
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return ValCount{}, ErrBSIGroupNotFound
|
|
}
|
|
|
|
view := f.view(viewBSIGroupPrefix + f.name)
|
|
if view == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
fragment := view.Fragment(shard)
|
|
if fragment == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
max, cnt, err := fragment.max(tx, filter, bsig.BitDepth)
|
|
if err != nil {
|
|
return ValCount{}, errors.Wrap(err, "calling fragment.max")
|
|
}
|
|
|
|
v, err := f.valCountize(max, cnt, bsig)
|
|
return v, err
|
|
}
|
|
|
|
// MinForShard returns the minimum value which appears in this shard
|
|
// (this field must be an Int or Decimal field). It also returns the
|
|
// number of times the minimum value appears.
|
|
func (f *Field) MinForShard(qcx *Qcx, shard uint64, filter *Row) (ValCount, error) {
|
|
tx, finisher, err := qcx.GetTx(Txo{Write: false, Index: f.idx, Shard: shard})
|
|
defer finisher(&err)
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return ValCount{}, ErrBSIGroupNotFound
|
|
}
|
|
|
|
view := f.view(viewBSIGroupPrefix + f.name)
|
|
if view == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
fragment := view.Fragment(shard)
|
|
if fragment == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
min, cnt, err := fragment.min(tx, filter, bsig.BitDepth)
|
|
if err != nil {
|
|
return ValCount{}, errors.Wrap(err, "calling fragment.min")
|
|
}
|
|
|
|
v, err := f.valCountize(min, cnt, bsig)
|
|
return v, err
|
|
}
|
|
|
|
// valCountize takes the "raw" value and count we get from the
|
|
// fragment and calculates the cooked values for this field
|
|
// (timestamping, decimaling, or just adding in the base). It always
|
|
// includes the int64 "Val\" value to make comparisons easier in the
|
|
// executor (at time of writing, Percentile takes advantage of this,
|
|
// but we might be able to simplify logic in other places as well).
|
|
func (f *Field) valCountize(val int64, cnt uint64, bsig *bsiGroup) (ValCount, error) {
|
|
if bsig == nil {
|
|
bsig = f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return ValCount{}, ErrBSIGroupNotFound
|
|
}
|
|
|
|
}
|
|
valCount := ValCount{Count: int64(cnt)}
|
|
|
|
if f.Options().Type == FieldTypeDecimal {
|
|
dec := pql.NewDecimal(val+bsig.Base, bsig.Scale)
|
|
valCount.DecimalVal = &dec
|
|
} else if f.Options().Type == FieldTypeTimestamp {
|
|
ts, err := ValToTimestamp(f.options.TimeUnit, val+bsig.Base)
|
|
if err != nil {
|
|
return ValCount{}, errors.Wrap(err, "translating value to timestamp")
|
|
}
|
|
valCount.TimestampVal = ts
|
|
// valCount.TimestampVal = time.Unix(0, (val+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
|
|
}
|
|
|
|
valCount.Val = val + bsig.Base
|
|
return valCount, nil
|
|
}
|
|
|
|
// Range performs a conditional operation on Field.
|
|
func (f *Field) Range(qcx *Qcx, name string, op pql.Token, predicate int64) (*Row, error) {
|
|
// Retrieve and validate bsiGroup.
|
|
bsig := f.bsiGroup(name)
|
|
if bsig == nil {
|
|
return nil, ErrBSIGroupNotFound
|
|
} else if predicate < bsig.Min || predicate > bsig.Max {
|
|
return nil, nil
|
|
}
|
|
|
|
// Retrieve bsiGroup's view.
|
|
view := f.view(viewBSIGroupPrefix + name)
|
|
if view == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
baseValue, outOfRange := bsig.baseValue(op, predicate)
|
|
if outOfRange {
|
|
return NewRow(), nil
|
|
}
|
|
|
|
return view.rangeOp(qcx, op, bsig.BitDepth, baseValue)
|
|
}
|
|
|
|
// existenceViewName reports the field we should use row 0 of
|
|
// for existence data. For a BSI field (integer, decimal,
|
|
// timestamp) this is the single BSI group. For other fields,
|
|
// it's viewExistence, which is probably "existence".
|
|
func (f *Field) existenceViewName() string {
|
|
if len(f.bsiGroups) > 0 {
|
|
return f.bsiGroups[0].Name
|
|
}
|
|
return viewExistence
|
|
}
|
|
|
|
// MarkExisting sets a range of column IDs as existing. The columnIDs
|
|
// are assumed to include the shard offset, but this will also work if
|
|
// they are shard-relative, as it's just stripping the offset.
|
|
//
|
|
// Positions aren't the same as column IDs; this function takes advantage
|
|
// of the fact that we're always doing row 0, so we don't have to think
|
|
// hard about this. It doesn't overwrite its input because the column IDs
|
|
// could be reused by other things.
|
|
//
|
|
// Note that this is subtly inefficient; if you're tracking existence for
|
|
// a field, we're computing the same column ID set to write to the index's
|
|
// existence field as we're using for the field's existence view. We don't
|
|
// have a good way to coalesce those, yet. (Also, that's not accurate in
|
|
// the ImportValue case, where we don't write to the existence view, etc.)
|
|
func (f *Field) MarkExisting(tx Tx, columnIDs []uint64, shard uint64) error {
|
|
return f.markExistingInView(tx, columnIDs, f.existenceViewName(), shard)
|
|
}
|
|
|
|
// markExistingInView implements the internals of MarkExisting, but lets
|
|
// you use a non-standard view. It's only interesting for the existence field.
|
|
func (f *Field) markExistingInView(tx Tx, columnIDs []uint64, viewName string, shard uint64) error {
|
|
copyCols := make([]uint64, len(columnIDs))
|
|
for i := range columnIDs {
|
|
copyCols[i] = columnIDs[i] % ShardWidth
|
|
}
|
|
eView, err := f.createViewIfNotExists(viewName)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "creating view %s", viewName)
|
|
}
|
|
|
|
eFrag, err := eView.CreateFragmentIfNotExists(shard)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating fragment")
|
|
}
|
|
return eFrag.importPositions(tx, copyCols, nil, map[uint64]struct{}{0: {}})
|
|
}
|
|
|
|
// MarkNotExisting is just like MarkExisting, except it is clearing bits,
|
|
// so it doesn't have to create the view or fragment if it doesn't exist.
|
|
// Because the bits reported to us in the case we wrote this for are likely
|
|
// to be sorted by position in the fragment, not by column ID, we sort the
|
|
// list after stripping the rows from the positions.
|
|
func (f *Field) MarkNotExisting(tx Tx, columnIDs []uint64, shard uint64) error {
|
|
viewName := f.existenceViewName()
|
|
v := f.view(viewName)
|
|
if v == nil {
|
|
return nil
|
|
}
|
|
frag := v.Fragment(shard)
|
|
if frag == nil {
|
|
return nil
|
|
}
|
|
copyCols := make([]uint64, len(columnIDs))
|
|
for i := range columnIDs {
|
|
copyCols[i] = columnIDs[i] % ShardWidth
|
|
}
|
|
sort.Slice(copyCols, func(i, j int) bool { return copyCols[i] < copyCols[j] })
|
|
return frag.importPositions(tx, nil, copyCols, map[uint64]struct{}{0: {}})
|
|
}
|
|
|
|
// Existing returns the existence row for this field, which
|
|
// comes from either the BSI view or the existence view.
|
|
func (f *Field) Existing(tx Tx, shard uint64) (*Row, error) {
|
|
viewName := f.existenceViewName()
|
|
v := f.view(viewName)
|
|
if v == nil {
|
|
return nil, nil
|
|
}
|
|
frag := v.Fragment(shard)
|
|
if frag == nil {
|
|
return nil, nil
|
|
}
|
|
return frag.row(tx, bsiExistsBit)
|
|
}
|
|
|
|
// Import bulk imports data.
|
|
func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, shard uint64, options *ImportOptions) (err0 error) {
|
|
// Determine quantum if timestamps are set.
|
|
q := f.TimeQuantum()
|
|
if len(timestamps) > 0 {
|
|
if q == "" {
|
|
return errors.New("time quantum not set in field")
|
|
} else if options.Clear {
|
|
return errors.New("import clear is not supported with timestamps")
|
|
}
|
|
} else {
|
|
if f.options.NoStandardView {
|
|
return errors.New("can't import data with no timestamps into a field with no standard view")
|
|
}
|
|
// short path: if we don't have any timestamps, we only need
|
|
// to write to exactly one view, which is always viewStandard,
|
|
// and *every* bit goes into that view, and we already verified that
|
|
// everything is in the same shard, so we can skip most of this.
|
|
fieldType := f.Type()
|
|
if fieldType == FieldTypeBool {
|
|
for _, rowID := range rowIDs {
|
|
if rowID > 1 {
|
|
return errors.New("bool field imports only support values 0 and 1")
|
|
}
|
|
}
|
|
}
|
|
tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: f.idx, Shard: shard})
|
|
if err != nil {
|
|
return errors.Wrap(err, "qcx.GetTx")
|
|
}
|
|
var err1 error
|
|
defer finisher(&err1)
|
|
view, err := f.createViewIfNotExists(viewStandard)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "creating view %s", viewStandard)
|
|
}
|
|
|
|
frag, err := view.CreateFragmentIfNotExists(shard)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating fragment")
|
|
}
|
|
if f.options.TrackExistence {
|
|
// if we're clearing a mutex, we do something fancy. otherwise,
|
|
// if we're not clearing, we mark the existence bits. either way,
|
|
// we then fall on out to the default behavior of importing the
|
|
// bits.
|
|
switch {
|
|
case options.Clear && fieldType == FieldTypeMutex:
|
|
// special fancy case; we have to try to clear the bits first, to
|
|
// find out WHICH bits we cleared, so we can mark those bits as
|
|
// null.
|
|
var changed []uint64
|
|
changed, err1 = frag.clearBitsReportingChanges(tx, rowIDs, columnIDs)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
err1 = f.MarkNotExisting(tx, changed, shard)
|
|
return err1
|
|
case !options.Clear:
|
|
err1 = f.MarkExisting(tx, columnIDs, shard)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
default:
|
|
// nothing to do. we'll fall on out of the TrackExistence
|
|
// case and go ahead and import those bits naively
|
|
}
|
|
}
|
|
err1 = frag.bulkImport(tx, rowIDs, columnIDs, options)
|
|
return err1
|
|
}
|
|
|
|
fieldType := f.Type()
|
|
|
|
// Split import data by fragment.
|
|
views := make(map[string]*importData)
|
|
var timeStringBuf []byte
|
|
var timeViews [][]byte
|
|
if len(q) > 0 {
|
|
// We're supporting time quantums, so we need to store bits in a
|
|
// number of views for every entry with a timestamp. We want to compute
|
|
// time quantum view names for whatever combination of YMDH views
|
|
// we have. But we don't want to allocate four strings per entry, or
|
|
// recompute and recreate the entire string. We know that only the
|
|
// YYYYMMDDHH part of the string changes over time.
|
|
timeStringBuf = make([]byte, len(viewStandard)+11)
|
|
copy(timeStringBuf, []byte(viewStandard))
|
|
copy(timeStringBuf[len(viewStandard):], []byte("_YYYYMMDDHH"))
|
|
// Now we have a buffer that contains
|
|
// `standard_YYYYMMDDHH`. We also need storage space to hold several
|
|
// slice headers, one per entry in q. These will hold the view names
|
|
// corresponding to each letter in q.
|
|
timeViews = make([][]byte, len(q))
|
|
}
|
|
// This helper function records that a given column/row pair is relevant
|
|
// to a specific view. We use a map lookup for the strings, but do the
|
|
// actual operations using a slice so we're only writing each map entry
|
|
// once, not once on every update.
|
|
see := func(name []byte, columnID uint64, rowID uint64) {
|
|
var ok bool
|
|
var data *importData
|
|
if data, ok = views[string(name)]; !ok {
|
|
data = &importData{}
|
|
views[string(name)] = data
|
|
}
|
|
data.RowIDs = append(data.RowIDs, rowID)
|
|
data.ColumnIDs = append(data.ColumnIDs, columnID)
|
|
}
|
|
for i := range rowIDs {
|
|
rowID, columnID := rowIDs[i], columnIDs[i]
|
|
|
|
// Bool-specific data validation.
|
|
if fieldType == FieldTypeBool && rowID > 1 {
|
|
return errors.New("bool field imports only support values 0 and 1")
|
|
}
|
|
|
|
hasTime := len(timestamps) > i && timestamps[i] != 0
|
|
|
|
// attach bit to standard view unless we have a timestamp and
|
|
// have the NoStandardView option set
|
|
if !hasTime || !f.options.NoStandardView {
|
|
see([]byte(viewStandard), columnID, rowID)
|
|
}
|
|
if hasTime {
|
|
// attach bit to all the views for this timestamp. note that the
|
|
// `timeViews` slice gets resliced and reused by this process, so
|
|
// we don't have to allocate millions of tiny slices of slice headers.
|
|
timeViews = viewsByTimeInto(timeStringBuf, timeViews, time.Unix(0, timestamps[i]).UTC(), q)
|
|
for _, v := range timeViews {
|
|
see(v, columnID, rowID)
|
|
}
|
|
}
|
|
}
|
|
tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: f.idx, Shard: shard})
|
|
if err != nil {
|
|
return errors.Wrap(err, "qcx.GetTx")
|
|
}
|
|
var err1 error
|
|
defer finisher(&err1)
|
|
for viewName, data := range views {
|
|
view, err := f.createViewIfNotExists(viewName)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "creating view %s", viewName)
|
|
}
|
|
|
|
frag, err := view.CreateFragmentIfNotExists(shard)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating fragment")
|
|
}
|
|
|
|
err1 = frag.bulkImport(tx, data.RowIDs, data.ColumnIDs, options)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
}
|
|
// If we are tracking existence, and don't have NoStandardView, we
|
|
// create the existence view.
|
|
if f.options.TrackExistence && !f.options.NoStandardView {
|
|
// this dance with err1 is so the finisher gets called with
|
|
// the right error value if we hit an error
|
|
err1 = f.MarkExisting(tx, columnIDs, shard)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// importFloatValue imports floating point values. In current usage, this
|
|
// should only ever be called with data for a single shard; the API calls
|
|
// around this are splitting it up per shard.
|
|
func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, shard uint64, options *ImportOptions) error {
|
|
// convert values to int64 values based on scale
|
|
ivalues := make([]int64, len(values))
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return errors.Wrap(ErrBSIGroupNotFound, f.name)
|
|
}
|
|
mult := math.Pow10(int(bsig.Scale))
|
|
for i, fval := range values {
|
|
ivalues[i] = int64(fval * mult)
|
|
}
|
|
// then call importValue
|
|
return f.importValue(qcx, columnIDs, ivalues, shard, options)
|
|
}
|
|
|
|
// importTimestampValue imports timestamp values. In current usage, this
|
|
// should only ever be called with data for a single shard; the API calls
|
|
// around this are splitting it up per shard.
|
|
func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time.Time, shard uint64, options *ImportOptions) error {
|
|
ivalues := make([]int64, len(values))
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return errors.Wrap(ErrBSIGroupNotFound, f.name)
|
|
}
|
|
|
|
for i, t := range values {
|
|
ivalues[i] = TimestampToVal(f.options.TimeUnit, t)
|
|
}
|
|
return f.importValue(qcx, columnIDs, ivalues, shard, options)
|
|
}
|
|
|
|
// importValue bulk imports range-encoded value data. This function should
|
|
// only be called with data for a single shard; the API calls that wrap
|
|
// this handle splitting the data up per-shard.
|
|
func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard uint64, options *ImportOptions) (err0 error) {
|
|
// no data to import
|
|
if len(columnIDs) == 0 {
|
|
return nil
|
|
}
|
|
if len(values) != len(columnIDs) {
|
|
return fmt.Errorf("importValue: mismatch between column IDs and values: %d != %d", len(columnIDs), len(values))
|
|
}
|
|
viewName := viewBSIGroupPrefix + f.name
|
|
// Get the bsiGroup so we know bitDepth.
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return errors.Wrap(ErrBSIGroupNotFound, f.name)
|
|
}
|
|
|
|
// We want to determine the required bit depth, in case the field doesn't
|
|
// have as many bits currently as would be needed to represent these values,
|
|
// but only if the values are in-range for the field.
|
|
min, max := values[0], values[0]
|
|
|
|
// Check for minimum/maximum in case we need to expand the field's
|
|
// stated bit depth.
|
|
for i := range columnIDs {
|
|
columnID, value := columnIDs[i], values[i]
|
|
if value > bsig.Max {
|
|
return errors.Wrapf(ErrBSIGroupValueTooHigh, "index = %v, field = %v, column ID = %v, value %v is larger than max allowed %v", f.index, f.name, columnID, value, bsig.Max)
|
|
} else if value < bsig.Min {
|
|
return errors.Wrapf(ErrBSIGroupValueTooLow, "index = %v, field = %v, column ID = %v, value %v is smaller than min allowed %v", f.index, f.name, columnID, value, bsig.Min)
|
|
}
|
|
if value > max {
|
|
max = value
|
|
}
|
|
if value < min {
|
|
min = value
|
|
}
|
|
|
|
}
|
|
|
|
// Timestamps differ from other BSI fields in that integer representations
|
|
// of timestamps are already relative to the epoch (base).
|
|
// So a user may set an epoch to 2022-03-01 as the start of a race
|
|
// and import finishing times in seconds.
|
|
// Timestamps ingested as timestamps are of course absolute, but by the time
|
|
// we get here it would be a relative integer.
|
|
if f.Type() != FieldTypeTimestamp {
|
|
min -= bsig.Base
|
|
max -= bsig.Base
|
|
}
|
|
|
|
// Determine the highest bit depth required by the min & max.
|
|
requiredDepth := bitDepthInt64(min)
|
|
if v := bitDepthInt64(max); v > requiredDepth {
|
|
requiredDepth = v
|
|
}
|
|
// Increase bit depth if required.
|
|
f.mu.Lock()
|
|
bitDepth := bsig.BitDepth
|
|
if requiredDepth > bitDepth {
|
|
bsig.BitDepth = requiredDepth
|
|
f.options.BitDepth = requiredDepth
|
|
} else {
|
|
requiredDepth = bitDepth
|
|
}
|
|
f.mu.Unlock()
|
|
|
|
if columnIDs[0]/ShardWidth != shard {
|
|
return fmt.Errorf("requested import for shard %d, got record ID for shard %d", shard, columnIDs[0]/ShardWidth)
|
|
}
|
|
|
|
view, err := f.createViewIfNotExists(viewName)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating view")
|
|
}
|
|
|
|
frag, err := view.CreateFragmentIfNotExists(shard)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating fragment")
|
|
}
|
|
|
|
if bsig.Base != 0 {
|
|
for i, v := range values {
|
|
// for Timestamps, values are already relative to their base (epoch)
|
|
// for other types (IntFields), values need to be subtracted from their base (either Min or Max)
|
|
if f.Type() == FieldTypeTimestamp {
|
|
values[i] = v
|
|
} else {
|
|
values[i] = v - bsig.Base
|
|
}
|
|
}
|
|
}
|
|
|
|
// now we know which shard we discovered.
|
|
tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: f.idx, Shard: frag.shard})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// defer the finisher, so it will check the error returned and
|
|
// possibly rollback.
|
|
defer finisher(&err0)
|
|
|
|
return frag.importValue(tx, columnIDs, values, requiredDepth, options.Clear)
|
|
}
|
|
|
|
func (f *Field) importRoaring(ctx context.Context, tx Tx, data []byte, shard uint64, viewName string, clear bool) error {
|
|
span, ctx := tracing.StartSpanFromContext(ctx, "Field.importRoaring")
|
|
defer span.Finish()
|
|
|
|
if viewName == "" {
|
|
viewName = viewStandard
|
|
}
|
|
span.LogKV("view", viewName, "bytes", len(data), "shard", shard)
|
|
view, err := f.createViewIfNotExists(viewName)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating view")
|
|
}
|
|
|
|
frag, err := view.CreateFragmentIfNotExists(shard)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating fragment")
|
|
}
|
|
if err := frag.importRoaring(ctx, tx, data, clear); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (f *Field) GetIndex() *Index {
|
|
return f.idx
|
|
}
|
|
|
|
func (f *Field) importRoaringOverwrite(ctx context.Context, tx Tx, data []byte, shard uint64, viewName string, block int) error {
|
|
span, ctx := tracing.StartSpanFromContext(ctx, "Field.importRoaringOverwrite")
|
|
defer span.Finish()
|
|
|
|
if viewName == "" {
|
|
viewName = viewStandard
|
|
}
|
|
span.LogKV("view", viewName, "bytes", len(data), "shard", shard)
|
|
view, err := f.createViewIfNotExists(viewName)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating view")
|
|
}
|
|
|
|
frag, err := view.CreateFragmentIfNotExists(shard)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating fragment")
|
|
}
|
|
if err := frag.importRoaringOverwrite(ctx, tx, data, block); err != nil {
|
|
return err
|
|
}
|
|
|
|
// If field is int, decimal, or timestamp, then we need to update
|
|
// field.options.BitDepth and bsiGroup.BitDepth based on the imported data.
|
|
switch f.Options().Type {
|
|
case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp:
|
|
frag.mu.Lock()
|
|
maxRowID, _, err := frag.maxRow(tx, nil)
|
|
frag.mu.Unlock()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var bitDepth uint64
|
|
if maxRowID+1 > bsiOffsetBit {
|
|
bitDepth = uint64(maxRowID + 1 - bsiOffsetBit)
|
|
}
|
|
|
|
bsig := f.bsiGroup(f.name)
|
|
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if bitDepth > f.options.BitDepth {
|
|
f.options.BitDepth = bitDepth
|
|
}
|
|
if bsig != nil {
|
|
bsig.BitDepth = bitDepth
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
type fieldSlice []*Field
|
|
|
|
func (p fieldSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
|
|
func (p fieldSlice) Len() int { return len(p) }
|
|
func (p fieldSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() }
|
|
|
|
// FieldInfo represents schema information for a field.
|
|
type FieldInfo struct {
|
|
Name string `json:"name"`
|
|
CreatedAt int64 `json:"createdAt,omitempty"`
|
|
Owner string `json:"owner"`
|
|
Options FieldOptions `json:"options"`
|
|
Cardinality *uint64 `json:"cardinality,omitempty"`
|
|
Views []*ViewInfo `json:"views,omitempty"`
|
|
}
|
|
|
|
type fieldInfoSlice []*FieldInfo
|
|
|
|
func (p fieldInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
|
|
func (p fieldInfoSlice) Len() int { return len(p) }
|
|
func (p fieldInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name }
|
|
|
|
// FieldOptions represents options to set when initializing a field.
|
|
type FieldOptions struct {
|
|
Base int64 `json:"base,omitempty"`
|
|
BitDepth uint64 `json:"bitDepth,omitempty"`
|
|
Min pql.Decimal `json:"min,omitempty"`
|
|
Max pql.Decimal `json:"max,omitempty"`
|
|
Scale int64 `json:"scale,omitempty"`
|
|
Keys bool `json:"keys"`
|
|
NoStandardView bool `json:"noStandardView,omitempty"`
|
|
TrackExistence bool `json:"trackExistence,omitempty"`
|
|
CacheSize uint32 `json:"cacheSize,omitempty"`
|
|
CacheType string `json:"cacheType,omitempty"`
|
|
Type string `json:"type,omitempty"`
|
|
TimeUnit string `json:"timeUnit,omitempty"`
|
|
TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"`
|
|
ForeignIndex string `json:"foreignIndex"`
|
|
TTL time.Duration `json:"ttl,omitempty"`
|
|
}
|
|
|
|
// newFieldOptions returns a new instance of FieldOptions
|
|
// with applied and validated functional options.
|
|
func newFieldOptions(opts ...FieldOption) (*FieldOptions, error) {
|
|
fo := FieldOptions{}
|
|
for _, opt := range opts {
|
|
err := opt(&fo)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
if fo.Keys {
|
|
switch fo.Type {
|
|
case FieldTypeInt:
|
|
return nil, ErrIntFieldWithKeys
|
|
|
|
case FieldTypeDecimal:
|
|
return nil, ErrDecimalFieldWithKeys
|
|
|
|
case FieldTypeTimestamp:
|
|
return nil, ErrTimestampFieldWithKeys
|
|
}
|
|
}
|
|
|
|
return &fo, nil
|
|
}
|
|
|
|
// applyDefaultOptions updates FieldOptions with the default
|
|
// values if o does not contain a valid type.
|
|
func applyDefaultOptions(o *FieldOptions) FieldOptions {
|
|
if o == nil {
|
|
o = &FieldOptions{}
|
|
}
|
|
if o.Type == "" {
|
|
o.Type = DefaultFieldType
|
|
o.CacheType = DefaultCacheType
|
|
o.CacheSize = DefaultCacheSize
|
|
}
|
|
return *o
|
|
}
|
|
|
|
// ActuallyTrackingExistence reflects the distinction between the
|
|
// TrackExistence bool, which is enabled by default for most fields,
|
|
// and whether we actually do existence tracking. Specifically,
|
|
// we don't do existence tracking for time quantum fields which don't
|
|
// have a standard view, or for BSI fields.
|
|
func (o *FieldOptions) ActuallyTrackingExistence() bool {
|
|
switch o.Type {
|
|
case FieldTypeTime:
|
|
return o.TrackExistence && !o.NoStandardView
|
|
case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp:
|
|
return false
|
|
default:
|
|
return o.TrackExistence
|
|
}
|
|
}
|
|
|
|
// MarshalJSON marshals FieldOptions to JSON such that
|
|
// only those attributes associated to the field type
|
|
// are included.
|
|
func (o *FieldOptions) MarshalJSON() ([]byte, error) {
|
|
switch o.Type {
|
|
case FieldTypeSet, "":
|
|
return json.Marshal(struct {
|
|
Type string `json:"type"`
|
|
CacheType string `json:"cacheType"`
|
|
CacheSize uint32 `json:"cacheSize"`
|
|
Keys bool `json:"keys"`
|
|
}{
|
|
o.Type,
|
|
o.CacheType,
|
|
o.CacheSize,
|
|
o.Keys,
|
|
})
|
|
case FieldTypeInt:
|
|
return json.Marshal(struct {
|
|
Type string `json:"type"`
|
|
Base int64 `json:"base"`
|
|
BitDepth uint64 `json:"bitDepth"`
|
|
Min pql.Decimal `json:"min"`
|
|
Max pql.Decimal `json:"max"`
|
|
Keys bool `json:"keys"`
|
|
ForeignIndex string `json:"foreignIndex"`
|
|
}{
|
|
o.Type,
|
|
o.Base,
|
|
o.BitDepth,
|
|
o.Min,
|
|
o.Max,
|
|
o.Keys,
|
|
o.ForeignIndex,
|
|
})
|
|
case FieldTypeDecimal:
|
|
return json.Marshal(struct {
|
|
Type string `json:"type"`
|
|
Base int64 `json:"base"`
|
|
Scale int64 `json:"scale"`
|
|
BitDepth uint64 `json:"bitDepth"`
|
|
Min pql.Decimal `json:"min"`
|
|
Max pql.Decimal `json:"max"`
|
|
Keys bool `json:"keys"`
|
|
}{
|
|
o.Type,
|
|
o.Base,
|
|
o.Scale,
|
|
o.BitDepth,
|
|
o.Min,
|
|
o.Max,
|
|
o.Keys,
|
|
})
|
|
case FieldTypeTimestamp:
|
|
epoch, err := ValToTimestamp(o.TimeUnit, o.Base)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "translating val to timestamp")
|
|
}
|
|
|
|
return json.Marshal(struct {
|
|
Type string `json:"type"`
|
|
Epoch time.Time `json:"epoch"`
|
|
BitDepth uint64 `json:"bitDepth"`
|
|
Min pql.Decimal `json:"min"`
|
|
Max pql.Decimal `json:"max"`
|
|
TimeUnit string `json:"timeUnit"`
|
|
}{
|
|
o.Type,
|
|
epoch,
|
|
o.BitDepth,
|
|
o.Min,
|
|
o.Max,
|
|
o.TimeUnit,
|
|
})
|
|
case FieldTypeTime:
|
|
return json.Marshal(struct {
|
|
Type string `json:"type"`
|
|
TimeQuantum TimeQuantum `json:"timeQuantum"`
|
|
Keys bool `json:"keys"`
|
|
NoStandardView bool `json:"noStandardView"`
|
|
TTL time.Duration `json:"ttl"`
|
|
}{
|
|
o.Type,
|
|
o.TimeQuantum,
|
|
o.Keys,
|
|
o.NoStandardView,
|
|
o.TTL,
|
|
})
|
|
case FieldTypeMutex:
|
|
return json.Marshal(struct {
|
|
Type string `json:"type"`
|
|
CacheType string `json:"cacheType"`
|
|
CacheSize uint32 `json:"cacheSize"`
|
|
Keys bool `json:"keys"`
|
|
}{
|
|
o.Type,
|
|
o.CacheType,
|
|
o.CacheSize,
|
|
o.Keys,
|
|
})
|
|
case FieldTypeBool:
|
|
return json.Marshal(struct {
|
|
Type string `json:"type"`
|
|
}{
|
|
o.Type,
|
|
})
|
|
}
|
|
return nil, errors.Errorf("invalid field type: '%s'", o.Type)
|
|
}
|
|
|
|
// List of bsiGroup types.
|
|
const (
|
|
bsiGroupTypeInt = "int"
|
|
)
|
|
|
|
func isValidBSIGroupType(v string) bool {
|
|
switch v {
|
|
case bsiGroupTypeInt:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// bsiBase is a helper function used to determine the default value
|
|
// for base. Because base is not exposed as a field option argument,
|
|
// it defaults to min, max, or 0 depending on the min/max range.
|
|
func bsiBase(min, max int64) int64 {
|
|
if min > 0 {
|
|
return min
|
|
} else if max < 0 {
|
|
return max
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// bsiGroup represents a group of range-encoded rows on a field.
|
|
type bsiGroup struct {
|
|
Name string `json:"name,omitempty"`
|
|
Type string `json:"type,omitempty"`
|
|
Min int64 `json:"min,omitempty"`
|
|
Max int64 `json:"max,omitempty"`
|
|
Base int64 `json:"base,omitempty"`
|
|
Scale int64 `json:"scale,omitempty"`
|
|
TimeUnit string `json:"timeUnit,omitempty"`
|
|
BitDepth uint64 `json:"bitDepth,omitempty"`
|
|
}
|
|
|
|
// baseValue adjusts the value to align with the range for Field for a certain
|
|
// operation type.
|
|
// Note: There is an edge case for GT and LT where this returns a baseValue
|
|
// that does not fully encompass the range.
|
|
// ex: Field.Min = 0, Field.Max = 1023
|
|
// baseValue(LT, 2000) returns 1023, which will perform "LT 1023" and effectively
|
|
// exclude any columns with value = 1023.
|
|
func (b *bsiGroup) baseValue(op pql.Token, value int64) (baseValue int64, outOfRange bool) {
|
|
min, max := b.bitDepthMin(), b.bitDepthMax()
|
|
|
|
if op == pql.GT || op == pql.GTE {
|
|
if value > max {
|
|
return baseValue, true
|
|
} else if value < min {
|
|
baseValue = int64(min - b.Base)
|
|
// Address edge case noted in comments above.
|
|
if op == pql.GT {
|
|
baseValue--
|
|
}
|
|
} else {
|
|
baseValue = int64(value - b.Base)
|
|
}
|
|
} else if op == pql.LT || op == pql.LTE {
|
|
if value < min {
|
|
return baseValue, true
|
|
} else if value > max {
|
|
baseValue = int64(max - b.Base)
|
|
// Address edge case noted in comments above.
|
|
if op == pql.LT {
|
|
baseValue++
|
|
}
|
|
} else {
|
|
baseValue = int64(value - b.Base)
|
|
}
|
|
} else if op == pql.EQ || op == pql.NEQ {
|
|
if value < min || value > max {
|
|
return baseValue, true
|
|
}
|
|
baseValue = int64(value - b.Base)
|
|
}
|
|
return baseValue, false
|
|
}
|
|
|
|
// baseValueBetween adjusts the min/max value to align with the range for Field.
|
|
func (b *bsiGroup) baseValueBetween(lo, hi int64) (baseValueLo, baseValueHi int64, outOfRange bool) {
|
|
min, max := b.bitDepthMin(), b.bitDepthMax()
|
|
if hi < min || lo > max || hi < lo {
|
|
return 0, 0, true
|
|
}
|
|
|
|
// Limit lo/hi to possible bit range.
|
|
if lo < min {
|
|
lo = min
|
|
}
|
|
if hi > max {
|
|
hi = max
|
|
}
|
|
return lo - b.Base, hi - b.Base, false
|
|
}
|
|
|
|
func (b *bsiGroup) validate() error {
|
|
if b.Name == "" {
|
|
return ErrBSIGroupNameRequired
|
|
} else if !isValidBSIGroupType(b.Type) {
|
|
return ErrInvalidBSIGroupType
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// bitDepthMin returns the minimum value possible for the current bit depth.
|
|
func (b *bsiGroup) bitDepthMin() int64 {
|
|
return b.Base - (1 << b.BitDepth) + 1
|
|
}
|
|
|
|
// bitDepthMax returns the maximum value possible for the current bit depth.
|
|
func (b *bsiGroup) bitDepthMax() int64 {
|
|
return b.Base + (1 << b.BitDepth) - 1
|
|
}
|
|
|
|
// Cache types.
|
|
const (
|
|
CacheTypeLRU = "lru"
|
|
CacheTypeRanked = "ranked"
|
|
CacheTypeNone = "none"
|
|
)
|
|
|
|
// isValidCacheType returns true if v is a valid cache type.
|
|
func isValidCacheType(v string) bool {
|
|
switch v {
|
|
case CacheTypeLRU, CacheTypeRanked, CacheTypeNone:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// bitDepth returns the number of bits required to store a value.
|
|
func bitDepth(v uint64) uint64 {
|
|
return uint64(bits.Len64(v))
|
|
}
|
|
|
|
// bitDepthInt64 returns the required bit depth for abs(v).
|
|
func bitDepthInt64(v int64) uint64 {
|
|
if v < 0 {
|
|
return bitDepth(uint64(-v))
|
|
}
|
|
return bitDepth(uint64(v))
|
|
}
|
|
|
|
// FormatQualifiedFieldName generates a qualified name for the field to be used with Tx operations.
|
|
func FormatQualifiedFieldName(index, field string) string {
|
|
return fmt.Sprintf("%s\x00%s\x00", index, field)
|
|
}
|
|
|
|
// persistView stores the view information in etcd.
|
|
func (f *Field) persistView(ctx context.Context, cvm *CreateViewMessage) error {
|
|
if cvm.Index == "" {
|
|
return ErrIndexRequired
|
|
} else if cvm.Field == "" {
|
|
return ErrFieldRequired
|
|
} else if cvm.View == "" {
|
|
return ErrViewRequired
|
|
}
|
|
|
|
return f.holder.Schemator.CreateView(ctx, cvm.Index, cvm.Field, cvm.View)
|
|
}
|
|
|
|
// Timestamp field ranges.
|
|
var (
|
|
DefaultEpoch = time.Unix(0, 0).UTC() // 1970-01-01T00:00:00Z
|
|
MinTimestampNano = time.Unix(-1<<32, 0).UTC() // 1833-11-24T17:31:44Z
|
|
MaxTimestampNano = time.Unix(1<<32, 0).UTC() // 2106-02-07T06:28:16Z
|
|
MinTimestamp = time.Unix(-62135596799, 0).UTC() // 0001-01-01T00:00:01Z
|
|
MaxTimestamp = time.Unix(253402300799, 0).UTC() // 9999-12-31T23:59:59Z
|
|
)
|
|
|
|
// Constants related to timestamp.
|
|
const (
|
|
TimeUnitSeconds = "s"
|
|
TimeUnitMilliseconds = "ms"
|
|
TimeUnitMicroseconds = "µs"
|
|
TimeUnitUSeconds = "us"
|
|
TimeUnitNanoseconds = "ns"
|
|
)
|
|
|
|
// IsValidTimeUnit returns true if unit is valid.
|
|
func IsValidTimeUnit(unit string) bool {
|
|
switch unit {
|
|
case TimeUnitSeconds, TimeUnitMilliseconds, TimeUnitMicroseconds, TimeUnitUSeconds, TimeUnitNanoseconds:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// TimeUnitNanos returns the number of nanoseconds in unit.
|
|
func TimeUnitNanos(unit string) int64 {
|
|
switch unit {
|
|
case TimeUnitSeconds:
|
|
return int64(time.Second)
|
|
case TimeUnitMilliseconds:
|
|
return int64(time.Millisecond)
|
|
case TimeUnitMicroseconds, TimeUnitUSeconds:
|
|
return int64(time.Microsecond)
|
|
default:
|
|
return int64(time.Nanosecond)
|
|
}
|
|
}
|
|
|
|
// CheckEpochOutOfRange checks if the epoch is after max or before min
|
|
func CheckEpochOutOfRange(epoch, min, max time.Time) error {
|
|
if epoch.After(max) || epoch.Before(min) {
|
|
return errors.Errorf("custom epoch too far from Unix epoch: %s", epoch)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (f *Field) SortShardRow(tx Tx, shard uint64, filter *Row, sort_desc bool) (*SortedRow, error) {
|
|
bsig := f.bsiGroup(f.name)
|
|
if bsig == nil {
|
|
return nil, errors.New("bsig is nil")
|
|
}
|
|
|
|
view := f.view(viewBSIGroupPrefix + f.name)
|
|
if view == nil {
|
|
return nil, errors.New("view is nil")
|
|
}
|
|
|
|
fragment := view.Fragment(shard)
|
|
if fragment == nil {
|
|
return nil, errors.New("fragment is nil")
|
|
}
|
|
|
|
return fragment.sortBsiData(tx, filter, bsig.BitDepth, sort_desc)
|
|
}
|