scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

clean up unused code/comments

Browse source
This commit is contained in:
Matthew Jaffee 2022-12-14 10:20:40 -06:00 • committed by Matthew Jaffee
parent bee666c07c
commit 3ddf79160f
11 changed files with 2 additions and 1418 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
api.go
View file

@ -3108,18 +3108,14 @@ func (api *API) SnapshotShardData(ctx context.Context, req *dax.SnapshotShardDat
partition := disco.ShardToShardPartition(string(req.TableKey), uint64(req.ShardNum), disco.DefaultPartitionN)
partitionNum := dax.PartitionNum(partition)
// Create the snapshot for the current version. How do we ensure
// here that any new writes go to the new write log since we are
// now reading a version of the shard which exists only at this
// exact point in time? Ans: we'll cut over to the new storage
// manager and call IncrementWriteLogVersion while a write Tx is
// held on RBF.
// Open a write Tx snapshotting current version.
rc, err := api.IndexShardSnapshot(ctx, string(req.TableKey), uint64(req.ShardNum), true)
if err != nil {
return errors.Wrap(err, "getting index/shard readcloser")
}
mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, req.ShardNum)
// Bump writelog version while write Tx is held.
if err := mgr.IncrementWLVersion(); err != nil {
return errors.Wrap(err, "incrementing write log version")
}

136
dax/computer/interfaces.go
View file

@ -35,78 +35,6 @@ type SnapshotService interface {
List(bucket, key string) ([]SnapInfo, error)
}
// ServerlessStorage is the interface to a particular shard or
// translation store that contains both writelogger and
// snapshotter. The interface is designed such that implementations
// are expected to be stateful.
//
// One must not call LoadWriteLog until after calling
// LoadLatestSnapshot. One must not call Append, IncrementWLVersion,
// or Snapshot until after successfully calling Lock.
// TODO(jaffee), this doesn't need to be an interface. Remove.
type ServerlessStorage interface {
// LoadLatestSnapshot loads the latest available snapshot in the snapshot store.
LoadLatestSnapshot() (data io.ReadCloser, err error)
// // Potential future methods to support getting older versions. SnapInfo would have timestamp information as well.
//
// ListSnapshots() []SnapInfo
// LoadSnapshot(version int) (data io.ReadCloser, err error)
// LoadWriteLog can be called after LoadLatestSnapshot. It loads
// any writelog data which has been written since the latest
// snapshot. Subsequent calls to LoadWriteLog will only return new
// data that hasn't previously been returned from LoadWriteLog.
LoadWriteLog() (data io.ReadCloser, err error)
// Lock acquires an advisory lock for this resource which grants
// us exclusive access to write to it. The normal pattern is to
// call:
//
// 1. LoadLatestSnapshot
// 2. LoadWriteLog
// 3. Lock
// 4. LoadWriteLog
//
// The second call to LoadWriteLog is necessary in case any writes
// occurred between the last load and acquiring the lock. Once the
// lock is acquired it should not be possible for any more writes
// to occur. Lock will error if (a) we fail to acquire the lock or
// (b) the state of the snapshot store for this resource is not
// identical to what is was before the lock was acquired. Case (b)
// means that quite a lot has happened in between LoadWriteLog and
// Lock, and we should probably just die and start over.
Lock() error
// Append appends the msg to the write log. It will fail if we
// haven't properly loaded and gotten a lock for the resource
// we're writing to.
Append(msg []byte) error
// IncrementWLVersion should be called during snapshotting with a
// write Tx held on the local resource. This ensures that any
// writes which completed prior to the snapshot are in the prior
// WL and any that complete after the snapshot are in the
// incremented WL.
IncrementWLVersion() error
// Snapshot takes a ReadCloser which has the contents of the
// resource being tracked at a particular point in time and writes
// them to the Snapshot Store. Upon a successful write it will
// truncate any write logs which are now incorporated into the
// snapshot.
Snapshot(rc io.ReadCloser) error
SnapshotTo(wt io.WriterTo) error
// Unlock releases the lock. This should be called if control of
// the underlying resource is being transitioned to another
// node. Ideally it's also called if the process crashes (e.g. via
// a defer), but an implementation based on filesystem locks
// should have those removed by the operating system when the
// process exits anyway.
Unlock() error
}
// SnapInfo holds metadata about a snapshot.
type SnapInfo struct {
Version int
@ -116,70 +44,6 @@ type SnapInfo struct {
// WriteLogInfo holds metadata about a write log.
type WriteLogInfo SnapInfo
// 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{}

162
dax/computer/noop.go
View file

@ -1,162 +0,0 @@
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 }

94
dax/computer/snapshot.go
View file

@ -1,94 +0,0 @@
package computer
import (
"context"
"io"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/errors"
)
// Ensure type implements interface.
var _ SnapshotReadWriter = &snapshotReadWriter{}
// 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 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 (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 (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 (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 (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 (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
}

307
dax/computer/writelog.go
View file

@ -1,307 +0,0 @@
package computer
import (
"bufio"
"context"
"encoding/json"
"io"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/errors"
)
// Ensure type implements interface.
var _ WriteLogReader = &writeLogReadWriter{}
var _ WriteLogWriter = &writeLogReadWriter{}
// 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 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 *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 *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 *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)
return w.wls.DeleteLog(bucket, shardKey, version)
}
////////////////////////////////////////////////
func (w *writeLogReadWriter) TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader {
return newTableKeyReader(w.wls, qtid, partition, version)
}
type tableKeyReader struct {
wl WriteLogService
table dax.TableKey
partition dax.PartitionNum
version int
scanner *bufio.Scanner
closer io.Closer
}
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)
readcloser, 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 = readcloser
r.scanner = bufio.NewScanner(readcloser)
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 (w *writeLogReadWriter) FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader {
return newFieldKeyReader(w.wls, qtid, field, version)
}
type fieldKeyReader struct {
wl WriteLogService
table dax.TableKey
field dax.FieldName
version int
scanner *bufio.Scanner
closer io.Closer
}
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)
readcloser, 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 = readcloser
r.scanner = bufio.NewScanner(readcloser)
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 (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)
}
type shardReader struct {
wl WriteLogService
table dax.TableKey
partition dax.PartitionNum
shard dax.ShardNum
version int
scanner *bufio.Scanner
closer io.Closer
}
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)
readcloser, 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 = readcloser
r.scanner = bufio.NewScanner(readcloser)
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 *shardReader) Close() error {
if r.closer != nil {
return r.closer.Close()
}
return nil
}

101
dax/snapshotter/api/openapi.yaml
View file

@ -1,101 +0,0 @@
openapi: 3.0.3
info:
title: Snapshotter
description: The alpha implementation of the Snapshotter interface.
version: 0.0.0
paths:
/snapshotter/health:
get:
summary: Health check endpoint.
description: Provides an endpoint to check the overall health of the Snapshotter service.
operationId: GetHealth
responses:
200:
description: Service is healthy.
/snapshotter/write-snapshot:
post:
summary: Write snapshot.
description: Write snapshot based on bucket/key.
operationId: PostWriteSnapshot
parameters:
- name: bucket
in: query
description: bucket containing snapshot key
required: true
schema:
type: string
- name: key
in: query
description: key identifying snapshot
required: true
schema:
type: string
- name: version
in: query
description: bucket/key version
required: true
schema:
type: integer
format: int64
requestBody:
content:
text/plain:
schema:
type: string
format: byte
responses:
200:
$ref: '#/components/responses/WriteSnapshotResponse'
/snapshotter/read-snapshot:
get:
summary: Read snapshot.
description: Read snapshot based on bucket/key.
operationId: GetReadSnapshot
parameters:
- name: bucket
in: query
description: bucket containing snapshot key
required: true
schema:
type: string
- name: key
in: query
description: key identifying snapshot
required: true
schema:
type: string
- name: version
in: query
description: bucket/key version
required: true
schema:
type: integer
format: int64
requestBody:
content:
text/plain:
schema:
type: string
format: byte
responses:
200:
description: Bytes making up the contents of the snapshot.
content:
text/plain:
schema:
type: string
format: byte
components:
responses:
WriteSnapshotResponse:
description: Placeholder response.
content:
application/json:
schema:
type: object

