datastore

Documentation

Index

• Constants
• Variables
• type Datastore
• type FieldLookupError
  • func NewFieldLookupError(message string) *FieldLookupError
  • func (self FieldLookupError) Error() string
• type PointIterator
  • func NewPointIterator(itrs []storage.Iterator, fields []*metastore.Field, ...) *PointIterator
  • func (pi *PointIterator) Close()
  • func (pi *PointIterator) Error() error
  • func (pi *PointIterator) Next()
  • func (pi *PointIterator) Point() *protocol.Point
  • func (pi *PointIterator) Valid() bool
• type PointIteratorStream
  • func (pis PointIteratorStream) Closed() bool
  • func (pis PointIteratorStream) HasPoint() bool
  • func (pis PointIteratorStream) Next() *protocol.Series
• type Shard
  • func NewShard(db storage.Engine, pointBatchSize, writeBatchSize int, ...) (*Shard, error)
  • func (self *Shard) DropFields(fields []*metastore.Field) error
  • func (self *Shard) IsClosed() bool
  • func (self *Shard) Query(querySpec *parser.QuerySpec, processor engine.Processor) error
  • func (self *Shard) Write(database string, series []*protocol.Series) error
• type ShardDatastore
  • func NewShardDatastore(config *configuration.Configuration, metaStore *metastore.Store) (*ShardDatastore, error)
  • func (self *ShardDatastore) BufferWrite(request *protocol.Request)
  • func (self *ShardDatastore) Close()
  • func (self *ShardDatastore) DeleteShard(shardId uint32) error
  • func (self *ShardDatastore) GetOrCreateShard(id uint32) (cluster.LocalShardDb, error)
  • func (self *ShardDatastore) ReturnShard(id uint32)
  • func (self *ShardDatastore) SetWriteBuffer(writeBuffer *cluster.WriteBuffer)
  • func (self *ShardDatastore) Write(request *protocol.Request) error

Constants

const (
	ONE_KILOBYTE                    = 1024
	ONE_MEGABYTE                    = 1024 * 1024
	SHARD_BLOOM_FILTER_BITS_PER_KEY = 10
	SHARD_DATABASE_DIR              = "shard_db_v2"
)

Variables

var (

	// This datastore implements the PersistentAtomicInteger interface. All of the persistent
	// integers start with this prefix, followed by their name
	ATOMIC_INCREMENT_PREFIX = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFD}
	// NEXT_ID_KEY holds the next id. ids are used to "intern" timeseries and column names
	NEXT_ID_KEY = []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
	// SERIES_COLUMN_INDEX_PREFIX is the prefix of the series to column names index
	SERIES_COLUMN_INDEX_PREFIX = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFE}
	// DATABASE_SERIES_INDEX_PREFIX is the prefix of the database to series names index
	DATABASE_SERIES_INDEX_PREFIX = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}
	MAX_SEQUENCE                 = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}

	TRUE = true
)

Types

type Datastore

type Datastore interface {
	ExecuteQuery(user common.User, database string,
		query *parser.SelectQuery, yield func(*protocol.Series) error,
		ringFilter func(database, series *string, time *int64) bool) error
	// Logs the request to a local store and assigns a sequence number that is unique per server id per day
	LogRequestAndAssignSequenceNumber(request *protocol.Request, replicationFactor *uint8, ownerServerId *uint32) error
	CurrentSequenceNumber(clusterVersion *uint32, replicationFactor *uint8, ownerServerId, originatingServerId *uint32) (uint64, error)
	// will replay all requests from a given number. If the number hasn't occured yet today, it replays from yesterday.
	// So this log replay is only meant to work for outages that last less than maybe 12 hours.
	ReplayRequestsFromSequenceNumber(*uint32, *uint32, *uint32, *uint8, *uint64, func(*[]byte) error) error
	// Increment the named integer by the given amount and return the new value
	AtomicIncrement(name string, val int) (uint64, error)
	WriteSeriesData(database string, series *protocol.Series) error
	DeleteSeriesData(database string, query *parser.DeleteQuery) error
	GetSeriesForDatabase(database string, yield func(string) error) error
	DropDatabase(database string) error
	DropSeries(database, series string) error
	Close()
}

type FieldLookupError

type FieldLookupError struct {
	// contains filtered or unexported fields
}

func NewFieldLookupError

func NewFieldLookupError(message string) *FieldLookupError

func (FieldLookupError) Error

func (self FieldLookupError) Error() string

type PointIterator

type PointIterator struct {
	// contains filtered or unexported fields
}

PointIterator takes a slice of iterators and their corresponding fields and turn it into a point iterator, i.e. an iterator that yields whole points instead of column values.

func NewPointIterator

func NewPointIterator(itrs []storage.Iterator, fields []*metastore.Field, startTime, endTime time.Time, asc bool) *PointIterator

Creates a new point iterator using the given column iterator, metadata columns, start and end time as well as the ascending flag. The iterator returned is already placed at the first point, there's no need to call Next() after the call to NewPointIterator, but the user should check Valid() to make sure the iterator is pointing at a valid point.

func (*PointIterator) Close

func (pi *PointIterator) Close()

Close the iterator and free any resources used by the iterator. Behavior of the iterator is undefined if the iterator is used after it was closed.

func (*PointIterator) Error

func (pi *PointIterator) Error() error

Returns an error if the iterator became invalid due to an error as opposed to reaching the end time.

func (*PointIterator) Next

func (pi *PointIterator) Next()

Advance the iterator to the next point

func (*PointIterator) Point

func (pi *PointIterator) Point() *protocol.Point

Returns the point that the iterator is pointing to.

func (*PointIterator) Valid

func (pi *PointIterator) Valid() bool

Returns true if the iterator is pointing at a valid location. Behavior of Point() is undefined if Valid() is false.

type PointIteratorStream

type PointIteratorStream struct {
	// contains filtered or unexported fields
}

PointIteratorStream is a struct that implements the StreamQuery interface and is used by the shard with the Merger to merge the data points locally to form a monotic stream of points (increasing or decreasing timestamps)

func (PointIteratorStream) Closed

func (pis PointIteratorStream) Closed() bool

Returns true if the point iterator is not valid

func (PointIteratorStream) HasPoint

func (pis PointIteratorStream) HasPoint() bool

Returns true if the point iterator is still valid

func (PointIteratorStream) Next

func (pis PointIteratorStream) Next() *protocol.Series

Returns the next point from the point iterator

type Shard

type Shard struct {
	// contains filtered or unexported fields
}

func NewShard

func NewShard(db storage.Engine, pointBatchSize, writeBatchSize int, metaStore *metastore.Store) (*Shard, error)

func (*Shard) DropFields

func (self *Shard) DropFields(fields []*metastore.Field) error

func (*Shard) IsClosed

func (self *Shard) IsClosed() bool

func (*Shard) Query

func (self *Shard) Query(querySpec *parser.QuerySpec, processor engine.Processor) error

func (*Shard) Write

func (self *Shard) Write(database string, series []*protocol.Series) error

type ShardDatastore

type ShardDatastore struct {
	// contains filtered or unexported fields
}

func NewShardDatastore

func NewShardDatastore(config *configuration.Configuration, metaStore *metastore.Store) (*ShardDatastore, error)

func (*ShardDatastore) BufferWrite

func (self *ShardDatastore) BufferWrite(request *protocol.Request)

func (*ShardDatastore) Close

func (self *ShardDatastore) Close()

func (*ShardDatastore) DeleteShard

func (self *ShardDatastore) DeleteShard(shardId uint32) error

func (*ShardDatastore) GetOrCreateShard

func (self *ShardDatastore) GetOrCreateShard(id uint32) (cluster.LocalShardDb, error)

func (*ShardDatastore) ReturnShard

func (self *ShardDatastore) ReturnShard(id uint32)

func (*ShardDatastore) SetWriteBuffer

func (self *ShardDatastore) SetWriteBuffer(writeBuffer *cluster.WriteBuffer)

func (*ShardDatastore) Write

func (self *ShardDatastore) Write(request *protocol.Request) error