From 12882ad14743ff3f377e47ae52d050f45dbaa1fb Mon Sep 17 00:00:00 2001
From: Seebs <seebs@molecula.com>
Date: Mon, 27 Sep 2021 11:54:02 -0500
Subject: [PATCH] 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.
---
 api.go | 21 +++++++++++++++------
 1 file changed, 15 insertions(+), 6 deletions(-)

diff --git a/api.go b/api.go
index 1caa34873..ca3506a0e 100644
--- a/api.go
+++ b/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 {