111
dax/snapshotter/client/client.go
View file

@ -1,111 +0,0 @@
// Package client contains an http implementation of the WriteLogger client.
package client
// import (
// "bytes"
// "encoding/json"
// "fmt"
// "io"
// "net/http"
// "net/url"
// "github.com/molecula/featurebase/v3/dax"
// snapshotterhttp "github.com/molecula/featurebase/v3/dax/snapshotter/http"
// "github.com/molecula/featurebase/v3/errors"
// )
// const defaultScheme = "http"
// // TODO(jaffee): remove this?
// // Snapshotter is a client for the Snapshotter API methods.
// type Snapshotter struct {
// address dax.Address
// }
// func New(address dax.Address) *Snapshotter {
// return &Snapshotter{
// address: address,
// }
// }
// func (s *Snapshotter) Write(bucket string, key string, version int, rc io.ReadCloser) error {
// url := fmt.Sprintf("%s/snapshotter/write-snapshot?bucket=%s&key=%s&version=%d",
// s.address.WithScheme(defaultScheme),
// url.QueryEscape(bucket),
// url.QueryEscape(key),
// version,
// )
// // Post the request.
// resp, err := http.Post(url, "", rc)
// if err != nil {
// return errors.Wrap(err, "posting write-snapshot")
// }
// defer resp.Body.Close()
// if resp.StatusCode != http.StatusOK {
// b, _ := io.ReadAll(resp.Body)
// return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
// }
// var wsr snapshotterhttp.WriteSnapshotResponse
// if err := json.NewDecoder(resp.Body).Decode(&wsr); err != nil {
// return errors.Wrap(err, "reading response body")
// }
// return nil
// }
// // WriteTo is exactly the same as Write, except that it takes an io.WriteTo
// // instead of an io.ReadCloser. This needs to be cleaned up so that we're only
// // using one or the other.
// func (s *Snapshotter) WriteTo(bucket string, key string, version int, wrTo io.WriterTo) error {
// url := fmt.Sprintf("%s/snapshotter/write-snapshot?bucket=%s&key=%s&version=%d",
// s.address.WithScheme(defaultScheme),
// url.QueryEscape(bucket),
// url.QueryEscape(key),
// version,
// )
// buf := &bytes.Buffer{}
// if _, err := wrTo.WriteTo(buf); err != nil {
// return errors.Wrap(err, "writing to buffer")
// }
// // Post the request.
// resp, err := http.Post(url, "", buf)
// if err != nil {
// return errors.Wrap(err, "posting write-snapshot")
// }
// defer resp.Body.Close()
// if resp.StatusCode != http.StatusOK {
// b, _ := io.ReadAll(resp.Body)
// return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
// }
// var wsr snapshotterhttp.WriteSnapshotResponse
// if err := json.NewDecoder(resp.Body).Decode(&wsr); err != nil {
// return errors.Wrap(err, "reading response body")
// }
// return nil
// }
// func (s *Snapshotter) Read(bucket string, key string, version int) (io.ReadCloser, error) {
// url := fmt.Sprintf("%s/snapshotter/read-snapshot?bucket=%s&key=%s&version=%d",
// s.address.WithScheme(defaultScheme),
// url.QueryEscape(bucket),
// url.QueryEscape(key),
// version,
// )
// // Get the request.
// resp, err := http.Get(url)
// if err != nil {
// return nil, errors.Wrap(err, "getting read-snapshot")
// }
// return resp.Body, nil
// }

120
dax/snapshotter/http/handler.go
View file

