Jackie-Jiang commented on code in PR #19958:
URL: https://github.com/apache/pinot/pull/19958#discussion_r4330992099
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -1448,6 +1515,37 @@ static void
validateInstanceAssignmentConfigs(TableConfig tableConfig) {
"numInstancesPerPartition must be 1 when using
IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR");
}
// TODO: Add more validations for other partition selectors here
+
+ // Partition based placement of OFFLINE segments reads the partition of
a segment from its metadata
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
+ instanceAssignmentConfig.getReplicaGroupPartitionConfig();
+ if (offlineSegmentKeys.contains(instancePartitionsType) &&
replicaGroupPartitionConfig.isReplicaGroupBased()
+ && replicaGroupPartitionConfig.getNumPartitions() > 1) {
+ validateSegmentAssignmentPartitionColumn(tableConfig);
+ }
+ }
+ }
+
+ /// Validates that the partition column for partition based placement of
OFFLINE segments can be resolved, which
+ /// is required when the instance partitions placing them, the OFFLINE ones
or those of a tier of an OFFLINE table,
+ /// have multiple partitions. The partition column configured for segment
assignment must be a column of the
+ /// segment partition config; 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.
+ public static void validateSegmentAssignmentPartitionColumn(TableConfig
tableConfig) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
getColumnPartitionMap(tableConfig);
+ Preconditions.checkState(!columnPartitionMap.isEmpty(),
Review Comment:
Kept, by design: an explicitly partitioned instance assignment now requires
the segments to be placed by partition, and hash placement defeats the
partitioning the table configured. The Compatibility section now lists every
case that fails on the next update, including a missing
`segmentPartitionConfig`.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/SegmentAssignmentConfig.java:
##########
@@ -21,21 +21,39 @@
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) {
Review Comment:
Only a config written after the upgrade can hit this: every pre-existing
entry has a strategy, since the factory required one. Rolling a controller back
after adopting a new config field is not something we guarantee, so the
strategy stays optional.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -431,26 +442,70 @@ private static void validateValidationConfig(TableConfig
tableConfig, Schema sch
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, the same across all entries, and the same as the
deprecated partition columns of the instance
+ /// assignment configs and `replicaGroupStrategyConfig` when those are still
configured.
+ @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();
+ String partitionColumn = null;
+ 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();
+ 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(SEGMENT_ASSIGNMENT_STRATEGIES.contains(assignmentStrategy),
Review Comment:
Kept. Such a string never selected a strategy; it ran as `balanced` by
accident. The rejection is disclosed in the Compatibility section, and the PR
carries the `backward-incompat` and `release-notes` labels.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -1448,6 +1515,37 @@ static void
validateInstanceAssignmentConfigs(TableConfig tableConfig) {
"numInstancesPerPartition must be 1 when using
IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR");
}
// TODO: Add more validations for other partition selectors here
+
+ // Partition based placement of OFFLINE segments reads the partition of
a segment from its metadata
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig =
+ instanceAssignmentConfig.getReplicaGroupPartitionConfig();
+ if (offlineSegmentKeys.contains(instancePartitionsType) &&
replicaGroupPartitionConfig.isReplicaGroupBased()
+ && replicaGroupPartitionConfig.getNumPartitions() > 1) {
+ validateSegmentAssignmentPartitionColumn(tableConfig);
+ }
+ }
+ }
+
+ /// Validates that the partition column for partition based placement of
OFFLINE segments can be resolved, which
+ /// is required when the instance partitions placing them, the OFFLINE ones
or those of a tier of an OFFLINE table,
+ /// have multiple partitions. The partition column configured for segment
assignment must be a column of the
+ /// segment partition config; 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.
+ public static void validateSegmentAssignmentPartitionColumn(TableConfig
tableConfig) {
+ 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) {
+ Preconditions.checkState(columnPartitionMap.containsKey(partitionColumn),
Review Comment:
Kept, for the same reason as the first comment: a column absent from
`segmentPartitionConfig` meant hash placement for a table that explicitly
configured partitions. Disclosed in the Compatibility section as one of the
rejected cases.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -431,26 +442,70 @@ private static void validateValidationConfig(TableConfig
tableConfig, Schema sch
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, the same across all entries, and the same as the
deprecated partition columns of the instance
+ /// assignment configs and `replicaGroupStrategyConfig` when those are still
configured.
+ @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();
+ String partitionColumn = null;
+ 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();
+ 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(SEGMENT_ASSIGNMENT_STRATEGIES.contains(assignmentStrategy),
+ "Invalid segment assignment strategy: %s for: %s of table: %s",
assignmentStrategy, key, tableName);
+ }
+ partitionBased =
PARTITION_BASED_SEGMENT_ASSIGNMENT_STRATEGIES.contains(assignmentStrategy);
+ }
+ String entryPartitionColumn =
segmentAssignmentConfig.getPartitionColumn();
+ if (StringUtils.isNotEmpty(entryPartitionColumn)) {
+ Preconditions.checkState(partitionBased,
+ "Partition column: %s does not apply to the segment assignment
strategy for: %s of table: %s",
+ entryPartitionColumn, key, tableName);
+ Preconditions.checkState(partitionColumn == null ||
partitionColumn.equals(entryPartitionColumn),
+ "Segment assignment configs of table: %s must configure the same
partition column, found: %s and %s",
+ tableName, partitionColumn, entryPartitionColumn);
+ partitionColumn = entryPartitionColumn;
+ }
+ }
+ if (partitionColumn != null) {
+
Preconditions.checkState(getColumnPartitionMap(tableConfig).containsKey(partitionColumn),
Review Comment:
A realtime segment's partition comes from its name, for LLC and uploaded
realtime segment names alike, so `replicaGroupStrategyConfig.partitionColumn`
on a realtime table was never read. The migration is to drop it, which is what
the removed fixture and test setup did.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java:
##########
@@ -233,13 +234,15 @@ public void setReplicasPerPartition(String
replicasPerPartition) {
_replicasPerPartition = replicasPerPartition;
}
- /// @deprecated Use
[org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig] instead.
+ /// @deprecated Configure the instance assignment in
`instanceAssignmentConfigMap` with
Review Comment:
The replacement for that path is `instanceAssignmentConfigMap`, as the
description says; the segment assignment column is not an instance assignment
input. The precondition message now points there instead of naming the
deprecated config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
