somandal commented on code in PR #15930:
URL: https://github.com/apache/pinot/pull/15930#discussion_r2121614980
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -913,6 +914,25 @@ static void
validatePartitionedReplicaGroupInstance(TableConfig tableConfig) {
}
}
+ @VisibleForTesting
+ static void validateInstanceAssignmentConfigs(TableConfig tableConfig) {
Review Comment:
thanks for adding this!
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/ImplicitRealtimeTablePartitionSelector.java:
##########
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.helix.core.assignment.instance;
+
+import com.google.common.annotations.VisibleForTesting;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.assignment.InstancePartitions;
+import org.apache.pinot.spi.config.table.TableConfig;
+import
org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
+import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
+import org.apache.pinot.spi.stream.StreamMetadataProvider;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
+
+
+/**
+ * Variation of {@link InstanceReplicaGroupPartitionSelector} that uses the
number of partitions from the stream
+ * to determine the number of partitions in each replica group.
+ */
+public class ImplicitRealtimeTablePartitionSelector extends
InstanceReplicaGroupPartitionSelector {
+ private final TableConfig _tableConfig;
+ private final int _numPartitions;
+
+ public ImplicitRealtimeTablePartitionSelector(TableConfig tableConfig,
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig, String
tableNameWithType,
+ @Nullable InstancePartitions existingInstancePartitions, boolean
minimizeDataMovement) {
+ this(tableConfig, replicaGroupPartitionConfig, tableNameWithType,
existingInstancePartitions, minimizeDataMovement,
+
StreamConsumerFactoryProvider.create(IngestionConfigUtils.getFirstStreamConfig(tableConfig))
+ .createStreamMetadataProvider(
+ ImplicitRealtimeTablePartitionSelector.class.getSimpleName() +
"-" + tableNameWithType));
+ }
+
+ @VisibleForTesting
+ ImplicitRealtimeTablePartitionSelector(TableConfig tableConfig,
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig, String
tableNameWithType,
+ @Nullable InstancePartitions existingInstancePartitions, boolean
minimizeDataMovement,
+ StreamMetadataProvider streamMetadataProvider) {
+ super(replicaGroupPartitionConfig, tableNameWithType,
existingInstancePartitions, minimizeDataMovement);
+ _tableConfig = tableConfig;
+ _numPartitions = getStreamNumPartitions(streamMetadataProvider);
+ }
+
+ private int getStreamNumPartitions(StreamMetadataProvider
streamMetadataProvider) {
+ try (streamMetadataProvider) {
+ return streamMetadataProvider.fetchPartitionCount(10_000L);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to retrieve partition info for table:
" + _tableNameWithType, e);
+ }
+ }
+
+ @Override
+ protected int getNumPartitions() {
+ return _numPartitions;
+ }
+
+ @Override
+ protected int getNumInstancesPerPartition(int numInstancesPerReplicaGroup) {
+ if (_tableConfig.isUpsertEnabled()) {
+ return 1; // For upsert enabled tables, we enforce one instance per
partition
Review Comment:
I don't see any harm in adding this, but If I remember correctly,
`RealtimeSegmentAssignment` and `StrictRealtimeSegmentAssignment` always assign
to the first instance of the RG in `assignConsumingSegment` already. So this
may not be needed (but please double check)
We don't allow any REALTIME tables to have more than 1 instance per
partition AFAIK
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -913,6 +914,25 @@ static void
validatePartitionedReplicaGroupInstance(TableConfig tableConfig) {
}
}
+ @VisibleForTesting
+ static void validateInstanceAssignmentConfigs(TableConfig tableConfig) {
+ if (tableConfig.getInstanceAssignmentConfigMap() == null) {
+ return;
+ }
+ for (InstanceAssignmentConfig instanceAssignmentConfig :
tableConfig.getInstanceAssignmentConfigMap().values()) {
+ if (instanceAssignmentConfig.getPartitionSelector()
+ ==
InstanceAssignmentConfig.PartitionSelector.IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR)
{
+ Preconditions.checkState(tableConfig.getTableType() ==
TableType.REALTIME,
+ "IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR can only be used for
REALTIME tables");
+
Preconditions.checkState(instanceAssignmentConfig.getReplicaGroupPartitionConfig().isReplicaGroupBased(),
Review Comment:
For REALTIME, we kind of implicitly do something similar to ReplicaGroup
based in terms of assignment, where we round-robin, but essentially every
partition is assigned to the same set of hosts. Is this limitation necessary?
What's the limitation to implicitly allow this for non-RG based REALTIME tables?
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TableRebalanceIntegrationTest.java:
##########
@@ -88,6 +88,78 @@ private String getRebalanceUrl(RebalanceConfig
rebalanceConfig, TableType tableT
+ "?type=" + tableType.toString() + "&" +
getQueryString(rebalanceConfig);
}
+ @Test
+ public void testImplicitRealtimeTableInstanceAssignment() throws Exception {
Review Comment:
Great test to have! Would it also make sense to add a test to
`TableRebalancerClusterStatelessTest` and actually validate the segment
assignment before and after rebalance for this instance selector? (similar to
your instance selector test, but with an actual rebalance in the mix)?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/ImplicitRealtimeTablePartitionSelector.java:
##########
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.helix.core.assignment.instance;
+
+import com.google.common.annotations.VisibleForTesting;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.assignment.InstancePartitions;
+import org.apache.pinot.spi.config.table.TableConfig;
+import
org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
+import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
+import org.apache.pinot.spi.stream.StreamMetadataProvider;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
+
+
+/**
+ * Variation of {@link InstanceReplicaGroupPartitionSelector} that uses the
number of partitions from the stream
+ * to determine the number of partitions in each replica group.
+ */
+public class ImplicitRealtimeTablePartitionSelector extends
InstanceReplicaGroupPartitionSelector {
+ private final TableConfig _tableConfig;
+ private final int _numPartitions;
+
+ public ImplicitRealtimeTablePartitionSelector(TableConfig tableConfig,
+ InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig, String
tableNameWithType,
+ @Nullable InstancePartitions existingInstancePartitions, boolean
minimizeDataMovement) {
+ this(tableConfig, replicaGroupPartitionConfig, tableNameWithType,
existingInstancePartitions, minimizeDataMovement,
+
StreamConsumerFactoryProvider.create(IngestionConfigUtils.getFirstStreamConfig(tableConfig))
Review Comment:
I am less familiar with the stream side, but is it correct to always pick
this up based on the `getFirstStreamConfig`? I see a lot of comments indicating
that it should ideally be picked up based on the partition.
Just for my understanding, when can there exist multiple StreamConfigs, and
what happens if they have a different number of partitions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