@ -1,120 +0,0 @@
package http
// import (
// "encoding/json"
// "io"
// "net/http"
// "strconv"
// "github.com/gorilla/mux"
// "github.com/molecula/featurebase/v3/dax/snapshotter"
// "github.com/molecula/featurebase/v3/rbf"
// )
// func Handler(s *snapshotter.Snapshotter) http.Handler {
// svr := &server{
// snapshotter: s,
// }
// router := mux.NewRouter()
// router.HandleFunc("/health", svr.getHealth).Methods("GET").Name("GetHealth")
// router.HandleFunc("/write-snapshot", svr.postWriteSnapshot).Methods("POST").Name("PostWriteSnapshot")
// router.HandleFunc("/read-snapshot", svr.getReadSnapshot).Methods("GET").Name("GetReadSnapshot")
// return router
// }
// type server struct {
// snapshotter *snapshotter.Snapshotter
// }
// // GET /health
// func (s *server) getHealth(w http.ResponseWriter, r *http.Request) {
// w.WriteHeader(http.StatusOK)
// }
// // POST /write-snapshot
// func (s *server) postWriteSnapshot(w http.ResponseWriter, r *http.Request) {
// bucket := r.URL.Query().Get("bucket")
// if bucket == "" {
// http.Error(w, "bucket required", http.StatusBadRequest)
// return
// }
// key := r.URL.Query().Get("key")
// if key == "" {
// http.Error(w, "key required", http.StatusBadRequest)
// return
// }
// versionArg := r.URL.Query().Get("version")
// versionInt64, err := strconv.ParseInt(versionArg, 10, 64)
// if err != nil {
// http.Error(w, "bad shard", http.StatusBadRequest)
// return
// }
// version := int(versionInt64)
// body := r.Body
// defer body.Close()
// if err := s.snapshotter.Write(bucket, key, version, body); err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// resp := &WriteSnapshotResponse{}
// if err := json.NewEncoder(w).Encode(resp); err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// }
// type WriteSnapshotResponse struct{}
// // GET /read-snapshot
// func (s *server) getReadSnapshot(w http.ResponseWriter, r *http.Request) {
// bucket := r.URL.Query().Get("bucket")
// if bucket == "" {
// http.Error(w, "bucket required", http.StatusBadRequest)
// return
// }
// key := r.URL.Query().Get("key")
// if key == "" {
// http.Error(w, "key required", http.StatusBadRequest)
// return
// }
// versionArg := r.URL.Query().Get("version")
// versionInt64, err := strconv.ParseInt(versionArg, 10, 64)
// if err != nil {
// http.Error(w, "bad shard", http.StatusBadRequest)
// return
// }
// version := int(versionInt64)
// rc, err := s.snapshotter.Read(bucket, key, version)
// if err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// defer rc.Close()
// // TODO: is rbf.PageSize a problem here for non-RBF snapshots (i.e. keys)?
// // Copy data to response body.
// if _, err := io.CopyBuffer(&passthroughWriter{w}, rc, make([]byte, rbf.PageSize)); err != nil {
// http.Error(w, err.Error(), http.StatusInternalServerError)
// return
// }
// }
// // passthroughWriter is used to remove non-Writer interfaces from an io.Writer.
// // For example, a writer that implements io.ReaderFrom can change io.Copy() behavior.
// type passthroughWriter struct {
// w io.Writer
// }
// func (w *passthroughWriter) Write(p []byte) (int, error) {
// return w.w.Write(p)
// }

113
dax/writelogger/api/openapi.yaml
View file

