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 f57c2285ec2 Validate segment assignment for strict replica-group
routing (#20006)
f57c2285ec2 is described below
commit f57c2285ec259041114eca09d11f6a882d9dcb99
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Oct 23 11:03:19 2026 -0700
Validate segment assignment for strict replica-group routing (#20006)
---
.../api/resources/TableConfigValidationUtils.java | 45 +++-
.../resources/TableConfigValidationUtilsTest.java | 76 ++++++
.../segment/local/utils/TableConfigUtils.java | 124 ++++++++-
.../segment/local/utils/TableConfigUtilsTest.java | 292 +++++++++++++++++++--
4 files changed, 500 insertions(+), 37 deletions(-)
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 4c7f7a6fd5b..19692575c01 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
@@ -19,7 +19,9 @@
package org.apache.pinot.controller.api.resources;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import java.util.List;
+import java.util.Map;
import javax.annotation.Nullable;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.assignment.InstanceAssignmentConfigUtils;
@@ -44,7 +46,7 @@ import org.apache.pinot.spi.utils.builder.TableNameBuilder;
///
/// This lives in `pinot-controller` (not `pinot-segment-local`'s
`TableConfigUtils`)
/// because validation requires controller-level dependencies like
[PinotHelixResourceManager],
-/// [ControllerConf], [PinotTaskManager], and [TableRebalancer].
+/// [ControllerConf], [PinotTaskManager], and [InstanceAssignmentUtils].
public final class TableConfigValidationUtils {
private TableConfigValidationUtils() {
@@ -121,9 +123,10 @@ 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, 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.
+ /// satisfied by the current instances is rejected before the table config
is persisted, and validates them: the
+ /// OFFLINE segments must be placed by partition when the OFFLINE instance
partitions have multiple partitions, and
+ /// the instance partitions must fit strict replica-group routing. 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) {
@@ -135,6 +138,7 @@ public final class TableConfigValidationUtils {
}
List<InstanceConfig> instanceConfigs =
resourceManager.getAllHelixInstanceConfigs();
InstancePartitions offlineInstancePartitions = null;
+ InstancePartitions consumingInstancePartitions = null;
try {
for (InstancePartitionsType instancePartitionsType :
instancePartitionsTypes) {
InstancePartitions instancePartitions;
@@ -147,6 +151,8 @@ public final class TableConfigValidationUtils {
}
if (instancePartitionsType == InstancePartitionsType.OFFLINE) {
offlineInstancePartitions = instancePartitions;
+ } else if (instancePartitionsType == InstancePartitionsType.CONSUMING)
{
+ consumingInstancePartitions = instancePartitions;
}
}
} catch (Exception e) {
@@ -157,5 +163,36 @@ public final class TableConfigValidationUtils {
if (offlineInstancePartitions != null &&
offlineInstancePartitions.getNumPartitions() > 1) {
TableConfigUtils.validatePartitionedSegmentAssignment(tableConfig,
InstancePartitionsType.OFFLINE.name());
}
+ validateStrictReplicaGroupInstancePartitions(tableConfig,
offlineInstancePartitions, consumingInstancePartitions);
+ }
+
+ /// Validates the computed or fetched instance partitions of a table using
strict replica-group routing, which is the
+ /// only place where pre-configured instance partitions can be checked; the
strict replica-group routing check in
+ /// [TableConfigUtils#validate] covers the instance assignment config. The
OFFLINE instance partitions must place each
+ /// partition on exactly one instance per replica-group. CONSUMING segments
are placed by their stream partition, so
+ /// the CONSUMING instance partitions need the same layout only when they
have multiple partitions. COMPLETED segments
+ /// are never relocated for such tables.
+ @VisibleForTesting
+ static void validateStrictReplicaGroupInstancePartitions(TableConfig
tableConfig,
+ @Nullable InstancePartitions offlineInstancePartitions,
+ @Nullable InstancePartitions consumingInstancePartitions) {
+ if (!TableConfigUtils.requiresStrictReplicaGroupAssignment(tableConfig)) {
+ return;
+ }
+ if (offlineInstancePartitions != null) {
+ validateOneInstancePerPartition(tableConfig, offlineInstancePartitions);
+ }
+ if (consumingInstancePartitions != null &&
consumingInstancePartitions.getNumPartitions() > 1) {
+ validateOneInstancePerPartition(tableConfig,
consumingInstancePartitions);
+ }
+ }
+
+ private static void validateOneInstancePerPartition(TableConfig tableConfig,
InstancePartitions instancePartitions) {
+ for (Map.Entry<String, List<String>> entry :
instancePartitions.getPartitionToInstancesMap().entrySet()) {
+ Preconditions.checkState(entry.getValue().size() == 1,
+ "Table: %s using strict replica-group routing must have exactly one
instance per partition in each "
+ + "replica-group, found instances: %s for
partition_replicaGroup: %s in instance partitions: %s",
+ tableConfig.getTableName(), entry.getValue(), entry.getKey(),
instancePartitions.getInstancePartitionsName());
+ }
}
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtilsTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtilsTest.java
index c59b1aab42a..0a1a798572f 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtilsTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtilsTest.java
@@ -18,13 +18,18 @@
*/
package org.apache.pinot.controller.api.resources;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
+import org.apache.pinot.common.assignment.InstancePartitions;
+import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableCustomConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.expectThrows;
@@ -73,4 +78,75 @@ public class TableConfigValidationUtilsTest {
TableConfig tableConfig = tableConfigWithCustomConfig("key", "plainValue");
TableConfigValidationUtils.validateEnvironmentVariables(tableConfig);
}
+
+ private static InstancePartitions createInstancePartitions(String name, int
numPartitions, int numReplicaGroups,
+ int numInstancesPerPartition) {
+ InstancePartitions instancePartitions = new InstancePartitions(name);
+ for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
+ for (int replicaGroupId = 0; replicaGroupId < numReplicaGroups;
replicaGroupId++) {
+ List<String> instances = new ArrayList<>(numInstancesPerPartition);
+ for (int i = 0; i < numInstancesPerPartition; i++) {
+ instances.add("Server_" + partitionId + "_" + replicaGroupId + "_" +
i);
+ }
+ instancePartitions.setInstances(partitionId, replicaGroupId,
instances);
+ }
+ }
+ return instancePartitions;
+ }
+
+ @Test
+ public void testValidateStrictReplicaGroupInstancePartitions() {
+ RoutingConfig strictRoutingConfig =
+ new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false);
+ InstancePartitions wideInstancePartitions =
createInstancePartitions("table_OFFLINE", 4, 2, 2);
+
+ // Tables without strict replica-group routing are not validated
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName("table").build();
+
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(tableConfig,
wideInstancePartitions, null);
+
+ // Dimension tables host every segment on every server
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName("table")
+ .setIsDimTable(true)
+ .setRoutingConfig(strictRoutingConfig)
+ .build();
+
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(tableConfig,
wideInstancePartitions, null);
+
+ // OFFLINE instance partitions with one instance per partition in each
replica-group
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName("table")
+ .setRoutingConfig(strictRoutingConfig)
+ .build();
+
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(tableConfig,
+ createInstancePartitions("table_OFFLINE", 4, 2, 1), null);
+
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(tableConfig,
+ createInstancePartitions("table_OFFLINE", 1, 2, 1), null);
+
+ // OFFLINE instance partitions with multiple instances per partition
+ TableConfig offlineTableConfig = tableConfig;
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(offlineTableConfig,
+ wideInstancePartitions, null));
+ assertEquals(e.getMessage(),
+ "Table: table_OFFLINE using strict replica-group routing must have
exactly one instance per partition in each "
+ + "replica-group, found instances: [Server_0_0_0, Server_0_0_1]
for partition_replicaGroup: 0_0 in "
+ + "instance partitions: table_OFFLINE");
+
+ // CONSUMING instance partitions with a single partition, or one instance
per partition
+ tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName("table")
+ .setRoutingConfig(strictRoutingConfig)
+ .build();
+
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(tableConfig,
null,
+ createInstancePartitions("table_CONSUMING", 1, 2, 3));
+
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(tableConfig,
null,
+ createInstancePartitions("table_CONSUMING", 4, 2, 1));
+
+ // CONSUMING instance partitions with multiple partitions and multiple
instances per partition
+ TableConfig realtimeTableConfig = tableConfig;
+ e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigValidationUtils.validateStrictReplicaGroupInstancePartitions(realtimeTableConfig,
null,
+ createInstancePartitions("table_CONSUMING", 4, 2, 2)));
+ assertEquals(e.getMessage(),
+ "Table: table_REALTIME using strict replica-group routing must have
exactly one instance per partition in each "
+ + "replica-group, found instances: [Server_0_0_0, Server_0_0_1]
for partition_replicaGroup: 0_0 in "
+ + "instance partitions: table_CONSUMING");
+ }
}
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 b933a997ddb..eb07e0c0640 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
@@ -39,6 +39,8 @@ import javax.annotation.Nullable;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.common.assignment.InstanceAssignmentConfigUtils;
+import org.apache.pinot.common.assignment.InstancePartitionsUtils;
import org.apache.pinot.common.evaluator.FunctionEvaluatorFactory;
import org.apache.pinot.common.function.FunctionInfo;
import org.apache.pinot.common.function.FunctionRegistry;
@@ -225,6 +227,7 @@ public final class TableConfigUtils {
validateInstancePartitionsTypeMapConfig(tableConfig);
validatePartitionedReplicaGroupInstance(tableConfig);
validateInstanceAssignmentConfigs(tableConfig);
+ validateStrictReplicaGroupRouting(tableConfig);
if (!skipTypes.contains(ValidationType.UPSERT)) {
validateUpsertAndDedupConfig(tableConfig, schema, existingTableConfig);
@@ -1057,12 +1060,15 @@ public final class TableConfigUtils {
/// Validates the upsert- and dedup-related configuration. Returns early
when neither is enabled; otherwise checks
/// the shared requirements (upsert and dedup are mutually exclusive,
primary keys exist and are single-valued,
- /// strict replica-group routing, no COMPLETED instance partitions, valid
tenant tag override for REALTIME, OFFLINE
- /// restrictions) and then, per the enabled mode:
+ /// strict replica-group routing, no tenant tag override for REALTIME,
OFFLINE restrictions) and then, per the enabled
+ /// mode:
/// - Upsert: no multi-tier / star-tree, comparison / delete-record /
out-of-order columns are valid, deleted-keys
/// compaction consistency, post-partial-upsert transforms,
consistency-mode constraints, and delegates to
/// [#validateTTLForUpsertConfig] and [#validatePartialUpsertStrategies];
rejects MD5 when disabled.
/// - Dedup: delegates to [#validateTTLForDedupConfig]; rejects MD5 when
disabled.
+ ///
+ /// The segment assignment requirements shared with strict replica-group
routing (one instance per partition, no
+ /// COMPLETED segment relocation) are validated by
[#validateStrictReplicaGroupRouting].
@VisibleForTesting
static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema
schema) {
validateUpsertAndDedupConfig(tableConfig, schema, null);
@@ -1102,10 +1108,6 @@ public final class TableConfigUtils {
tableConfig.getTenantConfig().getTagOverrideConfig().getRealtimeConsuming() ==
null
&&
tableConfig.getTenantConfig().getTagOverrideConfig().getRealtimeCompleted() ==
null),
"Invalid tenant tag override used for Upsert/Dedup table");
- Map<String, InstanceAssignmentConfig> instanceAssignmentConfigMap =
tableConfig.getInstanceAssignmentConfigMap();
- Preconditions.checkState(instanceAssignmentConfigMap == null ||
!instanceAssignmentConfigMap.containsKey(
- InstancePartitionsType.COMPLETED.name()),
- "COMPLETED instance partitions can't be configured for upsert /
dedup tables");
} else {
Preconditions.checkState(!dedupEnabled,
"Dedup is not supported for OFFLINE table. Only upsert is supported
for OFFLINE table");
@@ -1569,6 +1571,116 @@ public final class TableConfigUtils {
partitionColumn);
}
+ /// Validates that a table using strict replica-group routing has a segment
assignment that places all segments of
+ /// a partition on exactly one instance per replica-group. Strict
replica-group routing routes all segments with the
+ /// same instances in the ideal state to the same instance, which only
provides partition level consistency when the
+ /// segment assignment keeps each partition on one instance per
replica-group. Upsert and dedup tables rely on the
+ /// same property and are validated regardless of the configured instance
selector.
+ ///
+ /// - OFFLINE: the OFFLINE instance assignment must be replica-group based
with one instance per partition, and the
+ /// OFFLINE segment assignment strategy, if configured, must be partition
based (`replicagroup` or
+ /// `roundrobinreplicagroup`), since the other strategies ignore the
partition. The partition column of a
+ /// partitioned instance assignment is validated by
[#validatePartitionedSegmentAssignment]. The instance
+ /// assignment can also come from the deprecated
`replicaGroupStrategyConfig`, which takes effect only together with
+ /// the `replicagroup` strategy.
+ /// - REALTIME: CONSUMING segments are placed by their stream partition,
which keeps each partition on one instance
+ /// per replica-group when the CONSUMING instance partitions have a single
partition or one instance per
+ /// partition. COMPLETED segments must not be relocated, which would split
each partition between its CONSUMING
+ /// and COMPLETED instances.
+ ///
+ /// The layout of pre-configured instance partitions lives in ZK and is not
visible here, so the controller validates
+ /// it on the fetched instance partitions instead. Dimension tables are
exempt because every server hosts every
+ /// segment.
+ @VisibleForTesting
+ static void validateStrictReplicaGroupRouting(TableConfig tableConfig) {
+ if (!requiresStrictReplicaGroupAssignment(tableConfig)) {
+ return;
+ }
+ if (tableConfig.getTableType() == TableType.OFFLINE) {
+ validateStrictReplicaGroupOfflineInstanceAssignment(tableConfig);
+ } else {
+ validateStrictReplicaGroupRealtimeInstanceAssignment(tableConfig);
+ }
+ }
+
+ /// Returns `true` if the segment assignment of the table must place all
segments of a partition on exactly one
+ /// instance per replica-group: the table uses strict replica-group routing,
or is an upsert/dedup table, whose
+ /// allowed instance selectors all rely on this property. Dimension tables
are excluded because every server hosts
+ /// every segment.
+ public static boolean requiresStrictReplicaGroupAssignment(TableConfig
tableConfig) {
+ if (tableConfig.isDimTable()) {
+ return false;
+ }
+ if (tableConfig.isUpsertEnabled() || tableConfig.isDedupEnabled()) {
+ return true;
+ }
+ RoutingConfig routingConfig = tableConfig.getRoutingConfig();
+ String instanceSelectorType = routingConfig != null ?
routingConfig.getInstanceSelectorType() : null;
+ return
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE.equalsIgnoreCase(instanceSelectorType);
+ }
+
+ /// Validates the OFFLINE instance assignment config for strict
replica-group routing.
+ private static void
validateStrictReplicaGroupOfflineInstanceAssignment(TableConfig tableConfig) {
+ SegmentAssignmentConfig segmentAssignmentConfig =
+
MapUtils.emptyIfNull(tableConfig.getSegmentAssignmentConfigMap()).get(InstancePartitionsType.OFFLINE.name());
+ String assignmentStrategy =
+ segmentAssignmentConfig != null ?
segmentAssignmentConfig.getAssignmentStrategy() : null;
+ Preconditions.checkState(assignmentStrategy == null
+ ||
AssignmentStrategy.PARTITION_BASED_STRATEGIES.contains(assignmentStrategy.toLowerCase(Locale.ROOT)),
+ "Table using strict replica-group routing can only use '%s' or '%s'
OFFLINE segment assignment strategy, "
+ + "found: %s",
AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+
AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
assignmentStrategy);
+ if
(InstancePartitionsUtils.hasPreConfiguredInstancePartitions(tableConfig,
InstancePartitionsType.OFFLINE)) {
+ return;
+ }
+ // The instance assignment config comes from instanceAssignmentConfigMap,
or from the deprecated
+ // replicaGroupStrategyConfig when the replica-group segment assignment
strategy is configured
+ Preconditions.checkState(
+ InstanceAssignmentConfigUtils.allowInstanceAssignment(tableConfig,
InstancePartitionsType.OFFLINE),
+ "Table using strict replica-group routing must have OFFLINE instance
assignment config "
+ + "(instanceAssignmentConfigMap) to ensure all segments of a
partition are assigned to the same server");
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
+ InstanceAssignmentConfigUtils.getInstanceAssignmentConfig(tableConfig,
InstancePartitionsType.OFFLINE)
+ .getReplicaGroupPartitionConfig();
+ Preconditions.checkState(replicaGroupPartitionConfig.isReplicaGroupBased(),
+ "Table using strict replica-group routing must use replica-group based
OFFLINE instance assignment to ensure "
+ + "all segments of a partition are assigned to the same server");
+
Preconditions.checkState(getNumInstancesPerPartition(replicaGroupPartitionConfig)
<= 1,
+ "Table using strict replica-group routing must have one instance per
partition (numInstancesPerPartition, or "
+ + "numInstancesPerReplicaGroup when it is not set) in OFFLINE
instance assignment config to ensure all "
+ + "segments of a partition are assigned to the same server");
+ }
+
+ /// Validates the REALTIME instance assignment config for strict
replica-group routing.
+ private static void
validateStrictReplicaGroupRealtimeInstanceAssignment(TableConfig tableConfig) {
+
Preconditions.checkState(!InstanceAssignmentConfigUtils.shouldRelocateCompletedSegments(tableConfig),
+ "Table using strict replica-group routing must not relocate COMPLETED
segments (tagOverrideConfig with a "
+ + "different realtimeCompleted tag,
instanceAssignmentConfigMap.COMPLETED or "
+ + "instancePartitionsMap.COMPLETED) to ensure all segments of a
partition are assigned to the same server");
+ InstanceAssignmentConfig consumingInstanceAssignmentConfig =
+
MapUtils.emptyIfNull(tableConfig.getInstanceAssignmentConfigMap()).get(InstancePartitionsType.CONSUMING.name());
+ if (consumingInstanceAssignmentConfig != null) {
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
+ consumingInstanceAssignmentConfig.getReplicaGroupPartitionConfig();
+
Preconditions.checkState(!replicaGroupPartitionConfig.isReplicaGroupBased()
+ || replicaGroupPartitionConfig.getNumPartitions() <= 1
+ || getNumInstancesPerPartition(replicaGroupPartitionConfig) <= 1,
+ "Table using strict replica-group routing must have one instance per
partition (numInstancesPerPartition, or "
+ + "numInstancesPerReplicaGroup when it is not set) in CONSUMING
instance assignment config with multiple "
+ + "partitions to ensure all segments of a partition are assigned
to the same server");
+ }
+ }
+
+ /// Returns the number of instances per partition the instance assignment
config resolves to: the configured
+ /// `numInstancesPerPartition`, or `numInstancesPerReplicaGroup` when it is
not set, since each partition then gets
+ /// every instance of its replica-group. Returns 0 when neither is set, in
which case the count depends on the
+ /// instances in the pool and is validated on the computed instance
partitions by the controller.
+ private static int
getNumInstancesPerPartition(InstanceReplicaGroupPartitionConfig
replicaGroupPartitionConfig) {
+ int numInstancesPerPartition =
replicaGroupPartitionConfig.getNumInstancesPerPartition();
+ return numInstancesPerPartition > 0 ? numInstancesPerPartition
+ : replicaGroupPartitionConfig.getNumInstancesPerReplicaGroup();
+ }
+
/// Validates the partial upsert-related configurations:
/// - Null handling must be enabled
/// - Merger cannot be applied to private key columns
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 ed33075921d..ad89745e377 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
@@ -3012,33 +3012,6 @@ public class TableConfigUtilsTest {
assertEquals(e.getMessage(), "Invalid tenant tag override used for
Upsert/Dedup table");
}
- // Instance assignment config with just CONSUMING instance partitions
configured should be allowed
- InstanceAssignmentConfig instanceAssignmentConfig =
mock(InstanceAssignmentConfig.class);
- tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setTimeColumnName(TIME_COLUMN)
- .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
- .setStreamConfigs(getStreamConfigs())
- .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
-
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
instanceAssignmentConfig))
- .build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
-
- // Instance assignment config with CONSUMING and COMPLETED instance
partitions configured should not be allowed
- tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setTimeColumnName(TIME_COLUMN)
- .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
- .setStreamConfigs(getStreamConfigs())
- .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
-
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
instanceAssignmentConfig,
- InstancePartitionsType.COMPLETED.name(), instanceAssignmentConfig))
- .build();
- try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
- fail("Instance assignment config with COMPLETED instance partitions must
not be allowed with upsert");
- } catch (IllegalStateException e) {
- assertEquals(e.getMessage(), "COMPLETED instance partitions can't be
configured for upsert / dedup tables");
- }
-
// empty tag override with upsert should pass
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
@@ -3853,6 +3826,271 @@ public class TableConfigUtilsTest {
}
@SuppressWarnings("removal")
+ private static InstanceAssignmentConfig
createReplicaGroupInstanceAssignmentConfig(boolean replicaGroupBased,
+ int numInstancesPerReplicaGroup, int numPartitions, int
numInstancesPerPartition) {
+ return new InstanceAssignmentConfig(new
InstanceTagPoolConfig("DefaultTenant_OFFLINE", false, 0, null), null,
+ new InstanceReplicaGroupPartitionConfig(replicaGroupBased, 0, 2,
numInstancesPerReplicaGroup, numPartitions,
+ numInstancesPerPartition, false, null), null, false);
+ }
+
+ private static TableConfigBuilder
strictReplicaGroupTableConfigBuilder(TableType tableType) {
+ return new
TableConfigBuilder(tableType).setTableName(TABLE_NAME).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG);
+ }
+
+ private static void assertStrictReplicaGroupRoutingFails(TableConfig
tableConfig, String expectedMessage) {
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig));
+ assertEquals(e.getMessage(), expectedMessage);
+ }
+
+ @SuppressWarnings("removal")
+ @Test
+ public void testValidateStrictReplicaGroupRouting() {
+ String offlineKey = InstancePartitionsType.OFFLINE.name();
+ SegmentPartitionConfig segmentPartitionConfig =
+ new SegmentPartitionConfig(Map.of("myCol", new
ColumnPartitionConfig("murmur", 4)));
+ Map<String, InstanceAssignmentConfig> offlineInstanceAssignmentConfigMap =
+ Map.of(offlineKey, createReplicaGroupInstanceAssignmentConfig(true, 0,
4, 1));
+ SegmentAssignmentConfig replicaGroupSegmentAssignmentConfig =
+ new
SegmentAssignmentConfig(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
null);
+ SegmentAssignmentConfig balancedSegmentAssignmentConfig =
+ new
SegmentAssignmentConfig(AssignmentStrategy.BALANCE_NUM_SEGMENT_ASSIGNMENT_STRATEGY,
null);
+
+ // Tables without strict replica-group routing are not validated
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setRoutingConfig(new RoutingConfig(null, null,
RoutingConfig.REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // Dimension tables host every segment on every server
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setIsDimTable(true)
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // OFFLINE table with partitioned replica-group instance assignment and
one instance per partition
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setInstanceAssignmentConfigMap(offlineInstanceAssignmentConfigMap)
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // OFFLINE table with the replica-group segment assignment strategy
configured explicitly
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setInstanceAssignmentConfigMap(offlineInstanceAssignmentConfigMap)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
replicaGroupSegmentAssignmentConfig))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // OFFLINE table with the round-robin replica-group strategy, which is the
same as replica-group with one instance
+ // per partition
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setInstanceAssignmentConfigMap(offlineInstanceAssignmentConfigMap)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
+ new
SegmentAssignmentConfig(AssignmentStrategy.ROUND_ROBIN_REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY,
+ null)))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // Without numInstancesPerPartition, each partition gets
numInstancesPerReplicaGroup instances, and the layout is
+ // left to the controller when neither is set
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createReplicaGroupInstanceAssignmentConfig(true, 1, 0, 0)))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createReplicaGroupInstanceAssignmentConfig(true, 0, 0, 0)))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createReplicaGroupInstanceAssignmentConfig(true, 2, 0, 0)))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must have one instance per
partition (numInstancesPerPartition, or "
+ + "numInstancesPerReplicaGroup when it is not set) in OFFLINE
instance assignment config to ensure all "
+ + "segments of a partition are assigned to the same server");
+
+ // OFFLINE table with a single partition keeps every segment on one
instance per replica-group
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createReplicaGroupInstanceAssignmentConfig(true, 0, 0, 1)))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // OFFLINE table without OFFLINE instance assignment config
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must have OFFLINE instance
assignment config "
+ + "(instanceAssignmentConfigMap) to ensure all segments of a
partition are assigned to the same server");
+
+ // OFFLINE table with the deprecated replicaGroupStrategyConfig, which
takes effect only together with the
+ // replica-group segment assignment strategy
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setReplicaGroupStrategyConfig(new ReplicaGroupStrategyConfig("myCol",
1))
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
replicaGroupSegmentAssignmentConfig))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setReplicaGroupStrategyConfig(new ReplicaGroupStrategyConfig(null, 1))
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
replicaGroupSegmentAssignmentConfig))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setReplicaGroupStrategyConfig(new ReplicaGroupStrategyConfig("myCol",
2))
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
replicaGroupSegmentAssignmentConfig))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must have one instance per
partition (numInstancesPerPartition, or "
+ + "numInstancesPerReplicaGroup when it is not set) in OFFLINE
instance assignment config to ensure all "
+ + "segments of a partition are assigned to the same server");
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setReplicaGroupStrategyConfig(new ReplicaGroupStrategyConfig("myCol",
1))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must have OFFLINE instance
assignment config "
+ + "(instanceAssignmentConfigMap) to ensure all segments of a
partition are assigned to the same server");
+
+ // OFFLINE table with pre-configured instance partitions leaves the layout
check to the controller, but still
+ // cannot use a segment assignment strategy that ignores the partition
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setInstancePartitionsMap(Map.of(InstancePartitionsType.OFFLINE,
"test_OFFLINE"))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setInstancePartitionsMap(Map.of(InstancePartitionsType.OFFLINE,
"test_OFFLINE"))
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
balancedSegmentAssignmentConfig))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing can only use 'replicagroup'
or 'roundrobinreplicagroup' OFFLINE "
+ + "segment assignment strategy, found: balanced");
+
+ // OFFLINE table with non replica-group based instance assignment
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createReplicaGroupInstanceAssignmentConfig(false, 0, 4, 1)))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must use replica-group based
OFFLINE instance assignment to ensure "
+ + "all segments of a partition are assigned to the same server");
+
+ // OFFLINE table with multiple instances per partition
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setInstanceAssignmentConfigMap(Map.of(offlineKey,
createReplicaGroupInstanceAssignmentConfig(true, 0, 4, 2)))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must have one instance per
partition (numInstancesPerPartition, or "
+ + "numInstancesPerReplicaGroup when it is not set) in OFFLINE
instance assignment config to ensure all "
+ + "segments of a partition are assigned to the same server");
+
+ // OFFLINE table with a segment assignment strategy that ignores the
partition
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.OFFLINE)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setInstanceAssignmentConfigMap(offlineInstanceAssignmentConfigMap)
+ .setSegmentAssignmentConfigMap(Map.of(offlineKey,
balancedSegmentAssignmentConfig))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing can only use 'replicagroup'
or 'roundrobinreplicagroup' OFFLINE "
+ + "segment assignment strategy, found: balanced");
+
+ // REALTIME table without COMPLETED segment relocation places every
partition on one instance per replica-group
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
+ createReplicaGroupInstanceAssignmentConfig(true, 0, 0, 0)))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // REALTIME table with partitioned CONSUMING instance assignment needs one
instance per partition
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
+ createReplicaGroupInstanceAssignmentConfig(true, 0, 4, 1)))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
+ createReplicaGroupInstanceAssignmentConfig(true, 1, 4, 0)))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
+ createReplicaGroupInstanceAssignmentConfig(true, 0, 4, 2)))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must have one instance per
partition (numInstancesPerPartition, or "
+ + "numInstancesPerReplicaGroup when it is not set) in CONSUMING
instance assignment config with multiple "
+ + "partitions to ensure all segments of a partition are assigned
to the same server");
+
+ // REALTIME table with tag override that does not relocate COMPLETED
segments
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+ .setTagOverrideConfig(new TagOverrideConfig(null, null))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+ .setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME",
"T1_REALTIME"))
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+
+ // REALTIME table relocating COMPLETED segments through tag override,
instance assignment config or pre-configured
+ // instance partitions
+ String relocationMessage =
+ "Table using strict replica-group routing must not relocate COMPLETED
segments (tagOverrideConfig with a "
+ + "different realtimeCompleted tag,
instanceAssignmentConfigMap.COMPLETED or "
+ + "instancePartitionsMap.COMPLETED) to ensure all segments of a
partition are assigned to the same server";
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+ .setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME",
"T2_REALTIME"))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig, relocationMessage);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.COMPLETED.name(),
+ createReplicaGroupInstanceAssignmentConfig(true, 0, 4, 1)))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig, relocationMessage);
+ tableConfig = strictReplicaGroupTableConfigBuilder(TableType.REALTIME)
+ .setInstancePartitionsMap(Map.of(InstancePartitionsType.COMPLETED,
"test_COMPLETED"))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig, relocationMessage);
+
+ // Upsert and dedup tables are validated regardless of the instance
selector
+ RoutingConfig multiStageReplicaGroupRoutingConfig =
+ new RoutingConfig(null, null,
RoutingConfig.MULTI_STAGE_REPLICA_GROUP_SELECTOR_TYPE, false);
+ tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
+ .setRoutingConfig(multiStageReplicaGroupRoutingConfig)
+ .setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME",
"T2_REALTIME"))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig, relocationMessage);
+ tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setDedupConfig(new DedupConfig())
+ .setRoutingConfig(multiStageReplicaGroupRoutingConfig)
+ .setInstancePartitionsMap(Map.of(InstancePartitionsType.COMPLETED,
"test_COMPLETED"))
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig, relocationMessage);
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
+ .setRoutingConfig(multiStageReplicaGroupRoutingConfig)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .build();
+ assertStrictReplicaGroupRoutingFails(tableConfig,
+ "Table using strict replica-group routing must have OFFLINE instance
assignment config "
+ + "(instanceAssignmentConfigMap) to ensure all segments of a
partition are assigned to the same server");
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
+ .setRoutingConfig(multiStageReplicaGroupRoutingConfig)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setInstanceAssignmentConfigMap(offlineInstanceAssignmentConfigMap)
+ .build();
+ TableConfigUtils.validateStrictReplicaGroupRouting(tableConfig);
+ }
+
@Test
public void testValidatePartitionedReplicaGroupInstance() {
String partitionColumn = "testPartitionCol";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
