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 3019dc48bb1 Refactor upsert/dedup config validation into per-mode
branches (#18956)
3019dc48bb1 is described below
commit 3019dc48bb1ad46bdd286bf8f1ff3f9db9d6174c
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Thu Jul 9 23:53:45 2026 -0700
Refactor upsert/dedup config validation into per-mode branches (#18956)
---
.../segment/local/utils/TableConfigUtils.java | 155 ++--
.../segment/local/utils/TableConfigUtilsTest.java | 795 +++++++++------------
2 files changed, 405 insertions(+), 545 deletions(-)
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 f5330eafbde..5dd51a79e31 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
@@ -210,7 +210,6 @@ public final class TableConfigUtils {
if (!skipTypes.contains(ValidationType.UPSERT)) {
validateUpsertAndDedupConfig(tableConfig, schema);
- validatePartialUpsertStrategies(tableConfig, schema);
}
validateTaskConfig(tableConfig);
@@ -903,43 +902,25 @@ public final class TableConfigUtils {
}
}
- /**
- * Validates the upsert-related configurations
- * - check table type supports the configured mode
- * - the primary key exists on the schema
- * - strict replica-group is configured for routing type
- * - consumer type must be low-level
- * - comparison column exists
- */
+ /// 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 COMPLETED instance partitions, valid
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.
@VisibleForTesting
static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema
schema) {
- if (tableConfig.getUpsertMode() == UpsertConfig.Mode.NONE &&
(tableConfig.getDedupConfig() == null
- || !tableConfig.getDedupConfig().isDedupEnabled())) {
+ boolean upsertEnabled = tableConfig.isUpsertEnabled();
+ boolean dedupEnabled = tableConfig.isDedupEnabled();
+ if (!upsertEnabled && !dedupEnabled) {
return;
}
- boolean isUpsertEnabled = tableConfig.getUpsertMode() !=
UpsertConfig.Mode.NONE;
- DedupConfig dedupConfig = tableConfig.getDedupConfig();
- boolean isDedupEnabled = dedupConfig != null &&
dedupConfig.isDedupEnabled();
-
// check both upsert and dedup are not enabled simultaneously
- Preconditions.checkState(!(isUpsertEnabled && isDedupEnabled),
+ Preconditions.checkState(!(upsertEnabled && dedupEnabled),
"A table can have either Upsert or Dedup enabled, but not both");
- if (tableConfig.getTableType() == TableType.OFFLINE) {
- Preconditions.checkState(isUpsertEnabled && !isDedupEnabled,
- "Dedup is not supported for OFFLINE table. Only upsert is supported
for OFFLINE table");
- Preconditions.checkState(tableConfig.getUpsertMode() !=
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
- // based on partition, ensuring all segments of a partition land on the
same server for correct dedup.
- IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
- SegmentPartitionConfig segmentPartitionConfig =
- indexingConfig != null ? indexingConfig.getSegmentPartitionConfig()
: null;
- Preconditions.checkState(
- segmentPartitionConfig != null &&
MapUtils.isNotEmpty(segmentPartitionConfig.getColumnPartitionMap()),
- "Offline upsert table must have segment partition config to ensure
correct partition-based "
- + "segment assignment. Configure segmentPartitionConfig in the
indexingConfig.");
- }
// primary key exists
Preconditions.checkState(CollectionUtils.isNotEmpty(schema.getPrimaryKeyColumns()),
"Upsert/Dedup table must have primary key columns in the schema");
@@ -955,26 +936,39 @@ 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");
+ Map<String, InstanceAssignmentConfig> instanceAssignmentConfigMap =
tableConfig.getInstanceAssignmentConfigMap();
+ Preconditions.checkState(instanceAssignmentConfigMap == null ||
!instanceAssignmentConfigMap.containsKey(
+ InstancePartitionsType.COMPLETED.name()),
+ "COMPLETED instance partitions can't be configured for upsert /
dedup tables");
+ } else {
+ Preconditions.checkState(!dedupEnabled,
+ "Dedup is not supported for OFFLINE table. Only upsert is supported
for OFFLINE table");
+ assert upsertConfig != null;
+ 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
+ // based on partition, ensuring all segments of a partition land on the
same server for correct dedup.
+ IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
+ SegmentPartitionConfig segmentPartitionConfig =
+ indexingConfig != null ? indexingConfig.getSegmentPartitionConfig()
: null;
+ Preconditions.checkState(
+ segmentPartitionConfig != null &&
MapUtils.isNotEmpty(segmentPartitionConfig.getColumnPartitionMap()),
+ "Offline upsert table must have segment partition config to ensure
correct partition-based "
+ + "segment assignment. Configure segmentPartitionConfig in the
indexingConfig.");
}
- // specifically for upsert
- UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
- if (PinotMd5Mode.isPinotMd5Disabled()) {
- if (isUpsertEnabled && upsertConfig != null &&
upsertConfig.getHashFunction() == HashFunction.MD5) {
- throw new IllegalStateException(String.format(
- "Upsert hash function MD5 is disabled via '%s=true'",
CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED));
- }
- if (isDedupEnabled && dedupConfig.getHashFunction() == HashFunction.MD5)
{
- throw new IllegalStateException(String.format(
- "Dedup hash function MD5 is disabled via '%s=true'",
CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED));
- }
- }
- if (upsertConfig != null) {
+ 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
@@ -1006,7 +1000,7 @@ public final class TableConfigUtils {
String outOfOrderRecordColumn = upsertConfig.getOutOfOrderRecordColumn();
if (outOfOrderRecordColumn != null) {
-
Preconditions.checkState(!Boolean.TRUE.equals(upsertConfig.isDropOutOfOrderRecord()),
+ Preconditions.checkState(!upsertConfig.isDropOutOfOrderRecord(),
"outOfOrderRecordColumn and dropOutOfOrderRecord shouldn't exist
together for upsert table");
FieldSpec fieldSpec = schema.getFieldSpecFor(outOfOrderRecordColumn);
Preconditions.checkState(
@@ -1113,17 +1107,21 @@ public final class TableConfigUtils {
+ "Out-of-order record marking is only supported in NONE
consistency mode.",
upsertConfig.getConsistencyMode());
}
- }
- if (tableConfig.getTableType() == TableType.REALTIME) {
+ validateTTLForUpsertConfig(tableConfig, schema);
+ validatePartialUpsertStrategies(tableConfig, schema);
Preconditions.checkState(
- tableConfig.getInstanceAssignmentConfigMap() == null ||
!tableConfig.getInstanceAssignmentConfigMap()
- .containsKey(InstancePartitionsType.COMPLETED.name()),
- "COMPLETED instance partitions can't be configured for upsert /
dedup tables");
+ !(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);
}
- validateAggregateMetricsForUpsertConfig(tableConfig);
- validateTTLForUpsertConfig(tableConfig, schema);
- validateTTLForDedupConfig(tableConfig, schema);
}
/**
@@ -1144,7 +1142,8 @@ public final class TableConfigUtils {
@VisibleForTesting
static void validateTTLForUpsertConfig(TableConfig tableConfig, Schema
schema) {
UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
- if (upsertConfig == null || (upsertConfig.getMetadataTTL() == 0 &&
upsertConfig.getDeletedKeysTTL() == 0)) {
+ assert upsertConfig != null;
+ if (upsertConfig.getMetadataTTL() <= 0 && upsertConfig.getDeletedKeysTTL()
<= 0) {
return;
}
@@ -1185,7 +1184,8 @@ public final class TableConfigUtils {
@VisibleForTesting
static void validateTTLForDedupConfig(TableConfig tableConfig, Schema
schema) {
DedupConfig dedupConfig = tableConfig.getDedupConfig();
- if (dedupConfig == null || (dedupConfig.getMetadataTTL() <= 0)) {
+ assert dedupConfig != null;
+ if (dedupConfig.getMetadataTTL() <= 0) {
return;
}
@@ -1316,23 +1316,6 @@ public final class TableConfigUtils {
}
}
- /**
- * Validates metrics aggregation when upsert config is enabled
- * - Metrics aggregation cannot be enabled when Upsert Config is enabled.
- * - Aggregation cannot be enabled in the Ingestion Config and Indexing
Config at the same time.
- */
- private static void validateAggregateMetricsForUpsertConfig(TableConfig
config) {
- boolean isAggregateMetricsEnabledInIndexingConfig =
config.getIndexingConfig().isAggregateMetrics();
- boolean hasAggregationConfigs = config.getIngestionConfig() != null &&
CollectionUtils.isNotEmpty(
- config.getIngestionConfig().getAggregationConfigs());
- boolean bothAggregationConfigsEnabled =
isAggregateMetricsEnabledInIndexingConfig && hasAggregationConfigs;
- boolean anyOneAggregationConfigsEnabled =
isAggregateMetricsEnabledInIndexingConfig || hasAggregationConfigs;
- Preconditions.checkState(!bothAggregationConfigsEnabled,
- "Metrics aggregation cannot be enabled in the Indexing Config and
Ingestion Config at the same time");
- Preconditions.checkState(!anyOneAggregationConfigsEnabled,
- "Metrics aggregation and upsert cannot be enabled together");
- }
-
/**
* Validates the partial upsert-related configurations:
* - Null handling must be enabled
@@ -1344,7 +1327,9 @@ public final class TableConfigUtils {
*/
@VisibleForTesting
static void validatePartialUpsertStrategies(TableConfig tableConfig, Schema
schema) {
- if (tableConfig.getUpsertMode() != UpsertConfig.Mode.PARTIAL) {
+ UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
+ assert upsertConfig != null;
+ if (upsertConfig.getMode() != UpsertConfig.Mode.PARTIAL) {
return;
}
@@ -1352,8 +1337,6 @@ public final class TableConfigUtils {
schema.isEnableColumnBasedNullHandling() ||
tableConfig.getIndexingConfig().isNullHandlingEnabled(),
"Null handling must be enabled for partial upsert tables");
- UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
- assert upsertConfig != null;
Map<String, UpsertConfig.Strategy> partialUpsertStrategies =
upsertConfig.getPartialUpsertStrategies();
String partialUpsertMergerClass =
upsertConfig.getPartialUpsertMergerClass();
@@ -1453,25 +1436,25 @@ public final class TableConfigUtils {
if (!Objects.equals(existingUpsertConfig.getComparisonColumns(),
newUpsertConfig.getComparisonColumns())) {
violations.add(
String.format("upsertConfig.comparisonColumns (%s -> %s)",
existingUpsertConfig.getComparisonColumns(),
- newUpsertConfig.getComparisonColumns()));
+ newUpsertConfig.getComparisonColumns()));
}
List<String> existingComparisonColumns =
existingUpsertConfig.getComparisonColumns();
if (existingComparisonColumns == null ||
existingComparisonColumns.isEmpty()) {
String existingTimeColumn =
existingConfig.getValidationConfig() != null ?
existingConfig.getValidationConfig().getTimeColumnName()
- : null;
+ : null;
String newTimeColumn =
newConfig.getValidationConfig() != null ?
newConfig.getValidationConfig().getTimeColumnName() : null;
if (!Objects.equals(existingTimeColumn, newTimeColumn)) {
violations.add(
String.format("timeColumnName (%s -> %s) - used as default
comparison column", existingTimeColumn,
- newTimeColumn));
+ newTimeColumn));
}
}
if (existingUpsertConfig.isDropOutOfOrderRecord() !=
newUpsertConfig.isDropOutOfOrderRecord()) {
violations.add(
String.format("upsertConfig.dropOutOfOrderRecord (%s -> %s)",
existingUpsertConfig.isDropOutOfOrderRecord(),
- newUpsertConfig.isDropOutOfOrderRecord()));
+ newUpsertConfig.isDropOutOfOrderRecord()));
}
if (!Objects.equals(existingUpsertConfig.getOutOfOrderRecordColumn(),
newUpsertConfig.getOutOfOrderRecordColumn())) {
@@ -2167,10 +2150,8 @@ public final class TableConfigUtils {
}
public static boolean isCommitTimeCompactionEnabled(TableConfig tableConfig)
{
- if (tableConfig.getUpsertConfig() == null) {
- return false;
- }
- return tableConfig.getUpsertConfig().isEnableCommitTimeCompaction();
+ UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
+ return upsertConfig != null && upsertConfig.isEnableCommitTimeCompaction();
}
/**
@@ -2377,10 +2358,10 @@ public final class TableConfigUtils {
}
private static boolean
hasConsumingSegmentTierOverwriteForRealtimeTable(TableConfig tableConfig) {
- if (tableConfig.getTableType() != TableType.REALTIME
- || (CollectionUtils.isNotEmpty(tableConfig.getTierConfigsList())
- && tableConfig.getTierConfigsList().stream()
- .anyMatch(tierConfig ->
CONSUMING_SEGMENT_TIER.equals(tierConfig.getName())))) {
+ if (tableConfig.getTableType() != TableType.REALTIME || (
+ CollectionUtils.isNotEmpty(tableConfig.getTierConfigsList()) &&
tableConfig.getTierConfigsList()
+ .stream()
+ .anyMatch(tierConfig ->
CONSUMING_SEGMENT_TIER.equals(tierConfig.getName())))) {
return false;
}
IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
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 11aefa5d3c2..3bd12121812 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
@@ -20,7 +20,6 @@ package org.apache.pinot.segment.local.utils;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.common.collect.Lists;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -66,7 +65,7 @@ import
org.apache.pinot.spi.config.table.ingestion.SourceFieldConfig;
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.DimensionFieldSpec;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.MetricFieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -85,10 +84,11 @@ import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.PinotDataType;
import org.apache.pinot.spi.utils.PinotMd5Mode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
-import org.mockito.Mockito;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.*;
@@ -98,6 +98,9 @@ public class TableConfigUtilsTest {
private static final String TIME_COLUMN = "timeColumn";
private static final String PARTITION_COLUMN = "partitionColumn";
+ private static final RoutingConfig STRICT_REPLICA_ROUTING_CONFIG =
+ new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false);
+
@Test
public void validateTimeColumnValidationConfig() {
// REALTIME table
@@ -143,8 +146,7 @@ public class TableConfigUtilsTest {
}
// timeColumnName not present as valid time spec schema
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension(TIME_COLUMN, FieldSpec.DataType.LONG)
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension(TIME_COLUMN,
DataType.LONG)
.build();
tableConfig =
new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN).build();
@@ -157,7 +159,7 @@ public class TableConfigUtilsTest {
// valid
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setStreamConfigs(getStreamConfigs())
.setTableName(TABLE_NAME)
@@ -202,8 +204,7 @@ public class TableConfigUtilsTest {
}
// non-null schema and timeColumnName, but timeColumnName not present as a
time spec in schema
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension(TIME_COLUMN, FieldSpec.DataType.STRING)
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension(TIME_COLUMN,
DataType.STRING)
.build();
tableConfig =
new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN).build();
@@ -216,7 +217,7 @@ public class TableConfigUtilsTest {
// valid
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
tableConfig =
new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN).build();
@@ -227,7 +228,7 @@ public class TableConfigUtilsTest {
public void validateDimensionTableConfig() {
// dimension table with REALTIME type (should be OFFLINE)
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension(TIME_COLUMN, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(TIME_COLUMN, DataType.STRING)
.build();
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setIsDimTable(true)
@@ -253,8 +254,7 @@ public class TableConfigUtilsTest {
}
// dimension table without a Primary Key
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension(TIME_COLUMN, FieldSpec.DataType.STRING)
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension(TIME_COLUMN,
DataType.STRING)
.build();
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setIsDimTable(true)
@@ -269,8 +269,8 @@ public class TableConfigUtilsTest {
// valid dimension table
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIsDimTable(true).build();
TableConfigUtils.validate(tableConfig, schema);
@@ -279,8 +279,8 @@ public class TableConfigUtilsTest {
@Test
public void validateDimensionTableSegmentAssignmentStrategy() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
// Valid: null segment assignment strategy
@@ -372,15 +372,13 @@ public class TableConfigUtilsTest {
IndexingConfig indexingConfig = new IndexingConfig();
indexingConfig.setNoDictionaryColumns(List.of("twiceSum"));
tableConfig.setIndexingConfig(indexingConfig);
- schema =
- new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("twiceSum",
FieldSpec.DataType.DOUBLE).build();
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("twiceSum",
DataType.DOUBLE).build();
ingestionConfig.setTransformConfigs(List.of(new TransformConfig("twice",
"col * 2")));
ingestionConfig.setAggregationConfigs(List.of(new
AggregationConfig("twiceSum", "SUM(twice)")));
TableConfigUtils.validate(tableConfig, schema);
// valid transform configs
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol",
DataType.STRING)
.build();
indexingConfig.setNoDictionaryColumns(List.of("myCol"));
ingestionConfig.setAggregationConfigs(null);
@@ -389,8 +387,8 @@ public class TableConfigUtilsTest {
// valid transform configs
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addMetric("transformedCol", FieldSpec.DataType.LONG)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addMetric("transformedCol", DataType.LONG)
.build();
ingestionConfig.setTransformConfigs(Arrays.asList(new
TransformConfig("myCol", "reverse(anotherCol)"),
new TransformConfig("transformedCol", "Groovy({x+y}, x, y)")));
@@ -559,8 +557,7 @@ public class TableConfigUtilsTest {
ingestionConfig.setTransformConfigs(null);
ingestionConfig.setComplexTypeConfig(
new ComplexTypeConfig(null, ".", null, Map.of("after.", "")));
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addMultiValueDimension("after.test", FieldSpec.DataType.STRING)
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMultiValueDimension("after.test",
DataType.STRING)
.build();
try {
TableConfigUtils.validate(tableConfig, schema);
@@ -602,7 +599,7 @@ public class TableConfigUtilsTest {
@Test
public void ingestionAggregationConfigsTest() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addDateTime("timeColumn", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
IngestionConfig ingestionConfig = new IngestionConfig();
ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("d1",
"SUM(s1)")));
@@ -617,7 +614,7 @@ public class TableConfigUtilsTest {
// expected
}
- schema.addField(new DimensionFieldSpec("d1", FieldSpec.DataType.DOUBLE,
true));
+ schema.addField(new DimensionFieldSpec("d1", DataType.DOUBLE, true));
tableConfig.getIndexingConfig().setAggregateMetrics(true);
try {
TableConfigUtils.validateIngestionConfig(tableConfig, schema);
@@ -634,7 +631,7 @@ public class TableConfigUtilsTest {
// expected
}
- schema.addField(new MetricFieldSpec("m1", FieldSpec.DataType.DOUBLE));
+ schema.addField(new MetricFieldSpec("m1", DataType.DOUBLE));
// Mark the metric as no-dictionary up front (a requirement validated by
validateMetricsAggregation, covered in
// metricsAggregationValidationTest) so the steps below exercise the
per-aggregation-config function validation.
IndexingConfig indexingConfig = new IndexingConfig();
@@ -680,7 +677,7 @@ public class TableConfigUtilsTest {
ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1",
"SUM(s1)")));
TableConfigUtils.validateIngestionConfig(tableConfig, schema);
- schema.addField(new MetricFieldSpec("m2", FieldSpec.DataType.DOUBLE));
+ schema.addField(new MetricFieldSpec("m2", DataType.DOUBLE));
indexingConfig.setNoDictionaryColumns(List.of("m1", "m2"));
try {
TableConfigUtils.validateIngestionConfig(tableConfig, schema);
@@ -689,7 +686,7 @@ public class TableConfigUtilsTest {
// expected
}
- schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("d1",
FieldSpec.DataType.BYTES).build();
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("d1",
DataType.BYTES).build();
// distinctcounthllmv is not supported, we expect this to not validate
List<AggregationConfig> aggregationConfigs = Arrays.asList(new
AggregationConfig("d1", "DISTINCTCOUNTHLLMV(s1)"));
ingestionConfig.setAggregationConfigs(aggregationConfigs);
@@ -708,11 +705,11 @@ public class TableConfigUtilsTest {
// distinctcounthll, expect that the function name in various forms (with
and without underscores) still validates
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addMetric("d1", FieldSpec.DataType.BYTES)
- .addMetric("d2", FieldSpec.DataType.BYTES)
- .addMetric("d3", FieldSpec.DataType.BYTES)
- .addMetric("d4", FieldSpec.DataType.BYTES)
- .addMetric("d5", FieldSpec.DataType.BYTES)
+ .addMetric("d1", DataType.BYTES)
+ .addMetric("d2", DataType.BYTES)
+ .addMetric("d3", DataType.BYTES)
+ .addMetric("d4", DataType.BYTES)
+ .addMetric("d5", DataType.BYTES)
.build();
aggregationConfigs = Arrays.asList(new AggregationConfig("d1",
"distinct_count_hll(s1)"),
@@ -753,7 +750,7 @@ public class TableConfigUtilsTest {
}
// distinctcounthll, expect not specified log2m argument to default to 8
- schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("d1",
FieldSpec.DataType.BYTES).build();
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("d1",
DataType.BYTES).build();
aggregationConfigs = Arrays.asList(new AggregationConfig("d1",
"DISTINCTCOUNTHLL(s1)"));
ingestionConfig.setAggregationConfigs(aggregationConfigs);
@@ -786,11 +783,11 @@ public class TableConfigUtilsTest {
// sumprecision, expect that the function name in various forms (with and
without underscores) still validates
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("s1", FieldSpec.DataType.BIG_DECIMAL)
- .addMetric("d1", FieldSpec.DataType.BIG_DECIMAL)
- .addMetric("d2", FieldSpec.DataType.BIG_DECIMAL)
- .addMetric("d3", FieldSpec.DataType.BIG_DECIMAL)
- .addMetric("d4", FieldSpec.DataType.BIG_DECIMAL)
+ .addSingleValueDimension("s1", DataType.BIG_DECIMAL)
+ .addMetric("d1", DataType.BIG_DECIMAL)
+ .addMetric("d2", DataType.BIG_DECIMAL)
+ .addMetric("d3", DataType.BIG_DECIMAL)
+ .addMetric("d4", DataType.BIG_DECIMAL)
.build();
aggregationConfigs = Arrays.asList(new AggregationConfig("d1",
"sum_precision(s1, 10, 32)"),
@@ -807,8 +804,8 @@ public class TableConfigUtilsTest {
// with too many arguments should fail
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("s1", FieldSpec.DataType.BIG_DECIMAL)
- .addMetric("d1", FieldSpec.DataType.BIG_DECIMAL)
+ .addSingleValueDimension("s1", DataType.BIG_DECIMAL)
+ .addMetric("d1", DataType.BIG_DECIMAL)
.build();
aggregationConfigs = Arrays.asList(new AggregationConfig("d1",
"sum_precision(s1, 10, 32, 99)"));
@@ -833,9 +830,9 @@ public class TableConfigUtilsTest {
// A COMPLEX column is neither a valid aggregation key nor an aggregatable
metric, so metrics aggregation (via
// either the aggregateMetrics flag or ingestion aggregationConfigs)
cannot be enabled when the schema contains one.
Schema schemaWithComplex = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .addMetric("m1", FieldSpec.DataType.LONG)
- .addComplex("complexCol", FieldSpec.DataType.MAP, Map.of())
+ .addDateTime("timeColumn", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .addMetric("m1", DataType.LONG)
+ .addComplex("complexCol", DataType.MAP, Map.of())
.build();
TableConfig aggregateMetricsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
@@ -856,18 +853,18 @@ public class TableConfigUtilsTest {
// A multi-value dimension cannot be used as an aggregation key.
Schema schemaWithMvDimension = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .addMultiValueDimension("mvDim", FieldSpec.DataType.INT)
- .addMetric("m1", FieldSpec.DataType.LONG)
+ .addDateTime("timeColumn", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .addMultiValueDimension("mvDim", DataType.INT)
+ .addMetric("m1", DataType.LONG)
.build();
assertMetricsAggregationValidationFails(aggregationConfigsConfig,
schemaWithMvDimension,
"multi-value dimension column");
// Metric columns must be no-dictionary, for both mechanisms.
Schema schemaWithSvColumns = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .addSingleValueDimension("d1", FieldSpec.DataType.INT)
- .addMetric("m1", FieldSpec.DataType.LONG)
+ .addDateTime("timeColumn", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .addSingleValueDimension("d1", DataType.INT)
+ .addMetric("m1", DataType.LONG)
.build();
TableConfig dictMetricAggregateMetricsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName("timeColumn")
@@ -929,7 +926,7 @@ public class TableConfigUtilsTest {
@Test
public void ingestionStreamConfigsTest() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addDateTime("timeColumn", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
Map<String, String> streamConfigs = getStreamConfigs();
IngestionConfig ingestionConfig = new IngestionConfig();
@@ -1168,7 +1165,7 @@ public class TableConfigUtilsTest {
@Test
public void validateTierConfigs() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
// null tier configs
TableConfig tableConfig =
@@ -1182,16 +1179,14 @@ public class TableConfigUtilsTest {
TableConfigUtils.validate(tableConfig, schema);
// 1 tier configs
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null)))
.build();
TableConfigUtils.validate(tableConfig, schema);
// 2 tier configs, case insensitive check
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE.toLowerCase(), "30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null),
new TierConfig("tier2", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"40d", null,
@@ -1203,7 +1198,7 @@ public class TableConfigUtilsTest {
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setStreamConfigs(getStreamConfigs())
- .setTierConfigList(Lists.newArrayList(
+ .setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE.toLowerCase(),
"tier1_tag_OFFLINE", null, null),
new TierConfig("tier2",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE.toLowerCase(), "40d", null,
@@ -1213,7 +1208,7 @@ public class TableConfigUtilsTest {
// tier name empty
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(new TierConfig("",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "30d", null,
+ .setTierConfigList(List.of(new TierConfig("",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE", null,
null)))
.build();
try {
@@ -1224,8 +1219,7 @@ public class TableConfigUtilsTest {
}
// tier name repeats
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("sameTierName",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null),
new TierConfig("sameTierName",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "100d", null,
@@ -1239,8 +1233,7 @@ public class TableConfigUtilsTest {
}
// segmentSelectorType invalid
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null),
new TierConfig("tier2", "unsupportedSegmentSelector", "40d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE,
@@ -1254,8 +1247,7 @@ public class TableConfigUtilsTest {
}
// segmentAge not provided for TIME segmentSelectorType
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
null, null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null),
new TierConfig("tier2", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"40d", null,
@@ -1269,8 +1261,7 @@ public class TableConfigUtilsTest {
}
// segmentAge invalid
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null),
new TierConfig("tier2", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"3600", null,
@@ -1289,35 +1280,32 @@ public class TableConfigUtilsTest {
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setStreamConfigs(getStreamConfigs())
- .setTierConfigList(Lists.newArrayList(
+ .setTierConfigList(List.of(
new TierConfig("consuming",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE,
"consuming_tag_REALTIME", null, null)))
.build();
TableConfigUtils.validate(tableConfig, schema);
// fixedSegmentSelector
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.FIXED_SEGMENT_SELECTOR_TYPE,
null, null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null)))
.build();
TableConfigUtils.validate(tableConfig, schema);
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
- new TierConfig("tier1", TierFactory.FIXED_SEGMENT_SELECTOR_TYPE,
"30d", Lists.newArrayList(),
+ .setTierConfigList(List.of(new TierConfig("tier1",
TierFactory.FIXED_SEGMENT_SELECTOR_TYPE, "30d", List.of(),
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null)))
.build();
TableConfigUtils.validate(tableConfig, schema);
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
- new TierConfig("tier1", TierFactory.FIXED_SEGMENT_SELECTOR_TYPE,
null, Lists.newArrayList("seg0", "seg1"),
+ .setTierConfigList(
+ List.of(new TierConfig("tier1",
TierFactory.FIXED_SEGMENT_SELECTOR_TYPE, null, List.of("seg0", "seg1"),
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null)))
.build();
TableConfigUtils.validate(tableConfig, schema);
// storageType invalid
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"30d", null, "unsupportedStorageType",
"tier1_tag_OFFLINE", null, null),
new TierConfig("tier2", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"40d", null,
@@ -1332,8 +1320,7 @@ public class TableConfigUtilsTest {
}
// serverTag not provided for PINOT_SERVER storageType
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag_OFFLINE",
null, null),
new TierConfig("tier2", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"40d", null,
@@ -1347,8 +1334,7 @@ public class TableConfigUtilsTest {
}
// serverTag invalid
- tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTierConfigList(Lists.newArrayList(
+ tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"30d", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier1_tag", null,
null),
new TierConfig("tier2", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"40d", null,
@@ -1387,10 +1373,10 @@ public class TableConfigUtilsTest {
@Test
public void testValidateFieldConfig() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:HOURS:EPOCH",
"1:HOURS")
- .addSingleValueDimension("myCol1", FieldSpec.DataType.STRING)
- .addMultiValueDimension("myCol2", FieldSpec.DataType.INT)
- .addSingleValueDimension("intCol", FieldSpec.DataType.INT)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
+ .addSingleValueDimension("myCol1", DataType.STRING)
+ .addMultiValueDimension("myCol2", DataType.INT)
+ .addSingleValueDimension("intCol", DataType.INT)
.build();
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setNoDictionaryColumns(Arrays.asList("myCol1"))
@@ -1751,7 +1737,7 @@ public class TableConfigUtilsTest {
@Test
public void testValidateFieldConfigDuplicateColumnName() {
final Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol1", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol1", DataType.STRING)
.build();
final TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
.setTableName(TABLE_NAME).build();
@@ -1771,7 +1757,7 @@ public class TableConfigUtilsTest {
@Test
public void testValidateFieldConfigDuplicateColumnNameDifferentEncoding() {
final Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol1", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol1", DataType.STRING)
.build();
final TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
.setTableName(TABLE_NAME).build();
@@ -1793,7 +1779,7 @@ public class TableConfigUtilsTest {
public void testValidateFieldConfigDuplicateColumnNameWithIndexes()
throws Exception {
final Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol1", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol1", DataType.STRING)
.build();
final TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
.setTableName(TABLE_NAME).build();
@@ -1818,8 +1804,8 @@ public class TableConfigUtilsTest {
@Test
public void testValidateFieldConfigNoDuplicates() {
final Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol1", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myCol2", FieldSpec.DataType.INT)
+ .addSingleValueDimension("myCol1", DataType.STRING)
+ .addSingleValueDimension("myCol2", DataType.INT)
.build();
final TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
.setTableName(TABLE_NAME).build();
@@ -1839,7 +1825,7 @@ public class TableConfigUtilsTest {
@Test
public void testValidateFieldConfigSingleEntry() {
final Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol1", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol1", DataType.STRING)
.build();
final TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
.setTableName(TABLE_NAME).build();
@@ -1857,8 +1843,8 @@ public class TableConfigUtilsTest {
@Test
public void testValidateBFOnBoolean() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.BOOLEAN)
- .addSingleValueDimension("mycol2", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol", DataType.BOOLEAN)
+ .addSingleValueDimension("mycol2", DataType.STRING)
.build();
TableConfig tableconfig1 = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
@@ -1883,11 +1869,11 @@ public class TableConfigUtilsTest {
@Test
public void testValidateIndexingConfig() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("bytesCol", FieldSpec.DataType.BYTES)
- .addSingleValueDimension("intCol", FieldSpec.DataType.INT)
- .addSingleValueDimension("bigDecimalCol",
FieldSpec.DataType.BIG_DECIMAL)
- .addMultiValueDimension("multiValCol", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("bytesCol", DataType.BYTES)
+ .addSingleValueDimension("intCol", DataType.INT)
+ .addSingleValueDimension("bigDecimalCol", DataType.BIG_DECIMAL)
+ .addMultiValueDimension("multiValCol", DataType.STRING)
.build();
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setBloomFilterColumns(Arrays.asList("myCol2"))
@@ -2230,10 +2216,10 @@ public class TableConfigUtilsTest {
@Test
public void testValidateStarTreeIndexDuplicateFunctionColumnPair() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("bytesCol", FieldSpec.DataType.BYTES)
- .addSingleValueDimension("intCol", FieldSpec.DataType.INT)
- .addMultiValueDimension("multiValCol", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("bytesCol", DataType.BYTES)
+ .addSingleValueDimension("intCol", DataType.INT)
+ .addMultiValueDimension("multiValCol", DataType.STRING)
.build();
StarTreeIndexConfig starTreeIndexConfig =
@@ -2271,7 +2257,7 @@ public class TableConfigUtilsTest {
@Test
public void testValidateRetentionConfig() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("myCol", DataType.STRING)
.build();
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setRetentionTimeUnit("hours")
@@ -2295,83 +2281,76 @@ public class TableConfigUtilsTest {
@Test
public void testValidateDedupConfig() {
- Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ TableConfig validTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setDedupConfig(new DedupConfig())
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
+ Schema validSchema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
+ .build();
+ TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig,
validSchema);
+
// OFFLINE table should fail because dedup is not yet supported for
offline tables
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setTimeColumnName(TIME_COLUMN)
- .setDedupConfig(new DedupConfig())
+ .setDedupConfig(new
DedupConfig()).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Dedup is not supported for OFFLINE table.
Only upsert is supported for OFFLINE"
+ " table");
}
- tableConfig =
- new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setDedupConfig(new
DedupConfig()).build();
+ Schema schema =
+ new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol",
DataType.STRING).build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Upsert/Dedup table must have primary key
columns in the schema");
}
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addMultiValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addMultiValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
- tableConfig =
- new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setDedupConfig(new
DedupConfig()).build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Upsert/Dedup primary key column: myCol
cannot be of multi-value type");
}
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
- .build();
- Map<String, String> streamConfigs = getStreamConfigs();
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setDedupConfig(new DedupConfig())
- .setStreamConfigs(streamConfigs)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
"Upsert/Dedup table must use strict replica-group (i.e.
strictReplicaGroup) based routing");
}
- StarTreeIndexConfig starTreeIndexConfig = new
StarTreeIndexConfig(Lists.newArrayList("myCol"), null,
- List.of(
- new AggregationFunctionColumnPair(AggregationFunctionType.COUNT,
"myCol").toColumnName()), null, 10);
+
+ StarTreeIndexConfig starTreeIndexConfig = new
StarTreeIndexConfig(List.of("myCol"), null,
+ List.of(new
AggregationFunctionColumnPair(AggregationFunctionType.COUNT,
"myCol").toColumnName()), null, 10);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setDedupConfig(new DedupConfig())
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
- .setStarTreeIndexConfigs(Lists.newArrayList(starTreeIndexConfig))
- .setStreamConfigs(streamConfigs)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
// Dedup and upsert can't be enabled simultaneously
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setDedupConfig(new DedupConfig())
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
- .setStreamConfigs(streamConfigs)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "A table can have either Upsert or Dedup
enabled, but not both");
@@ -2382,15 +2361,12 @@ public class TableConfigUtilsTest {
dedupConfig.setMetadataTTL(10);
dedupConfig.setDedupTimeColumn(TIME_COLUMN);
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
- tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setDedupConfig(dedupConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
- .setStreamConfigs(streamConfigs)
+ tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
+
.setDedupConfig(dedupConfig).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
// Should not throw an exception - TIMESTAMP is a valid type for
dedupTimeColumn
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2400,9 +2376,9 @@ public class TableConfigUtilsTest {
public void testValidateInvalidDedupConfigs() {
// Invalid STRING time column
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.STRING,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.STRING, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
Map<String, String> streamConfigs = getStreamConfigs();
@@ -2410,8 +2386,7 @@ public class TableConfigUtilsTest {
dedupConfig.setMetadataTTL(10);
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setDedupConfig(dedupConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setTimeColumnName(TIME_COLUMN)
.setStreamConfigs(streamConfigs)
.build();
@@ -2427,15 +2402,14 @@ public class TableConfigUtilsTest {
@Test
public void testValidateUpsertConfigWithMd5Disabled() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setHashFunction(HashFunction.MD5);
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(getStreamConfigs())
.build();
try {
@@ -2452,15 +2426,14 @@ public class TableConfigUtilsTest {
@Test
public void testValidateDedupConfigWithMd5Disabled() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
DedupConfig dedupConfig = new DedupConfig();
dedupConfig.setHashFunction(HashFunction.MD5);
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setDedupConfig(dedupConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(getStreamConfigs())
.build();
try {
@@ -2477,18 +2450,15 @@ public class TableConfigUtilsTest {
@Test
public void testValidateDedupConfigWithMd5DisabledAllowsUpsertModeNoneMd5() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
DedupConfig dedupConfig = new DedupConfig();
dedupConfig.setHashFunction(HashFunction.NONE);
UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.NONE);
upsertConfig.setHashFunction(HashFunction.MD5);
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setUpsertConfig(upsertConfig)
- .setDedupConfig(dedupConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+
.setUpsertConfig(upsertConfig).setDedupConfig(dedupConfig).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(getStreamConfigs())
.build();
try {
@@ -2501,18 +2471,27 @@ public class TableConfigUtilsTest {
@Test
public void testValidateUpsertConfig() {
- Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
+ SegmentPartitionConfig segmentPartitionConfig =
+ new SegmentPartitionConfig(Map.of("myCol", new
ColumnPartitionConfig("murmur", 4)));
+ TableConfig validTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setUpsertConfig(upsertConfig)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ Schema validSchema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
+ TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig,
validSchema);
+
// OFFLINE table without segment partition config should fail with
partition config error
- UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
- .setTimeColumnName(TIME_COLUMN)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2524,35 +2503,35 @@ public class TableConfigUtilsTest {
UpsertConfig partialUpsertConfig = new
UpsertConfig(UpsertConfig.Mode.PARTIAL);
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setUpsertConfig(partialUpsertConfig)
- .setTimeColumnName(TIME_COLUMN)
+ .setSegmentPartitionConfig(segmentPartitionConfig)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Partial upsert is not supported for
OFFLINE table");
}
- tableConfig =
- new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig).build();
+ validTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setUpsertConfig(upsertConfig)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig,
validSchema);
+
+ Schema schema =
+ new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol",
DataType.STRING).build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(validTableConfig, schema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Upsert/Dedup table must have primary key
columns in the schema");
}
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
- .build();
- Map<String, String> streamConfigs = getStreamConfigs();
- tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setUpsertConfig(upsertConfig)
- .setStreamConfigs(streamConfigs)
- .build();
+ tableConfig =
+ new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig).build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2564,12 +2543,11 @@ public class TableConfigUtilsTest {
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
.setStreamConfigs(getStreamConfigs())
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME",
"T2_REALTIME"))
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(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");
@@ -2580,44 +2558,39 @@ public class TableConfigUtilsTest {
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
.setStreamConfigs(getStreamConfigs())
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME",
"T1_REALTIME"))
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(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");
}
// Instance assignment config with just CONSUMING instance partitions
configured should be allowed
- InstanceAssignmentConfig instanceAssignmentConfig =
Mockito.mock(InstanceAssignmentConfig.class);
+ InstanceAssignmentConfig instanceAssignmentConfig =
mock(InstanceAssignmentConfig.class);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
.setStreamConfigs(getStreamConfigs())
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
instanceAssignmentConfig))
.build();
-
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
// Instance assignment config with CONSUMING and COMPLETED instance
partitions configured should not be allowed
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
.setStreamConfigs(getStreamConfigs())
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setInstanceAssignmentConfigMap(Map.of(InstancePartitionsType.CONSUMING.name(),
instanceAssignmentConfig,
InstancePartitionsType.COMPLETED.name(), instanceAssignmentConfig))
.build();
-
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail("Instance assignment config with COMPLETED instance partitions must
not be allowed with upsert");
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "COMPLETED instance partitions can't be
configured for upsert / dedup tables");
@@ -2628,80 +2601,69 @@ public class TableConfigUtilsTest {
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
.setStreamConfigs(getStreamConfigs())
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setTagOverrideConfig(new TagOverrideConfig(null, null))
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
- .setStreamConfigs(streamConfigs)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
- StarTreeIndexConfig starTreeIndexConfig = new
StarTreeIndexConfig(Lists.newArrayList("myCol"), null,
- List.of(
- new AggregationFunctionColumnPair(AggregationFunctionType.COUNT,
"myCol").toColumnName()), null, 10);
+ StarTreeIndexConfig starTreeIndexConfig = new
StarTreeIndexConfig(List.of("myCol"), null,
+ List.of(new
AggregationFunctionColumnPair(AggregationFunctionType.COUNT,
"myCol").toColumnName()), null, 10);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
- .setStarTreeIndexConfigs(Lists.newArrayList(starTreeIndexConfig))
- .setStreamConfigs(streamConfigs)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "The upsert table cannot have star-tree
index.");
}
- //With Aggregate Metrics
+ // Metrics aggregation is incompatible with upsert; validated in
validateMetricsAggregation via
+ // validateIngestionConfig. With the aggregateMetrics flag:
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
- .setStreamConfigs(streamConfigs)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setAggregateMetrics(true)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateIngestionConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Metrics aggregation and upsert cannot be
enabled together");
}
- //With aggregation Configs in Ingestion Config
+ // With aggregationConfigs in the Ingestion Config:
IngestionConfig ingestionConfig = new IngestionConfig();
ingestionConfig.setAggregationConfigs(List.of(new
AggregationConfig("twiceSum", "SUM(twice)")));
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
- .setStreamConfigs(streamConfigs)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setIngestionConfig(ingestionConfig)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateIngestionConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(), "Metrics aggregation and upsert cannot be
enabled together");
}
- //With aggregation Configs in Ingestion Config and IndexingConfig at the
same time
+ // With aggregationConfigs in the Ingestion Config and aggregateMetrics in
the Indexing Config at the same time:
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
- .setStreamConfigs(streamConfigs)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setAggregateMetrics(true)
.setIngestionConfig(ingestionConfig)
.build();
try {
- TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
+ TableConfigUtils.validateIngestionConfig(tableConfig, validSchema);
fail();
} catch (IllegalStateException e) {
assertEquals(e.getMessage(),
@@ -2714,24 +2676,20 @@ public class TableConfigUtilsTest {
String mvCol = "mvCol";
String timestampCol = "timestampCol";
String invalidCol = "invalidCol";
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension(stringTypeDelCol, FieldSpec.DataType.STRING)
- .addSingleValueDimension(delCol, FieldSpec.DataType.BOOLEAN)
- .addSingleValueDimension(timestampCol, FieldSpec.DataType.TIMESTAMP)
- .addMultiValueDimension(mvCol, FieldSpec.DataType.STRING)
- .build();
- streamConfigs = getStreamConfigs();
-
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDeleteRecordColumn(stringTypeDelCol);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("myPkCol", DataType.STRING)
+ .addSingleValueDimension(stringTypeDelCol, DataType.STRING)
+ .addSingleValueDimension(delCol, DataType.BOOLEAN)
+ .addSingleValueDimension(timestampCol, DataType.TIMESTAMP)
+ .addMultiValueDimension(mvCol, DataType.STRING)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2742,10 +2700,8 @@ public class TableConfigUtilsTest {
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDeleteRecordColumn(delCol);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2756,10 +2712,8 @@ public class TableConfigUtilsTest {
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDeleteRecordColumn(timestampCol);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2771,10 +2725,8 @@ public class TableConfigUtilsTest {
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDeleteRecordColumn(invalidCol);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2786,10 +2738,8 @@ public class TableConfigUtilsTest {
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDeleteRecordColumn(mvCol);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2798,24 +2748,20 @@ public class TableConfigUtilsTest {
assertEquals(e.getMessage(), "The deleteRecordColumn - mvCol must be a
single-valued column");
}
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addMultiValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension(stringTypeDelCol, FieldSpec.DataType.STRING)
- .addSingleValueDimension(delCol, FieldSpec.DataType.BOOLEAN)
- .addSingleValueDimension(timestampCol, FieldSpec.DataType.TIMESTAMP)
- .addMultiValueDimension(mvCol, FieldSpec.DataType.STRING)
- .build();
- streamConfigs = getStreamConfigs();
-
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDeleteRecordColumn(stringTypeDelCol);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addMultiValueDimension("myPkCol", DataType.STRING)
+ .addSingleValueDimension(stringTypeDelCol, DataType.STRING)
+ .addSingleValueDimension(delCol, DataType.BOOLEAN)
+ .addSingleValueDimension(timestampCol, DataType.TIMESTAMP)
+ .addMultiValueDimension(mvCol, DataType.STRING)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2825,21 +2771,19 @@ public class TableConfigUtilsTest {
}
// upsert deleted-keys-ttl configs with no deleted column
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension(delCol, FieldSpec.DataType.BOOLEAN)
- .build();
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDeletedKeysTTL(3600);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("myPkCol", DataType.STRING)
+ .addSingleValueDimension(delCol, DataType.BOOLEAN)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2847,21 +2791,19 @@ public class TableConfigUtilsTest {
assertEquals(e.getMessage(), "Deleted Keys TTL can only be enabled with
deleteRecordColumn set.");
}
- upsertConfig.setDeleteRecordColumn(delCol);
// multiple comparison columns set for deleted-keys-ttl
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .addSingleValueDimension(delCol, FieldSpec.DataType.BOOLEAN)
- .build();
- upsertConfig.setComparisonColumns(Lists.newArrayList(TIME_COLUMN,
"myCol"));
+ upsertConfig.setDeleteRecordColumn(delCol);
+ upsertConfig.setComparisonColumns(List.of(TIME_COLUMN, "myCol"));
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("myPkCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .addSingleValueDimension(delCol, DataType.BOOLEAN)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2870,12 +2812,10 @@ public class TableConfigUtilsTest {
}
// comparison column with non-numeric type
- upsertConfig.setComparisonColumns(Lists.newArrayList("myCol"));
+ upsertConfig.setComparisonColumns(List.of("myCol"));
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2885,34 +2825,28 @@ public class TableConfigUtilsTest {
}
// time column as comparison column
- upsertConfig.setComparisonColumns(Lists.newArrayList(TIME_COLUMN));
+ upsertConfig.setComparisonColumns(List.of(TIME_COLUMN));
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
// upsert out-of-order configs
String outOfOrderRecordColumn = "outOfOrderRecordColumn";
boolean dropOutOfOrderRecord = true;
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension(outOfOrderRecordColumn,
FieldSpec.DataType.BOOLEAN)
- .build();
- streamConfigs = getStreamConfigs();
-
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDropOutOfOrderRecord(dropOutOfOrderRecord);
upsertConfig.setOutOfOrderRecordColumn(outOfOrderRecordColumn);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("myPkCol", DataType.STRING)
+ .addSingleValueDimension(outOfOrderRecordColumn, DataType.BOOLEAN)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2922,21 +2856,17 @@ public class TableConfigUtilsTest {
}
// outOfOrderRecordColumn not of type BOOLEAN
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension(outOfOrderRecordColumn,
FieldSpec.DataType.STRING)
- .build();
- streamConfigs = getStreamConfigs();
-
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setOutOfOrderRecordColumn(outOfOrderRecordColumn);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("myPkCol", DataType.STRING)
+ .addSingleValueDimension(outOfOrderRecordColumn, DataType.STRING)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2945,21 +2875,18 @@ public class TableConfigUtilsTest {
}
// test dropOutOfOrderRecord cannot be enabled with SYNC consistency mode
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .build();
- streamConfigs = getStreamConfigs();
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setDropOutOfOrderRecord(true);
upsertConfig.setConsistencyMode(UpsertConfig.ConsistencyMode.SYNC);
upsertConfig.setNewSegmentTrackingTimeMs(60000L);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("myPkCol", DataType.STRING)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2974,10 +2901,8 @@ public class TableConfigUtilsTest {
upsertConfig.setConsistencyMode(UpsertConfig.ConsistencyMode.SNAPSHOT);
upsertConfig.setNewSegmentTrackingTimeMs(60000L);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -2991,29 +2916,25 @@ public class TableConfigUtilsTest {
upsertConfig.setDropOutOfOrderRecord(true);
upsertConfig.setConsistencyMode(UpsertConfig.ConsistencyMode.NONE);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
// test outOfOrderRecordColumn cannot be enabled with SYNC consistency mode
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("isOutOfOrder", FieldSpec.DataType.BOOLEAN)
- .build();
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setOutOfOrderRecordColumn("isOutOfOrder");
upsertConfig.setConsistencyMode(UpsertConfig.ConsistencyMode.SYNC);
upsertConfig.setNewSegmentTrackingTimeMs(60000L);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .build();
+ schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .setPrimaryKeyColumns(List.of("myPkCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addSingleValueDimension("myPkCol", DataType.STRING)
+ .addSingleValueDimension("isOutOfOrder", DataType.BOOLEAN)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3028,10 +2949,8 @@ public class TableConfigUtilsTest {
upsertConfig.setConsistencyMode(UpsertConfig.ConsistencyMode.SNAPSHOT);
upsertConfig.setNewSegmentTrackingTimeMs(60000L);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3042,20 +2961,12 @@ public class TableConfigUtilsTest {
}
// test outOfOrderRecordColumn is allowed with NONE consistency mode
(should not throw)
- schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("myPkCol", FieldSpec.DataType.STRING)
- .addSingleValueDimension("isOutOfOrder", FieldSpec.DataType.BOOLEAN)
- .build();
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setOutOfOrderRecordColumn("isOutOfOrder");
upsertConfig.setConsistencyMode(UpsertConfig.ConsistencyMode.NONE);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3064,10 +2975,8 @@ public class TableConfigUtilsTest {
upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
upsertConfig.setMetadataTTL(1.0);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3081,10 +2990,8 @@ public class TableConfigUtilsTest {
upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
upsertConfig.setPreload(Enablement.ENABLE);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3098,10 +3005,8 @@ public class TableConfigUtilsTest {
upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
upsertConfig.setDeletedKeysTTL(0);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3116,10 +3021,8 @@ public class TableConfigUtilsTest {
upsertConfig.setDeletedKeysTTL(100);
upsertConfig.setSnapshot(Enablement.DISABLE);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3134,10 +3037,8 @@ public class TableConfigUtilsTest {
upsertConfig.setDeletedKeysTTL(100);
upsertConfig.setSnapshot(Enablement.ENABLE);
tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
- .setStreamConfigs(streamConfigs)
.setUpsertConfig(upsertConfig)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -3150,10 +3051,10 @@ public class TableConfigUtilsTest {
@Test
public void testValidatePartialUpsertConfig() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol1", FieldSpec.DataType.LONG)
- .addSingleValueDimension("myCol2", FieldSpec.DataType.STRING)
- .addDateTime("myTimeCol", FieldSpec.DataType.LONG, "1:DAYS:EPOCH",
"1:DAYS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol1"))
+ .addSingleValueDimension("myCol1", DataType.LONG)
+ .addSingleValueDimension("myCol2", DataType.STRING)
+ .addDateTime("myTimeCol", DataType.LONG, "1:DAYS:EPOCH", "1:DAYS")
+ .setPrimaryKeyColumns(List.of("myCol1"))
.build();
Map<String, String> streamConfigs = getStreamConfigs();
@@ -3166,8 +3067,7 @@ public class TableConfigUtilsTest {
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setUpsertConfig(partialUpsertConfig)
.setNullHandlingEnabled(true)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(streamConfigs)
.build();
try {
@@ -3184,8 +3084,7 @@ public class TableConfigUtilsTest {
.setTimeColumnName("myCol2")
.setUpsertConfig(partialUpsertConfig)
.setNullHandlingEnabled(true)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(streamConfigs)
.build();
try {
@@ -3200,8 +3099,7 @@ public class TableConfigUtilsTest {
.setTimeColumnName("timeCol")
.setUpsertConfig(partialUpsertConfig)
.setNullHandlingEnabled(true)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(streamConfigs)
.build();
try {
@@ -3268,17 +3166,16 @@ public class TableConfigUtilsTest {
partialUpsertConfig.setComparisonColumn("myCol2");
Schema.SchemaBuilder schemaBuilder = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol1", FieldSpec.DataType.LONG)
- .addSingleValueDimension("myCol2", FieldSpec.DataType.STRING)
- .addDateTime("myTimeCol", FieldSpec.DataType.LONG, "1:DAYS:EPOCH",
"1:DAYS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol1"));
+ .addSingleValueDimension("myCol1", DataType.LONG)
+ .addSingleValueDimension("myCol2", DataType.STRING)
+ .addDateTime("myTimeCol", DataType.LONG, "1:DAYS:EPOCH", "1:DAYS")
+ .setPrimaryKeyColumns(List.of("myCol1"));
TableConfigBuilder tableConfigBuilder = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName("timeCol")
.setUpsertConfig(partialUpsertConfig)
.setNullHandlingEnabled(true)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.setStreamConfigs(streamConfigs);
configureFun.accept(tableConfigBuilder, schemaBuilder);
@@ -3339,7 +3236,7 @@ public class TableConfigUtilsTest {
@Test
public void testValidateInstancePartitionsMap() {
- InstanceAssignmentConfig instanceAssignmentConfig =
Mockito.mock(InstanceAssignmentConfig.class);
+ InstanceAssignmentConfig instanceAssignmentConfig =
mock(InstanceAssignmentConfig.class);
TableConfig tableConfigWithoutInstancePartitionsMap =
new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
@@ -3415,9 +3312,9 @@ public class TableConfigUtilsTest {
public void testValidateTTLConfigForUpsertConfig() {
// Default comparison column (timestamp)
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setMetadataTTL(3600);
@@ -3447,7 +3344,7 @@ public class TableConfigUtilsTest {
// Invalid comparison columns: multiple comparison columns are not
supported for TTL-enabled upsert table.
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
- upsertConfig.setComparisonColumns(Lists.newArrayList(TIME_COLUMN,
"myCol"));
+ upsertConfig.setComparisonColumns(List.of(TIME_COLUMN, "myCol"));
upsertConfig.setSnapshot(Enablement.ENABLE);
upsertConfig.setMetadataTTL(3600);
TableConfig tableConfigWithInvalidComparisonColumn2 =
@@ -3464,9 +3361,9 @@ public class TableConfigUtilsTest {
// Invalid config with TTLConfig but Snapshot is not enabled
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setMetadataTTL(3600);
@@ -3484,9 +3381,9 @@ public class TableConfigUtilsTest {
// Invalid STRING time column
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.STRING,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.STRING, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setMetadataTTL(3600);
@@ -3504,9 +3401,9 @@ public class TableConfigUtilsTest {
// Valid TIMESTAMP time column
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("myCol"))
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("myCol"))
.build();
upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
upsertConfig.setMetadataTTL(3600);
@@ -3538,10 +3435,10 @@ public class TableConfigUtilsTest {
// Call validate with a table-config with replicaGroupStrategyConfig and
without replicaGroupPartitionConfig.
TableConfigUtils.validatePartitionedReplicaGroupInstance(tableConfigWithReplicaGroupStrategyConfig);
- InstanceAssignmentConfig instanceAssignmentConfig =
Mockito.mock(InstanceAssignmentConfig.class);
+ InstanceAssignmentConfig instanceAssignmentConfig =
mock(InstanceAssignmentConfig.class);
InstanceReplicaGroupPartitionConfig instanceReplicaGroupPartitionConfig =
new InstanceReplicaGroupPartitionConfig(true, 0, 0, 0, 2, 0, false,
partitionColumn);
- Mockito.doReturn(instanceReplicaGroupPartitionConfig)
+ doReturn(instanceReplicaGroupPartitionConfig)
.when(instanceAssignmentConfig)
.getReplicaGroupPartitionConfig();
@@ -3561,7 +3458,7 @@ public class TableConfigUtilsTest {
@Test
public void testValidateImplicitRealtimeTablePartitionSelectorConfigs() {
- InstanceAssignmentConfig instanceAssignmentConfig =
Mockito.mock(InstanceAssignmentConfig.class);
+ InstanceAssignmentConfig instanceAssignmentConfig =
mock(InstanceAssignmentConfig.class);
when(instanceAssignmentConfig.getPartitionSelector()).thenReturn(
InstanceAssignmentConfig.PartitionSelector.IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR);
@@ -3574,7 +3471,7 @@ public class TableConfigUtilsTest {
e.getMessage().contains("IMPLICIT_REALTIME_TABLE_PARTITION_SELECTOR
can only be used for REALTIME tables"));
InstanceReplicaGroupPartitionConfig instanceReplicaGroupPartitionConfig =
- Mockito.mock(InstanceReplicaGroupPartitionConfig.class);
+ mock(InstanceReplicaGroupPartitionConfig.class);
when(instanceReplicaGroupPartitionConfig.isReplicaGroupBased()).thenReturn(true);
when(instanceAssignmentConfig.getReplicaGroupPartitionConfig()).thenReturn(instanceReplicaGroupPartitionConfig);
TableConfig tableConfig2 = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
@@ -4062,8 +3959,8 @@ public class TableConfigUtilsTest {
@Test
public void testValidateTierConfigsForDedupTable() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
// Validate that table's timeColumn must be used as dedupTimeColumn.
DedupConfig dedupConfig = new DedupConfig();
@@ -4084,7 +3981,7 @@ public class TableConfigUtilsTest {
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setDedupConfig(dedupConfig)
- .setTierConfigList(Lists.newArrayList(new TierConfig("tier1",
TierFactory.FIXED_SEGMENT_SELECTOR_TYPE, "", null,
+ .setTierConfigList(List.of(new TierConfig("tier1",
TierFactory.FIXED_SEGMENT_SELECTOR_TYPE, "", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE.toLowerCase(),
"tier1_tag_OFFLINE", null, null)))
.build();
try {
@@ -4100,7 +3997,7 @@ public class TableConfigUtilsTest {
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setDedupConfig(dedupConfig)
- .setTierConfigList(Lists.newArrayList(
+ .setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"50s", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE.toLowerCase(),
"tier1_tag_OFFLINE", null, null),
new TierConfig("tier2",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE.toLowerCase(), "100s", null,
@@ -4110,8 +4007,8 @@ public class TableConfigUtilsTest {
TableConfigUtils.validateTTLAndTierConfigsForDedupTable(tableConfig,
schema);
// Got TTL=100s vs. minAge=50s, testing with timeColumn of sec/epoch
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:SECONDS:EPOCH",
"1:SECONDS")
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:SECONDS:EPOCH",
"1:SECONDS")
.build();
try {
TableConfigUtils.validateTTLAndTierConfigsForDedupTable(tableConfig,
schema);
@@ -4125,15 +4022,15 @@ public class TableConfigUtilsTest {
tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setDedupConfig(dedupConfig)
- .setTierConfigList(Lists.newArrayList(
+ .setTierConfigList(List.of(
new TierConfig("tier1", TierFactory.TIME_SEGMENT_SELECTOR_TYPE,
"50s", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE.toLowerCase(),
"tier1_tag_OFFLINE", null, null),
new TierConfig("tier2",
TierFactory.TIME_SEGMENT_SELECTOR_TYPE.toLowerCase(), "100s", null,
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tier2_tag_OFFLINE",
null, null)))
.build();
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTime(TIME_COLUMN, DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.build();
try {
TableConfigUtils.validateTTLAndTierConfigsForDedupTable(tableConfig,
schema);
@@ -4193,12 +4090,12 @@ public class TableConfigUtilsTest {
@Test
public void testValidatePostPartialUpsertTransformConfigs() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("pk", FieldSpec.DataType.STRING)
- .addMetric("score", FieldSpec.DataType.DOUBLE)
- .addMetric("bonus", FieldSpec.DataType.DOUBLE)
- .addMetric("total", FieldSpec.DataType.DOUBLE)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("pk"))
+ .addSingleValueDimension("pk", DataType.STRING)
+ .addMetric("score", DataType.DOUBLE)
+ .addMetric("bonus", DataType.DOUBLE)
+ .addMetric("total", DataType.DOUBLE)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("pk"))
.build();
Map<String, String> streamConfigs = getStreamConfigs();
@@ -4209,9 +4106,9 @@ public class TableConfigUtilsTest {
TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(upsertConfig)
+ .setNullHandlingEnabled(true)
.setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
@@ -4223,8 +4120,7 @@ public class TableConfigUtilsTest {
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(fullUpsertConfig)
.setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(fullTableConfig, schema);
@@ -4240,10 +4136,7 @@ public class TableConfigUtilsTest {
new TransformConfig("total", "minus(score,bonus)")));
TableConfig dupTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
- .setUpsertConfig(dupConfig)
- .setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+
.setUpsertConfig(dupConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(dupTableConfig, schema);
@@ -4260,8 +4153,7 @@ public class TableConfigUtilsTest {
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(missingColConfig)
.setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(missingColTableConfig,
schema);
@@ -4276,10 +4168,7 @@ public class TableConfigUtilsTest {
List.of(new TransformConfig("total", "plus(total,bonus)")));
TableConfig selfRefTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
- .setUpsertConfig(selfRefConfig)
- .setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+
.setUpsertConfig(selfRefConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(selfRefTableConfig,
schema);
@@ -4294,10 +4183,7 @@ public class TableConfigUtilsTest {
List.of(new TransformConfig("pk", "concat(score,bonus)")));
TableConfig pkTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
- .setUpsertConfig(pkConfig)
- .setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+
.setUpsertConfig(pkConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(pkTableConfig, schema);
@@ -4313,10 +4199,7 @@ public class TableConfigUtilsTest {
List.of(new TransformConfig(TIME_COLUMN, "plus(score,bonus)")));
TableConfig compTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
- .setUpsertConfig(compConfig)
- .setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+
.setUpsertConfig(compConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(compTableConfig, schema);
@@ -4334,8 +4217,7 @@ public class TableConfigUtilsTest {
.setTimeColumnName(TIME_COLUMN)
.setUpsertConfig(deleteColConfig)
.setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(deleteColTableConfig,
schema);
@@ -4346,13 +4228,13 @@ public class TableConfigUtilsTest {
// Transform targeting out-of-order record column should fail
Schema schemaWithBool = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("pk", FieldSpec.DataType.STRING)
- .addMetric("score", FieldSpec.DataType.DOUBLE)
- .addMetric("bonus", FieldSpec.DataType.DOUBLE)
- .addMetric("total", FieldSpec.DataType.DOUBLE)
- .addSingleValueDimension("isOoo", FieldSpec.DataType.BOOLEAN)
- .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
- .setPrimaryKeyColumns(Lists.newArrayList("pk"))
+ .addSingleValueDimension("pk", DataType.STRING)
+ .addMetric("score", DataType.DOUBLE)
+ .addMetric("bonus", DataType.DOUBLE)
+ .addMetric("total", DataType.DOUBLE)
+ .addSingleValueDimension("isOoo", DataType.BOOLEAN)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("pk"))
.build();
UpsertConfig oooColConfig = new UpsertConfig(UpsertConfig.Mode.PARTIAL);
oooColConfig.setOutOfOrderRecordColumn("isOoo");
@@ -4360,10 +4242,7 @@ public class TableConfigUtilsTest {
List.of(new TransformConfig("isOoo", "plus(bonus,total)")));
TableConfig oooColTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
.setTimeColumnName(TIME_COLUMN)
- .setUpsertConfig(oooColConfig)
- .setStreamConfigs(streamConfigs)
- .setRoutingConfig(
- new RoutingConfig(null, null,
RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
+
.setUpsertConfig(oooColConfig).setStreamConfigs(streamConfigs).setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
.build();
try {
TableConfigUtils.validateUpsertAndDedupConfig(oooColTableConfig,
schemaWithBool);
@@ -4544,14 +4423,14 @@ public class TableConfigUtilsTest {
// Valid: no '$' in any column name — OPEN_STRUCT and regular column both
clean.
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("event_type", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("event_type", DataType.STRING)
.addOpenStruct("payload", Map.of())
.build();
TableConfigUtils.validate(tableConfig, schema);
// Invalid: non-OPEN_STRUCT column contains '$'.
Schema schemaWithDollarInNonOpenStruct = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("event$type", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("event$type", DataType.STRING)
.addOpenStruct("payload", Map.of())
.build();
try {
@@ -4565,7 +4444,7 @@ public class TableConfigUtilsTest {
// 'metrics$v2' would produce children 'metrics$v2$key', and
parseParentColumn() would split
// on the first '$' and return 'metrics' instead of 'metrics$v2',
corrupting the name mapping.
Schema schemaWithDollarInOpenStructParent = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension("event_type", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("event_type", DataType.STRING)
.addOpenStruct("metrics$v2", Map.of())
.build();
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
