You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2019/08/01 20:37:39 UTC
[incubator-pinot] branch instace_assignment_config created (now
d9c4d82)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a change to branch instace_assignment_config
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
at d9c4d82 [Instance Assignment] Add configs for instance assignment
This branch includes the following new commits:
new d9c4d82 [Instance Assignment] Add configs for instance assignment
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/01: [Instance Assignment] Add configs for
instance assignment
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch instace_assignment_config
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit d9c4d8210eef52797055afc18240a3d3a43fe415
Author: Jackie (Xiaotian) Jiang <xa...@linkedin.com>
AuthorDate: Thu Aug 1 13:25:51 2019 -0700
[Instance Assignment] Add configs for instance assignment
- Also enhance JsonUtils to support TypeReference for deserialization
---
.../apache/pinot/common/config/TableConfig.java | 109 +++++++++++++++-----
.../apache/pinot/common/config/TagNameUtils.java | 15 +++
.../common/config/instance/ConstraintConfig.java | 44 ++++++++
.../config/instance/FinalSelectionConfig.java | 113 +++++++++++++++++++++
.../config/instance/InitialSelectionConfig.java | 86 ++++++++++++++++
.../config/instance/InstanceAssignmentConfig.java | 75 ++++++++++++++
...InstanceAssignmentConfigMapChildKeyHandler.java | 28 +++++
.../instance/InstanceAssignmentConfigUtils.java | 86 ++++++++++++++++
.../apache/pinot/common/utils/CommonConstants.java | 5 +
.../org/apache/pinot/common/utils/JsonUtils.java | 30 ++++--
.../pinot/common/config/TableConfigTest.java | 63 +++++++++++-
.../core/assignment/InstancePartitionsUtils.java | 25 +----
.../segment/SegmentAssignmentStrategyFactory.java | 7 +-
...eReplicaGroupSegmentAssignmentStrategyTest.java | 15 +--
...eReplicaGroupSegmentAssignmentStrategyTest.java | 9 +-
.../pinot/core/data/readers/JSONRecordReader.java | 23 ++---
.../tests/LLCRealtimeClusterIntegrationTest.java | 19 +---
17 files changed, 650 insertions(+), 102 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java
index 502a7a6..6c1ba88 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java
@@ -19,6 +19,7 @@
package org.apache.pinot.common.config;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Preconditions;
@@ -29,14 +30,15 @@ import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.helix.ZNRecord;
-import org.apache.pinot.common.data.StarTreeIndexSpec;
+import org.apache.pinot.common.config.instance.InstanceAssignmentConfig;
+import org.apache.pinot.common.config.instance.InstanceAssignmentConfigMapChildKeyHandler;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
import org.apache.pinot.common.utils.EqualityUtils;
+import org.apache.pinot.common.utils.InstancePartitionsType;
import org.apache.pinot.common.utils.JsonUtils;
-import org.apache.pinot.startree.hll.HllConfig;
-@SuppressWarnings("unused")
+@SuppressWarnings({"Duplicates", "unused"})
@ConfigDoc(value = "Configuration for a table", mandatory = true)
@ConfigKey("table")
public class TableConfig {
@@ -49,15 +51,16 @@ public class TableConfig {
public static final String QUOTA_CONFIG_KEY = "quota";
public static final String TASK_CONFIG_KEY = "task";
public static final String ROUTING_CONFIG_KEY = "routing";
+ public static final String INSTANCE_ASSIGNMENT_CONFIG_MAP_KEY = "instanceAssignmentConfigMap";
private static final String FIELD_MISSING_MESSAGE_TEMPLATE = "Mandatory field '%s' is missing";
@ConfigKey("name")
- @ConfigDoc(value = "The name for the table.", mandatory = true, exampleValue = "myTable")
+ @ConfigDoc(value = "The name for the table (with type suffix)", mandatory = true, exampleValue = "myTable_OFFLINE")
private String _tableName;
@ConfigKey("type")
- @ConfigDoc(value = "The type of the table, either realtime or offline", mandatory = true)
+ @ConfigDoc(value = "The type of the table (OFFLINE|REALTIME)", mandatory = true)
private TableType _tableType;
@NestedConfig
@@ -82,6 +85,10 @@ public class TableConfig {
@NestedConfig
private RoutingConfig _routingConfig;
+ @ConfigKey(INSTANCE_ASSIGNMENT_CONFIG_MAP_KEY)
+ @UseChildKeyHandler(InstanceAssignmentConfigMapChildKeyHandler.class)
+ private Map<InstancePartitionsType, InstanceAssignmentConfig> _instanceAssignmentConfigMap;
+
/**
* NOTE: DO NOT use this constructor, use builder instead. This constructor is for deserializer only.
*/
@@ -93,7 +100,8 @@ public class TableConfig {
private TableConfig(String tableName, TableType tableType, SegmentsValidationAndRetentionConfig validationConfig,
TenantConfig tenantConfig, IndexingConfig indexingConfig, TableCustomConfig customConfig,
- @Nullable QuotaConfig quotaConfig, @Nullable TableTaskConfig taskConfig, @Nullable RoutingConfig routingConfig) {
+ @Nullable QuotaConfig quotaConfig, @Nullable TableTaskConfig taskConfig, @Nullable RoutingConfig routingConfig,
+ @Nullable Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap) {
_tableName = TableNameBuilder.forType(tableType).tableNameWithType(tableName);
_tableType = tableType;
_validationConfig = validationConfig;
@@ -103,6 +111,7 @@ public class TableConfig {
_quotaConfig = quotaConfig;
_taskConfig = taskConfig;
_routingConfig = routingConfig;
+ _instanceAssignmentConfigMap = instanceAssignmentConfigMap;
}
public static TableConfig fromJsonString(String jsonString)
@@ -146,14 +155,19 @@ public class TableConfig {
RoutingConfig routingConfig = extractChildConfig(jsonConfig, ROUTING_CONFIG_KEY, RoutingConfig.class);
+ Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap =
+ extractChildConfig(jsonConfig, INSTANCE_ASSIGNMENT_CONFIG_MAP_KEY,
+ new TypeReference<Map<InstancePartitionsType, InstanceAssignmentConfig>>() {
+ });
+
return new TableConfig(tableName, tableType, validationConfig, tenantConfig, indexingConfig, customConfig,
- quotaConfig, taskConfig, routingConfig);
+ quotaConfig, taskConfig, routingConfig, instanceAssignmentConfigMap);
}
/**
* Extracts the child config from the table config. Returns {@code null} if child config does not exist.
* <p>
- * NOTE: for historical reason, we support two kinds of nested config values: normal json and serialized json string
+ * NOTE: For historical reason, we support two kinds of nested config values: normal json and serialized json string
*/
@Nullable
private static <T> T extractChildConfig(JsonNode jsonConfig, String childConfigKey, Class<T> childConfigClass)
@@ -169,6 +183,26 @@ public class TableConfig {
}
}
+ /**
+ * Extracts the child config from the table config. Returns {@code null} if child config does not exist.
+ * <p>
+ * NOTE: For historical reason, we support two kinds of nested config values: normal json and serialized json string
+ */
+ @Nullable
+ private static <T> T extractChildConfig(JsonNode jsonConfig, String childConfigKey,
+ TypeReference<T> childConfigTypeReference)
+ throws IOException {
+ JsonNode childConfigNode = jsonConfig.get(childConfigKey);
+ if (childConfigNode == null || childConfigNode.isNull()) {
+ return null;
+ }
+ if (childConfigNode.isObject()) {
+ return JsonUtils.jsonNodeToObject(childConfigNode, childConfigTypeReference);
+ } else {
+ return JsonUtils.stringToObject(childConfigNode.asText(), childConfigTypeReference);
+ }
+ }
+
public ObjectNode toJsonConfig() {
validate();
@@ -192,6 +226,9 @@ public class TableConfig {
if (_routingConfig != null) {
jsonConfig.set(ROUTING_CONFIG_KEY, JsonUtils.objectToJsonNode(_routingConfig));
}
+ if (_instanceAssignmentConfigMap != null) {
+ jsonConfig.set(INSTANCE_ASSIGNMENT_CONFIG_MAP_KEY, JsonUtils.objectToJsonNode(_instanceAssignmentConfigMap));
+ }
return jsonConfig;
}
@@ -250,8 +287,16 @@ public class TableConfig {
routingConfig = JsonUtils.stringToObject(routingConfigString, RoutingConfig.class);
}
+ Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap = null;
+ String instanceAssignmentConfigMapString = simpleFields.get(INSTANCE_ASSIGNMENT_CONFIG_MAP_KEY);
+ if (instanceAssignmentConfigMapString != null) {
+ instanceAssignmentConfigMap = JsonUtils.stringToObject(instanceAssignmentConfigMapString,
+ new TypeReference<Map<InstancePartitionsType, InstanceAssignmentConfig>>() {
+ });
+ }
+
return new TableConfig(tableName, tableType, validationConfig, tenantConfig, indexingConfig, customConfig,
- quotaConfig, taskConfig, routingConfig);
+ quotaConfig, taskConfig, routingConfig, instanceAssignmentConfigMap);
}
public ZNRecord toZNRecord()
@@ -278,6 +323,9 @@ public class TableConfig {
if (_routingConfig != null) {
simpleFields.put(ROUTING_CONFIG_KEY, JsonUtils.objectToString(_routingConfig));
}
+ if (_instanceAssignmentConfigMap != null) {
+ simpleFields.put(INSTANCE_ASSIGNMENT_CONFIG_MAP_KEY, JsonUtils.objectToString(_instanceAssignmentConfigMap));
+ }
ZNRecord znRecord = new ZNRecord(_tableName);
znRecord.setSimpleFields(simpleFields);
@@ -372,6 +420,16 @@ public class TableConfig {
_routingConfig = routingConfig;
}
+ @Nullable
+ public Map<InstancePartitionsType, InstanceAssignmentConfig> getInstanceAssignmentConfigMap() {
+ return _instanceAssignmentConfigMap;
+ }
+
+ public void setInstanceAssignmentConfigMap(
+ Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap) {
+ _instanceAssignmentConfigMap = instanceAssignmentConfigMap;
+ }
+
@Override
public String toString() {
try {
@@ -451,14 +509,13 @@ public class TableConfig {
private List<String> _bloomFilterColumns;
private Map<String, String> _streamConfigs;
private String _streamPartitionAssignmentStrategy = DEFAULT_STREAM_PARTITION_ASSIGNMENT_STRATEGY;
+ private SegmentPartitionConfig _segmentPartitionConfig;
private TableCustomConfig _customConfig;
private QuotaConfig _quotaConfig;
private TableTaskConfig _taskConfig;
private RoutingConfig _routingConfig;
- private HllConfig _hllConfig;
- private SegmentPartitionConfig _segmentPartitionConfig;
- private StarTreeIndexSpec _starTreeIndexSpec;
+ private Map<InstancePartitionsType, InstanceAssignmentConfig> _instanceAssignmentConfigMap;
public Builder(TableType tableType) {
_tableType = tableType;
@@ -564,11 +621,6 @@ public class TableConfig {
return this;
}
- public Builder setBloomFilterColumns(List<String> bloomFilterColumns) {
- _bloomFilterColumns = bloomFilterColumns;
- return this;
- }
-
public Builder setNoDictionaryColumns(List<String> noDictionaryColumns) {
_noDictionaryColumns = noDictionaryColumns;
return this;
@@ -579,8 +631,8 @@ public class TableConfig {
return this;
}
- public Builder setStreamPartitionAssignmentStrategy(String streamPartitionAssignmentStrategy) {
- _streamPartitionAssignmentStrategy = streamPartitionAssignmentStrategy;
+ public Builder setBloomFilterColumns(List<String> bloomFilterColumns) {
+ _bloomFilterColumns = bloomFilterColumns;
return this;
}
@@ -590,6 +642,16 @@ public class TableConfig {
return this;
}
+ public Builder setStreamPartitionAssignmentStrategy(String streamPartitionAssignmentStrategy) {
+ _streamPartitionAssignmentStrategy = streamPartitionAssignmentStrategy;
+ return this;
+ }
+
+ public Builder setSegmentPartitionConfig(SegmentPartitionConfig segmentPartitionConfig) {
+ _segmentPartitionConfig = segmentPartitionConfig;
+ return this;
+ }
+
public Builder setCustomConfig(TableCustomConfig customConfig) {
_customConfig = customConfig;
return this;
@@ -610,8 +672,9 @@ public class TableConfig {
return this;
}
- public Builder setSegmentPartitionConfig(SegmentPartitionConfig segmentPartitionConfig) {
- _segmentPartitionConfig = segmentPartitionConfig;
+ public Builder setInstanceAssignmentConfigMap(
+ Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap) {
+ _instanceAssignmentConfigMap = instanceAssignmentConfigMap;
return this;
}
@@ -647,8 +710,8 @@ public class TableConfig {
indexingConfig.setInvertedIndexColumns(_invertedIndexColumns);
indexingConfig.setNoDictionaryColumns(_noDictionaryColumns);
indexingConfig.setOnHeapDictionaryColumns(_onHeapDictionaryColumns);
- indexingConfig.setStreamConfigs(_streamConfigs);
indexingConfig.setBloomFilterColumns(_bloomFilterColumns);
+ indexingConfig.setStreamConfigs(_streamConfigs);
StreamConsumptionConfig streamConsumptionConfig = new StreamConsumptionConfig();
streamConsumptionConfig.setStreamPartitionAssignmentStrategy(_streamPartitionAssignmentStrategy);
indexingConfig.setStreamConsumptionConfig(streamConsumptionConfig);
@@ -660,7 +723,7 @@ public class TableConfig {
}
return new TableConfig(_tableName, _tableType, validationConfig, tenantConfig, indexingConfig, _customConfig,
- _quotaConfig, _taskConfig, _routingConfig);
+ _quotaConfig, _taskConfig, _routingConfig, _instanceAssignmentConfigMap);
}
}
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/TagNameUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/config/TagNameUtils.java
index 1702f53..6e0f0e1 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/config/TagNameUtils.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/TagNameUtils.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.common.config;
+import org.apache.pinot.common.utils.InstancePartitionsType;
import org.apache.pinot.common.utils.ServerType;
import org.apache.pinot.common.utils.TenantRole;
@@ -101,4 +102,18 @@ public class TagNameUtils {
}
return tag;
}
+
+ public static String getServerTagFromTableConfigAndInstancePartitionsType(TableConfig tableConfig,
+ InstancePartitionsType instancePartitionsType) {
+ switch (instancePartitionsType) {
+ case OFFLINE:
+ return new OfflineTagConfig(tableConfig).getOfflineServerTag();
+ case CONSUMING:
+ return new RealtimeTagConfig(tableConfig).getConsumingServerTag();
+ case COMPLETED:
+ return new RealtimeTagConfig(tableConfig).getCompletedServerTag();
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/instance/ConstraintConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/ConstraintConfig.java
new file mode 100644
index 0000000..455c853
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/ConstraintConfig.java
@@ -0,0 +1,44 @@
+/**
+ * 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.common.config.instance;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import org.apache.pinot.common.config.ConfigDoc;
+import org.apache.pinot.common.config.ConfigKey;
+
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ConstraintConfig {
+
+ @ConfigKey("constraints")
+ @ConfigDoc(value = "Name of the constraints to be applied", mandatory = true)
+ private List<String> _constraints;
+
+ @JsonProperty
+ public List<String> getConstraints() {
+ return _constraints;
+ }
+
+ @JsonProperty
+ public void setConstraints(List<String> constraints) {
+ _constraints = constraints;
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/instance/FinalSelectionConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/FinalSelectionConfig.java
new file mode 100644
index 0000000..84186f1
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/FinalSelectionConfig.java
@@ -0,0 +1,113 @@
+/**
+ * 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.common.config.instance;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.pinot.common.config.ConfigDoc;
+import org.apache.pinot.common.config.ConfigKey;
+
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class FinalSelectionConfig {
+
+ @ConfigKey("replicaGroupBased")
+ @ConfigDoc("Whether to use replica-group based selection, false by default")
+ private boolean _replicaGroupBased;
+
+ @ConfigKey("numServers")
+ @ConfigDoc("Number of servers to select for non-replica-group based selection, select all servers if not specified")
+ private int _numServers;
+
+ @ConfigKey("numReplicas")
+ @ConfigDoc("Number of replicas (replica-groups) for replica-group based selection")
+ private int _numReplicas;
+
+ @ConfigKey("numServersPerReplica")
+ @ConfigDoc("Number of servers per replica (replica-group) for replica-group based selection, select all servers if not specified")
+ private int _numServersPerReplica;
+
+ @ConfigKey("numPartitions")
+ @ConfigDoc("Number of partitions for replica-group based selection, do not partition the replica-group (1 partition) if not specified")
+ private int _numPartitions;
+
+ @ConfigKey("numServersPerPartition")
+ @ConfigDoc("Number of servers per partition (within a replica) for replica-group based selection, select all servers if not specified")
+ private int _numServersPerPartition;
+
+ @JsonProperty
+ public boolean isReplicaGroupBased() {
+ return _replicaGroupBased;
+ }
+
+ @JsonProperty
+ public void setReplicaGroupBased(boolean replicaGroupBased) {
+ _replicaGroupBased = replicaGroupBased;
+ }
+
+ @JsonProperty
+ public int getNumServers() {
+ return _numServers;
+ }
+
+ @JsonProperty
+ public void setNumServers(int numServers) {
+ _numServers = numServers;
+ }
+
+ @JsonProperty
+ public int getNumReplicas() {
+ return _numReplicas;
+ }
+
+ @JsonProperty
+ public void setNumReplicas(int numReplicas) {
+ _numReplicas = numReplicas;
+ }
+
+ @JsonProperty
+ public int getNumServersPerReplica() {
+ return _numServersPerReplica;
+ }
+
+ @JsonProperty
+ public void setNumServersPerReplica(int numServersPerReplica) {
+ _numServersPerReplica = numServersPerReplica;
+ }
+
+ @JsonProperty
+ public int getNumPartitions() {
+ return _numPartitions;
+ }
+
+ @JsonProperty
+ public void setNumPartitions(int numPartitions) {
+ _numPartitions = numPartitions;
+ }
+
+ @JsonProperty
+ public int getNumServersPerPartition() {
+ return _numServersPerPartition;
+ }
+
+ @JsonProperty
+ public void setNumServersPerPartition(int numServersPerPartition) {
+ _numServersPerPartition = numServersPerPartition;
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InitialSelectionConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InitialSelectionConfig.java
new file mode 100644
index 0000000..446c8bb
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InitialSelectionConfig.java
@@ -0,0 +1,86 @@
+/**
+ * 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.common.config.instance;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import org.apache.pinot.common.config.ConfigDoc;
+import org.apache.pinot.common.config.ConfigKey;
+
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class InitialSelectionConfig {
+
+ @ConfigKey("tag")
+ @ConfigDoc(value = "Tag of the instances to select", mandatory = true)
+ private String _tag;
+
+ @ConfigKey("poolBased")
+ @ConfigDoc("Whether to use pool based selection, false by default")
+ private boolean _poolBased;
+
+ @ConfigKey("numPools")
+ @ConfigDoc("Number of pools to select for pool based selection, contradict to pools")
+ private int _numPools;
+
+ @ConfigKey("pools")
+ @ConfigDoc("Pools to select for pool based selection, contradict to numPools")
+ private List<Integer> _pools;
+
+ @JsonProperty
+ public String getTag() {
+ return _tag;
+ }
+
+ @JsonProperty
+ public void setTag(String tag) {
+ _tag = tag;
+ }
+
+ @JsonProperty
+ public boolean isPoolBased() {
+ return _poolBased;
+ }
+
+ @JsonProperty
+ public void setPoolBased(boolean poolBased) {
+ _poolBased = poolBased;
+ }
+
+ @JsonProperty
+ public int getNumPools() {
+ return _numPools;
+ }
+
+ @JsonProperty
+ public void setNumPools(int numPools) {
+ _numPools = numPools;
+ }
+
+ @JsonProperty
+ public List<Integer> getPools() {
+ return _pools;
+ }
+
+ @JsonProperty
+ public void setPools(List<Integer> pools) {
+ _pools = pools;
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfig.java
new file mode 100644
index 0000000..256adca
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfig.java
@@ -0,0 +1,75 @@
+/**
+ * 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.common.config.instance;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.pinot.common.config.ConfigDoc;
+import org.apache.pinot.common.config.ConfigKey;
+import org.apache.pinot.common.config.NestedConfig;
+
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class InstanceAssignmentConfig {
+
+ @ConfigKey("initialSelectionConfig")
+ @ConfigDoc(value = "Configuration for the initial selection of the instance assignment, which picks instances based on the configured tenant and pool", mandatory = true)
+ @NestedConfig
+ private InitialSelectionConfig _initialSelectionConfig;
+
+ @ConfigKey("constraintConfig")
+ @ConfigDoc("Configuration for the constraints of the instance assignment, which filters out unqualified instances and sorts instances for picking priority")
+ @NestedConfig
+ private ConstraintConfig _constraintConfig;
+
+ @ConfigKey("finalSelectionConfig")
+ @ConfigDoc(value = "Configuration for the final selection of the instance assignment, which picks final instances for each replica and partition", mandatory = true)
+ @NestedConfig
+ private FinalSelectionConfig _finalSelectionConfig;
+
+ @JsonProperty
+ public InitialSelectionConfig getInitialSelectionConfig() {
+ return _initialSelectionConfig;
+ }
+
+ @JsonProperty
+ public void setInitialSelectionConfig(InitialSelectionConfig initialSelectionConfig) {
+ _initialSelectionConfig = initialSelectionConfig;
+ }
+
+ @JsonProperty
+ public ConstraintConfig getConstraintConfig() {
+ return _constraintConfig;
+ }
+
+ @JsonProperty
+ public void setConstraintConfig(ConstraintConfig constraintConfig) {
+ _constraintConfig = constraintConfig;
+ }
+
+ @JsonProperty
+ public FinalSelectionConfig getFinalSelectionConfig() {
+ return _finalSelectionConfig;
+ }
+
+ @JsonProperty
+ public void setFinalSelectionConfig(FinalSelectionConfig finalSelectionConfig) {
+ _finalSelectionConfig = finalSelectionConfig;
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfigMapChildKeyHandler.java b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfigMapChildKeyHandler.java
new file mode 100644
index 0000000..094a351
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfigMapChildKeyHandler.java
@@ -0,0 +1,28 @@
+/**
+ * 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.common.config.instance;
+
+import org.apache.pinot.common.config.TypedMapChildKeyHandler;
+
+
+public class InstanceAssignmentConfigMapChildKeyHandler extends TypedMapChildKeyHandler<InstanceAssignmentConfig> {
+ public InstanceAssignmentConfigMapChildKeyHandler() {
+ super(InstanceAssignmentConfig.class);
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfigUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfigUtils.java
new file mode 100644
index 0000000..0827dd3
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/instance/InstanceAssignmentConfigUtils.java
@@ -0,0 +1,86 @@
+/**
+ * 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.common.config.instance;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import org.apache.pinot.common.config.ReplicaGroupStrategyConfig;
+import org.apache.pinot.common.config.SegmentsValidationAndRetentionConfig;
+import org.apache.pinot.common.config.TableConfig;
+import org.apache.pinot.common.config.TagNameUtils;
+import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
+import org.apache.pinot.common.utils.CommonConstants.Segment.AssignmentStrategy;
+import org.apache.pinot.common.utils.InstancePartitionsType;
+
+
+public class InstanceAssignmentConfigUtils {
+ private InstanceAssignmentConfigUtils() {
+ }
+
+ public static InstanceAssignmentConfig getInstanceAssignmentConfig(TableConfig tableConfig,
+ InstancePartitionsType instancePartitionsType) {
+ // Use the instance assignment config from the table config if it exists
+ Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap =
+ tableConfig.getInstanceAssignmentConfigMap();
+ if (instanceAssignmentConfigMap != null && instanceAssignmentConfigMap.containsKey(instancePartitionsType)) {
+ return instanceAssignmentConfigMap.get(instancePartitionsType);
+ }
+
+ // Generate default instance assignment config if it does not exist
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig();
+
+ InitialSelectionConfig initialSelectionConfig = new InitialSelectionConfig();
+ String serverTag =
+ TagNameUtils.getServerTagFromTableConfigAndInstancePartitionsType(tableConfig, instancePartitionsType);
+ initialSelectionConfig.setTag(serverTag);
+ instanceAssignmentConfig.setInitialSelectionConfig(initialSelectionConfig);
+
+ FinalSelectionConfig finalSelectionConfig = new FinalSelectionConfig();
+ SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
+ if (AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY
+ .equalsIgnoreCase(validationConfig.getSegmentAssignmentStrategy())) {
+ finalSelectionConfig.setReplicaGroupBased(true);
+ if (tableConfig.getTableType() == TableType.OFFLINE) {
+ finalSelectionConfig.setNumReplicas(validationConfig.getReplicationNumber());
+ ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
+ tableConfig.getValidationConfig().getReplicaGroupStrategyConfig();
+ Preconditions
+ .checkState(replicaGroupStrategyConfig != null, "Failed to find the replica-group strategy config");
+ String partitionColumn = replicaGroupStrategyConfig.getPartitionColumn();
+ if (partitionColumn != null) {
+ int numPartitions =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getNumPartitions(partitionColumn);
+ Preconditions.checkState(numPartitions > 0, "Number of partitions for column: %s is not properly configured",
+ partitionColumn);
+ finalSelectionConfig.setNumPartitions(numPartitions);
+ finalSelectionConfig.setNumServersPerPartition(replicaGroupStrategyConfig.getNumInstancesPerPartition());
+ } else {
+ // If partition column is not configured, use replicaGroupStrategyConfig.getNumInstancesPerPartition() as
+ // number of servers per replica for backward-compatibility
+ finalSelectionConfig.setNumServersPerReplica(replicaGroupStrategyConfig.getNumInstancesPerPartition());
+ }
+ } else {
+ finalSelectionConfig.setNumReplicas(validationConfig.getReplicasPerPartitionNumber());
+ }
+ }
+ instanceAssignmentConfig.setFinalSelectionConfig(finalSelectionConfig);
+
+ return instanceAssignmentConfig;
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
index 26be6de..386621b 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
@@ -381,6 +381,11 @@ public class CommonConstants {
OFFLINE, REALTIME
}
+ public static class AssignmentStrategy {
+ public static String BALANCE_NUM_SEGMENT_ASSIGNMENT_STRATEGY = "BalanceNumSegmentAssignmentStrategy";
+ public static String REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY = "ReplicaGroupSegmentAssignmentStrategy";
+ }
+
@Deprecated
public static final String TABLE_NAME = "segment.table.name";
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/JsonUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/JsonUtils.java
index 7d97195..a1354fc 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/JsonUtils.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/JsonUtils.java
@@ -19,6 +19,7 @@
package org.apache.pinot.common.utils;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
@@ -40,14 +41,20 @@ public class JsonUtils {
private JsonUtils() {
}
- public static final ObjectMapper DEFAULT_MAPPER = new ObjectMapper();
+ // NOTE: Do not expose the ObjectMapper to prevent configuration change
+ private static final ObjectMapper DEFAULT_MAPPER = new ObjectMapper();
public static final ObjectReader DEFAULT_READER = DEFAULT_MAPPER.reader();
public static final ObjectWriter DEFAULT_WRITER = DEFAULT_MAPPER.writer();
public static final ObjectWriter DEFAULT_PRETTY_WRITER = DEFAULT_MAPPER.writerWithDefaultPrettyPrinter();
public static <T> T stringToObject(String jsonString, Class<T> valueType)
throws IOException {
- return DEFAULT_MAPPER.readValue(jsonString, valueType);
+ return DEFAULT_READER.forType(valueType).readValue(jsonString);
+ }
+
+ public static <T> T stringToObject(String jsonString, TypeReference<T> valueTypeRef)
+ throws IOException {
+ return DEFAULT_READER.forType(valueTypeRef).readValue(jsonString);
}
public static JsonNode stringToJsonNode(String jsonString)
@@ -57,19 +64,19 @@ public class JsonUtils {
public static <T> T fileToObject(File jsonFile, Class<T> valueType)
throws IOException {
- return DEFAULT_MAPPER.readValue(jsonFile, valueType);
+ return DEFAULT_READER.forType(valueType).readValue(jsonFile);
}
public static JsonNode fileToJsonNode(File jsonFile)
throws IOException {
try (InputStream inputStream = new FileInputStream(jsonFile)) {
- return inputStreamToJsonNode(inputStream);
+ return DEFAULT_READER.readTree(inputStream);
}
}
public static <T> T inputStreamToObject(InputStream jsonInputStream, Class<T> valueType)
throws IOException {
- return DEFAULT_MAPPER.readValue(jsonInputStream, valueType);
+ return DEFAULT_READER.forType(valueType).readValue(jsonInputStream);
}
public static JsonNode inputStreamToJsonNode(InputStream jsonInputStream)
@@ -79,17 +86,22 @@ public class JsonUtils {
public static <T> T bytesToObject(byte[] jsonBytes, Class<T> valueType)
throws IOException {
- return DEFAULT_MAPPER.readValue(jsonBytes, valueType);
+ return DEFAULT_READER.forType(valueType).readValue(jsonBytes);
}
public static JsonNode bytesToJsonNode(byte[] jsonBytes)
throws IOException {
- return inputStreamToJsonNode(new ByteArrayInputStream(jsonBytes));
+ return DEFAULT_READER.readTree(new ByteArrayInputStream(jsonBytes));
}
public static <T> T jsonNodeToObject(JsonNode jsonNode, Class<T> valueType)
- throws JsonProcessingException {
- return DEFAULT_READER.treeToValue(jsonNode, valueType);
+ throws IOException {
+ return DEFAULT_READER.forType(valueType).readValue(jsonNode);
+ }
+
+ public static <T> T jsonNodeToObject(JsonNode jsonNode, TypeReference<T> valueTypeRef)
+ throws IOException {
+ return DEFAULT_READER.forType(valueTypeRef).readValue(jsonNode);
}
public static String objectToString(Object object)
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/config/TableConfigTest.java b/pinot-common/src/test/java/org/apache/pinot/common/config/TableConfigTest.java
index 94fbe39..2112afd 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/config/TableConfigTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/config/TableConfigTest.java
@@ -19,11 +19,18 @@
package org.apache.pinot.common.config;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
+import org.apache.pinot.common.config.instance.ConstraintConfig;
+import org.apache.pinot.common.config.instance.FinalSelectionConfig;
+import org.apache.pinot.common.config.instance.InitialSelectionConfig;
+import org.apache.pinot.common.config.instance.InstanceAssignmentConfig;
import org.apache.pinot.common.data.StarTreeIndexSpec;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
+import org.apache.pinot.common.utils.InstancePartitionsType;
import org.apache.pinot.startree.hll.HllConfig;
import org.testng.annotations.Test;
@@ -112,7 +119,7 @@ public class TableConfigTest {
throws Exception {
TableConfig.Builder tableConfigBuilder = new TableConfig.Builder(TableType.OFFLINE).setTableName("myTable");
{
- // No quota config
+ // Only mandatory configs
TableConfig tableConfig = tableConfigBuilder.build();
assertEquals(tableConfig.getTableName(), "myTable_OFFLINE");
@@ -252,11 +259,11 @@ public class TableConfigTest {
checkTableConfigWithAssignmentConfig(tableConfig, tableConfigToCompare);
}
{
+ // With completion config
CompletionConfig completionConfig = new CompletionConfig();
completionConfig.setCompletionMode("DEFAULT");
- TableConfig tableConfig =
- tableConfigBuilder.build();
+ TableConfig tableConfig = tableConfigBuilder.build();
tableConfig.getValidationConfig().setCompletionConfig(completionConfig);
// Serialize then de-serialize
@@ -337,6 +344,30 @@ public class TableConfigTest {
tableConfigToCompare = TableConfig.fromZnRecord(tableConfig.toZNRecord());
checkTableConfigWithHllConfig(tableConfig, tableConfigToCompare);
}
+ {
+ // With instance assignment config
+ InitialSelectionConfig initialSelectionConfig = new InitialSelectionConfig();
+ initialSelectionConfig.setTag("tenant_OFFLINE");
+ initialSelectionConfig.setPoolBased(true);
+ initialSelectionConfig.setNumPools(3);
+ ConstraintConfig constraintConfig = new ConstraintConfig();
+ constraintConfig.setConstraints(Arrays.asList("constraint1", "constraint2"));
+ FinalSelectionConfig finalSelectionConfig = new FinalSelectionConfig();
+ finalSelectionConfig.setReplicaGroupBased(true);
+ finalSelectionConfig.setNumReplicas(3);
+ finalSelectionConfig.setNumServersPerReplica(5);
+ InstanceAssignmentConfig instanceAssignmentConfig = new InstanceAssignmentConfig();
+ instanceAssignmentConfig.setInitialSelectionConfig(initialSelectionConfig);
+ instanceAssignmentConfig.setConstraintConfig(constraintConfig);
+ instanceAssignmentConfig.setFinalSelectionConfig(finalSelectionConfig);
+ Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap =
+ Collections.singletonMap(InstancePartitionsType.OFFLINE, instanceAssignmentConfig);
+ TableConfig tableConfig = tableConfigBuilder.setInstanceAssignmentConfigMap(instanceAssignmentConfigMap).build();
+
+ // Serialize then de-serialize
+ checkTableConfigWithInstanceAssignmentConfigMap(TableConfig.fromJsonConfig(tableConfig.toJsonConfig()));
+ checkTableConfigWithInstanceAssignmentConfigMap(TableConfig.fromZnRecord(tableConfig.toZNRecord()));
+ }
}
private void checkTableConfigWithAssignmentConfig(TableConfig tableConfig, TableConfig tableConfigToCompare) {
@@ -362,8 +393,7 @@ public class TableConfigTest {
tableConfig.getValidationConfig().getCompletionConfig());
// Check that the configurations are correct.
- CompletionConfig completionConfig =
- tableConfigToCompare.getValidationConfig().getCompletionConfig();
+ CompletionConfig completionConfig = tableConfigToCompare.getValidationConfig().getCompletionConfig();
assertEquals(completionConfig.getCompletionMode(), "DEFAULT");
}
@@ -409,4 +439,27 @@ public class TableConfigTest {
assertEquals(hllConfig.getHllLog2m(), 9);
assertEquals(hllConfig.getHllDeriveColumnSuffix(), "suffix");
}
+
+ private void checkTableConfigWithInstanceAssignmentConfigMap(TableConfig tableConfig) {
+ Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap =
+ tableConfig.getInstanceAssignmentConfigMap();
+ assertNotNull(instanceAssignmentConfigMap);
+ assertEquals(instanceAssignmentConfigMap.size(), 1);
+ assertTrue(instanceAssignmentConfigMap.containsKey(InstancePartitionsType.OFFLINE));
+ InstanceAssignmentConfig instanceAssignmentConfig = instanceAssignmentConfigMap.get(InstancePartitionsType.OFFLINE);
+ InitialSelectionConfig initialSelectionConfig = instanceAssignmentConfig.getInitialSelectionConfig();
+ assertEquals(initialSelectionConfig.getTag(), "tenant_OFFLINE");
+ assertTrue(initialSelectionConfig.isPoolBased());
+ assertEquals(initialSelectionConfig.getNumPools(), 3);
+ assertNull(initialSelectionConfig.getPools());
+ ConstraintConfig constraintConfig = instanceAssignmentConfig.getConstraintConfig();
+ assertEquals(constraintConfig.getConstraints(), Arrays.asList("constraint1", "constraint2"));
+ FinalSelectionConfig finalSelectionConfig = instanceAssignmentConfig.getFinalSelectionConfig();
+ assertTrue(finalSelectionConfig.isReplicaGroupBased());
+ assertEquals(finalSelectionConfig.getNumServers(), 0);
+ assertEquals(finalSelectionConfig.getNumReplicas(), 3);
+ assertEquals(finalSelectionConfig.getNumServersPerReplica(), 5);
+ assertEquals(finalSelectionConfig.getNumPartitions(), 0);
+ assertEquals(finalSelectionConfig.getNumServersPerPartition(), 0);
+ }
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/InstancePartitionsUtils.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/InstancePartitionsUtils.java
index 4121246..be2c99d 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/InstancePartitionsUtils.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/InstancePartitionsUtils.java
@@ -23,12 +23,10 @@ import java.util.List;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixManager;
import org.apache.helix.ZNRecord;
-import org.apache.helix.model.InstanceConfig;
import org.apache.helix.store.HelixPropertyStore;
-import org.apache.pinot.common.config.OfflineTagConfig;
-import org.apache.pinot.common.config.RealtimeTagConfig;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.config.TableNameBuilder;
+import org.apache.pinot.common.config.TagNameUtils;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.utils.InstancePartitionsType;
import org.apache.pinot.common.utils.helix.HelixHelper;
@@ -60,26 +58,13 @@ public class InstancePartitionsUtils {
// Compute the instance partitions (for backward compatible)
// Sort all enabled instances with the server tag, rotate the list based on the table name to prevent creating
// hotspot servers
- InstancePartitions instancePartitions = new InstancePartitions(instancePartitionsName);
- List<InstanceConfig> instanceConfigs = HelixHelper.getInstanceConfigs(helixManager);
- String serverTag;
- switch (instancePartitionsType) {
- case OFFLINE:
- serverTag = new OfflineTagConfig(tableConfig).getOfflineServerTag();
- break;
- case CONSUMING:
- serverTag = new RealtimeTagConfig(tableConfig).getConsumingServerTag();
- break;
- case COMPLETED:
- serverTag = new RealtimeTagConfig(tableConfig).getCompletedServerTag();
- break;
- default:
- throw new IllegalArgumentException();
- }
- List<String> instances = HelixHelper.getEnabledInstancesWithTag(instanceConfigs, serverTag);
+ String serverTag =
+ TagNameUtils.getServerTagFromTableConfigAndInstancePartitionsType(tableConfig, instancePartitionsType);
+ List<String> instances = HelixHelper.getEnabledInstancesWithTag(helixManager, serverTag);
instances.sort(null);
int numInstances = instances.size();
Collections.rotate(instances, -(Math.abs(tableNameWithType.hashCode()) % numInstances));
+ InstancePartitions instancePartitions = new InstancePartitions(instancePartitionsName);
instancePartitions.setInstances(0, 0, instances);
return instancePartitions;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/SegmentAssignmentStrategyFactory.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/SegmentAssignmentStrategyFactory.java
index d71b1d9..875b856 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/SegmentAssignmentStrategyFactory.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/SegmentAssignmentStrategyFactory.java
@@ -21,6 +21,7 @@ package org.apache.pinot.controller.helix.core.assignment.segment;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
+import org.apache.pinot.common.utils.CommonConstants.Segment.AssignmentStrategy;
/**
@@ -30,14 +31,10 @@ public class SegmentAssignmentStrategyFactory {
private SegmentAssignmentStrategyFactory() {
}
- public enum Strategy {
- BalanceNumSegmentAssignmentStrategy, ReplicaGroupSegmentAssignmentStrategy
- }
-
public static SegmentAssignmentStrategy getSegmentAssignmentStrategy(HelixManager helixManager,
TableConfig tableConfig) {
SegmentAssignmentStrategy segmentAssignmentStrategy;
- if (Strategy.ReplicaGroupSegmentAssignmentStrategy.name()
+ if (AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY
.equalsIgnoreCase(tableConfig.getValidationConfig().getSegmentAssignmentStrategy())) {
if (tableConfig.getTableType() == TableType.OFFLINE) {
segmentAssignmentStrategy = new OfflineReplicaGroupSegmentAssignmentStrategy();
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/OfflineReplicaGroupSegmentAssignmentStrategyTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/OfflineReplicaGroupSegmentAssignmentStrategyTest.java
index 47ecdd8..28586f0 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/OfflineReplicaGroupSegmentAssignmentStrategyTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/OfflineReplicaGroupSegmentAssignmentStrategyTest.java
@@ -34,8 +34,9 @@ import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.ColumnPartitionMetadata;
import org.apache.pinot.common.metadata.segment.OfflineSegmentZKMetadata;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
-import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.CommonConstants.Helix.StateModel.SegmentOnlineOfflineStateModel;
+import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
+import org.apache.pinot.common.utils.CommonConstants.Segment.AssignmentStrategy;
import org.apache.pinot.common.utils.InstancePartitionsType;
import org.apache.pinot.controller.helix.core.assignment.InstancePartitions;
import org.testng.annotations.BeforeClass;
@@ -103,9 +104,9 @@ public class OfflineReplicaGroupSegmentAssignmentStrategyTest {
when(helixManagerWithoutPartitions.getHelixPropertyStore()).thenReturn(propertyStoreWithoutPartitions);
TableConfig tableConfigWithoutPartitions =
- new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITHOUT_PARTITION)
- .setNumReplicas(NUM_REPLICAS).setSegmentAssignmentStrategy(
- SegmentAssignmentStrategyFactory.Strategy.ReplicaGroupSegmentAssignmentStrategy.name()).build();
+ new TableConfig.Builder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITHOUT_PARTITION)
+ .setNumReplicas(NUM_REPLICAS)
+ .setSegmentAssignmentStrategy(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY).build();
_strategyWithoutPartition = SegmentAssignmentStrategyFactory
.getSegmentAssignmentStrategy(helixManagerWithoutPartitions, tableConfigWithoutPartitions);
@@ -163,9 +164,9 @@ public class OfflineReplicaGroupSegmentAssignmentStrategyTest {
ReplicaGroupStrategyConfig strategyConfig = new ReplicaGroupStrategyConfig();
strategyConfig.setPartitionColumn(PARTITION_COLUMN);
TableConfig tableConfigWithPartitions =
- new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITH_PARTITION)
- .setNumReplicas(NUM_REPLICAS).setSegmentAssignmentStrategy(
- SegmentAssignmentStrategyFactory.Strategy.ReplicaGroupSegmentAssignmentStrategy.name()).build();
+ new TableConfig.Builder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME_WITH_PARTITION)
+ .setNumReplicas(NUM_REPLICAS)
+ .setSegmentAssignmentStrategy(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY).build();
tableConfigWithPartitions.getValidationConfig().setReplicaGroupStrategyConfig(strategyConfig);
_strategyWithPartition = SegmentAssignmentStrategyFactory
.getSegmentAssignmentStrategy(helixManagerWithPartitions, tableConfigWithPartitions);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentStrategyTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentStrategyTest.java
index 6d83cc6..f74a7b2 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentStrategyTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentStrategyTest.java
@@ -29,8 +29,9 @@ import org.apache.helix.ZNRecord;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
-import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.CommonConstants.Helix.StateModel.RealtimeSegmentOnlineOfflineStateModel;
+import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
+import org.apache.pinot.common.utils.CommonConstants.Segment.AssignmentStrategy;
import org.apache.pinot.common.utils.InstancePartitionsType;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.helix.core.assignment.InstancePartitions;
@@ -125,9 +126,9 @@ public class RealtimeReplicaGroupSegmentAssignmentStrategyTest {
when(helixManager.getHelixPropertyStore()).thenReturn(propertyStore);
TableConfig tableConfig =
- new TableConfig.Builder(CommonConstants.Helix.TableType.REALTIME).setTableName(RAW_TABLE_NAME)
- .setNumReplicas(NUM_REPLICAS).setLLC(true).setSegmentAssignmentStrategy(
- SegmentAssignmentStrategyFactory.Strategy.ReplicaGroupSegmentAssignmentStrategy.name()).build();
+ new TableConfig.Builder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
+ .setLLC(true).setSegmentAssignmentStrategy(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY)
+ .build();
_strategy = SegmentAssignmentStrategyFactory.getSegmentAssignmentStrategy(helixManager, tableConfig);
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/readers/JSONRecordReader.java b/pinot-core/src/main/java/org/apache/pinot/core/data/readers/JSONRecordReader.java
index 23ec05b..cc4bcd7 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/readers/JSONRecordReader.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/readers/JSONRecordReader.java
@@ -18,11 +18,10 @@
*/
package org.apache.pinot.core.data.readers;
-import com.fasterxml.jackson.core.JsonFactory;
-import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.MappingIterator;
import java.io.File;
import java.io.IOException;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.apache.pinot.common.data.FieldSpec;
@@ -36,14 +35,11 @@ import org.apache.pinot.core.indexsegment.generator.SegmentGeneratorConfig;
* Record reader for JSON file.
*/
public class JSONRecordReader implements RecordReader {
- private final JsonFactory _factory = new JsonFactory();
private final File _dataFile;
private final Schema _schema;
private final List<FieldSpec> _fieldSpecs;
- private JsonParser _parser;
- private Iterator<Map> _iterator;
-
+ private MappingIterator<Map<String, Object>> _iterator;
public JSONRecordReader(File dataFile, Schema schema)
throws IOException {
@@ -56,18 +52,17 @@ public class JSONRecordReader implements RecordReader {
private void init()
throws IOException {
- _parser = _factory.createParser(RecordReaderUtils.getBufferedReader(_dataFile));
try {
- _iterator = JsonUtils.DEFAULT_MAPPER.readValues(_parser, Map.class);
+ _iterator = JsonUtils.DEFAULT_READER.forType(new TypeReference<Map<String, Object>>() {
+ }).readValues(_dataFile);
} catch (Exception e) {
- _parser.close();
+ _iterator.close();
throw e;
}
}
@Override
public void init(SegmentGeneratorConfig segmentGeneratorConfig) {
-
}
@Override
@@ -84,7 +79,7 @@ public class JSONRecordReader implements RecordReader {
@SuppressWarnings("Duplicates")
@Override
public GenericRow next(GenericRow reuse) {
- Map record = _iterator.next();
+ Map<String, Object> record = _iterator.next();
for (FieldSpec fieldSpec : _fieldSpecs) {
String fieldName = fieldSpec.getName();
Object value = record.get(fieldName);
@@ -99,7 +94,7 @@ public class JSONRecordReader implements RecordReader {
@Override
public void rewind()
throws IOException {
- _parser.close();
+ _iterator.close();
init();
}
@@ -111,6 +106,6 @@ public class JSONRecordReader implements RecordReader {
@Override
public void close()
throws IOException {
- _parser.close();
+ _iterator.close();
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
index 2f145a8..881b4dd 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
@@ -23,6 +23,7 @@ import com.google.common.base.Function;
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -38,6 +39,7 @@ import org.apache.pinot.common.config.TableCustomConfig;
import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.config.TenantConfig;
import org.apache.pinot.common.utils.CommonConstants;
+import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.util.TestUtils;
import org.testng.Assert;
@@ -161,20 +163,8 @@ public class LLCRealtimeClusterIntegrationTest extends RealtimeClusterIntegratio
@Test
public void testAddHLCTableShouldFail() {
- TableConfig tableConfig = new TableConfig();
- IndexingConfig indexingConfig = new IndexingConfig();
- Map<String, String> streamConfigs = new HashMap<>();
- streamConfigs.put("stream.kafka.consumer.type", "HIGHLEVEL");
- indexingConfig.setStreamConfigs(streamConfigs);
- tableConfig.setIndexingConfig(indexingConfig);
- tableConfig.setTableName("testTable");
- tableConfig.setTableType(CommonConstants.Helix.TableType.REALTIME);
- SegmentsValidationAndRetentionConfig validationAndRetentionConfig = new SegmentsValidationAndRetentionConfig();
- tableConfig.setValidationConfig(validationAndRetentionConfig);
- TenantConfig tenantConfig = new TenantConfig();
- tableConfig.setTenantConfig(tenantConfig);
- TableCustomConfig tableCustomConfig = new TableCustomConfig();
- tableConfig.setCustomConfig(tableCustomConfig);
+ TableConfig tableConfig = new TableConfig.Builder(TableType.REALTIME).setTableName("testTable")
+ .setStreamConfigs(Collections.singletonMap("stream.kafka.consumer.type", "HIGHLEVEL")).build();
try {
sendPostRequest(_controllerRequestURLBuilder.forTableCreate(), tableConfig.toJsonConfigString());
Assert.fail();
@@ -183,4 +173,3 @@ public class LLCRealtimeClusterIntegrationTest extends RealtimeClusterIntegratio
}
}
}
-
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org