This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 91fbab84f5d Move the segment assignment partition column to
SegmentAssignmentConfig (#19958)
91fbab84f5d is described below
commit 91fbab84f5d5100befd5171389dccf81c36db2bf
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Thu Oct 22 11:25:54 2026 -0700
Move the segment assignment partition column to SegmentAssignmentConfig
(#19958)
---
.../assignment/InstanceAssignmentConfigUtils.java | 4 +-
.../utils/config/TableConfigSerDeUtilsTest.java | 46 ++--
.../api/resources/TableConfigValidationUtils.java | 19 +-
.../api/upload/SegmentValidationUtils.java | 56 ++---
.../strategy/SegmentAssignmentStrategyFactory.java | 74 +++---
.../api/upload/SegmentValidationUtilsTest.java | 8 +-
.../controller/helix/PinotResourceManagerTest.java | 2 -
.../instance/InstanceAssignmentTest.java | 3 +-
.../RealtimeReplicaGroupSegmentAssignmentTest.java | 11 +-
.../StrictRealtimeSegmentAssignmentTest.java | 9 +-
.../ReplicaGroupSegmentAssignmentStrategyTest.java | 18 +-
...nReplicaGroupSegmentAssignmentStrategyTest.java | 14 +-
.../RoundRobinSegmentAssignmentStrategyTest.java | 3 +-
.../SegmentAssignmentStrategyFactoryTest.java | 9 +-
.../tests/BaseClusterIntegrationTest.java | 12 +-
.../tests/BaseDedupIntegrationTest.java | 2 -
.../tests/DedupPreloadIntegrationTest.java | 2 -
.../tests/custom/OfflineUpsertTableTest.java | 17 +-
.../tests/custom/VectorUpsertTableTest.java | 2 -
.../simpleMeetup_realtime_table_config.json | 6 +-
.../segment/local/utils/TableConfigUtils.java | 198 +++++++++++++---
.../segment/local/utils/TableConfigUtilsTest.java | 263 ++++++++++++++++++++-
.../config/table/ReplicaGroupStrategyConfig.java | 4 +-
.../SegmentsValidationAndRetentionConfig.java | 9 +-
.../table/assignment/InstanceAssignmentConfig.java | 35 +--
.../InstanceReplicaGroupPartitionConfig.java | 9 +-
.../table/assignment/SegmentAssignmentConfig.java | 22 +-
.../apache/pinot/spi/utils/CommonConstants.java | 13 +
.../spi/utils/builder/TableConfigBuilder.java | 3 +-
.../userAttributes_offline_table_config.json | 3 +-
.../userFactEvents_offline_table_config.json | 6 +-
.../userGroups_offline_table_config.json | 6 +-
...upsertJsonMeetupRsvp_realtime_table_config.json | 1 -
.../upsertMeetupRsvp_realtime_table_config.json | 1 -
...ertPartialMeetupRsvp_realtime_table_config.json | 1 -
35 files changed, 625 insertions(+), 266 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/assignment/InstanceAssignmentConfigUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/assignment/InstanceAssignmentConfigUtils.java
index d553cd377cd..0665f930f26 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/assignment/InstanceAssignmentConfigUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/assignment/InstanceAssignmentConfigUtils.java
@@ -115,7 +115,9 @@ public class InstanceAssignmentConfigUtils {
SegmentsValidationAndRetentionConfig segmentConfig =
tableConfig.getValidationConfig();
int numReplicaGroups = tableConfig.getReplication();
ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
segmentConfig.getReplicaGroupStrategyConfig();
- Preconditions.checkState(replicaGroupStrategyConfig != null, "Failed to
find the replica-group strategy config");
+ Preconditions.checkState(replicaGroupStrategyConfig != null,
+ "Replica-group segment assignment requires the OFFLINE instance
assignment to be configured in "
+ + "'instanceAssignmentConfigMap'");
String partitionColumn = replicaGroupStrategyConfig.getPartitionColumn();
boolean minimizeDataMovement = segmentConfig.isMinimizeDataMovement();
if (partitionColumn != null) {
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
index 4ddec5136ce..4aaa37e4363 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
@@ -32,9 +32,7 @@ import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.HashFunction;
import org.apache.pinot.spi.config.table.QueryConfig;
import org.apache.pinot.spi.config.table.QuotaConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
-import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TagOverrideConfig;
@@ -57,6 +55,7 @@ import
org.apache.pinot.spi.config.table.ingestion.ParallelSegmentConsumptionPol
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.CommonConstants.Segment.AssignmentStrategy;
import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
@@ -145,24 +144,26 @@ public class TableConfigSerDeUtilsTest {
checkTenantConfigWithTagOverride(tableConfigToCompare);
}
{
- // With SegmentAssignmentStrategyConfig
- ReplicaGroupStrategyConfig replicaGroupStrategyConfig = new
ReplicaGroupStrategyConfig("memberId", 5);
- TableConfig tableConfig =
tableConfigBuilder.setSegmentAssignmentConfigMap(
- Map.of(InstancePartitionsType.OFFLINE.toString(),
- new
SegmentAssignmentConfig("ReplicaGroupSegmentAssignmentStrategy")))
- .setReplicaGroupStrategyConfig(replicaGroupStrategyConfig)
+ // With partitioned instance assignment config and segment assignment
config
+ InstanceAssignmentConfig instanceAssignmentConfig =
+ new InstanceAssignmentConfig(new
InstanceTagPoolConfig("tenant_OFFLINE", false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, 2, 0, 4, 1,
false, null), null, false);
+ TableConfig tableConfig = tableConfigBuilder
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
instanceAssignmentConfig))
+
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
+ new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
"memberId")))
.build();
- checkSegmentAssignmentStrategyConfig(tableConfig);
+ checkSegmentAssignmentConfig(tableConfig);
// Serialize then de-serialize
TableConfig tableConfigToCompare =
JsonUtils.stringToObject(tableConfig.toJsonString(), TableConfig.class);
assertEquals(tableConfigToCompare, tableConfig);
- checkSegmentAssignmentStrategyConfig(tableConfigToCompare);
+ checkSegmentAssignmentConfig(tableConfigToCompare);
tableConfigToCompare =
TableConfigSerDeUtils.fromZNRecord(TableConfigSerDeUtils.toZNRecord(tableConfig));
assertEquals(tableConfigToCompare, tableConfig);
- checkSegmentAssignmentStrategyConfig(tableConfigToCompare);
+ checkSegmentAssignmentConfig(tableConfigToCompare);
}
{
// With completion config
@@ -441,18 +442,27 @@ public class TableConfigSerDeUtilsTest {
assertNull(tenantConfig.getTagOverrideConfig().getRealtimeCompleted());
}
- private void checkSegmentAssignmentStrategyConfig(TableConfig tableConfig) {
- SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
+ private void checkSegmentAssignmentConfig(TableConfig tableConfig) {
+ Map<String, InstanceAssignmentConfig> instanceAssignmentConfigMap =
tableConfig.getInstanceAssignmentConfigMap();
+ assertNotNull(instanceAssignmentConfigMap);
+ InstanceAssignmentConfig instanceAssignmentConfig =
+
instanceAssignmentConfigMap.get(InstancePartitionsType.OFFLINE.toString());
+ assertNotNull(instanceAssignmentConfig);
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
+ instanceAssignmentConfig.getReplicaGroupPartitionConfig();
+ assertTrue(replicaGroupPartitionConfig.isReplicaGroupBased());
+ assertEquals(replicaGroupPartitionConfig.getNumReplicaGroups(), 2);
+ assertEquals(replicaGroupPartitionConfig.getNumPartitions(), 4);
+ assertEquals(replicaGroupPartitionConfig.getNumInstancesPerPartition(), 1);
+
Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
tableConfig.getSegmentAssignmentConfigMap();
assertNotNull(segmentAssignmentConfigMap);
SegmentAssignmentConfig segmentAssignmentConfig =
segmentAssignmentConfigMap.get(InstancePartitionsType.OFFLINE.toString());
assertNotNull(segmentAssignmentConfig);
- assertEquals(segmentAssignmentConfig.getAssignmentStrategy(),
"ReplicaGroupSegmentAssignmentStrategy");
- ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
validationConfig.getReplicaGroupStrategyConfig();
- assertNotNull(replicaGroupStrategyConfig);
- assertEquals(replicaGroupStrategyConfig.getPartitionColumn(), "memberId");
- assertEquals(replicaGroupStrategyConfig.getNumInstancesPerPartition(), 5);
+ assertEquals(segmentAssignmentConfig.getAssignmentStrategy(),
+ AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY);
+ assertEquals(segmentAssignmentConfig.getPartitionColumn(), "memberId");
}
private void checkCompletionConfig(TableConfig tableConfig) {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
index 58c3e96532a..4c7f7a6fd5b 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
@@ -23,6 +23,7 @@ import java.util.List;
import javax.annotation.Nullable;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.assignment.InstanceAssignmentConfigUtils;
+import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.assignment.InstancePartitionsUtils;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
@@ -120,7 +121,9 @@ public final class TableConfigValidationUtils {
}
/// Computes the instance partitions the table would get on creation, so
that an instance assignment that cannot be
- /// satisfied by the current instances is rejected before the table config
is persisted.
+ /// satisfied by the current instances is rejected before the table config
is persisted, and validates that the
+ /// OFFLINE segments are placed by partition when the OFFLINE instance
partitions have multiple partitions. This
+ /// covers pre-configured instance partitions, whose layout is not visible
in the table config.
private static void validateInstanceAssignment(PinotHelixResourceManager
resourceManager, TableConfig tableConfig) {
List<InstancePartitionsType> instancePartitionsTypes;
if (tableConfig.getTableType() == TableType.OFFLINE) {
@@ -131,14 +134,19 @@ public final class TableConfigValidationUtils {
instancePartitionsTypes = List.of(InstancePartitionsType.CONSUMING);
}
List<InstanceConfig> instanceConfigs =
resourceManager.getAllHelixInstanceConfigs();
+ InstancePartitions offlineInstancePartitions = null;
try {
for (InstancePartitionsType instancePartitionsType :
instancePartitionsTypes) {
+ InstancePartitions instancePartitions;
if (InstanceAssignmentConfigUtils.allowInstanceAssignment(tableConfig,
instancePartitionsType)) {
- InstanceAssignmentUtils.computeInstancePartitions(tableConfig,
instancePartitionsType,
+ instancePartitions =
InstanceAssignmentUtils.computeInstancePartitions(tableConfig,
instancePartitionsType,
resourceManager.getPropertyStore(), instanceConfigs, null);
} else {
-
InstancePartitionsUtils.computeDefaultInstancePartitions(resourceManager.getHelixZkManager(),
tableConfig,
- instancePartitionsType);
+ instancePartitions =
InstancePartitionsUtils.computeDefaultInstancePartitions(
+ resourceManager.getHelixZkManager(), tableConfig,
instancePartitionsType);
+ }
+ if (instancePartitionsType == InstancePartitionsType.OFFLINE) {
+ offlineInstancePartitions = instancePartitions;
}
}
} catch (Exception e) {
@@ -146,5 +154,8 @@ public final class TableConfigValidationUtils {
"Failed to calculate instance partitions for table: " +
tableConfig.getTableName() + ", reason: "
+ e.getMessage(), e);
}
+ if (offlineInstancePartitions != null &&
offlineInstancePartitions.getNumPartitions() > 1) {
+ TableConfigUtils.validatePartitionedSegmentAssignment(tableConfig,
InstancePartitionsType.OFFLINE.name());
+ }
}
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
index 88acb5e6b3d..1127d331a40 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
@@ -20,22 +20,16 @@ package org.apache.pinot.controller.api.upload;
import java.util.Map;
import java.util.Set;
+import javax.annotation.Nullable;
import javax.ws.rs.core.Response;
-import org.apache.commons.collections4.MapUtils;
-import org.apache.commons.lang3.StringUtils;
import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.apache.pinot.controller.validation.StorageQuotaChecker;
+import org.apache.pinot.segment.local.utils.TableConfigUtils;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.SegmentMetadata;
-import org.apache.pinot.spi.config.table.IndexingConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
-import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig;
-import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
-import
org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
import org.apache.pinot.spi.utils.TimeUtils;
import org.joda.time.Interval;
import org.slf4j.Logger;
@@ -76,7 +70,7 @@ public class SegmentValidationUtils {
}
String partitionColumn = getPartitionColumn(tableConfig);
- if (StringUtils.isEmpty(partitionColumn)) {
+ if (partitionColumn == null) {
return;
}
@@ -90,46 +84,24 @@ public class SegmentValidationUtils {
}
}
+ /// Returns the partition column configured for segment assignment, falling
back to the single column of the segment
+ /// partition config, or `null` if neither is available.
+ @Nullable
private static String getPartitionColumn(TableConfig tableConfig) {
- if (!MapUtils.isEmpty(tableConfig.getInstanceAssignmentConfigMap())) {
- InstanceAssignmentConfig instanceAssignmentConfig =
-
tableConfig.getInstanceAssignmentConfigMap().get(InstancePartitionsType.OFFLINE.name());
- String partitionColumn = getPartitionColumn(instanceAssignmentConfig);
- if (StringUtils.isNotEmpty(partitionColumn)) {
- return partitionColumn;
- }
- }
-
- //noinspection deprecation
- SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
- ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
- validationConfig != null ?
validationConfig.getReplicaGroupStrategyConfig() : null;
- String partitionColumn =
- replicaGroupStrategyConfig != null ?
replicaGroupStrategyConfig.getPartitionColumn() : null;
- if (StringUtils.isNotEmpty(partitionColumn)) {
+ String partitionColumn = TableConfigUtils.getPartitionColumn(tableConfig);
+ if (partitionColumn != null) {
return partitionColumn;
}
-
- IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
- SegmentPartitionConfig segmentPartitionConfig =
- indexingConfig != null ? indexingConfig.getSegmentPartitionConfig() :
null;
- Map<String, ?> columnPartitionMap =
- segmentPartitionConfig != null ?
segmentPartitionConfig.getColumnPartitionMap() : null;
- if (MapUtils.isNotEmpty(columnPartitionMap) && columnPartitionMap.size()
== 1) {
- return columnPartitionMap.keySet().iterator().next();
+ SegmentPartitionConfig segmentPartitionConfig =
tableConfig.getIndexingConfig().getSegmentPartitionConfig();
+ if (segmentPartitionConfig != null) {
+ Map<String, ?> columnPartitionMap =
segmentPartitionConfig.getColumnPartitionMap();
+ if (columnPartitionMap.size() == 1) {
+ return columnPartitionMap.keySet().iterator().next();
+ }
}
return null;
}
- private static String getPartitionColumn(InstanceAssignmentConfig
instanceAssignmentConfig) {
- if (instanceAssignmentConfig == null) {
- return null;
- }
- InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
- instanceAssignmentConfig.getReplicaGroupPartitionConfig();
- return replicaGroupPartitionConfig != null ?
replicaGroupPartitionConfig.getPartitionColumn() : null;
- }
-
public static void checkStorageQuota(String segmentName, long
tarSegmentSizeInBytes, long untarredSegmentSizeInBytes,
TableConfig tableConfig,
StorageQuotaChecker quotaChecker) {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactory.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactory.java
index c7b901393e4..03e2da94cf8 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactory.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactory.java
@@ -19,6 +19,7 @@
package org.apache.pinot.controller.helix.core.assignment.segment.strategy;
import com.google.common.base.Preconditions;
+import java.util.Locale;
import java.util.Map;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.assignment.InstancePartitions;
@@ -30,58 +31,22 @@ import
org.apache.pinot.spi.utils.CommonConstants.Segment.AssignmentStrategy;
/// Factory for SegmentAssignmentStrategy
public class SegmentAssignmentStrategyFactory {
-
private SegmentAssignmentStrategyFactory() {
}
- /// Determine Segment Assignment strategy
+ /// Returns the segment assignment strategy for the given instance
partitions type (`OFFLINE` or `COMPLETED`) or
+ /// tier: the strategy configured under that key in
`segmentAssignmentConfigMap`, otherwise the one derived from the
+ /// instance partitions.
public static SegmentAssignmentStrategy
getSegmentAssignmentStrategy(HelixManager helixManager,
TableConfig tableConfig, String assignmentType, InstancePartitions
instancePartitions) {
- String assignmentStrategy = null;
-
- TableType currentTableType = tableConfig.getTableType();
- // TODO: Handle segment assignment strategy in future for CONSUMING
segments in follow up PR
- // See https://github.com/apache/pinot/issues/9047
- // Accommodate new changes for assignment strategy
- Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
tableConfig.getSegmentAssignmentConfigMap();
-
+ SegmentAssignmentStrategy segmentAssignmentStrategy;
if (tableConfig.isDimTable()) {
- // Segment Assignment Strategy for DIM tables
- Preconditions.checkState(currentTableType == TableType.OFFLINE,
+ // Dimension tables are always assigned to all servers
+ Preconditions.checkState(tableConfig.getTableType() == TableType.OFFLINE,
"All Servers Segment assignment Strategy is only applicable to Dim
OfflineTables");
- SegmentAssignmentStrategy segmentAssignmentStrategy = new
AllServersSegmentAssignmentStrategy();
- segmentAssignmentStrategy.init(helixManager, tableConfig);
- return segmentAssignmentStrategy;
+ segmentAssignmentStrategy = new AllServersSegmentAssignmentStrategy();
} else {
- // Try to determine segment assignment strategy from table config
- if (segmentAssignmentConfigMap != null) {
- SegmentAssignmentConfig segmentAssignmentConfig;
- // Use the pre-defined segment assignment strategy
- segmentAssignmentConfig =
segmentAssignmentConfigMap.get(assignmentType.toUpperCase());
- // Segment assignment config is only applicable to offline tables and
completed segments of realtime tables
- if (segmentAssignmentConfig != null) {
- assignmentStrategy =
segmentAssignmentConfig.getAssignmentStrategy().toLowerCase();
- }
- }
- }
-
- // Use the existing information to determine segment assignment strategy
- SegmentAssignmentStrategy segmentAssignmentStrategy;
- if (assignmentStrategy == null) {
- // Calculate numReplicaGroups and numPartitions to determine segment
assignment strategy
- Preconditions
- .checkState(instancePartitions != null, "Failed to find instance
partitions for segment assignment strategy");
- int numReplicaGroups = instancePartitions.getNumReplicaGroups();
- int numPartitions = instancePartitions.getNumPartitions();
-
- if (numReplicaGroups == 1 && numPartitions == 1) {
- segmentAssignmentStrategy = new BalancedNumSegmentAssignmentStrategy();
- } else {
- segmentAssignmentStrategy = new
ReplicaGroupSegmentAssignmentStrategy();
- }
- } else {
- // Set segment assignment strategy depending on strategy set in table
config
- switch (assignmentStrategy) {
+ switch (getAssignmentStrategy(tableConfig, assignmentType,
instancePartitions)) {
case AssignmentStrategy.ROUND_ROBIN_SEGMENT_ASSIGNMENT_STRATEGY:
segmentAssignmentStrategy = new
RoundRobinSegmentAssignmentStrategy();
break;
@@ -93,6 +58,7 @@ public class SegmentAssignmentStrategyFactory {
break;
case AssignmentStrategy.BALANCE_NUM_SEGMENT_ASSIGNMENT_STRATEGY:
default:
+ // Existing tables may carry a strategy name that was never
recognized, which has always run as balanced
segmentAssignmentStrategy = new
BalancedNumSegmentAssignmentStrategy();
break;
}
@@ -100,4 +66,24 @@ public class SegmentAssignmentStrategyFactory {
segmentAssignmentStrategy.init(helixManager, tableConfig);
return segmentAssignmentStrategy;
}
+
+ /// Returns the strategy name configured under the instance partitions type
or tier name, if any, otherwise the one
+ /// derived from the instance partitions: balanced unless they have multiple
replica-groups or partitions.
+ private static String getAssignmentStrategy(TableConfig tableConfig, String
assignmentType,
+ InstancePartitions instancePartitions) {
+ // TODO: Handle the segment assignment strategy for CONSUMING segments, see
+ // https://github.com/apache/pinot/issues/9047
+ Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
tableConfig.getSegmentAssignmentConfigMap();
+ if (segmentAssignmentConfigMap != null) {
+ SegmentAssignmentConfig segmentAssignmentConfig =
segmentAssignmentConfigMap.get(assignmentType);
+ if (segmentAssignmentConfig != null &&
segmentAssignmentConfig.getAssignmentStrategy() != null) {
+ return
segmentAssignmentConfig.getAssignmentStrategy().toLowerCase(Locale.ROOT);
+ }
+ }
+ Preconditions.checkState(instancePartitions != null,
+ "Failed to find instance partitions for segment assignment strategy");
+ return instancePartitions.getNumReplicaGroups() == 1 &&
instancePartitions.getNumPartitions() == 1
+ ? AssignmentStrategy.BALANCE_NUM_SEGMENT_ASSIGNMENT_STRATEGY
+ : AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY;
+ }
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
index c0e320cf62a..d58cb9db913 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
@@ -25,7 +25,6 @@ import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.SegmentMetadata;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -34,6 +33,7 @@ import
org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import
org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
import org.apache.pinot.spi.config.table.assignment.InstanceTagPoolConfig;
+import org.apache.pinot.spi.config.table.assignment.SegmentAssignmentConfig;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
@@ -127,10 +127,11 @@ public class SegmentValidationUtilsTest {
TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
.setTableName(RAW_TABLE_NAME)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(PARTITION_COLUMN, 1))
.setSegmentPartitionConfig(new
SegmentPartitionConfig(Map.of(PARTITION_COLUMN,
new ColumnPartitionConfig("Murmur", NUM_PARTITIONS),
"otherPartitionColumn",
new ColumnPartitionConfig("Modulo", NUM_PARTITIONS))))
+
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.name(),
+ new SegmentAssignmentConfig(null, PARTITION_COLUMN)))
.build();
SegmentValidationUtils.validateUpsertSegmentPartitionMetadata(mockSegmentMetadata(Set.of(1)),
tableConfig);
@@ -140,7 +141,8 @@ public class SegmentValidationUtilsTest {
return new TableConfigBuilder(TableType.OFFLINE)
.setTableName(RAW_TABLE_NAME)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(PARTITION_COLUMN, 1))
+
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.name(),
+ new SegmentAssignmentConfig(null, PARTITION_COLUMN)))
.build();
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java
index eea36b23659..d16331690b2 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java
@@ -31,7 +31,6 @@ import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.UpsertConfig;
@@ -75,7 +74,6 @@ public class PinotResourceManagerTest {
TableConfig realtimeTableConfig =
new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
.setStreamConfigs(streamConfigs)
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(PARTITION_COLUMN, 1))
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL)).build();
_resourceManager.addTable(realtimeTableConfig);
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java
index 23661254021..941f8c6a7cf 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceAssignmentTest.java
@@ -81,7 +81,7 @@ public class InstanceAssignmentTest {
.setServerTenant(TENANT_NAME)
.setNumReplicas(numReplicas)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY)))
+ new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
null)))
.build();
int numInstancesPerPartition = 2;
tableConfig.getValidationConfig()
@@ -2769,7 +2769,6 @@ public class InstanceAssignmentTest {
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
new InstanceAssignmentConfig(tagPoolConfig, null,
replicaPartitionConfig,
InstanceAssignmentConfig.PartitionSelector.FD_AWARE_INSTANCE_PARTITION_SELECTOR.toString(),
false)))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(partitionColumnName, numInstancesPerReplicaGroup))
.setSegmentPartitionConfig(segmentPartitionConfig)
.build();
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java
index 3952bcc2273..68dd2c110e5 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java
@@ -32,13 +32,10 @@ import
org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.restlet.resources.RebalanceConfig;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
-import org.apache.pinot.spi.config.table.assignment.SegmentAssignmentConfig;
import
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
-import org.apache.pinot.spi.utils.CommonConstants.Segment.AssignmentStrategy;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -55,7 +52,6 @@ import static org.testng.Assert.assertTrue;
@SuppressWarnings("unchecked")
public class RealtimeReplicaGroupSegmentAssignmentTest {
private static final int NUM_REPLICAS = 3;
- private static final String PARTITION_COLUMN = "partitionColumn";
private static final int NUM_PARTITIONS = 4;
private static final int NUM_SEGMENTS = 24;
private static final String CONSUMING_INSTANCE_NAME_PREFIX =
"consumingInstance_";
@@ -87,10 +83,7 @@ public class RealtimeReplicaGroupSegmentAssignmentTest {
Map<String, String> streamConfigs =
FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
TableConfig tableConfig =
new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
- .setStreamConfigs(streamConfigs)
-
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.COMPLETED.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY)))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(PARTITION_COLUMN, 1)).build();
+ .setStreamConfigs(streamConfigs).build();
_segmentAssignment =
SegmentAssignmentFactory.getSegmentAssignment(createHelixManager(),
tableConfig, null);
_instancePartitionsMap = new TreeMap<>();
@@ -654,8 +647,6 @@ public class RealtimeReplicaGroupSegmentAssignmentTest {
new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(numReplicas)
.setStreamConfigs(streamConfigs)
.setInstancePartitionsMap(importedInstancePartitions)
-
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.COMPLETED.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY)))
.build();
assertTrue(InstanceAssignmentConfigUtils.shouldRelocateCompletedSegments(tableConfig),
"Imported COMPLETED instance partitions must trigger completed-segment
relocation during rebalance");
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/StrictRealtimeSegmentAssignmentTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/StrictRealtimeSegmentAssignmentTest.java
index 833be993b47..85aa85afb75 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/StrictRealtimeSegmentAssignmentTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/StrictRealtimeSegmentAssignmentTest.java
@@ -40,15 +40,12 @@ import org.apache.pinot.common.tier.TierSegmentSelector;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import org.apache.pinot.spi.config.table.DedupConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
-import org.apache.pinot.spi.config.table.assignment.SegmentAssignmentConfig;
import
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
import org.apache.pinot.spi.utils.CommonConstants.Segment;
-import org.apache.pinot.spi.utils.CommonConstants.Segment.AssignmentStrategy;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
@@ -66,7 +63,6 @@ import static org.testng.Assert.assertTrue;
@SuppressWarnings("unchecked")
public class StrictRealtimeSegmentAssignmentTest {
private static final int NUM_REPLICAS = 3;
- private static final String PARTITION_COLUMN = "partitionColumn";
private static final int NUM_PARTITIONS = 4;
private static final int NUM_SEGMENTS = 24;
private static final String CONSUMING_INSTANCE_NAME_PREFIX =
"consumingInstance_";
@@ -140,10 +136,7 @@ public class StrictRealtimeSegmentAssignmentTest {
private static SegmentAssignment createSegmentAssignment(String tableType,
HelixManager helixManager) {
TableConfigBuilder builder = new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME)
.setNumReplicas(NUM_REPLICAS)
-
.setStreamConfigs(FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap())
-
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.COMPLETED.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY)))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(PARTITION_COLUMN, 1));
+
.setStreamConfigs(FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap());
TableConfig tableConfig;
if ("upsert".equalsIgnoreCase(tableType)) {
tableConfig = builder.setUpsertConfig(new
UpsertConfig(UpsertConfig.Mode.FULL)).build();
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/ReplicaGroupSegmentAssignmentStrategyTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/ReplicaGroupSegmentAssignmentStrategyTest.java
index a8996ed11b8..aacdfe5df65 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/ReplicaGroupSegmentAssignmentStrategyTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/ReplicaGroupSegmentAssignmentStrategyTest.java
@@ -38,7 +38,6 @@ import
org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignme
import
org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignmentTestUtils;
import
org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignmentUtils;
import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
@@ -92,7 +91,8 @@ public class ReplicaGroupSegmentAssignmentStrategyTest {
new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITHOUT_PARTITION)
.setNumReplicas(NUM_REPLICAS)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY))).build();
+ new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
null)))
+ .build();
_segmentAssignmentWithoutPartition =
SegmentAssignmentFactory.getSegmentAssignment(null,
tableConfigWithoutPartitions, null);
@@ -137,14 +137,13 @@ public class ReplicaGroupSegmentAssignmentStrategyTest {
when(helixManagerWithPartitions.getHelixPropertyStore()).thenReturn(propertyStoreWithPartitions);
int numInstancesPerPartition = numInstancesPerReplicaGroup / NUM_REPLICAS;
- ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
- new ReplicaGroupStrategyConfig(PARTITION_COLUMN,
numInstancesPerPartition);
TableConfig tableConfigWithPartitions =
new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITH_PARTITION)
.setNumReplicas(NUM_REPLICAS)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY)))
- .setReplicaGroupStrategyConfig(replicaGroupStrategyConfig).build();
+ new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ PARTITION_COLUMN)))
+ .build();
_segmentAssignmentWithPartition =
SegmentAssignmentFactory.getSegmentAssignment(helixManagerWithPartitions,
tableConfigWithPartitions, null);
@@ -397,13 +396,12 @@ public class ReplicaGroupSegmentAssignmentStrategyTest {
when(helixManager.getHelixPropertyStore()).thenReturn(propertyStore);
int numInstancesPerPartition = NUM_INSTANCES / NUM_PARTITIONS;
- ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
- new ReplicaGroupStrategyConfig(PARTITION_COLUMN,
numInstancesPerPartition);
TableConfig tableConfigWithPartitions =
new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITH_PARTITION).setNumReplicas(1)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY))).build();
-
tableConfigWithPartitions.getValidationConfig().setReplicaGroupStrategyConfig(replicaGroupStrategyConfig);
+ new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ PARTITION_COLUMN)))
+ .build();
SegmentAssignment segmentAssignment =
SegmentAssignmentFactory.getSegmentAssignment(helixManager,
tableConfigWithPartitions, null);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinReplicaGroupSegmentAssignmentStrategyTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinReplicaGroupSegmentAssignmentStrategyTest.java
index b018dce2613..23e64feb78f 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinReplicaGroupSegmentAssignmentStrategyTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinReplicaGroupSegmentAssignmentStrategyTest.java
@@ -37,7 +37,6 @@ import
org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignme
import
org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignmentTestUtils;
import
org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignmentUtils;
import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
@@ -94,8 +93,9 @@ public class
RoundRobinReplicaGroupSegmentAssignmentStrategyTest {
new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITHOUT_PARTITION)
.setNumReplicas(NUM_REPLICAS)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new SegmentAssignmentConfig(
-
AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY))).build();
+ new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ null)))
+ .build();
_segmentAssignmentWithoutPartition =
SegmentAssignmentFactory.getSegmentAssignment(null,
tableConfigWithoutPartitions, null);
@@ -139,15 +139,13 @@ public class
RoundRobinReplicaGroupSegmentAssignmentStrategyTest {
when(helixManagerWithPartitions.getHelixPropertyStore()).thenReturn(propertyStoreWithPartitions);
int numInstancesPerPartition = numInstancesPerReplicaGroup /
NUM_PARTITIONS;
- ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
- new ReplicaGroupStrategyConfig(PARTITION_COLUMN,
numInstancesPerPartition);
TableConfig tableConfigWithPartitions =
new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITH_PARTITION)
.setNumReplicas(NUM_REPLICAS)
- .setReplicaGroupStrategyConfig(replicaGroupStrategyConfig)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new SegmentAssignmentConfig(
-
AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY))).build();
+ new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ PARTITION_COLUMN)))
+ .build();
_segmentAssignmentWithPartition =
SegmentAssignmentFactory.getSegmentAssignment(helixManagerWithPartitions,
tableConfigWithPartitions, null);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinSegmentAssignmentStrategyTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinSegmentAssignmentStrategyTest.java
index 882d9257d6b..a3acd3ef0a3 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinSegmentAssignmentStrategyTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/RoundRobinSegmentAssignmentStrategyTest.java
@@ -64,7 +64,8 @@ public class RoundRobinSegmentAssignmentStrategyTest {
TableConfig tableConfig =
new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_SEGMENT_ASSIGNMENT_STRATEGY))).build();
+ new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_SEGMENT_ASSIGNMENT_STRATEGY,
null)))
+ .build();
_segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(null,
tableConfig, null);
// {
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java
index 0918f87d761..b1393f4d1db 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java
@@ -59,7 +59,8 @@ public class SegmentAssignmentStrategyFactoryTest {
public void testSegmentAssignmentStrategyFromTableConfig() {
// Set segment assignment config map in table config for balanced num
segment assignment strategy
Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap = new
HashMap<>();
- segmentAssignmentConfigMap.put(InstancePartitionsType.OFFLINE.toString(),
new SegmentAssignmentConfig("Balanced"));
+ segmentAssignmentConfigMap.put(InstancePartitionsType.OFFLINE.toString(),
+ new SegmentAssignmentConfig("Balanced", null));
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
.setSegmentAssignmentConfigMap(segmentAssignmentConfigMap).build();
@@ -119,10 +120,10 @@ public class SegmentAssignmentStrategyFactoryTest {
new ReplicaGroupStrategyConfig(PARTITION_COLUMN,
numInstancesPerPartition);
Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap = new
HashMap<>();
segmentAssignmentConfigMap.put(InstancePartitionsType.OFFLINE.toString(),
- new SegmentAssignmentConfig("ReplicaGroup"));
+ new SegmentAssignmentConfig("ReplicaGroup", null));
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITH_PARTITION)
-
.setNumReplicas(NUM_REPLICAS).setSegmentAssignmentConfigMap(segmentAssignmentConfigMap)
- .setReplicaGroupStrategyConfig(replicaGroupStrategyConfig).build();
+
.setNumReplicas(NUM_REPLICAS).setSegmentAssignmentConfigMap(segmentAssignmentConfigMap).build();
+
tableConfig.getValidationConfig().setReplicaGroupStrategyConfig(replicaGroupStrategyConfig);
// {
// 0_0=[instance_0, instance_1], 1_0=[instance_2, instance_3],
2_0=[instance_4, instance_5],
diff --git
a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index ba8218a01ce..327eae47fd0 100644
---
a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++
b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -73,7 +73,6 @@ import
org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.DedupConfig;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.QueryConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -305,11 +304,6 @@ public abstract class BaseClusterIntegrationTest extends
ClusterTest {
return null;
}
- @Nullable
- protected ReplicaGroupStrategyConfig getReplicaGroupStrategyConfig() {
- return null;
- }
-
/// Creates a new schema.
protected Schema createSchema()
throws IOException {
@@ -475,8 +469,7 @@ public abstract class BaseClusterIntegrationTest extends
ClusterTest {
.setStreamConfigs(getStreamConfigs())
.setNullHandlingEnabled(getNullHandlingEnabled())
.setSegmentPartitionConfig(getSegmentPartitionConfig())
- .setOptimizeNoDictStatsCollection(true)
- .setReplicaGroupStrategyConfig(getReplicaGroupStrategyConfig());
+ .setOptimizeNoDictStatsCollection(true);
}
/// Creates a new Upsert enabled table config.
@@ -504,7 +497,6 @@ public abstract class BaseClusterIntegrationTest extends
ClusterTest {
.setRoutingConfig(
new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
.setSegmentPartitionConfig(new
SegmentPartitionConfig(columnPartitionConfigMap))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(primaryKeyColumn, 1))
.setOptimizeNoDictStatsCollection(true)
.setUpsertConfig(upsertConfig)
.build();
@@ -563,7 +555,6 @@ public abstract class BaseClusterIntegrationTest extends
ClusterTest {
.setRoutingConfig(
new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
.setSegmentPartitionConfig(new
SegmentPartitionConfig(columnPartitionConfigMap))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(primaryKeyColumn, 1))
.setOptimizeNoDictStatsCollection(true)
.setUpsertConfig(upsertConfig)
.build();
@@ -589,7 +580,6 @@ public abstract class BaseClusterIntegrationTest extends
ClusterTest {
.setRoutingConfig(
new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
.setSegmentPartitionConfig(new
SegmentPartitionConfig(columnPartitionConfigMap))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(primaryKeyColumn, 1))
.setDedupConfig(new DedupConfig())
.setOptimizeNoDictStatsCollection(true)
.build();
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseDedupIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseDedupIntegrationTest.java
index be8adb5570c..b4bce719828 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseDedupIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseDedupIntegrationTest.java
@@ -33,7 +33,6 @@ import org.apache.pinot.core.routing.SegmentsToQuery;
import org.apache.pinot.core.transport.ServerInstance;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.DedupConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -169,7 +168,6 @@ public class BaseDedupIntegrationTest extends
BaseClusterIntegrationTestSet {
.setRoutingConfig(
new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
.setSegmentPartitionConfig(new
SegmentPartitionConfig(columnPartitionConfigMap))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(primaryKeyColumn, 2))
.setDedupConfig(dedupConfig)
.build();
if (PauselessConsumptionUtils.isPauselessEnabled(tableConfig)) {
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/DedupPreloadIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/DedupPreloadIntegrationTest.java
index aabda566d7e..731955fc06c 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/DedupPreloadIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/DedupPreloadIntegrationTest.java
@@ -27,7 +27,6 @@ import java.util.Map;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.DedupConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -180,7 +179,6 @@ public class DedupPreloadIntegrationTest extends
BaseClusterIntegrationTestSet {
.setRoutingConfig(
new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
.setSegmentPartitionConfig(new
SegmentPartitionConfig(columnPartitionConfigMap))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(primaryKeyColumn, 1))
.setDedupConfig(dedupConfig)
.build();
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/OfflineUpsertTableTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/OfflineUpsertTableTest.java
index c5c68de0797..104b80d9b06 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/OfflineUpsertTableTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/OfflineUpsertTableTest.java
@@ -25,16 +25,21 @@ import java.util.List;
import java.util.Map;
import org.apache.pinot.client.ResultSet;
import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.config.TagNameUtils;
import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.UpsertConfig;
+import org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig;
+import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
+import
org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
+import org.apache.pinot.spi.config.table.assignment.InstanceTagPoolConfig;
+import org.apache.pinot.spi.config.table.assignment.SegmentAssignmentConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -102,6 +107,12 @@ public class OfflineUpsertTableTest extends
CustomDataQueryClusterIntegrationTes
Map<String, ColumnPartitionConfig> columnPartitionConfigMap = new
HashMap<>();
columnPartitionConfigMap.put(PRIMARY_KEY_COL, new
ColumnPartitionConfig("Murmur", NUM_PARTITIONS));
+ InstanceTagPoolConfig tagPoolConfig = new InstanceTagPoolConfig(
+ TagNameUtils.getOfflineTagForTenant(TagNameUtils.DEFAULT_TENANT_NAME),
false, 0, null);
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
+ new InstanceReplicaGroupPartitionConfig(true, 0, 2, 0, NUM_PARTITIONS,
1, false, null);
+ InstanceAssignmentConfig instanceAssignmentConfig =
+ new InstanceAssignmentConfig(tagPoolConfig, null,
replicaGroupPartitionConfig, null, false);
return new TableConfigBuilder(TableType.OFFLINE)
.setTableName(TABLE_NAME)
@@ -111,7 +122,9 @@ public class OfflineUpsertTableTest extends
CustomDataQueryClusterIntegrationTes
.setRoutingConfig(new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
.setSegmentPartitionConfig(new
SegmentPartitionConfig(columnPartitionConfigMap))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(PRIMARY_KEY_COL, 1))
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.name(),
instanceAssignmentConfig))
+
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.name(),
+ new SegmentAssignmentConfig(null, PRIMARY_KEY_COL)))
.build();
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/VectorUpsertTableTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/VectorUpsertTableTest.java
index 1ab4107b123..a3003306956 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/VectorUpsertTableTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/VectorUpsertTableTest.java
@@ -42,7 +42,6 @@ import
org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.server.starter.helix.BaseServerStarter;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.FieldConfig;
-import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -317,7 +316,6 @@ public class VectorUpsertTableTest extends
CustomDataQueryClusterIntegrationTest
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
.setSegmentPartitionConfig(new SegmentPartitionConfig(
Map.of(ENTITY_ID_COLUMN, new ColumnPartitionConfig("Murmur",
NUM_PARTITIONS))))
- .setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig(ENTITY_ID_COLUMN, 1))
.build();
}
diff --git
a/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json
b/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json
index 838df06bb51..9def7e24d14 100644
---
a/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json
+++
b/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json
@@ -5,11 +5,7 @@
"timeColumnName": "mtime",
"timeType": "MILLISECONDS",
"segmentPushType": "APPEND",
- "replicasPerPartition": "1",
- "replicaGroupStrategyConfig": {
- "partitionColumn": "event_id",
- "numInstancesPerPartition": 1
- }
+ "replicasPerPartition": "1"
},
"tenants": {},
"tableIndexConfig": {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 4e76120982b..b933a997ddb 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -22,16 +22,17 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
@@ -84,6 +85,7 @@ import org.apache.pinot.spi.config.table.TimestampConfig;
import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
+import
org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
import org.apache.pinot.spi.config.table.assignment.SegmentAssignmentConfig;
import org.apache.pinot.spi.config.table.ingestion.AggregationConfig;
import org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig;
@@ -141,9 +143,10 @@ public final class TableConfigUtils {
private static final String KINESIS_STREAM_TYPE = "kinesis";
private static final String CONSUMING_SEGMENT_TIER = "consuming";
- private static final Set<String> UPSERT_DEDUP_ALLOWED_ROUTING_STRATEGIES =
-
ImmutableSet.of(RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE,
- RoutingConfig.MULTI_STAGE_REPLICA_GROUP_SELECTOR_TYPE);
+ private static final Set<String> UPSERT_DEDUP_ALLOWED_ROUTING_STRATEGIES =
Set.of(
+ RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE,
+ RoutingConfig.MULTI_STAGE_REPLICA_GROUP_SELECTOR_TYPE
+ );
private static boolean _disableGroovy;
@@ -211,7 +214,8 @@ public final class TableConfigUtils {
existingOverriddenColumns);
validateValidationConfig(tableConfig, schema);
- validateSegmentAssignmentConfig(tableConfig);
+ validateSegmentAssignmentConfigs(tableConfig);
+ validatePartitionColumnConsistency(tableConfig);
validateIngestionConfig(tableConfig, schema, existingTableConfig);
if (tableConfig.getTableType() == TableType.REALTIME) {
validateStreamConfigMaps(tableConfig);
@@ -431,29 +435,79 @@ public final class TableConfigUtils {
validateRetentionConfig(tableConfig);
}
- private static void validateSegmentAssignmentConfig(TableConfig tableConfig)
{
+ /// Validates the segment assignment configs: the strategy must be a known
one (`allservers` for a dimension table),
+ /// and the partition column, which only applies to partition based
strategies, must be a column of the segment
+ /// partition config.
+ @VisibleForTesting
+ static void validateSegmentAssignmentConfigs(TableConfig tableConfig) {
Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
tableConfig.getSegmentAssignmentConfigMap();
- if (segmentAssignmentConfigMap == null) {
+ if (MapUtils.isEmpty(segmentAssignmentConfigMap)) {
return;
}
- if (tableConfig.isDimTable()) {
- SegmentAssignmentConfig segmentAssignmentConfig =
-
segmentAssignmentConfigMap.get(InstancePartitionsType.OFFLINE.toString());
- if (segmentAssignmentConfig == null) {
- return;
- }
- String segmentAssignmentStrategy =
segmentAssignmentConfig.getAssignmentStrategy();
- if (segmentAssignmentStrategy != null
- &&
!segmentAssignmentStrategy.equalsIgnoreCase(AssignmentStrategy.DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY))
{
- throw new IllegalStateException(
- String.format("Dimension table: %s can only use '%s' segment
assignment strategy, found: %s",
- tableConfig.getTableName(),
-
CommonConstants.Segment.AssignmentStrategy.DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY,
- segmentAssignmentStrategy));
+ String tableName = tableConfig.getTableName();
+ boolean isDimTable = tableConfig.isDimTable();
+ for (Map.Entry<String, SegmentAssignmentConfig> entry :
segmentAssignmentConfigMap.entrySet()) {
+ String key = entry.getKey();
+ SegmentAssignmentConfig segmentAssignmentConfig = entry.getValue();
+ String assignmentStrategy =
segmentAssignmentConfig.getAssignmentStrategy();
+ boolean partitionBased = !isDimTable;
+ if (assignmentStrategy != null) {
+ assignmentStrategy = assignmentStrategy.toLowerCase(Locale.ROOT);
+ if (isDimTable) {
+
Preconditions.checkState(assignmentStrategy.equals(AssignmentStrategy.DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY),
+ "Dimension table: %s can only use '%s' segment assignment
strategy, found: %s", tableName,
+ AssignmentStrategy.DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY,
assignmentStrategy);
+ } else {
+
Preconditions.checkState(AssignmentStrategy.CONFIGURABLE_STRATEGIES.contains(assignmentStrategy),
+ "Invalid segment assignment strategy: %s for: %s of table: %s",
assignmentStrategy, key, tableName);
+ }
+ partitionBased =
AssignmentStrategy.PARTITION_BASED_STRATEGIES.contains(assignmentStrategy);
+ }
+ String partitionColumn = segmentAssignmentConfig.getPartitionColumn();
+ if (StringUtils.isNotEmpty(partitionColumn)) {
+ Preconditions.checkState(partitionBased,
+ "Partition column: %s does not apply to the segment assignment
strategy for: %s of table: %s",
+ partitionColumn, key, tableName);
+ validatePartitionColumnInSegmentPartitionConfig(tableConfig,
partitionColumn);
}
}
}
+ /// Validates that every place that configures the partition column for
segment assignment names the same column:
+ /// the segment assignment configs, and the deprecated instance assignment
configs and `replicaGroupStrategyConfig`
+ /// that existing tables may still carry. The column is resolved table-wide,
so a disagreement would be settled by
+ /// precedence and map iteration order.
+ @VisibleForTesting
+ static void validatePartitionColumnConsistency(TableConfig tableConfig) {
+ Set<String> partitionColumns = new TreeSet<>();
+ Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
tableConfig.getSegmentAssignmentConfigMap();
+ if (segmentAssignmentConfigMap != null) {
+ for (SegmentAssignmentConfig segmentAssignmentConfig :
segmentAssignmentConfigMap.values()) {
+ addIfNotEmpty(partitionColumns,
segmentAssignmentConfig.getPartitionColumn());
+ }
+ }
+ Map<String, InstanceAssignmentConfig> instanceAssignmentConfigMap =
tableConfig.getInstanceAssignmentConfigMap();
+ if (instanceAssignmentConfigMap != null) {
+ for (InstanceAssignmentConfig instanceAssignmentConfig :
instanceAssignmentConfigMap.values()) {
+ addIfNotEmpty(partitionColumns,
instanceAssignmentConfig.getReplicaGroupPartitionConfig().getPartitionColumn());
+ }
+ }
+ ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
+ tableConfig.getValidationConfig().getReplicaGroupStrategyConfig();
+ if (replicaGroupStrategyConfig != null) {
+ addIfNotEmpty(partitionColumns,
replicaGroupStrategyConfig.getPartitionColumn());
+ }
+ Preconditions.checkState(partitionColumns.size() <= 1,
+ "Table: %s must configure the same partition column in
segmentAssignmentConfigMap, instanceAssignmentConfigMap "
+ + "and replicaGroupStrategyConfig, found: %s",
tableConfig.getTableName(), partitionColumns);
+ }
+
+ private static void addIfNotEmpty(Set<String> values, @Nullable String
value) {
+ if (StringUtils.isNotEmpty(value)) {
+ values.add(value);
+ }
+ }
+
private static boolean isValidPeerDownloadScheme(String
peerSegmentDownloadScheme) {
return
CommonConstants.HTTP_PROTOCOL.equalsIgnoreCase(peerSegmentDownloadScheme)
||
CommonConstants.HTTPS_PROTOCOL.equalsIgnoreCase(peerSegmentDownloadScheme);
@@ -1426,6 +1480,18 @@ public final class TableConfigUtils {
if (tableConfig.getInstanceAssignmentConfigMap() == null) {
return;
}
+ // Keys of the instance assignment configs that place OFFLINE segments:
the OFFLINE instance partitions and the
+ // tiers of an OFFLINE table
+ Set<String> offlineSegmentKeys = new HashSet<>();
+ if (tableConfig.getTableType() == TableType.OFFLINE) {
+ offlineSegmentKeys.add(InstancePartitionsType.OFFLINE.name());
+ List<TierConfig> tierConfigs = tableConfig.getTierConfigsList();
+ if (tierConfigs != null) {
+ for (TierConfig tierConfig : tierConfigs) {
+ offlineSegmentKeys.add(tierConfig.getName());
+ }
+ }
+ }
for (Map.Entry<String, InstanceAssignmentConfig>
instanceAssignmentConfigMapEntry
: tableConfig.getInstanceAssignmentConfigMap()
.entrySet()) {
@@ -1448,9 +1514,61 @@ public final class TableConfigUtils {
"numInstancesPerPartition must be 1 when using
IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR");
}
// TODO: Add more validations for other partition selectors here
+
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
+ instanceAssignmentConfig.getReplicaGroupPartitionConfig();
+ // Explicitly partitioned instance partitions require the OFFLINE
segments to be placed by partition
+ if (offlineSegmentKeys.contains(instancePartitionsType) &&
replicaGroupPartitionConfig.isReplicaGroupBased()
+ && replicaGroupPartitionConfig.getNumPartitions() > 1) {
+ validatePartitionedSegmentAssignment(tableConfig,
instancePartitionsType);
+ }
+ }
+ }
+
+ /// Validates that OFFLINE segments are placed by partition for the given
instance partitions type or tier, which
+ /// is required when its instance partitions have multiple partitions. The
segment assignment strategy configured
+ /// for it, if any, must be partition based, and the partition column must
be resolvable: the column configured
+ /// for segment assignment must be a column of the segment partition config,
and without a configured column the
+ /// segment partition config must have exactly one column, which is how the
partition id of a segment is read from
+ /// its metadata. The OFFLINE instance partitions may also be
pre-configured, in which case the controller runs
+ /// this validation on the fetched ones. Dimension tables are exempt, since
their segments are assigned to all
+ /// servers regardless of the instance partitions.
+ public static void validatePartitionedSegmentAssignment(TableConfig
tableConfig, String key) {
+ if (tableConfig.isDimTable()) {
+ return;
+ }
+ Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
tableConfig.getSegmentAssignmentConfigMap();
+ SegmentAssignmentConfig segmentAssignmentConfig =
+ segmentAssignmentConfigMap != null ?
segmentAssignmentConfigMap.get(key) : null;
+ String assignmentStrategy =
+ segmentAssignmentConfig != null ?
segmentAssignmentConfig.getAssignmentStrategy() : null;
+ if (assignmentStrategy != null) {
+ Preconditions.checkState(
+
AssignmentStrategy.PARTITION_BASED_STRATEGIES.contains(assignmentStrategy.toLowerCase(Locale.ROOT)),
+ "Segment assignment strategy: %s for: %s of table: %s must be
partition based because the instance "
+ + "partitions are partitioned", assignmentStrategy, key,
tableConfig.getTableName());
+ }
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
getColumnPartitionMap(tableConfig);
+ Preconditions.checkState(!columnPartitionMap.isEmpty(),
+ "Offline table with partitioned instance partitions must have segment
partition config "
+ + "(segmentPartitionConfig in the indexingConfig) to resolve the
partition of a segment");
+ String partitionColumn = getPartitionColumn(tableConfig);
+ if (partitionColumn != null) {
+ validatePartitionColumnInSegmentPartitionConfig(tableConfig,
partitionColumn);
+ } else {
+ Preconditions.checkState(columnPartitionMap.size() == 1,
+ "Offline table with partitioned instance partitions must configure
the partition column (partitionColumn in "
+ + "the segment assignment config) when segment partition config
has multiple columns");
}
}
+ private static void
validatePartitionColumnInSegmentPartitionConfig(TableConfig tableConfig,
+ String partitionColumn) {
+
Preconditions.checkState(getColumnPartitionMap(tableConfig).containsKey(partitionColumn),
+ "Partition column: %s configured for segment assignment is not found
in segment partition config",
+ partitionColumn);
+ }
+
/// Validates the partial upsert-related configurations:
/// - Null handling must be enabled
/// - Merger cannot be applied to private key columns
@@ -2594,26 +2712,36 @@ public final class TableConfigUtils {
return tierOverwrites != null && tierOverwrites.isObject() ?
tierOverwrites.get(tier) : null;
}
- /// Get the partition column from tableConfig instance assignment config map.
- /// @param tableConfig table config
- /// @return partition column
+ /// Returns the partition column configured for segment assignment, or
`null` if none is configured. The column is
+ /// read from the segment assignment configs, falling back to the deprecated
partition column of the instance
+ /// assignment configs and then to the deprecated
`replicaGroupStrategyConfig`.
+ @Nullable
public static String getPartitionColumn(TableConfig tableConfig) {
- // check InstanceAssignmentConfigMap is null or empty,
- if (!MapUtils.isEmpty(tableConfig.getInstanceAssignmentConfigMap())) {
- for (InstanceAssignmentConfig instanceAssignmentConfig :
tableConfig.getInstanceAssignmentConfigMap().values()) {
- //check InstanceAssignmentConfig has the
InstanceReplicaGroupPartitionConfig with non-empty partitionColumn
- if
(StringUtils.isNotEmpty(instanceAssignmentConfig.getReplicaGroupPartitionConfig().getPartitionColumn()))
{
- return
instanceAssignmentConfig.getReplicaGroupPartitionConfig().getPartitionColumn();
+ // TODO: Revisit to resolve the partition column per key (instance
partitions type or tier) instead of returning
+ // the first configured one across all entries.
+ Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
tableConfig.getSegmentAssignmentConfigMap();
+ if (segmentAssignmentConfigMap != null) {
+ for (SegmentAssignmentConfig segmentAssignmentConfig :
segmentAssignmentConfigMap.values()) {
+ String partitionColumn = segmentAssignmentConfig.getPartitionColumn();
+ if (StringUtils.isNotEmpty(partitionColumn)) {
+ return partitionColumn;
+ }
+ }
+ }
+ Map<String, InstanceAssignmentConfig> instanceAssignmentConfigMap =
tableConfig.getInstanceAssignmentConfigMap();
+ if (instanceAssignmentConfigMap != null) {
+ for (InstanceAssignmentConfig instanceAssignmentConfig :
instanceAssignmentConfigMap.values()) {
+ String partitionColumn =
instanceAssignmentConfig.getReplicaGroupPartitionConfig().getPartitionColumn();
+ if (StringUtils.isNotEmpty(partitionColumn)) {
+ return partitionColumn;
}
}
}
-
- // for backward-compatibility, If partitionColumn value isn't there in
InstanceReplicaGroupPartitionConfig
- // check ReplicaGroupStrategyConfig for partitionColumn
- //noinspection deprecation
ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
tableConfig.getValidationConfig().getReplicaGroupStrategyConfig();
- return replicaGroupStrategyConfig != null ?
replicaGroupStrategyConfig.getPartitionColumn() : null;
+ String partitionColumn =
+ replicaGroupStrategyConfig != null ?
replicaGroupStrategyConfig.getPartitionColumn() : null;
+ return StringUtils.isNotEmpty(partitionColumn) ? partitionColumn : null;
}
public static Set<String> getRelevantTags(TableConfig tableConfig) {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 66eef1d9794..ed33075921d 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -21,6 +21,7 @@ package org.apache.pinot.segment.local.utils;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
+import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -299,7 +300,7 @@ public class TableConfigUtilsTest {
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setIsDimTable(true)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new
SegmentAssignmentConfig(AssignmentStrategy.DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY)))
+ new
SegmentAssignmentConfig(AssignmentStrategy.DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY,
null)))
.build();
TableConfigUtils.validate(tableConfig, schema);
@@ -307,7 +308,7 @@ public class TableConfigUtilsTest {
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setIsDimTable(true)
.setSegmentAssignmentConfigMap(Map.of(InstancePartitionsType.OFFLINE.toString(),
- new SegmentAssignmentConfig("replicaGroup")))
+ new SegmentAssignmentConfig("replicaGroup", null)))
.build();
try {
TableConfigUtils.validate(tableConfig, schema);
@@ -3851,6 +3852,7 @@ public class TableConfigUtilsTest {
TableConfigUtils.validateTTLForUpsertConfig(tableConfigWithValidTimestampTTLConfig,
schema);
}
+ @SuppressWarnings("removal")
@Test
public void testValidatePartitionedReplicaGroupInstance() {
String partitionColumn = "testPartitionCol";
@@ -4300,18 +4302,259 @@ public class TableConfigUtilsTest {
assertNull(TableConfigUtils.getPartitionColumn(tableConfig));
}
- @SuppressWarnings("deprecation")
+ @SuppressWarnings("removal")
@Test
- public void testGetPartitionColumnWithReplicaGroupConfig() {
- ReplicaGroupStrategyConfig replicaGroupStrategyConfig = new
ReplicaGroupStrategyConfig(PARTITION_COLUMN, 1);
- TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).build();
+ public void testValidateAcceptsReplicaGroupStrategyConfig()
+ throws IOException {
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).build();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ tableConfig.getValidationConfig().setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig("myCol", 1));
+ // The deprecated config must still be parsed and accepted
+
TableConfigUtils.validate(JsonUtils.stringToObject(tableConfig.toJsonString(),
TableConfig.class), schema);
+ }
+
+ @Test
+ public void testValidateSegmentAssignmentConfigs() {
+ String offlineKey = InstancePartitionsType.OFFLINE.name();
+ SegmentPartitionConfig segmentPartitionConfig =
+ new SegmentPartitionConfig(Map.of("myCol", new
ColumnPartitionConfig("murmur", 4)));
+
+ // Both the strategy and the partition column are optional, and the
strategy is case insensitive
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig(null, null)))
+ .build();
+ TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig);
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig(null, "myCol")));
+ TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig);
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey,
+ new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ "myCol")));
+ TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig);
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig("ReplicaGroup", "myCol")));
+ TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig);
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey,
+ new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_SEGMENT_ASSIGNMENT_STRATEGY,
null)));
+ TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig);
+
+ // The strategy must be a known one
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig("unknown", null)));
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig));
+ assertEquals(e.getMessage(),
+ "Invalid segment assignment strategy: unknown for: OFFLINE of table:
testTable_OFFLINE");
+
+ // The partition column only applies to partition based strategies
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey,
+ new
SegmentAssignmentConfig(AssignmentStrategy.BALANCE_NUM_SEGMENT_ASSIGNMENT_STRATEGY,
"myCol")));
+ e = expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig));
+ assertEquals(e.getMessage(), "Partition column: myCol does not apply to
the segment assignment strategy for: "
+ + "OFFLINE of table: testTable_OFFLINE");
+ TableConfig dimTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setIsDimTable(true)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig(null, "myCol")))
+ .build();
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validateSegmentAssignmentConfigs(dimTableConfig));
+ assertEquals(e.getMessage(), "Partition column: myCol does not apply to
the segment assignment strategy for: "
+ + "OFFLINE of table: testTable_OFFLINE");
+
+ // The partition column must be a column of the segment partition config
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig(null, "otherCol")));
+ e = expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validateSegmentAssignmentConfigs(tableConfig));
+ assertEquals(e.getMessage(),
+ "Partition column: otherCol configured for segment assignment is not
found in segment partition config");
+ }
+
+ private static InstanceAssignmentConfig
createInstanceAssignmentConfig(@Nullable String partitionColumn) {
+ return new InstanceAssignmentConfig(new
InstanceTagPoolConfig("DefaultTenant_OFFLINE", false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(true, 0, 2, 0, 4, 1, false,
partitionColumn), null, false);
+ }
+
+ @Test
+ public void testValidatePartitionedSegmentAssignment() {
+ String offlineKey = InstancePartitionsType.OFFLINE.name();
+ SegmentPartitionConfig singleColumnPartitionConfig =
+ new SegmentPartitionConfig(Map.of("myCol", new
ColumnPartitionConfig("murmur", 4)));
+ SegmentPartitionConfig multiColumnPartitionConfig = new
SegmentPartitionConfig(Map.of(
+ "myCol", new ColumnPartitionConfig("murmur", 4),
+ "otherCol", new ColumnPartitionConfig("murmur", 4)
+ ));
+
+ // Without configured partition column, the segment partition config must
have exactly one column
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ TableConfig noPartitionConfig = tableConfig;
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validatePartitionedSegmentAssignment(noPartitionConfig,
offlineKey));
+ assertEquals(e.getMessage(), "Offline table with partitioned instance
partitions must have segment partition "
+ + "config (segmentPartitionConfig in the indexingConfig) to resolve
the partition of a segment");
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(singleColumnPartitionConfig)
+ .build();
+ TableConfigUtils.validatePartitionedSegmentAssignment(tableConfig,
offlineKey);
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(multiColumnPartitionConfig)
+ .build();
+ TableConfig multiColumnConfig = tableConfig;
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validatePartitionedSegmentAssignment(multiColumnConfig,
offlineKey));
+ assertEquals(e.getMessage(), "Offline table with partitioned instance
partitions must configure the partition "
+ + "column (partitionColumn in the segment assignment config) when
segment partition config has multiple "
+ + "columns");
+
+ // The configured partition column must be a column of the segment
partition config
+ Map<String, SegmentAssignmentConfig> segmentAssignmentConfigMap =
+ Map.of(offlineKey, new SegmentAssignmentConfig(null, "myCol"));
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(multiColumnPartitionConfig)
+ .setSegmentAssignmentConfigMap(segmentAssignmentConfigMap)
+ .build();
+ TableConfigUtils.validatePartitionedSegmentAssignment(tableConfig,
offlineKey);
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(
+ new SegmentPartitionConfig(Map.of("otherCol", new
ColumnPartitionConfig("murmur", 4))))
+ .setSegmentAssignmentConfigMap(segmentAssignmentConfigMap)
+ .build();
+ TableConfig missingColumnConfig = tableConfig;
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validatePartitionedSegmentAssignment(missingColumnConfig,
offlineKey));
+ assertEquals(e.getMessage(),
+ "Partition column: myCol configured for segment assignment is not
found in segment partition config");
- // setting up ReplicaGroupStrategyConfig for backward compatibility test.
- SegmentsValidationAndRetentionConfig validationConfig = new
SegmentsValidationAndRetentionConfig();
- validationConfig.setReplicaGroupStrategyConfig(replicaGroupStrategyConfig);
- tableConfig.setValidationConfig(validationConfig);
+ // The deprecated partition column of the instance assignment config is
still honored
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(multiColumnPartitionConfig)
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createInstanceAssignmentConfig("myCol")))
+ .build();
+ TableConfigUtils.validatePartitionedSegmentAssignment(tableConfig,
offlineKey);
+ // The segment assignment strategy configured for the key, if any, must be
partition based
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(singleColumnPartitionConfig)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
+ new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ null)))
+ .build();
+ TableConfigUtils.validatePartitionedSegmentAssignment(tableConfig,
offlineKey);
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(singleColumnPartitionConfig)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
+ new
SegmentAssignmentConfig(AssignmentStrategy.BALANCE_NUM_SEGMENT_ASSIGNMENT_STRATEGY,
null)))
+ .build();
+ TableConfig balancedConfig = tableConfig;
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validatePartitionedSegmentAssignment(balancedConfig,
offlineKey));
+ assertEquals(e.getMessage(), "Segment assignment strategy: balanced for:
OFFLINE of table: testTable_OFFLINE must "
+ + "be partition based because the instance partitions are
partitioned");
+
+ // Dimension tables are assigned to all servers regardless of the instance
partitions
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setIsDimTable(true)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
+ new
SegmentAssignmentConfig(AssignmentStrategy.DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY,
null)))
+ .build();
+ TableConfigUtils.validatePartitionedSegmentAssignment(tableConfig,
offlineKey);
+
+ // Partitioned OFFLINE instance assignment config runs the validation
+ InstanceAssignmentConfig instanceAssignmentConfig =
createInstanceAssignmentConfig(null);
+ Map<String, InstanceAssignmentConfig> instanceAssignmentConfigMap =
Map.of(offlineKey, instanceAssignmentConfig);
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setInstanceAssignmentConfigMap(instanceAssignmentConfigMap)
+ .build();
+ TableConfig partitionedConfig = tableConfig;
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validateInstanceAssignmentConfigs(partitionedConfig));
+ assertEquals(e.getMessage(), "Offline table with partitioned instance
partitions must have segment partition "
+ + "config (segmentPartitionConfig in the indexingConfig) to resolve
the partition of a segment");
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentPartitionConfig(singleColumnPartitionConfig)
+ .setInstanceAssignmentConfigMap(instanceAssignmentConfigMap)
+ .build();
+ TableConfigUtils.validateInstanceAssignmentConfigs(tableConfig);
+
+ // Partitioned instance assignment config of a tier of an OFFLINE table
runs the validation too
+ List<TierConfig> tierConfigs = List.of(new TierConfig("hotTier",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "30d",
+ null, TierFactory.PINOT_SERVER_STORAGE_TYPE, "hotTier_OFFLINE", null,
null));
+ Map<String, InstanceAssignmentConfig> tierInstanceAssignmentConfigMap =
Map.of("hotTier", instanceAssignmentConfig);
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setTierConfigList(tierConfigs)
+ .setInstanceAssignmentConfigMap(tierInstanceAssignmentConfigMap)
+ .build();
+ TableConfig tierConfigWithoutPartitionConfig = tableConfig;
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validateInstanceAssignmentConfigs(tierConfigWithoutPartitionConfig));
+ assertEquals(e.getMessage(), "Offline table with partitioned instance
partitions must have segment partition "
+ + "config (segmentPartitionConfig in the indexingConfig) to resolve
the partition of a segment");
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setTierConfigList(tierConfigs)
+ .setSegmentPartitionConfig(singleColumnPartitionConfig)
+ .setInstanceAssignmentConfigMap(tierInstanceAssignmentConfigMap)
+ .build();
+ TableConfigUtils.validateInstanceAssignmentConfigs(tableConfig);
+
+ // Instance assignment configs of a REALTIME table are not validated
because the partition of a realtime segment
+ // is encoded in its name
+ tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.COMPLETED.name(),
instanceAssignmentConfig))
+ .build();
+ TableConfigUtils.validateInstanceAssignmentConfigs(tableConfig);
+ }
+
+ @SuppressWarnings("removal")
+ @Test
+ public void testValidatePartitionColumnConsistency() {
+ String offlineKey = InstancePartitionsType.OFFLINE.name();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentAssignmentConfigMap(Map.of(
+ offlineKey, new SegmentAssignmentConfig(null, "myCol"),
+ "hotTier", new SegmentAssignmentConfig(null, "myCol")
+ ))
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createInstanceAssignmentConfig("myCol")))
+ .build();
+ TableConfigUtils.validatePartitionColumnConsistency(tableConfig);
+
+ // A different column is rejected wherever it is configured
+ String expectedMessage = "Table: testTable_OFFLINE must configure the same
partition column in "
+ + "segmentAssignmentConfigMap, instanceAssignmentConfigMap and
replicaGroupStrategyConfig, found: [myCol, "
+ + "otherCol]";
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(
+ offlineKey, new SegmentAssignmentConfig(null, "myCol"),
+ "hotTier", new SegmentAssignmentConfig(null, "otherCol")
+ ));
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validatePartitionColumnConsistency(tableConfig));
+ assertEquals(e.getMessage(), expectedMessage);
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig(null, "myCol")));
+ tableConfig.setInstanceAssignmentConfigMap(Map.of(offlineKey,
createInstanceAssignmentConfig("otherCol")));
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validatePartitionColumnConsistency(tableConfig));
+ assertEquals(e.getMessage(), expectedMessage);
+ tableConfig.setInstanceAssignmentConfigMap(null);
+ tableConfig.getValidationConfig().setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig("otherCol", 1));
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validatePartitionColumnConsistency(tableConfig));
+ assertEquals(e.getMessage(), expectedMessage);
+ }
+
+ @SuppressWarnings("removal")
+ @Test
+ public void testGetPartitionColumn() {
+ String offlineKey = InstancePartitionsType.OFFLINE.name();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig(null, PARTITION_COLUMN)))
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createInstanceAssignmentConfig("instanceAssignmentCol")))
+ .build();
+ tableConfig.getValidationConfig().setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig("legacyCol", 1));
+
+ // The segment assignment config takes precedence over the deprecated
configs of existing tables
assertEquals(TableConfigUtils.getPartitionColumn(tableConfig),
PARTITION_COLUMN);
+ tableConfig.setSegmentAssignmentConfigMap(Map.of(offlineKey, new
SegmentAssignmentConfig(null, null)));
+ assertEquals(TableConfigUtils.getPartitionColumn(tableConfig),
"instanceAssignmentCol");
+ tableConfig.setInstanceAssignmentConfigMap(null);
+ assertEquals(TableConfigUtils.getPartitionColumn(tableConfig),
"legacyCol");
+ tableConfig.getValidationConfig().setReplicaGroupStrategyConfig(new
ReplicaGroupStrategyConfig("", 1));
+ assertNull(TableConfigUtils.getPartitionColumn(tableConfig));
}
@DataProvider(name = "tableTypeTestDataProvider")
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ReplicaGroupStrategyConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ReplicaGroupStrategyConfig.java
index 4507a3a083c..012bed291cc 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ReplicaGroupStrategyConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ReplicaGroupStrategyConfig.java
@@ -25,7 +25,9 @@ import javax.annotation.Nullable;
import org.apache.pinot.spi.config.BaseJsonConfig;
-/// Class representing configurations related to segment assignment strategy.
+/// Legacy replica-group config, superseded by `InstanceAssignmentConfig` for
the instance assignment and by the
+/// partition column of `SegmentAssignmentConfig`. It is still accepted and
honored until it is removed.
+@Deprecated(since = "1.3.0", forRemoval = true)
public class ReplicaGroupStrategyConfig extends BaseJsonConfig {
private final String _partitionColumn;
private final int _numInstancesPerPartition;
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
index f8c59b64d59..ae12ddae980 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
@@ -40,7 +40,8 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
private String _replicasPerPartition;
private String _timeColumnName;
private TimeUnit _timeType;
- @Deprecated // Use SegmentAssignmentConfig instead
+ // Superseded by the instance assignment config and the segment assignment
config, still accepted and honored
+ @Deprecated(since = "1.3.0", forRemoval = true)
private ReplicaGroupStrategyConfig _replicaGroupStrategyConfig;
private CompletionConfig _completionConfig;
private String _crypterClassName;
@@ -233,13 +234,15 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
_replicasPerPartition = replicasPerPartition;
}
- /// @deprecated Use
[org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig] instead.
+ /// @deprecated Configure the instance assignment in
`instanceAssignmentConfigMap` with
+ /// [org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig]
and the partition column in
+ /// `segmentAssignmentConfigMap` instead. This config is still accepted and
honored.
@Deprecated(since = "1.3.0", forRemoval = true)
public ReplicaGroupStrategyConfig getReplicaGroupStrategyConfig() {
return _replicaGroupStrategyConfig;
}
- @Deprecated
+ @Deprecated(since = "1.3.0", forRemoval = true)
public void setReplicaGroupStrategyConfig(ReplicaGroupStrategyConfig
replicaGroupStrategyConfig) {
_replicaGroupStrategyConfig = replicaGroupStrategyConfig;
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceAssignmentConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceAssignmentConfig.java
index 973cf368c7e..7ad00b20aca 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceAssignmentConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceAssignmentConfig.java
@@ -28,19 +28,20 @@ import org.apache.pinot.spi.config.BaseJsonConfig;
public class InstanceAssignmentConfig extends BaseJsonConfig {
- @JsonPropertyDescription("Configuration for the strategy to assign instances
to partitions")
- private final PartitionSelector _partitionSelector;
@JsonPropertyDescription("Configuration for the instance tag and pool of the
instance assignment (mandatory)")
private final InstanceTagPoolConfig _tagPoolConfig;
- @JsonPropertyDescription("Configuration for the instance constraints of the
instance assignment,"
- + " which filters out unqualified instances and sorts instances for
picking priority")
+ @JsonPropertyDescription("Configuration for the instance constraints of the
instance assignment, which filters out "
+ + "unqualified instances and sorts instances for picking priority")
private final InstanceConstraintConfig _constraintConfig;
@JsonPropertyDescription(
"Configuration for the instance replica-group and partition of the
instance assignment (mandatory)")
private final InstanceReplicaGroupPartitionConfig
_replicaGroupPartitionConfig;
+ @JsonPropertyDescription("Configuration for the strategy to assign instances
to partitions")
+ private final PartitionSelector _partitionSelector;
+
@JsonPropertyDescription("Configuration to minimize data movement for pool
and instance assignment")
private final boolean _minimizeDataMovement;
@@ -49,25 +50,21 @@ public class InstanceAssignmentConfig extends
BaseJsonConfig {
@JsonProperty(value = "tagPoolConfig", required = true)
InstanceTagPoolConfig tagPoolConfig,
@JsonProperty("constraintConfig") @Nullable InstanceConstraintConfig
constraintConfig,
@JsonProperty(value = "replicaGroupPartitionConfig", required = true)
- InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig,
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig,
@JsonProperty("partitionSelector") @Nullable String partitionSelector,
@JsonProperty("minimizeDataMovement") boolean minimizeDataMovement) {
Preconditions.checkArgument(tagPoolConfig != null, "'tagPoolConfig' must
be configured");
- Preconditions
- .checkArgument(replicaGroupPartitionConfig != null,
"'replicaGroupPartitionConfig' must be configured");
+ Preconditions.checkArgument(replicaGroupPartitionConfig != null,
+ "'replicaGroupPartitionConfig' must be configured");
_tagPoolConfig = tagPoolConfig;
_constraintConfig = constraintConfig;
_replicaGroupPartitionConfig = replicaGroupPartitionConfig;
- _partitionSelector =
- partitionSelector == null ?
PartitionSelector.INSTANCE_REPLICA_GROUP_PARTITION_SELECTOR
- : PartitionSelector.valueOf(partitionSelector);
+ _partitionSelector = partitionSelector == null
+ ? PartitionSelector.INSTANCE_REPLICA_GROUP_PARTITION_SELECTOR
+ : PartitionSelector.valueOf(partitionSelector);
_minimizeDataMovement = minimizeDataMovement;
}
- public PartitionSelector getPartitionSelector() {
- return _partitionSelector;
- }
-
public InstanceTagPoolConfig getTagPoolConfig() {
return _tagPoolConfig;
}
@@ -81,12 +78,18 @@ public class InstanceAssignmentConfig extends
BaseJsonConfig {
return _replicaGroupPartitionConfig;
}
+ public PartitionSelector getPartitionSelector() {
+ return _partitionSelector;
+ }
+
public boolean isMinimizeDataMovement() {
return _minimizeDataMovement;
}
public enum PartitionSelector {
- FD_AWARE_INSTANCE_PARTITION_SELECTOR,
INSTANCE_REPLICA_GROUP_PARTITION_SELECTOR,
- MIRROR_SERVER_SET_PARTITION_SELECTOR,
IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR;
+ FD_AWARE_INSTANCE_PARTITION_SELECTOR,
+ INSTANCE_REPLICA_GROUP_PARTITION_SELECTOR,
+ MIRROR_SERVER_SET_PARTITION_SELECTOR,
+ IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR;
}
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceReplicaGroupPartitionConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceReplicaGroupPartitionConfig.java
index 30628bdb68b..d427efb7a84 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceReplicaGroupPartitionConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceReplicaGroupPartitionConfig.java
@@ -52,8 +52,9 @@ public class InstanceReplicaGroupPartitionConfig extends
BaseJsonConfig {
+ "instances if not " + "specified")
private final int _numInstancesPerPartition;
- @JsonPropertyDescription(
- "Name of the column used for partition, if not provided table level
replica group will be used")
+ // Superseded by the partition column of the segment assignment config,
still accepted and honored
+ @Deprecated(since = "1.7.0", forRemoval = true)
+ @JsonPropertyDescription("Deprecated, configure the partition column in the
segment assignment config instead")
private final String _partitionColumn;
// TODO: remove this config in the next official release
@@ -108,6 +109,10 @@ public class InstanceReplicaGroupPartitionConfig extends
BaseJsonConfig {
return _minimizeDataMovement;
}
+ /// @deprecated Configure the partition column in
`segmentAssignmentConfigMap` with [SegmentAssignmentConfig]
+ /// instead, which is where the segment assignment reads it from. This
column is still accepted and honored, and the
+ /// segment assignment config takes precedence when both are configured.
+ @Deprecated(since = "1.7.0", forRemoval = true)
@Nullable
public String getPartitionColumn() {
return _partitionColumn;
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/SegmentAssignmentConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/SegmentAssignmentConfig.java
index eef95257f00..1238c6e59c0 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/SegmentAssignmentConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/SegmentAssignmentConfig.java
@@ -21,21 +21,39 @@ package org.apache.pinot.spi.config.table.assignment;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import javax.annotation.Nullable;
import org.apache.pinot.spi.config.BaseJsonConfig;
+/// Config of how segments are assigned to the instance partitions of an
instance partitions type (`OFFLINE` or
+/// `COMPLETED`) or a tier, keyed by its name in `segmentAssignmentConfigMap`
of the table config. Both fields are
+/// optional: the strategy defaults to the one derived from the instance
partitions, and the partition column defaults
+/// to the single column of the segment partition config.
public class SegmentAssignmentConfig extends BaseJsonConfig {
- @JsonPropertyDescription("Configuration for Segment Assignment Strategy")
+ @JsonPropertyDescription("Segment assignment strategy: balanced,
replicagroup, roundrobin or roundrobinreplicagroup "
+ + "(allservers for a dimension table), derived from the instance
partitions if not specified")
@JsonProperty("segmentAssignmentStrategy")
private final String _assignmentStrategy;
+ @JsonPropertyDescription("Column used to resolve the partition of a segment
for partition based segment assignment, "
+ + "the single column of the segment partition config if not specified")
+ private final String _partitionColumn;
+
@JsonCreator
- public SegmentAssignmentConfig(@JsonProperty(value =
"segmentAssignmentStrategy") String assignmentStrategy) {
+ public SegmentAssignmentConfig(@JsonProperty("segmentAssignmentStrategy")
@Nullable String assignmentStrategy,
+ @JsonProperty("partitionColumn") @Nullable String partitionColumn) {
_assignmentStrategy = assignmentStrategy;
+ _partitionColumn = partitionColumn;
}
+ @Nullable
public String getAssignmentStrategy() {
return _assignmentStrategy;
}
+
+ @Nullable
+ public String getPartitionColumn() {
+ return _partitionColumn;
+ }
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 46b2bd84ddc..6e935bb3cec 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -2536,6 +2536,19 @@ public class CommonConstants {
public static final String REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY =
"replicagroup";
public static final String
ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY =
"roundrobinreplicagroup";
public static final String DIM_TABLE_SEGMENT_ASSIGNMENT_STRATEGY =
"allservers";
+
+ // Strategies that can be configured for a table other than a dimension
table
+ public static final Set<String> CONFIGURABLE_STRATEGIES = Set.of(
+ BALANCE_NUM_SEGMENT_ASSIGNMENT_STRATEGY,
+ ROUND_ROBIN_SEGMENT_ASSIGNMENT_STRATEGY,
+ REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY
+ );
+ // Strategies that place a segment by its partition, the only ones the
partition column applies to
+ public static final Set<String> PARTITION_BASED_STRATEGIES = Set.of(
+ REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY
+ );
}
public static class BuiltInVirtualColumn {
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
index f61f433a4da..8c2950880bc 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
@@ -85,7 +85,7 @@ public class TableConfigBuilder {
@Deprecated
private String _segmentPushType = DEFAULT_SEGMENT_PUSH_TYPE;
private String _peerSegmentDownloadScheme;
- @Deprecated
+ @Deprecated(since = "1.3.0", forRemoval = true)
private ReplicaGroupStrategyConfig _replicaGroupStrategyConfig;
private CompletionConfig _completionConfig;
private String _crypterClassName;
@@ -242,6 +242,7 @@ public class TableConfigBuilder {
return this;
}
+ @Deprecated(since = "1.3.0", forRemoval = true)
public TableConfigBuilder
setReplicaGroupStrategyConfig(ReplicaGroupStrategyConfig
replicaGroupStrategyConfig) {
_replicaGroupStrategyConfig = replicaGroupStrategyConfig;
return this;
diff --git
a/pinot-tools/src/main/resources/examples/batch/colocated/userAttributes/userAttributes_offline_table_config.json
b/pinot-tools/src/main/resources/examples/batch/colocated/userAttributes/userAttributes_offline_table_config.json
index 6a5aecb8688..52655c24e49 100644
---
a/pinot-tools/src/main/resources/examples/batch/colocated/userAttributes/userAttributes_offline_table_config.json
+++
b/pinot-tools/src/main/resources/examples/batch/colocated/userAttributes/userAttributes_offline_table_config.json
@@ -32,8 +32,7 @@
"numReplicaGroups": 2,
"numInstancesPerReplicaGroup": 2,
"numPartitions": 2,
- "numInstancesPerPartition": 1,
- "partitionColumn": "userUUID"
+ "numInstancesPerPartition": 1
}
}
},
diff --git
a/pinot-tools/src/main/resources/examples/batch/colocated/userFactEvents/userFactEvents_offline_table_config.json
b/pinot-tools/src/main/resources/examples/batch/colocated/userFactEvents/userFactEvents_offline_table_config.json
index 99430f74485..33c51d507b9 100644
---
a/pinot-tools/src/main/resources/examples/batch/colocated/userFactEvents/userFactEvents_offline_table_config.json
+++
b/pinot-tools/src/main/resources/examples/batch/colocated/userFactEvents/userFactEvents_offline_table_config.json
@@ -3,11 +3,7 @@
"tableType": "OFFLINE",
"segmentsConfig": {
"segmentPushType": "APPEND",
- "replication": 2,
- "replicaGroupStrategyConfig": {
- "partitionColumn": "userUUID",
- "numInstancesPerPartition": 2
- }
+ "replication": 2
},
"tenants": {
"broker": "DefaultTenant",
diff --git
a/pinot-tools/src/main/resources/examples/batch/colocated/userGroups/userGroups_offline_table_config.json
b/pinot-tools/src/main/resources/examples/batch/colocated/userGroups/userGroups_offline_table_config.json
index 9f7d3fd0456..3386556dca6 100644
---
a/pinot-tools/src/main/resources/examples/batch/colocated/userGroups/userGroups_offline_table_config.json
+++
b/pinot-tools/src/main/resources/examples/batch/colocated/userGroups/userGroups_offline_table_config.json
@@ -3,11 +3,7 @@
"tableType": "OFFLINE",
"segmentsConfig": {
"segmentPushType": "APPEND",
- "replication": "2",
- "replicaGroupStrategyConfig": {
- "partitionColumn": "userUUID",
- "numInstancesPerPartition": 2
- }
+ "replication": "2"
},
"tenants": {
"broker": "DefaultTenant",
diff --git
a/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
index a5589b96703..70bee25d52d 100644
---
a/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
@@ -39,7 +39,6 @@
"replicaGroupPartitionConfig": {
"replicaGroupBased": true,
"numReplicaGroups": 1,
- "partitionColumn": "rsvp_id",
"numPartitions": 2,
"numInstancesPerPartition": 1
}
diff --git
a/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
index 393b5fa7e47..863f7fab41f 100644
---
a/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
@@ -27,7 +27,6 @@
"replicaGroupPartitionConfig": {
"replicaGroupBased": true,
"numReplicaGroups": 1,
- "partitionColumn": "event_id",
"numPartitions": 2,
"numInstancesPerPartition": 1
}
diff --git
a/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
index 1cc953a95d0..70c2f9f85aa 100644
---
a/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
@@ -27,7 +27,6 @@
"replicaGroupPartitionConfig": {
"replicaGroupBased": true,
"numReplicaGroups": 1,
- "partitionColumn": "event_id",
"numPartitions": 2,
"numInstancesPerPartition": 1
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