@ -1,113 +0,0 @@
openapi: 3.0.3
info:
title: WriteLogger
description: The alpha implementation of the WriteLogger interface.
version: 0.0.0
paths:
/writelogger/health:
get:
summary: Health check endpoint.
description: Provides an endpoint to check the overall health of the WriteLogger service.
operationId: GetHealth
responses:
200:
description: Service is healthy.
/writelogger/append-message:
post:
summary: Append message to WriteLogger.
description: Appends a message to a versioned bucket/key.
operationId: PostAppendMessage
requestBody:
content:
application/json:
example:
bucket: example-bucket
key: unique-key
version: 4
message: SGVsbG8gV29ybGQ=
schema:
type: object
properties:
bucket:
type: string
key:
type: string
version:
type: integer
format: int64
message:
type: string
format: byte
responses:
200:
$ref: '#/components/responses/AppendMessageResponse'
/writelogger/log-reader:
post:
summary: Read log.
description: Reads an entire log (collection of messages) at bucket/key for the given version.
operationId: PostLogReader
requestBody:
content:
application/json:
example:
bucket: example-bucket
key: unique-key
version: 4
schema:
type: object
properties:
bucket:
type: string
key:
type: string
version:
type: integer
format: int64
responses:
200:
description: Bytes making up the contents of the log.
content:
text/plain:
schema:
type: string
format: byte
/writelogger/delete-log:
post:
summary: Delete log.
description: Deletes the log at bucket/key for the given version.
operationId: PostDeleteLog
requestBody:
content:
application/json:
example:
bucket: example-bucket
key: unique-key
version: 4
schema:
type: object
properties:
bucket:
type: string
key:
type: string
version:
type: integer
format: int64
responses:
200:
description: Log was deleted.
components:
responses:
AppendMessageResponse:
description: Placeholder response.
content:
application/json:
schema:
type: object

147
dax/writelogger/client/client.go
View file

@ -1,147 +0,0 @@
// Package client contains an http implementation of the WriteLogger client.
package client
// import (
// "bytes"
// "encoding/json"
// "fmt"
// "io"
// "net/http"
// "github.com/molecula/featurebase/v3/dax"
// "github.com/molecula/featurebase/v3/errors"
// )
// // TODO(jaffee): remove this?
// const defaultScheme = "http"
// // WriteLogger is a client for the WriteLogger API methods.
// type WriteLogger struct {
// address dax.Address
// }
// func New(address dax.Address) *WriteLogger {
// return &WriteLogger{
// address: address,
// }
// }
// func (w *WriteLogger) AppendMessage(bucket string, key string, version int, msg []byte) error {
// url := fmt.Sprintf("%s/writelogger/append-message", w.address.WithScheme(defaultScheme))
// req := &AppendMessageRequest{
// Bucket: bucket,
// Key: key,
// Version: version,
// Message: msg,
// }
// // Encode the request.
// postBody, err := json.Marshal(req)
// if err != nil {
// return errors.Wrap(err, "marshalling post request")
// }
// requestBody := bytes.NewBuffer(postBody)
// // Post the request.
// resp, err := http.Post(url, "application/json", requestBody)
// if err != nil {
// return errors.Wrap(err, "posting append-message request")
// }
// defer resp.Body.Close()
// if resp.StatusCode != http.StatusOK {
// b, _ := io.ReadAll(resp.Body)
// return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
// }
// var isr *AppendMessageResponse
// if err := json.NewDecoder(resp.Body).Decode(&isr); err != nil {
// return errors.Wrap(err, "reading response body")
// }
// return nil
// }
// type AppendMessageRequest struct {
// Bucket string `json:"bucket"`
// Key string `json:"key"`
// Version int `json:"version"`
// Message []byte `json:"message"`
// }
// type AppendMessageResponse struct{}
// func (w *WriteLogger) LogReader(bucket string, key string, version int) (io.Reader, io.Closer, error) {
// url := fmt.Sprintf("%s/writelogger/log-reader", w.address.WithScheme(defaultScheme))
// req := &LogReaderRequest{
// Bucket: bucket,
// Version: version,
// Key: key,
// }
// // Encode the request.
// postBody, err := json.Marshal(req)
// if err != nil {
// return nil, nil, errors.Wrap(err, "marshalling post request")
// }
// requestBody := bytes.NewBuffer(postBody)
// // Post the request.
// resp, err := http.Post(url, "application/json", requestBody)
// if err != nil {
// return nil, nil, errors.Wrap(err, "posting log-reader request")
// }
// if resp.StatusCode != http.StatusOK {
// b, _ := io.ReadAll(resp.Body)
// defer resp.Body.Close()
// return nil, nil, errors.Errorf("status code: %d: %s", resp.StatusCode, b)
// }
// return resp.Body, resp.Body, nil
// }
// type LogReaderRequest struct {
// Bucket string `json:"bucket"`
// Version int `json:"version"`
// Key string `json:"key"`
// }
// func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error {
// url := fmt.Sprintf("%s/writelogger/delete-log", w.address.WithScheme(defaultScheme))
// req := &DeleteLogRequest{
// Bucket: bucket,
// Version: version,
// Key: key,
// }
// // Encode the request.
// postBody, err := json.Marshal(req)
// if err != nil {
// return errors.Wrap(err, "marshalling post request")
// }
// requestBody := bytes.NewBuffer(postBody)
// // Post the request.
// resp, err := http.Post(url, "application/json", requestBody)
// if err != nil {
// return errors.Wrap(err, "posting log-reader request")
// }
// if resp.StatusCode != http.StatusOK {
// b, _ := io.ReadAll(resp.Body)
// defer resp.Body.Close()
// return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
// }
// return nil
// }
// type DeleteLogRequest struct {
// Bucket string `json:"bucket"`
// Version int `json:"version"`
// Key string `json:"key"`
// }

