diff --git a/api.go b/api.go index 6b6f95cfe..88c7d289e 100644 --- a/api.go +++ b/api.go @@ -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 diff --git a/api_directive.go b/api_directive.go index a907ea442..2ecb3146a 100644 --- a/api_directive.go +++ b/api_directive.go @@ -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, diff --git a/cluster.go b/cluster.go index cd27ee603..3e8991820 100644 --- a/cluster.go +++ b/cluster.go @@ -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 diff --git a/dax/boltdb/versionstore.go b/dax/boltdb/versionstore.go index 20171abd6..4b857c285 100644 --- a/dax/boltdb/versionstore.go +++ b/dax/boltdb/versionstore.go @@ -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 { diff --git a/dax/boltdb/versionstore_test.go b/dax/boltdb/versionstore_test.go index cead07425..d6bde2254 100644 --- a/dax/boltdb/versionstore_test.go +++ b/dax/boltdb/versionstore_test.go @@ -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, } diff --git a/dax/directive.go b/dax/directive.go index 869bbdade..f5504864d 100644 --- a/dax/directive.go +++ b/dax/directive.go @@ -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 } diff --git a/dax/fieldversion.go b/dax/fieldversion.go deleted file mode 100644 index 8f1133c57..000000000 --- a/dax/fieldversion.go +++ /dev/null @@ -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] } diff --git a/dax/inmem/versionstore.go b/dax/inmem/versionstore.go index 5f30e41ce..64cbf2564 100644 --- a/dax/inmem/versionstore.go +++ b/dax/inmem/versionstore.go @@ -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 diff --git a/dax/inmem/versionstore_test.go b/dax/inmem/versionstore_test.go index 82db23b7b..35a7693cb 100644 --- a/dax/inmem/versionstore_test.go +++ b/dax/inmem/versionstore_test.go @@ -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. diff --git a/dax/mds/controller/controller.go b/dax/mds/controller/controller.go index c6f130fc3..7c8e44fe6 100644 --- a/dax/mds/controller/controller.go +++ b/dax/mds/controller/controller.go @@ -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() diff --git a/dax/mds/controller/controller_test.go b/dax/mds/controller/controller_test.go index eea00a241..22a226ef2 100644 --- a/dax/mds/controller/controller_test.go +++ b/dax/mds/controller/controller_test.go @@ -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), }, }, }, diff --git a/dax/mds/controller/stringers.go b/dax/mds/controller/stringers.go index 30269f664..59b71ae32 100644 --- a/dax/mds/controller/stringers.go +++ b/dax/mds/controller/stringers.go @@ -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, }, diff --git a/dax/partition.go b/dax/partition.go deleted file mode 100644 index b1c4e297c..000000000 --- a/dax/partition.go +++ /dev/null @@ -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 -} diff --git a/dax/role.go b/dax/role.go index db1e2ee93..48655a873 100644 --- a/dax/role.go +++ b/dax/role.go @@ -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 diff --git a/dax/shard.go b/dax/shard.go deleted file mode 100644 index 2d0264aa5..000000000 --- a/dax/shard.go +++ /dev/null @@ -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 -} diff --git a/dax/versioned_field.go b/dax/versioned_field.go new file mode 100644 index 000000000..3edfb2608 --- /dev/null +++ b/dax/versioned_field.go @@ -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] } diff --git a/dax/versioned_partition.go b/dax/versioned_partition.go new file mode 100644 index 000000000..9dd0c123d --- /dev/null +++ b/dax/versioned_partition.go @@ -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 +} diff --git a/dax/versioned_shard.go b/dax/versioned_shard.go new file mode 100644 index 000000000..68d568179 --- /dev/null +++ b/dax/versioned_shard.go @@ -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 +} diff --git a/dax/versionstore.go b/dax/versionstore.go index 112cdd00e..1d1dc051f 100644 --- a/dax/versionstore.go +++ b/dax/versionstore.go @@ -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 } diff --git a/holder.go b/holder.go index 55f356a54..f7c51e75e 100644 --- a/holder.go +++ b/holder.go @@ -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 { diff --git a/index.go b/index.go index 0a6205204..3f9cf52f6 100644 --- a/index.go +++ b/index.go @@ -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()