This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 d3893e27874 Parse validationTypesToSkip once and apply it consistently
(#20019)
d3893e27874 is described below
commit d3893e278740913c7733dd67d65db4e0d5c2e46c
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Oct 23 16:44:43 2026 -0700
Parse validationTypesToSkip once and apply it consistently (#20019)
---
.../pinot/client/admin/TableAdminClient.java | 6 +-
.../api/resources/PinotTableRestletResource.java | 29 +-
.../api/resources/TableConfigValidationUtils.java | 51 ++-
.../api/resources/TableConfigsRestletResource.java | 43 +--
.../pinot/controller/util/TaskConfigUtils.java | 24 +-
.../api/TableConfigsRestletResourceTest.java | 7 -
.../pinot/controller/util/TaskConfigUtilsTest.java | 16 +
.../segment/local/utils/TableConfigUtils.java | 424 +++++++++++----------
.../segment/local/utils/TableConfigUtilsTest.java | 200 ++++++----
9 files changed, 454 insertions(+), 346 deletions(-)
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
index da5adfe235f..7f64c2533ea 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
@@ -878,7 +878,7 @@ public class TableAdminClient extends
BaseServiceAdminClient {
/// Creates TableConfigs (schema + table configs). Supports skipping
validations and ignoring active tasks.
///
/// @param tableConfigsJson TableConfigs JSON payload
- /// @param validationTypesToSkip Optional comma-separated validation types
to skip (ALL|TASK|UPSERT)
+ /// @param validationTypesToSkip Optional comma-separated validation types
to skip
/// @param ignoreActiveTasks Optional flag to ignore active tasks during
creation
public String createTableConfigs(String tableConfigsJson, @Nullable String
validationTypesToSkip,
@Nullable Boolean ignoreActiveTasks)
@@ -902,7 +902,7 @@ public class TableAdminClient extends
BaseServiceAdminClient {
///
/// @param tableName Raw table name
/// @param tableConfigsJson Updated TableConfigs JSON payload
- /// @param validationTypesToSkip Optional comma-separated validation types
to skip (ALL|TASK|UPSERT)
+ /// @param validationTypesToSkip Optional comma-separated validation types
to skip
/// @param reload Whether to reload the table if schema change is backward
compatible
/// @param forceTableSchemaUpdate Whether to force schema update
public String updateTableConfigs(String tableName, String tableConfigsJson,
@Nullable String validationTypesToSkip,
@@ -937,7 +937,7 @@ public class TableAdminClient extends
BaseServiceAdminClient {
/// Validates a TableConfigs payload without applying it.
///
/// @param tableConfigsJson TableConfigs JSON payload
- /// @param validationTypesToSkip Optional comma-separated validation types
to skip (ALL|TASK|UPSERT)
+ /// @param validationTypesToSkip Optional comma-separated validation types
to skip
public String validateTableConfigs(String tableConfigsJson, @Nullable String
validationTypesToSkip)
throws PinotAdminException {
return validateTableConfigs(tableConfigsJson, validationTypesToSkip, null);
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 3a3ad2c7a87..1046b2a17c8 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
@@ -120,6 +120,7 @@ import org.apache.pinot.core.auth.Authorize;
import org.apache.pinot.core.auth.ManualAuthorization;
import org.apache.pinot.core.auth.TargetType;
import org.apache.pinot.segment.local.utils.TableConfigUtils;
+import org.apache.pinot.segment.local.utils.TableConfigUtils.ValidationType;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableConfigValidatorRegistry;
import org.apache.pinot.spi.config.table.TableStatsHumanReadable;
@@ -217,11 +218,12 @@ public class PinotTableRestletResource {
@ApiOperation(value = "Adds a table", notes = "Adds a table")
@ManualAuthorization // performed after parsing table configs
public ConfigSuccessResponse addTable(String tableConfigStr,
- @ApiParam(value = "comma separated list of validation type(s) to skip.
supported types: (ALL|TASK|UPSERT)")
+ @ApiParam(value =
TableConfigValidationUtils.VALIDATION_TYPES_TO_SKIP_DESCRIPTION)
@QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@DefaultValue("false") @QueryParam("ignoreActiveTasks") boolean
ignoreActiveTasks,
@Context HttpHeaders httpHeaders, @Context Request request)
throws IOException {
+ EnumSet<ValidationType> skipTypes =
TableConfigValidationUtils.parseValidationTypesToSkip(typesToSkip);
// TODO introduce a table config ctor with json string.
Pair<TableConfig, Map<String, Object>>
tableConfigAndUnrecognizedProperties;
TableConfig tableConfig;
@@ -251,7 +253,7 @@ public class PinotTableRestletResource {
TableConfigTunerUtils.applyTunerConfigs(_pinotHelixResourceManager,
tableConfig, schema, Map.of());
TableConfigValidationUtils.validateTableConfig(
- tableConfig, schema, typesToSkip, _pinotHelixResourceManager,
_controllerConf, _pinotTaskManager);
+ tableConfig, schema, skipTypes, _pinotHelixResourceManager,
_controllerConf, _pinotTaskManager);
} catch (TableAlreadyExistsException e) {
throw new ControllerApplicationException(LOGGER, e.getMessage(),
Response.Status.CONFLICT, e);
} catch (Exception e) {
@@ -761,13 +763,14 @@ public class PinotTableRestletResource {
@ApiOperation(value = "Updates table config for a table", notes = "Updates
table config for a table")
public ConfigSuccessResponse updateTableConfig(
@ApiParam(value = "Name of the table to update", required = true)
@PathParam("tableName") String tableName,
- @ApiParam(value = "comma separated list of validation type(s) to skip.
supported types: (ALL|TASK|UPSERT)")
+ @ApiParam(value =
TableConfigValidationUtils.VALIDATION_TYPES_TO_SKIP_DESCRIPTION)
@QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@ApiParam(value = "Force config changes")
@QueryParam("force") @DefaultValue("false") boolean force,
@Context HttpHeaders headers,
String tableConfigString)
throws Exception {
+ EnumSet<ValidationType> skipTypes =
TableConfigValidationUtils.parseValidationTypesToSkip(typesToSkip);
Pair<TableConfig, Map<String, Object>>
tableConfigAndUnrecognizedProperties;
TableConfig tableConfig;
String tableNameWithType;
@@ -788,9 +791,8 @@ public class PinotTableRestletResource {
schema = _pinotHelixResourceManager.getTableSchema(tableNameWithType);
Preconditions.checkState(schema != null, "Failed to find schema for
table: %s", tableNameWithType);
- TableConfigValidationUtils.validateTableConfig(
- tableConfig, schema, typesToSkip, _pinotHelixResourceManager,
_controllerConf, _pinotTaskManager,
- _pinotHelixResourceManager.getTableConfig(tableNameWithType));
+ TableConfigValidationUtils.validateTableConfig(tableConfig, schema,
skipTypes, _pinotHelixResourceManager,
+ _controllerConf, _pinotTaskManager,
_pinotHelixResourceManager.getTableConfig(tableNameWithType));
} 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);
@@ -827,7 +829,7 @@ public class PinotTableRestletResource {
+ " This allows us to validate table config before apply.")
@ManualAuthorization // performed after parsing TableConfig
public ObjectNode checkTableConfig(String tableConfigStr,
- @ApiParam(value = "comma separated list of validation type(s) to skip.
supported types: (ALL|TASK|UPSERT)")
+ @ApiParam(value =
TableConfigValidationUtils.VALIDATION_TYPES_TO_SKIP_DESCRIPTION)
@QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@Context HttpHeaders httpHeaders,
@Context Request request) {
Pair<TableConfig, Map<String, Object>>
tableConfigAndUnrecognizedProperties;
@@ -843,26 +845,27 @@ public class PinotTableRestletResource {
tableConfig.setTableName(tableNameWithType);
// validate permission
- ResourceUtils.checkPermissionAndAccess(tableNameWithType, request,
httpHeaders,
- AccessType.READ, Actions.Table.VALIDATE_TABLE_CONFIGS,
_accessControlFactory, LOGGER);
+ ResourceUtils.checkPermissionAndAccess(tableNameWithType, request,
httpHeaders, AccessType.READ,
+ Actions.Table.VALIDATE_TABLE_CONFIGS, _accessControlFactory, LOGGER);
- ObjectNode validationResponse = validateConfig(tableConfig, typesToSkip);
+ ObjectNode validationResponse =
+ validateConfig(tableConfig,
TableConfigValidationUtils.parseValidationTypesToSkip(typesToSkip));
validationResponse.set("unrecognizedProperties",
JsonUtils.objectToJsonNode(tableConfigAndUnrecognizedProperties.getRight()));
return validationResponse;
}
@VisibleForTesting
- ObjectNode validateConfig(TableConfig tableConfig, @Nullable String
typesToSkip) {
+ ObjectNode validateConfig(TableConfig tableConfig, @Nullable
EnumSet<ValidationType> skipTypes) {
String tableNameWithType = tableConfig.getTableName();
try {
Schema schema =
_pinotHelixResourceManager.getTableSchema(tableNameWithType);
if (schema == null) {
throw new SchemaNotFoundException("Failed to find schema for table: "
+ tableNameWithType);
}
- TableConfigUtils.validate(tableConfig, schema, typesToSkip,
+ TableConfigUtils.validate(tableConfig, schema, skipTypes,
_pinotHelixResourceManager.getTableConfig(tableNameWithType));
- TaskConfigUtils.validateTaskConfigs(tableConfig, schema,
_pinotTaskManager, typesToSkip);
+ TaskConfigUtils.validateTaskConfigs(tableConfig, schema,
_pinotTaskManager, skipTypes);
TableConfigValidatorRegistry.validate(tableConfig, schema);
ObjectNode tableConfigValidateStr = JsonUtils.newObjectNode();
if (tableConfig.getTableType() == TableType.OFFLINE) {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
index 19692575c01..ef68c576a76 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigValidationUtils.java
@@ -20,19 +20,24 @@ package org.apache.pinot.controller.api.resources;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
+import javax.ws.rs.core.Response;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.assignment.InstanceAssignmentConfigUtils;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.assignment.InstancePartitionsUtils;
import org.apache.pinot.controller.ControllerConf;
+import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import
org.apache.pinot.controller.helix.core.assignment.instance.InstanceAssignmentUtils;
import org.apache.pinot.controller.helix.core.minion.PinotTaskManager;
import org.apache.pinot.controller.util.TaskConfigUtils;
import org.apache.pinot.segment.local.utils.TableConfigUtils;
+import org.apache.pinot.segment.local.utils.TableConfigUtils.ValidationType;
import org.apache.pinot.spi.config.ConfigUtils;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableConfigValidatorRegistry;
@@ -40,6 +45,8 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/// Utility class that encapsulates table config validation logic.
@@ -52,31 +59,55 @@ public final class TableConfigValidationUtils {
private TableConfigValidationUtils() {
}
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TableConfigValidationUtils.class);
+
+ /// Description of the `validationTypesToSkip` request parameter parsed by
[#parseValidationTypesToSkip].
+ public static final String VALIDATION_TYPES_TO_SKIP_DESCRIPTION =
+ "Comma separated list of validation type(s) to skip. Supported types: "
+ +
"(ALL|TASK|UPSERT|DEDUP|INSTANCE_ASSIGNMENT|TENANT|MINION_INSTANCES). TENANT
and MINION_INSTANCES apply "
+ + "only to validating and tuning table configs, as creating or
updating a table always validates them";
+
+ /// Parses the `validationTypesToSkip` request parameter, rejecting an
unknown validation type as a bad request.
+ /// Returns `null` when the parameter is not set or blank.
+ @Nullable
+ public static EnumSet<ValidationType> parseValidationTypesToSkip(@Nullable
String typesToSkip) {
+ try {
+ return TableConfigUtils.parseTypesToSkipString(typesToSkip);
+ } catch (IllegalArgumentException e) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Invalid validationTypesToSkip: %s, supported types:
%s", typesToSkip,
+ Arrays.toString(ValidationType.values())),
Response.Status.BAD_REQUEST, e);
+ }
+ }
+
/// Validates a table config against the given schema and controller
configuration.
///
/// @param tableConfig the table config to validate
/// @param schema the schema for the table (must not be null)
- /// @param typesToSkip comma-separated list of validation types to skip
(ALL|TASK|UPSERT), or null
+ /// @param skipTypes the validation types to skip, or null to skip none,
see [ValidationType]
/// @param resourceManager the Helix resource manager
/// @param controllerConf the controller configuration
/// @param taskManager the task manager, or null to skip task validation
- public static void validateTableConfig(TableConfig tableConfig, Schema
schema, @Nullable String typesToSkip,
- PinotHelixResourceManager resourceManager, ControllerConf controllerConf,
- @Nullable PinotTaskManager taskManager) {
- validateTableConfig(tableConfig, schema, typesToSkip, resourceManager,
controllerConf, taskManager, null);
+ public static void validateTableConfig(TableConfig tableConfig, Schema
schema,
+ @Nullable EnumSet<ValidationType> skipTypes, PinotHelixResourceManager
resourceManager,
+ ControllerConf controllerConf, @Nullable PinotTaskManager taskManager) {
+ validateTableConfig(tableConfig, schema, skipTypes, resourceManager,
controllerConf, taskManager, null);
}
- public static void validateTableConfig(TableConfig tableConfig, Schema
schema, @Nullable String typesToSkip,
- PinotHelixResourceManager resourceManager, ControllerConf
controllerConf, @Nullable PinotTaskManager taskManager,
+ public static void validateTableConfig(TableConfig tableConfig, Schema
schema,
+ @Nullable EnumSet<ValidationType> skipTypes, PinotHelixResourceManager
resourceManager,
+ ControllerConf controllerConf, @Nullable PinotTaskManager taskManager,
@Nullable TableConfig existingTableConfig) {
validateEnvironmentVariables(tableConfig);
- TableConfigUtils.validate(tableConfig, schema, typesToSkip,
existingTableConfig);
+ TableConfigUtils.validate(tableConfig, schema, skipTypes,
existingTableConfig);
TableConfigUtils.validateTableName(tableConfig);
TableConfigUtils.ensureMinReplicas(tableConfig,
controllerConf.getDefaultTableMinReplicas());
TableConfigUtils.ensureStorageQuotaConstraints(tableConfig,
controllerConf.getDimTableMaxSize());
checkHybridTableConfig(resourceManager, tableConfig);
- TaskConfigUtils.validateTaskConfigs(tableConfig, schema, taskManager,
typesToSkip);
- validateInstanceAssignment(resourceManager, tableConfig);
+ TaskConfigUtils.validateTaskConfigs(tableConfig, schema, taskManager,
skipTypes);
+ if (!ValidationType.INSTANCE_ASSIGNMENT.isSkipped(skipTypes)) {
+ validateInstanceAssignment(resourceManager, tableConfig);
+ }
resourceManager.validateTableTenantConfig(tableConfig);
resourceManager.validateTableTaskMinionInstanceTagConfig(tableConfig);
TableConfigValidatorRegistry.validate(tableConfig, schema);
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 579322765a1..825feaaed31 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
@@ -30,9 +30,9 @@ import io.swagger.annotations.SecurityDefinition;
import io.swagger.annotations.SwaggerDefinition;
import java.io.IOException;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import javax.annotation.Nullable;
import javax.inject.Inject;
import javax.ws.rs.DELETE;
@@ -76,6 +76,7 @@ import org.apache.pinot.core.auth.ManualAuthorization;
import org.apache.pinot.core.auth.TargetType;
import org.apache.pinot.segment.local.utils.SchemaUtils;
import org.apache.pinot.segment.local.utils.TableConfigUtils;
+import org.apache.pinot.segment.local.utils.TableConfigUtils.ValidationType;
import org.apache.pinot.spi.config.TableConfigs;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableConfigValidatorRegistry;
@@ -190,11 +191,12 @@ public class TableConfigsRestletResource {
@ManualAuthorization // performed after parsing table configs
public ConfigSuccessResponse addConfig(
String tableConfigsStr,
- @ApiParam(value = "comma separated list of validation type(s) to skip.
supported types: (ALL|TASK|UPSERT)")
+ @ApiParam(value =
TableConfigValidationUtils.VALIDATION_TYPES_TO_SKIP_DESCRIPTION)
@QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@DefaultValue("false") @QueryParam("ignoreActiveTasks") boolean
ignoreActiveTasks,
@Context HttpHeaders httpHeaders, @Context Request request)
throws Exception {
+ EnumSet<ValidationType> skipTypes =
TableConfigValidationUtils.parseValidationTypesToSkip(typesToSkip);
Pair<TableConfigs, Map<String, Object>> tableConfigsAndUnrecognizedProps;
try {
tableConfigsAndUnrecognizedProps =
@@ -213,7 +215,7 @@ public class TableConfigsRestletResource {
Response.Status.BAD_REQUEST);
}
- validateConfig(tableConfigs, databaseName, typesToSkip);
+ validateConfig(tableConfigs, databaseName, skipTypes);
tableConfigs.setTableName(rawTableName);
TableConfig offlineTableConfig = tableConfigs.getOffline();
@@ -353,13 +355,14 @@ public class TableConfigsRestletResource {
public ConfigSuccessResponse 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)")
+ @ApiParam(value =
TableConfigValidationUtils.VALIDATION_TYPES_TO_SKIP_DESCRIPTION)
@QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@ApiParam(value = "Reload the table if the new schema is backward
compatible") @DefaultValue("false")
@QueryParam("reload") boolean reload,
@ApiParam(value = "Force update the table schema")
@DefaultValue("false") @QueryParam("forceTableSchemaUpdate")
boolean forceTableSchemaUpdate, String tableConfigsStr, @Context
HttpHeaders headers)
throws Exception {
+ EnumSet<ValidationType> skipTypes =
TableConfigValidationUtils.parseValidationTypesToSkip(typesToSkip);
String databaseName =
DatabaseUtils.extractDatabaseFromHttpHeaders(headers);
tableName = DatabaseUtils.translateTableName(tableName, databaseName);
Pair<TableConfigs, Map<String, Object>> tableConfigsAndUnrecognizedProps;
@@ -368,7 +371,7 @@ public class TableConfigsRestletResource {
tableConfigsAndUnrecognizedProps =
JsonUtils.stringToObjectAndUnrecognizedProperties(tableConfigsStr,
TableConfigs.class);
tableConfigs = tableConfigsAndUnrecognizedProps.getLeft();
- validateConfig(tableConfigs, databaseName, typesToSkip);
+ validateConfig(tableConfigs, databaseName, skipTypes);
Preconditions.checkState(
DatabaseUtils.translateTableName(tableConfigs.getTableName(),
databaseName).equals(tableName),
"'tableName' in TableConfigs: %s must match provided tableName: %s",
tableConfigs.getTableName(), tableName);
@@ -440,8 +443,7 @@ public class TableConfigsRestletResource {
@ApiOperation(value = "Validate the TableConfigs", notes = "Validate the
TableConfigs")
@ManualAuthorization // performed after parsing TableConfigs
public String validateConfig(String tableConfigsStr,
- @ApiParam(value = "comma separated list of validation type(s) to skip.
supported types: "
- + "(ALL|TASK|UPSERT|TENANT|MINION_INSTANCES)")
+ @ApiParam(value =
TableConfigValidationUtils.VALIDATION_TYPES_TO_SKIP_DESCRIPTION)
@QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@Context HttpHeaders httpHeaders,
@Context Request request) {
Pair<TableConfigs, Map<String, Object>> tableConfigsAndUnrecognizedProps =
@@ -462,8 +464,7 @@ public class TableConfigsRestletResource {
+ "returning the result that would be stored on create/update")
@ManualAuthorization // performed after parsing TableConfigs
public String tuneConfig(String tableConfigsStr,
- @ApiParam(value = "comma separated list of validation type(s) to skip.
supported types: "
- + "(ALL|TASK|UPSERT|TENANT|MINION_INSTANCES)")
+ @ApiParam(value =
TableConfigValidationUtils.VALIDATION_TYPES_TO_SKIP_DESCRIPTION)
@QueryParam("validationTypesToSkip") @Nullable String typesToSkip,
@Context HttpHeaders httpHeaders,
@Context Request request) {
Pair<TableConfigs, Map<String, Object>> tableConfigsAndUnrecognizedProps =
@@ -483,6 +484,7 @@ public class TableConfigsRestletResource {
private Pair<TableConfigs, Map<String, Object>>
parseAndValidateTableConfigs(String tableConfigsStr,
@Nullable String typesToSkip, HttpHeaders httpHeaders, Request request) {
+ EnumSet<ValidationType> skipTypes =
TableConfigValidationUtils.parseValidationTypesToSkip(typesToSkip);
Pair<TableConfigs, Map<String, Object>> tableConfigsAndUnrecognizedProps;
try {
tableConfigsAndUnrecognizedProps =
@@ -494,7 +496,7 @@ public class TableConfigsRestletResource {
}
String databaseName =
DatabaseUtils.extractDatabaseFromHttpHeaders(httpHeaders);
TableConfigs tableConfigs = tableConfigsAndUnrecognizedProps.getLeft();
- validateConfig(tableConfigs, databaseName, typesToSkip);
+ validateConfig(tableConfigs, databaseName, skipTypes);
String rawTableName =
DatabaseUtils.translateTableName(tableConfigs.getTableName(), databaseName);
tableConfigs.setTableName(rawTableName);
@@ -503,7 +505,6 @@ public class TableConfigsRestletResource {
// equivalent checks inline or via PinotHelixResourceManager). Active-task
validation is intentionally excluded
// here: it applies only on the create/update path (gated by the
ignoreActiveTasks flag) so that validate/tune of an
// existing table with running tasks is not blocked.
- Set<TableConfigUtils.ValidationType> skipTypes =
TableConfigUtils.parseTypesToSkipString(typesToSkip);
try {
if (tableConfigs.getOffline() != null) {
validateClusterAwareConfig(tableConfigs.getOffline(), skipTypes);
@@ -529,14 +530,11 @@ public class TableConfigsRestletResource {
return tableConfigsAndUnrecognizedProps;
}
- private void validateClusterAwareConfig(TableConfig tableConfig,
Set<TableConfigUtils.ValidationType> skipTypes) {
- if (skipTypes.contains(TableConfigUtils.ValidationType.ALL)) {
- return;
- }
- if (!skipTypes.contains(TableConfigUtils.ValidationType.TENANT)) {
+ private void validateClusterAwareConfig(TableConfig tableConfig, @Nullable
EnumSet<ValidationType> skipTypes) {
+ if (!ValidationType.TENANT.isSkipped(skipTypes)) {
_pinotHelixResourceManager.validateTableTenantConfig(tableConfig);
}
- if (!skipTypes.contains(TableConfigUtils.ValidationType.MINION_INSTANCES))
{
+ if (!ValidationType.MINION_INSTANCES.isSkipped(skipTypes)) {
_pinotHelixResourceManager.validateTableTaskMinionInstanceTagConfig(tableConfig);
}
}
@@ -550,7 +548,8 @@ public class TableConfigsRestletResource {
/// Validates the provided TableConfigs. Hybrid table validation is
performed only on the provided
/// configs and does not check for conflicts with existing tables in the
cluster.
- private void validateConfig(TableConfigs tableConfigs, String database,
@Nullable String typesToSkip) {
+ private void validateConfig(TableConfigs tableConfigs, String database,
+ @Nullable EnumSet<ValidationType> skipTypes) {
String rawTableName =
DatabaseUtils.translateTableName(tableConfigs.getTableName(), database);
TableConfig offlineTableConfig = tableConfigs.getOffline();
TableConfig realtimeTableConfig = tableConfigs.getRealtime();
@@ -573,9 +572,9 @@ 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, typesToSkip,
+ TableConfigUtils.validate(offlineTableConfig, schema, skipTypes,
_pinotHelixResourceManager.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(rawTableName)));
- TaskConfigUtils.validateTaskConfigs(tableConfigs.getOffline(), schema,
_pinotTaskManager, typesToSkip);
+ TaskConfigUtils.validateTaskConfigs(tableConfigs.getOffline(), schema,
_pinotTaskManager, skipTypes);
TableConfigValidatorRegistry.validate(offlineTableConfig, schema);
}
if (realtimeTableConfig != null) {
@@ -584,9 +583,9 @@ public class TableConfigsRestletResource {
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, typesToSkip,
+ TableConfigUtils.validate(realtimeTableConfig, schema, skipTypes,
_pinotHelixResourceManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(rawTableName)));
- TaskConfigUtils.validateTaskConfigs(tableConfigs.getRealtime(),
schema, _pinotTaskManager, typesToSkip);
+ TaskConfigUtils.validateTaskConfigs(tableConfigs.getRealtime(),
schema, _pinotTaskManager, skipTypes);
TableConfigValidatorRegistry.validate(realtimeTableConfig, schema);
}
if (offlineTableConfig != null && realtimeTableConfig != null) {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TaskConfigUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TaskConfigUtils.java
index 53ba3009347..ca7ec801055 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TaskConfigUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TaskConfigUtils.java
@@ -20,10 +20,13 @@ package org.apache.pinot.controller.util;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import java.util.EnumSet;
import java.util.Map;
+import javax.annotation.Nullable;
import org.apache.pinot.controller.helix.core.minion.PinotTaskManager;
import
org.apache.pinot.controller.helix.core.minion.generator.PinotTaskGenerator;
import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorRegistry;
+import org.apache.pinot.segment.local.utils.TableConfigUtils.ValidationType;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
import org.apache.pinot.spi.data.Schema;
@@ -33,22 +36,21 @@ import org.slf4j.LoggerFactory;
public class TaskConfigUtils {
- private static final Logger LOGGER =
LoggerFactory.getLogger(TaskConfigUtils.class);
- private static final String SCHEDULE_KEY = "schedule";
- private static final String ALL_VALIDATION_TYPE = "ALL";
- private static final String TASK_VALIDATION_TYPE = "TASK";
-
private TaskConfigUtils() {
}
- public static void validateTaskConfigs(TableConfig tableConfig, Schema
schema, PinotTaskManager pinotTaskManager,
- String validationTypesToSkip) {
- if (tableConfig == null || tableConfig.getTaskConfig() == null) {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TaskConfigUtils.class);
+ private static final String SCHEDULE_KEY = "schedule";
+
+ /// Validates the task configs of the table with the task generators of the
task manager. Skips the validation when
+ /// the table has no task config, the task manager is `null`, or the task
validation is skipped.
+ public static void validateTaskConfigs(TableConfig tableConfig, Schema
schema,
+ @Nullable PinotTaskManager pinotTaskManager, @Nullable
EnumSet<ValidationType> skipTypes) {
+ if (tableConfig == null || tableConfig.getTaskConfig() == null ||
pinotTaskManager == null) {
return;
}
- if (ALL_VALIDATION_TYPE.equalsIgnoreCase(validationTypesToSkip) ||
TASK_VALIDATION_TYPE.equalsIgnoreCase(
- validationTypesToSkip)) {
- LOGGER.info("Skipping task validation as validationTypesToSkip is set
to: {}", validationTypesToSkip);
+ if (ValidationType.TASK.isSkipped(skipTypes)) {
+ LOGGER.info("Skipping task validation as validationTypesToSkip is set
to: {}", skipTypes);
return;
}
TableTaskConfig taskConfig = tableConfig.getTaskConfig();
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
index f3b332271c1..f1577d559ca 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
@@ -894,13 +894,6 @@ public class TableConfigsRestletResourceTest extends
ControllerTest {
adminClient.getTableClient().validateTableConfigs(tableConfigs.toPrettyJsonString(),
"MINION_INSTANCES");
Assert.assertNotNull(responseWithMinionSkip);
- // ACTIVE_TASKS is still accepted as a skip type for backward
compatibility, but the validate/tune preflight
- // endpoints no longer run active-task validation (it applies only on the
create/update path), so passing it is a
- // no-op here.
- String responseWithTasksSkip =
-
adminClient.getTableClient().validateTableConfigs(tableConfigs.toPrettyJsonString(),
"ACTIVE_TASKS");
- Assert.assertNotNull(responseWithTasksSkip);
-
// Test validation with multiple skip types
String responseWithMultipleSkips = adminClient.getTableClient()
.validateTableConfigs(tableConfigs.toPrettyJsonString(),
"TENANT,MINION_INSTANCES");
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/util/TaskConfigUtilsTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/util/TaskConfigUtilsTest.java
index bd06d422c55..2bdefb36ee9 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/util/TaskConfigUtilsTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/util/TaskConfigUtilsTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.util;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -26,6 +27,7 @@ import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator
import
org.apache.pinot.controller.helix.core.minion.generator.PinotTaskGenerator;
import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorRegistry;
import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.segment.local.utils.TableConfigUtils.ValidationType;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -91,6 +93,20 @@ public class TaskConfigUtilsTest {
TaskConfigUtils.validateTaskConfigs(tableConfig, new Schema(),
_mockTaskManager, null);
}
+ @Test
+ public void testValidateTableTaskConfigsSkipped() {
+ TableTaskConfig tableTaskConfig =
+ new TableTaskConfig(Map.of(TEST_TASK_TYPE, Map.of("schedule", "0 */10
* ? * * *")));
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(TEST_TABLE_NAME).setTaskConfig(tableTaskConfig).build();
+ // The task generator rejects every task config, so these pass only
because the task validation is skipped
+ TaskConfigUtils.validateTaskConfigs(tableConfig, new Schema(),
_mockTaskManager, EnumSet.of(ValidationType.TASK));
+ TaskConfigUtils.validateTaskConfigs(tableConfig, new Schema(),
_mockTaskManager,
+ EnumSet.of(ValidationType.UPSERT, ValidationType.TASK));
+ TaskConfigUtils.validateTaskConfigs(tableConfig, new Schema(),
_mockTaskManager, EnumSet.of(ValidationType.ALL));
+ TaskConfigUtils.validateTaskConfigs(tableConfig, new Schema(), null, null);
+ }
+
@Test
public void testCommonTaskValidations() {
// invalid schedule
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 eb07e0c0640..705b218b38f 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
@@ -23,9 +23,9 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
-import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
@@ -162,7 +162,7 @@ public final class TableConfigUtils {
_enforcePoolBasedAssignment = enforcePoolBasedAssignment;
}
- /// @see TableConfigUtils#validate(TableConfig, Schema, String)
+ /// @see TableConfigUtils#validate(TableConfig, Schema, EnumSet)
public static void validate(TableConfig tableConfig, Schema schema) {
validate(tableConfig, schema, null);
}
@@ -175,17 +175,18 @@ public final class TableConfigUtils {
/// 5. Field Config List
/// 6. Instance pool and replica group, if enabled
///
+ /// The validation types in `skipTypes` are skipped, or none when it is
`null`, see [ValidationType].
+ ///
/// TODO: Add more validations for each section (e.g. validate conditions
are met for aggregateMetrics)
- public static void validate(TableConfig tableConfig, Schema schema,
@Nullable String typesToSkip) {
- validate(tableConfig, schema, typesToSkip, null);
+ public static void validate(TableConfig tableConfig, Schema schema,
@Nullable EnumSet<ValidationType> skipTypes) {
+ validate(tableConfig, schema, skipTypes, null);
}
- public static void validate(TableConfig tableConfig, Schema schema,
@Nullable String typesToSkip,
+ public static void validate(TableConfig tableConfig, Schema schema,
@Nullable EnumSet<ValidationType> skipTypes,
@Nullable TableConfig existingTableConfig) {
Preconditions.checkArgument(schema != null, "Schema should not be null for
table: %s", tableConfig.getTableName());
- Set<ValidationType> skipTypes = parseTypesToSkipString(typesToSkip);
validateEffectiveTableConfig(tableConfig, schema, skipTypes,
existingTableConfig);
- if (skipTypes.contains(ValidationType.ALL) ||
!hasConsumingSegmentTierOverwriteForRealtimeTable(tableConfig)) {
+ if (ValidationType.ALL.isSkipped(skipTypes) ||
!hasConsumingSegmentTierOverwriteForRealtimeTable(tableConfig)) {
return;
}
try {
@@ -198,11 +199,11 @@ public final class TableConfigUtils {
}
private static void validateEffectiveTableConfig(TableConfig tableConfig,
Schema schema,
- Set<ValidationType> skipTypes, @Nullable TableConfig
existingTableConfig) {
+ @Nullable EnumSet<ValidationType> skipTypes, @Nullable TableConfig
existingTableConfig) {
// Sanitize the table config before validation
sanitize(tableConfig);
- if (skipTypes.contains(ValidationType.ALL)) {
+ if (ValidationType.ALL.isSkipped(skipTypes)) {
return;
}
@@ -229,8 +230,15 @@ public final class TableConfigUtils {
validateInstanceAssignmentConfigs(tableConfig);
validateStrictReplicaGroupRouting(tableConfig);
- if (!skipTypes.contains(ValidationType.UPSERT)) {
- validateUpsertAndDedupConfig(tableConfig, schema, existingTableConfig);
+ boolean upsertEnabled = tableConfig.isUpsertEnabled();
+ boolean dedupEnabled = tableConfig.isDedupEnabled();
+ Preconditions.checkState(!(upsertEnabled && dedupEnabled),
+ "A table can have either Upsert or Dedup enabled, but not both");
+ if (upsertEnabled && !ValidationType.UPSERT.isSkipped(skipTypes)) {
+ validateUpsertConfig(tableConfig, schema, existingTableConfig);
+ }
+ if (dedupEnabled && !ValidationType.DEDUP.isSkipped(skipTypes)) {
+ validateDedupConfig(tableConfig, schema);
}
validateTaskConfig(tableConfig);
@@ -278,14 +286,16 @@ public final class TableConfigUtils {
"Instance pool and replica group configurations must be enabled");
}
- public static Set<ValidationType> parseTypesToSkipString(@Nullable String
typesToSkipStr) {
+ /// Parses the comma separated validation types to skip, ignoring case and
surrounding whitespace. Returns `null` when
+ /// there is nothing to skip, that is when the input is `null` or blank.
+ @Nullable
+ public static EnumSet<ValidationType> parseTypesToSkipString(@Nullable
String typesToSkipStr) {
if (StringUtils.isBlank(typesToSkipStr)) {
- return Set.of();
+ return null;
}
- String[] split = StringUtils.split(typesToSkipStr, ',');
- Set<ValidationType> typesToSkip =
Sets.newHashSetWithExpectedSize(split.length);
- for (String type : split) {
- typesToSkip.add(ValidationType.valueOf(type.trim().toUpperCase()));
+ EnumSet<ValidationType> typesToSkip = EnumSet.noneOf(ValidationType.class);
+ for (String type : StringUtils.split(typesToSkipStr, ',')) {
+
typesToSkip.add(ValidationType.valueOf(type.trim().toUpperCase(Locale.ROOT)));
}
return typesToSkip;
}
@@ -1058,33 +1068,12 @@ public final class TableConfigUtils {
}
}
- /// Validates the upsert- and dedup-related configuration. Returns early
when neither is enabled; otherwise checks
- /// the shared requirements (upsert and dedup are mutually exclusive,
primary keys exist and are single-valued,
- /// strict replica-group routing, no tenant tag override for REALTIME,
OFFLINE restrictions) and then, per the enabled
- /// mode:
- /// - Upsert: no multi-tier / star-tree, comparison / delete-record /
out-of-order columns are valid, deleted-keys
- /// compaction consistency, post-partial-upsert transforms,
consistency-mode constraints, and delegates to
- /// [#validateTTLForUpsertConfig] and [#validatePartialUpsertStrategies];
rejects MD5 when disabled.
- /// - Dedup: delegates to [#validateTTLForDedupConfig]; rejects MD5 when
disabled.
+ /// Validates the requirements shared by upsert and dedup tables: the
primary key columns exist in the schema and are
+ /// single-valued, the routing uses strict replica-group, and a REALTIME
table has no tenant tag override.
///
/// The segment assignment requirements shared with strict replica-group
routing (one instance per partition, no
/// COMPLETED segment relocation) are validated by
[#validateStrictReplicaGroupRouting].
- @VisibleForTesting
- static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema
schema) {
- validateUpsertAndDedupConfig(tableConfig, schema, null);
- }
-
- static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema
schema,
- @Nullable TableConfig existingTableConfig) {
- boolean upsertEnabled = tableConfig.isUpsertEnabled();
- boolean dedupEnabled = tableConfig.isDedupEnabled();
- if (!upsertEnabled && !dedupEnabled) {
- return;
- }
-
- // check both upsert and dedup are not enabled simultaneously
- Preconditions.checkState(!(upsertEnabled && dedupEnabled),
- "A table can have either Upsert or Dedup enabled, but not both");
+ private static void validateUpsertAndDedupCommonConfig(TableConfig
tableConfig, Schema schema) {
// primary key exists
Preconditions.checkState(CollectionUtils.isNotEmpty(schema.getPrimaryKeyColumns()),
"Upsert/Dedup table must have primary key columns in the schema");
@@ -1100,18 +1089,30 @@ public final class TableConfigUtils {
Preconditions.checkState(
tableConfig.getRoutingConfig() != null &&
isRoutingStrategyAllowedForUpsert(tableConfig.getRoutingConfig()),
"Upsert/Dedup table must use strict replica-group (i.e.
strictReplicaGroup) based routing");
-
- UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
- DedupConfig dedupConfig = tableConfig.getDedupConfig();
if (tableConfig.getTableType() == TableType.REALTIME) {
Preconditions.checkState(tableConfig.getTenantConfig().getTagOverrideConfig()
== null || (
tableConfig.getTenantConfig().getTagOverrideConfig().getRealtimeConsuming() ==
null
&&
tableConfig.getTenantConfig().getTagOverrideConfig().getRealtimeCompleted() ==
null),
"Invalid tenant tag override used for Upsert/Dedup table");
- } else {
- Preconditions.checkState(!dedupEnabled,
- "Dedup is not supported for OFFLINE table. Only upsert is supported
for OFFLINE table");
- assert upsertConfig != null;
+ }
+ }
+
+ /// Validates the upsert config of a table with upsert enabled: checks the
requirements shared with dedup (see
+ /// [#validateUpsertAndDedupCommonConfig]), the OFFLINE restrictions (no
partial upsert, segment partition config,
+ /// comparison or time column), no multi-tier / star-tree, comparison /
delete-record / out-of-order columns are
+ /// valid, deleted-keys compaction consistency, post-partial-upsert
transforms, consistency-mode constraints, and
+ /// delegates to [#validateTTLForUpsertConfig] and
[#validatePartialUpsertStrategies]; rejects MD5 when disabled.
+ @VisibleForTesting
+ static void validateUpsertConfig(TableConfig tableConfig, Schema schema) {
+ validateUpsertConfig(tableConfig, schema, null);
+ }
+
+ static void validateUpsertConfig(TableConfig tableConfig, Schema schema,
@Nullable TableConfig existingTableConfig) {
+ UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
+ assert upsertConfig != null;
+ validateUpsertAndDedupCommonConfig(tableConfig, schema);
+
+ if (tableConfig.getTableType() == TableType.OFFLINE) {
Preconditions.checkState(upsertConfig.getMode() !=
UpsertConfig.Mode.PARTIAL,
"Partial upsert is not supported for OFFLINE table");
// Offline upsert tables require segment partition config so that
segments are assigned to servers
@@ -1126,174 +1127,168 @@ public final class TableConfigUtils {
// A comparison column is required, same as realtime upsert. It resolves
to the configured comparison columns, or
// the table's time column when comparison columns are not set. Unlike
realtime, offline tables do not require a
// time column, so reject the case where neither is configured (no
implicit segment creation time fallback).
- Preconditions.checkState(
- CollectionUtils.isNotEmpty(upsertConfig.getComparisonColumns())
+
Preconditions.checkState(CollectionUtils.isNotEmpty(upsertConfig.getComparisonColumns())
|| tableConfig.getValidationConfig().getTimeColumnName() != null,
"Offline upsert table must have a comparison column or a time column
configured");
}
- if (upsertEnabled) {
- // specifically for upsert
- assert upsertConfig != null;
-
- // Currently, only one tier is allowed for upsert table, as the
committed segments can't be moved to other tiers.
- Preconditions.checkState(tableConfig.getTierConfigsList() == null, "The
upsert table cannot have multi-tiers");
- // no startree index
-
Preconditions.checkState(CollectionUtils.isEmpty(tableConfig.getIndexingConfig().getStarTreeIndexConfigs())
- && !tableConfig.getIndexingConfig().isEnableDefaultStarTree(),
- "The upsert table cannot have star-tree index.");
+ // Currently, only one tier is allowed for upsert table, as the committed
segments can't be moved to other tiers.
+ Preconditions.checkState(tableConfig.getTierConfigsList() == null, "The
upsert table cannot have multi-tiers");
+ // no startree index
+
Preconditions.checkState(CollectionUtils.isEmpty(tableConfig.getIndexingConfig().getStarTreeIndexConfigs())
+ && !tableConfig.getIndexingConfig().isEnableDefaultStarTree(), "The
upsert table cannot have star-tree index.");
- // comparison column exists
- List<String> comparisonColumns = upsertConfig.getComparisonColumns();
- if (comparisonColumns != null) {
- for (String column : comparisonColumns) {
- Preconditions.checkState(schema.hasColumn(column), "The comparison
column does not exist on schema");
- }
+ // comparison column exists
+ List<String> comparisonColumns = upsertConfig.getComparisonColumns();
+ if (comparisonColumns != null) {
+ for (String column : comparisonColumns) {
+ Preconditions.checkState(schema.hasColumn(column), "The comparison
column does not exist on schema");
}
+ }
- // Delete record column exist and is a BOOLEAN field
- String deleteRecordColumn = upsertConfig.getDeleteRecordColumn();
- if (deleteRecordColumn != null) {
- FieldSpec fieldSpec = schema.getFieldSpecFor(deleteRecordColumn);
- Preconditions.checkState(fieldSpec != null,
- String.format("Column %s specified in deleteRecordColumn does not
exist", deleteRecordColumn));
- Preconditions.checkState(fieldSpec.isSingleValueField(),
- String.format("The deleteRecordColumn - %s must be a single-valued
column", deleteRecordColumn));
- DataType dataType = fieldSpec.getDataType();
- Preconditions.checkState(dataType == DataType.BOOLEAN || dataType ==
DataType.STRING || dataType.isNumeric(),
- String.format("The deleteRecordColumn - %s must be of type: String
/ Boolean / Numeric",
- deleteRecordColumn));
- }
-
- String outOfOrderRecordColumn = upsertConfig.getOutOfOrderRecordColumn();
- if (outOfOrderRecordColumn != null) {
- Preconditions.checkState(!upsertConfig.isDropOutOfOrderRecord(),
- "outOfOrderRecordColumn and dropOutOfOrderRecord shouldn't exist
together for upsert table");
- FieldSpec fieldSpec = schema.getFieldSpecFor(outOfOrderRecordColumn);
- Preconditions.checkState(
- fieldSpec != null && fieldSpec.isSingleValueField() &&
fieldSpec.getDataType() == DataType.BOOLEAN,
- "The outOfOrderRecordColumn must be a single-valued BOOLEAN
column");
- }
+ // Delete record column exist and is a BOOLEAN field
+ String deleteRecordColumn = upsertConfig.getDeleteRecordColumn();
+ if (deleteRecordColumn != null) {
+ FieldSpec fieldSpec = schema.getFieldSpecFor(deleteRecordColumn);
+ Preconditions.checkState(fieldSpec != null,
+ String.format("Column %s specified in deleteRecordColumn does not
exist", deleteRecordColumn));
+ Preconditions.checkState(fieldSpec.isSingleValueField(),
+ String.format("The deleteRecordColumn - %s must be a single-valued
column", deleteRecordColumn));
+ DataType dataType = fieldSpec.getDataType();
+ Preconditions.checkState(dataType == DataType.BOOLEAN || dataType ==
DataType.STRING || dataType.isNumeric(),
+ String.format("The deleteRecordColumn - %s must be of type: String /
Boolean / Numeric", deleteRecordColumn));
+ }
- if (upsertConfig.isEnableDeletedKeysCompactionConsistency()) {
- // enableDeletedKeysCompactionConsistency shouldn't exist with
metadataTTL
- Preconditions.checkState(upsertConfig.getMetadataTTL() == 0,
- "enableDeletedKeysCompactionConsistency and metadataTTL shouldn't
exist together for upsert table");
+ String outOfOrderRecordColumn = upsertConfig.getOutOfOrderRecordColumn();
+ if (outOfOrderRecordColumn != null) {
+ Preconditions.checkState(!upsertConfig.isDropOutOfOrderRecord(),
+ "outOfOrderRecordColumn and dropOutOfOrderRecord shouldn't exist
together for upsert table");
+ FieldSpec fieldSpec = schema.getFieldSpecFor(outOfOrderRecordColumn);
+ Preconditions.checkState(
+ fieldSpec != null && fieldSpec.isSingleValueField() &&
fieldSpec.getDataType() == DataType.BOOLEAN,
+ "The outOfOrderRecordColumn must be a single-valued BOOLEAN column");
+ }
- // enableDeletedKeysCompactionConsistency shouldn't exist with preload
enabled
- Preconditions.checkState(upsertConfig.getPreload() !=
Enablement.ENABLE,
- "enableDeletedKeysCompactionConsistency and preload shouldn't
exist together for upsert table");
+ if (upsertConfig.isEnableDeletedKeysCompactionConsistency()) {
+ // enableDeletedKeysCompactionConsistency shouldn't exist with
metadataTTL
+ Preconditions.checkState(upsertConfig.getMetadataTTL() == 0,
+ "enableDeletedKeysCompactionConsistency and metadataTTL shouldn't
exist together for upsert table");
- // enableDeletedKeysCompactionConsistency should exist with
deletedKeysTTL
- Preconditions.checkState(upsertConfig.getDeletedKeysTTL() > 0,
- "enableDeletedKeysCompactionConsistency should exist with
deletedKeysTTL for upsert table");
+ // enableDeletedKeysCompactionConsistency shouldn't exist with preload
enabled
+ Preconditions.checkState(upsertConfig.getPreload() != Enablement.ENABLE,
+ "enableDeletedKeysCompactionConsistency and preload shouldn't exist
together for upsert table");
- // enableDeletedKeysCompactionConsistency should exist with snapshot
enabled
- // NOTE: Allow snapshot to be DEFAULT because it might be enabled at
server level.
- Preconditions.checkState(upsertConfig.getSnapshot() !=
Enablement.DISABLE,
- "enableDeletedKeysCompactionConsistency should exist with snapshot
for upsert table");
+ // enableDeletedKeysCompactionConsistency should exist with
deletedKeysTTL
+ Preconditions.checkState(upsertConfig.getDeletedKeysTTL() > 0,
+ "enableDeletedKeysCompactionConsistency should exist with
deletedKeysTTL for upsert table");
- // enableDeletedKeysCompactionConsistency should exist with
UpsertCompactionTask / UpsertCompactMergeTask
- TableTaskConfig taskConfig = tableConfig.getTaskConfig();
- Preconditions.checkState(
- taskConfig != null &&
(taskConfig.getTaskTypeConfigsMap().containsKey(UPSERT_COMPACTION_TASK_TYPE)
- ||
taskConfig.getTaskTypeConfigsMap().containsKey(UPSERT_COMPACT_MERGE_TASK_TYPE)),
- "enableDeletedKeysCompactionConsistency should exist with
UpsertCompactionTask"
- + " / UpsertCompactMergeTask for upsert table");
- }
-
- // Validate post-partial-upsert transform configs
- List<TransformConfig> postPartialUpsertTransformConfigs =
- upsertConfig.getPostPartialUpsertTransformConfigs();
- if (postPartialUpsertTransformConfigs != null) {
- List<TransformConfig> existingPostPartialUpsertTransformConfigs =
- getPostPartialUpsertTransformConfigs(existingTableConfig);
- Preconditions.checkState(upsertConfig.getMode() ==
UpsertConfig.Mode.PARTIAL,
- "postPartialUpsertTransformConfigs can only be configured for
PARTIAL upsert tables");
- Set<String> primaryKeyColumns = new
HashSet<>(schema.getPrimaryKeyColumns());
- Set<String> transformColumns = new HashSet<>();
- for (TransformConfig transformConfig :
postPartialUpsertTransformConfigs) {
- String columnName = transformConfig.getColumnName();
- String transformFunction = transformConfig.getTransformFunction();
- if (columnName == null || transformFunction == null) {
- throw new IllegalStateException(
- "columnName/transformFunction cannot be null in
postPartialUpsertTransformConfigs "
- + transformConfig);
- }
- Preconditions.checkState(!primaryKeyColumns.contains(columnName),
- "Post-partial-upsert transform cannot target primary key column
'%s'", columnName);
- Preconditions.checkState(comparisonColumns == null ||
!comparisonColumns.contains(columnName),
- "Post-partial-upsert transform cannot target comparison column
'%s'", columnName);
- Preconditions.checkState(!columnName.equals(deleteRecordColumn),
- "Post-partial-upsert transform cannot target delete record
column '%s'", columnName);
- Preconditions.checkState(!columnName.equals(outOfOrderRecordColumn),
- "Post-partial-upsert transform cannot target out-of-order record
column '%s'", columnName);
- if (!transformColumns.add(columnName)) {
- throw new IllegalStateException(
- "Duplicate post-partial-upsert transform config found for
column '" + columnName + "'");
- }
- Preconditions.checkState(schema.hasColumn(columnName),
- "The destination column '%s' of the post-partial-upsert
transform function must be present in the "
- + "schema", columnName);
- if (_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
- throw new IllegalStateException(
- "Groovy transform functions are disabled. Found '" +
transformFunction + "' for column '"
- + columnName + "' in postPartialUpsertTransformConfigs");
- }
- try {
- validateIngestionTransformFunctionVolatility(transformConfig,
- existingPostPartialUpsertTransformConfigs);
- FunctionEvaluator expressionEvaluator =
-
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
- List<String> arguments = expressionEvaluator.getArguments();
- if (arguments.contains(columnName)) {
- throw new IllegalStateException(
- "Arguments of a post-partial-upsert transform function '" +
arguments
- + "' cannot contain the destination column '" +
columnName + "'");
- }
- } catch (IllegalStateException e) {
- throw e;
- } catch (Exception e) {
- throw new IllegalStateException(
- "Invalid post-partial-upsert transform function '" +
transformFunction + "' for column '"
- + columnName + "'", e);
+ // enableDeletedKeysCompactionConsistency should exist with snapshot
enabled
+ // NOTE: Allow snapshot to be DEFAULT because it might be enabled at
server level.
+ Preconditions.checkState(upsertConfig.getSnapshot() !=
Enablement.DISABLE,
+ "enableDeletedKeysCompactionConsistency should exist with snapshot
for upsert table");
+
+ // enableDeletedKeysCompactionConsistency should exist with
UpsertCompactionTask / UpsertCompactMergeTask
+ TableTaskConfig taskConfig = tableConfig.getTaskConfig();
+ Preconditions.checkState(
+ taskConfig != null &&
(taskConfig.getTaskTypeConfigsMap().containsKey(UPSERT_COMPACTION_TASK_TYPE)
+ ||
taskConfig.getTaskTypeConfigsMap().containsKey(UPSERT_COMPACT_MERGE_TASK_TYPE)),
+ "enableDeletedKeysCompactionConsistency should exist with
UpsertCompactionTask"
+ + " / UpsertCompactMergeTask for upsert table");
+ }
+
+ // Validate post-partial-upsert transform configs
+ List<TransformConfig> postPartialUpsertTransformConfigs =
upsertConfig.getPostPartialUpsertTransformConfigs();
+ if (postPartialUpsertTransformConfigs != null) {
+ List<TransformConfig> existingPostPartialUpsertTransformConfigs =
+ getPostPartialUpsertTransformConfigs(existingTableConfig);
+ Preconditions.checkState(upsertConfig.getMode() ==
UpsertConfig.Mode.PARTIAL,
+ "postPartialUpsertTransformConfigs can only be configured for
PARTIAL upsert tables");
+ Set<String> primaryKeyColumns = new
HashSet<>(schema.getPrimaryKeyColumns());
+ Set<String> transformColumns = new HashSet<>();
+ for (TransformConfig transformConfig :
postPartialUpsertTransformConfigs) {
+ String columnName = transformConfig.getColumnName();
+ String transformFunction = transformConfig.getTransformFunction();
+ if (columnName == null || transformFunction == null) {
+ throw new IllegalStateException(
+ "columnName/transformFunction cannot be null in
postPartialUpsertTransformConfigs " + transformConfig);
+ }
+ Preconditions.checkState(!primaryKeyColumns.contains(columnName),
+ "Post-partial-upsert transform cannot target primary key column
'%s'", columnName);
+ Preconditions.checkState(comparisonColumns == null ||
!comparisonColumns.contains(columnName),
+ "Post-partial-upsert transform cannot target comparison column
'%s'", columnName);
+ Preconditions.checkState(!columnName.equals(deleteRecordColumn),
+ "Post-partial-upsert transform cannot target delete record column
'%s'", columnName);
+ Preconditions.checkState(!columnName.equals(outOfOrderRecordColumn),
+ "Post-partial-upsert transform cannot target out-of-order record
column '%s'", columnName);
+ if (!transformColumns.add(columnName)) {
+ throw new IllegalStateException(
+ "Duplicate post-partial-upsert transform config found for column
'" + columnName + "'");
+ }
+ Preconditions.checkState(schema.hasColumn(columnName),
+ "The destination column '%s' of the post-partial-upsert transform
function must be present in the "
+ + "schema", columnName);
+ if (_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
+ throw new IllegalStateException(
+ "Groovy transform functions are disabled. Found '" +
transformFunction + "' for column '" + columnName
+ + "' in postPartialUpsertTransformConfigs");
+ }
+ try {
+ validateIngestionTransformFunctionVolatility(transformConfig,
existingPostPartialUpsertTransformConfigs);
+ FunctionEvaluator expressionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
+ List<String> arguments = expressionEvaluator.getArguments();
+ if (arguments.contains(columnName)) {
+ throw new IllegalStateException("Arguments of a
post-partial-upsert transform function '" + arguments
+ + "' cannot contain the destination column '" + columnName +
"'");
}
+ } catch (IllegalStateException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Invalid post-partial-upsert transform function '" +
transformFunction + "' for column '" + columnName
+ + "'", e);
}
}
+ }
- if (upsertConfig.getConsistencyMode() !=
UpsertConfig.ConsistencyMode.NONE) {
- Preconditions.checkState(upsertConfig.getNewSegmentTrackingTimeMs() >
0,
- "Positive newSegmentTrackingTimeMs is required to enable
consistency mode: "
- + upsertConfig.getConsistencyMode());
- // dropOutOfOrderRecord and outOfOrderRecordColumn are not supported
with SYNC/SNAPSHOT consistency mode
- // because records must be indexed before metadata is updated. By the
time we know if a record is
- // out-of-order, it's already indexed - As of today, we can't skip
indexing or update the
- // out-of-order column value.
- Preconditions.checkState(!upsertConfig.isDropOutOfOrderRecord(),
- "dropOutOfOrderRecord cannot be enabled when consistencyMode is
%s. "
- + "Out-of-order records can only be dropped in NONE
consistency mode.",
- upsertConfig.getConsistencyMode());
- Preconditions.checkState(upsertConfig.getOutOfOrderRecordColumn() ==
null,
- "outOfOrderRecordColumn cannot be configured when consistencyMode
is %s. "
- + "Out-of-order record marking is only supported in NONE
consistency mode.",
- upsertConfig.getConsistencyMode());
- }
-
- validateTTLForUpsertConfig(tableConfig, schema);
- validatePartialUpsertStrategies(tableConfig, schema);
- validateConsumptionDuringUpsertRevert(tableConfig);
- Preconditions.checkState(
- !(PinotMd5Mode.isPinotMd5Disabled() &&
upsertConfig.getHashFunction() == HashFunction.MD5),
- "Upsert hash function MD5 is disabled via '%s=true'",
CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED);
- } else {
- // specifically for dedup
- assert dedupConfig != null;
-
- validateTTLForDedupConfig(tableConfig, schema);
- Preconditions.checkState(
- !(PinotMd5Mode.isPinotMd5Disabled() && dedupConfig.getHashFunction()
== HashFunction.MD5),
- "Dedup hash function MD5 is disabled via '%s=true'",
CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED);
+ if (upsertConfig.getConsistencyMode() !=
UpsertConfig.ConsistencyMode.NONE) {
+ Preconditions.checkState(upsertConfig.getNewSegmentTrackingTimeMs() > 0,
+ "Positive newSegmentTrackingTimeMs is required to enable consistency
mode: "
+ + upsertConfig.getConsistencyMode());
+ // dropOutOfOrderRecord and outOfOrderRecordColumn are not supported
with SYNC/SNAPSHOT consistency mode
+ // because records must be indexed before metadata is updated. By the
time we know if a record is
+ // out-of-order, it's already indexed - As of today, we can't skip
indexing or update the
+ // out-of-order column value.
+ Preconditions.checkState(!upsertConfig.isDropOutOfOrderRecord(),
+ "dropOutOfOrderRecord cannot be enabled when consistencyMode is %s. "
+ + "Out-of-order records can only be dropped in NONE consistency
mode.",
+ upsertConfig.getConsistencyMode());
+ Preconditions.checkState(upsertConfig.getOutOfOrderRecordColumn() ==
null,
+ "outOfOrderRecordColumn cannot be configured when consistencyMode is
%s. "
+ + "Out-of-order record marking is only supported in NONE
consistency mode.",
+ upsertConfig.getConsistencyMode());
}
+
+ validateTTLForUpsertConfig(tableConfig, schema);
+ validatePartialUpsertStrategies(tableConfig, schema);
+ validateConsumptionDuringUpsertRevert(tableConfig);
+ Preconditions.checkState(!(PinotMd5Mode.isPinotMd5Disabled() &&
upsertConfig.getHashFunction() == HashFunction.MD5),
+ "Upsert hash function MD5 is disabled via '%s=true'",
CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED);
+ }
+
+ /// Validates the dedup config of a table with dedup enabled: checks the
requirements shared with upsert (see
+ /// [#validateUpsertAndDedupCommonConfig]), rejects OFFLINE tables,
delegates to [#validateTTLForDedupConfig], and
+ /// rejects MD5 when disabled.
+ @VisibleForTesting
+ static void validateDedupConfig(TableConfig tableConfig, Schema schema) {
+ DedupConfig dedupConfig = tableConfig.getDedupConfig();
+ assert dedupConfig != null;
+ validateUpsertAndDedupCommonConfig(tableConfig, schema);
+ Preconditions.checkState(tableConfig.getTableType() == TableType.REALTIME,
+ "Dedup is not supported for OFFLINE table. Only upsert is supported
for OFFLINE table");
+
+ validateTTLForDedupConfig(tableConfig, schema);
+ Preconditions.checkState(!(PinotMd5Mode.isPinotMd5Disabled() &&
dedupConfig.getHashFunction() == HashFunction.MD5),
+ "Dedup hash function MD5 is disabled via '%s=true'",
CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED);
}
/// Rejects consuming the next segment during a download on tables that
revert upsert metadata in PROTECTED mode,
@@ -2491,13 +2486,32 @@ public final class TableConfigUtils {
|| upsertConfig.getOutOfOrderRecordColumn() != null);
}
- // enum of all the skip-able validation types.
- // ACTIVE_TASKS is retained for backward compatibility: it is still accepted
as a validationTypesToSkip value so
- // existing clients do not break, but it is no longer consumed by any
production path. Active-task validation runs
- // only on the create/update path (see
PinotTableRestletResource.tableTasksValidation, gated by ignoreActiveTasks),
- // not on the validate/tune preflight endpoints, so passing ACTIVE_TASKS
there is a no-op.
+ /// Validation types that can be skipped through the `validationTypesToSkip`
parameter. A type is skipped when the
+ /// types to skip contain it or `ALL`, see [#isSkipped].
public enum ValidationType {
- ALL, TASK, UPSERT, TENANT, MINION_INSTANCES, ACTIVE_TASKS
+ /// Every type below, and the whole table config validation in
[TableConfigUtils#validate]
+ ALL,
+ /// Validation of the task configs
+ TASK,
+ /// Validation of the upsert config of an upsert table
+ UPSERT,
+ /// Validation of the dedup config of a dedup table
+ DEDUP,
+ /// Validation of the instance assignment on table creation and update:
the controller computes the instance
+ /// partitions the table would get from the current instances, and
validates them
+ INSTANCE_ASSIGNMENT,
+ /// Validation of the tenants on the validate and tune endpoints; creating
or updating a table always validates
+ /// them
+ TENANT,
+ /// Validation of the minion instances for the task configs on the
validate and tune endpoints; creating or
+ /// updating a table always validates them
+ MINION_INSTANCES;
+
+ /// Returns whether this validation type is skipped, that is whether the
given types to skip contain it or `ALL`.
+ /// `null` skips nothing.
+ public boolean isSkipped(@Nullable EnumSet<ValidationType> skipTypes) {
+ return skipTypes != null && (skipTypes.contains(this) ||
skipTypes.contains(ALL));
+ }
}
/// needsEmptySegmentPruner checks if EmptySegmentPruner is needed for a
TableConfig.
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 ad89745e377..9fe062216e6 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
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
import java.io.IOException;
import java.util.Arrays;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -30,6 +31,7 @@ import java.util.Set;
import java.util.function.BiConsumer;
import javax.annotation.Nullable;
import org.apache.pinot.common.tier.TierFactory;
+import org.apache.pinot.segment.local.utils.TableConfigUtils.ValidationType;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.Constants;
import
org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
@@ -2620,14 +2622,14 @@ public class TableConfigUtilsTest {
.addSingleValueDimension("myCol", DataType.STRING)
.setPrimaryKeyColumns(List.of("myCol"))
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig,
validSchema);
+ TableConfigUtils.validateDedupConfig(validTableConfig, validSchema);
// OFFLINE table should fail because dedup is not yet supported for
offline tables
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setDedupConfig(new
DedupConfig()).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Dedup is not supported for OFFLINE table.
Only upsert is supported for OFFLINE"
@@ -2637,7 +2639,7 @@ public class TableConfigUtilsTest {
Schema schema =
new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol",
DataType.STRING).build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig, schema);
+ TableConfigUtils.validateDedupConfig(validTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Upsert/Dedup table must have primary key
columns in the schema");
@@ -2648,7 +2650,7 @@ public class TableConfigUtilsTest {
.setPrimaryKeyColumns(List.of("myCol"))
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig, schema);
+ TableConfigUtils.validateDedupConfig(validTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Upsert/Dedup primary key column: myCol
cannot be of multi-value type");
@@ -2658,7 +2660,7 @@ public class TableConfigUtilsTest {
.setDedupConfig(new DedupConfig())
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2672,20 +2674,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
-
- // Dedup and upsert can't be enabled simultaneously
- tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setDedupConfig(new DedupConfig())
- .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
- .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
- .build();
- try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
- fail();
- } catch (IllegalStateException e) {
- assertEquals(e.getMessage(), "A table can have either Upsert or Dedup
enabled, but not both");
- }
+ TableConfigUtils.validateDedupConfig(tableConfig, validSchema);
// Valid TIMESTAMP dedupTimeColumn
DedupConfig dedupConfig = new DedupConfig();
@@ -2700,7 +2689,7 @@ public class TableConfigUtilsTest {
.setDedupConfig(dedupConfig).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
// Should not throw an exception - TIMESTAMP is a valid type for
dedupTimeColumn
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateDedupConfig(tableConfig, schema);
}
@Test
@@ -2722,7 +2711,7 @@ public class TableConfigUtilsTest {
.setStreamConfigs(streamConfigs)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateDedupConfig(tableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2776,7 +2765,7 @@ public class TableConfigUtilsTest {
TableConfig tableConfig =
createTableConfigWithConsumptionPolicy(dropOutOfOrderConfig,
ParallelSegmentConsumptionPolicy.ALLOW_ALWAYS);
IllegalStateException e = expectThrows(IllegalStateException.class,
- () -> TableConfigUtils.validateUpsertAndDedupConfig(tableConfig,
schema));
+ () -> TableConfigUtils.validateUpsertConfig(tableConfig, schema));
assertTrue(e.getMessage().endsWith("Set parallelSegmentConsumptionPolicy
to DISALLOW_ALWAYS"), e.getMessage());
} finally {
consistencyModeListener.reset();
@@ -2833,7 +2822,7 @@ public class TableConfigUtilsTest {
try {
PinotMd5Mode.setPinotMd5Disabled(true);
IllegalStateException exception =
- expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig,
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validateUpsertConfig(tableConfig,
schema));
assertTrue(exception.getMessage().contains(CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED));
} finally {
@@ -2857,7 +2846,7 @@ public class TableConfigUtilsTest {
try {
PinotMd5Mode.setPinotMd5Disabled(true);
IllegalStateException exception =
- expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig,
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validateDedupConfig(tableConfig,
schema));
assertTrue(exception.getMessage().contains(CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED));
} finally {
@@ -2869,6 +2858,7 @@ public class TableConfigUtilsTest {
public void testValidateDedupConfigWithMd5DisabledAllowsUpsertModeNoneMd5() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
.addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.setPrimaryKeyColumns(List.of("myCol"))
.build();
DedupConfig dedupConfig = new DedupConfig();
@@ -2876,12 +2866,13 @@ public class TableConfigUtilsTest {
UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.NONE);
upsertConfig.setHashFunction(HashFunction.MD5);
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(upsertConfig).setDedupConfig(dedupConfig).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(getStreamConfigs())
.build();
try {
PinotMd5Mode.setPinotMd5Disabled(true);
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validate(tableConfig, schema);
} finally {
PinotMd5Mode.setPinotMd5Disabled(false);
}
@@ -2902,7 +2893,7 @@ public class TableConfigUtilsTest {
.addSingleValueDimension("myCol", DataType.STRING)
.setPrimaryKeyColumns(List.of("myCol"))
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig,
validSchema);
+ TableConfigUtils.validateUpsertConfig(validTableConfig, validSchema);
// OFFLINE table without segment partition config should fail with
partition config error
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
@@ -2910,7 +2901,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2926,7 +2917,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2940,7 +2931,7 @@ public class TableConfigUtilsTest {
.setSegmentPartitionConfig(segmentPartitionConfig)
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
// OFFLINE table with partial upsert should fail
UpsertConfig partialUpsertConfig = new
UpsertConfig(UpsertConfig.Mode.PARTIAL);
@@ -2950,7 +2941,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Partial upsert is not supported for
OFFLINE table");
@@ -2960,12 +2951,12 @@ public class TableConfigUtilsTest {
.setUpsertConfig(upsertConfig)
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig,
validSchema);
+ TableConfigUtils.validateUpsertConfig(validTableConfig, validSchema);
Schema schema =
new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol",
DataType.STRING).build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(validTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Upsert/Dedup table must have primary key
columns in the schema");
@@ -2974,7 +2965,7 @@ public class TableConfigUtilsTest {
tableConfig =
new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig).build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2990,7 +2981,7 @@ public class TableConfigUtilsTest {
.setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME",
"T2_REALTIME"))
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
fail("Tag override must not be allowed with upsert");
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Invalid tenant tag override used for
Upsert/Dedup table");
@@ -3006,7 +2997,7 @@ public class TableConfigUtilsTest {
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
fail("Tag override must not be allowed with upsert");
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Invalid tenant tag override used for
Upsert/Dedup table");
@@ -3020,13 +3011,13 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setTagOverrideConfig(new TagOverrideConfig(null, null))
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
StarTreeIndexConfig starTreeIndexConfig = new
StarTreeIndexConfig(List.of("myCol"), null,
List.of(new
AggregationFunctionColumnPair(AggregationFunctionType.COUNT,
"myCol").toColumnName()), null, 10);
@@ -3036,7 +3027,7 @@ public class TableConfigUtilsTest {
.setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "The upsert table cannot have star-tree
index.");
@@ -3108,7 +3099,7 @@ public class TableConfigUtilsTest {
.addMultiValueDimension(mvCol, DataType.STRING)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
fail("Shouldn't fail table creation when delete column type is
single-valued.");
}
@@ -3120,7 +3111,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
fail("Shouldn't fail table creation when delete column type is
single-valued.");
}
@@ -3132,7 +3123,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Should have failed table creation when delete column type is
timestamp.");
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "The deleteRecordColumn - timestampCol must
be of type: String / Boolean / Numeric");
@@ -3145,7 +3136,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Should have failed table creation when invalid delete column
entered.");
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Column invalidCol specified in
deleteRecordColumn does not exist");
@@ -3158,7 +3149,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Should have failed table creation when delete column type is
multi-valued.");
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "The deleteRecordColumn - mvCol must be a
single-valued column");
@@ -3180,7 +3171,7 @@ public class TableConfigUtilsTest {
.addMultiValueDimension(mvCol, DataType.STRING)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Should fail table creation when PK column type is multi-valued.");
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Upsert/Dedup primary key column: myPkCol
cannot be of multi-value type");
@@ -3202,7 +3193,7 @@ public class TableConfigUtilsTest {
.addSingleValueDimension(delCol, DataType.BOOLEAN)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Deleted Keys TTL can only be enabled with
deleteRecordColumn set.");
}
@@ -3222,7 +3213,7 @@ public class TableConfigUtilsTest {
.addSingleValueDimension(delCol, DataType.BOOLEAN)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "MetadataTTL / DeletedKeysTTL does not work
with multiple comparison columns");
}
@@ -3234,7 +3225,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
"MetadataTTL / DeletedKeysTTL must have comparison column: myCol in
numeric type, found: STRING");
@@ -3246,7 +3237,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(upsertConfig)
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
// upsert out-of-order configs
String outOfOrderRecordColumn = "outOfOrderRecordColumn";
@@ -3265,7 +3256,7 @@ public class TableConfigUtilsTest {
.addSingleValueDimension(outOfOrderRecordColumn, DataType.BOOLEAN)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
"outOfOrderRecordColumn and dropOutOfOrderRecord shouldn't exist
together for upsert table");
@@ -3285,7 +3276,7 @@ public class TableConfigUtilsTest {
.addSingleValueDimension(outOfOrderRecordColumn, DataType.STRING)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "The outOfOrderRecordColumn must be a
single-valued BOOLEAN column");
}
@@ -3305,7 +3296,7 @@ public class TableConfigUtilsTest {
.addSingleValueDimension("myPkCol", DataType.STRING)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Expected IllegalStateException for dropOutOfOrderRecord with SYNC
consistency mode");
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("dropOutOfOrderRecord cannot be
enabled when consistencyMode is SYNC"));
@@ -3321,7 +3312,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Expected IllegalStateException for dropOutOfOrderRecord with
SNAPSHOT consistency mode");
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("dropOutOfOrderRecord cannot be
enabled when consistencyMode is SNAPSHOT"));
@@ -3335,7 +3326,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(upsertConfig)
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
// test outOfOrderRecordColumn cannot be enabled with SYNC consistency mode
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
@@ -3353,7 +3344,7 @@ public class TableConfigUtilsTest {
.addSingleValueDimension("isOutOfOrder", DataType.BOOLEAN)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Expected IllegalStateException for outOfOrderRecordColumn with
SYNC consistency mode");
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("outOfOrderRecordColumn cannot be
configured when consistencyMode is SYNC"));
@@ -3369,7 +3360,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
fail("Expected IllegalStateException for outOfOrderRecordColumn with
SNAPSHOT consistency mode");
} catch (IllegalStateException e) {
assertTrue(
@@ -3384,7 +3375,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(upsertConfig)
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
// test enableDeletedKeysCompactionConsistency shouldn't exist with
metadataTTL
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
@@ -3395,7 +3386,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
"enableDeletedKeysCompactionConsistency and metadataTTL shouldn't
exist together for upsert table");
@@ -3410,7 +3401,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
"enableDeletedKeysCompactionConsistency and preload shouldn't exist
together for upsert table");
@@ -3425,7 +3416,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
"enableDeletedKeysCompactionConsistency should exist with
deletedKeysTTL for upsert table");
@@ -3441,7 +3432,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
"enableDeletedKeysCompactionConsistency should exist with snapshot
for upsert table");
@@ -3457,7 +3448,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "enableDeletedKeysCompactionConsistency
should exist with UpsertCompactionTask "
+ "/ UpsertCompactMergeTask for upsert table");
@@ -4795,6 +4786,65 @@ public class TableConfigUtilsTest {
assertNull(TableConfigUtils.getPartitionColumn(tableConfig));
}
+ @Test
+ public void testParseTypesToSkipString() {
+ assertNull(TableConfigUtils.parseTypesToSkipString(null));
+ assertNull(TableConfigUtils.parseTypesToSkipString(" "));
+ assertEquals(TableConfigUtils.parseTypesToSkipString("task,
Instance_Assignment"),
+ EnumSet.of(ValidationType.TASK, ValidationType.INSTANCE_ASSIGNMENT));
+ expectThrows(IllegalArgumentException.class, () ->
TableConfigUtils.parseTypesToSkipString("TASK,UNKNOWN"));
+
+ // A validation type is skipped when the types to skip contain it or ALL
+ assertTrue(ValidationType.TASK.isSkipped(EnumSet.of(ValidationType.TASK)));
+ assertTrue(ValidationType.TASK.isSkipped(EnumSet.of(ValidationType.ALL)));
+
assertFalse(ValidationType.TASK.isSkipped(EnumSet.of(ValidationType.UPSERT)));
+ assertFalse(ValidationType.TASK.isSkipped(null));
+ }
+
+ @Test
+ public void testValidateSkipUpsertAndDedup() {
+ // Upsert and dedup tables require primary key columns, which the schema
does not have
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ String noPrimaryKeyMessage = "Upsert/Dedup table must have primary key
columns in the schema";
+
+ TableConfig upsertTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
+ .setStreamConfigs(getStreamConfigs())
+ .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(upsertTableConfig, schema,
EnumSet.of(ValidationType.DEDUP)));
+ assertEquals(e.getMessage(), noPrimaryKeyMessage);
+ TableConfigUtils.validate(upsertTableConfig, schema,
EnumSet.of(ValidationType.UPSERT));
+
+ TableConfig dedupTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
+ .setStreamConfigs(getStreamConfigs())
+ .setDedupConfig(new DedupConfig())
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ e = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(dedupTableConfig, schema,
EnumSet.of(ValidationType.UPSERT)));
+ assertEquals(e.getMessage(), noPrimaryKeyMessage);
+ TableConfigUtils.validate(dedupTableConfig, schema,
EnumSet.of(ValidationType.DEDUP));
+
+ // A table cannot enable both upsert and dedup, even when both validations
are skipped
+ TableConfig upsertAndDedupTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN)
+ .setStreamConfigs(getStreamConfigs())
+ .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
+ .setDedupConfig(new DedupConfig())
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ EnumSet<ValidationType> skipUpsertAndDedup =
EnumSet.of(ValidationType.UPSERT, ValidationType.DEDUP);
+ e = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(upsertAndDedupTableConfig, schema,
skipUpsertAndDedup));
+ assertEquals(e.getMessage(), "A table can have either Upsert or Dedup
enabled, but not both");
+ }
+
@DataProvider(name = "tableTypeTestDataProvider")
public Object[][] tableTypeTestDataProvider() {
return new Object[][]{
@@ -5025,7 +5075,7 @@ public class TableConfigUtilsTest {
.setStreamConfigs(streamConfigs)
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(tableConfig, schema);
// postPartialUpsertTransformConfigs on FULL upsert should fail
UpsertConfig fullUpsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
@@ -5038,7 +5088,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(fullTableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(fullTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("PARTIAL"));
@@ -5054,7 +5104,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(dupConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(dupTableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(dupTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("Duplicate"));
@@ -5071,7 +5121,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(missingColTableConfig,
schema);
+ TableConfigUtils.validateUpsertConfig(missingColTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("must be present in the schema"));
@@ -5086,7 +5136,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(selfRefConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(selfRefTableConfig,
schema);
+ TableConfigUtils.validateUpsertConfig(selfRefTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("cannot contain the destination
column"));
@@ -5101,7 +5151,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(pkConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(pkTableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(pkTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("primary key column"));
@@ -5117,7 +5167,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(compConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(compTableConfig, schema);
+ TableConfigUtils.validateUpsertConfig(compTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("comparison column"));
@@ -5135,7 +5185,7 @@ public class TableConfigUtilsTest {
.setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(deleteColTableConfig,
schema);
+ TableConfigUtils.validateUpsertConfig(deleteColTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("delete record column"));
@@ -5160,7 +5210,7 @@ public class TableConfigUtilsTest {
.setUpsertConfig(oooColConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(oooColTableConfig,
schemaWithBool);
+ TableConfigUtils.validateUpsertConfig(oooColTableConfig, schemaWithBool);
fail();
} catch (IllegalStateException e) {
assertTrue(e.getMessage().contains("out-of-order record column"));
@@ -5179,24 +5229,24 @@ public class TableConfigUtilsTest {
TableConfig nonDeterministicConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "now()");
IllegalStateException nonDeterministicError =
expectThrows(IllegalStateException.class,
- () ->
TableConfigUtils.validateUpsertAndDedupConfig(nonDeterministicConfig, schema));
+ () -> TableConfigUtils.validateUpsertConfig(nonDeterministicConfig,
schema));
assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
TableConfig existingConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "now()");
- TableConfigUtils.validateUpsertAndDedupConfig(nonDeterministicConfig,
schema, existingConfig);
+ TableConfigUtils.validateUpsertConfig(nonDeterministicConfig, schema,
existingConfig);
TableConfig changedConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "plus(now(), 1)");
nonDeterministicError = expectThrows(IllegalStateException.class,
- () -> TableConfigUtils.validateUpsertAndDedupConfig(changedConfig,
schema, existingConfig));
+ () -> TableConfigUtils.validateUpsertConfig(changedConfig, schema,
existingConfig));
assertTrue(nonDeterministicError.getMessage().contains("Function 'now' has
VOLATILE volatility"));
TableConfig randomConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "rand()");
nonDeterministicError = expectThrows(IllegalStateException.class,
- () -> TableConfigUtils.validateUpsertAndDedupConfig(randomConfig,
schema));
+ () -> TableConfigUtils.validateUpsertConfig(randomConfig, schema));
assertTrue(nonDeterministicError.getMessage().contains("Function 'rand'
has VOLATILE volatility"));
TableConfig seededRandomConfig =
buildPostPartialUpsertTransformTableConfig(streamConfigs, "rand(123)");
- TableConfigUtils.validateUpsertAndDedupConfig(seededRandomConfig, schema);
+ TableConfigUtils.validateUpsertConfig(seededRandomConfig, schema);
}
private static TableConfig
buildPostPartialUpsertTransformTableConfig(Map<String, String> streamConfigs,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