121
dax/writelogger/http/handler.go
View file

@ -1,121 +0,0 @@
package http
// import (
// "encoding/json"
// "io"
// "net/http"
// "github.com/gorilla/mux"
// "github.com/molecula/featurebase/v3/dax/writelogger"
// "github.com/molecula/featurebase/v3/logger"
// )
// func Handler(w *writelogger.WriteLogger, logger logger.Logger) http.Handler {
// svr := &server{
// writeLogger: w,
// logger: logger,
// }
// router := mux.NewRouter()
// router.HandleFunc("/health", svr.getHealth).Methods("GET").Name("GetHealth")
// router.HandleFunc("/append-message", svr.postAppendMessage).Methods("POST").Name("PostAppendMessage")
// router.HandleFunc("/log-reader", svr.postLogReader).Methods("POST").Name("PostLogReader")
// router.HandleFunc("/delete-log", svr.postDeleteLog).Methods("POST").Name("PostDeleteLog")
// return router
// }
// type server struct {
// writeLogger *writelogger.WriteLogger
// logger logger.Logger
// }
// // GET /health
// func (s *server) getHealth(w http.ResponseWriter, r *http.Request) {
// w.WriteHeader(http.StatusOK)
// }
// // POST /append-message
// func (s *server) postAppendMessage(w http.ResponseWriter, r *http.Request) {
// body := r.Body
// defer body.Close()
// req := AppendMessageRequest{}
// if err := json.NewDecoder(body).Decode(&req); err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// err := s.writeLogger.AppendMessage(req.Bucket, req.Key, req.Version, req.Message)
// if err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// resp := AppendMessageResponse{}
// if err := json.NewEncoder(w).Encode(resp); err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// }
// type AppendMessageRequest struct {
// Bucket string `json:"bucket"`
// Key string `json:"key"`
// Version int `json:"version"`
// Message []byte `json:"message"`
// }
// type AppendMessageResponse struct{}
// // POST /log-reader
// func (s *server) postLogReader(w http.ResponseWriter, r *http.Request) {
// body := r.Body
// defer body.Close()
// req := LogReaderRequest{}
// if err := json.NewDecoder(body).Decode(&req); err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// reader, closer, err := s.writeLogger.LogReader(req.Bucket, req.Key, req.Version)
// if err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// defer closer.Close()
// if _, err := io.Copy(w, reader); err != nil {
// s.logger.Printf("error streaming log data: %s", err)
// }
// }
// type LogReaderRequest struct {
// Bucket string `json:"bucket"`
// Version int `json:"version"`
// Key string `json:"key"`
// }
// // POST /delete-log
// func (s *server) postDeleteLog(w http.ResponseWriter, r *http.Request) {
// body := r.Body
// defer body.Close()
// req := DeleteLogRequest{}
// if err := json.NewDecoder(body).Decode(&req); err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// if err := s.writeLogger.DeleteLog(req.Bucket, req.Key, req.Version); err != nil {
// http.Error(w, err.Error(), http.StatusBadRequest)
// return
// }
// }
// type DeleteLogRequest struct {
// Bucket string `json:"bucket"`
// Version int `json:"version"`
// Key string `json:"key"`
// }