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

rename shard partition field (#2302)

Browse source 
* Rename dax.Shard to dax.VersionedShard

* Rename dax.Partition to dax.VersionedPartition

* Rename dax.FieldVersion to dax.VersionedField

* Rename go files to a standard

(cherry picked from commit 49d0e0fbc8)
This commit is contained in:
Travis Turner 2022-11-16 15:58:20 -06:00 • committed by Fletcher Haynes
parent 10026d0a90
commit 202a296131
21 changed files with 615 additions and 614 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
api.go
View file

@ -722,7 +722,7 @@ func (api *API) getOrCreateShardVersion(ctx context.Context, indexName string, s
} else if !found {
version = 0
api.server.logger.Printf("could not find version for shard: %s, %d, so creating 0", tableName, shardNum)
if err := api.holder.versionStore.AddShards(ctx, qtid, dax.NewShard(shardNum, version)); err != nil {
if err := api.holder.versionStore.AddShards(ctx, qtid, dax.NewVersionedShard(shardNum, version)); err != nil {
return -1, errors.Wrap(err, "adding shard 0")
}
}
@ -3103,7 +3103,7 @@ func (api *API) SnapshotShardData(ctx context.Context, req *dax.SnapshotShardDat
// Increment the version of the shard managed by this node.
if err := api.holder.versionStore.AddShards(ctx, qtid,
dax.NewShard(req.ShardNum, req.ToVersion),
dax.NewVersionedShard(req.ShardNum, req.ToVersion),
); err != nil {
return errors.Wrap(err, "incrementing shard version locally")
}
@ -3151,7 +3151,7 @@ func (api *API) SnapshotTableKeys(ctx context.Context, req *dax.SnapshotTableKey
// Increment the version of the partition managed by this node.
if err := api.holder.versionStore.AddPartitions(ctx, qtid,
dax.NewPartition(req.PartitionNum, req.ToVersion),
dax.NewVersionedPartition(req.PartitionNum, req.ToVersion),
); err != nil {
return errors.Wrap(err, "incrementing partition version locally")
}
@ -3192,7 +3192,7 @@ func (api *API) SnapshotFieldKeys(ctx context.Context, req *dax.SnapshotFieldKey
// Increment the version of the field managed by this node.
if err := api.holder.versionStore.AddFields(ctx, qtid,
dax.NewFieldVersion(req.Field, req.ToVersion),
dax.NewVersionedField(req.Field, req.ToVersion),
); err != nil {
return errors.Wrap(err, "incrementing field version locally")
}
@ -3324,7 +3324,7 @@ var methodsNormal = map[apiMethod]struct{}{
apiMutexCheck: {},
}
func shardInShards(i dax.ShardNum, s dax.Shards) bool {
func shardInShards(i dax.ShardNum, s dax.VersionedShards) bool {
for _, o := range s {
if i == o.Num {
return true

62
api_directive.go
View file

@ -101,19 +101,19 @@ type directiveJobTableKeys struct {
directiveJobType
idx *Index
tkey dax.TableKey
partition dax.Partition
partition dax.VersionedPartition
}
type directiveJobFieldKeys struct {
directiveJobType
tkey dax.TableKey
field dax.FieldVersion
field dax.VersionedField
}
type directiveJobShards struct {
directiveJobType
tkey dax.TableKey
shard dax.Shard
shard dax.VersionedShard
}
// directiveWorker is a worker in a worker pool which handles portions of a
@ -347,7 +347,7 @@ func (api *API) pushJobsTableKeys(ctx context.Context, jobs chan<- directiveJobT
}
}
func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey, partition dax.Partition) error {
func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey, partition dax.VersionedPartition) error {
qtid := tkey.QualifiedTableID()
// Load the previous snapshot. Version 0 doesn't have a snapshot
@ -422,7 +422,7 @@ func (api *API) pushJobsFieldKeys(ctx context.Context, jobs chan<- directiveJobT
}
}
func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax.FieldVersion) error {
func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax.VersionedField) error {
qtid := tkey.QualifiedTableID()
// Load the previous snapshot. Version 0 doesn't have a snapshot
@ -505,7 +505,7 @@ func (api *API) pushJobsShards(ctx context.Context, jobs chan<- directiveJobType
}
}
func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.Shard) error {
func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.VersionedShard) error {
qtid := tkey.QualifiedTableID()
partition := disco.ShardToShardPartition(string(tkey), uint64(shard.Num), disco.DefaultPartitionN)
@ -706,11 +706,11 @@ func thingsAdded[K comparable](from []K, to []K) []K {
// partitionsComparer is used to compare the differences between two maps of
// table:[]partition.
type partitionsComparer struct {
from map[dax.TableKey]dax.Partitions
to map[dax.TableKey]dax.Partitions
from map[dax.TableKey]dax.VersionedPartitions
to map[dax.TableKey]dax.VersionedPartitions
}
func newPartitionsComparer(from map[dax.TableKey]dax.Partitions, to map[dax.TableKey]dax.Partitions) *partitionsComparer {
func newPartitionsComparer(from map[dax.TableKey]dax.VersionedPartitions, to map[dax.TableKey]dax.VersionedPartitions) *partitionsComparer {
return &partitionsComparer{
from: from,
to: to,
@ -719,23 +719,23 @@ func newPartitionsComparer(from map[dax.TableKey]dax.Partitions, to map[dax.Tabl
// added returns the partitions which are present in `to` but not in `from`. The
// results remain in the format of a map of table:[]partition.
func (p *partitionsComparer) added() map[dax.TableKey]dax.Partitions {
func (p *partitionsComparer) added() map[dax.TableKey]dax.VersionedPartitions {
return partitionsAdded(p.from, p.to)
}
// removed returns the partitions which are present in `from` but not in `to`.
// The results remain in the format of a map of table:[]partition.
func (p *partitionsComparer) removed() map[dax.TableKey]dax.Partitions {
func (p *partitionsComparer) removed() map[dax.TableKey]dax.VersionedPartitions {
return partitionsAdded(p.to, p.from)
}
// partitionsAdded returns the partitions which are present in `to` but not in `from`.
func partitionsAdded(from map[dax.TableKey]dax.Partitions, to map[dax.TableKey]dax.Partitions) map[dax.TableKey]dax.Partitions {
func partitionsAdded(from map[dax.TableKey]dax.VersionedPartitions, to map[dax.TableKey]dax.VersionedPartitions) map[dax.TableKey]dax.VersionedPartitions {
if from == nil {
return to
}
added := make(map[dax.TableKey]dax.Partitions)
added := make(map[dax.TableKey]dax.VersionedPartitions)
for tt, tps := range to {
fps, found := from[tt]
if !found {
@ -743,7 +743,7 @@ func partitionsAdded(from map[dax.TableKey]dax.Partitions, to map[dax.TableKey]d
continue
}
addedPartitions := dax.Partitions{}
addedPartitions := dax.VersionedPartitions{}
for i := range tps {
var found bool
for j := range fps {
@ -767,11 +767,11 @@ func partitionsAdded(from map[dax.TableKey]dax.Partitions, to map[dax.TableKey]d
// fieldsComparer is used to compare the differences between two maps of
// table:[]fieldVersion.
type fieldsComparer struct {
from map[dax.TableKey]dax.FieldVersions
to map[dax.TableKey]dax.FieldVersions
from map[dax.TableKey]dax.VersionedFields
to map[dax.TableKey]dax.VersionedFields
}
func newFieldsComparer(from map[dax.TableKey]dax.FieldVersions, to map[dax.TableKey]dax.FieldVersions) *fieldsComparer {
func newFieldsComparer(from map[dax.TableKey]dax.VersionedFields, to map[dax.TableKey]dax.VersionedFields) *fieldsComparer {
return &fieldsComparer{
from: from,
to: to,
@ -780,23 +780,23 @@ func newFieldsComparer(from map[dax.TableKey]dax.FieldVersions, to map[dax.Table
// added returns the fields which are present in `to` but not in `from`. The
// results remain in the format of a map of table:[]field.
func (f *fieldsComparer) added() map[dax.TableKey]dax.FieldVersions {
func (f *fieldsComparer) added() map[dax.TableKey]dax.VersionedFields {
return fieldsAdded(f.from, f.to)
}
// removed returns the fields which are present in `from` but not in `to`.
// The results remain in the format of a map of table:[]field.
func (f *fieldsComparer) removed() map[dax.TableKey]dax.FieldVersions {
func (f *fieldsComparer) removed() map[dax.TableKey]dax.VersionedFields {
return fieldsAdded(f.to, f.from)
}
// fieldsAdded returns the fields which are present in `to` but not in `from`.
func fieldsAdded(from map[dax.TableKey]dax.FieldVersions, to map[dax.TableKey]dax.FieldVersions) map[dax.TableKey]dax.FieldVersions {
func fieldsAdded(from map[dax.TableKey]dax.VersionedFields, to map[dax.TableKey]dax.VersionedFields) map[dax.TableKey]dax.VersionedFields {
if from == nil {
return to
}
added := make(map[dax.TableKey]dax.FieldVersions)
added := make(map[dax.TableKey]dax.VersionedFields)
for tt, tps := range to {
fps, found := from[tt]
if !found {
@ -804,7 +804,7 @@ func fieldsAdded(from map[dax.TableKey]dax.FieldVersions, to map[dax.TableKey]da
continue
}
addedFieldVersions := dax.FieldVersions{}
addedFieldVersions := dax.VersionedFields{}
for i := range tps {
var found bool
for j := range fps {
@ -828,11 +828,11 @@ func fieldsAdded(from map[dax.TableKey]dax.FieldVersions, to map[dax.TableKey]da
// shardsComparer is used to compare the differences between two maps of
// table:[]shardV.
type shardsComparer struct {
from map[dax.TableKey]dax.Shards
to map[dax.TableKey]dax.Shards
from map[dax.TableKey]dax.VersionedShards
to map[dax.TableKey]dax.VersionedShards
}
func newShardsComparer(from map[dax.TableKey]dax.Shards, to map[dax.TableKey]dax.Shards) *shardsComparer {
func newShardsComparer(from map[dax.TableKey]dax.VersionedShards, to map[dax.TableKey]dax.VersionedShards) *shardsComparer {
return &shardsComparer{
from: from,
to: to,
@ -841,23 +841,23 @@ func newShardsComparer(from map[dax.TableKey]dax.Shards, to map[dax.TableKey]dax
// added returns the shards which are present in `to` but not in `from`. The
// results remain in the format of a map of table:[]shard.
func (s *shardsComparer) added() map[dax.TableKey]dax.Shards {
func (s *shardsComparer) added() map[dax.TableKey]dax.VersionedShards {
return shardsAdded(s.from, s.to)
}
// removed returns the shards which are present in `from` but not in `to`. The
// results remain in the format of a map of table:[]shard.
func (s *shardsComparer) removed() map[dax.TableKey]dax.Shards {
func (s *shardsComparer) removed() map[dax.TableKey]dax.VersionedShards {
return shardsAdded(s.to, s.from)
}
// shardsAdded returns the shards which are present in `to` but not in `from`.
func shardsAdded(from map[dax.TableKey]dax.Shards, to map[dax.TableKey]dax.Shards) map[dax.TableKey]dax.Shards {
func shardsAdded(from map[dax.TableKey]dax.VersionedShards, to map[dax.TableKey]dax.VersionedShards) map[dax.TableKey]dax.VersionedShards {
if from == nil {
return to
}
added := make(map[dax.TableKey]dax.Shards)
added := make(map[dax.TableKey]dax.VersionedShards)
for tt, tss := range to {
fss, found := from[tt]
if !found {
@ -865,7 +865,7 @@ func shardsAdded(from map[dax.TableKey]dax.Shards, to map[dax.TableKey]dax.Shard
continue
}
addedShards := dax.Shards{}
addedShards := dax.VersionedShards{}
for i := range tss {
var found bool
for j := range fss {
@ -888,7 +888,7 @@ func shardsAdded(from map[dax.TableKey]dax.Shards, to map[dax.TableKey]dax.Shard
// createTableAndFields creates the FeatureBase Tables and Fields provided in
// the dax.Directive format.
func (api *API) createTableAndFields(tbl *dax.QualifiedTable, partitions dax.Partitions) error {
func (api *API) createTableAndFields(tbl *dax.QualifiedTable, partitions dax.VersionedPartitions) error {
cim := &CreateIndexMessage{
Index: string(tbl.Key()),
CreatedAt: 0,

2
cluster.go
View file

@ -993,7 +993,7 @@ type TransactionMessage struct {
Action string
}
func intInPartitions(i int, s dax.Partitions) bool {
func intInPartitions(i int, s dax.VersionedPartitions) bool {
for _, a := range s {
if int(a.Num) == i {
return true

38
dax/boltdb/versionstore.go
View file

@ -99,7 +99,7 @@ func (s *VersionStore) AddTable(ctx context.Context, qtid dax.QualifiedTableID)
return tx.Commit()
}
func (s *VersionStore) RemoveTable(ctx context.Context, qtid dax.QualifiedTableID) (dax.Shards, dax.Partitions, error) {
func (s *VersionStore) RemoveTable(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedShards, dax.VersionedPartitions, error) {
tx, err := s.db.BeginTx(ctx, true)
if err != nil {
return nil, nil, err
@ -189,7 +189,7 @@ func deleteByPrefix(tx *Tx, bucket Bucket, prefix []byte) error {
return nil
}
func (s *VersionStore) AddShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.Shard) error {
func (s *VersionStore) AddShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.VersionedShard) error {
tx, err := s.db.BeginTx(ctx, true)
if err != nil {
return errors.Wrap(err, "getting transaction")
@ -205,7 +205,7 @@ func (s *VersionStore) AddShards(ctx context.Context, qtid dax.QualifiedTableID,
return tx.Commit()
}
func createShard(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, shard dax.Shard) error {
func createShard(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, shard dax.VersionedShard) error {
// TODO: validate data more formally
if shard.Version < 0 {
return errors.New(errors.ErrUncoded, fmt.Sprintf("invalid shard version: %d", shard.Version))
@ -227,7 +227,7 @@ func createShard(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, shard d
return bkt.Put(shardKey(qtid, shard.Num), vsn)
}
func (s *VersionStore) Shards(ctx context.Context, qtid dax.QualifiedTableID) (dax.Shards, bool, error) {
func (s *VersionStore) Shards(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedShards, bool, error) {
tx, err := s.db.BeginTx(ctx, false)
if err != nil {
return nil, false, errors.Wrap(err, "getting tx")
@ -242,11 +242,11 @@ func (s *VersionStore) Shards(ctx context.Context, qtid dax.QualifiedTableID) (d
return shards, true, nil
}
func (s *VersionStore) getShards(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID) (dax.Shards, error) {
func (s *VersionStore) getShards(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID) (dax.VersionedShards, error) {
c := tx.Bucket(bucketShards).Cursor()
// Deserialize rows into Shard objects.
shards := make(dax.Shards, 0)
shards := make(dax.VersionedShards, 0)
prefix := []byte(fmt.Sprintf(prefixFmtShards, qtid.OrganizationID, qtid.DatabaseID, qtid.ID))
for k, v := c.Seek(prefix); k != nil && bytes.HasPrefix(k, prefix); k, v = c.Next() {
@ -255,7 +255,7 @@ func (s *VersionStore) getShards(ctx context.Context, tx *Tx, qtid dax.Qualified
continue
}
var shard dax.Shard
var shard dax.VersionedShard
shardNum, err := keyShardNum(k)
if err != nil {
@ -368,7 +368,7 @@ func (s *VersionStore) bucketTables(ctx context.Context, bucket Bucket) ([]dax.Q
// AddPartitions adds new partitions to be managed by VersionStore. It returns
// the number of partitions added or an error.
func (s *VersionStore) AddPartitions(ctx context.Context, qtid dax.QualifiedTableID, partitions ...dax.Partition) error {
func (s *VersionStore) AddPartitions(ctx context.Context, qtid dax.QualifiedTableID, partitions ...dax.VersionedPartition) error {
tx, err := s.db.BeginTx(ctx, true)
if err != nil {
return errors.Wrap(err, "getting transaction")
@ -384,7 +384,7 @@ func (s *VersionStore) AddPartitions(ctx context.Context, qtid dax.QualifiedTabl
return tx.Commit()
}
func createPartition(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, partition dax.Partition) error {
func createPartition(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, partition dax.VersionedPartition) error {
// TODO: validate data more formally
if partition.Version < 0 {
return errors.New(errors.ErrUncoded, fmt.Sprintf("invalid partition version: %d", partition.Version))
@ -406,7 +406,7 @@ func createPartition(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, par
return bkt.Put(partitionKey(qtid, partition.Num), vsn)
}
func (s *VersionStore) Partitions(ctx context.Context, qtid dax.QualifiedTableID) (dax.Partitions, bool, error) {
func (s *VersionStore) Partitions(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedPartitions, bool, error) {
tx, err := s.db.BeginTx(ctx, false)
if err != nil {
return nil, false, errors.Wrap(err, "getting tx")
@ -421,11 +421,11 @@ func (s *VersionStore) Partitions(ctx context.Context, qtid dax.QualifiedTableID
return partitions, true, nil
}
func (s *VersionStore) getPartitions(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID) (dax.Partitions, error) {
func (s *VersionStore) getPartitions(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID) (dax.VersionedPartitions, error) {
c := tx.Bucket(bucketTableKeys).Cursor()
// Deserialize rows into Partition objects.
partitions := make(dax.Partitions, 0)
partitions := make(dax.VersionedPartitions, 0)
prefix := []byte(fmt.Sprintf(prefixFmtTableKeys, qtid.OrganizationID, qtid.DatabaseID, qtid.ID))
for k, v := c.Seek(prefix); k != nil && bytes.HasPrefix(k, prefix); k, v = c.Next() {
@ -434,7 +434,7 @@ func (s *VersionStore) getPartitions(ctx context.Context, tx *Tx, qtid dax.Quali
continue
}
var partition dax.Partition
var partition dax.VersionedPartition
partitionNum, err := keyPartitionNum(k)
if err != nil {
@ -489,7 +489,7 @@ func (s *VersionStore) PartitionTables(ctx context.Context, qual dax.TableQualif
// AddFields adds new fields to be managed by VersionStore. It returns the
// number of fields added or an error.
func (s *VersionStore) AddFields(ctx context.Context, qtid dax.QualifiedTableID, fields ...dax.FieldVersion) error {
func (s *VersionStore) AddFields(ctx context.Context, qtid dax.QualifiedTableID, fields ...dax.VersionedField) error {
tx, err := s.db.BeginTx(ctx, true)
if err != nil {
return err
@ -505,7 +505,7 @@ func (s *VersionStore) AddFields(ctx context.Context, qtid dax.QualifiedTableID,
return tx.Commit()
}
func createFieldVersion(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, field dax.FieldVersion) error {
func createFieldVersion(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID, field dax.VersionedField) error {
// TODO: validate data more formally
if field.Version < 0 {
return errors.New(errors.ErrUncoded, fmt.Sprintf("invalid field version: %d", field.Version))
@ -527,7 +527,7 @@ func createFieldVersion(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID,
return bkt.Put(fieldKey(qtid, field.Name), vsn)
}
func (s *VersionStore) Fields(ctx context.Context, qtid dax.QualifiedTableID) (dax.FieldVersions, bool, error) {
func (s *VersionStore) Fields(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedFields, bool, error) {
tx, err := s.db.BeginTx(ctx, false)
if err != nil {
return nil, false, errors.Wrap(err, "getting tx")
@ -542,11 +542,11 @@ func (s *VersionStore) Fields(ctx context.Context, qtid dax.QualifiedTableID) (d
return fields, true, nil
}
func (s *VersionStore) getFields(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID) (dax.FieldVersions, error) {
func (s *VersionStore) getFields(ctx context.Context, tx *Tx, qtid dax.QualifiedTableID) (dax.VersionedFields, error) {
c := tx.Bucket(bucketFieldKeys).Cursor()
// Deserialize rows into FieldVersion objects.
fieldVersions := make(dax.FieldVersions, 0)
fieldVersions := make(dax.VersionedFields, 0)
prefix := []byte(fmt.Sprintf(prefixFmtFieldKeys, qtid.OrganizationID, qtid.DatabaseID, qtid.ID))
for k, v := c.Seek(prefix); k != nil && bytes.HasPrefix(k, prefix); k, v = c.Next() {
@ -555,7 +555,7 @@ func (s *VersionStore) getFields(ctx context.Context, tx *Tx, qtid dax.Qualified
continue
}
var fieldVersion dax.FieldVersion
var fieldVersion dax.VersionedField
fieldName, err := keyFieldName(k)
if err != nil {

38
dax/boltdb/versionstore_test.go
View file

@ -69,9 +69,9 @@ func TestVersionStore(t *testing.T) {
defer vs.RemoveTable(ctx, qtid3)
// Create some shards to insert into the table.
shards := make(dax.Shards, 3)
shards := make(dax.VersionedShards, 3)
for i := range shards {
shards[i] = dax.Shard{
shards[i] = dax.VersionedShard{
Num: dax.ShardNum(i),
Version: i * 2,
}
@ -118,7 +118,7 @@ func TestVersionStore(t *testing.T) {
shards, partitions, err := vs.RemoveTable(ctx, qtid1)
assert.NoError(t, err)
assert.Equal(t, shards, shards)
assert.Equal(t, dax.Partitions{}, partitions)
assert.Equal(t, dax.VersionedPartitions{}, partitions)
}
// Fetch all shards and compare.
@ -126,7 +126,7 @@ func TestVersionStore(t *testing.T) {
shrds, found, err := vs.Shards(ctx, qtid1)
assert.NoError(t, err)
assert.True(t, found)
assert.Equal(t, dax.Shards{}, shrds)
assert.Equal(t, dax.VersionedShards{}, shrds)
}
})
@ -134,9 +134,9 @@ func TestVersionStore(t *testing.T) {
vs := boltdb.NewVersionStore(db, logger.NopLogger)
// Create some partitions to insert into the table.
partitions := make(dax.Partitions, 3)
partitions := make(dax.VersionedPartitions, 3)
for i := range partitions {
partitions[i] = dax.Partition{
partitions[i] = dax.VersionedPartition{
Num: dax.PartitionNum(i),
Version: i * 2,
}
@ -192,7 +192,7 @@ func TestVersionStore(t *testing.T) {
{
shards, partitions, err := vs.RemoveTable(ctx, qtid1)
assert.NoError(t, err)
assert.Equal(t, dax.Shards{}, shards)
assert.Equal(t, dax.VersionedShards{}, shards)
assert.Equal(t, partitions, partitions)
}
@ -201,7 +201,7 @@ func TestVersionStore(t *testing.T) {
parts, found, err := vs.Partitions(ctx, qtid1)
assert.NoError(t, err)
assert.True(t, found)
assert.Equal(t, dax.Partitions{}, parts)
assert.Equal(t, dax.VersionedPartitions{}, parts)
}
})
@ -219,9 +219,9 @@ func TestVersionStore(t *testing.T) {
defer vs.RemoveTable(ctx, qtid2)
// Create some fieldVersions to insert into the table.
fieldVersions := make(dax.FieldVersions, 3)
fieldVersions := make(dax.VersionedFields, 3)
for i := range fieldVersions {
fieldVersions[i] = dax.FieldVersion{
fieldVersions[i] = dax.VersionedField{
Name: dax.FieldName(fmt.Sprintf("fld-%d", i)),
Version: i * 2,
}
@ -267,8 +267,8 @@ func TestVersionStore(t *testing.T) {
{
shards, partitions, err := vs.RemoveTable(ctx, qtid1)
assert.NoError(t, err)
assert.Equal(t, dax.Shards{}, shards)
assert.Equal(t, dax.Partitions{}, partitions)
assert.Equal(t, dax.VersionedShards{}, shards)
assert.Equal(t, dax.VersionedPartitions{}, partitions)
}
// Fetch all fieldVersions and compare.
@ -276,7 +276,7 @@ func TestVersionStore(t *testing.T) {
flds, found, err := vs.Fields(ctx, qtid1)
assert.NoError(t, err)
assert.True(t, found)
assert.Equal(t, dax.FieldVersions{}, flds)
assert.Equal(t, dax.VersionedFields{}, flds)
}
})
@ -291,27 +291,27 @@ func TestVersionStore(t *testing.T) {
defer vs.RemoveTable(ctx, qtid1)
// Create some shards to insert into the table.
shards := make(dax.Shards, 3)
shards := make(dax.VersionedShards, 3)
for i := range shards {
shards[i] = dax.Shard{
shards[i] = dax.VersionedShard{
Num: dax.ShardNum(i),
Version: i * 2,
}
}
// Create some partitions to insert into the table.
partitions := make(dax.Partitions, 3)
partitions := make(dax.VersionedPartitions, 3)
for i := range partitions {
partitions[i] = dax.Partition{
partitions[i] = dax.VersionedPartition{
Num: dax.PartitionNum(i),
Version: i * 2,
}
}
// Create some fieldVersions to insert into the table.
fieldVersions := make(dax.FieldVersions, 3)
fieldVersions := make(dax.VersionedFields, 3)
for i := range fieldVersions {
fieldVersions[i] = dax.FieldVersion{
fieldVersions[i] = dax.VersionedField{
Name: dax.FieldName(fmt.Sprintf("fld-%d", i)),
Version: i * 2,
}

24
dax/directive.go
View file

@ -57,9 +57,9 @@ func (d *Directive) Table(qtid QualifiedTableID) (*QualifiedTable, error) {
// compute node is responsible. It assumes that the Directive does not contain
// more than one ComputeRole for the same table; in that case, we would need to
// return the union of Shards.
func (d *Directive) ComputeShards(tbl TableKey) Shards {
func (d *Directive) ComputeShards(tbl TableKey) VersionedShards {
if d == nil || d.ComputeRoles == nil {
return Shards{}
return VersionedShards{}
}
for _, cr := range d.ComputeRoles {
@ -68,14 +68,14 @@ func (d *Directive) ComputeShards(tbl TableKey) Shards {
}
}
return Shards{}
return VersionedShards{}
}
// ComputeShardsMap returns a map of table to shards. It assumes that the
// Directive does not contain more than one ComputeRole for the same table; in
// that case, we would need to return the union of Shards.
func (d *Directive) ComputeShardsMap() map[TableKey]Shards {
m := make(map[TableKey]Shards)
func (d *Directive) ComputeShardsMap() map[TableKey]VersionedShards {
m := make(map[TableKey]VersionedShards)
if d == nil || d.ComputeRoles == nil {
return m
}
@ -91,9 +91,9 @@ func (d *Directive) ComputeShardsMap() map[TableKey]Shards {
// which this translate node is responsible. It assumes that the Directive does
// not contain more than one TranslateRole for the same table; in that case, we
// would need to return the union of Shards.
func (d *Directive) TranslatePartitions(tbl TableKey) Partitions {
func (d *Directive) TranslatePartitions(tbl TableKey) VersionedPartitions {
if d == nil || d.TranslateRoles == nil {
return Partitions{}
return VersionedPartitions{}
}
for _, tr := range d.TranslateRoles {
@ -101,14 +101,14 @@ func (d *Directive) TranslatePartitions(tbl TableKey) Partitions {
return tr.Partitions
}
}
return Partitions{}
return VersionedPartitions{}
}
// TranslatePartitionsMap returns a map of table to partitions. It assumes that
// the Directive does not contain more than one TranslateRole for the same
// table; in that case, we would need to return the union of Partitions.
func (d *Directive) TranslatePartitionsMap() map[TableKey]Partitions {
m := make(map[TableKey]Partitions)
func (d *Directive) TranslatePartitionsMap() map[TableKey]VersionedPartitions {
m := make(map[TableKey]VersionedPartitions)
if d == nil || d.TranslateRoles == nil {
return m
}
@ -129,8 +129,8 @@ func (d *Directive) TranslatePartitionsMap() map[TableKey]Partitions {
// TranslateFieldsMap returns a map of table to fields. It assumes that
// the Directive does not contain more than one TranslateRole for the same
// table; in that case, we would need to return the union of FieldValues.
func (d *Directive) TranslateFieldsMap() map[TableKey]FieldVersions {
m := make(map[TableKey]FieldVersions)
func (d *Directive) TranslateFieldsMap() map[TableKey]VersionedFields {
m := make(map[TableKey]VersionedFields)
if d == nil || d.TranslateRoles == nil {
return m
}

32
dax/fieldversion.go
View file

@ -1,32 +0,0 @@
package dax
import "fmt"
// FieldVersion is used in a similar way to Shard and Partition in that they all
// contain a snapshot version. It would have been confusing to use the Field
// type which already exists, because versioning that would mean something else.
// This is really snapshot specific (as are Shard and Partition).
type FieldVersion struct {
Name FieldName `json:"name"`
Version int `json:"version"`
}
// String returns the FieldVersion (i.e. its Name and Version) as a string.
func (f FieldVersion) String() string {
return fmt.Sprintf("%s.%d", f.Name, f.Version)
}
// NewFieldVersion returns a FieldVersion with the provided name and version.
func NewFieldVersion(name FieldName, version int) FieldVersion {
return FieldVersion{
Name: name,
Version: version,
}
}
// FieldVersions is a sortable slice of FieldVersion.
type FieldVersions []FieldVersion
func (f FieldVersions) Len() int { return len(f) }
func (f FieldVersions) Less(i, j int) bool { return f[i].Name < f[j].Name }
func (f FieldVersions) Swap(i, j int) { f[i], f[j] = f[j], f[i] }

44
dax/inmem/versionstore.go
View file

@ -18,7 +18,7 @@ type VersionStore struct {
// shards is a map of all shards, by table, by shard number, known to
// contain data.
shards map[dax.TableQualifierKey]map[dax.TableID]map[dax.ShardNum]dax.Shard
shards map[dax.TableQualifierKey]map[dax.TableID]map[dax.ShardNum]dax.VersionedShard
// tableKeys is a map of all partitions, by table, by partition number,
// known to contain key data.
@ -31,7 +31,7 @@ type VersionStore struct {
// NewVersionStore returns a new instance of VersionStore with default values.
func NewVersionStore() *VersionStore {
return &VersionStore{
shards: make(map[dax.TableQualifierKey]map[dax.TableID]map[dax.ShardNum]dax.Shard),
shards: make(map[dax.TableQualifierKey]map[dax.TableID]map[dax.ShardNum]dax.VersionedShard),
tableKeys: make(map[dax.TableQualifierKey]map[dax.TableID]map[dax.PartitionNum]int),
fieldKeys: make(map[dax.TableQualifierKey]map[dax.TableID]map[dax.FieldName]int),
}
@ -52,7 +52,7 @@ func (s *VersionStore) AddTable(ctx context.Context, qtid dax.QualifiedTableID)
// Initialize the maps in case VersionStore wasn't created with NewVersionStore().
if s.shards == nil {
s.shards = make(map[dax.TableQualifierKey]map[dax.TableID]map[dax.ShardNum]dax.Shard)
s.shards = make(map[dax.TableQualifierKey]map[dax.TableID]map[dax.ShardNum]dax.VersionedShard)
}
if s.tableKeys == nil {
s.tableKeys = make(map[dax.TableQualifierKey]map[dax.TableID]map[dax.PartitionNum]int)
@ -63,10 +63,10 @@ func (s *VersionStore) AddTable(ctx context.Context, qtid dax.QualifiedTableID)
// shards.
if _, ok := s.shards[qtid.TableQualifier.Key()]; !ok {
s.shards[qtid.TableQualifier.Key()] = make(map[dax.TableID]map[dax.ShardNum]dax.Shard, 0)
s.shards[qtid.TableQualifier.Key()] = make(map[dax.TableID]map[dax.ShardNum]dax.VersionedShard, 0)
}
if _, ok := s.shards[qtid.TableQualifier.Key()][qtid.ID]; !ok {
s.shards[qtid.TableQualifier.Key()][qtid.ID] = make(map[dax.ShardNum]dax.Shard, 0)
s.shards[qtid.TableQualifier.Key()][qtid.ID] = make(map[dax.ShardNum]dax.VersionedShard, 0)
}
// tableKeys.
@ -90,13 +90,13 @@ func (s *VersionStore) AddTable(ctx context.Context, qtid dax.QualifiedTableID)
// RemoveTable removes the given table. An error will be returned if the table
// does not exist.
func (s *VersionStore) RemoveTable(ctx context.Context, qtid dax.QualifiedTableID) (dax.Shards, dax.Partitions, error) {
func (s *VersionStore) RemoveTable(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedShards, dax.VersionedPartitions, error) {
s.mu.Lock()
defer s.mu.Unlock()
var foundTable bool
var shards dax.Shards
var partitions dax.Partitions
var shards dax.VersionedShards
var partitions dax.VersionedPartitions
var err error
// Remove shards for table.
@ -150,7 +150,7 @@ func (s *VersionStore) RemoveTable(ctx context.Context, qtid dax.QualifiedTableI
// AddShards adds new shards to be managed by VersionStore. It returns the
// number of shards added or an error.
func (s *VersionStore) AddShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.Shard) error {
func (s *VersionStore) AddShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.VersionedShard) error {
s.mu.Lock()
defer s.mu.Unlock()
@ -173,7 +173,7 @@ func (s *VersionStore) AddShards(ctx context.Context, qtid dax.QualifiedTableID,
// Shards returns the list of shards available for the give table. It returns
// false if the table does not exist.
func (s *VersionStore) Shards(ctx context.Context, qtid dax.QualifiedTableID) (dax.Shards, bool, error) {
func (s *VersionStore) Shards(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedShards, bool, error) {
s.mu.RLock()
defer s.mu.RUnlock()
@ -181,13 +181,13 @@ func (s *VersionStore) Shards(ctx context.Context, qtid dax.QualifiedTableID) (d
}
// shardSlice is an unprotected version of Shards().
func (s *VersionStore) shardSlice(qtid dax.QualifiedTableID) (dax.Shards, bool, error) {
func (s *VersionStore) shardSlice(qtid dax.QualifiedTableID) (dax.VersionedShards, bool, error) {
if s.shards == nil {
return nil, false, nil
}
if shardNumMap, ok := s.shards[qtid.TableQualifier.Key()][qtid.ID]; ok {
rtn := make(dax.Shards, 0, len(shardNumMap))
rtn := make(dax.VersionedShards, 0, len(shardNumMap))
for _, shard := range shardNumMap {
rtn = append(rtn, shard)
}
@ -233,7 +233,7 @@ func (s *VersionStore) ShardTables(ctx context.Context, qual dax.TableQualifier)
// AddPartitions adds new partitions to be managed by VersionStore. It returns
// the number of partitions added or an error.
func (s *VersionStore) AddPartitions(ctx context.Context, qtid dax.QualifiedTableID, partitions ...dax.Partition) error {
func (s *VersionStore) AddPartitions(ctx context.Context, qtid dax.QualifiedTableID, partitions ...dax.VersionedPartition) error {
s.mu.Lock()
defer s.mu.Unlock()
@ -251,7 +251,7 @@ func (s *VersionStore) AddPartitions(ctx context.Context, qtid dax.QualifiedTabl
// Partitions returns the list of partitions available for the give table. It
// returns false if the table does not exist.
func (s *VersionStore) Partitions(ctx context.Context, qtid dax.QualifiedTableID) (dax.Partitions, bool, error) {
func (s *VersionStore) Partitions(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedPartitions, bool, error) {
s.mu.RLock()
defer s.mu.RUnlock()
@ -259,15 +259,15 @@ func (s *VersionStore) Partitions(ctx context.Context, qtid dax.QualifiedTableID
}
// partitionSlice is an unprotected version of Partitions().
func (s *VersionStore) partitionSlice(qtid dax.QualifiedTableID) (dax.Partitions, bool, error) {
func (s *VersionStore) partitionSlice(qtid dax.QualifiedTableID) (dax.VersionedPartitions, bool, error) {
if s.tableKeys == nil {
return nil, false, nil
}
if partitionNumMap, ok := s.tableKeys[qtid.TableQualifier.Key()][qtid.ID]; ok {
rtn := make(dax.Partitions, 0, len(partitionNumMap))
rtn := make(dax.VersionedPartitions, 0, len(partitionNumMap))
for partitionNum, version := range partitionNumMap {
rtn = append(rtn, dax.NewPartition(partitionNum, version))
rtn = append(rtn, dax.NewVersionedPartition(partitionNum, version))
}
sort.Sort(rtn)
return rtn, true, nil
@ -309,7 +309,7 @@ func (s *VersionStore) PartitionTables(ctx context.Context, qual dax.TableQualif
// AddFields adds new fields to be managed by VersionStore. It returns the
// number of fields added or an error.
func (s *VersionStore) AddFields(ctx context.Context, qtid dax.QualifiedTableID, fields ...dax.FieldVersion) error {
func (s *VersionStore) AddFields(ctx context.Context, qtid dax.QualifiedTableID, fields ...dax.VersionedField) error {
s.mu.Lock()
defer s.mu.Unlock()
@ -327,7 +327,7 @@ func (s *VersionStore) AddFields(ctx context.Context, qtid dax.QualifiedTableID,
// Fields returns the list of fields available for the give table. It returns
// false if the table does not exist.
func (s *VersionStore) Fields(ctx context.Context, qtid dax.QualifiedTableID) (dax.FieldVersions, bool, error) {
func (s *VersionStore) Fields(ctx context.Context, qtid dax.QualifiedTableID) (dax.VersionedFields, bool, error) {
s.mu.RLock()
defer s.mu.RUnlock()
@ -335,15 +335,15 @@ func (s *VersionStore) Fields(ctx context.Context, qtid dax.QualifiedTableID) (d
}
// fieldSlice is an unprotected version of Fields().
func (s *VersionStore) fieldSlice(qtid dax.QualifiedTableID) (dax.FieldVersions, bool, error) {
func (s *VersionStore) fieldSlice(qtid dax.QualifiedTableID) (dax.VersionedFields, bool, error) {
if s.fieldKeys == nil {
return nil, false, nil
}
if fieldNameMap, ok := s.fieldKeys[qtid.TableQualifier.Key()][qtid.ID]; ok {
rtn := make(dax.FieldVersions, 0, len(fieldNameMap))
rtn := make(dax.VersionedFields, 0, len(fieldNameMap))
for fieldName, version := range fieldNameMap {
rtn = append(rtn, dax.NewFieldVersion(fieldName, version))
rtn = append(rtn, dax.NewVersionedField(fieldName, version))
}
sort.Sort(rtn)
return rtn, true, nil

54
dax/inmem/versionstore_test.go
View file

@ -53,7 +53,7 @@ func TestVersionStore(t *testing.T) {
})
t.Run("AddShards", func(t *testing.T) {
err := s.AddShards(ctx, invalidQtid, dax.NewShard(1, 0))
err := s.AddShards(ctx, invalidQtid, dax.NewVersionedShard(1, 0))
if assert.Error(t, err) {
assert.True(t, errors.Is(err, dax.ErrTableIDDoesNotExist))
}
@ -69,16 +69,16 @@ func TestVersionStore(t *testing.T) {
sh, ok, err := s.Shards(ctx, qtid)
assert.NoError(t, err)
assert.True(t, ok)
assert.Equal(t, sh, dax.Shards{})
assert.Equal(t, sh, dax.VersionedShards{})
}
// Add the first set of shards (with a duplicate (8)).
{
err := s.AddShards(ctx, qtid,
dax.NewShard(8, 0),
dax.NewShard(9, 0),
dax.NewShard(8, 0),
dax.NewShard(10, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(9, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(10, 0),
)
assert.NoError(t, err)
}
@ -87,10 +87,10 @@ func TestVersionStore(t *testing.T) {
sh, ok, err := s.Shards(ctx, qtid)
assert.NoError(t, err)
assert.True(t, ok)
assert.Equal(t, dax.Shards{
dax.NewShard(8, 0),
dax.NewShard(9, 0),
dax.NewShard(10, 0),
assert.Equal(t, dax.VersionedShards{
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(9, 0),
dax.NewVersionedShard(10, 0),
}, sh)
}
@ -98,10 +98,10 @@ func TestVersionStore(t *testing.T) {
// existing (10)).
{
err := s.AddShards(ctx, qtid,
dax.NewShard(10, 0),
dax.NewShard(11, 0),
dax.NewShard(12, 0),
dax.NewShard(11, 0),
dax.NewVersionedShard(10, 0),
dax.NewVersionedShard(11, 0),
dax.NewVersionedShard(12, 0),
dax.NewVersionedShard(11, 0),
)
assert.NoError(t, err)
}
@ -110,12 +110,12 @@ func TestVersionStore(t *testing.T) {
sh, ok, err := s.Shards(ctx, qtid)
assert.NoError(t, err)
assert.True(t, ok)
assert.Equal(t, dax.Shards{
dax.NewShard(8, 0),
dax.NewShard(9, 0),
dax.NewShard(10, 0),
dax.NewShard(11, 0),
dax.NewShard(12, 0),
assert.Equal(t, dax.VersionedShards{
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(9, 0),
dax.NewVersionedShard(10, 0),
dax.NewVersionedShard(11, 0),
dax.NewVersionedShard(12, 0),
}, sh)
}
})
@ -123,13 +123,13 @@ func TestVersionStore(t *testing.T) {
t.Run("RemoveTable", func(t *testing.T) {
shards, partitions, err := s.RemoveTable(ctx, qtid)
assert.NoError(t, err)
assert.Equal(t, dax.Partitions{}, partitions)
assert.Equal(t, dax.Shards{
dax.NewShard(8, 0),
dax.NewShard(9, 0),
dax.NewShard(10, 0),
dax.NewShard(11, 0),
dax.NewShard(12, 0),
assert.Equal(t, dax.VersionedPartitions{}, partitions)
assert.Equal(t, dax.VersionedShards{
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(9, 0),
dax.NewVersionedShard(10, 0),
dax.NewVersionedShard(11, 0),
dax.NewVersionedShard(12, 0),
}, shards)
// Make sure the table was removed.

70
dax/mds/controller/controller.go
View file

@ -520,7 +520,7 @@ func (c *Controller) nodesTranslateReadOrWrite(ctx context.Context, role *dax.Tr
// Initialize the partition version to 0.
qtid := j.table().QualifiedTableID()
if err := c.versionStore.AddPartitions(ctx, qtid, dax.NewPartition(j.partitionNum(), 0)); err != nil {
if err := c.versionStore.AddPartitions(ctx, qtid, dax.NewVersionedPartition(j.partitionNum(), 0)); err != nil {
return nil, false, NewErrInternal(err.Error())
}
}
@ -543,7 +543,7 @@ func (c *Controller) nodesTranslateReadOrWrite(ctx context.Context, role *dax.Tr
for _, worker := range workers {
// covert worker.Jobs []string to map[string][]Partition
translateMap := make(map[dax.TableKey]dax.Partitions)
translateMap := make(map[dax.TableKey]dax.VersionedPartitions)
for _, job := range worker.Jobs {
j, err := decodePartition(job)
if err != nil {
@ -561,7 +561,7 @@ func (c *Controller) nodesTranslateReadOrWrite(ctx context.Context, role *dax.Tr
}
translateMap[tkey] = append(translateMap[tkey],
dax.NewPartition(j.partitionNum(), partitionVersion),
dax.NewVersionedPartition(j.partitionNum(), partitionVersion),
)
}
@ -707,7 +707,7 @@ func (c *Controller) nodesComputeReadOrWrite(ctx context.Context, role *dax.Comp
// Initialize the shard version to 0.
qtid := j.table().QualifiedTableID()
if err := c.versionStore.AddShards(ctx, qtid, dax.NewShard(j.shardNum(), 0)); err != nil {
if err := c.versionStore.AddShards(ctx, qtid, dax.NewVersionedShard(j.shardNum(), 0)); err != nil {
return nil, false, NewErrInternal(err.Error())
}
}
@ -736,7 +736,7 @@ func (c *Controller) workersToAssignedNodes(ctx context.Context, workers []dax.W
nodes := []dax.AssignedNode{}
for _, worker := range workers {
// convert worker.Jobs []string to map[TableName][]Shard
computeMap := make(map[dax.TableKey]dax.Shards)
computeMap := make(map[dax.TableKey]dax.VersionedShards)
for _, job := range worker.Jobs {
j, err := decodeShard(job)
if err != nil {
@ -754,7 +754,7 @@ func (c *Controller) workersToAssignedNodes(ctx context.Context, workers []dax.W
}
computeMap[tkey] = append(computeMap[tkey],
dax.NewShard(j.shardNum(), shardVersion),
dax.NewVersionedShard(j.shardNum(), shardVersion),
)
}
@ -788,13 +788,13 @@ func (c *Controller) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable)
}
// Add fields which have string keys to the local versionStore.
fieldVersions := make(dax.FieldVersions, 0)
fieldVersions := make(dax.VersionedFields, 0)
for _, field := range qtbl.Fields {
if !field.StringKeys() {
continue
}
fieldVersions = append(fieldVersions, dax.FieldVersion{
fieldVersions = append(fieldVersions, dax.VersionedField{
Name: field.Name,
Version: 0,
})
@ -812,9 +812,9 @@ func (c *Controller) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable)
workerSet := NewAddressSet()
// Generate the list of partitions to be added.
partitions := make(dax.Partitions, qtbl.PartitionN)
partitions := make(dax.VersionedPartitions, qtbl.PartitionN)
for partitionNum := 0; partitionNum < qtbl.PartitionN; partitionNum++ {
partitions[partitionNum] = dax.NewPartition(dax.PartitionNum(partitionNum), 0)
partitions[partitionNum] = dax.NewVersionedPartition(dax.PartitionNum(partitionNum), 0)
}
// Add partitions to versionStore. Version is intentionally set to 0
@ -855,7 +855,7 @@ func (c *Controller) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable)
// and therefore need to be sent an updated Directive.
workerSet := NewAddressSet()
p := dax.NewPartition(0, 0)
p := dax.NewVersionedPartition(0, 0)
// Add partition 0 to versionStore. Version is intentionally set to 0
// here as this is the initial instance of the partition.
@ -963,7 +963,7 @@ func (c *Controller) Tables(ctx context.Context, qual dax.TableQualifier, ids ..
// AddShards registers the table/shard combinations with the controller and
// sends the necessary directive.
func (c *Controller) AddShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.Shard) error {
func (c *Controller) AddShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.VersionedShard) error {
c.mu.Lock()
defer c.mu.Unlock()
@ -989,7 +989,7 @@ func (c *Controller) AddShards(ctx context.Context, qtid dax.QualifiedTableID, s
}
// Initialize the shard version to 0.
if err := c.versionStore.AddShards(ctx, qtid, dax.NewShard(s.Num, 0)); err != nil {
if err := c.versionStore.AddShards(ctx, qtid, dax.NewVersionedShard(s.Num, 0)); err != nil {
return NewErrInternal(err.Error())
}
}
@ -1007,7 +1007,7 @@ func (c *Controller) AddShards(ctx context.Context, qtid dax.QualifiedTableID, s
// RemoveShards deregisters the table/shard combinations with the controller and
// sends the necessary directives.
func (c *Controller) RemoveShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.Shard) error {
func (c *Controller) RemoveShards(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.VersionedShard) error {
c.mu.Lock()
defer c.mu.Unlock()
@ -1138,12 +1138,12 @@ func (c *Controller) buildDirectives(ctx context.Context, addrs []addressMethod,
// computeMap maps a table to a list of shards for that table. We need
// to aggregate them here because the list of jobs from WorkerState()
// can contain a mixture of table/shards.
computeMap := make(map[dax.TableKey][]dax.Shard)
computeMap := make(map[dax.TableKey][]dax.VersionedShard)
// translateMap maps a table to a list of partitions for that table. We
// need to aggregate them here because the list of jobs from
// WorkerState() can contain a mixture of table/partitions.
translateMap := make(map[dax.TableKey]dax.Partitions)
translateMap := make(map[dax.TableKey]dax.VersionedPartitions)
// tableSet maintains the set of tables which have a job assignment
// change and therefore need to be included in the Directive schema.
@ -1187,7 +1187,7 @@ func (c *Controller) buildDirectives(ctx context.Context, addrs []addressMethod,
}
computeMap[tkey] = append(computeMap[tkey],
dax.NewShard(j.shardNum(), shardVersion),
dax.NewVersionedShard(j.shardNum(), shardVersion),
)
tableSet.Add(tkey)
}
@ -1222,7 +1222,7 @@ func (c *Controller) buildDirectives(ctx context.Context, addrs []addressMethod,
}
translateMap[tkey] = append(translateMap[tkey],
dax.NewPartition(j.partitionNum(), partitionVersion),
dax.NewVersionedPartition(j.partitionNum(), partitionVersion),
)
tableSet.Add(tkey)
}
@ -1234,7 +1234,7 @@ func (c *Controller) buildDirectives(ctx context.Context, addrs []addressMethod,
// Because these were encoded as strings in the balancer and may be
// out of order numerically, sort them as integers.
//sort.Slice(v, func(i, j int) bool { return v[i] < v[j] })
sort.Sort(dax.Shards(v))
sort.Sort(dax.VersionedShards(v))
d.ComputeRoles = append(d.ComputeRoles, dax.ComputeRole{
TableKey: k,
@ -1268,7 +1268,7 @@ func (c *Controller) buildDirectives(ctx context.Context, addrs []addressMethod,
return nil, errors.Wrapf(err, "getting table: %s", tkey)
}
fieldVersions := make(dax.FieldVersions, 0)
fieldVersions := make(dax.VersionedFields, 0)
for _, field := range table.Fields {
if !field.StringKeys() {
continue
@ -1286,7 +1286,7 @@ func (c *Controller) buildDirectives(ctx context.Context, addrs []addressMethod,
return nil, NewErrInternal("field version not found in cache")
}
fieldVersions = append(fieldVersions, dax.FieldVersion{
fieldVersions = append(fieldVersions, dax.VersionedField{
Name: field.Name,
Version: fieldVersion,
})
@ -1414,7 +1414,7 @@ func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTa
// Get the node responsible for the shard.
bal := c.ComputeBalancer
job := shard(qtid.Key(), dax.NewShard(shardNum, -1))
job := shard(qtid.Key(), dax.NewVersionedShard(shardNum, -1))
workers, err := bal.WorkersForJobs(ctx, []dax.Job{dax.Job(job.String())})
if err != nil {
@ -1435,7 +1435,7 @@ func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTa
return errors.Wrap(err, "copying version store")
}
if err := versionStoreCopy.AddShards(ctx, qtid,
dax.NewShard(shardNum, toShardVersion),
dax.NewVersionedShard(shardNum, toShardVersion),
); err != nil {
return NewErrInternal(err.Error())
}
@ -1470,7 +1470,7 @@ func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTa
// A successful request means the shard version can be incremented.
if err := c.versionStore.AddShards(ctx, qtid,
dax.NewShard(shardNum, toShardVersion),
dax.NewVersionedShard(shardNum, toShardVersion),
); err != nil {
return NewErrInternal(err.Error())
}
@ -1496,7 +1496,7 @@ func (c *Controller) SnapshotTableKeys(ctx context.Context, qtid dax.QualifiedTa
// Get the node responsible for the partition.
bal := c.TranslateBalancer
job := partition(qtid.Key(), dax.NewPartition(partitionNum, -1))
job := partition(qtid.Key(), dax.NewVersionedPartition(partitionNum, -1))
workers, err := bal.WorkersForJobs(ctx, []dax.Job{dax.Job(job.String())})
if err != nil {
@ -1517,7 +1517,7 @@ func (c *Controller) SnapshotTableKeys(ctx context.Context, qtid dax.QualifiedTa
return errors.Wrap(err, "copying version store")
}
if err := versionStoreCopy.AddPartitions(ctx, qtid,
dax.NewPartition(partitionNum, toPartitionVersion),
dax.NewVersionedPartition(partitionNum, toPartitionVersion),
); err != nil {
return NewErrInternal(err.Error())
}
@ -1552,7 +1552,7 @@ func (c *Controller) SnapshotTableKeys(ctx context.Context, qtid dax.QualifiedTa
// A successful request means the partition version can be incremented.
if err := c.versionStore.AddPartitions(ctx, qtid,
dax.NewPartition(partitionNum, toPartitionVersion),
dax.NewVersionedPartition(partitionNum, toPartitionVersion),
); err != nil {
return NewErrInternal(err.Error())
}
@ -1580,7 +1580,7 @@ func (c *Controller) SnapshotFieldKeys(ctx context.Context, qtid dax.QualifiedTa
// Field translation is currently handled by partition 0.
partitionNum := dax.PartitionNum(0)
job := partition(qtid.Key(), dax.NewPartition(partitionNum, -1))
job := partition(qtid.Key(), dax.NewVersionedPartition(partitionNum, -1))
workers, err := bal.WorkersForJobs(ctx, []dax.Job{dax.Job(job.String())})
if err != nil {
@ -1601,7 +1601,7 @@ func (c *Controller) SnapshotFieldKeys(ctx context.Context, qtid dax.QualifiedTa
return errors.Wrap(err, "copying version store")
}
if err := versionStoreCopy.AddFields(ctx, qtid,
dax.NewFieldVersion(field, toFieldVersion),
dax.NewVersionedField(field, toFieldVersion),
); err != nil {
return NewErrInternal(err.Error())
}
@ -1636,7 +1636,7 @@ func (c *Controller) SnapshotFieldKeys(ctx context.Context, qtid dax.QualifiedTa
// A successful request means the field version can be incremented.
if err := c.versionStore.AddFields(ctx, qtid,
dax.NewFieldVersion(field, toFieldVersion),
dax.NewVersionedField(field, toFieldVersion),
); err != nil {
return NewErrInternal(err.Error())
}
@ -1649,7 +1649,7 @@ func (c *Controller) SnapshotFieldKeys(ctx context.Context, qtid dax.QualifiedTa
func (c *Controller) ComputeNodes(ctx context.Context, qtid dax.QualifiedTableID, shards dax.ShardNums, isWrite bool) ([]ComputeNode, error) {
inRole := &dax.ComputeRole{
TableKey: qtid.Key(),
Shards: dax.NewShards(shards...),
Shards: dax.NewVersionedShards(shards...),
}
nodes, err := c.Nodes(ctx, inRole, isWrite)
@ -1680,7 +1680,7 @@ func (c *Controller) ComputeNodes(ctx context.Context, qtid dax.QualifiedTableID
func (c *Controller) TranslateNodes(ctx context.Context, qtid dax.QualifiedTableID, partitions dax.PartitionNums, isWrite bool) ([]TranslateNode, error) {
inRole := &dax.TranslateRole{
TableKey: qtid.Key(),
Partitions: dax.NewPartitions(partitions...),
Partitions: dax.NewVersionedPartitions(partitions...),
}
nodes, err := c.Nodes(ctx, inRole, isWrite)
@ -1760,7 +1760,7 @@ func (c *Controller) CreateField(ctx context.Context, qtid dax.QualifiedTableID,
// If the field has string keys, add it to the local versionStore.
if fld.StringKeys() {
fieldVersion := dax.FieldVersion{
fieldVersion := dax.VersionedField{
Name: fld.Name,
Version: 0,
}
@ -1775,7 +1775,7 @@ func (c *Controller) CreateField(ctx context.Context, qtid dax.QualifiedTableID,
return errors.Wrapf(err, "getting partition version: %s/0", qtid)
} else if found {
// Get the worker(s) responsible for partition 0.
job := partition(qtid.Key(), dax.Partition{
job := partition(qtid.Key(), dax.VersionedPartition{
Num: 0,
Version: v,
}).String()
@ -1834,7 +1834,7 @@ func (c *Controller) DropField(ctx context.Context, qtid dax.QualifiedTableID, f
return errors.Wrapf(err, "getting partition version: %s/0", qtid)
} else if found {
// Get the worker(s) responsible for partition 0.
job := partition(qtid.Key(), dax.Partition{
job := partition(qtid.Key(), dax.VersionedPartition{
Num: 0,
Version: v,
}).String()

504
dax/mds/controller/controller_test.go
View file

@ -119,7 +119,7 @@ func TestController(t *testing.T) {
// Add a shard.
assert.NoError(t, con.AddShards(ctx, tbl0.QualifiedID(),
dax.NewShard(0, 0),
dax.NewVersionedShard(0, 0),
))
exp = []*dax.Directive{
@ -132,8 +132,8 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
},
},
},
@ -186,11 +186,11 @@ func TestController(t *testing.T) {
// Add more shards.
assert.NoError(t, con.AddShards(ctx, tbl0.QualifiedID(),
dax.NewShard(1, 0),
dax.NewShard(2, 0),
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
))
exp = []*dax.Directive{
@ -203,9 +203,9 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(3, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(3, 0),
},
},
},
@ -221,9 +221,9 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(1, 0),
dax.NewShard(5, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(5, 0),
},
},
},
@ -239,9 +239,9 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(2, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(8, 0),
},
},
},
@ -258,10 +258,10 @@ func TestController(t *testing.T) {
// Add more shards.
assert.NoError(t, con.AddShards(ctx, tbl1.QualifiedID(),
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
dax.NewShard(13, 0),
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(13, 0),
))
exp = []*dax.Directive{
@ -275,16 +275,16 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(3, 0),
dax.NewShard(13, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(13, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(3, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(3, 0),
},
},
},
@ -301,15 +301,15 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(5, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(5, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(1, 0),
dax.NewShard(5, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(5, 0),
},
},
},
@ -326,15 +326,15 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(8, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(2, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(8, 0),
},
},
},
@ -358,17 +358,17 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(3, 0),
dax.NewShard(13, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(13, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(1, 0),
dax.NewShard(3, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(3, 0),
},
},
},
@ -385,17 +385,17 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(5, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(2, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
},
},
},
@ -419,22 +419,22 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
dax.NewShard(13, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(13, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(1, 0),
dax.NewShard(2, 0),
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
},
},
},
@ -470,22 +470,22 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
dax.NewShard(13, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(13, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(1, 0),
dax.NewShard(2, 0),
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
},
},
},
@ -497,8 +497,8 @@ func TestController(t *testing.T) {
// Remove shards.
assert.NoError(t, con.RemoveShards(ctx, tbl0.QualifiedID(),
dax.NewShard(2, 0),
dax.NewShard(5, 0),
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(5, 0),
))
exp = []*dax.Directive{
@ -512,20 +512,20 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
dax.NewShard(13, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(13, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(1, 0),
dax.NewShard(3, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(8, 0),
},
},
},
@ -539,8 +539,8 @@ func TestController(t *testing.T) {
// Currently that doesn't result in an error, it simply no-ops on trying
// to remove 99.
assert.NoError(t, con.RemoveShards(ctx, tbl0.QualifiedID(),
dax.NewShard(3, 0),
dax.NewShard(99, 0),
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(99, 0),
))
exp = []*dax.Directive{
@ -554,19 +554,19 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
dax.NewShard(13, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(13, 0),
},
},
{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(1, 0),
dax.NewShard(8, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(8, 0),
},
},
},
@ -589,11 +589,11 @@ func TestController(t *testing.T) {
ComputeRoles: []dax.ComputeRole{
{
TableKey: tbl1.Key(),
Shards: dax.Shards{
dax.NewShard(3, 0),
dax.NewShard(5, 0),
dax.NewShard(8, 0),
dax.NewShard(13, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(5, 0),
dax.NewVersionedShard(8, 0),
dax.NewVersionedShard(13, 0),
},
},
},
@ -684,15 +684,15 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
dax.NewPartition(1, 0),
dax.NewPartition(2, 0),
dax.NewPartition(3, 0),
dax.NewPartition(4, 0),
dax.NewPartition(5, 0),
dax.NewPartition(6, 0),
dax.NewPartition(7, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
dax.NewVersionedPartition(1, 0),
dax.NewVersionedPartition(2, 0),
dax.NewVersionedPartition(3, 0),
dax.NewVersionedPartition(4, 0),
dax.NewVersionedPartition(5, 0),
dax.NewVersionedPartition(6, 0),
dax.NewVersionedPartition(7, 0),
},
},
},
@ -721,11 +721,11 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
dax.NewPartition(1, 0),
dax.NewPartition(2, 0),
dax.NewPartition(3, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
dax.NewVersionedPartition(1, 0),
dax.NewVersionedPartition(2, 0),
dax.NewVersionedPartition(3, 0),
},
},
},
@ -741,11 +741,11 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(4, 0),
dax.NewPartition(5, 0),
dax.NewPartition(6, 0),
dax.NewPartition(7, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(4, 0),
dax.NewVersionedPartition(5, 0),
dax.NewVersionedPartition(6, 0),
dax.NewVersionedPartition(7, 0),
},
},
},
@ -773,10 +773,10 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
dax.NewPartition(1, 0),
dax.NewPartition(2, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
dax.NewVersionedPartition(1, 0),
dax.NewVersionedPartition(2, 0),
},
},
},
@ -792,10 +792,10 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(4, 0),
dax.NewPartition(5, 0),
dax.NewPartition(6, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(4, 0),
dax.NewVersionedPartition(5, 0),
dax.NewVersionedPartition(6, 0),
},
},
},
@ -811,9 +811,9 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(3, 0),
dax.NewPartition(7, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(3, 0),
dax.NewVersionedPartition(7, 0),
},
},
},
@ -843,23 +843,23 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl1.Key(),
Partitions: dax.Partitions{
dax.NewPartition(1, 0),
dax.NewPartition(4, 0),
dax.NewPartition(7, 0),
dax.NewPartition(10, 0),
dax.NewPartition(13, 0),
dax.NewPartition(16, 0),
dax.NewPartition(19, 0),
dax.NewPartition(22, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(1, 0),
dax.NewVersionedPartition(4, 0),
dax.NewVersionedPartition(7, 0),
dax.NewVersionedPartition(10, 0),
dax.NewVersionedPartition(13, 0),
dax.NewVersionedPartition(16, 0),
dax.NewVersionedPartition(19, 0),
dax.NewVersionedPartition(22, 0),
},
},
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
dax.NewPartition(1, 0),
dax.NewPartition(2, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
dax.NewVersionedPartition(1, 0),
dax.NewVersionedPartition(2, 0),
},
},
},
@ -876,23 +876,23 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl1.Key(),
Partitions: dax.Partitions{
dax.NewPartition(2, 0),
dax.NewPartition(5, 0),
dax.NewPartition(8, 0),
dax.NewPartition(11, 0),
dax.NewPartition(14, 0),
dax.NewPartition(17, 0),
dax.NewPartition(20, 0),
dax.NewPartition(23, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(2, 0),
dax.NewVersionedPartition(5, 0),
dax.NewVersionedPartition(8, 0),
dax.NewVersionedPartition(11, 0),
dax.NewVersionedPartition(14, 0),
dax.NewVersionedPartition(17, 0),
dax.NewVersionedPartition(20, 0),
dax.NewVersionedPartition(23, 0),
},
},
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(4, 0),
dax.NewPartition(5, 0),
dax.NewPartition(6, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(4, 0),
dax.NewVersionedPartition(5, 0),
dax.NewVersionedPartition(6, 0),
},
},
},
@ -909,22 +909,22 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl1.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
dax.NewPartition(3, 0),
dax.NewPartition(6, 0),
dax.NewPartition(9, 0),
dax.NewPartition(12, 0),
dax.NewPartition(15, 0),
dax.NewPartition(18, 0),
dax.NewPartition(21, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
dax.NewVersionedPartition(3, 0),
dax.NewVersionedPartition(6, 0),
dax.NewVersionedPartition(9, 0),
dax.NewVersionedPartition(12, 0),
dax.NewVersionedPartition(15, 0),
dax.NewVersionedPartition(18, 0),
dax.NewVersionedPartition(21, 0),
},
},
{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(3, 0),
dax.NewPartition(7, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(3, 0),
dax.NewVersionedPartition(7, 0),
},
},
},
@ -948,15 +948,15 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl1.Key(),
Partitions: dax.Partitions{
dax.NewPartition(1, 0),
dax.NewPartition(4, 0),
dax.NewPartition(7, 0),
dax.NewPartition(10, 0),
dax.NewPartition(13, 0),
dax.NewPartition(16, 0),
dax.NewPartition(19, 0),
dax.NewPartition(22, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(1, 0),
dax.NewVersionedPartition(4, 0),
dax.NewVersionedPartition(7, 0),
dax.NewVersionedPartition(10, 0),
dax.NewVersionedPartition(13, 0),
dax.NewVersionedPartition(16, 0),
dax.NewVersionedPartition(19, 0),
dax.NewVersionedPartition(22, 0),
},
},
},
@ -972,15 +972,15 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl1.Key(),
Partitions: dax.Partitions{
dax.NewPartition(2, 0),
dax.NewPartition(5, 0),
dax.NewPartition(8, 0),
dax.NewPartition(11, 0),
dax.NewPartition(14, 0),
dax.NewPartition(17, 0),
dax.NewPartition(20, 0),
dax.NewPartition(23, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(2, 0),
dax.NewVersionedPartition(5, 0),
dax.NewVersionedPartition(8, 0),
dax.NewVersionedPartition(11, 0),
dax.NewVersionedPartition(14, 0),
dax.NewVersionedPartition(17, 0),
dax.NewVersionedPartition(20, 0),
dax.NewVersionedPartition(23, 0),
},
},
},
@ -996,15 +996,15 @@ func TestController(t *testing.T) {
TranslateRoles: []dax.TranslateRole{
{
TableKey: tbl1.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
dax.NewPartition(3, 0),
dax.NewPartition(6, 0),
dax.NewPartition(9, 0),
dax.NewPartition(12, 0),
dax.NewPartition(15, 0),
dax.NewPartition(18, 0),
dax.NewPartition(21, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
dax.NewVersionedPartition(3, 0),
dax.NewVersionedPartition(6, 0),
dax.NewVersionedPartition(9, 0),
dax.NewVersionedPartition(12, 0),
dax.NewVersionedPartition(15, 0),
dax.NewVersionedPartition(18, 0),
dax.NewVersionedPartition(21, 0),
},
},
},
@ -1021,8 +1021,8 @@ func TestController(t *testing.T) {
// Add shards to a table which doesn't exist.
err = con.AddShards(ctx, invalidQtid,
dax.NewShard(1, 0),
dax.NewShard(2, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(2, 0),
)
if assert.Error(t, err) {
assert.True(t, errors.Is(err, dax.ErrTableIDDoesNotExist))
@ -1083,12 +1083,12 @@ func TestController(t *testing.T) {
// Add shards.
assert.NoError(t, con.AddShards(ctx, tbl0.QualifiedID(),
dax.NewShard(0, 0),
dax.NewShard(1, 0),
dax.NewShard(2, 0),
dax.NewShard(3, 0),
dax.NewShard(11, 0),
dax.NewShard(12, 0),
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(3, 0),
dax.NewVersionedShard(11, 0),
dax.NewVersionedShard(12, 0),
))
t.Run("ComputeRole", func(t *testing.T) {
@ -1100,16 +1100,16 @@ func TestController(t *testing.T) {
{
role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.NewShards(0, 1, 2, 3),
Shards: dax.NewVersionedShards(0, 1, 2, 3),
},
exp: []dax.AssignedNode{
{
Address: node0.Address,
Role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(0, 0),
dax.NewShard(2, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(0, 0),
dax.NewVersionedShard(2, 0),
},
},
},
@ -1117,9 +1117,9 @@ func TestController(t *testing.T) {
Address: node1.Address,
Role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(1, 0),
dax.NewShard(3, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(3, 0),
},
},
},
@ -1128,15 +1128,15 @@ func TestController(t *testing.T) {
{
role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.NewShards(1),
Shards: dax.NewVersionedShards(1),
},
exp: []dax.AssignedNode{
{
Address: node1.Address,
Role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(1, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(1, 0),
},
},
},
@ -1146,7 +1146,7 @@ func TestController(t *testing.T) {
// Add unassigned shards.
role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.NewShards(1, 888, 889),
Shards: dax.NewVersionedShards(1, 888, 889),
},
isWrite: true,
exp: []dax.AssignedNode{
@ -1154,8 +1154,8 @@ func TestController(t *testing.T) {
Address: node0.Address,
Role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(888, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(888, 0),
},
},
},
@ -1163,9 +1163,9 @@ func TestController(t *testing.T) {
Address: node1.Address,
Role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(1, 0),
dax.NewShard(889, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(1, 0),
dax.NewVersionedShard(889, 0),
},
},
},
@ -1175,16 +1175,16 @@ func TestController(t *testing.T) {
// Ensure shards are not returned sorted as strings.
role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.NewShards(2, 11),
Shards: dax.NewVersionedShards(2, 11),
},
exp: []dax.AssignedNode{
{
Address: node0.Address,
Role: &dax.ComputeRole{
TableKey: tbl0.Key(),
Shards: dax.Shards{
dax.NewShard(2, 0),
dax.NewShard(11, 0),
Shards: dax.VersionedShards{
dax.NewVersionedShard(2, 0),
dax.NewVersionedShard(11, 0),
},
},
},
@ -1210,8 +1210,8 @@ func TestController(t *testing.T) {
{
role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, -1),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, -1),
},
},
isWrite: true,
@ -1220,8 +1220,8 @@ func TestController(t *testing.T) {
Address: node0.Address,
Role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
},
},
},
@ -1230,12 +1230,12 @@ func TestController(t *testing.T) {
{
role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, -1),
dax.NewPartition(1, -1),
dax.NewPartition(2, -1),
dax.NewPartition(3, -1),
dax.NewPartition(999, -1),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, -1),
dax.NewVersionedPartition(1, -1),
dax.NewVersionedPartition(2, -1),
dax.NewVersionedPartition(3, -1),
dax.NewVersionedPartition(999, -1),
},
},
isWrite: false,
@ -1244,9 +1244,9 @@ func TestController(t *testing.T) {
Address: node0.Address,
Role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(0, 0),
dax.NewPartition(2, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(0, 0),
dax.NewVersionedPartition(2, 0),
},
},
},
@ -1254,9 +1254,9 @@ func TestController(t *testing.T) {
Address: node1.Address,
Role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(1, 0),
dax.NewPartition(3, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(1, 0),
dax.NewVersionedPartition(3, 0),
},
},
},
@ -1265,8 +1265,8 @@ func TestController(t *testing.T) {
{
role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(1, -1),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(1, -1),
},
},
isWrite: false,
@ -1275,8 +1275,8 @@ func TestController(t *testing.T) {
Address: node1.Address,
Role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(1, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(1, 0),
},
},
},
@ -1286,9 +1286,9 @@ func TestController(t *testing.T) {
// Ensure partitions are not returned sorted as strings.
role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(2, -1),
dax.NewPartition(10, -1),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(2, -1),
dax.NewVersionedPartition(10, -1),
},
},
isWrite: false,
@ -1297,9 +1297,9 @@ func TestController(t *testing.T) {
Address: node0.Address,
Role: &dax.TranslateRole{
TableKey: tbl0.Key(),
Partitions: dax.Partitions{
dax.NewPartition(2, 0),
dax.NewPartition(10, 0),
Partitions: dax.VersionedPartitions{
dax.NewVersionedPartition(2, 0),
dax.NewVersionedPartition(10, 0),
},
},
},

12
dax/mds/controller/stringers.go
View file

@ -13,7 +13,7 @@ import (
// used as a job in the Balancer.
type pUnit struct {
t dax.TableKey
p dax.Partition
p dax.VersionedPartition
}
func (p pUnit) String() string {
@ -28,7 +28,7 @@ func (p pUnit) partitionNum() dax.PartitionNum {
return p.p.Num
}
func partition(t dax.TableKey, p dax.Partition) pUnit {
func partition(t dax.TableKey, p dax.VersionedPartition) pUnit {
return pUnit{t, p}
}
@ -49,7 +49,7 @@ func decodePartition(j dax.Job) (pUnit, error) {
return pUnit{
t: dax.TableKey(parts[0]),
p: dax.Partition{
p: dax.VersionedPartition{
Num: dax.PartitionNum(intVar),
Version: -1,
},
@ -60,7 +60,7 @@ func decodePartition(j dax.Job) (pUnit, error) {
// a job in the Balancer.
type sUnit struct {
t dax.TableKey
s dax.Shard
s dax.VersionedShard
}
func (s sUnit) String() string {
@ -75,7 +75,7 @@ func (s sUnit) shardNum() dax.ShardNum {
return s.s.Num
}
func shard(t dax.TableKey, s dax.Shard) sUnit {
func shard(t dax.TableKey, s dax.VersionedShard) sUnit {
return sUnit{t, s}
}
@ -96,7 +96,7 @@ func decodeShard(j dax.Job) (sUnit, error) {
return sUnit{
t: dax.TableKey(parts[0]),
s: dax.Shard{
s: dax.VersionedShard{
Num: dax.ShardNum(uint64Var),
Version: -1,
},

65
dax/partition.go
View file

@ -1,65 +0,0 @@
package dax
import "fmt"
// PartitionNum is the numerical (int) partition value.
type PartitionNum int
// PartitionNums is a slice of PartitionNum.
type PartitionNums []PartitionNum
// String returns the PartitionNum as a string.
func (p PartitionNum) String() string {
return fmt.Sprintf("%d", p)
}
// Partition is a versioned partition.
type Partition struct {
Num PartitionNum `json:"num"`
Version int `json:"version"`
}
// NewPartition returns a Partition with the provided num and version.
func NewPartition(num PartitionNum, version int) Partition {
return Partition{
Num: num,
Version: version,
}
}
// String returns the Partition (i.e. its Num and Version) as a string.
func (p Partition) String() string {
return fmt.Sprintf("%d.%d", p.Num, p.Version)
}
// Partitions is a sortable slice of Partition.
type Partitions []Partition
func (p Partitions) Len() int { return len(p) }
func (p Partitions) Less(i, j int) bool { return p[i].Num < p[j].Num }
func (p Partitions) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
// NewPartitions returns the provided list of partition nums as a list of
// Partition with an invalid version (-1). This is to use for cases where the
// request should not be aware of a partition versioning.
func NewPartitions(partitionNums ...PartitionNum) Partitions {
pvs := make(Partitions, len(partitionNums))
for i := range partitionNums {
pvs[i] = Partition{
Num: partitionNums[i],
Version: -1,
}
}
return pvs
}
// Nums returns a slice of all the partition numbers in Partitions.
func (p Partitions) Nums() []PartitionNum {
pp := make([]PartitionNum, len(p))
for i := range p {
pp[i] = p[i].Num
}
return pp
}

10
dax/role.go
View file

@ -33,8 +33,8 @@ var _ Role = &TranslateRole{}
// ComputeRole is a role specific to compute nodes.
type ComputeRole struct {
TableKey TableKey `json:"table-key"`
Shards Shards `json:"shards"`
TableKey TableKey `json:"table-key"`
Shards VersionedShards `json:"shards"`
}
// Type returns the type for ComputeRole. This is mainly to impolement the Role
@ -45,9 +45,9 @@ func (cr *ComputeRole) Type() RoleType {
// TranslateRole is a role specific to translate nodes.
type TranslateRole struct {
TableKey TableKey `json:"table-key"`
Partitions Partitions `json:"partitions"`
Fields FieldVersions `json:"fields"`
TableKey TableKey `json:"table-key"`
Partitions VersionedPartitions `json:"partitions"`
Fields VersionedFields `json:"fields"`
}
// Type returns the type for TransteRole. This is mainly to impolement the Role

64
dax/shard.go
View file

@ -1,64 +0,0 @@
package dax
import "fmt"
// ShardNum is the numerical (uint64) shard value.
type ShardNum uint64
// ShardNums is a slice of ShardNum.
type ShardNums []ShardNum
func (s ShardNum) String() string {
return fmt.Sprintf("%d", s)
}
// Shard is a versioned shard.
type Shard struct {
Num ShardNum `json:"num"`
Version int `json:"version"`
}
// NewShard returns a Shard with the provided num and version.
func NewShard(num ShardNum, version int) Shard {
return Shard{
Num: num,
Version: version,
}
}
// String returns the Shard (i.e. its Num and Version) as a string.
func (s Shard) String() string {
return fmt.Sprintf("%d.%d", s.Num, s.Version)
}
// Shards is a sortable slice of Shard.
type Shards []Shard
func (s Shards) Len() int { return len(s) }
func (s Shards) Less(i, j int) bool { return s[i].Num < s[j].Num }
func (s Shards) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
// NewShards returns the provided list of shard nums as a list of Shard with an
// invalid version (-1). This is to use for cases where the request should not
// be aware of shard versioning.
func NewShards(shardNums ...ShardNum) Shards {
svs := make(Shards, len(shardNums))
for i := range shardNums {
svs[i] = Shard{
Num: shardNums[i],
Version: -1,
}
}
return svs
}
// Nums returns a slice of all the shard numbers in Shards.
func (s Shards) Nums() []ShardNum {
ss := make([]ShardNum, len(s))
for i := range s {
ss[i] = s[i].Num
}
return ss
}

30
dax/versioned_field.go Normal file
View file

@ -0,0 +1,30 @@
package dax
import "fmt"
// VersionedField is used in a similar way to VersionedShard and
// VersionedPartition in that they all contain a snapshot version.
type VersionedField struct {
Name FieldName `json:"name"`
Version int `json:"version"`
}
// String returns the VersionedField (i.e. its Name and Version) as a string.
func (f VersionedField) String() string {
return fmt.Sprintf("%s.%d", f.Name, f.Version)
}
// NewVersionedField returns a VersionedField with the provided name and version.
func NewVersionedField(name FieldName, version int) VersionedField {
return VersionedField{
Name: name,
Version: version,
}
}
// VersionedFields is a sortable slice of VersionedField.
type VersionedFields []VersionedField
func (f VersionedFields) Len() int { return len(f) }
func (f VersionedFields) Less(i, j int) bool { return f[i].Name < f[j].Name }
func (f VersionedFields) Swap(i, j int) { f[i], f[j] = f[j], f[i] }

67
dax/versioned_partition.go Normal file
View file

@ -0,0 +1,67 @@
package dax
import "fmt"
// PartitionNum is the numerical (int) partition value.
type PartitionNum int
// PartitionNums is a slice of PartitionNum.
type PartitionNums []PartitionNum
// String returns the PartitionNum as a string.
func (p PartitionNum) String() string {
return fmt.Sprintf("%d", p)
}
// VersionedPartition is a partition number along with the snapshot version
// which it is currently writing at.
type VersionedPartition struct {
Num PartitionNum `json:"num"`
Version int `json:"version"`
}
// NewVersionedPartition returns a VersionedPartition with the provided
// partition number and version.
func NewVersionedPartition(num PartitionNum, version int) VersionedPartition {
return VersionedPartition{
Num: num,
Version: version,
}
}
// String returns the VersionedPartition (i.e. its Num and Version) as a string.
func (p VersionedPartition) String() string {
return fmt.Sprintf("%d.%d", p.Num, p.Version)
}
// VersionedPartitions is a sortable slice of VersionedPartition.
type VersionedPartitions []VersionedPartition
func (p VersionedPartitions) Len() int { return len(p) }
func (p VersionedPartitions) Less(i, j int) bool { return p[i].Num < p[j].Num }
func (p VersionedPartitions) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
// NewVersionedPartitions returns the provided list of partition nums as a list
// of VersionedPartition with an invalid version (-1). This is to use for cases
// where the request should not be aware of a partition versioning.
func NewVersionedPartitions(partitionNums ...PartitionNum) VersionedPartitions {
pvs := make(VersionedPartitions, len(partitionNums))
for i := range partitionNums {
pvs[i] = VersionedPartition{
Num: partitionNums[i],
Version: -1,
}
}
return pvs
}
// Nums returns a slice of all the partition numbers in VersionedPartitions.
func (p VersionedPartitions) Nums() []PartitionNum {
pp := make([]PartitionNum, len(p))
for i := range p {
pp[i] = p[i].Num
}
return pp
}

65
dax/versioned_shard.go Normal file
View file

@ -0,0 +1,65 @@
package dax
import "fmt"
// ShardNum is the numerical (uint64) shard value.
type ShardNum uint64
// ShardNums is a slice of ShardNum.
type ShardNums []ShardNum
func (s ShardNum) String() string {
return fmt.Sprintf("%d", s)
}
// VersionedShard is a shard number along with the snapshot version which it is
// currently writing at.
type VersionedShard struct {
Num ShardNum `json:"num"`
Version int `json:"version"`
}
// NewVersionedShard returns a VersionedShard with the provided shard number and version.
func NewVersionedShard(num ShardNum, version int) VersionedShard {
return VersionedShard{
Num: num,
Version: version,
}
}
// String returns the VersionedShard (i.e. its Num and Version) as a string.
func (s VersionedShard) String() string {
return fmt.Sprintf("%d.%d", s.Num, s.Version)
}
// VersionedShards is a sortable slice of VersionedShard.
type VersionedShards []VersionedShard
func (s VersionedShards) Len() int { return len(s) }
func (s VersionedShards) Less(i, j int) bool { return s[i].Num < s[j].Num }
func (s VersionedShards) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
// NewVersionedShards returns the provided list of shard nums as a list of
// VersionedShard with an invalid version (-1). This is to use for cases where
// the request should not be aware of shard versioning.
func NewVersionedShards(shardNums ...ShardNum) VersionedShards {
svs := make(VersionedShards, len(shardNums))
for i := range shardNums {
svs[i] = VersionedShard{
Num: shardNums[i],
Version: -1,
}
}
return svs
}
// Nums returns a slice of all the shard numbers in VersionedShards.
func (s VersionedShards) Nums() []ShardNum {
ss := make([]ShardNum, len(s))
for i := range s {
ss[i] = s[i].Num
}
return ss
}

28
dax/versionstore.go
View file

@ -11,23 +11,23 @@ import (
// is the interface through which various services read/write that version.
type VersionStore interface {
AddTable(ctx context.Context, qtid QualifiedTableID) error
RemoveTable(ctx context.Context, qtid QualifiedTableID) (Shards, Partitions, error)
RemoveTable(ctx context.Context, qtid QualifiedTableID) (VersionedShards, VersionedPartitions, error)
// Shards (shardData)
AddShards(ctx context.Context, qtid QualifiedTableID, shards ...Shard) error
Shards(ctx context.Context, qtid QualifiedTableID) (Shards, bool, error)
AddShards(ctx context.Context, qtid QualifiedTableID, shards ...VersionedShard) error
Shards(ctx context.Context, qtid QualifiedTableID) (VersionedShards, bool, error)
ShardVersion(ctx context.Context, qtid QualifiedTableID, shardNum ShardNum) (int, bool, error)
ShardTables(ctx context.Context, qual TableQualifier) (TableIDs, error)
// Partitions (tableKeys)
AddPartitions(ctx context.Context, qtid QualifiedTableID, partitions ...Partition) error
Partitions(ctx context.Context, qtid QualifiedTableID) (Partitions, bool, error)
AddPartitions(ctx context.Context, qtid QualifiedTableID, partitions ...VersionedPartition) error
Partitions(ctx context.Context, qtid QualifiedTableID) (VersionedPartitions, bool, error)
PartitionVersion(ctx context.Context, qtid QualifiedTableID, partitionNum PartitionNum) (int, bool, error)
PartitionTables(ctx context.Context, qual TableQualifier) (TableIDs, error)
// Fields (fieldKeys)
AddFields(ctx context.Context, qtid QualifiedTableID, fields ...FieldVersion) error
Fields(ctx context.Context, qtid QualifiedTableID) (FieldVersions, bool, error)
AddFields(ctx context.Context, qtid QualifiedTableID, fields ...VersionedField) error
Fields(ctx context.Context, qtid QualifiedTableID) (VersionedFields, bool, error)
FieldVersion(ctx context.Context, qtid QualifiedTableID, field FieldName) (int, bool, error)
FieldTables(ctx context.Context, qual TableQualifier) (TableIDs, error)
@ -53,15 +53,15 @@ func (s *nopVersionStore) AddTable(ctx context.Context, qtid QualifiedTableID) e
return nil
}
func (s *nopVersionStore) RemoveTable(ctx context.Context, qtid QualifiedTableID) (Shards, Partitions, error) {
func (s *nopVersionStore) RemoveTable(ctx context.Context, qtid QualifiedTableID) (VersionedShards, VersionedPartitions, error) {
return nil, nil, nil
}
func (s *nopVersionStore) AddShards(ctx context.Context, qtid QualifiedTableID, shards ...Shard) error {
func (s *nopVersionStore) AddShards(ctx context.Context, qtid QualifiedTableID, shards ...VersionedShard) error {
return nil
}
func (s *nopVersionStore) Shards(ctx context.Context, qtid QualifiedTableID) (Shards, bool, error) {
func (s *nopVersionStore) Shards(ctx context.Context, qtid QualifiedTableID) (VersionedShards, bool, error) {
return nil, false, nil
}
@ -73,11 +73,11 @@ func (s *nopVersionStore) ShardTables(ctx context.Context, qual TableQualifier)
return TableIDs{}, nil
}
func (s *nopVersionStore) AddPartitions(ctx context.Context, qtid QualifiedTableID, partitions ...Partition) error {
func (s *nopVersionStore) AddPartitions(ctx context.Context, qtid QualifiedTableID, partitions ...VersionedPartition) error {
return nil
}
func (s *nopVersionStore) Partitions(ctx context.Context, qtid QualifiedTableID) (Partitions, bool, error) {
func (s *nopVersionStore) Partitions(ctx context.Context, qtid QualifiedTableID) (VersionedPartitions, bool, error) {
return nil, false, nil
}
@ -89,11 +89,11 @@ func (s *nopVersionStore) PartitionTables(ctx context.Context, qual TableQualifi
return TableIDs{}, nil
}
func (s *nopVersionStore) AddFields(ctx context.Context, qtid QualifiedTableID, fields ...FieldVersion) error {
func (s *nopVersionStore) AddFields(ctx context.Context, qtid QualifiedTableID, fields ...VersionedField) error {
return nil
}
func (s *nopVersionStore) Fields(ctx context.Context, qtid QualifiedTableID) (FieldVersions, bool, error) {
func (s *nopVersionStore) Fields(ctx context.Context, qtid QualifiedTableID) (VersionedFields, bool, error) {
return nil, false, nil
}

6
holder.go
View file

@ -1017,7 +1017,7 @@ func (h *Holder) createIndex(cim *CreateIndexMessage, broadcast bool) (*Index, e
// createIndexWithPartitions is similar to createIndex, but it takes a list of
// partitions for which this node is responsible. This ensures that the node
// doesn't instantiate more partition TranslateStores than is necessary.
func (h *Holder) createIndexWithPartitions(cim *CreateIndexMessage, translatePartitions dax.Partitions) (*Index, error) {
func (h *Holder) createIndexWithPartitions(cim *CreateIndexMessage, translatePartitions dax.VersionedPartitions) (*Index, error) {
if cim.Index == "" {
return nil, errors.New("index name required")
}
@ -1049,9 +1049,9 @@ func (h *Holder) createIndexWithPartitions(cim *CreateIndexMessage, translatePar
}
// Initialize a list of partitions at version 0.
newPartitions := make(dax.Partitions, len(translatePartitions))
newPartitions := make(dax.VersionedPartitions, len(translatePartitions))
for i := range translatePartitions {
newPartitions[i] = dax.NewPartition(translatePartitions[i].Num, 0)
newPartitions[i] = dax.NewVersionedPartition(translatePartitions[i].Num, 0)
}
if err := h.versionStore.AddPartitions(context.Background(), qtid, newPartitions...); err != nil {

4
index.go
View file

@ -50,7 +50,7 @@ type Index struct {
holder *Holder
// Per-partition translation stores
translatePartitions dax.Partitions
translatePartitions dax.VersionedPartitions
translateStores map[int]TranslateStore
translationSyncer TranslationSyncer
@ -967,7 +967,7 @@ func (i *Index) DeleteField(name string) error {
// case, we need to update this cached value. Really, this is kind of hacky and
// we need to revisit the ApplyDirective logic so that it's more intuitive with
// respect to index.translatePartitions.
func (i *Index) SetTranslatePartitions(tp dax.Partitions) {
func (i *Index) SetTranslatePartitions(tp dax.VersionedPartitions) {
i.mu.Lock()
defer i.mu.Unlock()