scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-06 16:45:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Implement translator store sharding

Browse source
This commit is contained in:
Ben Johnson 2019-12-02 14:14:33 -07:00
parent cabfa3c456
commit 7215bfd16c
23 changed files with 4717 additions and 2704 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
api.go
View file

@ -545,7 +545,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
var err error
if field.keys() {
if rowStr, err = field.translateStore.TranslateID(rowID); err != nil {
if rowStr, err = api.cluster.translateFieldID(indexName, fieldName, rowID); err != nil {
return errors.Wrap(err, "translating row")
}
} else {
@ -553,7 +553,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
}
if index.Keys() {
if colStr, err = index.translateStore.TranslateID(columnID); err != nil {
if colStr, err = api.cluster.translateIndexPartitionID(ctx, indexName, api.cluster.idPartition(indexName, columnID), columnID); err != nil {
return errors.Wrap(err, "translating column")
}
} else {
@ -963,7 +963,7 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
if len(req.RowIDs) != 0 {
return errors.New("row ids cannot be used because field uses string keys")
}
if req.RowIDs, err = field.translateStore.TranslateKeys(req.RowKeys); err != nil {
if req.RowIDs, err = api.cluster.translateFieldKeys(req.Index, req.Field, req.RowKeys); err != nil {
return errors.Wrap(err, "translating rows")
}
}
@ -973,7 +973,7 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = index.translateStore.TranslateKeys(req.ColumnKeys); err != nil {
if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
}
}
@ -1078,7 +1078,7 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = index.translateStore.TranslateKeys(req.ColumnKeys); err != nil {
if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
}
req.Shard = math.MaxUint64
@ -1371,11 +1371,11 @@ func (api *API) Info() serverInfo {
func (api *API) GetTranslateEntryReader(ctx context.Context, offsets TranslateOffsetMap) (TranslateEntryReader, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "API.GetTranslateEntryReader")
defer span.Finish()
return api.holder.TranslateEntryReader(ctx, offsets)
return api.cluster.translateEntryReader(ctx, offsets)
}
// TranslateKeys handles a TranslateKeyRequest.
func (api *API) TranslateKeys(r io.Reader) ([]byte, error) {
func (api *API) TranslateKeys(ctx context.Context, r io.Reader) (_ []byte, err error) {
var req TranslateKeysRequest
if buf, err := ioutil.ReadAll(r); err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "read translate keys request error"))
@ -1384,13 +1384,15 @@ func (api *API) TranslateKeys(r io.Reader) ([]byte, error) {
}
// Lookup store for either index or field and translate keys.
store, err := api.holder.TranslateStore(req.Index, req.Field)
if err != nil {
return nil, err
}
ids, err := store.TranslateKeys(req.Keys)
if err != nil {
return nil, err
var ids []uint64
if req.Field == "" {
if ids, err = api.cluster.translateIndexKeys(ctx, req.Index, req.Keys); err != nil {
return nil, err
}
} else {
if ids, err = api.cluster.translateFieldKeys(req.Index, req.Field, req.Keys); err != nil {
return nil, err
}
}
// Encode response.

17
boltdb/translate.go
View file

@ -32,8 +32,8 @@ var (
)
// OpenTranslateStore opens and initializes a boltdb translation store.
func OpenTranslateStore(path, index, field string) (pilosa.TranslateStore, error) {
s := NewTranslateStore(index, field)
func OpenTranslateStore(path, index, field string, partitionID int) (pilosa.TranslateStore, error) {
s := NewTranslateStore(index, field, partitionID)
s.Path = path
if err := s.Open(); err != nil {
return nil, err
@ -49,8 +49,9 @@ type TranslateStore struct {
mu sync.RWMutex
db *bolt.DB
index string
field string
index string
field string
partitionID int
once sync.Once
closing chan struct{}
@ -63,10 +64,11 @@ type TranslateStore struct {
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore(index, field string) *TranslateStore {
func NewTranslateStore(index, field string, partitionID int) *TranslateStore {
return &TranslateStore{
index: index,
field: field,
partitionID: partitionID,
closing: make(chan struct{}),
writeNotify: make(chan struct{}),
}
@ -108,6 +110,11 @@ func (s *TranslateStore) Close() (err error) {
return nil
}
// PartitionID returns the partition id the store was initialized with.
func (s *TranslateStore) PartitionID() int {
return s.partitionID
}
// ReadOnly returns true if the store is in read-only mode.
func (s *TranslateStore) ReadOnly() bool {
s.mu.RLock()

19
client.go
View file

@ -44,6 +44,8 @@ type FieldValue struct {
// While I understand that putting the entire Client behind an interface might require this many methods,
// I don't want to let it go unquestioned.
type InternalClient interface {
InternalQueryClient
MaxShardByIndex(ctx context.Context) (map[string]uint64, error)
Schema(ctx context.Context) ([]*IndexInfo, error)
PostSchema(ctx context.Context, uri *URI, s *Schema, remote bool) error
@ -51,7 +53,6 @@ type InternalClient interface {
FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error)
Nodes(ctx context.Context) ([]*Node, error)
Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error)
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error
ImportK(ctx context.Context, index, field string, bits []Bit, opts ...ImportOption) error
EnsureIndex(ctx context.Context, name string, options IndexOptions) error
@ -78,6 +79,8 @@ type InternalClient interface {
// InternalQueryClient is the internal interface for querying a node.
type InternalQueryClient interface {
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string) ([]uint64, error)
TranslateIDsNode(ctx context.Context, uri *URI, index, field string, id []uint64) ([]string, error)
}
type nopInternalQueryClient struct{}
@ -86,6 +89,14 @@ func (n *nopInternalQueryClient) QueryNode(ctx context.Context, uri *URI, index
return nil, nil
}
func (n nopInternalQueryClient) TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string) ([]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) TranslateIDsNode(ctx context.Context, uri *URI, index, field string, ids []uint64) ([]string, error) {
return nil, nil
}
func newNopInternalQueryClient() *nopInternalQueryClient {
return &nopInternalQueryClient{}
}
@ -125,6 +136,12 @@ func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest
func (n nopInternalClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n nopInternalClient) TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string) ([]uint64, error) {
return nil, nil
}
func (n nopInternalClient) TranslateIDsNode(ctx context.Context, uri *URI, index, field string, ids []uint64) ([]string, error) {
return nil, nil
}
func (n nopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error {
return nil
}

532
cluster.go
View file

@ -215,6 +215,10 @@ type cluster struct { // nolint: maligned
holder *Holder
broadcaster broadcaster
// translation stores
indexTranslateStoreMap map[string]map[int]TranslateStore // store by index+partition
fieldTranslateStoreMap map[string]map[string]TranslateStore // store by index+field
joiningLeavingNodes chan nodeAction
// joining is held open until this node
@ -235,6 +239,10 @@ type cluster struct { // nolint: maligned
logger logger.Logger
InternalClient InternalClient
// Instantiates new translation stores
OpenTranslateStore OpenTranslateStoreFunc
OpenTranslateReader OpenTranslateReaderFunc
}
// newCluster returns a new instance of Cluster with defaults.
@ -249,8 +257,13 @@ func newCluster() *cluster {
closing: make(chan struct{}),
joining: make(chan struct{}),
indexTranslateStoreMap: make(map[string]map[int]TranslateStore),
fieldTranslateStoreMap: make(map[string]map[string]TranslateStore),
InternalClient: newNopInternalClient(),
OpenTranslateStore: OpenInMemTranslateStore,
logger: logger.NopLogger,
}
}
@ -390,14 +403,6 @@ func (c *cluster) addNode(node *Node) error {
return nil
}
// If the cluster membership has changed, reset the primary for
// translate store replication.
if c.holder != nil {
if err := c.holder.setPrimaryTranslateStore(c.unprotectedPrimaryReplicaNode()); err != nil {
return err
}
}
// add to topology
if c.Topology == nil {
return fmt.Errorf("Cluster.Topology is nil")
@ -417,14 +422,6 @@ func (c *cluster) removeNode(nodeID string) error {
// remove from cluster
c.removeNodeBasicSorted(nodeID)
// If the cluster membership has changed, reset the primary for
// translate store replication.
if c.holder != nil {
if err := c.holder.setPrimaryTranslateStore(c.unprotectedPrimaryReplicaNode()); err != nil {
return err
}
}
// remove from topology
if c.Topology == nil {
return fmt.Errorf("Cluster.Topology is nil")
@ -867,8 +864,8 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
return m, nil
}
// partition returns the partition that a shard belongs to.
func (c *cluster) partition(index string, shard uint64) int {
// shardPartition returns the partition that a shard belongs to.
func (c *cluster) shardPartition(index string, shard uint64) int {
var buf [8]byte
binary.BigEndian.PutUint64(buf[:], shard)
@ -879,6 +876,20 @@ func (c *cluster) partition(index string, shard uint64) int {
return int(h.Sum64() % uint64(c.partitionN))
}
// keyPartition returns the partition that a shard belongs to.
func (c *cluster) keyPartition(index, key string) int {
// Hash the bytes and mod by partition count.
h := fnv.New64a()
_, _ = h.Write([]byte(index))
_, _ = h.Write([]byte(key))
return int(h.Sum64() % uint64(c.partitionN))
}
// idPartition returns the partition that an id belongs to.
func (c *cluster) idPartition(index string, id uint64) int {
return c.shardPartition(index, id/ShardWidth)
}
// ShardNodes returns a list of nodes that own a fragment. Safe for concurrent use.
func (c *cluster) ShardNodes(index string, shard uint64) []*Node {
c.mu.RLock()
@ -888,7 +899,19 @@ func (c *cluster) ShardNodes(index string, shard uint64) []*Node {
// shardNodes returns a list of nodes that own a fragment. unprotected
func (c *cluster) shardNodes(index string, shard uint64) []*Node {
return c.partitionNodes(c.partition(index, shard))
return c.partitionNodes(c.shardPartition(index, shard))
}
// KeyNodes returns a list of nodes that own a fragment. Safe for concurrent use.
func (c *cluster) KeyNodes(index, key string) []*Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.keyNodes(index, key)
}
// keyNodes returns a list of nodes that own a key. unprotected
func (c *cluster) keyNodes(index, key string) []*Node {
return c.partitionNodes(c.keyPartition(index, key))
}
// ownsShard returns true if a host owns a fragment.
@ -900,7 +923,6 @@ func (c *cluster) ownsShard(nodeID string, index string, shard uint64) bool {
// partitionNodes returns a list of nodes that own a partition. unprotected.
func (c *cluster) partitionNodes(partitionID int) []*Node {
// Default replica count to between one and the number of nodes.
// The replica count can be zero if there are no nodes.
replicaN := c.ReplicaN
@ -922,11 +944,18 @@ func (c *cluster) partitionNodes(partitionID int) []*Node {
return nodes
}
// ownsPartition returns true if a host owns a partition.
func (c *cluster) ownsPartition(nodeID string, partition int) bool {
c.mu.RLock()
defer c.mu.RUnlock()
return Nodes(c.partitionNodes(partition)).ContainsID(nodeID)
}
// containsShards is like OwnsShards, but it includes replicas.
func (c *cluster) containsShards(index string, availableShards *roaring.Bitmap, node *Node) []uint64 {
var shards []uint64
availableShards.ForEach(func(i uint64) {
p := c.partition(index, i)
p := c.shardPartition(index, i)
// Determine the nodes for partition.
nodes := c.partitionNodes(p)
for _, n := range nodes {
@ -982,6 +1011,10 @@ func (c *cluster) setup() error {
if err != nil {
return errors.Wrap(err, "adding local node")
}
if err := c.updateTranslateStores(); err != nil {
return err
}
return nil
}
@ -1991,6 +2024,11 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
}
}
// Open appropriate translate stores.
if err := c.updateTranslateStores(); err != nil {
return err
}
c.unprotectedSetState(cs.State)
c.markAsJoined()
@ -2047,6 +2085,458 @@ func (c *cluster) setStatic(hosts []string) error {
return nil
}
// updateTranslateStores starts stores for partitions & fields owned by this
// node and stops stores for ones this node does not own.
func (c *cluster) updateTranslateStores() error {
if err := c.closeUnownedTranslateStores(); err != nil {
return err
} else if err := c.openOwnedTranslateStores(); err != nil {
return err
}
return nil
}
func (c *cluster) closeUnownedTranslateStores() error {
for indexName, m := range c.indexTranslateStoreMap {
idx := c.holder.Index(indexName)
// Close unowned partition stores.
for partitionID, store := range m {
if idx != nil && c.ownsShard(c.Node.ID, indexName, uint64(partitionID)) {
continue
}
if err := store.Close(); err != nil {
return err
}
delete(m, partitionID)
}
if len(c.indexTranslateStoreMap) == 0 {
delete(c.indexTranslateStoreMap, indexName)
}
}
// Close index partition stores.
for indexName, m := range c.fieldTranslateStoreMap {
idx := c.holder.Index(indexName)
for fieldName, store := range m {
if idx != nil && idx.Field(fieldName) != nil {
continue
}
if err := store.Close(); err != nil {
return err
}
delete(m, fieldName)
}
if len(c.fieldTranslateStoreMap) == 0 {
delete(c.fieldTranslateStoreMap, indexName)
}
}
return nil
}
// openOwnedTranslateStores ensures that all owned partition & field stores are open.
func (c *cluster) openOwnedTranslateStores() error {
// Open partition stores.
for _, index := range c.holder.Indexes() {
m := c.indexTranslateStoreMap[index.Name()]
if m == nil {
m = make(map[int]TranslateStore)
c.indexTranslateStoreMap[index.Name()] = m
}
for partitionID := 0; partitionID < c.partitionN; partitionID++ {
if m[partitionID] != nil {
continue
}
store, err := c.OpenTranslateStore(index.TranslateStorePath(partitionID), index.Name(), "", partitionID)
if err != nil {
return err
}
m[partitionID] = store
}
}
// Open field stores.
for _, index := range c.holder.Indexes() {
m := c.fieldTranslateStoreMap[index.Name()]
if m == nil {
m = make(map[string]TranslateStore)
c.fieldTranslateStoreMap[index.Name()] = m
}
for _, field := range index.Fields() {
if m[field.Name()] != nil {
continue
}
store, err := c.OpenTranslateStore(field.TranslateStorePath(), index.Name(), field.Name(), 0)
if err != nil {
return err
}
m[field.Name()] = store
}
}
return nil
}
func (c *cluster) indexPartitionTranslateStore(index string, partitionID int) TranslateStore {
m := c.indexTranslateStoreMap[index]
if m == nil {
return nil
}
return m[partitionID]
}
func (c *cluster) fieldTranslateStore(index, field string) TranslateStore {
m := c.fieldTranslateStoreMap[index]
if m == nil {
return nil
}
return m[field]
}
func (c *cluster) translateIndexKeys(ctx context.Context, index string, keys []string) ([]uint64, error) {
keySet := make(map[string]struct{})
for _, key := range keys {
keySet[key] = struct{}{}
}
keyMap, err := c.translateIndexKeySet(ctx, index, keySet)
if err != nil {
return nil, err
}
ids := make([]uint64, len(keys))
for i := range keys {
ids[i] = keyMap[keys[i]]
}
return ids, nil
}
func (c *cluster) translateIndexKeySet(ctx context.Context, index string, keySet map[string]struct{}) (map[string]uint64, error) {
keyMap := make(map[string]uint64)
// Split keys by partition.
keysByPartition := make(map[int][]string, c.partitionN)
for key := range keySet {
partitionID := c.keyPartition(index, key)
keysByPartition[partitionID] = append(keysByPartition[partitionID], key)
}
// Translate keys by partition.
var g errgroup.Group
var mu sync.Mutex
for partitionID := range keysByPartition {
keys := keysByPartition[partitionID]
g.Go(func() (err error) {
var ids []uint64
if c.ownsPartition(c.Node.ID, partitionID) {
if ids, err = c.translateIndexPartitionKeys(ctx, index, partitionID, keys); err != nil {
return err
}
} else {
nodes := c.partitionNodes(partitionID)
if ids, err = c.InternalClient.TranslateKeysNode(ctx, &nodes[0].URI, index, "", keys); err != nil {
return err
}
}
mu.Lock()
defer mu.Unlock()
for i := range keys {
keyMap[keys[i]] = ids[i]
}
return nil
})
}
if err := g.Wait(); err != nil {
return nil, err
}
return keyMap, nil
}
func (c *cluster) translateIndexIDSet(ctx context.Context, index string, idSet map[uint64]struct{}) (map[uint64]string, error) {
idMap := make(map[uint64]string)
// Split ids by partition.
idsByPartition := make(map[int][]uint64, c.partitionN)
for id := range idSet {
partitionID := c.idPartition(index, id)
idsByPartition[partitionID] = append(idsByPartition[partitionID], id)
}
// Translate ids by partition.
var g errgroup.Group
var mu sync.Mutex
for partitionID, ids := range idsByPartition {
g.Go(func() error {
nodes := c.partitionNodes(partitionID)
keys, err := c.InternalClient.TranslateIDsNode(ctx, &nodes[0].URI, index, "", ids)
if err != nil {
return err
}
mu.Lock()
defer mu.Unlock()
for i := range ids {
idMap[ids[i]] = keys[i]
}
return nil
})
}
if err := g.Wait(); err != nil {
return nil, err
}
return idMap, nil
}
func (c *cluster) translateIndexPartitionKeys(ctx context.Context, index string, partitionID int, keys []string) ([]uint64, error) {
s := c.indexPartitionTranslateStore(index, partitionID)
if s == nil {
return nil, ErrTranslateStoreNotFound
}
return s.TranslateKeys(keys)
}
func (c *cluster) translateIndexPartitionIDs(ctx context.Context, index string, partitionID int, ids []uint64) ([]string, error) {
s := c.indexPartitionTranslateStore(index, partitionID)
if s == nil {
return nil, ErrTranslateStoreNotFound
}
return s.TranslateIDs(ids)
}
func (c *cluster) translateIndexPartitionID(ctx context.Context, index string, partitionID int, id uint64) (string, error) {
s := c.indexPartitionTranslateStore(index, partitionID)
if s == nil {
return "", ErrTranslateStoreNotFound
}
return s.TranslateID(id)
}
func (c *cluster) translateFieldKey(index, field string, key string) (uint64, error) {
s := c.fieldTranslateStore(index, field)
if s == nil {
return 0, ErrTranslateStoreNotFound
}
return s.TranslateKey(key)
}
func (c *cluster) translateFieldKeys(index, field string, keys []string) ([]uint64, error) {
s := c.fieldTranslateStore(index, field)
if s == nil {
return nil, ErrTranslateStoreNotFound
}
return s.TranslateKeys(keys)
}
func (c *cluster) translateFieldID(index, field string, id uint64) (string, error) {
s := c.fieldTranslateStore(index, field)
if s == nil {
return "", ErrTranslateStoreNotFound
}
return s.TranslateID(id)
}
func (c *cluster) translateFieldIDs(index, field string, ids []uint64) ([]string, error) {
s := c.fieldTranslateStore(index, field)
if s == nil {
return nil, ErrTranslateStoreNotFound
}
return s.TranslateIDs(ids)
}
// TranslateOffsetMap returns a map of offsets for all indexes & fields.
func (c *cluster) TranslateOffsetMap() (TranslateOffsetMap, error) {
m := make(TranslateOffsetMap)
for index, partitionMap := range c.indexTranslateStoreMap {
for partitionID, store := range partitionMap {
id, err := store.MaxID()
if err != nil {
return nil, err
}
m.SetIndexPartitionOffset(index, partitionID, id+1)
}
}
for index, fieldMap := range c.fieldTranslateStoreMap {
for field, store := range fieldMap {
id, err := store.MaxID()
if err != nil {
return nil, err
}
m.SetFieldOffset(index, field, id+1)
}
}
return m, nil
}
func (c *cluster) setTranslateStoreReadOnly(v bool) {
for _, partitionMap := range c.indexTranslateStoreMap {
for _, store := range partitionMap {
store.SetReadOnly(v)
}
}
for _, fieldMap := range c.fieldTranslateStoreMap {
for _, store := range fieldMap {
store.SetReadOnly(v)
}
}
}
// translateEntryReader returns a reader that merges all index & field reader
// that are specified in the offsets map.
func (c *cluster) translateEntryReader(ctx context.Context, offsets TranslateOffsetMap) (_ TranslateEntryReader, err error) {
// Ensure all readers are cleaned up if any error.
var a []TranslateEntryReader
defer func() {
if err != nil {
for i := range a {
a[i].Close()
}
}
}()
// Fetch all index partition readers.
for indexName, indexMap := range offsets {
for partitionID, offset := range indexMap.Partitions {
store := c.indexPartitionTranslateStore(indexName, partitionID)
if store == nil {
return nil, ErrTranslateStoreNotFound
}
r, err := store.EntryReader(ctx, uint64(offset))
if err != nil {
return nil, errors.Wrap(err, "index partition translate reader")
}
a = append(a, r)
}
}
// Fetch all field readers.
for indexName, indexMap := range offsets {
for fieldName, offset := range indexMap.Fields {
store := c.fieldTranslateStore(indexName, fieldName)
if store == nil {
return nil, ErrTranslateStoreNotFound
}
r, err := store.EntryReader(ctx, uint64(offset))
if err != nil {
return nil, errors.Wrap(err, "field translate reader")
}
a = append(a, r)
}
}
return NewMultiTranslateEntryReader(ctx, a), nil
}
/*
// holderTranslateStoreReplicator manages the replication of translation store
// data from a primary store to the local replica. Continually tries to
// reconnect on disconnect.
type holderTranslateStoreReplicator struct {
ctx context.Context
cancel func()
wg sync.WaitGroup
holder *Holder
nodeURL string
logger logger.Logger
}
func newHolderTranslateStoreReplicator(h *Holder, nodeURL string) *holderTranslateStoreReplicator {
r := &holderTranslateStoreReplicator{
holder: h,
nodeURL: nodeURL,
logger: logger.NopLogger,
}
r.ctx, r.cancel = context.WithCancel(context.Background())
return r
}
// Open starts the background monitoring goroutine.
func (r *holderTranslateStoreReplicator) Open() error {
r.wg.Add(1)
go func() { defer r.wg.Done(); r.monitor() }()
return nil
}
// Close stops the replicator.
func (r *holderTranslateStoreReplicator) Close() error {
r.cancel()
return nil
}
// monitor runs in a background goroutine and continually tries to connect and
// stream translate changes from the primary store.
func (r *holderTranslateStoreReplicator) monitor() {
for {
select {
case <-r.ctx.Done():
return
default:
if err := r.replicate(); err != nil {
r.logger.Printf("cannot replicate: nodeURL=%s err=%s", r.nodeURL, err)
}
time.Sleep(1 * time.Second)
}
}
}
func (r *holderTranslateStoreReplicator) replicate() error {
// Determine the offsets of every index & field store.
offsets, err := r.holder.TranslateOffsetMap()
if err != nil {
return err
} else if len(offsets) == 0 {
return nil
}
// Begin streaming from remote primary.
rd, err := r.holder.OpenTranslateReader(r.ctx, r.nodeURL, offsets)
if err != nil {
return err
}
defer rd.Close()
for {
var entry TranslateEntry
if err := rd.ReadEntry(&entry); err != nil {
return err
}
// Find appropriate store.
var store TranslateStore
if entry.Field == "" {
idx := r.holder.Index(entry.Index)
if idx == nil {
return ErrIndexNotFound
}
store = idx.TranslateStore()
} else {
f := r.holder.Field(entry.Index, entry.Field)
if f == nil {
return ErrFieldNotFound
}
store = f.TranslateStore()
}
// Apply replication to store.
if err := store.ForceSet(entry.ID, entry.Key); err != nil {
return err
}
}
}
*/
// ClusterStatus describes the status of the cluster including its
// state and node topology.
type ClusterStatus struct {

2
cluster_internal_test.go
View file

@ -352,7 +352,7 @@ func TestCluster_Partition(t *testing.T) {
c := newCluster()
c.partitionN = partitionN
partitionID := c.partition(index, shard)
partitionID := c.shardPartition(index, shard)
if partitionID < 0 || partitionID >= partitionN {
t.Errorf("partition out of range: shard=%d, p=%d, n=%d", shard, partitionID, partitionN)
}

30
encoding/proto/proto.go
View file

@ -758,17 +758,31 @@ func encodeRecalculateCaches(*pilosa.RecalculateCaches) *internal.RecalculateCac
return &internal.RecalculateCaches{}
}
func encodeTranslateKeysRequest(request *pilosa.TranslateKeysRequest) *internal.TranslateKeysRequest {
return &internal.TranslateKeysRequest{
Index: request.Index,
Field: request.Field,
Keys: request.Keys,
}
}
func encodeTranslateKeysResponse(response *pilosa.TranslateKeysResponse) *internal.TranslateKeysResponse {
return &internal.TranslateKeysResponse{
IDs: response.IDs,
}
}
func encodeTranslateKeysRequest(request *pilosa.TranslateKeysRequest) *internal.TranslateKeysRequest {
return &internal.TranslateKeysRequest{
func encodeTranslateIDsRequest(request *pilosa.TranslateIDsRequest) *internal.TranslateIDsRequest {
return &internal.TranslateIDsRequest{
Index: request.Index,
Field: request.Field,
Keys: request.Keys,
IDs: request.IDs,
}
}
func encodeTranslateIDsResponse(response *pilosa.TranslateIDsResponse) *internal.TranslateIDsResponse {
return &internal.TranslateIDsResponse{
Keys: response.Keys,
}
}
@ -1102,6 +1116,16 @@ func decodeTranslateKeysResponse(pb *internal.TranslateKeysResponse, m *pilosa.T
m.IDs = pb.IDs
}
func decodeTranslateIDsRequest(pb *internal.TranslateIDsRequest, m *pilosa.TranslateIDsRequest) {
m.Index = pb.Index
m.Field = pb.Field
m.IDs = pb.IDs
}
func decodeTranslateIDsResponse(pb *internal.TranslateIDsResponse, m *pilosa.TranslateIDsResponse) {
m.Keys = pb.Keys
}
// QueryResult types.
const (
queryResultTypeNil uint32 = iota

187
executor.go
View file

@ -29,6 +29,7 @@ import (
"github.com/pilosa/pilosa/v2/shardwidth"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
// defaultField is the field used if one is not specified.
@ -230,12 +231,18 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
// Translate column attributes, if necessary.
if idx.Keys() {
idSet := make(map[uint64]struct{})
for _, col := range columnAttrSets {
v, err := idx.translateStore.TranslateID(col.ID)
if err != nil {
return resp, err
}
col.Key, col.ID = v, 0
idSet[col.ID] = struct{}{}
}
idMap, err := e.Cluster.translateIndexIDSet(ctx, index, idSet)
if err != nil {
return resp, errors.Wrap(err, "translating id set")
}
for _, col := range columnAttrSets {
col.Key, col.ID = idMap[col.ID], 0
}
}
@ -3520,10 +3527,27 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
}
}
func (e *executor) translateCalls(ctx context.Context, index string, idx *Index, calls []*pql.Call) error {
func (e *executor) translateCalls(ctx context.Context, index string, idx *Index, calls []*pql.Call) (err error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.translateCalls")
defer span.Finish()
// TODO(BBJ): Handle cross-index boundaries.
keyMap := make(map[string]uint64)
if idx.Keys() {
// Collect all index keys.
keySet := make(map[string]struct{})
for i := range calls {
if err := e.collectCallIndexKeys(index, idx, calls[i], keySet); err != nil {
return err
}
}
if keyMap, err = e.Cluster.translateIndexKeySet(ctx, index, keySet); err != nil {
return err
}
}
// Translate calls.
for i := range calls {
// Possibly change to another index for translation, if this
// call crosses index boundaries.
@ -3538,51 +3562,57 @@ func (e *executor) translateCalls(ctx context.Context, index string, idx *Index,
return fmt.Errorf("unknown index %q specified in cross-index call", newIdxName)
}
}
if err := e.translateCall(newIdxName, newIdx, calls[i]); err != nil {
if err := e.translateCall(newIdxName, newIdx, calls[i], keyMap); err != nil {
return err
}
}
return nil
}
func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
var colKey, rowKey, fieldName string
switch c.Name {
case "Set", "Clear", "Row", "Range", "SetColumnAttrs", "ClearRow":
// Positional args in new PQL syntax require special handling here.
colKey = "_" + columnLabel
fieldName, _ = c.FieldArg()
rowKey = fieldName
case "SetRowAttrs":
// Positional args in new PQL syntax require special handling here.
rowKey = "_" + rowLabel
fieldName = callArgString(c, "_field")
case "Rows":
fieldName = callArgString(c, "_field")
rowKey = "previous"
colKey = "column"
case "GroupBy":
return errors.Wrap(e.translateGroupByCall(index, idx, c), "translating GroupBy")
case "IncludesColumn":
colKey = "column"
default:
colKey = "col"
fieldName = callArgString(c, "field")
rowKey = "row"
func (e *executor) collectCallIndexKeys(index string, idx *Index, c *pql.Call, keySet map[string]struct{}) error {
// Handle group by separately.
if c.Name == "GroupBy" {
for _, child := range c.Children {
if err := e.collectCallIndexKeys(index, idx, child, keySet); err != nil {
return errors.Wrapf(err, "translating %s", child)
}
}
if filter, ok, err := c.CallArg("filter"); ok {
if err != nil {
return errors.Wrap(err, "getting filter call")
}
err = e.collectCallIndexKeys(index, idx, filter, keySet)
if err != nil {
return errors.Wrap(err, "translating filter call")
}
}
return nil
}
colKey, _, _ := c.TranslateInfo(columnLabel, rowLabel)
if c.Args[colKey] != nil && !isString(c.Args[colKey]) {
return errors.New("column value must be a string when index 'keys' option enabled")
} else if value := callArgString(c, colKey); value != "" {
keySet[value] = struct{}{}
}
return nil
}
func (e *executor) translateCall(index string, idx *Index, c *pql.Call, keyMap map[string]uint64) error {
if c.Name == "GroupBy" {
return errors.Wrap(e.translateGroupByCall(index, idx, c, keyMap), "translating GroupBy")
}
// Translate column key.
colKey, rowKey, fieldName := c.TranslateInfo(columnLabel, rowLabel)
if idx.Keys() {
if c.Args[colKey] != nil && !isString(c.Args[colKey]) {
if !isValidID(c.Args[colKey]) {
return errors.Errorf("column value must be a string or non-negative integer, but got: %v of %[1]T", c.Args[colKey])
}
} else if value := callArgString(c, colKey); value != "" {
id, err := idx.translateStore.TranslateKey(value)
if err != nil {
return err
}
c.Args[colKey] = id
c.Args[colKey] = keyMap[value]
}
} else {
if isString(c.Args[colKey]) {
@ -3626,7 +3656,7 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
return errors.Errorf("row value must be a string or non-negative integer, but got: %v of %[1]T", c.Args[rowKey])
}
} else if value := callArgString(c, rowKey); value != "" {
id, err := field.translateStore.TranslateKey(value)
id, err := e.Cluster.translateFieldKey(index, fieldName, value)
if err != nil {
return err
}
@ -3654,7 +3684,7 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
return fmt.Errorf("unknown index %q specified in cross-index call", newIdxName)
}
}
if err := e.translateCall(newIdxName, newIdx, child); err != nil {
if err := e.translateCall(newIdxName, newIdx, child, keyMap); err != nil {
return err
}
}
@ -3662,13 +3692,13 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
return nil
}
func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call) error {
func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call, keyMap map[string]uint64) error {
if c.Name != "GroupBy" {
panic("translateGroupByCall called with '" + c.Name + "'")
}
for _, child := range c.Children {
if err := e.translateCall(index, idx, child); err != nil {
if err := e.translateCall(index, idx, child, keyMap); err != nil {
return errors.Wrapf(err, "translating %s", child)
}
}
@ -3677,7 +3707,7 @@ func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call) e
if err != nil {
return errors.Wrap(err, "getting filter call")
}
err = e.translateCall(index, idx, filter)
err = e.translateCall(index, idx, filter, keyMap)
if err != nil {
return errors.Wrap(err, "translating filter call")
}
@ -3687,7 +3717,7 @@ func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call) e
if err != nil {
return errors.Wrap(err, "getting aggregate call")
}
err = e.translateCall(index, idx, aggregate)
err = e.translateCall(index, idx, aggregate, keyMap)
if err != nil {
return errors.Wrap(err, "translating aggregate call")
}
@ -3722,7 +3752,7 @@ func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call) e
if !ok {
return errors.New("prev value must be a string when field 'keys' option enabled")
}
id, err := field.translateStore.TranslateKey(prevStr)
id, err := e.Cluster.translateFieldKey(index, field.Name(), prevStr)
if err != nil {
return errors.Wrapf(err, "translating row key '%s'", prevStr)
}
@ -3741,8 +3771,49 @@ func (e *executor) translateResults(ctx context.Context, index string, idx *Inde
span, _ := tracing.StartSpanFromContext(ctx, "Executor.translateResults")
defer span.Finish()
idMap := make(map[uint64]string)
if idx.Keys() {
// Collect all index ids.
idSet := make(map[uint64]struct{})
for i := range calls {
if err := e.collectResultIDs(index, idx, calls[i], results[i], idSet); err != nil {
return err
}
}
// Split ids by partition.
idsByPartition := make(map[int][]uint64, e.Cluster.partitionN)
for id := range idSet {
partitionID := e.Cluster.shardPartition(index, id/ShardWidth)
idsByPartition[partitionID] = append(idsByPartition[partitionID], id)
}
// Translate ids by partition.
var g errgroup.Group
var mu sync.Mutex
for partitionID, ids := range idsByPartition {
g.Go(func() error {
nodes := e.Cluster.partitionNodes(partitionID)
keys, err := e.client.TranslateIDsNode(ctx, &nodes[0].URI, index, "", ids)
if err != nil {
return err
}
mu.Lock()
defer mu.Unlock()
for i := range ids {
idMap[ids[i]] = keys[i]
}
return nil
})
}
if err := g.Wait(); err != nil {
return err
}
}
for i := range results {
results[i], err = e.translateResult(index, idx, calls[i], results[i])
results[i], err = e.translateResult(index, idx, calls[i], results[i], idMap)
if err != nil {
return err
}
@ -3750,18 +3821,30 @@ func (e *executor) translateResults(ctx context.Context, index string, idx *Inde
return nil
}
func (e *executor) translateResult(index string, idx *Index, call *pql.Call, result interface{}) (interface{}, error) {
func (e *executor) collectResultIDs(index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]struct{}) error {
row, ok := result.(*Row)
if !ok {
return nil
} else if !idx.Keys() {
return nil
}
for _, segment := range row.Segments() {
for _, col := range segment.Columns() {
idSet[col] = struct{}{}
}
}
return nil
}
func (e *executor) translateResult(index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]string) (interface{}, error) {
switch result := result.(type) {
case *Row:
if idx.Keys() {
other := &Row{Attrs: result.Attrs}
for _, segment := range result.Segments() {
for _, col := range segment.Columns() {
key, err := idx.translateStore.TranslateID(col)
if err != nil {
return nil, err
}
other.Keys = append(other.Keys, key)
other.Keys = append(other.Keys, idSet[col])
}
}
return other, nil
@ -3825,7 +3908,7 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res
return nil, ErrFieldNotFound
}
if field.keys() {
key, err := field.translateStore.TranslateID(g.RowID)
key, err := e.Cluster.translateFieldID(index, g.Field, g.RowID)
if err != nil {
return nil, errors.Wrap(err, "translating row ID in Group")
}
@ -3856,7 +3939,7 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res
} else if field.keys() {
other.Keys = make([]string, len(result))
for i, id := range result {
key, err := field.translateStore.TranslateID(id)
key, err := e.Cluster.translateFieldID(index, fieldName, id)
if err != nil {
return nil, errors.Wrap(err, "translating row ID")
}

18
executor_internal_test.go
View file

@ -25,8 +25,16 @@ import (
)
func TestExecutor_TranslateGroupByCall(t *testing.T) {
holder := NewHolder()
cluster := newCluster()
cluster.holder = holder
cluster.Node = &Node{ID: "node1", URI: NewTestURIFromHostPort("node1", 0)}
cluster.addNode(cluster.Node)
e := &executor{
Holder: NewHolder(),
Holder: holder,
Cluster: cluster,
}
e.Holder.Path, _ = ioutil.TempDir(*TempDir, "")
err := e.Holder.Open()
@ -46,12 +54,16 @@ func TestExecutor_TranslateGroupByCall(t *testing.T) {
t.Fatalf("creating fields %v, %v, %v", erra, errb, errc)
}
if err := cluster.updateTranslateStores(); err != nil {
t.Fatal(err)
}
query, err := pql.ParseString(`GroupBy(Rows(ak), Rows(b), Rows(ck), previous=["la", 0, "ha"], having=Condition(count > 10))`)
if err != nil {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
err = e.translateGroupByCall("i", idx, c)
err = e.translateGroupByCall("i", idx, c, make(map[string]uint64))
if err != nil {
t.Fatalf("translating call: %v", err)
}
@ -115,7 +127,7 @@ func TestExecutor_TranslateGroupByCall(t *testing.T) {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
err = e.translateGroupByCall("i", idx, c)
err = e.translateGroupByCall("i", idx, c, make(map[string]uint64))
if err == nil {
t.Fatalf("expected error, but translated call is '%s", c)
}

26
field.go
View file

@ -75,9 +75,6 @@ type Field struct {
// Row attribute storage and cache
rowAttrStore AttrStore
// Key/ID translation store.
translateStore TranslateStore
broadcaster broadcaster
Stats stats.StatsClient
@ -92,8 +89,6 @@ type Field struct {
logger logger.Logger
snapshotQueue snapshotQueue
// Instantiates new translation store on open.
OpenTranslateStore OpenTranslateStoreFunc
}
// FieldOption is a functional option type for pilosa.fieldOptions.
@ -265,8 +260,6 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) {
remoteAvailableShards: roaring.NewBitmap(),
logger: logger.NopLogger,
OpenTranslateStore: OpenInMemTranslateStore,
}
return f, nil
}
@ -280,12 +273,14 @@ func (f *Field) Index() string { return f.index }
// Path returns the path the field was initialized with.
func (f *Field) Path() string { return f.path }
// TranslateStorePath returns the translation database path for the field.
func (f *Field) TranslateStorePath() string {
return filepath.Join(f.path, "keys")
}
// RowAttrStore returns the attribute storage.
func (f *Field) RowAttrStore() AttrStore { return f.rowAttrStore }
// TranslateStore returns the underlying translation store for the field.
func (f *Field) TranslateStore() TranslateStore { return f.translateStore }
// AvailableShards returns a bitmap of shards that contain data.
func (f *Field) AvailableShards() *roaring.Bitmap {
f.mu.RLock()
@ -473,11 +468,6 @@ func (f *Field) Open() error {
return errors.Wrap(err, "opening attrstore")
}
// Instantiate & open translation store.
if f.translateStore, err = f.OpenTranslateStore(filepath.Join(f.path, "keys"), f.index, f.name); err != nil {
return errors.Wrap(err, "opening translate store")
}
return nil
}(); err != nil {
f.Close()
@ -726,12 +716,6 @@ func (f *Field) Close() error {
}
f.viewMap = make(map[string]*view)
if f.translateStore != nil {
if err := f.translateStore.Close(); err != nil {
return err
}
}
return nil
}

14
handler.go
View file

@ -212,3 +212,17 @@ type TranslateKeysRequest struct {
type TranslateKeysResponse struct {
IDs []uint64
}
// TranslateIDsRequest describes the structure of a request
// for a batch of id translations.
type TranslateIDsRequest struct {
Index string
Field string
IDs []uint64
}
// TranslateIDsResponse is the structured response of a id
// translation request.
type TranslateIDsResponse struct {
Keys []string
}

257
holder.go
View file

@ -78,12 +78,7 @@ type Holder struct {
snapshotQueue snapshotQueue
// Manages replication from the primary node.
primaryTranslateNode *Node
translateStoreReplicator *holderTranslateStoreReplicator
// Instantiates new translation stores for indexes & fields.
OpenTranslateStore OpenTranslateStoreFunc // local store
OpenTranslateReader OpenTranslateReaderFunc // replication
primaryTranslateNode *Node
}
// lockedChan looks a little ridiculous admittedly, but exists for good reason.
@ -128,8 +123,6 @@ func NewHolder() *Holder {
cacheFlushInterval: defaultCacheFlushInterval,
Logger: logger.NopLogger,
OpenTranslateStore: OpenInMemTranslateStore,
}
}
@ -232,13 +225,6 @@ func (h *Holder) Close() error {
h.opened.ch = make(chan struct{})
h.opened.mu.Unlock()
h.mu.Lock()
if h.translateStoreReplicator != nil {
h.translateStoreReplicator.Close()
h.translateStoreReplicator = nil
}
h.mu.Unlock()
return nil
}
@ -452,9 +438,6 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
// Update options.
h.indexes[index.Name()] = index
// Restart replication.
go h.refreshTranslateStoreReplicator()
return index, nil
}
@ -470,7 +453,6 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
index.snapshotQueue = h.snapshotQueue
index.holder = h
index.OpenTranslateStore = h.OpenTranslateStore
return index, nil
}
@ -667,243 +649,6 @@ func (h *Holder) logStartup() error {
return nil
}
// TranslateStore returns store for the given index or field.
func (h *Holder) TranslateStore(index, field string) (TranslateStore, error) {
if field == "" {
idx := h.Index(index)
if idx == nil {
return nil, ErrIndexNotFound
}
return idx.TranslateStore(), nil
}
f := h.Field(index, field)
if f == nil {
return nil, ErrFieldNotFound
}
return f.TranslateStore(), nil
}
// TranslateOffsetMap returns a map of offsets for all indexes & fields.
func (h *Holder) TranslateOffsetMap() (TranslateOffsetMap, error) {
m := make(TranslateOffsetMap)
for _, idx := range h.Indexes() {
id, err := idx.TranslateStore().MaxID()
if err != nil {
return nil, err
}
m.SetIndexOffset(idx.Name(), id+1)
for _, field := range idx.Fields() {
id, err := field.TranslateStore().MaxID()
if err != nil {
return nil, err
}
m.SetFieldOffset(idx.Name(), field.Name(), id+1)
}
}
return m, nil
}
func (h *Holder) setTranslateStoreReadOnly(v bool) {
for _, idx := range h.Indexes() {
idx.TranslateStore().SetReadOnly(v)
for _, field := range idx.Fields() {
field.TranslateStore().SetReadOnly(v)
}
}
}
func (h *Holder) setPrimaryTranslateStore(node *Node) error {
if node != nil && h.OpenTranslateReader == nil {
return nil
}
h.mu.Lock()
h.primaryTranslateNode = node.Clone()
h.mu.Unlock()
go h.refreshTranslateStoreReplicator()
return nil
}
func (h *Holder) refreshTranslateStoreReplicator() {
h.mu.RLock()
node := h.primaryTranslateNode
h.mu.RUnlock()
var nodeURL string
if node != nil {
u := node.URI.URL()
nodeURL = u.String()
}
// Stop existing replication, if running.
h.mu.Lock()
if h.translateStoreReplicator != nil {
h.translateStoreReplicator.Close()
h.translateStoreReplicator = nil
}
h.mu.Unlock()
// Set all stores read only mode based on if we have a primary.
h.setTranslateStoreReadOnly(node != nil)
// Start replication monitor, if needed.
h.mu.Lock()
defer h.mu.Unlock()
if nodeURL != "" {
h.translateStoreReplicator = newHolderTranslateStoreReplicator(h, nodeURL)
h.translateStoreReplicator.logger = h.Logger
if err := h.translateStoreReplicator.Open(); err != nil {
h.Logger.Printf("cannot open translate store replicator: %s", err)
}
}
}
// TranslateEntryReader returns a reader that merges all index & field reader
// that are specified in the offsets map.
func (h *Holder) TranslateEntryReader(ctx context.Context, offsets TranslateOffsetMap) (_ TranslateEntryReader, err error) {
// Ensure all readers are cleaned up if any error.
var a []TranslateEntryReader
defer func() {
if err != nil {
for i := range a {
a[i].Close()
}
}
}()
// Fetch all readers.
for indexName, m := range offsets {
for fieldName, offset := range m {
var store TranslateStore
idx := h.Index(indexName)
if idx == nil {
return nil, ErrIndexNotFound
}
// Fetch from index or field store.
if fieldName == "" {
store = idx.TranslateStore()
} else {
f := idx.Field(fieldName)
if f == nil {
return nil, ErrFieldNotFound
}
store = f.TranslateStore()
}
// Generate reader and append to multireader.
r, err := store.EntryReader(ctx, uint64(offset))
if err != nil {
return nil, errors.Wrap(err, "translate reader")
}
a = append(a, r)
}
}
return NewMultiTranslateEntryReader(ctx, a), nil
}
// holderTranslateStoreReplicator manages the replication of translation store
// data from a primary store to the local replica. Continually tries to
// reconnect on disconnect.
type holderTranslateStoreReplicator struct {
ctx context.Context
cancel func()
wg sync.WaitGroup
holder *Holder
nodeURL string
logger logger.Logger
}
func newHolderTranslateStoreReplicator(h *Holder, nodeURL string) *holderTranslateStoreReplicator {
r := &holderTranslateStoreReplicator{
holder: h,
nodeURL: nodeURL,
logger: logger.NopLogger,
}
r.ctx, r.cancel = context.WithCancel(context.Background())
return r
}
// Open starts the background monitoring goroutine.
func (r *holderTranslateStoreReplicator) Open() error {
r.wg.Add(1)
go func() { defer r.wg.Done(); r.monitor() }()
return nil
}
// Close stops the replicator.
func (r *holderTranslateStoreReplicator) Close() error {
r.cancel()
return nil
}
// monitor runs in a background goroutine and continually tries to connect and
// stream translate changes from the primary store.
func (r *holderTranslateStoreReplicator) monitor() {
for {
select {
case <-r.ctx.Done():
return
default:
if err := r.replicate(); err != nil {
r.logger.Printf("cannot replicate: nodeURL=%s err=%s", r.nodeURL, err)
}
time.Sleep(1 * time.Second)
}
}
}
func (r *holderTranslateStoreReplicator) replicate() error {
// Determine the offsets of every index & field store.
offsets, err := r.holder.TranslateOffsetMap()
if err != nil {
return err
} else if len(offsets) == 0 {
return nil
}
// Begin streaming from remote primary.
rd, err := r.holder.OpenTranslateReader(r.ctx, r.nodeURL, offsets)
if err != nil {
return err
}
defer rd.Close()
for {
var entry TranslateEntry
if err := rd.ReadEntry(&entry); err != nil {
return err
}
// Find appropriate store.
var store TranslateStore
if entry.Field == "" {
idx := r.holder.Index(entry.Index)
if idx == nil {
return ErrIndexNotFound
}
store = idx.TranslateStore()
} else {
f := r.holder.Field(entry.Index, entry.Field)
if f == nil {
return ErrFieldNotFound
}
store = f.TranslateStore()
}
// Apply replication to store.
if err := store.ForceSet(entry.ID, entry.Key); err != nil {
return err
}
}
}
// holderSyncer is an active anti-entropy tool that compares the local holder
// with a remote holder based on block checksums and resolves differences.
type holderSyncer struct {

100
http/client.go
View file

@ -1124,6 +1124,106 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg [
return errors.Wrap(resp.Body.Close(), "closing response body")
}
// TranslateKeysNode sends a key translation request to a specific node.
func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI, index, field string, keys []string) ([]uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "TranslateKeysNode")
defer span.Finish()
if index == "" {
return nil, pilosa.ErrIndexRequired
}
buf, err := c.serializer.Marshal(&pilosa.TranslateKeysRequest{
Index: index,
Field: field,
Keys: keys,
})
if err != nil {
return nil, errors.Wrap(err, "marshaling TranslateKeysRequest")
}
// Create HTTP request.
u := uri.Path("/internal/translate/keys")
req, err := http.NewRequest("POST", u, bytes.NewReader(buf))
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, err
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading")
}
tkresp := &pilosa.TranslateKeysResponse{}
if err := c.serializer.Unmarshal(body, tkresp); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
return tkresp.IDs, nil
}
// TranslateIDsNode sends an id translation request to a specific node.
func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI, index, field string, ids []uint64) ([]string, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "TranslateIDsNode")
defer span.Finish()
if index == "" {
return nil, pilosa.ErrIndexRequired
}
buf, err := c.serializer.Marshal(pilosa.TranslateIDsRequest{
Index: index,
Field: field,
IDs: ids,
})
if err != nil {
return nil, errors.Wrap(err, "marshaling TranslateIDsRequest")
}
// Create HTTP request.
u := uri.Path("/internal/translate/ids")
req, err := http.NewRequest("POST", u, bytes.NewReader(buf))
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, err
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading")
}
tkresp := &pilosa.TranslateIDsResponse{}
if err := c.serializer.Unmarshal(body, tkresp); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
return tkresp.Keys, nil
}
// executeRequest executes the given request and checks the Response. For
// responses with non-2XX status, the body is read and closed, and an error is
// returned. If the error is nil, the caller must ensure that the response body

2
http/handler.go
View file

@ -1752,7 +1752,7 @@ func (h *Handler) handlePostTranslateKeys(w http.ResponseWriter, r *http.Request
return
}
buf, err := h.api.TranslateKeys(r.Body)
buf, err := h.api.TranslateKeys(r.Context(), r.Body)
if err != nil {
http.Error(w, fmt.Sprintf("translate keys: %v", err), http.StatusInternalServerError)
}

33
index.go
View file

@ -21,6 +21,7 @@ import (
"os"
"path/filepath"
"sort"
"strconv"
"sync"
"time"
@ -52,8 +53,6 @@ type Index struct {
// Column attribute storage and cache.
columnAttrs AttrStore
translateStore TranslateStore
broadcaster broadcaster
Stats stats.StatsClient
@ -62,9 +61,6 @@ type Index struct {
// Used for notifying holder when a field is added.
holder *Holder
// Instantiates new translation stores for fields.
OpenTranslateStore OpenTranslateStoreFunc
}
// NewIndex returns a new instance of Index.
@ -86,8 +82,6 @@ func NewIndex(path, name string) (*Index, error) {
Stats: stats.NopStatsClient,
logger: logger.NopLogger,
trackExistence: true,
OpenTranslateStore: OpenInMemTranslateStore,
}, nil
}
@ -97,15 +91,17 @@ func (i *Index) Name() string { return i.name }
// Path returns the path the index was initialized with.
func (i *Index) Path() string { return i.path }
// TranslateStorePath returns the translation database path for a partition.
func (i *Index) TranslateStorePath(partitionID int) string {
return filepath.Join(i.path, "keys", strconv.Itoa(partitionID))
}
// Keys returns true if the index uses string keys.
func (i *Index) Keys() bool { return i.keys }
// ColumnAttrStore returns the storage for column attributes.
func (i *Index) ColumnAttrStore() AttrStore { return i.columnAttrs }
// TranslateStore returns the underlying translation store for the index.
func (i *Index) TranslateStore() TranslateStore { return i.translateStore }
// Options returns all options for this index.
func (i *Index) Options() IndexOptions {
i.mu.RLock()
@ -149,11 +145,6 @@ func (i *Index) Open() (err error) {
return errors.Wrap(err, "opening attrstore")
}
// Instantiate & open translation store.
if i.translateStore, err = i.OpenTranslateStore(filepath.Join(i.path, "keys"), i.name, ""); err != nil {
return errors.Wrap(err, "opening translate store")
}
return nil
}
@ -279,12 +270,6 @@ func (i *Index) Close() error {
}
i.fields = make(map[string]*Field)
if i.translateStore != nil {
if err := i.translateStore.Close(); err != nil {
return err
}
}
return nil
}
@ -445,11 +430,6 @@ func (i *Index) createField(name string, opt FieldOptions) (*Field, error) {
// Add to index's field lookup.
i.fields[name] = f
// Update replication, if needed.
if i.holder != nil {
go i.holder.refreshTranslateStoreReplicator()
}
return f, nil
}
@ -465,7 +445,6 @@ func (i *Index) newField(path, name string) (*Field, error) {
if i.snapshotQueue != nil {
f.snapshotQueue = i.snapshotQueue
}
f.OpenTranslateStore = i.OpenTranslateStore
return f, nil
}

2846
internal/private.pb.go
View file

File diff suppressed because it is too large Load diff

3191
internal/public.pb.go
View file

File diff suppressed because it is too large Load diff

10
internal/public.proto
View file

@ -132,6 +132,16 @@ message TranslateKeysResponse {
repeated uint64 IDs = 3;
}
message TranslateIDsRequest {
string Index = 1;
string Field = 2;
repeated uint64 IDs = 3;
}
message TranslateIDsResponse {
repeated string Keys = 3;
}
message ImportRoaringRequestView {
string Name = 1;
bytes Data = 2;

5
mock/translator.go
View file

@ -25,6 +25,7 @@ var _ pilosa.TranslateStore = (*TranslateStore)(nil)
type TranslateStore struct {
CloseFunc func() error
MaxIDFunc func() (uint64, error)
PartitionIDFunc func() int
ReadOnlyFunc func() bool
SetReadOnlyFunc func(v bool)
TranslateKeyFunc func(key string) (uint64, error)
@ -43,6 +44,10 @@ func (s *TranslateStore) MaxID() (uint64, error) {
return s.MaxIDFunc()
}
func (s *TranslateStore) PartitionID() int {
return s.PartitionIDFunc()
}
func (s *TranslateStore) ReadOnly() bool {
return s.ReadOnlyFunc()
}

28
pql/ast.go
View file

@ -742,6 +742,34 @@ func (c *Call) HasConditionArg() bool {
return false
}
// TranslateInfo returns the relevant translation fields.
func (c *Call) TranslateInfo(columnLabel, rowLabel string) (colKey, rowKey, fieldName string) {
switch c.Name {
case "Set", "Clear", "Row", "Range", "SetColumnAttrs", "ClearRow":
// Positional args in new PQL syntax require special handling here.
fieldName, _ = c.FieldArg()
return "_" + columnLabel, fieldName, fieldName
case "SetRowAttrs":
// Positional args in new PQL syntax require special handling here.
return "", "_" + rowLabel, c.ArgString("_field")
case "Rows":
return "column", "previous", c.ArgString("_field")
case "GroupBy":
return "", "", ""
default:
return "col", "row", c.ArgString("_field")
}
}
func (c *Call) ArgString(key string) string {
value, ok := c.Args[key]
if !ok {
return ""
}
s, _ := value.(string)
return s
}
// Condition represents an operation & value.
// When used in an argument map it represents a binary expression.
type Condition struct {

4
server.go
View file

@ -284,7 +284,7 @@ func OptServerClusterHasher(h Hasher) ServerOption {
// used to specify the translation data store type.
func OptServerOpenTranslateStore(fn OpenTranslateStoreFunc) ServerOption {
return func(s *Server) error {
s.holder.OpenTranslateStore = fn
s.cluster.OpenTranslateStore = fn
return nil
}
}
@ -293,7 +293,7 @@ func OptServerOpenTranslateStore(fn OpenTranslateStoreFunc) ServerOption {
// used to specify the remote translation data reader.
func OptServerOpenTranslateReader(fn OpenTranslateReaderFunc) ServerOption {
return func(s *Server) error {
s.holder.OpenTranslateReader = fn
s.cluster.OpenTranslateReader = fn
return nil
}
}

1
server/server.go
View file

@ -137,7 +137,6 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption
// Start starts the pilosa server - it returns once the server is running.
func (m *Command) Start() (err error) {
// Seed random number generator
rand.Seed(time.Now().UTC().UnixNano())

65
translate.go
View file

@ -28,6 +28,7 @@ var (
ErrTranslateStoreReaderClosed = errors.New("translate store reader closed")
ErrReplicationNotSupported = errors.New("replication not supported")
ErrTranslateStoreReadOnly = errors.New("translate store could not find or create key, translate store read only")
ErrTranslateStoreNotFound = errors.New("translate store not found")
ErrCannotOpenV1TranslateFile = errors.New("cannot open v1 translate .keys file")
)
@ -38,11 +39,18 @@ type TranslateStore interface {
// Returns the maximum ID set on the store.
MaxID() (uint64, error)
// Retrieves the partition ID associated with the store.
// Only applies to index stores.
PartitionID() int
// Sets & retrieves whether the store is read-only.
ReadOnly() bool
SetReadOnly(v bool)
// Converts a string key to its autoincrementing integer ID value.
//
// Translated id must be associated with a shard in the store's partition
// unless partition is set to -1.
TranslateKey(key string) (uint64, error)
TranslateKeys(key []string) ([]uint64, error)
@ -58,7 +66,7 @@ type TranslateStore interface {
}
// OpenTranslateStoreFunc represents a function for instantiating and opening a TranslateStore.
type OpenTranslateStoreFunc func(path, index, field string) (TranslateStore, error)
type OpenTranslateStoreFunc func(path, index, field string, partitionID int) (TranslateStore, error)
// TranslateEntryReader represents a stream of translation entries.
type TranslateEntryReader interface {
@ -154,22 +162,22 @@ type readEntryResponse struct {
}
// TranslateOffsetMap maintains a set of offsets for both indexes & fields.
type TranslateOffsetMap map[string]map[string]uint64
type TranslateOffsetMap map[string]*IndexTranslateOffsetMap
// IndexOffset returns the offset for the given index.
func (m TranslateOffsetMap) IndexOffset(name string) uint64 {
func (m TranslateOffsetMap) IndexPartitionOffset(name string, partitionID int) uint64 {
if m[name] == nil {
return 0
}
return m[name][""]
return m[name].Partitions[partitionID]
}
// SetIndexOffset sets the offset for the given index.
func (m TranslateOffsetMap) SetIndexOffset(name string, offset uint64) {
func (m TranslateOffsetMap) SetIndexPartitionOffset(name string, partitionID int, offset uint64) {
if m[name] == nil {
m[name] = make(map[string]uint64)
m[name] = NewIndexTranslateOffsetMap()
}
m[name][""] = offset
m[name].Partitions[partitionID] = offset
}
// FieldOffset returns the offset for the given field.
@ -177,15 +185,27 @@ func (m TranslateOffsetMap) FieldOffset(index, name string) uint64 {
if m[index] == nil {
return 0
}
return m[index][name]
return m[index].Fields[name]
}
// SetFieldOffset sets the offset for the given field.
func (m TranslateOffsetMap) SetFieldOffset(index, name string, offset uint64) {
if m[index] == nil {
m[index] = make(map[string]uint64)
m[index] = NewIndexTranslateOffsetMap()
}
m[index].Fields[name] = offset
}
type IndexTranslateOffsetMap struct {
Partitions map[int]uint64 `json:"partitions"`
Fields map[string]uint64 `json:"fields"`
}
func NewIndexTranslateOffsetMap() *IndexTranslateOffsetMap {
return &IndexTranslateOffsetMap{
Partitions: make(map[int]uint64),
Fields: make(map[string]uint64),
}
m[index][name] = offset
}
// Ensure type implements interface.
@ -193,19 +213,21 @@ var _ TranslateStore = &InMemTranslateStore{}
// InMemTranslateStore is an in-memory storage engine for mapping keys to int values.
type InMemTranslateStore struct {
mu sync.RWMutex
index string
field string
readOnly bool
keys []string
lookup map[string]uint64
mu sync.RWMutex
partitionID int
index string
field string
readOnly bool
keys []string
lookup map[string]uint64
writeNotify chan struct{}
}
// NewInMemTranslateStore returns a new instance of InMemTranslateStore.
func NewInMemTranslateStore(index, field string) *InMemTranslateStore {
func NewInMemTranslateStore(index, field string, partitionID int) *InMemTranslateStore {
return &InMemTranslateStore{
partitionID: partitionID,
index: index,
field: field,
lookup: make(map[string]uint64),
@ -217,14 +239,19 @@ var _ OpenTranslateStoreFunc = OpenInMemTranslateStore
// OpenInMemTranslateStore returns a new instance of InMemTranslateStore.
// Implements OpenTranslateStoreFunc.
func OpenInMemTranslateStore(rawurl, index, field string) (TranslateStore, error) {
return NewInMemTranslateStore(index, field), nil
func OpenInMemTranslateStore(rawurl, index, field string, partitionID int) (TranslateStore, error) {
return NewInMemTranslateStore(index, field, partitionID), nil
}
func (s *InMemTranslateStore) Close() error {
return nil
}
// PartitionID returns the partition id the store was initialized with.
func (s *InMemTranslateStore) PartitionID() int {
return s.partitionID
}
// ReadOnly returns true if the store is in read-only mode.
func (s *InMemTranslateStore) ReadOnly() bool {
s.mu.Lock()

4
translator_test.go
View file

@ -26,7 +26,7 @@ import (
)
func TestInMemTranslateStore_TranslateKey(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD")
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0)
// Ensure initial key translates to ID 1.
if id, err := s.TranslateKey("foo"); err != nil {
@ -51,7 +51,7 @@ func TestInMemTranslateStore_TranslateKey(t *testing.T) {
}
func TestInMemTranslateStore_TranslateID(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD")
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0)
// Setup initial keys.
if _, err := s.TranslateKey("foo"); err != nil {