add pinot upsert features to pinot common #5175
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -30,17 +30,21 @@ | |
| import java.util.List; | ||
| import java.util.Map; | ||
| import javax.annotation.Nullable; | ||
| import javax.validation.constraints.Null; | ||
|
|
||
| import org.apache.helix.ZNRecord; | ||
| import org.apache.pinot.common.assignment.InstancePartitionsType; | ||
| import org.apache.pinot.common.config.instance.InstanceAssignmentConfig; | ||
| import org.apache.pinot.common.utils.CommonConstants.Helix.TableType; | ||
| import org.apache.pinot.spi.utils.JsonUtils; | ||
| import org.apache.pinot.common.utils.CommonConstants.UpdateSemantic; | ||
|
|
||
|
|
||
| @SuppressWarnings({"Duplicates", "unused"}) | ||
| public class TableConfig extends BaseJsonConfig { | ||
| public static final String TABLE_NAME_KEY = "tableName"; | ||
| public static final String TABLE_TYPE_KEY = "tableType"; | ||
| private static final String UPDATE_SEMANTIC_CONFIG_KEY = "updateSemantic"; | ||
| public static final String VALIDATION_CONFIG_KEY = "segmentsConfig"; | ||
| public static final String TENANT_CONFIG_KEY = "tenants"; | ||
| public static final String INDEXING_CONFIG_KEY = "tableIndexConfig"; | ||
|
|
@@ -75,6 +79,9 @@ public class TableConfig extends BaseJsonConfig { | |
| private Map<InstancePartitionsType, InstanceAssignmentConfig> _instanceAssignmentConfigMap; | ||
| private List<FieldConfig> _fieldConfigList; | ||
|
|
||
| @JsonPropertyDescription(value = "The update semantic of the table, either append or upsert, default as append") | ||
| private UpdateSemantic _updateSemantic; | ||
|
There was a problem hiding this comment. rename to IngestMode or just have a boolean enableUpsert? There was a problem hiding this comment. Is this for real-time? We already have push type of either APPEND or REFRESH for offline; aggregateMetrics for real-time. What does UPSERT stand for? There was a problem hiding this comment. @kishoreg I am thinking if using enum here will be more flexible in the case of future features changes or whatnot. Boolean seems to be only limiting two options and it might limit what other changes we might want to do with pinot. Please let me what you think. ingestionMode definitely sounds more explicit than what I am using here and I will definite consider about this. @Jackie-Jiang yes. So basically it is previous old design about we want to support update in pinot realtime ingestion by adding override by primary key (eg, a second message in Kafka with the same primary key represents the update of existing key instead of two records with the same primary key). Upsert stands for this new ingestion semantic (insert if no duplicated primary key, update if duplicate primary key). This feature will be applying to realtime table only |
||
|
|
||
| /** | ||
| * NOTE: DO NOT use this constructor, use builder instead. This constructor is for deserializer only. | ||
| */ | ||
|
|
@@ -90,6 +97,16 @@ private TableConfig(String tableName, TableType tableType, SegmentsValidationAnd | |
| @Nullable QueryConfig queryConfig, | ||
| @Nullable Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap, | ||
| @Nullable List<FieldConfig> fieldConfigList) { | ||
| this(tableName, tableType, validationConfig, tenantConfig, indexingConfig, customConfig, quotaConfig, taskConfig, | ||
| routingConfig, queryConfig, instanceAssignmentConfigMap, fieldConfigList, UpdateSemantic.APPEND); | ||
| } | ||
|
|
||
| private TableConfig(String tableName, TableType tableType, SegmentsValidationAndRetentionConfig validationConfig, | ||
| TenantConfig tenantConfig, IndexingConfig indexingConfig, TableCustomConfig customConfig, | ||
| @Nullable QuotaConfig quotaConfig, @Nullable TableTaskConfig taskConfig, @Nullable RoutingConfig routingConfig, | ||
| @Null QueryConfig queryConfig, | ||
| @Nullable Map<InstancePartitionsType, InstanceAssignmentConfig> instanceAssignmentConfigMap, | ||
| @Nullable List<FieldConfig> fieldConfigList, @Nullable UpdateSemantic updateSemantic) { | ||
| _tableName = TableNameBuilder.forType(tableType).tableNameWithType(tableName); | ||
| _tableType = tableType; | ||
| _validationConfig = validationConfig; | ||
|
|
@@ -102,6 +119,7 @@ private TableConfig(String tableName, TableType tableType, SegmentsValidationAnd | |
| _queryConfig = queryConfig; | ||
| _instanceAssignmentConfigMap = instanceAssignmentConfigMap; | ||
| _fieldConfigList = fieldConfigList; | ||
| _updateSemantic = updateSemantic; | ||
| } | ||
|
|
||
| public static TableConfig fromJsonString(String jsonString) | ||
|
|
@@ -156,8 +174,15 @@ public static TableConfig fromJsonConfig(JsonNode jsonConfig) | |
| extractChildConfig(jsonConfig, FIELD_CONFIG_LIST_KEY, new TypeReference<List<FieldConfig>>() { | ||
| }); | ||
|
|
||
| // generate update schematic | ||
| UpdateSemantic updateSemantic = UpdateSemantic.DEFAULT_SEMANTIC; | ||
| if (jsonConfig.has(UPDATE_SEMANTIC_CONFIG_KEY)) { | ||
| updateSemantic = UpdateSemantic.getUpdateSemantic(jsonConfig.get(UPDATE_SEMANTIC_CONFIG_KEY).asText()); | ||
| } | ||
|
|
||
| return new TableConfig(tableName, tableType, validationConfig, tenantConfig, indexingConfig, customConfig, | ||
| quotaConfig, taskConfig, routingConfig, queryConfig, instanceAssignmentConfigMap, fieldConfigList, | ||
| updateSemantic); | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -232,6 +257,11 @@ public ObjectNode toJsonConfig() { | |
| jsonConfig.put(FIELD_CONFIG_LIST_KEY, JsonUtils.objectToJsonNode(_fieldConfigList)); | ||
| } | ||
|
|
||
| if (_updateSemantic != null) { | ||
| jsonConfig.put(UPDATE_SEMANTIC_CONFIG_KEY, _updateSemantic.toString()); | ||
| } else { | ||
| jsonConfig.put(UPDATE_SEMANTIC_CONFIG_KEY, UpdateSemantic.APPEND.toString()); | ||
| } | ||
| return jsonConfig; | ||
| } | ||
|
|
||
|
|
@@ -315,8 +345,11 @@ public static TableConfig fromZnRecord(ZNRecord znRecord) | |
| }); | ||
| } | ||
|
|
||
| UpdateSemantic updateSemantic = UpdateSemantic.getUpdateSemantic(simpleFields.get(UPDATE_SEMANTIC_CONFIG_KEY)); | ||
|
|
||
| return new TableConfig(tableName, tableType, validationConfig, tenantConfig, indexingConfig, customConfig, | ||
| quotaConfig, taskConfig, routingConfig, queryConfig, instanceAssignmentConfigMap, fieldConfigList, | ||
| updateSemantic); | ||
| } | ||
|
|
||
| public ZNRecord toZNRecord() | ||
|
|
@@ -354,6 +387,7 @@ public ZNRecord toZNRecord() | |
| } | ||
|
|
||
| ZNRecord znRecord = new ZNRecord(_tableName); | ||
| simpleFields.put(UPDATE_SEMANTIC_CONFIG_KEY, _updateSemantic.toString()); | ||
jamesyfshao marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| znRecord.setSimpleFields(simpleFields); | ||
| return znRecord; | ||
| } | ||
|
|
@@ -470,6 +504,18 @@ public List<FieldConfig> getFieldConfigList() { | |
| return _fieldConfigList; | ||
| } | ||
|
|
||
| public UpdateSemantic getUpdateSemantic() { | ||
| return _updateSemantic; | ||
| } | ||
|
|
||
| public void setUpdateSemantic(UpdateSemantic updateSemantic) { | ||
| _updateSemantic = updateSemantic; | ||
| } | ||
|
|
||
| public boolean isTableForUpsert() { | ||
|
There was a problem hiding this comment. let's avoid adding additional utility methods to config classes if possible |
||
| return _updateSemantic == UpdateSemantic.UPSERT; | ||
| } | ||
|
|
||
| public static class Builder { | ||
| private static final String DEFAULT_SEGMENT_PUSH_TYPE = "APPEND"; | ||
| private static final String REFRESH_SEGMENT_PUSH_TYPE = "REFRESH"; | ||
|
|
@@ -517,6 +563,8 @@ public static class Builder { | |
| private Map<InstancePartitionsType, InstanceAssignmentConfig> _instanceAssignmentConfigMap; | ||
| private List<FieldConfig> _fieldConfigList; | ||
|
|
||
| private UpdateSemantic _updateSemantic; | ||
|
|
||
| public Builder(TableType tableType) { | ||
| _tableType = tableType; | ||
| } | ||
|
|
@@ -683,6 +731,11 @@ public Builder setFieldConfigList(List<FieldConfig> fieldConfigList) { | |
| return this; | ||
| } | ||
|
|
||
| public Builder setUpdateSemantic(UpdateSemantic updateSemantic) { | ||
| _updateSemantic = updateSemantic; | ||
| return this; | ||
| } | ||
|
|
||
| public TableConfig build() { | ||
| // Validation config | ||
| SegmentsValidationAndRetentionConfig validationConfig = new SegmentsValidationAndRetentionConfig(); | ||
|
|
@@ -720,14 +773,20 @@ public TableConfig build() { | |
| _customConfig = new TableCustomConfig(null); | ||
| } | ||
|
|
||
| if (_updateSemantic == null) { | ||
| _updateSemantic = UpdateSemantic.DEFAULT_SEMANTIC; | ||
| } | ||
|
|
||
| // eventually this validation will be generic but since we are initially | ||
| // using FieldConfig only for text columns (and migrate to expand its usage | ||
| // soon after), just validate the field config list from text index creation | ||
| // perspective. | ||
| TextIndexConfigValidator.validate(_fieldConfigList, _noDictionaryColumns); | ||
|
|
||
| return new TableConfig(_tableName, _tableType, validationConfig, tenantConfig, indexingConfig, _customConfig, | ||
| _quotaConfig, _taskConfig, _routingConfig, _queryConfig, _instanceAssignmentConfigMap, _fieldConfigList, | ||
| _updateSemantic); | ||
|
|
||
| } | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -132,4 +132,19 @@ public static boolean isOfflineTableResource(String resourceName) { | |
| public static boolean isRealtimeTableResource(String resourceName) { | ||
| return REALTIME.tableHasTypeSuffix(resourceName); | ||
| } | ||
|
|
||
| /** | ||
| * ensure that table name ends with type info, if no, create one with the given type | ||
| * @param tableName the name of the table | ||
| * @param type the type of the table for it to fill in if the type info is missing | ||
| * @return the table type name with the type info | ||
| */ | ||
| public static String ensureTableNameWithType(String tableName, TableType type) { | ||
|
There was a problem hiding this comment. Please use There was a problem hiding this comment. removed this method, @kishoreg this method is used in other components that I will submit PR later |
||
| for (TableType tableType: TableType.values()) { | ||
| if (tableName.endsWith(TYPE_SUFFIX_SEPARATOR + tableType.toString())) { | ||
| return tableName; | ||
| } | ||
| } | ||
| return tableName + TYPE_SUFFIX_SEPARATOR + type.toString(); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| /** | ||
| * 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.restlet.resources; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import org.codehaus.jackson.annotate.JsonIgnoreProperties; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.Map; | ||
|
|
||
| @JsonIgnoreProperties(ignoreUnknown = true) | ||
| public class TableLowWaterMarksInfo { | ||
| // A mapping from table name to the table's partiton -> low_water_mark mappings. | ||
| private Map<String, Map<Integer, Long>> tableLowWaterMarks; | ||
|
|
||
| public TableLowWaterMarksInfo(@JsonProperty("lowWaterMarks") Map<String, Map<Integer, Long>> tableLowWaterMarks) { | ||
| this.tableLowWaterMarks = tableLowWaterMarks; | ||
| } | ||
|
|
||
| public void setTableLowWaterMarks(Map<String, Map<Integer, Long>> tableLowWaterMarks) { | ||
| this.tableLowWaterMarks = tableLowWaterMarks; | ||
| } | ||
|
|
||
| public Map<String, Map<Integer, Long>> getTableLowWaterMarks() { | ||
| return Collections.unmodifiableMap(tableLowWaterMarks); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Wrong import
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
issue with merging, fixed. Thanks for pointing it out