You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "J-HowHuang (via GitHub)" <gi...@apache.org> on 2025/02/22 02:52:58 UTC
[PR] Add minimizeDataMovement to Rebalacne API [pinot]
J-HowHuang opened a new pull request, #15110:
URL: https://github.com/apache/pinot/pull/15110
Add new boolean property `minimizeDataMovement` to `org.apache.pinot.controller.helix.core.rebalance.RebalanceConfig`.
This flag enforces [minimal data movement algorithm](https://docs.pinot.apache.org/operators/operating-pinot/rebalance/rebalance-servers#minimal-data-movement-algorithm) if set to true. Otherwise, fall back to the `minimizeDataMovement` tag set in the table config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984345215
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
oh, now I understand what your suggestion is @Jackie-Jiang, I think I misunderstood before. I think your suggestion makes sense, especially the use case to try out and see the difference between `true` and `false`. I know I've wanted to do this while testing to see when it even matters and when it doesn't apply. @J-HowHuang let's model this flag as: `true`, `false`, `use table config` - you'll need to figure out how to model this in `PinotTableRestletResource`
(btw good catch on missing `PinotTableRestletResource`, somehow I completely missed this in the review - @J-HowHuang please add the flag there otherwise it won't show up in swagger - please test this out in swagger locally on your machine)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972438842
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,226 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+ @Test
+ public void testRebalanceWithMinimizeDataMovementBalanced() throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
+
+
+ // Check if the instance assignment is the same as the one without minimizeDataMovement flag set
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(false);
+ RebalanceResult rebalanceResultWithoutMinimized = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ assertEquals(rebalanceResult.getInstanceAssignment(), rebalanceResultWithoutMinimized.getInstanceAssignment());
+
+ // Rebalance
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ // Should see the added server in the instance assignment
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getInstances(0, 0).size(), numServers + 1);
+
+ }
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // One instance per replica group, no partition
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS, 1, 0, 0, false,
+ null), null, false);
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig)).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(RAW_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
Review Comment:
nit: is this the correct table? `OFFLINE_TIERED_TABLE_NAME`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985795069
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -83,6 +83,13 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _bestEfforts = false;
+ // Whether to enforce Minimal Data Movement Algorithm (only effective if instance assignment config is set, and if
+ // bootstrap is false). If set to false, the minimizeDataMovement flag in the table config will be used to determine
+ // whether to run the Minimal Data Movement Algorithm.
+ @JsonProperty("minimizeDataMovement")
+ @ApiModelProperty(example = "TRUE")
+ private String _minimizeDataMovement = "TRUE";
Review Comment:
I just tried this locally and it works fine (and correctly converts the string and sets it as a Boolean in the RebalanceConfig). `ApiModel` is for annotating the parameters to be passed in. `ApiModelProperty` is for the model used to hold the properties. I don't believe there has to be a 1-1 mapping
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972439379
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,226 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+ @Test
+ public void testRebalanceWithMinimizeDataMovementBalanced() throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
Review Comment:
nit: is this the correct table? `OFFLINE_TIERED_TABLE_NAME`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1978439542
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
Should we model this as a override, i.e. pass in a `Boolean` object, and override the value in table config when it is set; use the value in table config when it is `null`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984398264
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
I don't think this will be only for debugging (but I can't say how this will be used by customers). Will it be overly complicated to add the behavior recommended?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince merged PR #15110:
URL: https://github.com/apache/pinot/pull/15110
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985770339
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -83,6 +83,13 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _bestEfforts = false;
+ // Whether to enforce Minimal Data Movement Algorithm (only effective if instance assignment config is set, and if
+ // bootstrap is false). If set to false, the minimizeDataMovement flag in the table config will be used to determine
+ // whether to run the Minimal Data Movement Algorithm.
+ @JsonProperty("minimizeDataMovement")
+ @ApiModelProperty(example = "TRUE")
+ private String _minimizeDataMovement = "TRUE";
Review Comment:
This class `RebalanceConfig` is annotated with Swagger's `@ApiModel` so its member variable here (annotated with `@ApiModelProperty`) is supposed to reflect as it is in the API?
I'll have to make another `Boolean` member variable without `@ApiModelProperty` and then convert the `String` variable read from API to the new `Boolean` variable in `RebalanceConfig` class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984400252
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,6 +615,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
+ @ApiParam(value = "If `minimizeDataMovement` is set to false in the table config, set this to true to enforce"
+ + " minimizeDataMovement during the rebalance.") @DefaultValue("true")
+ @QueryParam("forceMinimizeDataMovement") boolean forceMinimizeDataMovement,
Review Comment:
yeah that's also why it may be a better idea to allow the 3 values for this flag: a) true -> set minimizeDataMovement, b) false -> don't set minimizeDataMovement, c) none -> default to table config
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985758187
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
Review Comment:
no, this is still relevant. minimizeDataMovement might not be carried out if there's no existing instance partition or when `bootstrap=true`. See https://github.com/apache/pinot/blob/8dab53ae4efd4475cbd23b7b808fb684018a4cb0/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceTagPoolSelector.java#L55
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on PR #15110:
URL: https://github.com/apache/pinot/pull/15110#issuecomment-2709255220
@J-HowHuang can you also add the new `minimizeDataMovement` option to `RebalanceTableCommand` and set it to `ENABLE` by default there as well? do validate it by running the command locally to ensure it is able to parse correctly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972435892
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
Review Comment:
you're still using `OFFLINE_TIERED_TABLE_NAME`, is this just for testing a different tenant? if so it may be a good idea to either use a non-tier based tenant if this isn't testing tiering, or at least add a comment on what the purpose of this is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972443473
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,226 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+ @Test
+ public void testRebalanceWithMinimizeDataMovementBalanced() throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
+
+
+ // Check if the instance assignment is the same as the one without minimizeDataMovement flag set
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(false);
+ RebalanceResult rebalanceResultWithoutMinimized = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ assertEquals(rebalanceResult.getInstanceAssignment(), rebalanceResultWithoutMinimized.getInstanceAssignment());
+
+ // Rebalance
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ // Should see the added server in the instance assignment
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getInstances(0, 0).size(), numServers + 1);
Review Comment:
i see some cleanup done for the instance assignment based test, is that needed here too? or is that not needed there?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #15110:
URL: https://github.com/apache/pinot/pull/15110#issuecomment-2679977418
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/15110?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 55.90%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`1bb364c`)](https://app.codecov.io/gh/apache/pinot/commit/1bb364cf30e5e42671f459dbb781848cf8ff54bb?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1753 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (1bb364c). Click for more details.
>
> <details><summary>HEAD has 48 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (1bb364c) |
>|------|------|------|
>|integration|7|0|
>|integration2|3|0|
>|temurin|12|2|
>|java-21|7|2|
>|skip-bytebuffers-true|3|1|
>|skip-bytebuffers-false|7|1|
>|unittests|5|2|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #15110 +/- ##
============================================
- Coverage 61.75% 55.90% -5.85%
- Complexity 207 711 +504
============================================
Files 2436 2149 -287
Lines 133233 115662 -17571
Branches 20636 18584 -2052
============================================
- Hits 82274 64665 -17609
- Misses 44911 45680 +769
+ Partials 6048 5317 -731
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.90% <ø> (-5.72%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.87% <ø> (-5.88%)` | :arrow_down: |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.89% <ø> (+28.16%)` | :arrow_up: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.90% <ø> (-5.85%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.90% <ø> (-5.84%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.90% <ø> (+9.01%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/15110/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/15110?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984407159
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985799022
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
+ // if existingInstancePartitions is null.
+ boolean minimizeDataMovement =
+ minimizeDataMovementFlag == null ? instanceAssignmentConfig.isMinimizeDataMovement() : minimizeDataMovementFlag;
+ LOGGER.info("Starting {} instance assignment for table {}, instanceAssignmentConfig.isMinimizeDataMovement()={}, "
+ + "minimizeDataMovement={}", instancePartitionsName, tableNameWithType,
Review Comment:
So this log might be looked at for debugging issues, so it should contain all the relevant information that we might need for that purpose. It should include the following:
- Table config value
- Boolean override value
- value we set based on the above
So my ask was to add all three, and for the Boolean, call it the "Override"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985901053
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -83,6 +83,13 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _bestEfforts = false;
+ // Whether to enforce Minimal Data Movement Algorithm (only effective if instance assignment config is set, and if
+ // bootstrap is false). If set to false, the minimizeDataMovement flag in the table config will be used to determine
+ // whether to run the Minimal Data Movement Algorithm.
+ @JsonProperty("minimizeDataMovement")
+ @ApiModelProperty(example = "TRUE")
+ private String _minimizeDataMovement = "TRUE";
Review Comment:
Ended up with an Enum method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972321436
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -1303,7 +1303,7 @@ private void handleLegacySchemaConfig(TableConfig tableConfig, HttpHeaders httpH
private void validateInstanceAssignment(TableConfig tableConfig) {
TableRebalancer tableRebalancer = new TableRebalancer(_pinotHelixResourceManager.getHelixZkManager());
try {
- tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true);
+ tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true, false);
Review Comment:
8a02789
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972440509
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,226 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+ @Test
+ public void testRebalanceWithMinimizeDataMovementBalanced() throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
+
+
+ // Check if the instance assignment is the same as the one without minimizeDataMovement flag set
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(false);
+ RebalanceResult rebalanceResultWithoutMinimized = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ assertEquals(rebalanceResult.getInstanceAssignment(), rebalanceResultWithoutMinimized.getInstanceAssignment());
+
+ // Rebalance
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ // Should see the added server in the instance assignment
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getInstances(0, 0).size(), numServers + 1);
+
+ }
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // One instance per replica group, no partition
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS, 1, 0, 0, false,
+ null), null, false);
+
+ // Create the table with default balanced segment assignment
Review Comment:
nit: is this comment correct? thought this test is for replicaGroups?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984378250
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
Would the debugging scenario you mentioned be ad hoc or intended to make it a feature? If it's only for ad hoc testing and playing around, I would still suggest that the one who wants to test go find a table with `minimizeDataMovement=false` or modify the table config.
Imho the interface is explanatory and should be designed as the way that fits the intended future use.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984396711
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -655,6 +658,7 @@ public RebalanceResult rebalance(
rebalanceConfig.setPreChecks(preChecks);
rebalanceConfig.setReassignInstances(reassignInstances);
rebalanceConfig.setIncludeConsuming(includeConsuming);
+ rebalanceConfig.setForceMinimizeDataMovement(forceMinimizeDataMovement);
Review Comment:
same, remove "force" from the name of the config and the getter/setters
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985859192
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,6 +615,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
+ @ApiParam(value = "If `minimizeDataMovement` is set to false in the table config, set this to true to enforce"
+ + " minimizeDataMovement during the rebalance.") @DefaultValue("true")
+ @QueryParam("forceMinimizeDataMovement") boolean forceMinimizeDataMovement,
Review Comment:
Discussed with @J-HowHuang and recommended exploring enum (ENABLE, DISABLE, DEFAULT) instead of this string boolean thing.
cc @Jackie-Jiang any thoughts on modeling the API to take 3 values? I kind of think string is clunky here, but I'm okay with keeping this as string if there isn't a better way
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985796437
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
+ // if existingInstancePartitions is null.
+ boolean minimizeDataMovement =
+ minimizeDataMovementFlag == null ? instanceAssignmentConfig.isMinimizeDataMovement() : minimizeDataMovementFlag;
+ LOGGER.info("Starting {} instance assignment for table {}, instanceAssignmentConfig.isMinimizeDataMovement()={}, "
+ + "minimizeDataMovement={}", instancePartitionsName, tableNameWithType,
Review Comment:
The log message is showing the resolved `boolean` minimizeDataMovement parameter, not the `Boolean` overriding parameter. The overriding value should be found in `TableRebalancer`'s logger.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968483886
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
just capturing offline discussion here:
`InstanceTagPoolSelector::selectInstances` uses the `_minimizeDataMovement` flag:
```
if (_minimizeDataMovement) {
assert _existingInstancePartitions != null;
Map<Integer, Integer> poolToNumExistingInstancesMap = new TreeMap<>();
...
...
```
Based on the above, if we blindly pass in `_minimizeDataMovement` even for balanced (default) instance assignment, then this assert will fail since `_existingInstancePartitions` should be null. So we should only set `_minimizeDataMovement` to 'true' if `allowInstanceAssignment` != `false`
@J-HowHuang to verify this via testing as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968443719
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
I think in `InstanceAssignmentConfigUtils::getInstanceAssignmentConfig` it handles the case when instance config is not set (this covers cases when `allowInstanceAssignment` is false) by falling back to look at the `minimizeDataMovement` flag in segment validation config.
Can we safely assume that `InstanceAssignmentConfigUtils::getInstanceAssignmentConfig` is always called when making `InstanceAssignmentConfig` instances?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
I think in `InstanceAssignmentConfigUtils::getInstanceAssignmentConfig` it handles the case when instance config is not set (this covers cases when `allowInstanceAssignment` is false) by falling back to look at the `minimizeDataMovement` flag in segment validation config.
Can we safely assume that `InstanceAssignmentConfigUtils::getInstanceAssignmentConfig` is always called when making `InstanceAssignmentConfig` instances?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968634440
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
sounds good. please just verify that rebalance with default instance assignment and replica group instance assignment both work in that case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972258491
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + numServers, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
+
+ // Rebalance
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ // Should see the added server in the instance assignment
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getInstances(0, 0).size(), numServers + 1);
+
+ // One instance per replica group, no partition
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS, 1, 0, 0, false,
+ null), null, false);
+
+ tableConfig.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig));
+ _helixResourceManager.updateTableConfig(tableConfig);
+
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // With instance assignment config, the number of servers is reduced to the number of replica groups
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers + 1);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), NUM_REPLICAS);
+
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getNumReplicaGroups(), NUM_REPLICAS);
+
+
+ // increase replica group size by 1
+ instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS + 1, 1, 0, 0, false,
+ null), null, true);
+
+ tableConfig.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig));
+ _helixResourceManager.updateTableConfig(tableConfig);
+
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ rebalanceServerInfo = rebalanceResult.getRebalanceSummaryResult().getServerInfo();
+
+ // with minimize data movement, we should add 1 server only
Review Comment:
It was not able, but it's fixed now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972259795
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + numServers, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
Review Comment:
a299268
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
Review Comment:
a299268
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972321929
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,67 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, false);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
0e7e98b
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985774736
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
+ // if existingInstancePartitions is null.
+ boolean minimizeDataMovement =
+ minimizeDataMovementFlag == null ? instanceAssignmentConfig.isMinimizeDataMovement() : minimizeDataMovementFlag;
+ LOGGER.info("Starting {} instance assignment for table {}, instanceAssignmentConfig.isMinimizeDataMovement()={}, "
+ + "minimizeDataMovement={}", instancePartitionsName, tableNameWithType,
Review Comment:
you mean the comment or the log msg?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985772549
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -191,17 +191,21 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
boolean bestEfforts = rebalanceConfig.isBestEfforts();
long externalViewCheckIntervalInMs = rebalanceConfig.getExternalViewCheckIntervalInMs();
long externalViewStabilizationTimeoutInMs = rebalanceConfig.getExternalViewStabilizationTimeoutInMs();
+ String minimizeDataMovementStr = rebalanceConfig.getMinimizeDataMovement();
+ Boolean minimizeDataMovement =
+ minimizeDataMovementStr.toLowerCase().matches("^(true|false)$") ? Boolean.valueOf(minimizeDataMovementStr)
Review Comment:
Right now all values other than "true" or "false" (case insensitive) will be considered as "default", no exception will be raised, is this okay?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984248686
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
@Jackie-Jiang Consider that enabling `minimizeDataMovement` is always encouraged, based on the philosophy I think for this API it's better to be kept like this (only to override false with true but not the other way round).
If anyone wants to override a table config that enables `minimizeDataMovement` with false by all means, he can modify the table config per se.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1970330120
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -604,7 +814,8 @@ public void testMirrorServerSetBasedRandomInner(int loopCount) throws FileNotFou
}
System.out.println("");
InstancePartitions instancePartitions =
- driver.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, existing, preConfigured);
+ driver.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, existing, preConfigured,
Review Comment:
I see you added method that passes `false` to assignInstances(), if so, I assume these changes to pass `false` when calling assignInstances() can be avoided?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
We can print both `forceMinimizeDataMovement` and `instanceAssignmentConfig.isMinimizeDataMovement()` in this log msg, to help debug
```
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984400252
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,6 +615,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
+ @ApiParam(value = "If `minimizeDataMovement` is set to false in the table config, set this to true to enforce"
+ + " minimizeDataMovement during the rebalance.") @DefaultValue("true")
+ @QueryParam("forceMinimizeDataMovement") boolean forceMinimizeDataMovement,
Review Comment:
yeah that's also why it may be a better idea to allow the 3 values for this flag: a) true -> set minimizeDataMovement, b) false -> don't set minimizeDataMovement, c) none -> default to table config
it makes the expected behavior much clearer and captures the user's actual intent. otherwise users will need to think about what's in table config. e.g. say i set this to false, but it is true in table config, now have to change in 2 places
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984399204
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,6 +615,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
+ @ApiParam(value = "If `minimizeDataMovement` is set to false in the table config, set this to true to enforce"
+ + " minimizeDataMovement during the rebalance.") @DefaultValue("true")
+ @QueryParam("forceMinimizeDataMovement") boolean forceMinimizeDataMovement,
Review Comment:
It would be possible to have minimizeDataMovement algorithm take place though `minimizeDataMovement=false`, which is misleading. That's why I added "force" to clarify it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985790604
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
+ // if existingInstancePartitions is null.
+ boolean minimizeDataMovement =
+ minimizeDataMovementFlag == null ? instanceAssignmentConfig.isMinimizeDataMovement() : minimizeDataMovementFlag;
+ LOGGER.info("Starting {} instance assignment for table {}, instanceAssignmentConfig.isMinimizeDataMovement()={}, "
+ + "minimizeDataMovement={}", instancePartitionsName, tableNameWithType,
Review Comment:
i mean in the log message
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984400252
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,6 +615,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
+ @ApiParam(value = "If `minimizeDataMovement` is set to false in the table config, set this to true to enforce"
+ + " minimizeDataMovement during the rebalance.") @DefaultValue("true")
+ @QueryParam("forceMinimizeDataMovement") boolean forceMinimizeDataMovement,
Review Comment:
yeah that's also why it may be a better idea to allow the 3 values for this flag: a) true -> set minimizeDataMovement, b) false -> don't set minimizeDataMovement, c) none -> default to table config
it makes the expected behavior much clearer and captures the user's actual intent. otherwise users will need to think about what's in table config. e.g. say i set this to false, but it is true in table config, now have to change in 2 places if my intent is to disable it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985741776
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -83,6 +83,13 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _bestEfforts = false;
+ // Whether to enforce Minimal Data Movement Algorithm (only effective if instance assignment config is set, and if
+ // bootstrap is false). If set to false, the minimizeDataMovement flag in the table config will be used to determine
+ // whether to run the Minimal Data Movement Algorithm.
+ @JsonProperty("minimizeDataMovement")
+ @ApiModelProperty(example = "TRUE")
+ private String _minimizeDataMovement = "TRUE";
Review Comment:
Can we keep this `Boolean` here?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
Review Comment:
nit: let's rename it to `minimizeDataMovementOverride`?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -51,9 +51,7 @@
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.fail;
+import static org.testng.Assert.*;
Review Comment:
nit: I usually avoid "*" imports. can you just add the newly added functions here?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -651,6 +654,7 @@ public RebalanceResult rebalance(
rebalanceConfig.setPreChecks(preChecks);
rebalanceConfig.setReassignInstances(reassignInstances);
rebalanceConfig.setIncludeConsuming(includeConsuming);
+ rebalanceConfig.setMinimizeDataMovement(minimizeDataMovement);
Review Comment:
Can the conversion of this String to `Boolean` be done here itself? i.e. the parsing for TRUE/FALSE and DEFAULT
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
+ // if existingInstancePartitions is null.
+ boolean minimizeDataMovement =
+ minimizeDataMovementFlag == null ? instanceAssignmentConfig.isMinimizeDataMovement() : minimizeDataMovementFlag;
+ LOGGER.info("Starting {} instance assignment for table {}, instanceAssignmentConfig.isMinimizeDataMovement()={}, "
+ + "minimizeDataMovement={}", instancePartitionsName, tableNameWithType,
Review Comment:
nit: in the comment change `minimizeDataMovement=` to `minimizeDataMovementOverride=`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -191,17 +191,21 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
boolean bestEfforts = rebalanceConfig.isBestEfforts();
long externalViewCheckIntervalInMs = rebalanceConfig.getExternalViewCheckIntervalInMs();
long externalViewStabilizationTimeoutInMs = rebalanceConfig.getExternalViewStabilizationTimeoutInMs();
+ String minimizeDataMovementStr = rebalanceConfig.getMinimizeDataMovement();
+ Boolean minimizeDataMovement =
+ minimizeDataMovementStr.toLowerCase().matches("^(true|false)$") ? Boolean.valueOf(minimizeDataMovementStr)
Review Comment:
let's move this to `PinotTableRestletResource`. Also throw an exception if the value is not one of "true" "false" or "default" from there itself.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -858,6 +867,227 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+ @Test
+ public void testRebalanceWithMinimizeDataMovementBalanced()
+ throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster("minimizeDataMovement_balance_" + SERVER_INSTANCE_ID_PREFIX + i,
+ true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(RAW_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement("true");
Review Comment:
can you add a test for `setMinimizeDataMovement("false")` where TableConfig indicates it is true?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
Review Comment:
nit: do you need to cast null to `Boolean` here?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementFlag) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
Review Comment:
does this comment need to be updated to reflect the new change?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -83,6 +83,13 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _bestEfforts = false;
+ // Whether to enforce Minimal Data Movement Algorithm (only effective if instance assignment config is set, and if
+ // bootstrap is false). If set to false, the minimizeDataMovement flag in the table config will be used to determine
Review Comment:
nit: update comment, that if set to DEFAULT we fallback on table config, otherwise use this as the override value
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968443719
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
I think in `InstanceAssignmentConfigUtils::getInstanceAssignmentConfig` it handles the case when instance config is not set (this covers cases when `allowInstanceAssignment` is false) by falling back to look at the minimizeDataMovement flag in segment validation config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968611873
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
```java
public InstanceTagPoolSelector(InstanceTagPoolConfig tagPoolConfig, String tableNameWithType,
boolean minimizeDataMovement, @Nullable InstancePartitions existingInstancePartitions) {
...
_minimizeDataMovement = minimizeDataMovement && existingInstancePartitions != null;
_existingInstancePartitions = existingInstancePartitions;
}
```
It does the check in the constructor of `InstanceTagPoolSelector` so it should not be a problem. Though need to document this to manifest the behavior in the api docs, especially the case that `existingInstancePartitions` is set to `null` when `bootstrap=true` in the rebalance api
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972234452
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
+ // This test case is using the same instance rebalance plot as testMinimizeDataMovement, and test whether
+ // forceMinimizeDataMovement flag in InstanceAssignmentDriver works as the minimizeDataMovement flag in
+ // TableConfig does.
+ int numReplicas = 3;
+ int numPartitions = 2;
+ int numInstancesPerPartition = 2;
+ String partitionColumn = "partition";
+
+ // Configs and driver that minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, true);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(numReplicas)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig))
+ .build();
+
+ InstanceAssignmentDriver driver = new InstanceAssignmentDriver(tableConfig);
+
+ // Configs and driver that DO NOT minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfigNotMinimized = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, false);
+
+ TableConfig tableConfigNotMinimized = new TableConfig(tableConfig);
+ tableConfigNotMinimized.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfigNotMinimized));
+ InstanceAssignmentDriver driverNotMinimized = new InstanceAssignmentDriver(tableConfigNotMinimized);
+
+
+ int numInstances = 10;
+ List<InstanceConfig> instanceConfigs = new ArrayList<>(numInstances);
+ for (int i = 0; i < numInstances; i++) {
+ InstanceConfig instanceConfig = new InstanceConfig(SERVER_INSTANCE_ID_PREFIX + i);
+ instanceConfig.addTag(OFFLINE_TAG);
+ instanceConfigs.add(instanceConfig);
+ }
+
+ // Instances should be assigned to 3 replica-groups with a round-robin fashion, each with 2 instances
+ InstancePartitions instancePartitions =
+ driver.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, null, false);
+
+ InstancePartitions instancePartitionsForcedMinimize =
+ driverNotMinimized.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, null, true);
+
+ InstancePartitions instancePartitionsNotMinimize =
+ driverNotMinimized.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, null, false);
+
+ // Assignment without existing InstancePartitions should be the same for all scenarios
Review Comment:
"Initial assignment should be the same for all scenarios" would be clearer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1970742135
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
+ // This test case is using the same instance rebalance plot as testMinimizeDataMovement, and test whether
+ // forceMinimizeDataMovement flag in InstanceAssignmentDriver works as the minimizeDataMovement flag in
+ // TableConfig does.
+ int numReplicas = 3;
+ int numPartitions = 2;
+ int numInstancesPerPartition = 2;
+ String partitionColumn = "partition";
+
+ // Configs and driver that minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, true);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(numReplicas)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig))
+ .build();
+
+ InstanceAssignmentDriver driver = new InstanceAssignmentDriver(tableConfig);
+
+ // Configs and driver that DO NOT minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfigNotMinimized = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, false);
+
+ TableConfig tableConfigNotMinimized = new TableConfig(tableConfig);
+ tableConfigNotMinimized.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfigNotMinimized));
+ InstanceAssignmentDriver driverNotMinimized = new InstanceAssignmentDriver(tableConfigNotMinimized);
Review Comment:
can you add an assert that when you check if this TableConfig has minimizeDataMovement = false?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + numServers, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
Review Comment:
is it possible to validate that the expected changes are the same for `setMinimizeDataMovement(true)` and `setMinimizeDataMovement(false)`?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
+ // This test case is using the same instance rebalance plot as testMinimizeDataMovement, and test whether
+ // forceMinimizeDataMovement flag in InstanceAssignmentDriver works as the minimizeDataMovement flag in
+ // TableConfig does.
+ int numReplicas = 3;
+ int numPartitions = 2;
+ int numInstancesPerPartition = 2;
+ String partitionColumn = "partition";
+
+ // Configs and driver that minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, true);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(numReplicas)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig))
+ .build();
Review Comment:
can you add an assert that when you check if this TableConfig has minimizeDataMovement = true?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
Review Comment:
nit: is this still a TODO?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
+ // This test case is using the same instance rebalance plot as testMinimizeDataMovement, and test whether
+ // forceMinimizeDataMovement flag in InstanceAssignmentDriver works as the minimizeDataMovement flag in
+ // TableConfig does.
+ int numReplicas = 3;
+ int numPartitions = 2;
+ int numInstancesPerPartition = 2;
+ String partitionColumn = "partition";
+
+ // Configs and driver that minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, true);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(numReplicas)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig))
+ .build();
+
+ InstanceAssignmentDriver driver = new InstanceAssignmentDriver(tableConfig);
+
+ // Configs and driver that DO NOT minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfigNotMinimized = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, false);
+
+ TableConfig tableConfigNotMinimized = new TableConfig(tableConfig);
+ tableConfigNotMinimized.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfigNotMinimized));
+ InstanceAssignmentDriver driverNotMinimized = new InstanceAssignmentDriver(tableConfigNotMinimized);
+
+
+ int numInstances = 10;
+ List<InstanceConfig> instanceConfigs = new ArrayList<>(numInstances);
+ for (int i = 0; i < numInstances; i++) {
+ InstanceConfig instanceConfig = new InstanceConfig(SERVER_INSTANCE_ID_PREFIX + i);
+ instanceConfig.addTag(OFFLINE_TAG);
+ instanceConfigs.add(instanceConfig);
+ }
+
+ // Instances should be assigned to 3 replica-groups with a round-robin fashion, each with 2 instances
+ InstancePartitions instancePartitions =
+ driver.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, null, false);
+
+ InstancePartitions instancePartitionsForcedMinimize =
+ driverNotMinimized.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, null, true);
+
+ InstancePartitions instancePartitionsNotMinimize =
+ driverNotMinimized.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, null, false);
+
+ // Assignment without existing InstancePartitions should be the same for all scenarios
Review Comment:
what does this comment mean?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,67 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, false);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
Can you also add a comment here stating that blindly setting `minimizeDataMovement` to `true` is okay, and why (which is that for balanced mode there is a check that resets this if instance partitions don't actually exist)
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -1303,7 +1303,7 @@ private void handleLegacySchemaConfig(TableConfig tableConfig, HttpHeaders httpH
private void validateInstanceAssignment(TableConfig tableConfig) {
TableRebalancer tableRebalancer = new TableRebalancer(_pinotHelixResourceManager.getHelixZkManager());
try {
- tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true);
+ tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true, false);
Review Comment:
let's use the method signature without the `forceMinimizeDataMovement`?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
Review Comment:
nit: are you testing tiering here? why the tiered table name? i'd recommend adding a separate test for tiering if you want to test tier and non-tier
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,67 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, false);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
+ if (forceMinimizeDataMovement) {
+ LOGGER.info("Table: {} is forced with minimizeDataMovement", tableNameWithType);
+ } else if (instanceAssignmentConfig.isMinimizeDataMovement()) {
+ LOGGER.info("Table: {} has configured minimizeDataMovement", tableNameWithType);
+ } else {
+ LOGGER.warn("Table: {} is being rebalanced without minimizeDataMovement", tableNameWithType);
+ }
Review Comment:
Let's just add a single log line where you print `tableNameWithType`, `instanceAssignmentConfig.isMinimizeDataMovement()`, and `forceMinimizeDataMovement`. No need for this kind of if-else.
Also, enough to keep this an info log for now
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
Review Comment:
can we add a similar test for balanced - just show that it is always the same (pick 1 scenario like increasing RF)?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
Review Comment:
can we split this into 2 tests - 1 for balanced assignment and 1 for replicaGroup assignment? it'll be easier to read then
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + numServers, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
+
+ // Rebalance
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ // Should see the added server in the instance assignment
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getInstances(0, 0).size(), numServers + 1);
+
+ // One instance per replica group, no partition
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS, 1, 0, 0, false,
+ null), null, false);
+
+ tableConfig.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig));
+ _helixResourceManager.updateTableConfig(tableConfig);
+
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // With instance assignment config, the number of servers is reduced to the number of replica groups
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers + 1);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), NUM_REPLICAS);
+
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getNumReplicaGroups(), NUM_REPLICAS);
+
+
+ // increase replica group size by 1
+ instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS + 1, 1, 0, 0, false,
+ null), null, true);
+
+ tableConfig.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig));
+ _helixResourceManager.updateTableConfig(tableConfig);
+
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ rebalanceServerInfo = rebalanceResult.getRebalanceSummaryResult().getServerInfo();
+
+ // with minimize data movement, we should add 1 server only
Review Comment:
what happens for this test case if we use `setMinimizeDataMovement(false)`? Similarly for the case when a replica is removed? Just wondering if your test is able to capture the difference between the two modes or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985895623
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -612,9 +612,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
- @ApiParam(value = "TRUE|FALSE|DEFAULT, whether to enable minimize data movement on rebalance, DEFAULT will use "
- + "the minimizeDataMovement in table config") @DefaultValue("TRUE")
- @QueryParam("minimizeDataMovement") String minimizeDataMovement,
+ @ApiParam(value = "Whether to enable minimize data movement on rebalance, DEFAULT will use "
+ + "the minimizeDataMovement in table config") @DefaultValue("ENABLE")
+ @QueryParam("minimizeDataMovement") RebalanceConfig.MinimizeDataMovementOptions minimizeDataMovement,
Review Comment:
it returns HTTP 404
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984335201
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
@somandal What's your opinion on this? I agree in most cases `minimizeDataMovement` is preferred, but it would be more flexible if we allow users to override the value to false without updating table config, even for debugging purpose to see the difference with/without minimize movement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985760510
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,65 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
Review Comment:
There will be an ambiguity with this method if type was not specified:
https://github.com/apache/pinot/blob/28dfb4a54eed7affad67eaa52fc61629816e4cd6/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java#L72-L74
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985795968
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -191,17 +191,21 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
boolean bestEfforts = rebalanceConfig.isBestEfforts();
long externalViewCheckIntervalInMs = rebalanceConfig.getExternalViewCheckIntervalInMs();
long externalViewStabilizationTimeoutInMs = rebalanceConfig.getExternalViewStabilizationTimeoutInMs();
+ String minimizeDataMovementStr = rebalanceConfig.getMinimizeDataMovement();
+ Boolean minimizeDataMovement =
+ minimizeDataMovementStr.toLowerCase().matches("^(true|false)$") ? Boolean.valueOf(minimizeDataMovementStr)
Review Comment:
I didn't understand - here you're checking if they are true or false and you use that, otherwise set it to null. I'm saying if you expect the 3rd value to be "DEFAULT" then check for that explicitly and throw an exception as a parameter check failed kind of error
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1994044837
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,66 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, (Boolean) null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
@Nullable InstancePartitions preConfiguredInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions,
+ preConfiguredInstancePartitions, null);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ minimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ return assignInstances(tierName, instanceConfigs, existingInstancePartitions, instanceAssignmentConfig, null);
+ }
+
+ public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ @Nullable Boolean minimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, minimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, @Nullable Boolean minimizeDataMovementOverride) {
String tableNameWithType = _tableConfig.getTableName();
- LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ // minimizeDataMovement might be set back to false within InstanceTagPoolSelector and InstancePartitionSelector
+ // if existingInstancePartitions is null.
+ boolean minimizeDataMovement =
+ minimizeDataMovementOverride == null ? instanceAssignmentConfig.isMinimizeDataMovement()
+ : minimizeDataMovementOverride;
+ LOGGER.info("Starting {} instance assignment for table {}, instanceAssignmentConfig.isMinimizeDataMovement()={}, "
+ + "minimizeDataMovementOverride={}, minimizeDataMovement={}", instancePartitionsName, tableNameWithType,
Review Comment:
nit: it's not enforced but most log msgs use this convention `... minimizeDataMovementOverride: {}` instead of `=`
and better rename `instanceAssignmentConfig.isMinimizeDataMovement()=` to something more readable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972269922
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
Review Comment:
https://github.com/apache/pinot/commit/a2992680f7d47553b08e8dda9ff6da92076a1747
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968428441
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
Review Comment:
[5c68e49](https://github.com/apache/pinot/pull/15110/commits/5c68e49acb5517f9dc09c8d1f2c02d87e703039b)
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -185,17 +185,18 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
boolean bestEfforts = rebalanceConfig.isBestEfforts();
long externalViewCheckIntervalInMs = rebalanceConfig.getExternalViewCheckIntervalInMs();
long externalViewStabilizationTimeoutInMs = rebalanceConfig.getExternalViewStabilizationTimeoutInMs();
+ boolean forceMinimizeDataMovement = rebalanceConfig.isMinimizeDataMovement();
boolean enableStrictReplicaGroup = tableConfig.getRoutingConfig() != null
&& RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE.equalsIgnoreCase(
tableConfig.getRoutingConfig().getInstanceSelectorType());
LOGGER.info(
"Start rebalancing table: {} with dryRun: {}, preChecks: {}, reassignInstances: {}, includeConsuming: {}, "
+ "bootstrap: {}, downtime: {}, minReplicasToKeepUpForNoDowntime: {}, enableStrictReplicaGroup: {}, "
+ "lowDiskMode: {}, bestEfforts: {}, externalViewCheckIntervalInMs: {}, "
- + "externalViewStabilizationTimeoutInMs: {}",
+ + "externalViewStabilizationTimeoutInMs: {}, forceMinimizeDataMovement: {}",
Review Comment:
[8e9dccf](https://github.com/apache/pinot/pull/15110/commits/8e9dccfa74f974bb11e63f10875625c16420d525)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on PR #15110:
URL: https://github.com/apache/pinot/pull/15110#issuecomment-2683530503
> Is it possible to add a rebalance specific test case as well for the new flag?
>
> `OfflineClusterIntegrationTest` had some tests with dry-run, but can probably be enhanced to add one for this. You can also look at `PartialUpsertTableRebalanceIntegrationTest` as an example
>
> `TableRebalancerClusterStatelessTest` might be a good one too
Added test for rebalance with the flag in `TableRebalancerClusterStatelessTest`. Please look at 46f398f
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968349627
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
Review Comment:
recommend adding separate method signatures with the added parameter. Have the existing one pass 'forceMinimizeDataMovement' as 'false'
e.g.
```
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
}
```
That way you only need to modify the Rebalance code to call the newly added method signatures. Remaining code that doesn't need to pass this flag can continue using the original signature.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -185,17 +185,18 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
boolean bestEfforts = rebalanceConfig.isBestEfforts();
long externalViewCheckIntervalInMs = rebalanceConfig.getExternalViewCheckIntervalInMs();
long externalViewStabilizationTimeoutInMs = rebalanceConfig.getExternalViewStabilizationTimeoutInMs();
+ boolean forceMinimizeDataMovement = rebalanceConfig.isMinimizeDataMovement();
boolean enableStrictReplicaGroup = tableConfig.getRoutingConfig() != null
&& RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE.equalsIgnoreCase(
tableConfig.getRoutingConfig().getInstanceSelectorType());
LOGGER.info(
"Start rebalancing table: {} with dryRun: {}, preChecks: {}, reassignInstances: {}, includeConsuming: {}, "
+ "bootstrap: {}, downtime: {}, minReplicasToKeepUpForNoDowntime: {}, enableStrictReplicaGroup: {}, "
+ "lowDiskMode: {}, bestEfforts: {}, externalViewCheckIntervalInMs: {}, "
- + "externalViewStabilizationTimeoutInMs: {}",
+ + "externalViewStabilizationTimeoutInMs: {}, forceMinimizeDataMovement: {}",
Review Comment:
nit: in the log just use `minimizeDataMovement` as that is the config name
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
Also, there are some types of `TableConfigs` for which it is not allowed to set an instance config.
There is a method `InstanceAssignmentConfigUtils::allowInstanceAssignment` which checks for this. I think it makes sense to only set `forceMinimizeDataMovement` to `true` if instance assignment config is even allowed. You can perhaps log a warning if `forceMinimizeDataMovement` is set to `true` but instance assignment isn't allowed, and just set `minimizeDataMovement` to `false` in that case
nit: perhaps also add a comment that explains all of the above
cc @Jackie-Jiang @klsince can you folks confirm if my understanding of the above is correct?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -83,6 +83,12 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _bestEfforts = false;
+ // Whether to enforce Minimal Data Movement Algorithm. If set to false, the minimizeDataMovement flag in the table
+ // config will be used to determine whether to run the Minimal Data Movement Algorithm.
+ @JsonProperty("minimizeDataMovement")
+ @ApiModelProperty(example = "true")
+ private boolean _minimizeDataMovement = true;
Review Comment:
Just a thought, should we start with defaulting this to `false`, test it out in a cluster with a few different scenarios, and then add a new change to default this to `true` once we confirm it always does the right thing?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968483886
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
just capturing offline discussion here:
`InstanceTagPoolSelector::selectInstances` uses the `_minimizeDataMovement` flag:
```
if (_minimizeDataMovement) {
assert _existingInstancePartitions != null;
Map<Integer, Integer> poolToNumExistingInstancesMap = new TreeMap<>();
...
...
```
Based on the above, if we blindly pass in `_minimizeDataMovement` even for balanced (default) instance assignment, then this assert will fail since `_existingInstancePartitions` should be null. So we should only set `_minimizeDataMovement` to 'true' if `allowInstanceAssignment` != `false`
@J-HowHuang to verify this via testing as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972442713
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,226 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+ @Test
+ public void testRebalanceWithMinimizeDataMovementBalanced() throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(TIERED_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX, true);
+
+ // Table without instance assignment config should work fine (ignore) with the minimizeDataMovement flag set
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ // Should see the added server
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceServerInfo.getNumServers().getValueBeforeRebalance(), numServers);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), numServers + 1);
+
+
+ // Check if the instance assignment is the same as the one without minimizeDataMovement flag set
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(false);
+ RebalanceResult rebalanceResultWithoutMinimized = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ assertEquals(rebalanceResult.getInstanceAssignment(), rebalanceResultWithoutMinimized.getInstanceAssignment());
+
+ // Rebalance
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ // Should see the added server in the instance assignment
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getInstances(0, 0).size(), numServers + 1);
+
+ }
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // One instance per replica group, no partition
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS, 1, 0, 0, false,
+ null), null, false);
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig)).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadataWithEndTimeInfo(RAW_TABLE_NAME, SEGMENT_NAME_PREFIX + i,
+ nowInDays), null);
+ }
+
+ Map<String, Map<String, String>> oldSegmentAssignment =
+ _helixResourceManager.getTableIdealState(OFFLINE_TIERED_TABLE_NAME).getRecord().getMapFields();
+
+ TableRebalancer tableRebalancer = new TableRebalancer(_helixManager);
+
+ // Try dry-run summary mode
+ RebalanceConfig rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ RebalanceResult rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+
+ RebalanceSummaryResult rebalanceSummaryResult = rebalanceResult.getRebalanceSummaryResult();
+ assertNotNull(rebalanceSummaryResult);
+ assertNotNull(rebalanceSummaryResult.getServerInfo());
+ RebalanceSummaryResult.ServerInfo rebalanceServerInfo = rebalanceSummaryResult.getServerInfo();
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ assertEquals(rebalanceServerInfo.getNumServers().getExpectedValueAfterRebalance(), NUM_REPLICAS);
+
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, new RebalanceConfig(), null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.NO_OP);
+ // Segment assignment should not change
+ assertEquals(rebalanceResult.getSegmentAssignment(), oldSegmentAssignment);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + numServers, true);
+
+
+ // increase replica group size by 1
+ instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS + 1, 1, 0, 0, false,
+ null), null, false);
+
+ tableConfig.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig));
+
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ rebalanceServerInfo = rebalanceResult.getRebalanceSummaryResult().getServerInfo();
+
+ // with minimize data movement, we should add 1 server only
+ assertEquals(rebalanceServerInfo.getServersAdded().size(), 1);
+ assertEquals(rebalanceServerInfo.getServersRemoved().size(), 0);
+
+
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getNumReplicaGroups(), NUM_REPLICAS + 1);
+
+ // add one server instance
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + (numServers + 1), true);
+
+ // decrease replica group size by 1
+ instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(null), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, NUM_REPLICAS, 1, 0, 0, false,
+ null), null, false);
+
+ tableConfig.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig));
+ _helixResourceManager.updateTableConfig(tableConfig);
+
+ // Try dry-run summary mode
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setDryRun(true);
+ rebalanceConfig.setSummary(true);
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ rebalanceServerInfo = rebalanceResult.getRebalanceSummaryResult().getServerInfo();
+
+ // with minimize data movement, we should remove 1 server only
+ assertEquals(rebalanceServerInfo.getServersAdded().size(), 0);
+ assertEquals(rebalanceServerInfo.getServersRemoved().size(), 1);
+
+
+ rebalanceConfig = new RebalanceConfig();
+ rebalanceConfig.setReassignInstances(true);
+ rebalanceConfig.setMinimizeDataMovement(true);
+ rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null);
+ assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE);
+ assertEquals(rebalanceResult.getInstanceAssignment().get(InstancePartitionsType.OFFLINE).getNumReplicaGroups(), NUM_REPLICAS);
+
+ _helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
Review Comment:
nit: is this deleting the right offline table? `TIERED_TABLE_NAME`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972437026
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -886,6 +886,188 @@ public void testRebalanceWithTiersAndInstanceAssignments()
_helixResourceManager.deleteOfflineTable(TIERED_TABLE_NAME);
}
+
+ @Test
+ public void testRebalanceWithMinimizeDataMovementInstanceAssignments()
+ throws Exception {
+ // TODO: try balanced (default) instance assignment
+ int numServers = 6;
+ for (int i = 0; i < numServers; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, true);
+ }
+
+ // Create the table with default balanced segment assignment
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS).build();
+
+ addDummySchema(RAW_TABLE_NAME);
+ _helixResourceManager.addTable(tableConfig);
+
+ // Add the segments
+ int numSegments = 10;
+ long nowInDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
+
+ for (int i = 0; i < numSegments; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TIERED_TABLE_NAME,
Review Comment:
the table config created above uses `RAW_TABLE_NAME`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972069279
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -1303,7 +1303,7 @@ private void handleLegacySchemaConfig(TableConfig tableConfig, HttpHeaders httpH
private void validateInstanceAssignment(TableConfig tableConfig) {
TableRebalancer tableRebalancer = new TableRebalancer(_pinotHelixResourceManager.getHelixZkManager());
try {
- tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true);
+ tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true, false);
Review Comment:
I did not write one for `TableRebalancer.getInstancePartitionsMap`. Since this is the only place that uses this method outside of `TableRebalancer`, should we add another signature to default the argument just for this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1984396192
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,6 +615,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
+ @ApiParam(value = "If `minimizeDataMovement` is set to false in the table config, set this to true to enforce"
+ + " minimizeDataMovement during the rebalance.") @DefaultValue("true")
+ @QueryParam("forceMinimizeDataMovement") boolean forceMinimizeDataMovement,
Review Comment:
can you name it `minimizeDataMovement`? this is the user-facing flag
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972270210
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
+ // This test case is using the same instance rebalance plot as testMinimizeDataMovement, and test whether
+ // forceMinimizeDataMovement flag in InstanceAssignmentDriver works as the minimizeDataMovement flag in
+ // TableConfig does.
+ int numReplicas = 3;
+ int numPartitions = 2;
+ int numInstancesPerPartition = 2;
+ String partitionColumn = "partition";
+
+ // Configs and driver that minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, true);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(numReplicas)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig))
+ .build();
+
+ InstanceAssignmentDriver driver = new InstanceAssignmentDriver(tableConfig);
+
+ // Configs and driver that DO NOT minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfigNotMinimized = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, false);
+
+ TableConfig tableConfigNotMinimized = new TableConfig(tableConfig);
+ tableConfigNotMinimized.setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfigNotMinimized));
+ InstanceAssignmentDriver driverNotMinimized = new InstanceAssignmentDriver(tableConfigNotMinimized);
Review Comment:
https://github.com/apache/pinot/commit/a2992680f7d47553b08e8dda9ff6da92076a1747
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
+ // This test case is using the same instance rebalance plot as testMinimizeDataMovement, and test whether
+ // forceMinimizeDataMovement flag in InstanceAssignmentDriver works as the minimizeDataMovement flag in
+ // TableConfig does.
+ int numReplicas = 3;
+ int numPartitions = 2;
+ int numInstancesPerPartition = 2;
+ String partitionColumn = "partition";
+
+ // Configs and driver that minimize data movement
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig(
+ new InstanceTagPoolConfig(TagNameUtils.getOfflineTagForTenant(TENANT_NAME), false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, numReplicas, 0, numPartitions, numInstancesPerPartition, false,
+ partitionColumn), null, true);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(numReplicas)
+ .setInstanceAssignmentConfigMap(Map.of("OFFLINE", instanceAssignmentConfig))
+ .build();
Review Comment:
https://github.com/apache/pinot/commit/a2992680f7d47553b08e8dda9ff6da92076a1747
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1970687275
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -604,7 +814,8 @@ public void testMirrorServerSetBasedRandomInner(int loopCount) throws FileNotFou
}
System.out.println("");
InstancePartitions instancePartitions =
- driver.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, existing, preConfigured);
+ driver.assignInstances(InstancePartitionsType.OFFLINE, instanceConfigs, existing, preConfigured,
Review Comment:
[edba795](https://github.com/apache/pinot/pull/15110/commits/edba795c6d4d7caf859fcc80dcab729ed150deb0)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972101287
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -1303,7 +1303,7 @@ private void handleLegacySchemaConfig(TableConfig tableConfig, HttpHeaders httpH
private void validateInstanceAssignment(TableConfig tableConfig) {
TableRebalancer tableRebalancer = new TableRebalancer(_pinotHelixResourceManager.getHelixZkManager());
try {
- tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true);
+ tableRebalancer.getInstancePartitionsMap(tableConfig, true, true, true, false);
Review Comment:
yeah add one, it's okay even if this is the only place since in the future it might be used elsewhere.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1972316987
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -504,6 +505,208 @@ public void testMirrorServerSetBasedRandom() throws FileNotFoundException {
testMirrorServerSetBasedRandomInner(10000000);
}
+ @Test
+ public void testForceMinimizeDataMovement() {
Review Comment:
I think a balanced strategy table is not relevant here as this `InstanceAssignmentTest` is under the context of an `InstanceAssignmentDriver`.
A balanced strategy table should never touch this part of code (`InstanceAssignmentDriver`) and [the test here](https://github.com/apache/pinot/pull/15110/files#diff-ae3bd7ea7215864e0439399375581799850e67a32c2d320d42f1d96f12ac6b71R890) should be sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1979922019
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -55,40 +55,64 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ return assignInstances(instancePartitionsType, instanceConfigs, existingInstancePartitions, false);
+ }
+
+ public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
Review Comment:
Is it safe to assume that the scope of the value override could be constrained within this single rebalance operation? One concern I had was that I was not sure whether there will be other operations trying to make use of this `minimizeDataMovement` value from the table config. Current implementation is to avoid that potential misreading.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968443719
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
I think in `InstanceAssignmentConfigUtils::getInstanceAssignmentConfig` it handles the case when instance config is not set (this covers cases when `allowInstanceAssignment` is false) by falling back to look at the `minimizeDataMovement` flag in segment validation config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968447325
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentDriver.java:
##########
@@ -54,41 +54,44 @@ public InstanceAssignmentDriver(TableConfig tableConfig) {
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
- List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions) {
+ List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
+ boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(InstancePartitionsType instancePartitionsType,
List<InstanceConfig> instanceConfigs, @Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
InstanceAssignmentConfig assignmentConfig =
InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(_tableConfig, instancePartitionsType);
return getInstancePartitions(
instancePartitionsType.getInstancePartitionsName(TableNameBuilder.extractRawTableName(tableNameWithType)),
- assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions);
+ assignmentConfig, instanceConfigs, existingInstancePartitions, preConfiguredInstancePartitions,
+ forceMinimizeDataMovement);
}
public InstancePartitions assignInstances(String tierName, List<InstanceConfig> instanceConfigs,
- @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig) {
+ @Nullable InstancePartitions existingInstancePartitions, InstanceAssignmentConfig instanceAssignmentConfig,
+ boolean forceMinimizeDataMovement) {
return getInstancePartitions(
InstancePartitionsUtils.getInstancePartitionsNameForTier(_tableConfig.getTableName(), tierName),
- instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null);
+ instanceAssignmentConfig, instanceConfigs, existingInstancePartitions, null, forceMinimizeDataMovement);
}
private InstancePartitions getInstancePartitions(String instancePartitionsName,
InstanceAssignmentConfig instanceAssignmentConfig, List<InstanceConfig> instanceConfigs,
@Nullable InstancePartitions existingInstancePartitions,
- @Nullable InstancePartitions preConfiguredInstancePartitions) {
+ @Nullable InstancePartitions preConfiguredInstancePartitions, boolean forceMinimizeDataMovement) {
String tableNameWithType = _tableConfig.getTableName();
LOGGER.info("Starting {} instance assignment for table {}", instancePartitionsName, tableNameWithType);
- boolean minimizeDataMovement = instanceAssignmentConfig.isMinimizeDataMovement();
+ boolean minimizeDataMovement = forceMinimizeDataMovement || instanceAssignmentConfig.isMinimizeDataMovement();
Review Comment:
it has a Precondition check for `allowInstanceAssignment` != `false`, indicating that it just throws an exception in that case
```
Preconditions.checkState(allowInstanceAssignment(tableConfig, instancePartitionsType),
"Instance assignment is not allowed for the given table config");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1968433216
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -83,6 +83,12 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _bestEfforts = false;
+ // Whether to enforce Minimal Data Movement Algorithm. If set to false, the minimizeDataMovement flag in the table
+ // config will be used to determine whether to run the Minimal Data Movement Algorithm.
+ @JsonProperty("minimizeDataMovement")
+ @ApiModelProperty(example = "true")
+ private boolean _minimizeDataMovement = true;
Review Comment:
agree. [1bb364c](https://github.com/apache/pinot/pull/15110/commits/1bb364cf30e5e42671f459dbb781848cf8ff54bb)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985893639
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -612,9 +612,9 @@ public RebalanceResult rebalance(
@QueryParam("reassignInstances") boolean reassignInstances,
@ApiParam(value = "Whether to reassign CONSUMING segments for real-time table") @DefaultValue("false")
@QueryParam("includeConsuming") boolean includeConsuming,
- @ApiParam(value = "TRUE|FALSE|DEFAULT, whether to enable minimize data movement on rebalance, DEFAULT will use "
- + "the minimizeDataMovement in table config") @DefaultValue("TRUE")
- @QueryParam("minimizeDataMovement") String minimizeDataMovement,
+ @ApiParam(value = "Whether to enable minimize data movement on rebalance, DEFAULT will use "
+ + "the minimizeDataMovement in table config") @DefaultValue("ENABLE")
+ @QueryParam("minimizeDataMovement") RebalanceConfig.MinimizeDataMovementOptions minimizeDataMovement,
Review Comment:
quick question, what happens if we pass in a random string like "none"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add minimizeDataMovement to Rebalacne API [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #15110:
URL: https://github.com/apache/pinot/pull/15110#discussion_r1985759022
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java:
##########
@@ -51,9 +51,7 @@
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.fail;
+import static org.testng.Assert.*;
Review Comment:
Okay I think my IDE did this though
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org