You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ne...@apache.org on 2022/01/03 18:24:52 UTC
[pinot] branch master updated: addking skip key for realtimeToOffline job validation (#7921)
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1e27f01 addking skip key for realtimeToOffline job validation (#7921)
1e27f01 is described below
commit 1e27f01c85ef20b503ea7ae6cd166c14a08402aa
Author: Rong Rong <wa...@gmail.com>
AuthorDate: Mon Jan 3 10:24:28 2022 -0800
addking skip key for realtimeToOffline job validation (#7921)
* add skipValidate QueryParam in and endpoints
* fix test failures
* address diff comments
* fix jdk8 compile
* address diff comments
Co-authored-by: Rong Rong <ro...@startree.ai>
---
.../api/resources/PinotTableRestletResource.java | 32 +++++++++-----
.../api/resources/TableConfigsRestletResource.java | 25 +++++++----
.../segment/local/utils/TableConfigUtils.java | 49 +++++++++++++++++-----
.../segment/local/utils/TableConfigUtilsTest.java | 12 ++++--
4 files changed, 86 insertions(+), 32 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
index 7c22851..91e2a39 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
@@ -153,7 +153,11 @@ public class PinotTableRestletResource {
@Produces(MediaType.APPLICATION_JSON)
@Path("/tables")
@ApiOperation(value = "Adds a table", notes = "Adds a table")
- public SuccessResponse addTable(String tableConfigStr, @Context HttpHeaders httpHeaders, @Context Request request) {
+ public SuccessResponse addTable(
+ String tableConfigStr,
+ @ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip, @Context HttpHeaders httpHeaders,
+ @Context Request request) {
// TODO introduce a table config ctor with json string.
TableConfig tableConfig;
String tableName;
@@ -171,7 +175,7 @@ public class PinotTableRestletResource {
TableConfigTunerUtils.applyTunerConfigs(_pinotHelixResourceManager, tableConfig, schema, Collections.emptyMap());
// TableConfigUtils.validate(...) is used across table create/update.
- TableConfigUtils.validate(tableConfig, schema);
+ TableConfigUtils.validate(tableConfig, schema, typesToSkip);
// TableConfigUtils.validateTableName(...) checks table name rules.
// So it won't effect already created tables.
TableConfigUtils.validateTableName(tableConfig);
@@ -435,13 +439,14 @@ public class PinotTableRestletResource {
@ApiOperation(value = "Updates table config for a table", notes = "Updates table config for a table")
public SuccessResponse updateTableConfig(
@ApiParam(value = "Name of the table to update", required = true) @PathParam("tableName") String tableName,
- String tableConfigString)
+ @ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip, String tableConfigString)
throws Exception {
TableConfig tableConfig;
try {
tableConfig = JsonUtils.stringToObject(tableConfigString, TableConfig.class);
Schema schema = _pinotHelixResourceManager.getSchemaForTableConfig(tableConfig);
- TableConfigUtils.validate(tableConfig, schema);
+ TableConfigUtils.validate(tableConfig, schema, typesToSkip);
} catch (Exception e) {
String msg = String.format("Invalid table config: %s with error: %s", tableName, e.getMessage());
throw new ControllerApplicationException(LOGGER, msg, Response.Status.BAD_REQUEST, e);
@@ -487,7 +492,10 @@ public class PinotTableRestletResource {
@ApiOperation(value = "Validate table config for a table",
notes = "This API returns the table config that matches the one you get from 'GET /tables/{tableName}'."
+ " This allows us to validate table config before apply.")
- public String checkTableConfig(String tableConfigStr) {
+ public String checkTableConfig(
+ String tableConfigStr,
+ @ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip) {
TableConfig tableConfig;
try {
tableConfig = JsonUtils.stringToObject(tableConfigStr, TableConfig.class);
@@ -495,7 +503,7 @@ public class PinotTableRestletResource {
String msg = String.format("Invalid table config json string: %s", tableConfigStr);
throw new ControllerApplicationException(LOGGER, msg, Response.Status.BAD_REQUEST, e);
}
- return validateConfig(tableConfig, _pinotHelixResourceManager.getSchemaForTableConfig(tableConfig));
+ return validateConfig(tableConfig, _pinotHelixResourceManager.getSchemaForTableConfig(tableConfig), typesToSkip);
}
@Deprecated
@@ -508,22 +516,24 @@ public class PinotTableRestletResource {
+ "Validate given table config and schema. If specified schema is null, attempt to retrieve schema using the "
+ "table name. This API returns the table config that matches the one you get from 'GET /tables/{tableName}'."
+ " This allows us to validate table config before apply.")
- public String validateTableAndSchema(TableAndSchemaConfig tableSchemaConfig) {
+ public String validateTableAndSchema(
+ TableAndSchemaConfig tableSchemaConfig,
+ @ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip) {
TableConfig tableConfig = tableSchemaConfig.getTableConfig();
Schema schema = tableSchemaConfig.getSchema();
if (schema == null) {
schema = _pinotHelixResourceManager.getSchemaForTableConfig(tableConfig);
}
- return validateConfig(tableSchemaConfig.getTableConfig(), schema);
+ return validateConfig(tableSchemaConfig.getTableConfig(), schema, typesToSkip);
}
- private String validateConfig(TableConfig tableConfig, Schema schema) {
+ private String validateConfig(TableConfig tableConfig, Schema schema, @Nullable String typesToSkip) {
try {
if (schema == null) {
throw new SchemaNotFoundException("Got empty schema");
}
-
- TableConfigUtils.validate(tableConfig, schema);
+ TableConfigUtils.validate(tableConfig, schema, typesToSkip);
ObjectNode tableConfigValidateStr = JsonUtils.newObjectNode();
if (tableConfig.getTableType() == TableType.OFFLINE) {
tableConfigValidateStr.set(TableType.OFFLINE.name(), tableConfig.toJsonNode());
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java
index 0713a94..1fa5d7d 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java
@@ -26,6 +26,7 @@ import io.swagger.annotations.ApiParam;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
+import javax.annotation.Nullable;
import javax.inject.Inject;
import javax.ws.rs.DELETE;
import javax.ws.rs.DefaultValue;
@@ -148,11 +149,15 @@ public class TableConfigsRestletResource {
@Path("/tableConfigs")
@ApiOperation(value = "Add the TableConfigs using the tableConfigsStr json",
notes = "Add the TableConfigs using the tableConfigsStr json")
- public SuccessResponse addConfig(String tableConfigsStr, @Context HttpHeaders httpHeaders, @Context Request request) {
+ public SuccessResponse addConfig(
+ String tableConfigsStr,
+ @ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip, @Context HttpHeaders httpHeaders,
+ @Context Request request) {
TableConfigs tableConfigs;
try {
tableConfigs = JsonUtils.stringToObject(tableConfigsStr, TableConfigs.class);
- validateConfig(tableConfigs);
+ validateConfig(tableConfigs, typesToSkip);
} catch (Exception e) {
throw new ControllerApplicationException(LOGGER, String.format("Invalid TableConfigs. %s", e.getMessage()),
Response.Status.BAD_REQUEST, e);
@@ -278,6 +283,8 @@ public class TableConfigsRestletResource {
public SuccessResponse updateConfig(
@ApiParam(value = "TableConfigs name i.e. raw table name", required = true) @PathParam("tableName")
String tableName,
+ @ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@ApiParam(value = "Reload the table if the new schema is backward compatible") @DefaultValue("false")
@QueryParam("reload") boolean reload, String tableConfigsStr)
throws Exception {
@@ -286,7 +293,7 @@ public class TableConfigsRestletResource {
tableConfigs = JsonUtils.stringToObject(tableConfigsStr, TableConfigs.class);
Preconditions.checkState(tableConfigs.getTableName().equals(tableName),
"'tableName' in TableConfigs: %s must match provided tableName: %s", tableConfigs.getTableName(), tableName);
- validateConfig(tableConfigs);
+ validateConfig(tableConfigs, typesToSkip);
} catch (Exception e) {
throw new ControllerApplicationException(LOGGER, String.format("Invalid TableConfigs: %s", tableName),
Response.Status.BAD_REQUEST, e);
@@ -349,7 +356,9 @@ public class TableConfigsRestletResource {
@Path("/tableConfigs/validate")
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Validate the TableConfigs", notes = "Validate the TableConfigs")
- public String validateConfig(String tableConfigsStr) {
+ public String validateConfig(String tableConfigsStr,
+ @ApiParam(value = "comma separated list of validation type(s) to skip. supported types: (ALL|TASK|UPSERT)")
+ @QueryParam("validationTypesToSkip") @Nullable String typesToSkip) {
TableConfigs tableConfigs;
try {
tableConfigs = JsonUtils.stringToObject(tableConfigsStr, TableConfigs.class);
@@ -357,7 +366,7 @@ public class TableConfigsRestletResource {
throw new ControllerApplicationException(LOGGER,
String.format("Invalid TableConfigs json string: %s", tableConfigsStr), Response.Status.BAD_REQUEST, e);
}
- return validateConfig(tableConfigs);
+ return validateConfig(tableConfigs, typesToSkip);
}
private void tuneConfig(TableConfig tableConfig, Schema schema) {
@@ -366,7 +375,7 @@ public class TableConfigsRestletResource {
TableConfigUtils.ensureStorageQuotaConstraints(tableConfig, _controllerConf.getDimTableMaxSize());
}
- private String validateConfig(TableConfigs tableConfigs) {
+ private String validateConfig(TableConfigs tableConfigs, @Nullable String typesToSkip) {
String rawTableName = tableConfigs.getTableName();
TableConfig offlineTableConfig = tableConfigs.getOffline();
TableConfig realtimeTableConfig = tableConfigs.getRealtime();
@@ -387,14 +396,14 @@ public class TableConfigsRestletResource {
Preconditions.checkState(offlineRawTableName.equals(rawTableName),
"Name in 'offline' table config: %s must be equal to 'tableName': %s", offlineRawTableName, rawTableName);
TableConfigUtils.validateTableName(offlineTableConfig);
- TableConfigUtils.validate(offlineTableConfig, schema);
+ TableConfigUtils.validate(offlineTableConfig, schema, typesToSkip);
}
if (realtimeTableConfig != null) {
String realtimeRawTableName = TableNameBuilder.extractRawTableName(realtimeTableConfig.getTableName());
Preconditions.checkState(realtimeRawTableName.equals(rawTableName),
"Name in 'realtime' table config: %s must be equal to 'tableName': %s", realtimeRawTableName, rawTableName);
TableConfigUtils.validateTableName(realtimeTableConfig);
- TableConfigUtils.validate(realtimeTableConfig, schema);
+ TableConfigUtils.validate(realtimeTableConfig, schema, typesToSkip);
}
if (offlineTableConfig != null && realtimeTableConfig != null) {
TableConfigUtils.verifyHybridTableConfigs(rawTableName, offlineTableConfig, realtimeTableConfig);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 3425bf6..60f6bda 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -22,6 +22,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableSet;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -77,10 +79,18 @@ public final class TableConfigUtils {
private static final Logger LOGGER = LoggerFactory.getLogger(TableConfigUtils.class);
private static final String SCHEDULE_KEY = "schedule";
private static final String STAR_TREE_CONFIG_NAME = "StarTreeIndex Config";
+
// supported TableTaskTypes, must be identical to the one return in the impl of {@link PinotTaskGenerator}.
private static final String REALTIME_TO_OFFLINE_TASK_TYPE = "RealtimeToOfflineSegmentsTask";
/**
+ * @see TableConfigUtils#validate(TableConfig, Schema, String)
+ */
+ public static void validate(TableConfig tableConfig, @Nullable Schema schema) {
+ validate(tableConfig, schema, null);
+ }
+
+ /**
* Performs table config validations. Includes validations for the following:
* 1. Validation config
* 2. IngestionConfig
@@ -90,20 +100,34 @@ public final class TableConfigUtils {
*
* TODO: Add more validations for each section (e.g. validate conditions are met for aggregateMetrics)
*/
- public static void validate(TableConfig tableConfig, @Nullable Schema schema) {
+ public static void validate(TableConfig tableConfig, @Nullable Schema schema, @Nullable String typesToSkip) {
+ Set<ValidationType> skipTypes = parseTypesToSkipString(typesToSkip);
if (tableConfig.getTableType() == TableType.REALTIME) {
Preconditions.checkState(schema != null, "Schema should not be null for REALTIME table");
}
// Sanitize the table config before validation
sanitize(tableConfig);
- validateValidationConfig(tableConfig, schema);
- validateIngestionConfig(tableConfig, schema);
- validateTierConfigList(tableConfig.getTierConfigsList());
- validateIndexingConfig(tableConfig.getIndexingConfig(), schema);
- validateFieldConfigList(tableConfig.getFieldConfigList(), tableConfig.getIndexingConfig(), schema);
- validateUpsertConfig(tableConfig, schema);
- validatePartialUpsertStrategies(tableConfig, schema);
- validateTaskConfigs(tableConfig, schema);
+ // skip all validation if skip type ALL is selected.
+ if (!skipTypes.contains(ValidationType.ALL)) {
+ validateValidationConfig(tableConfig, schema);
+ validateIngestionConfig(tableConfig, schema);
+ validateTierConfigList(tableConfig.getTierConfigsList());
+ validateIndexingConfig(tableConfig.getIndexingConfig(), schema);
+ validateFieldConfigList(tableConfig.getFieldConfigList(), tableConfig.getIndexingConfig(), schema);
+ if (!skipTypes.contains(ValidationType.UPSERT)) {
+ validateUpsertConfig(tableConfig, schema);
+ validatePartialUpsertStrategies(tableConfig, schema);
+ }
+ if (!skipTypes.contains(ValidationType.TASK)) {
+ validateTaskConfigs(tableConfig, schema);
+ }
+ }
+ }
+
+ private static Set<ValidationType> parseTypesToSkipString(@Nullable String typesToSkip) {
+ return typesToSkip == null ? Collections.emptySet() : Arrays.stream(typesToSkip.split(","))
+ .map(s -> ValidationType.valueOf(s.toUpperCase()))
+ .collect(Collectors.toSet());
}
/**
@@ -331,7 +355,7 @@ public final class TableConfigUtils {
if (taskTypeConfigName.equals(REALTIME_TO_OFFLINE_TASK_TYPE)) {
// check table is not upsert
Preconditions.checkState(tableConfig.getUpsertMode() == UpsertConfig.Mode.NONE,
- "RealtimeToOfflineTask doesn't support upsert ingestion mode!");
+ "RealtimeToOfflineTask doesn't support upsert table!");
// check no malformed period
TimeUtils.convertPeriodToMillis(taskTypeConfig.getOrDefault("bufferTimePeriod", "2d"));
TimeUtils.convertPeriodToMillis(taskTypeConfig.getOrDefault("bucketTimePeriod", "1d"));
@@ -828,4 +852,9 @@ public final class TableConfigUtils {
rawTableName, offlineTimeColumnName, realtimeTimeColumnName));
}
}
+
+ // enum of all the skip-able validation types.
+ public enum ValidationType {
+ ALL, TASK, UPSERT
+ }
}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 3ff9c06..eacb587 100644
--- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -1191,7 +1191,10 @@ public class TableConfigUtilsTest {
@Test
public void testTaskConfig() {
Schema schema =
- new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
+ new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
+ .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
.build();
Map<String, String> realtimeToOfflineTaskConfig =
ImmutableMap.of("schedule", "0 */10 * ? * * *", "bucketTimePeriod", "6h", "bufferTimePeriod", "5d", "mergeType",
@@ -1218,7 +1221,8 @@ public class TableConfigUtilsTest {
}
// invalid Upsert config with RealtimeToOfflineTask
- tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL, null, null, null)).setTaskConfig(new TableTaskConfig(
ImmutableMap.of("RealtimeToOfflineSegmentsTask", realtimeToOfflineTaskConfig,
"SegmentGenerationAndPushTask", segmentGenerationAndPushTaskConfig))).build();
@@ -1226,8 +1230,10 @@ public class TableConfigUtilsTest {
TableConfigUtils.validateTaskConfigs(tableConfig, schema);
Assert.fail();
} catch (IllegalStateException e) {
- Assert.assertTrue(e.getMessage().contains("RealtimeToOfflineTask doesn't support upsert ingestion mode"));
+ Assert.assertTrue(e.getMessage().contains("RealtimeToOfflineTask doesn't support upsert table"));
}
+ // validate that TASK config will be skipped with skip string.
+ TableConfigUtils.validate(tableConfig, schema, "TASK,UPSERT");
// invalid period
HashMap<String, String> invalidPeriodConfig = new HashMap<>(realtimeToOfflineTaskConfig);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org