[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8989: [colocated-join] Adds Support for instancePartitionsMap in Table Config

[GitBox <gi...@apache.org>]

Jackie-Jiang commented on code in PR #8989:
URL: https://github.com/apache/pinot/pull/8989#discussion_r939243821


##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java:
##########
@@ -53,6 +53,7 @@ public class TableConfig extends BaseJsonConfig {
   public static final String INGESTION_CONFIG_KEY = "ingestionConfig";
   public static final String TIER_CONFIGS_LIST_KEY = "tierConfigs";
   public static final String TUNER_CONFIG_LIST_KEY = "tunerConfigs";
+  public static final String INSTANCE_PARTITIONS_MAP_CONFIG_KEY = "instancePartitionsMap";

Review Comment:
   (minor) Put it next to `INSTANCE_ASSIGNMENT_CONFIG_MAP_KEY` since they are both for the instance assignment. Same for the order of variable declaration, argument, getter, setter



##########
pinot-common/src/main/java/org/apache/pinot/common/utils/config/TableConfigUtils.java:
##########
@@ -294,4 +305,20 @@ public static void convertFromLegacyTableConfig(TableConfig tableConfig) {
     validationConfig.setSegmentPushFrequency(null);
     validationConfig.setSegmentPushType(null);
   }
+
+  /**
+   * Returns true if the table has pre-configured instance partitions for any type (OFFLINE/CONSUMING/COMPLETED).
+   */
+  public static boolean hasPreConfiguredInstancePartitions(TableConfig tableConfig) {
+    return tableConfig.getInstancePartitionsMap() != null && tableConfig.getInstancePartitionsMap().size() > 0;

Review Comment:
   (nit) 
   ```suggestion
       return MapUtils.isNotEmpty(tableConfig.getInstancePartitionsMap());
   ```



##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceAssignmentRestletResource.java:
##########
@@ -210,6 +211,13 @@ private void assignInstancesForInstancePartitionsType(
       Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap, TableConfig tableConfig,
       List<InstanceConfig> instanceConfigs, InstancePartitionsType instancePartitionsType) {
     String tableNameWithType = tableConfig.getTableName();
+    String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);

Review Comment:
   (nit) Move this into the following if block (can also be inlined)



##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -416,8 +418,22 @@ private Map<InstancePartitionsType, InstancePartitions> getInstancePartitionsMap
   private InstancePartitions getInstancePartitions(TableConfig tableConfig,
       InstancePartitionsType instancePartitionsType, boolean reassignInstances, boolean dryRun) {
     String tableNameWithType = tableConfig.getTableName();
+    String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
+    boolean hasPreConfiguredInstancePartitions = TableConfigUtils.hasPreConfiguredInstancePartitions(tableConfig,
+        instancePartitionsType);

Review Comment:
   (nit) Move them into the following if block (can also be inlined)



##########
pinot-common/src/main/java/org/apache/pinot/common/assignment/InstancePartitions.java:
##########
@@ -127,6 +127,13 @@ public ZNRecord toZNRecord() {
     return znRecord;
   }
 
+  /**
+   * Returns a new instance of InstancePartitions with the given name
+   */
+  public InstancePartitions withName(String newName) {

Review Comment:
   (minor) I think we can just add `setInstancePartitionsName(String instancePartitionsName)` so that we can reuse the same object



##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java:
##########
@@ -115,6 +115,7 @@ public class TableConfigBuilder {
   private IngestionConfig _ingestionConfig;
   private List<TierConfig> _tierConfigList;
   private List<TunerConfig> _tunerConfigList;
+  private Map<InstancePartitionsType, String> _instancePartitionsMap;

Review Comment:
   (minor) Put it next to `_instanceAssignmentConfigMap`



##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1693,10 +1693,20 @@ private void assignInstances(TableConfig tableConfig, boolean override) {
       InstanceAssignmentDriver instanceAssignmentDriver = new InstanceAssignmentDriver(tableConfig);
       List<InstanceConfig> instanceConfigs = getAllHelixInstanceConfigs();
       for (InstancePartitionsType instancePartitionsType : instancePartitionsTypesToAssign) {
-        InstancePartitions instancePartitions =
-            instanceAssignmentDriver.assignInstances(instancePartitionsType, instanceConfigs, null);
-        LOGGER.info("Persisting instance partitions: {}", instancePartitions);
-        InstancePartitionsUtils.persistInstancePartitions(_propertyStore, instancePartitions);
+        boolean hasPreConfiguredInstancePartitions = TableConfigUtils.hasPreConfiguredInstancePartitions(tableConfig,
+            instancePartitionsType);
+        InstancePartitions instancePartitions;
+        if (!hasPreConfiguredInstancePartitions) {
+          instancePartitions = instanceAssignmentDriver.assignInstances(instancePartitionsType, instanceConfigs, null);
+          LOGGER.info("Persisting instance partitions: {}", instancePartitions);
+          InstancePartitionsUtils.persistInstancePartitions(_propertyStore, instancePartitions);
+        } else {
+          instancePartitions = InstancePartitionsUtils.fetchInstancePartitionsWithRename(_propertyStore,
+              tableConfig.getInstancePartitionsMap().get(instancePartitionsType),
+              instancePartitionsType.getInstancePartitionsName(rawTableName));
+          LOGGER.info("Persisting instance partitions: {}", instancePartitions);

Review Comment:
   (minor) Suggest mentioning the instance partitions is referencing another instance partitions in the log, something like `Persisting instance partitions: {} (referencing {})`



##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -559,6 +561,26 @@ static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema)
     validateAggregateMetricsForUpsertConfig(tableConfig);
   }
 
+  /**
+   * Detects whether both InstanceAssignmentConfig and InstancePartitionsMap are set for a given
+   * instance partitions type. Validation fails because the table would ignore InstanceAssignmentConfig
+   * when the partitions are already set.
+   */
+  @VisibleForTesting
+  static void validateInstancePartitionsTypeMapConfig(TableConfig tableConfig) {
+    if (!org.apache.pinot.common.utils.config.TableConfigUtils.hasPreConfiguredInstancePartitions(tableConfig)) {
+      return;
+    }
+    for (InstancePartitionsType instancePartitionsType : tableConfig.getInstancePartitionsMap().keySet()) {
+      if (tableConfig.getInstanceAssignmentConfigMap() == null) {
+        break;
+      }
+      Preconditions.checkState(!tableConfig.getInstanceAssignmentConfigMap().containsKey(instancePartitionsType),
+          String.format("Both InstanceAssignmentConfigMap and InstancePartitionsMap set for %s",
+              instancePartitionsType));
+    }

Review Comment:
   (nit) might be more readable, also avoid accessing the other `TableConfigUtils`
   ```suggestion
       if (MapUtils.isEmpty(tableConfig.getInstancePartitionsMap()) || MapUtils.isEmpty(tableConfig.getInstanceAssignmentConfigMap())) {
         return;
       }
       for (InstancePartitionsType instancePartitionsType : tableConfig.getInstancePartitionsMap().keySet()) {
         Preconditions.checkState(!tableConfig.getInstanceAssignmentConfigMap().containsKey(instancePartitionsType),
             String.format("Both InstanceAssignmentConfigMap and InstancePartitionsMap set for %s",
                 instancePartitionsType));
       }
   ```



##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -416,8 +418,22 @@ private Map<InstancePartitionsType, InstancePartitions> getInstancePartitionsMap
   private InstancePartitions getInstancePartitions(TableConfig tableConfig,
       InstancePartitionsType instancePartitionsType, boolean reassignInstances, boolean dryRun) {
     String tableNameWithType = tableConfig.getTableName();
+    String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
+    boolean hasPreConfiguredInstancePartitions = TableConfigUtils.hasPreConfiguredInstancePartitions(tableConfig,
+        instancePartitionsType);
     if (InstanceAssignmentConfigUtils.allowInstanceAssignment(tableConfig, instancePartitionsType)) {
       if (reassignInstances) {
+        if (hasPreConfiguredInstancePartitions) {
+          InstancePartitions instancePartitions = InstancePartitionsUtils.fetchInstancePartitionsWithRename(
+              _helixManager.getHelixPropertyStore(), tableConfig.getInstancePartitionsMap().get(instancePartitionsType),
+              instancePartitionsType.getInstancePartitionsName(rawTableName));
+          if (!dryRun) {
+            LOGGER.info("Persisting instance partitions: {} to ZK", instancePartitions);

Review Comment:
   (minor) Mention it is referencing an existing instance partitions



-- 
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