// 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). // // Note that the ValCount returned has bsig.Base included, or if // you specify a nil bsig, includes the field's bsig.Base. Which is // to say, don't use this if you have a value that's already been // adjusted by base. 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) }