Skip to content

Commit b0b5cbd

Browse files
nik9000nik9000
authored
Pull index routing into strategy object (#77211)
* Pull index routing into strategy object This pulls the calculation of the shard id for an (id, routing) pair into a little strategy class, `IndexRouting`. This is easier to test and should be easier to extend. My hope is that this is an incremental readability improvement. My ulterior motive is that this is where I want to land our new routing-by-dimensions work for tsdb. * WIP * Switch build check
1 parent 95a8c80 commit b0b5cbd

File tree

11 files changed

+561
-392
lines changed

11 files changed

+561
-392
lines changed

‎server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,13 @@
4242
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
4343
import org.elasticsearch.cluster.metadata.MappingMetadata;
4444
import org.elasticsearch.cluster.metadata.Metadata;
45+
import org.elasticsearch.cluster.routing.IndexRouting;
4546
import org.elasticsearch.cluster.service.ClusterService;
4647
import org.elasticsearch.common.inject.Inject;
4748
import org.elasticsearch.common.unit.ByteSizeUnit;
49+
import org.elasticsearch.common.util.concurrent.AtomicArray;
4850
import org.elasticsearch.core.Releasable;
4951
import org.elasticsearch.core.TimeValue;
50-
import org.elasticsearch.common.util.concurrent.AtomicArray;
5152
import org.elasticsearch.index.Index;
5253
import org.elasticsearch.index.IndexNotFoundException;
5354
import org.elasticsearch.index.IndexingPressure;
@@ -423,6 +424,7 @@ protected void doRun() {
423424
Metadata metadata = clusterState.metadata();
424425
// Group the requests by ShardId -> Operations mapping
425426
Map<ShardId, List<BulkItemRequest>> requestsByShard = new HashMap<>();
427+
Map<Index, IndexRouting> indexRoutings = new HashMap<>();
426428
for (int i = 0; i < bulkRequest.requests.size(); i++) {
427429
DocWriteRequest<?> docWriteRequest = bulkRequest.requests.get(i);
428430
//the request can only be null because we set it to null in the previous step, so it gets ignored
@@ -474,8 +476,13 @@ protected void doRun() {
474476
break;
475477
default: throw new AssertionError("request type not supported: [" + docWriteRequest.opType() + "]");
476478
}
477-
ShardId shardId = clusterService.operationRouting().indexShards(clusterState, concreteIndex.getName(),
478-
docWriteRequest.id(), docWriteRequest.routing()).shardId();
479+
IndexRouting indexRouting = indexRoutings.computeIfAbsent(
480+
concreteIndex,
481+
idx -> IndexRouting.fromIndexMetadata(clusterState.metadata().getIndexSafe(idx))
482+
);
483+
ShardId shardId = clusterService.operationRouting()
484+
.indexShards(clusterState, concreteIndex.getName(), indexRouting, docWriteRequest.id(), docWriteRequest.routing())
485+
.shardId();
479486
List<BulkItemRequest> shardRequests = requestsByShard.computeIfAbsent(shardId, shard -> new ArrayList<>());
480487
shardRequests.add(new BulkItemRequest(i, docWriteRequest));
481488
} catch (ElasticsearchParseException | IllegalArgumentException | RoutingMissingException e) {

‎server/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,19 +25,21 @@
2525
import org.elasticsearch.action.support.single.instance.TransportInstanceSingleOperationAction;
2626
import org.elasticsearch.client.node.NodeClient;
2727
import org.elasticsearch.cluster.ClusterState;
28+
import org.elasticsearch.cluster.metadata.IndexMetadata;
2829
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
2930
import org.elasticsearch.cluster.metadata.Metadata;
31+
import org.elasticsearch.cluster.routing.IndexRouting;
3032
import org.elasticsearch.cluster.routing.PlainShardIterator;
3133
import org.elasticsearch.cluster.routing.ShardIterator;
3234
import org.elasticsearch.cluster.routing.ShardRouting;
3335
import org.elasticsearch.cluster.service.ClusterService;
3436
import org.elasticsearch.common.bytes.BytesReference;
35-
import org.elasticsearch.core.Tuple;
3637
import org.elasticsearch.common.inject.Inject;
3738
import org.elasticsearch.common.io.stream.NotSerializableExceptionWrapper;
3839
import org.elasticsearch.common.io.stream.StreamInput;
3940
import org.elasticsearch.common.xcontent.XContentHelper;
4041
import org.elasticsearch.common.xcontent.XContentType;
42+
import org.elasticsearch.core.Tuple;
4143
import org.elasticsearch.index.IndexNotFoundException;
4244
import org.elasticsearch.index.IndexService;
4345
import org.elasticsearch.index.engine.VersionConflictEngineException;
@@ -157,8 +159,13 @@ protected ShardIterator shards(ClusterState clusterState, UpdateRequest request)
157159
if (request.getShardId() != null) {
158160
return clusterState.routingTable().index(request.concreteIndex()).shard(request.getShardId().getId()).primaryShardIt();
159161
}
162+
IndexMetadata indexMetadata = clusterState.metadata().index(request.concreteIndex());
163+
if (indexMetadata == null) {
164+
throw new IndexNotFoundException(request.concreteIndex());
165+
}
166+
IndexRouting indexRouting = IndexRouting.fromIndexMetadata(indexMetadata);
160167
ShardIterator shardIterator = clusterService.operationRouting()
161-
.indexShards(clusterState, request.concreteIndex(), request.id(), request.routing());
168+
.indexShards(clusterState, request.concreteIndex(), indexRouting, request.id(), request.routing());
162169
ShardRouting shard;
163170
while ((shard = shardIterator.nextOrNull()) != null) {
164171
if (shard.primary()) {

‎server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import com.carrotsearch.hppc.cursors.IntObjectCursor;
1313
import com.carrotsearch.hppc.cursors.ObjectCursor;
1414
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
15+
1516
import org.elasticsearch.Assertions;
1617
import org.elasticsearch.Version;
1718
import org.elasticsearch.action.admin.indices.rollover.RolloverInfo;
@@ -22,15 +23,8 @@
2223
import org.elasticsearch.cluster.block.ClusterBlock;
2324
import org.elasticsearch.cluster.block.ClusterBlockLevel;
2425
import org.elasticsearch.cluster.node.DiscoveryNodeFilters;
26+
import org.elasticsearch.cluster.routing.IndexRouting;
2527
import org.elasticsearch.cluster.routing.allocation.IndexMetadataUpdater;
26-
import org.elasticsearch.common.xcontent.ToXContent;
27-
import org.elasticsearch.common.xcontent.ToXContentFragment;
28-
import org.elasticsearch.common.xcontent.XContentBuilder;
29-
import org.elasticsearch.common.xcontent.XContentFactory;
30-
import org.elasticsearch.common.xcontent.XContentHelper;
31-
import org.elasticsearch.common.xcontent.XContentParser;
32-
import org.elasticsearch.common.xcontent.XContentParserUtils;
33-
import org.elasticsearch.core.Nullable;
3428
import org.elasticsearch.common.collect.ImmutableOpenIntMap;
3529
import org.elasticsearch.common.collect.ImmutableOpenMap;
3630
import org.elasticsearch.common.collect.MapBuilder;
@@ -41,6 +35,14 @@
4135
import org.elasticsearch.common.settings.Setting;
4236
import org.elasticsearch.common.settings.Setting.Property;
4337
import org.elasticsearch.common.settings.Settings;
38+
import org.elasticsearch.common.xcontent.ToXContent;
39+
import org.elasticsearch.common.xcontent.ToXContentFragment;
40+
import org.elasticsearch.common.xcontent.XContentBuilder;
41+
import org.elasticsearch.common.xcontent.XContentFactory;
42+
import org.elasticsearch.common.xcontent.XContentHelper;
43+
import org.elasticsearch.common.xcontent.XContentParser;
44+
import org.elasticsearch.common.xcontent.XContentParserUtils;
45+
import org.elasticsearch.core.Nullable;
4446
import org.elasticsearch.gateway.MetadataStateFormat;
4547
import org.elasticsearch.index.Index;
4648
import org.elasticsearch.index.mapper.MapperService;
@@ -1623,7 +1625,7 @@ public IndexMetadata fromXContent(XContentParser parser) throws IOException {
16231625

16241626
/**
16251627
* Returns the number of shards that should be used for routing. This basically defines the hash space we use in
1626-
* {@link org.elasticsearch.cluster.routing.OperationRouting#generateShardId(IndexMetadata, String, String)} to route documents
1628+
* {@link IndexRouting#shardId(String, String)} to route documents
16271629
* to shards based on their ID or their specific routing value. The default value is {@link #getNumberOfShards()}. This value only
16281630
* changes if and index is shrunk.
16291631
*/
@@ -1736,7 +1738,7 @@ public static Set<ShardId> selectShrinkShards(int shardId, IndexMetadata sourceI
17361738
/**
17371739
* Returns the routing factor for and shrunk index with the given number of target shards.
17381740
* This factor is used in the hash function in
1739-
* {@link org.elasticsearch.cluster.routing.OperationRouting#generateShardId(IndexMetadata, String, String)} to guarantee consistent
1741+
* {@link IndexRouting#shardId(String, String)} to guarantee consistent
17401742
* hashing / routing of documents even if the number of shards changed (ie. a shrunk index).
17411743
*
17421744
* @param sourceNumberOfShards the total number of shards in the source index

‎server/src/main/java/org/elasticsearch/cluster/routing/IndexRouting.java

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License
4+
* 2.0 and the Server Side Public License, v 1; you may not use this file except
5+
* in compliance with, at your election, the Elastic License 2.0 or the Server
6+
* Side Public License, v 1.
7+
*/
8+
9+
package org.elasticsearch.cluster.routing;
10+
11+
import org.elasticsearch.cluster.metadata.IndexMetadata;
12+
import org.elasticsearch.core.Nullable;
13+
14+
import java.util.function.IntConsumer;
15+
16+
/**
17+
* Generates the shard id for {@code (id, routing)} pairs.
18+
*/
19+
public abstract class IndexRouting {
20+
/**
21+
* Build the routing from {@link IndexMetadata}.
22+
*/
23+
public static IndexRouting fromIndexMetadata(IndexMetadata indexMetadata) {
24+
if (indexMetadata.isRoutingPartitionedIndex()) {
25+
return new Partitioned(
26+
indexMetadata.getRoutingNumShards(),
27+
indexMetadata.getRoutingFactor(),
28+
indexMetadata.getRoutingPartitionSize()
29+
);
30+
}
31+
return new Unpartitioned(indexMetadata.getRoutingNumShards(), indexMetadata.getRoutingFactor());
32+
}
33+
34+
private final int routingNumShards;
35+
private final int routingFactor;
36+
37+
private IndexRouting(int routingNumShards, int routingFactor) {
38+
this.routingNumShards = routingNumShards;
39+
this.routingFactor = routingFactor;
40+
}
41+
42+
/**
43+
* Generate the single shard id that should contain a document with the
44+
* provided {@code id} and {@code routing}.
45+
*/
46+
public abstract int shardId(String id, @Nullable String routing);
47+
48+
/**
49+
* Collect all of the shard ids that *may* contain documents with the
50+
* provided {@code routing}. Indices with a {@code routing_partition}
51+
* will collect more than one shard. Indices without a partition
52+
* will collect the same shard id as would be returned
53+
* by {@link #shardId}.
54+
*/
55+
public abstract void collectSearchShards(String routing, IntConsumer consumer);
56+
57+
/**
58+
* Convert a hash generated from an {@code (id, routing}) pair into a
59+
* shard id.
60+
*/
61+
protected final int hashToShardId(int hash) {
62+
return Math.floorMod(hash, routingNumShards) / routingFactor;
63+
}
64+
65+
/**
66+
* Convert a routing value into a hash.
67+
*/
68+
private static int effectiveRoutingToHash(String effectiveRouting) {
69+
return Murmur3HashFunction.hash(effectiveRouting);
70+
}
71+
72+
/**
73+
* Strategy for indices that are not partitioned.
74+
*/
75+
private static class Unpartitioned extends IndexRouting {
76+
Unpartitioned(int routingNumShards, int routingFactor) {
77+
super(routingNumShards, routingFactor);
78+
}
79+
80+
@Override
81+
public int shardId(String id, @Nullable String routing) {
82+
return hashToShardId(effectiveRoutingToHash(routing == null ? id : routing));
83+
}
84+
85+
@Override
86+
public void collectSearchShards(String routing, IntConsumer consumer) {
87+
consumer.accept(hashToShardId(effectiveRoutingToHash(routing)));
88+
}
89+
}
90+
91+
/**
92+
* Strategy for partitioned indices.
93+
*/
94+
private static class Partitioned extends IndexRouting {
95+
private final int routingPartitionSize;
96+
97+
Partitioned(int routingNumShards, int routingFactor, int routingPartitionSize) {
98+
super(routingNumShards, routingFactor);
99+
this.routingPartitionSize = routingPartitionSize;
100+
}
101+
102+
@Override
103+
public int shardId(String id, @Nullable String routing) {
104+
if (routing == null) {
105+
throw new IllegalArgumentException("A routing value is required for gets from a partitioned index");
106+
}
107+
int offset = Math.floorMod(effectiveRoutingToHash(id), routingPartitionSize);
108+
return hashToShardId(effectiveRoutingToHash(routing) + offset);
109+
}
110+
111+
@Override
112+
public void collectSearchShards(String routing, IntConsumer consumer) {
113+
int hash = effectiveRoutingToHash(routing);
114+
for (int i = 0; i < routingPartitionSize; i++) {
115+
consumer.accept(hashToShardId(hash + i));
116+
}
117+
}
118+
}
119+
}

‎server/src/main/java/org/elasticsearch/cluster/routing/OperationRouting.java

Lines changed: 32 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -11,11 +11,11 @@
1111
import org.elasticsearch.cluster.ClusterState;
1212
import org.elasticsearch.cluster.metadata.IndexMetadata;
1313
import org.elasticsearch.cluster.node.DiscoveryNodes;
14-
import org.elasticsearch.core.Nullable;
1514
import org.elasticsearch.common.Strings;
1615
import org.elasticsearch.common.settings.ClusterSettings;
1716
import org.elasticsearch.common.settings.Setting;
1817
import org.elasticsearch.common.settings.Settings;
18+
import org.elasticsearch.core.Nullable;
1919
import org.elasticsearch.index.IndexNotFoundException;
2020
import org.elasticsearch.index.shard.ShardId;
2121
import org.elasticsearch.node.ResponseCollectorService;
@@ -45,14 +45,30 @@ void setUseAdaptiveReplicaSelection(boolean useAdaptiveReplicaSelection) {
4545
this.useAdaptiveReplicaSelection = useAdaptiveReplicaSelection;
4646
}
4747

48-
public ShardIterator indexShards(ClusterState clusterState, String index, String id, @Nullable String routing) {
49-
return shards(clusterState, index, id, routing).shardsIt();
48+
public ShardIterator indexShards(
49+
ClusterState clusterState,
50+
String index,
51+
IndexRouting indexRouting,
52+
String id,
53+
@Nullable String routing
54+
) {
55+
return shards(clusterState, index, indexRouting, id, routing).shardsIt();
5056
}
5157

58+
/**
59+
* Shards to use for a {@code GET} operation.
60+
*/
5261
public ShardIterator getShards(ClusterState clusterState, String index, String id, @Nullable String routing,
5362
@Nullable String preference) {
54-
return preferenceActiveShardIterator(shards(clusterState, index, id, routing), clusterState.nodes().getLocalNodeId(),
55-
clusterState.nodes(), preference, null, null);
63+
IndexRouting indexRouting = IndexRouting.fromIndexMetadata(indexMetadata(clusterState, index));
64+
return preferenceActiveShardIterator(
65+
shards(clusterState, index, indexRouting, id, routing),
66+
clusterState.nodes().getLocalNodeId(),
67+
clusterState.nodes(),
68+
preference,
69+
null,
70+
null
71+
);
5672
}
5773

5874
public ShardIterator getShards(ClusterState clusterState, String index, int shardId, @Nullable String preference) {
@@ -100,18 +116,16 @@ private Set<IndexShardRoutingTable> computeTargetedShards(ClusterState clusterSt
100116
final Set<IndexShardRoutingTable> set = new HashSet<>();
101117
// we use set here and not list since we might get duplicates
102118
for (String index : concreteIndices) {
103-
final IndexRoutingTable indexRouting = indexRoutingTable(clusterState, index);
119+
final IndexRoutingTable indexRoutingTable = indexRoutingTable(clusterState, index);
104120
final IndexMetadata indexMetadata = indexMetadata(clusterState, index);
105-
final Set<String> effectiveRouting = routing.get(index);
106-
if (effectiveRouting != null) {
107-
for (String r : effectiveRouting) {
108-
final int routingPartitionSize = indexMetadata.getRoutingPartitionSize();
109-
for (int partitionOffset = 0; partitionOffset < routingPartitionSize; partitionOffset++) {
110-
set.add(RoutingTable.shardRoutingTable(indexRouting, calculateScaledShardId(indexMetadata, r, partitionOffset)));
111-
}
121+
final Set<String> indexSearchRouting = routing.get(index);
122+
if (indexSearchRouting != null) {
123+
IndexRouting indexRouting = IndexRouting.fromIndexMetadata(indexMetadata);
124+
for (String r : indexSearchRouting) {
125+
indexRouting.collectSearchShards(r, s -> set.add(RoutingTable.shardRoutingTable(indexRoutingTable, s)));
112126
}
113127
} else {
114-
for (IndexShardRoutingTable indexShard : indexRouting) {
128+
for (IndexShardRoutingTable indexShard : indexRoutingTable) {
115129
set.add(indexShard);
116130
}
117131
}
@@ -198,51 +212,20 @@ protected IndexRoutingTable indexRoutingTable(ClusterState clusterState, String
198212
return indexRouting;
199213
}
200214

201-
protected IndexMetadata indexMetadata(ClusterState clusterState, String index) {
215+
private IndexMetadata indexMetadata(ClusterState clusterState, String index) {
202216
IndexMetadata indexMetadata = clusterState.metadata().index(index);
203217
if (indexMetadata == null) {
204218
throw new IndexNotFoundException(index);
205219
}
206220
return indexMetadata;
207221
}
208222

209-
protected IndexShardRoutingTable shards(ClusterState clusterState, String index, String id, String routing) {
210-
int shardId = generateShardId(indexMetadata(clusterState, index), id, routing);
211-
return clusterState.getRoutingTable().shardRoutingTable(index, shardId);
223+
private IndexShardRoutingTable shards(ClusterState clusterState, String index, IndexRouting indexRouting, String id, String routing) {
224+
return clusterState.getRoutingTable().shardRoutingTable(index, indexRouting.shardId(id, routing));
212225
}
213226

214227
public ShardId shardId(ClusterState clusterState, String index, String id, @Nullable String routing) {
215228
IndexMetadata indexMetadata = indexMetadata(clusterState, index);
216-
return new ShardId(indexMetadata.getIndex(), generateShardId(indexMetadata, id, routing));
217-
}
218-
219-
public static int generateShardId(IndexMetadata indexMetadata, @Nullable String id, @Nullable String routing) {
220-
final String effectiveRouting;
221-
final int partitionOffset;
222-
223-
if (routing == null) {
224-
assert(indexMetadata.isRoutingPartitionedIndex() == false) : "A routing value is required for gets from a partitioned index";
225-
effectiveRouting = id;
226-
} else {
227-
effectiveRouting = routing;
228-
}
229-
230-
if (indexMetadata.isRoutingPartitionedIndex()) {
231-
partitionOffset = Math.floorMod(Murmur3HashFunction.hash(id), indexMetadata.getRoutingPartitionSize());
232-
} else {
233-
// we would have still got 0 above but this check just saves us an unnecessary hash calculation
234-
partitionOffset = 0;
235-
}
236-
237-
return calculateScaledShardId(indexMetadata, effectiveRouting, partitionOffset);
229+
return new ShardId(indexMetadata.getIndex(), IndexRouting.fromIndexMetadata(indexMetadata).shardId(id, routing));
238230
}
239-
240-
private static int calculateScaledShardId(IndexMetadata indexMetadata, String effectiveRouting, int partitionOffset) {
241-
final int hash = Murmur3HashFunction.hash(effectiveRouting) + partitionOffset;
242-
243-
// we don't use IMD#getNumberOfShards since the index might have been shrunk such that we need to use the size
244-
// of original index to hash documents
245-
return Math.floorMod(hash, indexMetadata.getRoutingNumShards()) / indexMetadata.getRoutingFactor();
246-
}
247-
248231
}

0 commit comments

Comments
 (0)