diff --git a/api.go b/api.go index ef2d7bba2..0cec23368 100644 --- a/api.go +++ b/api.go @@ -3427,39 +3427,11 @@ type CreateFieldObj struct { Options []FieldOption } -// ComputeAPI is a subset of the API methods which have to do with compute -// operations such as import. -type ComputeAPI interface { - Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts ...ImportOption) error - ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, opts ...ImportOption) error - Txf() *TxFactory -} - // QueryAPI is a subset of the API methods which have to do with query. type QueryAPI interface { Query(ctx context.Context, req *QueryRequest) (QueryResponse, error) } -// Ensure type implements interface. -var _ ComputeAPI = (*NopComputeAPI)(nil) - -// NopComputeAPI is a no-op implementation of the ComputeAPI interface. -type NopComputeAPI struct{} - -func NewNopComputeAPI() *NopComputeAPI { - return &NopComputeAPI{} -} - -func (c *NopComputeAPI) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts ...ImportOption) error { - return nil -} - -func (c *NopComputeAPI) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, opts ...ImportOption) error { - return nil -} - -func (c *NopComputeAPI) Txf() *TxFactory { return nil } - // Ensure type implements interface. var _ SchemaAPI = (*FeatureBaseSchemaAPI)(nil) diff --git a/dax/computer/alpha/snapshot.go b/dax/computer/alpha/snapshot.go deleted file mode 100644 index 919dbf2ca..000000000 --- a/dax/computer/alpha/snapshot.go +++ /dev/null @@ -1,103 +0,0 @@ -// Package alpha contains an implementation of the SnapshotReadWriter interface. -// In the case where a sub-service (such as snapshotter) implements these -// interfaces directly with both its service and its http client, then we don't -// need this middle implementation layer. But in this case, the Snapshotter -// operates as a third-party service might, meaning its API methods don't align -// with what FeatureBase needs to call. So this implementation acts as a -// translation later between the featurebase-to-snapshotter interface, and the -// third-party Snapshotter service. -package alpha - -import ( - "context" - "io" - - "github.com/molecula/featurebase/v3/dax" - "github.com/molecula/featurebase/v3/dax/computer" - "github.com/molecula/featurebase/v3/errors" -) - -// Ensure type implements interface. -var _ computer.SnapshotReadWriter = &alphaSnapshot{} - -// alphaSnapshot uses a Snapshotter implementation (which could be, for -// example, an http client or a locally running sub-service) to store its -// snapshots. -type alphaSnapshot struct { - ss computer.Snapshotter -} - -func NewAlphaSnapshot(sser computer.Snapshotter) *alphaSnapshot { - return &alphaSnapshot{ - ss: sser, - } -} - -func (s *alphaSnapshot) WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error { - bucket := partitionBucket(qtid.Key(), partition) - key := shardKey(shard) - - if err := s.ss.Write(bucket, key, version, rc); err != nil { - return errors.Wrapf(err, "writing shard data: %s, %d", key, version) - } - - return nil -} - -func (s *alphaSnapshot) ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) { - bucket := partitionBucket(qtid.Key(), partition) - key := shardKey(shard) - - rc, err := s.ss.Read(bucket, key, version) - if err != nil { - return nil, errors.Wrapf(err, "reading shard data: %s, %s, %d", bucket, key, version) - } - - return rc, nil -} - -func (s *alphaSnapshot) WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error { - bucket := partitionBucket(qtid.Key(), partition) - key := keysFileName - - if err := s.ss.WriteTo(bucket, key, version, wrTo); err != nil { - return errors.Wrapf(err, "writing table keys: %s, %d", key, version) - } - - return nil -} - -func (s *alphaSnapshot) ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) { - bucket := partitionBucket(qtid.Key(), partition) - key := keysFileName - - rc, err := s.ss.Read(bucket, key, version) - if err != nil { - return nil, errors.Wrapf(err, "reading table keys: %s, %s, %d", bucket, key, version) - } - - return rc, nil -} - -func (s *alphaSnapshot) WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error { - bucket := fieldBucket(qtid.Key(), field) - key := keysFileName - - if err := s.ss.WriteTo(bucket, key, version, wrTo); err != nil { - return errors.Wrapf(err, "writing field keys: %s, %d", key, version) - } - - return nil -} - -func (s *alphaSnapshot) ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) { - bucket := fieldBucket(qtid.Key(), field) - key := keysFileName - - rc, err := s.ss.Read(bucket, key, version) - if err != nil { - return nil, errors.Wrapf(err, "reading field keys: %s, %s, %d", bucket, key, version) - } - - return rc, nil -} diff --git a/dax/computer/alpha/writelog.go b/dax/computer/alpha/writelog.go deleted file mode 100644 index 24df9231d..000000000 --- a/dax/computer/alpha/writelog.go +++ /dev/null @@ -1,322 +0,0 @@ -// Package alpha contains an implementation of the WriteLogReader and -// WriteLogWriter interfaces. In the case where a sub-service (such as -// writelogger) implements these interfaces directly with both its service and -// its http client, then we don't need this middle implementation layer. But in -// this case, the WriteLogger operates as a third-party service might, meaning -// its API methods don't align with what FeatureBase needs to call. So this -// implementation acts as a translation later between the -// featurebase-to-writelogger interface, and the third-party WriteLogger -// service. -package alpha - -import ( - "bufio" - "context" - "encoding/json" - "io" - - "github.com/molecula/featurebase/v3/dax" - "github.com/molecula/featurebase/v3/dax/computer" - "github.com/molecula/featurebase/v3/errors" -) - -// Ensure type implements interface. -var _ computer.WriteLogReader = &alphaWriteLog{} -var _ computer.WriteLogWriter = &alphaWriteLog{} - -// alphaWriteLog is an implementation of the WriteLogReader and WriteLogWriter -// interfaces. It uses a WriteLogger implementation (which could be, for -// example, an http client or a locally running sub-service) to store its log -// messages. I can't remember why we put this implementation in its own package -// (a meaningless name called "alpha"); it probably has something to do with -// wanting to adhere to a typical interface/implementation package structure, -// but was confused by the current state of things where the "storage" package -// (or perhaps "computer") is the top level package (i.e. "pilosa"). Until we -// can correct the packaging, we will likely have weird cases like this. -type alphaWriteLog struct { - wl computer.WriteLogger -} - -func NewAlphaWriteLog(wler computer.WriteLogger) *alphaWriteLog { - return &alphaWriteLog{ - wl: wler, - } -} - -func (w *alphaWriteLog) CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, m map[string]uint64) error { - msg := computer.PartitionKeyMap{ - TableKey: qtid.Key(), - Partition: partition, - StringToID: m, - } - - b, err := json.Marshal(msg) - if err != nil { - return errors.Wrap(err, "marshalling partition key map to json") - } - - bucket := partitionBucket(qtid.Key(), partition) - - if err := w.wl.AppendMessage(bucket, keysFileName, version, b); err != nil { - return errors.Wrapf(err, "appending partition key message: %s, %d", keysFileName, version) - } - - return nil -} - -func (w *alphaWriteLog) DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error { - bucket := partitionBucket(qtid.Key(), partition) - return w.wl.DeleteLog(bucket, keysFileName, version) -} - -func (w *alphaWriteLog) CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, m map[string]uint64) error { - msg := computer.FieldKeyMap{ - TableKey: qtid.Key(), - Field: field, - StringToID: m, - } - - b, err := json.Marshal(msg) - if err != nil { - return errors.Wrap(err, "marshalling field key map to json") - } - - bucket := fieldBucket(qtid.Key(), field) - - if err := w.wl.AppendMessage(bucket, keysFileName, version, b); err != nil { - return errors.Wrapf(err, "appending field key message: %s, %d", keysFileName, version) - } - - return nil -} - -func (w *alphaWriteLog) DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error { - bucket := fieldBucket(qtid.Key(), field) - return w.wl.DeleteLog(bucket, keysFileName, version) -} - -func (w *alphaWriteLog) WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg computer.LogMessage) error { - b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON) - if err != nil { - return errors.Wrap(err, "marshalling log message") - } - - bucket := partitionBucket(qtid.Key(), partition) - shardKey := shardKey(shard) - - if err := w.wl.AppendMessage(bucket, shardKey, version, b); err != nil { - return errors.Wrapf(err, "appending shard key message: %s, %d", shardKey, version) - } - - return nil -} - -func (w *alphaWriteLog) DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error { - bucket := partitionBucket(qtid.Key(), partition) - shardKey := shardKey(shard) - - return w.wl.DeleteLog(bucket, shardKey, version) -} - -//////////////////////////////////////////////// - -func (w *alphaWriteLog) TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) computer.TableKeyReader { - return newTableKeyReader(w.wl, qtid, partition, version) -} - -type tableKeyReader struct { - wl computer.WriteLogger - table dax.TableKey - partition dax.PartitionNum - version int - scanner *bufio.Scanner - closer io.Closer -} - -func newTableKeyReader(wl computer.WriteLogger, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) *tableKeyReader { - r := &tableKeyReader{ - wl: wl, - table: qtid.Key(), - partition: partition, - version: version, - } - - return r -} - -func (r *tableKeyReader) Open() error { - bucket := partitionBucket(r.table, r.partition) - - reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version) - if err != nil { - return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version) - } - - r.closer = closer - r.scanner = bufio.NewScanner(reader) - - return nil -} - -func (r *tableKeyReader) Read() (computer.PartitionKeyMap, error) { - if r.scanner == nil { - return computer.PartitionKeyMap{}, io.EOF - } - - var b []byte - var out computer.PartitionKeyMap - - if r.scanner.Scan() { - b = r.scanner.Bytes() - if err := json.Unmarshal(b, &out); err != nil { - return out, err - } - return out, nil - } - if err := r.scanner.Err(); err != nil { - return out, err - } - - return out, io.EOF -} - -func (r *tableKeyReader) Close() error { - if r.closer != nil { - return r.closer.Close() - } - return nil -} - -//////////////////////////////////////////////// - -func (w *alphaWriteLog) FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) computer.FieldKeyReader { - return newFieldKeyReader(w.wl, qtid, field, version) -} - -type fieldKeyReader struct { - wl computer.WriteLogger - table dax.TableKey - field dax.FieldName - version int - scanner *bufio.Scanner - closer io.Closer -} - -func newFieldKeyReader(wl computer.WriteLogger, qtid dax.QualifiedTableID, field dax.FieldName, version int) *fieldKeyReader { - r := &fieldKeyReader{ - wl: wl, - table: qtid.Key(), - field: field, - version: version, - } - - return r -} - -func (r *fieldKeyReader) Open() error { - bucket := fieldBucket(r.table, r.field) - - reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version) - if err != nil { - return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version) - } - - r.closer = closer - r.scanner = bufio.NewScanner(reader) - - return nil -} - -func (r *fieldKeyReader) Read() (computer.FieldKeyMap, error) { - if r.scanner == nil { - return computer.FieldKeyMap{}, io.EOF - } - - var b []byte - var out computer.FieldKeyMap - - if r.scanner.Scan() { - b = r.scanner.Bytes() - if err := json.Unmarshal(b, &out); err != nil { - return out, err - } - return out, nil - } - if err := r.scanner.Err(); err != nil { - return out, err - } - - return out, io.EOF -} - -func (r *fieldKeyReader) Close() error { - if r.closer != nil { - return r.closer.Close() - } - return nil -} - -//////////////////////////////////////////////// - -func (w *alphaWriteLog) ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) computer.ShardReader { - return newShardReader(w.wl, qtid, partition, shard, version) -} - -type shardReader struct { - wl computer.WriteLogger - table dax.TableKey - partition dax.PartitionNum - shard dax.ShardNum - version int - scanner *bufio.Scanner - closer io.Closer -} - -func newShardReader(wl computer.WriteLogger, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) *shardReader { - r := &shardReader{ - wl: wl, - table: qtid.Key(), - partition: partition, - shard: shard, - version: version, - } - - return r -} - -func (r *shardReader) Open() error { - bucket := partitionBucket(r.table, r.partition) - shardKey := shardKey(r.shard) - - reader, closer, err := r.wl.LogReader(bucket, shardKey, r.version) - if err != nil { - return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, shardKey, r.version) - } - - r.closer = closer - r.scanner = bufio.NewScanner(reader) - - return nil -} - -func (r *shardReader) Read() (computer.LogMessage, error) { - if r.scanner == nil { - return nil, io.EOF - } - - if r.scanner.Scan() { - return computer.UnmarshalLogMessage(r.scanner.Bytes()) - } - if err := r.scanner.Err(); err != nil { - return nil, err - } - - return nil, io.EOF -} - -func (r *shardReader) Close() error { - if r.closer != nil { - return r.closer.Close() - } - return nil -} diff --git a/dax/computer/interfaces.go b/dax/computer/interfaces.go index e1554fc2d..1ef6135e3 100644 --- a/dax/computer/interfaces.go +++ b/dax/computer/interfaces.go @@ -14,18 +14,86 @@ type Registrar interface { CheckInNode(ctx context.Context, node *dax.Node) error } -// WriteLogger represents the WriteLogger methods which Computer uses. These are -// typically implemented by the WriteLogger client. -type WriteLogger interface { +// WriteLogService represents the WriteLogService methods which Computer uses. +// These are typically implemented by the WriteLogger client. +type WriteLogService interface { AppendMessage(bucket string, key string, version int, msg []byte) error LogReader(bucket string, key string, version int) (io.Reader, io.Closer, error) DeleteLog(bucket string, key string, version int) error } -// Snapshotter represents the Snapshotter methods which Computer uses. These are -// typically implemented by both the Snapshotter client. -type Snapshotter interface { +// SnapshotService represents the SnapshotService methods which Computer uses. +// These are typically implemented by both the Snapshotter client. +type SnapshotService interface { Read(bucket string, key string, version int) (io.ReadCloser, error) Write(bucket string, key string, version int, rc io.ReadCloser) error WriteTo(bucket string, key string, version int, wrTo io.WriterTo) error } + +// SnapshotReadWriter provides the interface for all snapshot read and writes in +// FeatureBase. +type SnapshotReadWriter interface { + WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error + ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) + + WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error + ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) + + WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error + ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) +} + +// WriteLogWriter provides the interface for all data writes to FeatureBase. After +// data has been written to the local FeatureBase node, the respective interface +// method(s) will be called. +type WriteLogWriter interface { + // CreateTableKeys sends a map of string key to uint64 ID for the table and + // partition provided. + CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, _ map[string]uint64) error + + // DeleteTableKeys deletes all table keys for the table and partition + // provided. + DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error + + // CreateFieldKeys sends a map of string key to uint64 ID for the table and + // field provided. + CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, _ map[string]uint64) error + + // DeleteTableKeys deletes all field keys for the table and field provided. + DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error + + // WriteShard sends shard data for the table and shard provided. + WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error + + // DeleteShard deletes all data for the table and shard provided. + DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error +} + +// WriteLogReader provides the interface for all reads from the write log. +type WriteLogReader interface { + ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader + TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader + FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader +} + +type TableKeyReader interface { + Open() error + Read() (PartitionKeyMap, error) + Close() error +} + +type FieldKeyReader interface { + Open() error + Read() (FieldKeyMap, error) + Close() error +} + +type ShardReader interface { + Open() error + Read() (LogMessage, error) + Close() error +} + +// LogMessage is implemented by a variety of types which can be serialized as +// messages to the WriteLogger. +type LogMessage interface{} diff --git a/dax/computer/alpha/alpha.go b/dax/computer/keys.go similarity index 96% rename from dax/computer/alpha/alpha.go rename to dax/computer/keys.go index c8f8bfd1a..185c70ee5 100644 --- a/dax/computer/alpha/alpha.go +++ b/dax/computer/keys.go @@ -1,4 +1,4 @@ -package alpha +package computer import ( "fmt" diff --git a/dax/computer/logmessage.go b/dax/computer/logmessage.go new file mode 100644 index 000000000..74d33031f --- /dev/null +++ b/dax/computer/logmessage.go @@ -0,0 +1,265 @@ +package computer + +import ( + "encoding/json" + "time" + + "github.com/molecula/featurebase/v3/dax" + "github.com/molecula/featurebase/v3/errors" +) + +//////////////// Messages /////////////////////// + +type PartitionKeyMap struct { + TableKey dax.TableKey `json:"table-key"` + Partition dax.PartitionNum `json:"partition"` + StringToID map[string]uint64 `json:"string-to-id"` +} + +type FieldKeyMap struct { + TableKey dax.TableKey `json:"table-key"` + Field dax.FieldName `json:"field"` + StringToID map[string]uint64 `json:"string-to-id"` +} + +const ( + logMessageTypeImportRoaring byte = iota + logMessageTypeImport + logMessageTypeImportValue + logMessageTypeImportRoaringShard +) + +// encoderKey* are part of the log message header. They indicate what encoding +// type a specific log message is serialized with. +const ( + encoderKeyJSON byte = iota +) + +const ( + EncodeTypeJSON string = "json" + + // encodeVersion refers to the version of the structs used to represent the + // log messages. If we change structs, we'll need to modify this version + // number and maintain the previous version of the structs somewhere for + // deserialization. + encodeVersion byte = 1 +) + +// logMessageEncoder is implemented by any encoder used to serialize LogMessages +// to []byte. +type logMessageEncoder interface { + Key() byte + Marshal(LogMessage) ([]byte, error) + Unmarshal([]byte, LogMessage) error +} + +// MarshalLogMessage serializes the log message and prepends additional encoding +// information to each message. Currently, we prepend three bytes to each log +// message: +// byte[0]: encodeVersion - this is currently a constant within the code. If we +// modify structs such that they encode differently, we'll have to change the +// constant and keep previous versions of structs for deserialization. +// byte[1]: encodeType (e.g. "json", etc.) +// byte[2]: logMessageType +// +// If we get into a situation where we want more flexibility in these message +// header bytes—for example, if we want to use more than three bytes—we could do +// something with the first bit of the encodeVersion: if it's 1, that could +// indicate that there are additional header bytes, and the following seven bits +// could indicate how many. +func MarshalLogMessage(msg LogMessage, encode string) ([]byte, error) { + encoder, err := getEncoderByType(encode) + if err != nil { + return nil, errors.Wrap(err, "getting encoder by type") + } + + logMessageType, err := getLogMessageType(msg) + if err != nil { + return nil, errors.Wrap(err, "getting log message type") + } + + var buf []byte + + buf, err = encoder.Marshal(msg) + if err != nil { + return nil, errors.Wrap(err, "marshaling log message") + } + + return append([]byte{encodeVersion, encoder.Key(), logMessageType}, buf...), nil +} + +// UnmarshalLogMessage deserializes the log message based on the log message +// type info. +func UnmarshalLogMessage(b []byte) (LogMessage, error) { + if len(b) < 3 { + return nil, errors.New(errors.ErrUncoded, "log record does not contain a full header") + } + + encVersion := b[0] + encKey := b[1] + logMessageType := b[2] + + // Ensure that the log message is able to be handled by this code. If we + // increment the constant encodeVersion, we'll need to modify this to handle + // the log based on previous encodeVersions. + if encVersion != encodeVersion { + return nil, errors.Errorf("encode version is unsupported: %d", encVersion) + } + + msg, err := logMessageByType(logMessageType) + if err != nil { + return nil, errors.Wrap(err, "getting log message by type") + } + + encoder, err := getEncoderByKey(encKey) + if err != nil { + return nil, errors.Wrap(err, "getting encoder by key") + } + + if err := encoder.Unmarshal(b[3:], &msg); err != nil { + return nil, errors.Wrap(err, "unmarshaling log message") + } + + return msg, nil +} + +func logMessageByType(typ byte) (LogMessage, error) { + switch typ { + case logMessageTypeImportRoaring: + return &ImportRoaringMessage{}, nil + case logMessageTypeImport: + return &ImportMessage{}, nil + case logMessageTypeImportValue: + return &ImportValueMessage{}, nil + case logMessageTypeImportRoaringShard: + return &ImportRoaringShardMessage{}, nil + default: + return nil, errors.Errorf("unknown message type %d", typ) + } +} + +func getLogMessageType(m LogMessage) (byte, error) { + switch m.(type) { + case *ImportRoaringMessage: + return logMessageTypeImportRoaring, nil + case *ImportMessage: + return logMessageTypeImport, nil + case *ImportValueMessage: + return logMessageTypeImportValue, nil + case *ImportRoaringShardMessage: + return logMessageTypeImportRoaringShard, nil + default: + return 0, errors.Errorf("don't have type for message %#v", m) + } +} + +func getEncoderByType(encode string) (logMessageEncoder, error) { + switch encode { + case EncodeTypeJSON: + return &encoderJSON{}, nil + default: + return nil, errors.Errorf("invalid encode type: %s", encode) + } +} + +func getEncoderByKey(id byte) (logMessageEncoder, error) { + switch id { + case encoderKeyJSON: + return &encoderJSON{}, nil + default: + return nil, errors.Errorf("invalid encode type: %d", id) + } +} + +type ImportRoaringMessage struct { + LogMessage `json:"-"` + + Table string `json:"table"` + Field string `json:"field"` + Partition int `json:"partition"` + Shard uint64 `json:"shard"` + Clear bool `json:"clear"` + Action string `json:"action"` // [set, clear, overwrite] + Block int `json:"block"` + Views map[string][]byte `json:"views"` + UpdateExistence bool `json:"update-existence"` +} + +type ImportMessage struct { + LogMessage `json:"-"` + + Table string `json:"table"` + Field string `json:"field"` + Partition int `json:"partition"` + Shard uint64 `json:"shard"` + RowIDs []uint64 `json:"row-ids"` + ColumnIDs []uint64 `json:"column-ids"` + RowKeys []string `json:"row-keys"` + ColumnKeys []string `json:"column-keys"` + Timestamps []int64 `json:"timestamps"` + Clear bool `json:"clear"` + + // options + IgnoreKeyCheck bool `json:"ignore-key-check"` + Presorted bool `json:"presorted"` +} + +type ImportValueMessage struct { + LogMessage `json:"-"` + + Table string `json:"table"` + Field string `json:"field"` + Partition int `json:"partition"` + Shard uint64 `json:"shard"` + ColumnIDs []uint64 `json:"column-ids"` + ColumnKeys []string `json:"column-keys"` + Values []int64 `json:"values"` + FloatValues []float64 `json:"float-values"` + TimestampValues []time.Time `json:"timestamp-values"` + StringValues []string `json:"string-values"` + Clear bool `json:"clear"` + + // options + IgnoreKeyCheck bool `json:"ignore-key-check"` + Presorted bool `json:"presorted"` +} + +type ImportRoaringShardMessage struct { + LogMessage `json:"-"` + + Table string `json:"table"` + Partition int `json:"partition"` + Shard uint64 `json:"shard"` + Views []RoaringUpdate `json:"views"` +} + +// RoaringUpdate is identical to featurebase.RoaringUpdate, but we +// can't import it due to import cycles. TODO featurebase top level +// shouldn't import dax stuff... all the types it needs should just be +// in the top level. +type RoaringUpdate struct { + Field string `json:"field"` + View string `json:"view"` + Clear []byte `json:"clear"` + Set []byte `json:"set"` + ClearRecords bool `json:"clear-records"` +} + +// Ensure type implements interface. +var _ logMessageEncoder = (*encoderJSON)(nil) + +// encoderJSON is an implementation of the logMessageEncoder interface which +// encodes LogMessages as JSON. +type encoderJSON struct{} + +func (e *encoderJSON) Key() byte { + return encoderKeyJSON +} + +func (e *encoderJSON) Marshal(msg LogMessage) ([]byte, error) { + return json.Marshal(msg) +} + +func (e *encoderJSON) Unmarshal(b []byte, msg LogMessage) error { + return json.Unmarshal(b, msg) +} diff --git a/dax/computer/writelog_test.go b/dax/computer/logmessage_test.go similarity index 100% rename from dax/computer/writelog_test.go rename to dax/computer/logmessage_test.go diff --git a/dax/computer/noop.go b/dax/computer/noop.go new file mode 100644 index 000000000..01e6cc41c --- /dev/null +++ b/dax/computer/noop.go @@ -0,0 +1,162 @@ +package computer + +import ( + "context" + "io" + + "github.com/molecula/featurebase/v3/dax" +) + +// Ensure type implements interface. +var _ WriteLogWriter = (*NopWriteLogWriter)(nil) + +// NopWriteLogWriter is a no-op implementation of the WriteLogWriter interface. +type NopWriteLogWriter struct{} + +func NewNopWriteLogWriter() *NopWriteLogWriter { + return &NopWriteLogWriter{} +} + +func (w *NopWriteLogWriter) CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, m map[string]uint64) error { + return nil +} + +func (w *NopWriteLogWriter) DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error { + return nil +} + +func (w *NopWriteLogWriter) CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, m map[string]uint64) error { + return nil +} + +func (w *NopWriteLogWriter) DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error { + return nil +} + +func (w *NopWriteLogWriter) WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error { + return nil +} + +func (w *NopWriteLogWriter) DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error { + return nil +} + +// Ensure type implements interface. +var _ WriteLogReader = (*NopWriteLogReader)(nil) + +// NopWriteLogReader is a no-op implementation of the WriteLogReader interface. +type NopWriteLogReader struct{} + +func NewNopWriteLogReader() *NopWriteLogReader { + return &NopWriteLogReader{} +} + +func (w *NopWriteLogReader) TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader { + return NewNopTableKeyReader() +} + +func (w *NopWriteLogReader) FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader { + return NewNopFieldKeyReader() +} + +func (w *NopWriteLogReader) ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader { + return NewNopShardReader() +} + +//////////////////////////////////////////////// + +// Ensure type implements interface. +var _ TableKeyReader = &NopTableKeyReader{} + +// NopTableKeyReader is a no-op implementation of the TableKeyReader +// interface. +type NopTableKeyReader struct{} + +func NewNopTableKeyReader() *NopTableKeyReader { + return &NopTableKeyReader{} +} + +func (r *NopTableKeyReader) Open() error { return nil } +func (r *NopTableKeyReader) Read() (PartitionKeyMap, error) { + return PartitionKeyMap{}, io.EOF +} +func (r *NopTableKeyReader) Close() error { return nil } + +//////////////////////////////////////////////// + +// Ensure type implements interface. +var _ FieldKeyReader = &NopFieldKeyReader{} + +// NopFieldKeyReader is a no-op implementation of the FieldKeyReader +// interface. +type NopFieldKeyReader struct{} + +func NewNopFieldKeyReader() *NopFieldKeyReader { + return &NopFieldKeyReader{} +} + +func (r *NopFieldKeyReader) Open() error { return nil } +func (r *NopFieldKeyReader) Read() (FieldKeyMap, error) { + return FieldKeyMap{}, io.EOF +} +func (r *NopFieldKeyReader) Close() error { return nil } + +//////////////////////////////////////////////// + +// Ensure type implements interface. +var _ ShardReader = &NopShardReader{} + +// NopShardReader is a no-op implementation of the ShardReader interface. +type NopShardReader struct{} + +func NewNopShardReader() *NopShardReader { + return &NopShardReader{} +} + +func (r *NopShardReader) Open() error { return nil } +func (r *NopShardReader) Read() (LogMessage, error) { + return nil, io.EOF +} +func (r *NopShardReader) Close() error { return nil } + +////////////// SNAPSHOT //////////////////////// + +// Ensure type implements interface. +var _ SnapshotReadWriter = &NopSnapshotReadWriter{} + +// NopSnapshotReadWriter is a no-op implementation of the SnapshotReadWriter +// interface. +type NopSnapshotReadWriter struct{} + +func NewNopSnapshotReadWriter() *NopSnapshotReadWriter { + return &NopSnapshotReadWriter{} +} + +func (w *NopSnapshotReadWriter) WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error { + return nil +} + +func (w *NopSnapshotReadWriter) ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) { + return &nopReadCloser{}, nil +} + +func (w *NopSnapshotReadWriter) WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error { + return nil +} + +func (w *NopSnapshotReadWriter) ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) { + return &nopReadCloser{}, nil +} + +func (w *NopSnapshotReadWriter) WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error { + return nil +} + +func (w *NopSnapshotReadWriter) ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) { + return &nopReadCloser{}, nil +} + +type nopReadCloser struct{} + +func (n *nopReadCloser) Read([]byte) (int, error) { return 0, nil } +func (n *nopReadCloser) Close() error { return nil } diff --git a/dax/computer/service/computer.go b/dax/computer/service/computer.go index ba2dcb419..80ee002f6 100644 --- a/dax/computer/service/computer.go +++ b/dax/computer/service/computer.go @@ -155,7 +155,7 @@ func newCommand(addr dax.Address, cfg CommandConfig) *fbserver.Command { cfg.ComputerConfig.GRPCListener = &nopListener{} cfg.ComputerConfig.DataDir = cfg.RootDataDir + "/" + cfg.Name - var writeLoggerImpl computer.WriteLogger + var writeLoggerImpl computer.WriteLogService if cfg.ComputerConfig.WriteLogger != "" { writeLoggerImpl = writeloggerclient.New(dax.Address(cfg.ComputerConfig.WriteLogger)) } else if wlSvc != nil { @@ -164,7 +164,7 @@ func newCommand(addr dax.Address, cfg CommandConfig) *fbserver.Command { cfg.Logger.Warnf("No writelogger configured, dynamic scaling will not function properly.") } - var snapshotterImpl computer.Snapshotter + var snapshotterImpl computer.SnapshotService if cfg.ComputerConfig.Snapshotter != "" { snapshotterImpl = snapshotterclient.New(dax.Address(cfg.ComputerConfig.Snapshotter)) } else if ssSvc != nil { diff --git a/dax/computer/snapshot.go b/dax/computer/snapshot.go index 3b19e93ed..ac30f8b72 100644 --- a/dax/computer/snapshot.go +++ b/dax/computer/snapshot.go @@ -5,57 +5,90 @@ import ( "io" "github.com/molecula/featurebase/v3/dax" + "github.com/molecula/featurebase/v3/errors" ) -// SnapshotReadWriter provides the interface for all snapshot read and writes in -// FeatureBase. -type SnapshotReadWriter interface { - WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error - ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) - - WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error - ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) - - WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error - ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) -} - // Ensure type implements interface. -var _ SnapshotReadWriter = &NopSnapshotReadWriter{} +var _ SnapshotReadWriter = &snapshotReadWriter{} -// NopSnapshotReadWriter is a no-op implementation of the SnapshotReadWriter -// interface. -type NopSnapshotReadWriter struct{} - -func NewNopSnapshotReadWriter() *NopSnapshotReadWriter { - return &NopSnapshotReadWriter{} +// snapshotReadWriter uses a SnapshotService implementation (which could be, for +// example, an http client or a locally running sub-service) to store its +// snapshots. +type snapshotReadWriter struct { + ss SnapshotService } -func (w *NopSnapshotReadWriter) WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error { +func NewSnapshotReadWriter(ss SnapshotService) *snapshotReadWriter { + return &snapshotReadWriter{ + ss: ss, + } +} + +func (s *snapshotReadWriter) WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error { + bucket := partitionBucket(qtid.Key(), partition) + key := shardKey(shard) + + if err := s.ss.Write(bucket, key, version, rc); err != nil { + return errors.Wrapf(err, "writing shard data: %s, %d", key, version) + } + return nil } -func (w *NopSnapshotReadWriter) ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) { - return &nopReadCloser{}, nil +func (s *snapshotReadWriter) ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) { + bucket := partitionBucket(qtid.Key(), partition) + key := shardKey(shard) + + rc, err := s.ss.Read(bucket, key, version) + if err != nil { + return nil, errors.Wrapf(err, "reading shard data: %s, %s, %d", bucket, key, version) + } + + return rc, nil } -func (w *NopSnapshotReadWriter) WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error { +func (s *snapshotReadWriter) WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error { + bucket := partitionBucket(qtid.Key(), partition) + key := keysFileName + + if err := s.ss.WriteTo(bucket, key, version, wrTo); err != nil { + return errors.Wrapf(err, "writing table keys: %s, %d", key, version) + } + return nil } -func (w *NopSnapshotReadWriter) ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) { - return &nopReadCloser{}, nil +func (s *snapshotReadWriter) ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) { + bucket := partitionBucket(qtid.Key(), partition) + key := keysFileName + + rc, err := s.ss.Read(bucket, key, version) + if err != nil { + return nil, errors.Wrapf(err, "reading table keys: %s, %s, %d", bucket, key, version) + } + + return rc, nil } -func (w *NopSnapshotReadWriter) WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error { +func (s *snapshotReadWriter) WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error { + bucket := fieldBucket(qtid.Key(), field) + key := keysFileName + + if err := s.ss.WriteTo(bucket, key, version, wrTo); err != nil { + return errors.Wrapf(err, "writing field keys: %s, %d", key, version) + } + return nil } -func (w *NopSnapshotReadWriter) ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) { - return &nopReadCloser{}, nil +func (s *snapshotReadWriter) ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) { + bucket := fieldBucket(qtid.Key(), field) + key := keysFileName + + rc, err := s.ss.Read(bucket, key, version) + if err != nil { + return nil, errors.Wrapf(err, "reading field keys: %s, %s, %d", bucket, key, version) + } + + return rc, nil } - -type nopReadCloser struct{} - -func (n *nopReadCloser) Read([]byte) (int, error) { return 0, nil } -func (n *nopReadCloser) Close() error { return nil } diff --git a/dax/computer/writelog.go b/dax/computer/writelog.go index d8694e115..4c023b67b 100644 --- a/dax/computer/writelog.go +++ b/dax/computer/writelog.go @@ -1,434 +1,307 @@ package computer import ( + "bufio" "context" "encoding/json" "io" - "time" "github.com/molecula/featurebase/v3/dax" "github.com/molecula/featurebase/v3/errors" ) -// WriteLogWriter provides the interface for all data writes to FeatureBase. After -// data has been written to the local FeatureBase node, the respective interface -// method(s) will be called. -type WriteLogWriter interface { - // CreateTableKeys sends a map of string key to uint64 ID for the table and - // partition provided. - CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, _ map[string]uint64) error - - // DeleteTableKeys deletes all table keys for the table and partition - // provided. - DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error - - // CreateFieldKeys sends a map of string key to uint64 ID for the table and - // field provided. - CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, _ map[string]uint64) error - - // DeleteTableKeys deletes all field keys for the table and field provided. - DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error - - // WriteShard sends shard data for the table and shard provided. - WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error - - // DeleteShard deletes all data for the table and shard provided. - DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error -} - // Ensure type implements interface. -var _ WriteLogWriter = (*NopWriteLogWriter)(nil) +var _ WriteLogReader = &writeLogReadWriter{} +var _ WriteLogWriter = &writeLogReadWriter{} -// NopWriteLogWriter is a no-op implementation of the WriteLogWriter interface. -type NopWriteLogWriter struct{} - -func NewNopWriteLogWriter() *NopWriteLogWriter { - return &NopWriteLogWriter{} +// writeLogReadWriter is an implementation of the WriteLogReader and WriteLogWriter +// interfaces. It uses a WriteLogService implementation (which could be, for +// example, an http client or a locally running sub-service) to store its log +// messages. +type writeLogReadWriter struct { + wls WriteLogService } -func (w *NopWriteLogWriter) CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, m map[string]uint64) error { +func NewWriteLogReadWriter(wls WriteLogService) *writeLogReadWriter { + return &writeLogReadWriter{ + wls: wls, + } +} + +func (w *writeLogReadWriter) CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, m map[string]uint64) error { + msg := PartitionKeyMap{ + TableKey: qtid.Key(), + Partition: partition, + StringToID: m, + } + + b, err := json.Marshal(msg) + if err != nil { + return errors.Wrap(err, "marshalling partition key map to json") + } + + bucket := partitionBucket(qtid.Key(), partition) + + if err := w.wls.AppendMessage(bucket, keysFileName, version, b); err != nil { + return errors.Wrapf(err, "appending partition key message: %s, %d", keysFileName, version) + } + return nil } -func (w *NopWriteLogWriter) DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error { +func (w *writeLogReadWriter) DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error { + bucket := partitionBucket(qtid.Key(), partition) + return w.wls.DeleteLog(bucket, keysFileName, version) +} + +func (w *writeLogReadWriter) CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, m map[string]uint64) error { + msg := FieldKeyMap{ + TableKey: qtid.Key(), + Field: field, + StringToID: m, + } + + b, err := json.Marshal(msg) + if err != nil { + return errors.Wrap(err, "marshalling field key map to json") + } + + bucket := fieldBucket(qtid.Key(), field) + + if err := w.wls.AppendMessage(bucket, keysFileName, version, b); err != nil { + return errors.Wrapf(err, "appending field key message: %s, %d", keysFileName, version) + } + return nil } -func (w *NopWriteLogWriter) CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, m map[string]uint64) error { +func (w *writeLogReadWriter) DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error { + bucket := fieldBucket(qtid.Key(), field) + return w.wls.DeleteLog(bucket, keysFileName, version) +} + +func (w *writeLogReadWriter) WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error { + b, err := MarshalLogMessage(msg, EncodeTypeJSON) + if err != nil { + return errors.Wrap(err, "marshalling log message") + } + + bucket := partitionBucket(qtid.Key(), partition) + shardKey := shardKey(shard) + + if err := w.wls.AppendMessage(bucket, shardKey, version, b); err != nil { + return errors.Wrapf(err, "appending shard key message: %s, %d", shardKey, version) + } + return nil } -func (w *NopWriteLogWriter) DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error { - return nil -} +func (w *writeLogReadWriter) DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error { + bucket := partitionBucket(qtid.Key(), partition) + shardKey := shardKey(shard) -func (w *NopWriteLogWriter) WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error { - return nil -} - -func (w *NopWriteLogWriter) DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error { - return nil -} - -// WriteLogReader provides the interface for all reads from the write log. -type WriteLogReader interface { - ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader - TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader - FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader -} - -// Ensure type implements interface. -var _ WriteLogReader = (*NopWriteLogReader)(nil) - -// NopWriteLogReader is a no-op implementation of the WriteLogReader interface. -type NopWriteLogReader struct{} - -func NewNopWriteLogReader() *NopWriteLogReader { - return &NopWriteLogReader{} -} - -func (w *NopWriteLogReader) TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader { - return NewNopTableKeyReader() -} - -func (w *NopWriteLogReader) FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader { - return NewNopFieldKeyReader() -} - -func (w *NopWriteLogReader) ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader { - return NewNopShardReader() + return w.wls.DeleteLog(bucket, shardKey, version) } //////////////////////////////////////////////// -type TableKeyReader interface { - Open() error - Read() (PartitionKeyMap, error) - Close() error +func (w *writeLogReadWriter) TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader { + return newTableKeyReader(w.wls, qtid, partition, version) } -// Ensure type implements interface. -var _ TableKeyReader = &NopTableKeyReader{} - -// NopTableKeyReader is a no-op implementation of the TableKeyReader -// interface. -type NopTableKeyReader struct{} - -func NewNopTableKeyReader() *NopTableKeyReader { - return &NopTableKeyReader{} +type tableKeyReader struct { + wl WriteLogService + table dax.TableKey + partition dax.PartitionNum + version int + scanner *bufio.Scanner + closer io.Closer } -func (r *NopTableKeyReader) Open() error { return nil } -func (r *NopTableKeyReader) Read() (PartitionKeyMap, error) { - return PartitionKeyMap{}, io.EOF +func newTableKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) *tableKeyReader { + r := &tableKeyReader{ + wl: wl, + table: qtid.Key(), + partition: partition, + version: version, + } + + return r +} + +func (r *tableKeyReader) Open() error { + bucket := partitionBucket(r.table, r.partition) + + reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version) + if err != nil { + return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version) + } + + r.closer = closer + r.scanner = bufio.NewScanner(reader) + + return nil +} + +func (r *tableKeyReader) Read() (PartitionKeyMap, error) { + if r.scanner == nil { + return PartitionKeyMap{}, io.EOF + } + + var b []byte + var out PartitionKeyMap + + if r.scanner.Scan() { + b = r.scanner.Bytes() + if err := json.Unmarshal(b, &out); err != nil { + return out, err + } + return out, nil + } + if err := r.scanner.Err(); err != nil { + return out, err + } + + return out, io.EOF +} + +func (r *tableKeyReader) Close() error { + if r.closer != nil { + return r.closer.Close() + } + return nil } -func (r *NopTableKeyReader) Close() error { return nil } //////////////////////////////////////////////// -type FieldKeyReader interface { - Open() error - Read() (FieldKeyMap, error) - Close() error +func (w *writeLogReadWriter) FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader { + return newFieldKeyReader(w.wls, qtid, field, version) } -// Ensure type implements interface. -var _ FieldKeyReader = &NopFieldKeyReader{} - -// NopFieldKeyReader is a no-op implementation of the FieldKeyReader -// interface. -type NopFieldKeyReader struct{} - -func NewNopFieldKeyReader() *NopFieldKeyReader { - return &NopFieldKeyReader{} +type fieldKeyReader struct { + wl WriteLogService + table dax.TableKey + field dax.FieldName + version int + scanner *bufio.Scanner + closer io.Closer } -func (r *NopFieldKeyReader) Open() error { return nil } -func (r *NopFieldKeyReader) Read() (FieldKeyMap, error) { - return FieldKeyMap{}, io.EOF +func newFieldKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, field dax.FieldName, version int) *fieldKeyReader { + r := &fieldKeyReader{ + wl: wl, + table: qtid.Key(), + field: field, + version: version, + } + + return r +} + +func (r *fieldKeyReader) Open() error { + bucket := fieldBucket(r.table, r.field) + + reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version) + if err != nil { + return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version) + } + + r.closer = closer + r.scanner = bufio.NewScanner(reader) + + return nil +} + +func (r *fieldKeyReader) Read() (FieldKeyMap, error) { + if r.scanner == nil { + return FieldKeyMap{}, io.EOF + } + + var b []byte + var out FieldKeyMap + + if r.scanner.Scan() { + b = r.scanner.Bytes() + if err := json.Unmarshal(b, &out); err != nil { + return out, err + } + return out, nil + } + if err := r.scanner.Err(); err != nil { + return out, err + } + + return out, io.EOF +} + +func (r *fieldKeyReader) Close() error { + if r.closer != nil { + return r.closer.Close() + } + return nil } -func (r *NopFieldKeyReader) Close() error { return nil } //////////////////////////////////////////////// -type ShardReader interface { - Open() error - Read() (LogMessage, error) - Close() error +func (w *writeLogReadWriter) ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader { + return newShardReader(w.wls, qtid, partition, shard, version) } -// Ensure type implements interface. -var _ ShardReader = &NopShardReader{} - -// NopShardReader is a no-op implementation of the ShardReader interface. -type NopShardReader struct{} - -func NewNopShardReader() *NopShardReader { - return &NopShardReader{} +type shardReader struct { + wl WriteLogService + table dax.TableKey + partition dax.PartitionNum + shard dax.ShardNum + version int + scanner *bufio.Scanner + closer io.Closer } -func (r *NopShardReader) Open() error { return nil } -func (r *NopShardReader) Read() (LogMessage, error) { +func newShardReader(wl WriteLogService, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) *shardReader { + r := &shardReader{ + wl: wl, + table: qtid.Key(), + partition: partition, + shard: shard, + version: version, + } + + return r +} + +func (r *shardReader) Open() error { + bucket := partitionBucket(r.table, r.partition) + shardKey := shardKey(r.shard) + + reader, closer, err := r.wl.LogReader(bucket, shardKey, r.version) + if err != nil { + return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, shardKey, r.version) + } + + r.closer = closer + r.scanner = bufio.NewScanner(reader) + + return nil +} + +func (r *shardReader) Read() (LogMessage, error) { + if r.scanner == nil { + return nil, io.EOF + } + + if r.scanner.Scan() { + return UnmarshalLogMessage(r.scanner.Bytes()) + } + if err := r.scanner.Err(); err != nil { + return nil, err + } + return nil, io.EOF } -func (r *NopShardReader) Close() error { return nil } -//////////////// Messages /////////////////////// - -type PartitionKeyMap struct { - TableKey dax.TableKey `json:"table-key"` - Partition dax.PartitionNum `json:"partition"` - StringToID map[string]uint64 `json:"string-to-id"` -} - -type FieldKeyMap struct { - TableKey dax.TableKey `json:"table-key"` - Field dax.FieldName `json:"field"` - StringToID map[string]uint64 `json:"string-to-id"` -} - -const ( - logMessageTypeImportRoaring byte = iota - logMessageTypeImport - logMessageTypeImportValue - logMessageTypeImportRoaringShard -) - -// encoderKey* are part of the log message header. They indicate what encoding -// type a specific log message is serialized with. -const ( - encoderKeyJSON byte = iota -) - -const ( - EncodeTypeJSON string = "json" - - // encodeVersion refers to the version of the structs used to represent the - // log messages. If we change structs, we'll need to modify this version - // number and maintain the previous version of the structs somewhere for - // deserialization. - encodeVersion byte = 1 -) - -// logMessageEncoder is implemented by any encoder used to serialize LogMessages -// to []byte. -type logMessageEncoder interface { - Key() byte - Marshal(LogMessage) ([]byte, error) - Unmarshal([]byte, LogMessage) error -} - -// LogMessage is implemented by a variety of types which can be serialized as -// messages to the WriteLogger. -type LogMessage interface{} - -// MarshalLogMessage serializes the log message and prepends additional encoding -// information to each message. Currently, we prepend three bytes to each log -// message: -// byte[0]: encodeVersion - this is currently a constant within the code. If we -// modify structs such that they encode differently, we'll have to change the -// constant and keep previous versions of structs for deserialization. -// byte[1]: encodeType (e.g. "json", etc.) -// byte[2]: logMessageType -// -// If we get into a situation where we want more flexibility in these message -// header bytes—for example, if we want to use more than three bytes—we could do -// something with the first bit of the encodeVersion: if it's 1, that could -// indicate that there are additional header bytes, and the following seven bits -// could indicate how many. -func MarshalLogMessage(msg LogMessage, encode string) ([]byte, error) { - encoder, err := getEncoderByType(encode) - if err != nil { - return nil, errors.Wrap(err, "getting encoder by type") +func (r *shardReader) Close() error { + if r.closer != nil { + return r.closer.Close() } - - logMessageType, err := getLogMessageType(msg) - if err != nil { - return nil, errors.Wrap(err, "getting log message type") - } - - var buf []byte - - buf, err = encoder.Marshal(msg) - if err != nil { - return nil, errors.Wrap(err, "marshaling log message") - } - - return append([]byte{encodeVersion, encoder.Key(), logMessageType}, buf...), nil -} - -// UnmarshalLogMessage deserializes the log message based on the log message -// type info. -func UnmarshalLogMessage(b []byte) (LogMessage, error) { - if len(b) < 3 { - return nil, errors.New(errors.ErrUncoded, "log record does not contain a full header") - } - - encVersion := b[0] - encKey := b[1] - logMessageType := b[2] - - // Ensure that the log message is able to be handled by this code. If we - // increment the constant encodeVersion, we'll need to modify this to handle - // the log based on previous encodeVersions. - if encVersion != encodeVersion { - return nil, errors.Errorf("encode version is unsupported: %d", encVersion) - } - - msg, err := logMessageByType(logMessageType) - if err != nil { - return nil, errors.Wrap(err, "getting log message by type") - } - - encoder, err := getEncoderByKey(encKey) - if err != nil { - return nil, errors.Wrap(err, "getting encoder by key") - } - - if err := encoder.Unmarshal(b[3:], &msg); err != nil { - return nil, errors.Wrap(err, "unmarshaling log message") - } - - return msg, nil -} - -func logMessageByType(typ byte) (LogMessage, error) { - switch typ { - case logMessageTypeImportRoaring: - return &ImportRoaringMessage{}, nil - case logMessageTypeImport: - return &ImportMessage{}, nil - case logMessageTypeImportValue: - return &ImportValueMessage{}, nil - case logMessageTypeImportRoaringShard: - return &ImportRoaringShardMessage{}, nil - default: - return nil, errors.Errorf("unknown message type %d", typ) - } -} - -func getLogMessageType(m LogMessage) (byte, error) { - switch m.(type) { - case *ImportRoaringMessage: - return logMessageTypeImportRoaring, nil - case *ImportMessage: - return logMessageTypeImport, nil - case *ImportValueMessage: - return logMessageTypeImportValue, nil - case *ImportRoaringShardMessage: - return logMessageTypeImportRoaringShard, nil - default: - return 0, errors.Errorf("don't have type for message %#v", m) - } -} - -func getEncoderByType(encode string) (logMessageEncoder, error) { - switch encode { - case EncodeTypeJSON: - return &encoderJSON{}, nil - default: - return nil, errors.Errorf("invalid encode type: %s", encode) - } -} - -func getEncoderByKey(id byte) (logMessageEncoder, error) { - switch id { - case encoderKeyJSON: - return &encoderJSON{}, nil - default: - return nil, errors.Errorf("invalid encode type: %d", id) - } -} - -type ImportRoaringMessage struct { - LogMessage `json:"-"` - - Table string `json:"table"` - Field string `json:"field"` - Partition int `json:"partition"` - Shard uint64 `json:"shard"` - Clear bool `json:"clear"` - Action string `json:"action"` // [set, clear, overwrite] - Block int `json:"block"` - Views map[string][]byte `json:"views"` - UpdateExistence bool `json:"update-existence"` -} - -type ImportMessage struct { - LogMessage `json:"-"` - - Table string `json:"table"` - Field string `json:"field"` - Partition int `json:"partition"` - Shard uint64 `json:"shard"` - RowIDs []uint64 `json:"row-ids"` - ColumnIDs []uint64 `json:"column-ids"` - RowKeys []string `json:"row-keys"` - ColumnKeys []string `json:"column-keys"` - Timestamps []int64 `json:"timestamps"` - Clear bool `json:"clear"` - - // options - IgnoreKeyCheck bool `json:"ignore-key-check"` - Presorted bool `json:"presorted"` -} - -type ImportValueMessage struct { - LogMessage `json:"-"` - - Table string `json:"table"` - Field string `json:"field"` - Partition int `json:"partition"` - Shard uint64 `json:"shard"` - ColumnIDs []uint64 `json:"column-ids"` - ColumnKeys []string `json:"column-keys"` - Values []int64 `json:"values"` - FloatValues []float64 `json:"float-values"` - TimestampValues []time.Time `json:"timestamp-values"` - StringValues []string `json:"string-values"` - Clear bool `json:"clear"` - - // options - IgnoreKeyCheck bool `json:"ignore-key-check"` - Presorted bool `json:"presorted"` -} - -type ImportRoaringShardMessage struct { - LogMessage `json:"-"` - - Table string `json:"table"` - Partition int `json:"partition"` - Shard uint64 `json:"shard"` - Views []RoaringUpdate `json:"views"` -} - -// RoaringUpdate is identical to featurebase.RoaringUpdate, but we -// can't import it due to import cycles. TODO featurebase top level -// shouldn't import dax stuff... all the types it needs should just be -// in the top level. -type RoaringUpdate struct { - Field string `json:"field"` - View string `json:"view"` - Clear []byte `json:"clear"` - Set []byte `json:"set"` - ClearRecords bool `json:"clear-records"` -} - -// Ensure type implements interface. -var _ logMessageEncoder = (*encoderJSON)(nil) - -// encoderJSON is an implementation of the logMessageEncoder interface which -// encodes LogMessages as JSON. -type encoderJSON struct{} - -func (e *encoderJSON) Key() byte { - return encoderKeyJSON -} - -func (e *encoderJSON) Marshal(msg LogMessage) ([]byte, error) { - return json.Marshal(msg) -} - -func (e *encoderJSON) Unmarshal(b []byte, msg LogMessage) error { - return json.Unmarshal(b, msg) + return nil } diff --git a/dax/queryer/compute_api.go b/dax/queryer/compute_api.go deleted file mode 100644 index aef166fb5..000000000 --- a/dax/queryer/compute_api.go +++ /dev/null @@ -1,318 +0,0 @@ -package queryer - -import ( - "context" - "time" - - featurebase "github.com/molecula/featurebase/v3" - "github.com/molecula/featurebase/v3/dax" - "github.com/molecula/featurebase/v3/dax/mds/controller/partitioner" - "github.com/molecula/featurebase/v3/errors" -) - -// Ensure type implements interface. -var _ featurebase.ComputeAPI = &qualifiedComputeAPI{} - -type qualifiedComputeAPI struct { - mds MDS - qual dax.TableQualifier -} - -func NewQualifiedComputeAPI(qual dax.TableQualifier, mds MDS) *qualifiedComputeAPI { - return &qualifiedComputeAPI{ - mds: mds, - qual: qual, - } -} - -// importer is used to get the Importer based on the provided address. We used -// to maintain a map of different importers (pointers to computers) running -// in-process, but since getting rid of that logic this method is currently just -// a wrapper around NewComputeImporter. I'm leaving it like this for now in case -// it makes sense for this to become a cache of computer clients. -func (c *qualifiedComputeAPI) importer(addr dax.Address) (Importer, error) { - return NewComputeImporter(addr), nil -} - -func (c *qualifiedComputeAPI) Import(ctx context.Context, qcx *featurebase.Qcx, req *featurebase.ImportRequest, opts ...featurebase.ImportOption) error { - // If the request is empty, return early. - if len(req.ColumnKeys) == 0 && len(req.ColumnIDs) == 0 { - return nil - } - - // Determine if the columns use string keys or not. - var hasColKeys bool - if len(req.ColumnKeys) > 0 { - hasColKeys = true - if len(req.ColumnIDs) > 0 { - return errors.Errorf("import request has both column ids and keys") - } - } - // Determine if the rows use string keys or not. - var hasRowKeys bool - if len(req.RowKeys) > 0 { - hasRowKeys = true - if len(req.RowIDs) > 0 { - return errors.Errorf("import request has both row ids and keys") - } - } - - partitioner := partitioner.NewPartitioner() - - reqPerShard := make(map[uint64]*featurebase.ImportRequest) - - tkey, err := c.indexToQualifiedTableKey(ctx, req.Index) - if err != nil { - return errors.Wrap(err, "converting index to qualified table key") - } - stkey := string(tkey) - - qtid := tkey.QualifiedTableID() - - qtbl, err := c.mds.Table(ctx, qtid) - if err != nil { - return errors.Wrapf(err, "getting table for import: %s", req.Index) - } - - // Translate column keys. - if hasColKeys { - // Get the partitions (and therefore, nodes) responsible for the keys. - pMap := partitioner.PartitionsForKeys(qtbl.Key(), qtbl.PartitionN, req.ColumnKeys...) - - colIDs := make([]uint64, 0, len(req.ColumnKeys)) - for pNum := range pMap { - addr, err := c.mds.IngestPartition(ctx, qtid, pNum) - if err != nil { - return errors.Wrapf(err, "getting ingest partition: %d", pNum) - } - - importer, err := c.importer(addr) - if err != nil { - return errors.Wrapf(err, "getting importer for address: %s", addr) - } - - colKeyMap, err := importer.CreateIndexKeys(ctx, stkey, req.ColumnKeys...) - if err != nil { - return errors.Wrap(err, "creating index keys") - } - for i := range req.ColumnKeys { - colIDs = append(colIDs, colKeyMap[req.ColumnKeys[i]]) - } - } - req.ColumnIDs = colIDs - } - - // Translate row keys. - if hasRowKeys { - addr, err := c.mds.IngestPartition(ctx, qtid, 0) - if err != nil { - return errors.Wrapf(err, "getting ingest partition: %d", 0) - } - - importer, err := c.importer(addr) - if err != nil { - return errors.Wrapf(err, "getting importer for address: %s", addr) - } - - rowKeyMap, err := importer.CreateFieldKeys(ctx, stkey, req.Field, req.RowKeys...) - if err != nil { - return errors.Wrap(err, "creating field keys") - } - rowIDs := make([]uint64, len(req.RowKeys)) - for i := range req.RowKeys { - rowIDs[i] = rowKeyMap[req.RowKeys[i]] - } - req.RowIDs = rowIDs - } - - // Loop over the column ids and split them up by shard. - for ii := range req.ColumnIDs { - // Determine shard. - shard := req.ColumnIDs[ii] / featurebase.ShardWidth - - // Get or create the ImportRequest for this shard. - shardedReq, found := reqPerShard[shard] - if !found { - shardedReq = &featurebase.ImportRequest{ - Index: stkey, - IndexCreatedAt: req.IndexCreatedAt, - Field: req.Field, - FieldCreatedAt: req.FieldCreatedAt, - Shard: shard, - RowIDs: []uint64{}, - ColumnIDs: []uint64{}, - RowKeys: []string{}, - ColumnKeys: []string{}, - Timestamps: []int64{}, - Clear: req.Clear, - } - reqPerShard[shard] = shardedReq - } - - shardedReq.ColumnIDs = append(shardedReq.ColumnIDs, req.ColumnIDs[ii]) - if len(req.RowIDs) > 0 { - shardedReq.RowIDs = append(shardedReq.RowIDs, req.RowIDs[ii]) - } - if len(req.Timestamps) > 0 { - shardedReq.Timestamps = append(shardedReq.Timestamps, req.Timestamps[ii]) - } - } - - // Send each of the sharded ImportRequests to the appropriate compute node. - for shard, req := range reqPerShard { - addr, err := c.mds.IngestShard(ctx, qtid, dax.ShardNum(shard)) - if err != nil { - return errors.Wrapf(err, "getting ingest shard: %d", shard) - } - - importer, err := c.importer(addr) - if err != nil { - return errors.Wrapf(err, "getting importer for address: %s", addr) - } - - importer.Import(ctx, req, - featurebase.OptImportOptionsClear(req.Clear), - featurebase.OptImportOptionsIgnoreKeyCheck(true), - ) - } - - return nil -} - -func (c *qualifiedComputeAPI) ImportValue(ctx context.Context, qcx *featurebase.Qcx, req *featurebase.ImportValueRequest, opts ...featurebase.ImportOption) error { - // If the request is empty, return early. - if len(req.ColumnKeys) == 0 && len(req.ColumnIDs) == 0 { - return nil - } - - // Determine if the columns use string keys or not. - var hasColKeys bool - if len(req.ColumnKeys) > 0 { - hasColKeys = true - if len(req.ColumnIDs) > 0 { - return errors.Errorf("import value request has both column ids and keys") - } - } - - partitioner := partitioner.NewPartitioner() - - reqPerShard := make(map[uint64]*featurebase.ImportValueRequest) - - tkey, err := c.indexToQualifiedTableKey(ctx, req.Index) - if err != nil { - return errors.Wrap(err, "converting index to qualified table key") - } - stkey := string(tkey) - - qtid := tkey.QualifiedTableID() - - qtbl, err := c.mds.Table(ctx, qtid) - if err != nil { - return errors.Wrapf(err, "getting table for importvalue: %s", req.Index) - } - - // Translate column keys. - if hasColKeys { - // Get the partitions (and therefore, nodes) responsible for the keys. - pMap := partitioner.PartitionsForKeys(qtbl.Key(), qtbl.PartitionN, req.ColumnKeys...) - - colIDs := make([]uint64, 0, len(req.ColumnKeys)) - for pNum := range pMap { - addr, err := c.mds.IngestPartition(ctx, qtid, pNum) - if err != nil { - return errors.Wrapf(err, "getting ingest partition: %d", pNum) - } - - importer, err := c.importer(addr) - if err != nil { - return errors.Wrapf(err, "getting importer for address: %s", addr) - } - - colKeyMap, err := importer.CreateIndexKeys(ctx, stkey, req.ColumnKeys...) - if err != nil { - return errors.Wrap(err, "creating index keys") - } - for i := range req.ColumnKeys { - colIDs = append(colIDs, colKeyMap[req.ColumnKeys[i]]) - } - } - req.ColumnIDs = colIDs - } - - // Loop over the column ids and split them up by shard. - for ii := range req.ColumnIDs { - // Determine shard. - shard := req.ColumnIDs[ii] / featurebase.ShardWidth - - // Get or create the ImportRequest for this shard. - shardedReq, found := reqPerShard[shard] - if !found { - shardedReq = &featurebase.ImportValueRequest{ - Index: stkey, - IndexCreatedAt: req.IndexCreatedAt, - Field: req.Field, - FieldCreatedAt: req.FieldCreatedAt, - Shard: shard, - ColumnIDs: []uint64{}, - ColumnKeys: []string{}, - Values: []int64{}, - FloatValues: []float64{}, - TimestampValues: []time.Time{}, - StringValues: []string{}, - Clear: req.Clear, - } - reqPerShard[shard] = shardedReq - } - - shardedReq.ColumnIDs = append(shardedReq.ColumnIDs, req.ColumnIDs[ii]) - if len(req.Values) > 0 { - shardedReq.Values = append(shardedReq.Values, req.Values[ii]) - } - // TODO: The following would populate the other value types, but the - // EncodeImportValues doesn't seem to use this data. So we need to track - // down how this is being used. - // - // if len(req.FloatValues) > 0 { - // shardedReq.FloatValues = append(shardedReq.FloatValues, req.FloatValues[ii]) - // } - // if len(req.TimestampValues) > 0 { - // shardedReq.TimestampValues = append(shardedReq.TimestampValues, req.TimestampValues[ii]) - // } - // if len(req.StringValues) > 0 { - // shardedReq.StringValues = append(shardedReq.StringValues, req.StringValues[ii]) - // } - } - - // Send each of the sharded ImportValueRequests to the appropriate compute - // node. - for shard, req := range reqPerShard { - addr, err := c.mds.IngestShard(ctx, qtid, dax.ShardNum(shard)) - if err != nil { - return errors.Wrapf(err, "getting ingest shard: %d", shard) - } - - importer, err := c.importer(addr) - if err != nil { - return errors.Wrapf(err, "getting importer for address: %s", addr) - } - - importer.ImportValue(ctx, req, - featurebase.OptImportOptionsClear(req.Clear), - featurebase.OptImportOptionsIgnoreKeyCheck(true), - ) - } - - return nil -} - -func (c *qualifiedComputeAPI) Txf() *featurebase.TxFactory { - return &featurebase.TxFactory{} -} - -func (c *qualifiedComputeAPI) indexToQualifiedTableKey(ctx context.Context, index string) (dax.TableKey, error) { - qtid, err := c.mds.TableID(ctx, c.qual, dax.TableName(index)) - if err != nil { - return "", errors.Wrap(err, "converting index to qualified table id") - } - return qtid.Key(), nil -} diff --git a/dax/queryer/compute_importer.go b/dax/queryer/compute_importer.go deleted file mode 100644 index 6f22fa8ed..000000000 --- a/dax/queryer/compute_importer.go +++ /dev/null @@ -1,68 +0,0 @@ -package queryer - -import ( - "context" - - featurebase "github.com/molecula/featurebase/v3" - "github.com/molecula/featurebase/v3/client" - "github.com/molecula/featurebase/v3/dax" - "github.com/molecula/featurebase/v3/errors" -) - -// Ensure type implements interface. -var _ Importer = &ComputeImporter{} - -// ComputeImporter is an implementation of the Importer interface which uses a -// featurebase client to communicate with the compute node. -type ComputeImporter struct { - addr dax.Address -} - -func NewComputeImporter(addr dax.Address) *ComputeImporter { - return &ComputeImporter{ - addr: addr, - } -} - -func (ci *ComputeImporter) CreateIndexKeys(ctx context.Context, index string, keys ...string) (map[string]uint64, error) { - fbClient, err := fbClient(ci.addr) - if err != nil { - return nil, errors.Wrap(err, "getting fb client") - } - - idx := client.NewIndex(index) - return fbClient.CreateIndexKeys(idx, keys...) -} - -func (ci *ComputeImporter) CreateFieldKeys(ctx context.Context, index, field string, keys ...string) (map[string]uint64, error) { - fbClient, err := fbClient(ci.addr) - if err != nil { - return nil, errors.Wrap(err, "getting fb client") - } - - idx := client.NewIndex(index) - fld := idx.Field(field) - return fbClient.CreateFieldKeys(fld, keys...) -} - -func (ci *ComputeImporter) Import(ctx context.Context, req *featurebase.ImportRequest, opts ...featurebase.ImportOption) error { - fbClient, err := fbClient(ci.addr) - if err != nil { - return errors.Wrap(err, "getting fb client") - } - - idx := client.NewIndex(req.Index) - fld := idx.Field(req.Field) - return fbClient.Import(fld, req.Shard, req.RowIDs, req.ColumnIDs, req.Clear) -} - -func (ci *ComputeImporter) ImportValue(ctx context.Context, req *featurebase.ImportValueRequest, opts ...featurebase.ImportOption) error { - fbClient, err := fbClient(ci.addr) - if err != nil { - return errors.Wrap(err, "getting fb client") - } - - idx := client.NewIndex(req.Index) - fld := idx.Field(req.Field) - return fbClient.ImportValues(fld, req.Shard, req.Values, req.ColumnIDs, req.Clear) -} diff --git a/dax/queryer/queryer.go b/dax/queryer/queryer.go index 4be9c5b79..7b0a98b35 100644 --- a/dax/queryer/queryer.go +++ b/dax/queryer/queryer.go @@ -122,9 +122,6 @@ func (q *Queryer) QuerySQL(ctx context.Context, qual dax.TableQualifier, sql str return ret, nil } - // ComputeAPI - capi := NewQualifiedComputeAPI(qual, q.mds) - // SchemaAPI sapi := NewQualifiedSchemaAPI(qual, q.mds) @@ -141,7 +138,7 @@ func (q *Queryer) QuerySQL(ctx context.Context, qual dax.TableQualifier, sql str systemLayer := systemlayer.NewSystemLayer() - pl := planner.NewExecutionPlanner(orch, sapi, sysapi, capi, systemLayer, imp, q.orchestrator.logger, sql) + pl := planner.NewExecutionPlanner(orch, sapi, sysapi, systemLayer, imp, q.orchestrator.logger, sql) planOp, err := pl.CompilePlan(ctx, st) if err != nil { diff --git a/qa/scripts/setupTLS.sh b/qa/scripts/setupTLS.sh index 2956c30e2..109d05902 100755 --- a/qa/scripts/setupTLS.sh +++ b/qa/scripts/setupTLS.sh @@ -16,10 +16,15 @@ ifErr() { fi } -echo "installing go" +echo "installing wget" sudo yum install wget -y ifErr "installing wget" +echo "installing nss-tools" +sudo yum install nss-tools -y +ifErr "installing nss-tools" + +echo "installing git" sudo yum install git -y ifErr "installing git" diff --git a/server/server.go b/server/server.go index 101f3eabc..ff896ef10 100644 --- a/server/server.go +++ b/server/server.go @@ -33,7 +33,6 @@ import ( "github.com/molecula/featurebase/v3/boltdb" "github.com/molecula/featurebase/v3/dax" "github.com/molecula/featurebase/v3/dax/computer" - "github.com/molecula/featurebase/v3/dax/computer/alpha" "github.com/molecula/featurebase/v3/disco" "github.com/molecula/featurebase/v3/encoding/proto" petcd "github.com/molecula/featurebase/v3/etcd" @@ -77,9 +76,9 @@ type Command struct { logger loggerLogger queryLogger loggerLogger - Registrar computer.Registrar - writeLogger computer.WriteLogger - snapshotter computer.Snapshotter + Registrar computer.Registrar + writeLogService computer.WriteLogService + snapshotService computer.SnapshotService Handler pilosa.HandlerI httpHandler http.Handler @@ -151,10 +150,10 @@ func OptCommandSetConfig(config *Config) CommandOption { func OptCommandInjections(inj Injections) CommandOption { return func(c *Command) error { if inj.WriteLogger != nil { - c.writeLogger = inj.WriteLogger + c.writeLogService = inj.WriteLogger } if inj.Snapshotter != nil { - c.snapshotter = inj.Snapshotter + c.snapshotService = inj.Snapshotter } c.isComputeNode = inj.IsComputeNode return nil @@ -162,8 +161,8 @@ func OptCommandInjections(inj Injections) CommandOption { } type Injections struct { - WriteLogger computer.WriteLogger - Snapshotter computer.Snapshotter + WriteLogger computer.WriteLogService + Snapshotter computer.SnapshotService IsComputeNode bool } @@ -555,16 +554,16 @@ func (m *Command) setupServer() error { // WriteLogger setup. var wlw computer.WriteLogWriter = computer.NewNopWriteLogWriter() var wlr computer.WriteLogReader = computer.NewNopWriteLogReader() - if m.writeLogger != nil { - alphaWriteLog := alpha.NewAlphaWriteLog(m.writeLogger) - wlr = alphaWriteLog - wlw = alphaWriteLog + if m.writeLogService != nil { + wlrw := computer.NewWriteLogReadWriter(m.writeLogService) + wlr = wlrw + wlw = wlrw } // Snapshotter setup. var snap computer.SnapshotReadWriter = computer.NewNopSnapshotReadWriter() - if m.snapshotter != nil { - snap = alpha.NewAlphaSnapshot(m.snapshotter) + if m.snapshotService != nil { + snap = computer.NewSnapshotReadWriter(m.snapshotService) } executionPlannerFn := func(e pilosa.Executor, api *pilosa.API, sql string) sql3.CompilePlanner { @@ -572,7 +571,7 @@ func (m *Command) setupServer() error { fsapi := &pilosa.FeatureBaseSystemAPI{API: api} fimp := &batch.FeaturebaseImporter{API: api} - return planner.NewExecutionPlanner(e, fapi, fsapi, api, m.Server.SystemLayer, fimp, m.logger, sql) + return planner.NewExecutionPlanner(e, fapi, fsapi, m.Server.SystemLayer, fimp, m.logger, sql) } serverOptions := []pilosa.ServerOption{ diff --git a/sql3/planner/executionplanner.go b/sql3/planner/executionplanner.go index 11ab0a9fb..605269ed4 100644 --- a/sql3/planner/executionplanner.go +++ b/sql3/planner/executionplanner.go @@ -26,7 +26,6 @@ type ExecutionPlanner struct { executor pilosa.Executor schemaAPI pilosa.SchemaAPI systemAPI pilosa.SystemAPI - computeAPI pilosa.ComputeAPI systemLayerAPI pilosa.SystemLayerAPI importer batch.Importer logger logger.Logger @@ -34,12 +33,11 @@ type ExecutionPlanner struct { scopeStack *scopeStack } -func NewExecutionPlanner(executor pilosa.Executor, schemaAPI pilosa.SchemaAPI, systemAPI pilosa.SystemAPI, computeAPI pilosa.ComputeAPI, systemLayerAPI pilosa.SystemLayerAPI, importer batch.Importer, logger logger.Logger, sql string) *ExecutionPlanner { +func NewExecutionPlanner(executor pilosa.Executor, schemaAPI pilosa.SchemaAPI, systemAPI pilosa.SystemAPI, systemLayerAPI pilosa.SystemLayerAPI, importer batch.Importer, logger logger.Logger, sql string) *ExecutionPlanner { return &ExecutionPlanner{ executor: executor, schemaAPI: newSystemTableDefintionsWrapper(schemaAPI), systemAPI: systemAPI, - computeAPI: computeAPI, systemLayerAPI: systemLayerAPI, importer: importer, logger: logger,