· 10 years ago · Sep 06, 2016, 02:20 PM
1
2# entry point for shard rebalancer
3def rebalance_table_shards(relation regclass,
4 threshold float4 default 0.1,
5 max_shard_moves int default 1000000,
6 excluded_shard_list bigint[] default '{}'):
7
8 # TODO: should we acquire lock on collocationId to prevent concurrent rebalance operations
9
10 # acquire lock on the relation to prevent concurrent rebalance operations
11 # TODO: should we acquire lock on all collocated placements
12 AcquireLockForAllCollocatedTables(relation)
13
14 # get the required shard placement list
15 shard_placement_list = ShardPlacementList (relation, excluded_shard_list)
16
17 # get the active worker node list
18 reponsive_worker_node_list = master_get_active_worker_nodes() # {filter by worker_node_responsive}
19
20 rebalance_placement_update_list = shard_placement_rebalance_array (worker_node_list, shard_placement_list,
21 threshold))[1:max_shard_moves];
22
23 for placement_update_data in rebalance_placement_update_list:
24 PERFORM update_shard_placement(placement_update_data, relation, reponsive_worker_node_list)
25
26##
27## Connect to itself and execute master_copy_shard_placement in a seperate connection.
28## TODO: Do we really need relation parameter?
29##
30def update_shard_placement(placement_update_data, relation, reponsive_worker_node_list):
31
32 updateType, sshardId = placement_update_data.updateType, placement_update_data.shardId
33 sourceHost, sourcePort, do_drop = placement_update_data.sourceHost, placement_update_data.sourcePort
34 destHost, destPort = placement_update_data.destHost, placement_update_data.destPort
35 do_drop = placement_update_data.do_drop
36
37 # now, do the check here
38 if {source or destination} not in reponsive_worker_node_list:
39 ERROR: "source or destination is not responsive"
40
41 queryToExecute = "SELECT master_copy_shard_placement(%s, %s, %d, %s, %s, %d, %d)", shardId, sourceHost, sourcePort, destHost, destPort, doDrop
42
43 ExecuteRemoteCommand(LOCAL_HOST_NAME, PostPortNumber, queryToExecute)
44
45##
46## We already have master_copy_shard_placement(), which is to repair shards. So, we
47## remove the pre-requisits the function has. Also, we make it so that it becomes aware
48## of co-located placements.
49##
50## TODO: this implementation skips the tableId parameter since it is only used to get the
51## shardTableName, which we can generate without table name as well
52def master_copy_shard_placement(shardId, sourceHost, sourcePort, destHost, destPort, do_drop):
53
54 # get all collocatedPlacements
55 collocatedPlacamentList = CollocatedPlacementList(shardId, sourceHost, sourcePort)
56
57 # prepare for the rebalance operations
58 for collocatedPlacament in collocatedPlacamentList:
59
60 # acquire lock on the realtion id to prevent schema changes
61 # TODO: Is that OK to lock the relation both here and rebalance_table_shards()
62 AcquireShareTableLock(collocatedPlacament.relationId)
63
64 shardTableName = ShardTableName(collocatedPlacament.shardId);
65
66 # get only CREATE TABLE command without any constraints
67 appendStringInfo(createTableQuery, CREATE_TABLE_COMMAND, shardTableName, shardLength,
68 sourceHost, sourcePort);
69
70 appendStringInfo(dropTableQuery, DROP_TABLE_COMMAND, shardTableName)
71
72 # DROP all tables in a single command, then CREATE tables without the references
73 ExecuteCommandsOnTheHost({dropTableQuery, createTableQuery}, destHost, destPort)
74
75 # now that fetch the table without creating it, and then apply the constraints
76 for collocatedPlacament in collocatedPlacamentList:
77
78 bool createTable = False
79 AcquireShardMetadataLock(collocatedPlacament.shardId)
80
81 shardTableName = ShardTableName(collocatedPlacament.shardId);
82
83 # fetch table without CREATING THE TABLE
84 appendStringInfo(dataFetchQuery, TABLE_FETCH_COMMAND, shardTableName, shardLength,
85 sourceHost, sourcePort, createTable);
86
87 appendStringInfo(createConstraintsCommand, CREATE_CONSTRAINTS_COMMAND, shardTableName)
88
89 # go and drop the table if already exists, and fetch the fresh table
90 ExecuteCommandsOnTheHost({dataFetchQuery, createConstraintsCommand}, destHost, destPort)
91
92 # now update the metadata
93 InsertIntoShardPlacement(shardId, destHost, destPort)
94
95 # if drop specified, dropping the placement done in 2PC
96 if do_drop:
97 DeleteShardPlacement(shardId, destHost, destPort)
98
99 appendStringInfo(dropTableQuery, DROP_TABLE_COMMAND, shardTableName)
100
101 # we have to run this via 2PC so that placement is not dropped
102 # until the metadata changes committed
103 ExecuteCommandVia2PC(dropTableQuery, sourceHost, sourcePort)
104
105
106###
107### A psudo-code for shard_placement_rebalance_array
108### Note that some parts (i.e., mapping function parameters to the variables, mapping return list to JSON etc.)
109### But, still gives a good overview of what is going on while rebalancing.
110###
111def shard_placement_rebalance_array():
112
113 # get shard placement list for each of the nodes
114 # Format:
115 # [nodeIndex: 0] => {S1,S2, ..., SN}
116 # [nodeIndex: 1] => {S4,S5, ..., SX}
117 # [nodeIndex: 2] => {S2,S4, ..., SY}
118 nodeShardPlacementList[] = GenerateNodeShardList(tableName)
119
120 # first get the total placement count
121 totalPlacementCount = TotalPlacementCount(nodeShardPlacementList)
122
123 # get the average placement count
124 placementCountAverage = totalPlacementCount / workerNodeCount
125
126 # get the lower bound for any of the placement
127 placementCountLowerbound = (int) ((1.0 - threshold) * placementCountAverage);
128
129 # now itedate over the worker nodes to get each worker to have at least
130 # placementCountLowerbound placements
131 while (underUtilizedNodeExists)
132
133 # find the workers to operat on this iteration
134 workerThatHasMaxCountOfPlacement = findWorkerWithMaxPlacementCount(nodeShardPlacementList)
135 workerThatHasMinCountOfPlacement = findWorkerWithMinPlacementCount(nodeShardPlacementList)
136
137 # we already hit the expected distribution, do not proceed
138 if workerCount(workerThatHasMinCountOfPlacement) >= placementCountLowerbound
139 break
140
141 # now find the placement to move
142 # (i.e., the target does not already have the same placement)
143 for placement in workerThatHasMaxCountOfPlacement
144
145 # skip this placement if already exists in the min worker
146 if placement in workerThatHasMinCountOfPlacement:
147 continue
148
149 # now that we found the placement to move
150 placementToMove = placement
151 break
152
153 # create an event to update placements
154 placementUpdateEvent = PlacementUpdateEvent(placementToMove,
155 sourceNode: workerThatHasMinCountOfPlacement,
156 targetNode: workerThatHasMaxCountOfPlacement)
157
158 # add this placement update event to the return list
159 placementUpdateList.append(placementUpdateEvent)
160
161 # update the shard placement list according to the move
162 UpdateShardLists(nodeShardPlacementList)
163
164 return placementUpdateList