mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
handle replication
I assumed the existing import code handled replicas. It doesn't, actually. It just assumes they're handled. So, in the new import code, when splitting things up by-shard, send each shard's data to *every* node that has that shard, not just the first one.
This commit is contained in:
parent
b3f82ac894
commit
12882ad147
1 changed files with 15 additions and 6 deletions
21
api.go
21
api.go
|
|
@ -1942,16 +1942,25 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string
|
|||
if len(snap.Nodes) == 1 {
|
||||
return api.ingestNodeOperationsForFields(ctx, qcx, index, knownFields, sharded)
|
||||
}
|
||||
// split up by fields in some way
|
||||
// Created new ShardedRequest objects for every node, giving each of them
|
||||
// all the shards that apply to them.
|
||||
byNode := make(map[string]*ingest.ShardedRequest)
|
||||
for shard, ops := range sharded.Ops {
|
||||
nodes := snap.ShardNodes(indexName, shard)
|
||||
forThisShard := byNode[nodes[0].ID]
|
||||
if forThisShard == nil {
|
||||
byNode[nodes[0].ID] = &ingest.ShardedRequest{Ops: map[uint64][]*ingest.Operation{shard: ops}}
|
||||
continue
|
||||
for _, node := range nodes {
|
||||
forThisShard := byNode[node.ID]
|
||||
if forThisShard == nil {
|
||||
// Create new ShardedRequest for the target node, with its op map
|
||||
// mapping this shard to the ops for this shard.
|
||||
byNode[node.ID] = &ingest.ShardedRequest{Ops: map[uint64][]*ingest.Operation{shard: ops}}
|
||||
continue
|
||||
}
|
||||
// Add this shard to the existing ShardedRequest's Ops map. Note that
|
||||
// we don't have to worry about overwrites; we can't have seen this
|
||||
// shard before, because we're in a range loop on a map where the shard
|
||||
// is the key.
|
||||
forThisShard.Ops[shard] = ops
|
||||
}
|
||||
forThisShard.Ops[shard] = ops
|
||||
}
|
||||
eg, ctx := errgroup.WithContext(ctx)
|
||||
for _, node := range snap.Nodes {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue