rm physical shard

Author: yao-xiao-github

fdbclient/ServerKnobs.cpp

@@ -1118,6 +1118,9 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
 	init( FETCH_SHARD_BUFFER_BYTE_LIMIT,                        20e6 ); if( randomize && BUGGIFY ) FETCH_SHARD_BUFFER_BYTE_LIMIT = 1;
 	init( FETCH_SHARD_UPDATES_BYTE_LIMIT,                    2500000 ); if( randomize && BUGGIFY ) FETCH_SHARD_UPDATES_BYTE_LIMIT = 100;
 
+	// Storage Server with Physical Shard
+	init( SS_GET_DATA_MOVE_ID, false); if ( isSimulated ) SS_GET_DATA_MOVE_ID = SHARD_ENCODE_LOCATION_METADATA;
+
 	//Wait Failure
 	init( MAX_OUTSTANDING_WAIT_FAILURE_REQUESTS,                 250 ); if( randomize && BUGGIFY ) MAX_OUTSTANDING_WAIT_FAILURE_REQUESTS = 2;
 	init( WAIT_FAILURE_DELAY_LIMIT,                              1.0 ); if( randomize && BUGGIFY ) WAIT_FAILURE_DELAY_LIMIT = 5.0;

fdbclient/include/fdbclient/ServerKnobs.h

@@ -1157,6 +1157,9 @@ class SWIFT_CXX_IMMORTAL_SINGLETON_TYPE ServerKnobs : public KnobsImpl<ServerKno
 	int FETCH_SHARD_BUFFER_BYTE_LIMIT;
 	int FETCH_SHARD_UPDATES_BYTE_LIMIT;
 
+	// Storage Server with Physical Shard
+	bool SS_GET_DATA_MOVE_ID;
+
 	// Wait Failure
 	int MAX_OUTSTANDING_WAIT_FAILURE_REQUESTS;
 	double WAIT_FAILURE_DELAY_LIMIT;

fdbclient/include/fdbclient/StorageCheckpoint.h

@@ -175,6 +175,7 @@ struct DataMoveMetaData {
 	int16_t phase; // DataMoveMetaData::Phase.
 	int8_t mode;
 	Optional<BulkLoadTaskState> bulkLoadTaskState; // set if the data move is a bulk load data move
+	Optional<std::unordered_map<std::string, std::string>> dcTeamIds; // map of dcId to teamId
 
 	DataMoveMetaData() = default;
 	DataMoveMetaData(UID id, Version version, KeyRange range) : id(id), version(version), priority(0), mode(0) {
@@ -202,7 +203,8 @@ struct DataMoveMetaData {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, id, version, ranges, priority, src, dest, checkpoints, phase, mode, bulkLoadTaskState);
+		serializer(
+		    ar, id, version, ranges, priority, src, dest, checkpoints, phase, mode, bulkLoadTaskState, dcTeamIds);
 	}
 };
 

fdbclient/include/fdbclient/StorageServerShard.h

@@ -104,6 +104,7 @@ struct StorageServerShard {
 	uint64_t desiredId; // The intended shard ID.
 	int8_t shardState;
 	Optional<UID> moveInShardId; // If present, it is the associated MoveInShardMetaData.
+	UID teamId;
 };
 
 #endif

fdbserver/MoveKeys.actor.cpp

@@ -1712,7 +1712,7 @@ ACTOR static Future<Void> startMoveShards(Database occ,
 					serverListEntries.push_back(tr.get(serverListKeyFor(servers[s])));
 				}
 				std::vector<Optional<Value>> serverListValues = wait(getAll(serverListEntries));
-
+				state std::unordered_map<std::string, std::vector<std::string>> dcServers;
 				for (int s = 0; s < serverListValues.size(); s++) {
 					if (!serverListValues[s].present()) {
 						// Attempt to move onto a server that isn't in serverList (removed or never added to the
@@ -1721,6 +1721,13 @@ ACTOR static Future<Void> startMoveShards(Database occ,
 						// TODO(psm): Mark the data move as 'deleting'.
 						throw move_to_removed_server();
 					}
+					auto si = decodeServerListValue(serverListValues[s].get());
+					ASSERT(si.id() == servers[s]);
+					auto it = dcServers.find(si.locality.describeDcId());
+					if (it == dcServers.end()) {
+						dcServers[si.locality.describeDcId()] = std::vector<std::string>();
+					}
+					dcServers[si.locality.describeDcId()].push_back(si.id().shortString());
 				}
 
 				currentKeys = KeyRangeRef(begin, keys.end);
@@ -1733,6 +1740,15 @@ ACTOR static Future<Void> startMoveShards(Database occ,
 					state Key endKey = old.back().key;
 					currentKeys = KeyRangeRef(currentKeys.begin, endKey);
 
+					if (ranges.front() != currentKeys) {
+						TraceEvent("MoveShardsPartialRange")
+						    .detail("ExpectedRange", ranges.front())
+						    .detail("ActualRange", currentKeys)
+						    .detail("DataMoveId", dataMoveId)
+						    .detail("RowLimit", SERVER_KNOBS->MOVE_SHARD_KRM_ROW_LIMIT)
+						    .detail("ByteLimit", SERVER_KNOBS->MOVE_SHARD_KRM_BYTE_LIMIT);
+					}
+
 					// Check that enough servers for each shard are in the correct state
 					state RangeResult UIDtoTagMap = wait(tr.getRange(serverTagKeys, CLIENT_KNOBS->TOO_MANY));
 					ASSERT(!UIDtoTagMap.more && UIDtoTagMap.size() < CLIENT_KNOBS->TOO_MANY);
@@ -1806,6 +1822,7 @@ ACTOR static Future<Void> startMoveShards(Database occ,
 									TraceEvent(
 									    SevWarn, "StartMoveShardsCancelConflictingDataMove", relocationIntervalId)
 									    .detail("Range", rangeIntersectKeys)
+									    .detail("CurrentDataMoveRange", ranges[0])
 									    .detail("DataMoveID", dataMoveId.toString())
 									    .detail("ExistingDataMoveID", destId.toString());
 									wait(cleanUpDataMove(occ, destId, lock, startMoveKeysLock, keys, ddEnabledState));
@@ -1868,6 +1885,16 @@ ACTOR static Future<Void> startMoveShards(Database occ,
 					dataMove.ranges.clear();
 					dataMove.ranges.push_back(KeyRangeRef(keys.begin, currentKeys.end));
 					dataMove.dest.insert(servers.begin(), servers.end());
+					dataMove.dcTeamIds = std::unordered_map<std::string, std::string>();
+					for (auto& [dc, serverIds] : dcServers) {
+						std::sort(serverIds.begin(), serverIds.end());
+						std::string teamId;
+						for (const auto& serverId : serverIds) {
+							teamId += serverId;
+						}
+						// Use the concatenated server ids as the team id to avoid conflicts.
+						dataMove.dcTeamIds.get()[dc] = teamId;
+					}
 				}
 
 				if (currentKeys.end == keys.end) {