xiangfu0 commented on code in PR #19958:
URL: https://github.com/apache/pinot/pull/19958#discussion_r4329944969
##########
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:
**[CRITICAL] Upgrade break: existing partitioned OFFLINE tables without
`segmentPartitionConfig` can no longer be updated.**
Pre-PR, an OFFLINE table with replica-group instance assignment and
`numPartitions > 1` but no `segmentPartitionConfig` validated fine and worked:
`ReplicaGroupSegmentAssignmentStrategy` resolves each segment via
`SegmentUtils.getSegmentPartitionIdOrDefault`, which falls back to a hash of
the segment name when no partition metadata/column exists. This new check
(triggered from `validateInstanceAssignmentConfigs` and again from
`TableConfigValidationUtils.validateInstanceAssignment` for
computed/fetched/pre-configured OFFLINE instance partitions) now throws on
**every** table config update — even unrelated ones like changing retention —
until the operator adds a `segmentPartitionConfig` they never needed. The PR
description only discloses the "multiple columns without a configured column"
case; the no-`segmentPartitionConfig`-at-all case is an undisclosed break for
working production tables.
##########
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:
**[CRITICAL] Mixed-version hazard: a strategy-less `SegmentAssignmentConfig`
NPEs on pre-PR controllers.**
This PR makes `segmentAssignmentStrategy` optional and blesses `{"OFFLINE":
{"partitionColumn": "..."}}` (validation accepts it;
`SegmentValidationUtilsTest`/`OfflineUpsertTableTest` model it). But the null
guard on `getAssignmentStrategy()` only ships in this PR's
`SegmentAssignmentStrategyFactory` change — a controller still running the
previous release executes
`segmentAssignmentConfig.getAssignmentStrategy().toLowerCase()` and throws NPE
at segment assignment time. During a rolling upgrade (or after a rollback,
since the config persists in ZK), every segment upload/rebalance for such a
table fails on the old controller. Per repo guidance, mixed-version
broker/server/controller compatibility must be preserved; consider gating
strategy-less entries for one release or documenting the ordering constraint.
##########
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:
**[MAJOR] Strategy whitelist bricks config updates for tables carrying
legacy long-form strategy names.**
Pre-PR, validation never checked the strategy for non-dim tables and
`SegmentAssignmentStrategyFactory` silently ran unknown strings as balanced.
Long-form names demonstrably exist in the wild — this repo's own
`TableConfigSerDeUtilsTest` used `"ReplicaGroupSegmentAssignmentStrategy"`
until this PR — and they lowercase to strings outside `{balanced, roundrobin,
replicagroup, roundrobinreplicagroup}`. After upgrade, any config update of
such a table fails with `Invalid segment assignment strategy`, and "fixing" the
name to `replicagroup` silently flips live assignment behavior from balanced to
replica-group, requiring an unexpected rebalance. The PR acknowledges rejecting
unknown strategies, but blocking all updates of previously-accepted persisted
configs deserves at least a release-note/migration callout — or a warn-only
grace period.
##########
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:
**[MAJOR] Legacy-only partition columns are now validated against
`segmentPartitionConfig`, contradicting "still accepted and honored".**
`getPartitionColumn` falls back to the deprecated
`replicaGroupPartitionConfig.partitionColumn` /
`replicaGroupStrategyConfig.partitionColumn`, and this check then requires that
legacy column to exist in `segmentPartitionConfig`. Pre-PR, a legacy column
absent from `segmentPartitionConfig` merely meant the segment ZK-metadata
lookup missed and assignment fell back to name-hash — never a validation error.
Example: an existing table with
`replicaGroupStrategyConfig.partitionColumn="memberId"` (legacy) and
`segmentPartitionConfig` on `"userId"` (added later for query pruning)
validates today, but post-upgrade every config update fails with `Partition
column: memberId ... not found in segment partition config`, even though the
user never touched the new field.
##########
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:
**[MAJOR] No migration path for REALTIME tables: the new `partitionColumn`
requires `segmentPartitionConfig`, which the deprecated column never did.**
A realtime (e.g. upsert) table with
`replicaGroupStrategyConfig.partitionColumn="event_id"` and no
`segmentPartitionConfig` works today — uploaded/committed segments carry
partition metadata in ZK independently of the table-level config (exactly the
shape this PR removed from `simpleMeetup_realtime_table_config.json` and
`StrictRealtimeSegmentAssignmentTest`). Migrating to the advertised replacement
`{"COMPLETED": {"segmentAssignmentStrategy": "replicagroup", "partitionColumn":
"event_id"}}` is rejected here because `getColumnPartitionMap` is empty. The
deprecated field is marked `forRemoval` yet has no valid replacement for such
tables.
##########
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:
**[MAJOR] The advertised new-style config cannot actually replace
`replicaGroupStrategyConfig` on the derived instance-assignment path.**
`InstanceAssignmentConfigUtils.getInstanceAssignmentConfig` (pinot-common,
~line 118) still hard-requires the deprecated config: a table configured the
new way — `segmentAssignmentConfigMap: {OFFLINE: {segmentAssignmentStrategy:
"replicagroup", partitionColumn: "col"}}` with no `instanceAssignmentConfigMap`
and no `replicaGroupStrategyConfig` — passes `allowInstanceAssignment()`
(replica-group strategy implies OFFLINE instance assignment) and then dies with
`Failed to find the replica-group strategy config` during instance-partition
computation at table creation. The new `partitionColumn` is never consulted on
this path, so deleting the `forRemoval` config per the deprecation guidance
breaks table creation. (Pre-existing precondition, but this PR's deprecation
story makes it a dead end.)
##########
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()
Review Comment:
**[MAJOR] This trigger ignores an explicitly configured non-partition-based
strategy.**
An OFFLINE table with `segmentAssignmentConfigMap.OFFLINE =
{segmentAssignmentStrategy: "balanced"}` (or `roundrobin`) plus replica-group
instance assignment with `numPartitions > 1` builds
`BalancedNumSegmentAssignmentStrategy`, which never reads a partition column —
yet this condition only checks `isReplicaGroupBased() && getNumPartitions() >
1` and rejects the config with a misleading "must have segment partition
config" error. `validateSegmentAssignmentConfigs` already computes exactly the
`partitionBased` property of the configured strategy but it is not consulted
here (nor in the parallel controller-side check in
`TableConfigValidationUtils`).
--
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]
