You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ne...@apache.org on 2020/11/18 04:03:10 UTC
[incubator-pinot] branch master updated: Add stream and batch to
ingestionConfig (#6247)
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 9eaea99 Add stream and batch to ingestionConfig (#6247)
9eaea99 is described below
commit 9eaea998fc74d4b66f5dc9ea6c2de9b8f758aae5
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Tue Nov 17 20:02:47 2020 -0800
Add stream and batch to ingestionConfig (#6247)
---
.../routing/timeboundary/TimeBoundaryManager.java | 3 +-
.../common/utils/config/TableConfigUtils.java | 2 +-
.../common/utils/config/TableConfigSerDeTest.java | 27 +++-
.../helix/core/PinotHelixResourceManager.java | 14 +-
.../helix/core/PinotTableIdealStateBuilder.java | 3 +-
.../RealtimeToOfflineSegmentsTaskGenerator.java | 5 +-
.../realtime/PinotLLCRealtimeSegmentManager.java | 9 +-
.../core/realtime/PinotRealtimeSegmentManager.java | 7 +-
.../core/realtime/SegmentCompletionManager.java | 5 +-
.../helix/core/rebalance/TableRebalancer.java | 3 +-
.../helix/core/relocation/SegmentRelocator.java | 4 +-
.../helix/core/retention/RetentionManager.java | 3 +-
.../controller/util/TableRetentionValidator.java | 6 +-
.../validation/OfflineSegmentIntervalChecker.java | 4 +-
.../RealtimeSegmentValidationManager.java | 6 +-
.../PinotLLCRealtimeSegmentManagerTest.java | 18 +--
.../realtime/HLRealtimeSegmentDataManager.java | 3 +-
.../realtime/LLRealtimeSegmentDataManager.java | 4 +-
.../org/apache/pinot/core/util/IngestionUtils.java | 5 +-
.../apache/pinot/core/util/ReplicationUtils.java | 5 +-
.../apache/pinot/core/util/TableConfigUtils.java | 49 ++++--
.../ExpressionTransformerTest.java | 10 +-
.../recordtransformer/RecordTransformerTest.java | 18 ++-
.../SegmentGenerationWithFilterRecordsTest.java | 10 +-
.../apache/pinot/core/util/IngestionUtilsTest.java | 41 ++---
.../apache/pinot/core/util/SchemaUtilsTest.java | 6 +-
.../pinot/core/util/TableConfigUtilsTest.java | 120 ++++++++++++---
.../tests/BaseClusterIntegrationTest.java | 47 +++---
.../IngestionConfigHybridIntegrationTest.java | 17 +-
.../LuceneRealtimeClusterIntegrationTest.java | 1 -
.../tests/MapTypeClusterIntegrationTest.java | 7 +-
.../RealtimeToOfflineSegmentsTaskExecutorTest.java | 6 +-
.../batch/common/SegmentGenerationTaskRunner.java | 8 +-
.../pinot/hadoop/job/HadoopSegmentCreationJob.java | 3 +-
.../hadoop/job/HadoopSegmentPreprocessingJob.java | 11 +-
.../hadoop/job/mappers/SegmentCreationMapper.java | 6 +-
.../spark/jobs/SparkSegmentCreationFunction.java | 4 +-
.../pinot/spark/jobs/SparkSegmentCreationJob.java | 3 +-
.../pinot/spi/config/table/IndexingConfig.java | 5 +
.../SegmentsValidationAndRetentionConfig.java | 9 ++
.../apache/pinot/spi/config/table/TableConfig.java | 1 +
.../table/ingestion/BatchIngestionConfig.java | 65 ++++++++
.../table/{ => ingestion}/IngestionConfig.java | 30 +++-
.../table/ingestion/StreamIngestionConfig.java | 46 ++++++
.../pinot/spi/ingestion/batch/BatchConfig.java | 133 ++++++++++++++++
.../spi/ingestion/batch/BatchConfigProperties.java | 49 ++++++
.../pinot/spi/utils/IngestionConfigUtils.java | 95 ++++++++++++
.../spi/utils/builder/TableConfigBuilder.java | 2 +-
.../pinot/spi/ingestion/batch/BatchConfigTest.java | 171 +++++++++++++++++++++
.../pinot/spi/utils/IngestionConfigUtilsTest.java | 149 ++++++++++++++++++
.../segment/converter/SegmentMergeCommand.java | 5 +-
51 files changed, 1080 insertions(+), 183 deletions(-)
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java
index 4a8c357..3097a1b 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java
@@ -38,6 +38,7 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DateTimeFormatSpec;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -83,7 +84,7 @@ public class TimeBoundaryManager {
// For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
// (maxEndTime - 1 DAY)
boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY
- .equalsIgnoreCase(tableConfig.getValidationConfig().getSegmentPushFrequency())
+ .equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig))
&& _timeFormatSpec.getColumnUnit() != TimeUnit.DAYS;
_timeOffsetMs = isHourlyTable ? TimeUnit.HOURS.toMillis(1) : TimeUnit.DAYS.toMillis(1);
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/TableConfigUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/TableConfigUtils.java
index e82533c..1685e74 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/TableConfigUtils.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/TableConfigUtils.java
@@ -28,7 +28,7 @@ import java.util.Map;
import org.apache.helix.ZNRecord;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.IndexingConfig;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.QueryConfig;
import org.apache.pinot.spi.config.table.QuotaConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeTest.java b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeTest.java
index 757f239..bd7df17 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeTest.java
@@ -21,6 +21,7 @@ package org.apache.pinot.common.utils.config;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Lists;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -30,7 +31,8 @@ import org.apache.pinot.common.tier.TierFactory;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.spi.config.table.CompletionConfig;
import org.apache.pinot.spi.config.table.FieldConfig;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.QueryConfig;
import org.apache.pinot.spi.config.table.QuotaConfig;
import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
@@ -48,6 +50,7 @@ import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
import org.apache.pinot.spi.config.table.assignment.InstanceTagPoolConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
@@ -261,7 +264,17 @@ public class TableConfigSerDeTest {
// With ingestion config
List<TransformConfig> transformConfigs =
Lists.newArrayList(new TransformConfig("bar", "func(moo)"), new TransformConfig("zoo", "myfunc()"));
- IngestionConfig ingestionConfig = new IngestionConfig(new FilterConfig("filterFunc(foo)"), transformConfigs);
+ Map<String, String> batchConfigMap = new HashMap<>();
+ batchConfigMap.put("batchType", "s3");
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("streamType", "kafka");
+ List<Map<String, String>> streamConfigMaps = new ArrayList<>();
+ streamConfigMaps.add(streamConfigMap);
+ List<Map<String, String>> batchConfigMaps = new ArrayList<>();
+ batchConfigMaps.add(batchConfigMap);
+ IngestionConfig ingestionConfig =
+ new IngestionConfig(new BatchIngestionConfig(batchConfigMaps, "APPEND", "HOURLY"),
+ new StreamIngestionConfig(streamConfigMaps), new FilterConfig("filterFunc(foo)"), transformConfigs);
TableConfig tableConfig = tableConfigBuilder.setIngestionConfig(ingestionConfig).build();
checkIngestionConfig(tableConfig);
@@ -400,6 +413,16 @@ public class TableConfigSerDeTest {
assertEquals(transformConfigs.get(0).getTransformFunction(), "func(moo)");
assertEquals(transformConfigs.get(1).getColumnName(), "zoo");
assertEquals(transformConfigs.get(1).getTransformFunction(), "myfunc()");
+ assertNotNull(ingestionConfig.getBatchIngestionConfig());
+ assertNotNull(ingestionConfig.getBatchIngestionConfig().getBatchConfigMaps());
+ assertEquals(ingestionConfig.getBatchIngestionConfig().getBatchConfigMaps().size(), 1);
+ assertEquals(ingestionConfig.getBatchIngestionConfig().getBatchConfigMaps().get(0).get("batchType"), "s3");
+ assertEquals(ingestionConfig.getBatchIngestionConfig().getSegmentPushType(), "APPEND");
+ assertEquals(ingestionConfig.getBatchIngestionConfig().getSegmentPushFrequency(), "HOURLY");
+ assertNotNull(ingestionConfig.getStreamIngestionConfig());
+ assertNotNull(ingestionConfig.getStreamIngestionConfig().getStreamConfigMaps());
+ assertEquals(ingestionConfig.getStreamIngestionConfig().getStreamConfigMaps().size(), 1);
+ assertEquals(ingestionConfig.getStreamIngestionConfig().getStreamConfigMaps().get(0).get("streamType"), "kafka");
}
private void checkTierConfigList(TableConfig tableConfig) {
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index 26c2dac..fdc005f 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -119,6 +119,7 @@ import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.config.tenant.Tenant;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.apache.pinot.spi.utils.retry.RetryPolicy;
@@ -1142,8 +1143,7 @@ public class PinotHelixResourceManager {
break;
case REALTIME:
- IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
- verifyIndexingConfig(tableNameWithType, indexingConfig);
+ verifyStreamConfig(tableNameWithType, tableConfig);
// Ensure that realtime table is not created if schema is not present
Schema schema =
@@ -1307,9 +1307,10 @@ public class PinotHelixResourceManager {
_pinotLLCRealtimeSegmentManager = pinotLLCRealtimeSegmentManager;
}
- private void verifyIndexingConfig(String tableNameWithType, IndexingConfig indexingConfig) {
+ private void verifyStreamConfig(String tableNameWithType, TableConfig tableConfig) {
// Check if HLC table is allowed.
- StreamConfig streamConfig = new StreamConfig(tableNameWithType, indexingConfig.getStreamConfigs());
+ StreamConfig streamConfig =
+ new StreamConfig(tableNameWithType, IngestionConfigUtils.getStreamConfigMap(tableConfig));
if (streamConfig.hasHighLevelConsumerType() && !_allowHLCTables) {
throw new InvalidTableConfigException(
"Creating HLC realtime table is not allowed for Table: " + tableNameWithType);
@@ -1319,7 +1320,7 @@ public class PinotHelixResourceManager {
private void ensureRealtimeClusterIsSetUp(TableConfig realtimeTableConfig) {
String realtimeTableName = realtimeTableConfig.getTableName();
StreamConfig streamConfig = new StreamConfig(realtimeTableConfig.getTableName(),
- realtimeTableConfig.getIndexingConfig().getStreamConfigs());
+ IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig));
IdealState idealState = getTableIdealState(realtimeTableName);
if (streamConfig.hasHighLevelConsumerType()) {
@@ -1431,8 +1432,7 @@ public class PinotHelixResourceManager {
break;
case REALTIME:
- IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
- verifyIndexingConfig(tableNameWithType, indexingConfig);
+ verifyStreamConfig(tableNameWithType, tableConfig);
ZKMetadataProvider
.setRealtimeTableConfig(_propertyStore, tableNameWithType, TableConfigUtils.toZNRecord(tableConfig));
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
index 05b61ba..a564542 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
@@ -34,6 +34,7 @@ import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentMa
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.stream.PartitionCountFetcher;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -88,7 +89,7 @@ public class PinotTableIdealStateBuilder {
}
setupInstanceConfigForHighLevelConsumer(realtimeTableName, realtimeInstances.size(),
Integer.parseInt(realtimeTableConfig.getValidationConfig().getReplication()),
- realtimeTableConfig.getIndexingConfig().getStreamConfigs(), zkHelixPropertyStore, realtimeInstances);
+ IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig), zkHelixPropertyStore, realtimeInstances);
return idealState;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java
index e27c045..a0e519b 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java
@@ -39,6 +39,7 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -100,8 +101,8 @@ public class RealtimeToOfflineSegmentsTaskGenerator implements PinotTaskGenerato
LOGGER.warn("Skip generating task: {} for non-REALTIME table: {}", taskType, realtimeTableName);
continue;
}
- if (new StreamConfig(realtimeTableName, tableConfig.getIndexingConfig().getStreamConfigs())
- .hasHighLevelConsumerType()) {
+ StreamConfig streamConfig = new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ if (streamConfig.hasHighLevelConsumerType()) {
LOGGER.warn("Skip generating task: {} for HLC REALTIME table: {}", taskType, realtimeTableName);
continue;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index e60a428..00433b3 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -82,6 +82,7 @@ import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.apache.zookeeper.data.Stat;
@@ -210,7 +211,7 @@ public class PinotLLCRealtimeSegmentManager {
_flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName);
PartitionLevelStreamConfig streamConfig =
- new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs());
+ new PartitionLevelStreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
int numPartitions = getNumPartitions(streamConfig);
int numReplicas = getNumReplicas(tableConfig, instancePartitions);
@@ -452,7 +453,7 @@ public class PinotLLCRealtimeSegmentManager {
LLCSegmentName newLLCSegmentName =
getNextLLCSegmentName(new LLCSegmentName(committingSegmentName), newSegmentCreationTimeMs);
createNewSegmentZKMetadata(tableConfig,
- new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs()),
+ new PartitionLevelStreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig)),
newLLCSegmentName, newSegmentCreationTimeMs, committingSegmentDescriptor, committingSegmentZKMetadata,
instancePartitions, numPartitions, numReplicas);
@@ -611,8 +612,8 @@ public class PinotLLCRealtimeSegmentManager {
return commitTimeoutMS;
}
TableConfig tableConfig = getTableConfig(realtimeTableName);
- final Map<String, String> streamConfigs = tableConfig.getIndexingConfig().getStreamConfigs();
- if (streamConfigs != null && streamConfigs.containsKey(StreamConfigProperties.SEGMENT_COMMIT_TIMEOUT_SECONDS)) {
+ final Map<String, String> streamConfigs = IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ if (streamConfigs.containsKey(StreamConfigProperties.SEGMENT_COMMIT_TIMEOUT_SECONDS)) {
final String commitTimeoutSecondsStr = streamConfigs.get(StreamConfigProperties.SEGMENT_COMMIT_TIMEOUT_SECONDS);
try {
return TimeUnit.MILLISECONDS.convert(Integer.parseInt(commitTimeoutSecondsStr), TimeUnit.SECONDS);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
index 6a22b63..b17f20f 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
@@ -56,6 +56,7 @@ import org.apache.pinot.core.query.utils.Pair;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.apache.zookeeper.data.Stat;
@@ -130,12 +131,12 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
continue;
}
- StreamConfig metadata = new StreamConfig(realtimeTableName, tableConfig.getIndexingConfig().getStreamConfigs());
+ StreamConfig metadata = new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(tableConfig));
if (metadata.hasHighLevelConsumerType()) {
idealStateMap.put(realtimeTableName, _pinotHelixResourceManager.getHelixAdmin()
.getResourceIdealState(_pinotHelixResourceManager.getHelixClusterName(), realtimeTableName));
} else {
- LOGGER.debug("Not considering table {} for realtime segment assignment");
+ LOGGER.debug("Not considering table {} for realtime segment assignment", realtimeTableName);
}
}
@@ -335,7 +336,7 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
if (TableNameBuilder.getTableTypeFromTableName(znRecordId) == TableType.REALTIME) {
TableConfig tableConfig = TableConfigUtils.fromZNRecord(tableConfigZnRecord);
StreamConfig metadata = new StreamConfig(tableConfig.getTableName(),
- tableConfig.getIndexingConfig().getStreamConfigs());
+ IngestionConfigUtils.getStreamConfigMap(tableConfig));
if (metadata.hasHighLevelConsumerType()) {
String realtimeTable = tableConfig.getTableName();
String realtimeSegmentsPathForTable = _propertyStorePath + SEGMENTS_PATH + "/" + realtimeTable;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
index b898790..a6be7e8 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
@@ -42,6 +42,7 @@ import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -123,8 +124,8 @@ public class SegmentCompletionManager {
protected StreamPartitionMsgOffsetFactory getStreamPartitionMsgOffsetFactory(LLCSegmentName llcSegmentName) {
final String rawTableName = llcSegmentName.getTableName();
TableConfig tableConfig = _segmentManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(rawTableName));
- PartitionLevelStreamConfig streamConfig =
- new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs());
+ PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
+ IngestionConfigUtils.getStreamConfigMap(tableConfig));
return StreamConsumerFactoryProvider.create(streamConfig).createStreamMsgOffsetFactory();
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
index 081e52c..bc1fe0e 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
@@ -53,6 +53,7 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -144,7 +145,7 @@ public class TableRebalancer {
try {
// Do not allow rebalancing HLC real-time table
if (tableConfig.getTableType() == TableType.REALTIME && new StreamConfig(tableNameWithType,
- tableConfig.getIndexingConfig().getStreamConfigs()).hasHighLevelConsumerType()) {
+ IngestionConfigUtils.getStreamConfigMap(tableConfig)).hasHighLevelConsumerType()) {
LOGGER.warn("Cannot rebalance table: {} with high-level consumer, aborting the rebalance", tableNameWithType);
return new RebalanceResult(RebalanceResult.Status.FAILED, "Cannot rebalance table with high-level consumer",
null, null);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
index 25179ad..5861448 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
@@ -34,7 +34,7 @@ import org.apache.pinot.controller.helix.core.rebalance.RebalanceResult;
import org.apache.pinot.controller.helix.core.rebalance.TableRebalancer;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.utils.TimeUtils;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,7 +67,7 @@ public class SegmentRelocator extends ControllerPeriodicTask<Void> {
// Segment relocation doesn't apply to HLC
boolean isRealtimeTable = TableNameBuilder.isRealtimeTableResource(tableNameWithType);
- if (isRealtimeTable && new StreamConfig(tableNameWithType, tableConfig.getIndexingConfig().getStreamConfigs())
+ if (isRealtimeTable && new StreamConfig(tableNameWithType, IngestionConfigUtils.getStreamConfigMap(tableConfig))
.hasHighLevelConsumerType()) {
return;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
index 2793087..2dcd3c4 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
@@ -45,6 +45,7 @@ import org.apache.pinot.controller.helix.core.retention.strategy.TimeRetentionSt
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.apache.pinot.spi.utils.retry.RetryPolicy;
@@ -100,7 +101,7 @@ public class RetentionManager extends ControllerPeriodicTask<Void> {
// For offline tables, ensure that the segmentPushType is APPEND.
SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
- String segmentPushType = validationConfig.getSegmentPushType();
+ String segmentPushType = IngestionConfigUtils.getBatchSegmentPushType(tableConfig);
if (tableConfig.getTableType() == TableType.OFFLINE && !"APPEND".equalsIgnoreCase(segmentPushType)) {
LOGGER.info("Segment push type is not APPEND for table: {}, skip", tableNameWithType);
return;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableRetentionValidator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableRetentionValidator.java
index c3d3df3..ba44225 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableRetentionValidator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableRetentionValidator.java
@@ -34,6 +34,7 @@ import org.apache.pinot.common.metadata.segment.OfflineSegmentZKMetadata;
import org.apache.pinot.common.utils.config.TableConfigUtils;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.TimeUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
@@ -102,12 +103,13 @@ public class TableRetentionValidator {
}
// Get the retention config
- SegmentsValidationAndRetentionConfig retentionConfig = getTableConfig(tableName).getValidationConfig();
+ TableConfig tableConfig = getTableConfig(tableName);
+ SegmentsValidationAndRetentionConfig retentionConfig = tableConfig.getValidationConfig();
if (retentionConfig == null) {
LOGGER.error("Table: {}, \"segmentsConfig\" field is missing in table config", tableName);
continue;
}
- String segmentPushType = retentionConfig.getSegmentPushType();
+ String segmentPushType = IngestionConfigUtils.getBatchSegmentPushType(tableConfig);
if (segmentPushType == null) {
LOGGER.error("Table: {}, null push type", tableName);
continue;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java
index 7e593b3..e21c72e 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java
@@ -33,6 +33,7 @@ import org.apache.pinot.controller.util.SegmentIntervalUtils;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.TimeUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.joda.time.Duration;
@@ -100,7 +101,8 @@ public class OfflineSegmentIntervalChecker extends ControllerPeriodicTask<Void>
LOGGER
.warn("Table: {} has {} segments with invalid interval", offlineTableName, numSegmentsWithInvalidIntervals);
}
- Duration frequency = SegmentIntervalUtils.convertToDuration(validationConfig.getSegmentPushFrequency());
+ Duration frequency =
+ SegmentIntervalUtils.convertToDuration(IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig));
numMissingSegments = computeNumMissingSegments(segmentIntervals, frequency);
}
// Update the gauge that contains the number of missing segments
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
index 732a60e..d611433 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
@@ -36,6 +36,7 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -96,7 +97,7 @@ public class RealtimeSegmentValidationManager extends ControllerPeriodicTask<Rea
}
PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
- tableConfig.getIndexingConfig().getStreamConfigs());
+ IngestionConfigUtils.getStreamConfigMap(tableConfig));
if (streamConfig.hasLowLevelConsumerType()) {
_llcRealtimeSegmentManager.ensureAllPartitionsConsuming(tableConfig, streamConfig);
}
@@ -108,7 +109,8 @@ public class RealtimeSegmentValidationManager extends ControllerPeriodicTask<Rea
List<RealtimeSegmentZKMetadata> metadataList =
_pinotHelixResourceManager.getRealtimeSegmentMetadata(realtimeTableName);
boolean countHLCSegments = true; // false if this table has ONLY LLC segments (i.e. fully migrated)
- StreamConfig streamConfig = new StreamConfig(realtimeTableName, tableConfig.getIndexingConfig().getStreamConfigs());
+ StreamConfig streamConfig =
+ new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(tableConfig));
if (streamConfig.hasLowLevelConsumerType() && !streamConfig.hasHighLevelConsumerType()) {
countHLCSegments = false;
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
index 4db0266..e57d70f 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
@@ -18,14 +18,7 @@
*/
package org.apache.pinot.controller.helix.core.realtime;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
-
+import com.google.common.base.Preconditions;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
@@ -37,9 +30,7 @@ import java.util.Map;
import java.util.Random;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
-
import javax.annotation.Nullable;
-
import org.apache.commons.io.FileUtils;
import org.apache.helix.model.IdealState;
import org.apache.pinot.common.assignment.InstancePartitions;
@@ -68,6 +59,7 @@ import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.OffsetCriteria;
import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.zookeeper.data.Stat;
@@ -77,7 +69,9 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-import com.google.common.base.Preconditions;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.*;
public class PinotLLCRealtimeSegmentManagerTest {
@@ -843,7 +837,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(_numReplicas)
.setLLC(true).setStreamConfigs(streamConfigs).build();
_streamConfig = new PartitionLevelStreamConfig(_tableConfig.getTableName(),
- _tableConfig.getIndexingConfig().getStreamConfigs());
+ IngestionConfigUtils.getStreamConfigMap(_tableConfig));
}
void makeConsumingInstancePartitions() {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/HLRealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/HLRealtimeSegmentDataManager.java
index 311d2a7..060f19b 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/HLRealtimeSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/HLRealtimeSegmentDataManager.java
@@ -56,6 +56,7 @@ import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamLevelConsumer;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.slf4j.Logger;
@@ -155,7 +156,7 @@ public class HLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
_varLengthDictionaryColumns = new ArrayList<>(indexLoadingConfig.getVarLengthDictionaryColumns());
- _streamConfig = new StreamConfig(_tableNameWithType, tableConfig.getIndexingConfig().getStreamConfigs());
+ _streamConfig = new StreamConfig(_tableNameWithType, IngestionConfigUtils.getStreamConfigMap(tableConfig));
_segmentLogger = LoggerFactory.getLogger(
HLRealtimeSegmentDataManager.class.getName() + "_" + _segmentName + "_" + _streamConfig.getTopicName());
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
index 01e69ae..13deed5 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
@@ -83,6 +83,7 @@ import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
import org.apache.pinot.spi.stream.TransientConsumerException;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.slf4j.Logger;
@@ -1102,7 +1103,8 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
String timeColumnName = tableConfig.getValidationConfig().getTimeColumnName();
// TODO Validate configs
IndexingConfig indexingConfig = _tableConfig.getIndexingConfig();
- _partitionLevelStreamConfig = new PartitionLevelStreamConfig(_tableNameWithType, indexingConfig.getStreamConfigs());
+ _partitionLevelStreamConfig =
+ new PartitionLevelStreamConfig(_tableNameWithType, IngestionConfigUtils.getStreamConfigMap(_tableConfig));
_streamConsumerFactory = StreamConsumerFactoryProvider.create(_partitionLevelStreamConfig);
_streamPartitionMsgOffsetFactory =
StreamConsumerFactoryProvider.create(_partitionLevelStreamConfig).createStreamMsgOffsetFactory();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/util/IngestionUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/util/IngestionUtils.java
index fac986d..fa7ce1a 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/util/IngestionUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/util/IngestionUtils.java
@@ -25,8 +25,8 @@ import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.core.data.function.FunctionEvaluator;
import org.apache.pinot.core.data.function.FunctionEvaluatorFactory;
-import org.apache.pinot.spi.config.table.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
@@ -88,7 +88,8 @@ public class IngestionUtils {
FunctionEvaluator expressionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformConfig.getTransformFunction());
fields.addAll(expressionEvaluator.getArguments());
- fields.add(transformConfig.getColumnName()); // add the column itself too, so that if it is already transformed, we won't transform again
+ fields.add(transformConfig
+ .getColumnName()); // add the column itself too, so that if it is already transformed, we won't transform again
}
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/util/ReplicationUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/util/ReplicationUtils.java
index 6d33bc6..1edf233 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/util/ReplicationUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/util/ReplicationUtils.java
@@ -22,6 +22,7 @@ import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
/**
@@ -37,7 +38,7 @@ public class ReplicationUtils {
TableType tableType = tableConfig.getTableType();
if (tableType.equals(TableType.REALTIME)) {
StreamConfig streamConfig = new StreamConfig(tableConfig.getTableName(),
- tableConfig.getIndexingConfig().getStreamConfigs());
+ IngestionConfigUtils.getStreamConfigMap(tableConfig));
return streamConfig.hasHighLevelConsumerType();
}
return true;
@@ -51,7 +52,7 @@ public class ReplicationUtils {
TableType tableType = tableConfig.getTableType();
if (tableType.equals(TableType.REALTIME)) {
StreamConfig streamConfig = new StreamConfig(tableConfig.getTableName(),
- tableConfig.getIndexingConfig().getStreamConfigs());
+ IngestionConfigUtils.getStreamConfigMap(tableConfig));
return streamConfig.hasLowLevelConsumerType();
}
return false;
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/util/TableConfigUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/util/TableConfigUtils.java
index 882f017..e6cafaf 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/util/TableConfigUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/util/TableConfigUtils.java
@@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
import org.apache.pinot.common.tier.TierFactory;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.config.TagNameUtils;
@@ -37,7 +38,6 @@ import org.apache.pinot.core.data.function.FunctionEvaluatorFactory;
import org.apache.pinot.core.startree.v2.AggregationFunctionColumnPair;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.IndexingConfig;
-import org.apache.pinot.spi.config.table.IngestionConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.StarTreeIndexConfig;
@@ -46,9 +46,12 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TierConfig;
import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.ingestion.batch.BatchConfig;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.TimeUtils;
@@ -79,7 +82,7 @@ public final class TableConfigUtils {
// Sanitize the table config before validation
sanitize(tableConfig);
validateValidationConfig(tableConfig, schema);
- validateIngestionConfig(tableConfig.getIngestionConfig(), schema);
+ validateIngestionConfig(tableConfig, schema);
validateTierConfigList(tableConfig.getTierConfigsList());
validateIndexingConfig(tableConfig.getIndexingConfig(), schema);
validateFieldConfigList(tableConfig.getFieldConfigList(), schema);
@@ -113,7 +116,7 @@ public final class TableConfigUtils {
String.format("Table: %s, \"segmentsConfig\" field is missing in table config", tableName));
}
- String segmentPushType = segmentsConfig.getSegmentPushType();
+ String segmentPushType = IngestionConfigUtils.getBatchSegmentPushType(tableConfig);
// segmentPushType is not needed for Realtime table
if (tableConfig.getTableType() == TableType.OFFLINE && segmentPushType != null && !segmentPushType.isEmpty()) {
if (!segmentPushType.equalsIgnoreCase("REFRESH") && !segmentPushType.equalsIgnoreCase("APPEND")) {
@@ -179,8 +182,39 @@ public final class TableConfigUtils {
* 4. validity of transform function string
* 5. checks for source fields used in destination columns
*/
- private static void validateIngestionConfig(@Nullable IngestionConfig ingestionConfig, @Nullable Schema schema) {
+ public static void validateIngestionConfig(TableConfig tableConfig, @Nullable Schema schema) {
+ IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
+
if (ingestionConfig != null) {
+ String tableNameWithType = tableConfig.getTableName();
+
+ // Batch
+ if (ingestionConfig.getBatchIngestionConfig() != null) {
+ List<Map<String, String>> batchConfigMaps = ingestionConfig.getBatchIngestionConfig().getBatchConfigMaps();
+ try {
+ if (CollectionUtils.isNotEmpty(batchConfigMaps)) {
+ // Validate that BatchConfig can be created
+ batchConfigMaps.forEach(b -> new BatchConfig(tableNameWithType, b));
+ }
+ } catch (Exception e) {
+ throw new IllegalStateException("Could not create BatchConfig using the batchConfig map", e);
+ }
+ }
+
+ // Stream
+ if (ingestionConfig.getStreamIngestionConfig() != null) {
+ IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
+ Preconditions.checkState(indexingConfig == null || MapUtils.isEmpty(indexingConfig.getStreamConfigs()),
+ "Should not use indexingConfig#getStreamConfigs if ingestionConfig#StreamIngestionConfig is provided");
+ List<Map<String, String>> streamConfigMaps = ingestionConfig.getStreamIngestionConfig().getStreamConfigMaps();
+ Preconditions.checkState(streamConfigMaps.size() == 1, "Only 1 stream is supported in REALTIME table");
+ try {
+ // Validate that StreamConfig can be created
+ new StreamConfig(tableNameWithType, streamConfigMaps.get(0));
+ } catch (Exception e) {
+ throw new IllegalStateException("Could not create StreamConfig using the streamConfig map", e);
+ }
+ }
// Filter config
FilterConfig filterConfig = ingestionConfig.getFilterConfig();
@@ -256,11 +290,8 @@ public final class TableConfigUtils {
Preconditions.checkState(CollectionUtils.isNotEmpty(schema.getPrimaryKeyColumns()),
"Upsert table must have primary key columns in the schema");
// consumer type must be low-level
- Preconditions.checkState(
- tableConfig.getIndexingConfig() != null && tableConfig.getIndexingConfig().getStreamConfigs() != null,
- "streamConfig must exist in the table config");
- StreamConfig streamConfig =
- new StreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs());
+ Map<String, String> streamConfigsMap = IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ StreamConfig streamConfig = new StreamConfig(tableConfig.getTableName(), streamConfigsMap);
Preconditions.checkState(streamConfig.hasLowLevelConsumerType() && !streamConfig.hasHighLevelConsumerType(),
"Upsert table must use low-level streaming consumer type");
// replica group is configured for routing
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/ExpressionTransformerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/ExpressionTransformerTest.java
index 70d3948..e14f67a 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/ExpressionTransformerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/ExpressionTransformerTest.java
@@ -23,7 +23,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
@@ -60,7 +60,7 @@ public class ExpressionTransformerTest {
transformConfigs.add(new TransformConfig("map2_values", "Groovy({map2.sort()*.value}, map2)"));
transformConfigs.add(new TransformConfig("hoursSinceEpoch", "Groovy({timestamp/(1000*60*60)}, timestamp)"));
TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTransformFunctions")
- .setIngestionConfig(new IngestionConfig(null, transformConfigs)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, null, transformConfigs)).build();
ExpressionTransformer expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);
DataTypeTransformer dataTypeTransformer = new DataTypeTransformer(pinotSchema);
@@ -148,7 +148,7 @@ public class ExpressionTransformerTest {
transformConfigs.add(new TransformConfig("fullName", "Groovy({firstName+' '+lastName}, firstName, lastName)"));
transformConfigs.add(new TransformConfig("hoursSinceEpoch", "Groovy({timestamp/(1000*60*60)}, timestamp)"));
TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTransformFunctions")
- .setIngestionConfig(new IngestionConfig(null, transformConfigs)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, null, transformConfigs)).build();
ExpressionTransformer expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);
@@ -202,7 +202,7 @@ public class ExpressionTransformerTest {
List<TransformConfig> transformConfigs = new ArrayList<>();
transformConfigs.add(new TransformConfig("fullName", "Groovy({firstName + ' ' + lastName}, firstName, lastName)"));
TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testValueExists")
- .setIngestionConfig(new IngestionConfig(null, transformConfigs)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, null, transformConfigs)).build();
ExpressionTransformer expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);
GenericRow genericRow = new GenericRow();
@@ -218,7 +218,7 @@ public class ExpressionTransformerTest {
.addTime(new TimeGranularitySpec(FieldSpec.DataType.LONG, TimeUnit.MILLISECONDS, "incoming"),
new TimeGranularitySpec(FieldSpec.DataType.INT, TimeUnit.DAYS, "outgoing")).build();
tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testValueExists")
- .setIngestionConfig(new IngestionConfig(null, null)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, null, null)).build();
expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);
genericRow = new GenericRow();
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/RecordTransformerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/RecordTransformerTest.java
index 5ea38de..3495815 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/RecordTransformerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/recordtransformer/RecordTransformerTest.java
@@ -19,7 +19,7 @@
package org.apache.pinot.core.data.recordtransformer;
import java.util.Collections;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
@@ -79,28 +79,30 @@ public class RecordTransformerTest {
// expression false, not filtered
GenericRow genericRow = getRecord();
- tableConfig.setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy({svInt > 123}, svInt)"), null));
+ tableConfig
+ .setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("Groovy({svInt > 123}, svInt)"), null));
RecordTransformer transformer = new FilterTransformer(tableConfig);
transformer.transform(genericRow);
Assert.assertFalse(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
// expression true, filtered
genericRow = getRecord();
- tableConfig.setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy({svInt <= 123}, svInt)"), null));
+ tableConfig
+ .setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("Groovy({svInt <= 123}, svInt)"), null));
transformer = new FilterTransformer(tableConfig);
transformer.transform(genericRow);
Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
// value not found
genericRow = getRecord();
- tableConfig
- .setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy({notPresent == 123}, notPresent)"), null));
+ tableConfig.setIngestionConfig(
+ new IngestionConfig(null, null, new FilterConfig("Groovy({notPresent == 123}, notPresent)"), null));
transformer = new FilterTransformer(tableConfig);
transformer.transform(genericRow);
Assert.assertFalse(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
// invalid function
- tableConfig.setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy(svInt == 123)"), null));
+ tableConfig.setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("Groovy(svInt == 123)"), null));
try {
new FilterTransformer(tableConfig);
Assert.fail("Should have failed constructing FilterTransformer");
@@ -110,8 +112,8 @@ public class RecordTransformerTest {
// multi value column
genericRow = getRecord();
- tableConfig
- .setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy({svFloat.max() < 500}, svFloat)"), null));
+ tableConfig.setIngestionConfig(
+ new IngestionConfig(null, null, new FilterConfig("Groovy({svFloat.max() < 500}, svFloat)"), null));
transformer = new FilterTransformer(tableConfig);
transformer.transform(genericRow);
Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/segment/index/creator/SegmentGenerationWithFilterRecordsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/segment/index/creator/SegmentGenerationWithFilterRecordsTest.java
index abc3c3c..3c5d29e 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/segment/index/creator/SegmentGenerationWithFilterRecordsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/segment/index/creator/SegmentGenerationWithFilterRecordsTest.java
@@ -29,7 +29,7 @@ import org.apache.pinot.core.indexsegment.generator.SegmentGeneratorConfig;
import org.apache.pinot.core.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.core.segment.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.core.segment.store.SegmentDirectory;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
@@ -65,9 +65,11 @@ public class SegmentGenerationWithFilterRecordsTest {
@BeforeClass
public void setup() {
- String filterFunction = "Groovy({((col2 < 1589007600000L) && (col3.max() < 4)) || col1 == \"B\"}, col1, col2, col3)";
- IngestionConfig ingestionConfig = new IngestionConfig(new FilterConfig(filterFunction), null);
- _tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").setIngestionConfig(ingestionConfig).build();
+ String filterFunction =
+ "Groovy({((col2 < 1589007600000L) && (col3.max() < 4)) || col1 == \"B\"}, col1, col2, col3)";
+ IngestionConfig ingestionConfig = new IngestionConfig(null, null, new FilterConfig(filterFunction), null);
+ _tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").setIngestionConfig(ingestionConfig).build();
_schema = new Schema.SchemaBuilder().addSingleValueDimension(STRING_COLUMN, FieldSpec.DataType.STRING)
.addMetric(LONG_COLUMN, FieldSpec.DataType.LONG).addMultiValueDimension(MV_INT_COLUMN, FieldSpec.DataType.INT)
.build();
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/util/IngestionUtilsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/util/IngestionUtilsTest.java
index 122d190..90d7396 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/util/IngestionUtilsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/util/IngestionUtilsTest.java
@@ -25,8 +25,8 @@ import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import org.apache.pinot.spi.config.table.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DimensionFieldSpec;
@@ -134,7 +134,7 @@ public class IngestionUtilsTest {
Schema schema = new Schema();
// filter config
- IngestionConfig ingestionConfig = new IngestionConfig(new FilterConfig("Groovy({x > 100}, x)"), null);
+ IngestionConfig ingestionConfig = new IngestionConfig(null, null, new FilterConfig("Groovy({x > 100}, x)"), null);
Set<String> fields = IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema);
Assert.assertEquals(fields.size(), 1);
Assert.assertTrue(fields.containsAll(Sets.newHashSet("x")));
@@ -144,18 +144,18 @@ public class IngestionUtilsTest {
Assert.assertEquals(fields.size(), 2);
Assert.assertTrue(fields.containsAll(Sets.newHashSet("x", "y")));
-
// transform configs
schema = new Schema.SchemaBuilder().addSingleValueDimension("d1", FieldSpec.DataType.STRING).build();
- List<TransformConfig> transformConfigs = Lists.newArrayList(new TransformConfig("d1", "Groovy({function}, argument1, argument2)"));
- ingestionConfig = new IngestionConfig(null, transformConfigs);
+ List<TransformConfig> transformConfigs =
+ Lists.newArrayList(new TransformConfig("d1", "Groovy({function}, argument1, argument2)"));
+ ingestionConfig = new IngestionConfig(null, null, null, transformConfigs);
List<String> extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
Assert.assertEquals(extract.size(), 3);
Assert.assertTrue(extract.containsAll(Arrays.asList("d1", "argument1", "argument2")));
// groovy function, no arguments
transformConfigs = Lists.newArrayList(new TransformConfig("d1", "Groovy({function})"));
- ingestionConfig = new IngestionConfig(null, transformConfigs);
+ ingestionConfig = new IngestionConfig(null, null, null, transformConfigs);
extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
Assert.assertEquals(extract.size(), 1);
Assert.assertTrue(extract.contains("d1"));
@@ -163,36 +163,39 @@ public class IngestionUtilsTest {
// inbuilt functions
schema = new Schema.SchemaBuilder().addSingleValueDimension("hoursSinceEpoch", FieldSpec.DataType.LONG).build();
transformConfigs = Lists.newArrayList(new TransformConfig("hoursSinceEpoch", "toEpochHours(timestampColumn)"));
- ingestionConfig = new IngestionConfig(null, transformConfigs);
+ ingestionConfig = new IngestionConfig(null, null, null, transformConfigs);
extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
Assert.assertEquals(extract.size(), 2);
Assert.assertTrue(extract.containsAll(Arrays.asList("timestampColumn", "hoursSinceEpoch")));
// inbuilt functions with literal
- schema = new Schema.SchemaBuilder().addSingleValueDimension("tenMinutesSinceEpoch", FieldSpec.DataType.LONG).build();
- transformConfigs = Lists.newArrayList(new TransformConfig("tenMinutesSinceEpoch", "toEpochMinutesBucket(timestampColumn, 10)"));
- ingestionConfig = new IngestionConfig(null, transformConfigs);
+ schema =
+ new Schema.SchemaBuilder().addSingleValueDimension("tenMinutesSinceEpoch", FieldSpec.DataType.LONG).build();
+ transformConfigs =
+ Lists.newArrayList(new TransformConfig("tenMinutesSinceEpoch", "toEpochMinutesBucket(timestampColumn, 10)"));
+ ingestionConfig = new IngestionConfig(null, null, null, transformConfigs);
extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
Assert.assertEquals(extract.size(), 2);
Assert.assertTrue(extract.containsAll(Lists.newArrayList("tenMinutesSinceEpoch", "timestampColumn")));
// inbuilt functions on DateTimeFieldSpec
- schema = new Schema.SchemaBuilder().addDateTime("dateColumn", FieldSpec.DataType.STRING, "1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd", "1:DAYS").build();
- transformConfigs = Lists.newArrayList(new TransformConfig("dateColumn", "toDateTime(timestampColumn, 'yyyy-MM-dd')"));
- ingestionConfig = new IngestionConfig(null, transformConfigs);
+ schema = new Schema.SchemaBuilder()
+ .addDateTime("dateColumn", FieldSpec.DataType.STRING, "1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd", "1:DAYS").build();
+ transformConfigs =
+ Lists.newArrayList(new TransformConfig("dateColumn", "toDateTime(timestampColumn, 'yyyy-MM-dd')"));
+ ingestionConfig = new IngestionConfig(null, null, null, transformConfigs);
extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
Assert.assertEquals(extract.size(), 2);
Assert.assertTrue(extract.containsAll(Lists.newArrayList("dateColumn", "timestampColumn")));
// filter + transform configs + schema fields + schema transform
- schema = new Schema.SchemaBuilder()
- .addSingleValueDimension("d1", FieldSpec.DataType.STRING)
- .addSingleValueDimension("d2", FieldSpec.DataType.STRING)
- .addMetric("m1", FieldSpec.DataType.INT)
+ schema = new Schema.SchemaBuilder().addSingleValueDimension("d1", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("d2", FieldSpec.DataType.STRING).addMetric("m1", FieldSpec.DataType.INT)
.addDateTime("dateColumn", FieldSpec.DataType.STRING, "1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd", "1:DAYS").build();
schema.getFieldSpecFor("d2").setTransformFunction("reverse(xy)");
- transformConfigs = Lists.newArrayList(new TransformConfig("dateColumn", "toDateTime(timestampColumn, 'yyyy-MM-dd')"));
- ingestionConfig = new IngestionConfig(new FilterConfig("Groovy({d1 == \"10\"}, d1)"), transformConfigs);
+ transformConfigs =
+ Lists.newArrayList(new TransformConfig("dateColumn", "toDateTime(timestampColumn, 'yyyy-MM-dd')"));
+ ingestionConfig = new IngestionConfig(null, null, new FilterConfig("Groovy({d1 == \"10\"}, d1)"), transformConfigs);
extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
Assert.assertEquals(extract.size(), 6);
Assert.assertTrue(extract.containsAll(Lists.newArrayList("d1", "d2", "m1", "dateColumn", "xy", "timestampColumn")));
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
index ef3aa90..19d7fab 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java
@@ -23,7 +23,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
@@ -90,7 +90,7 @@ public class SchemaUtilsTest {
// schema doesn't have destination columns from transformConfigs
schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).build();
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("colA", "round(colB, 1000)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("colA", "round(colB, 1000)")))).build();
try {
SchemaUtils.validate(schema, Lists.newArrayList(tableConfig));
Assert.fail("Should fail schema validation, as colA is not present in schema");
@@ -138,7 +138,7 @@ public class SchemaUtilsTest {
.addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH", "1:HOURS").build();
tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
.setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("colA", "round(colB, 1000)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("colA", "round(colB, 1000)")))).build();
try {
SchemaUtils.validate(schema, Lists.newArrayList(tableConfig));
Assert.fail("Should fail schema validation, as colA is not present in schema");
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/util/TableConfigUtilsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/util/TableConfigUtilsTest.java
index 9ddcb9c..ae313c0 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/util/TableConfigUtilsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/util/TableConfigUtilsTest.java
@@ -24,13 +24,14 @@ import java.util.Collections;
import java.util.List;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
import org.apache.pinot.common.function.AggregationFunctionType;
import org.apache.pinot.common.tier.TierFactory;
+import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import org.apache.pinot.core.startree.v2.AggregationFunctionColumnPair;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.FieldConfig;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.RoutingConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.StarTreeIndexConfig;
@@ -39,13 +40,15 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TierConfig;
import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties;
+import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.Assert;
import org.testng.annotations.Test;
-import org.testng.collections.Sets;
/**
@@ -180,27 +183,27 @@ public class TableConfigUtilsTest {
// null filter config, transform config
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(null, null)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, null, null)).build();
TableConfigUtils.validate(tableConfig, schema);
// null filter function
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(new FilterConfig(null), null)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, new FilterConfig(null), null)).build();
TableConfigUtils.validate(tableConfig, schema);
// valid filterFunction
- tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(new FilterConfig("startsWith(columnX, \"myPrefix\")"), null)).build();
+ tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
+ new IngestionConfig(null, null, new FilterConfig("startsWith(columnX, \"myPrefix\")"), null)).build();
TableConfigUtils.validate(tableConfig, schema);
// valid filterFunction
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy({x == 10}, x)"), null)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("Groovy({x == 10}, x)"), null)).build();
TableConfigUtils.validate(tableConfig, schema);
// invalid filter function
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy(badExpr)"), null)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("Groovy(badExpr)"), null)).build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail on invalid filter function string");
@@ -209,7 +212,7 @@ public class TableConfigUtilsTest {
}
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(new FilterConfig("fakeFunction(xx)"), null)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("fakeFunction(xx)"), null)).build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail for invalid filter function");
@@ -219,12 +222,13 @@ public class TableConfigUtilsTest {
// empty transform configs
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(null, Collections.emptyList())).build();
+ .setIngestionConfig(new IngestionConfig(null, null, null, Collections.emptyList())).build();
TableConfigUtils.validate(tableConfig, schema);
// transformed column not in schema
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)"))))
+ .build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail for transformedColumn not present in schema");
@@ -237,7 +241,8 @@ public class TableConfigUtilsTest {
.build();
// valid transform configs
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)"))))
+ .build();
TableConfigUtils.validate(tableConfig, schema);
schema =
@@ -245,13 +250,14 @@ public class TableConfigUtilsTest {
.addMetric("transformedCol", FieldSpec.DataType.LONG).build();
// valid transform configs
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)"),
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)"),
new TransformConfig("transformedCol", "Groovy({x+y}, x, y)")))).build();
TableConfigUtils.validate(tableConfig, schema);
// null transform column name
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig(null, "reverse(anotherCol)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig(null, "reverse(anotherCol)"))))
+ .build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail for null column name in transform config");
@@ -260,8 +266,8 @@ public class TableConfigUtilsTest {
}
// null transform function string
- tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
- .setIngestionConfig(new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", null)))).build();
+ tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("myCol", null)))).build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail for null transform function in transform config");
@@ -271,7 +277,8 @@ public class TableConfigUtilsTest {
// invalid function
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "fakeFunction(col)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("myCol", "fakeFunction(col)"))))
+ .build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail for invalid transform function in transform config");
@@ -281,7 +288,8 @@ public class TableConfigUtilsTest {
// invalid function
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "Groovy(badExpr)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("myCol", "Groovy(badExpr)"))))
+ .build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail for invalid transform function in transform config");
@@ -291,7 +299,8 @@ public class TableConfigUtilsTest {
// input field name used as destination field
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(myCol)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig("myCol", "reverse(myCol)"))))
+ .build();
try {
TableConfigUtils.validate(tableConfig, schema);
Assert.fail("Should fail due to use of myCol as arguments and columnName");
@@ -301,7 +310,7 @@ public class TableConfigUtilsTest {
// input field name used as destination field
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null,
+ new IngestionConfig(null, null, null,
Lists.newArrayList(new TransformConfig("myCol", "Groovy({x + y + myCol}, x, myCol, y)")))).build();
try {
TableConfigUtils.validate(tableConfig, schema);
@@ -312,7 +321,7 @@ public class TableConfigUtilsTest {
// duplicate transform config
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null,
+ new IngestionConfig(null, null, null,
Lists.newArrayList(new TransformConfig("myCol", "reverse(x)"), new TransformConfig("myCol", "lower(y)"))))
.build();
try {
@@ -324,7 +333,7 @@ public class TableConfigUtilsTest {
// chained transform functions
tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(
- new IngestionConfig(null,
+ new IngestionConfig(null, null, null,
Lists.newArrayList(new TransformConfig("a", "reverse(x)"), new TransformConfig("b", "lower(a)")))).build();
try {
TableConfigUtils.validate(tableConfig, schema);
@@ -335,6 +344,68 @@ public class TableConfigUtilsTest {
}
@Test
+ public void ingestionStreamConfigsTest() {
+ Map<String, String> fakeMap = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
+ IngestionConfig ingestionConfig =
+ new IngestionConfig(null, new StreamIngestionConfig(Lists.newArrayList(fakeMap, fakeMap)), null, null);
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.REALTIME).setTableName("myTable_REALTIME").setTimeColumnName("timeColumn")
+ .setIngestionConfig(ingestionConfig).build();
+
+ // only 1 stream config allowed
+ try {
+ TableConfigUtils.validateIngestionConfig(tableConfig, null);
+ Assert.fail("Should fail for more than 1 stream config");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ // stream config should be valid
+ ingestionConfig = new IngestionConfig(null, new StreamIngestionConfig(Lists.newArrayList(fakeMap)), null, null);
+ tableConfig.setIngestionConfig(ingestionConfig);
+ TableConfigUtils.validateIngestionConfig(tableConfig, null);
+
+ fakeMap.remove(StreamConfigProperties.STREAM_TYPE);
+ try {
+ TableConfigUtils.validateIngestionConfig(tableConfig, null);
+ Assert.fail("Should fail for invalid stream configs map");
+ } catch (Exception e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void ingestionBatchConfigsTest() {
+ Map<String, String> batchConfigMap = new HashMap<>();
+ batchConfigMap.put(BatchConfigProperties.BATCH_TYPE, "s3");
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty("s3", BatchConfigProperties.INPUT_DIR_URI), "s3://foo");
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty("s3", BatchConfigProperties.OUTPUT_DIR_URI), "s3://bar");
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty("s3", BatchConfigProperties.FS_CLASS), "org.foo.S3FS");
+ batchConfigMap.put(BatchConfigProperties.constructBatchProperty("s3", BatchConfigProperties.INPUT_FORMAT), "avro");
+ batchConfigMap.put(BatchConfigProperties.constructBatchProperty("s3", BatchConfigProperties.RECORD_READER_CLASS),
+ "org.foo.Reader");
+
+ IngestionConfig ingestionConfig =
+ new IngestionConfig(new BatchIngestionConfig(Lists.newArrayList(batchConfigMap, batchConfigMap), null, null),
+ null, null, null);
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable_OFFLINE").setIngestionConfig(ingestionConfig)
+ .build();
+ TableConfigUtils.validateIngestionConfig(tableConfig, null);
+
+ batchConfigMap.remove(BatchConfigProperties.BATCH_TYPE);
+ try {
+ TableConfigUtils.validateIngestionConfig(tableConfig, null);
+ Assert.fail("Should fail for invalid batch config map");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+ }
+
+ @Test
public void validateTierConfigs() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
.addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
@@ -723,7 +794,8 @@ public class TableConfigUtilsTest {
try {
TableConfigUtils.validateUpsertConfig(tableConfig, schema);
} catch (Exception e) {
- Assert.assertEquals(e.getMessage(), "streamConfig must exist in the table config");
+ Assert
+ .assertEquals(e.getMessage(), "Could not find streamConfigs for REALTIME table: " + TABLE_NAME + "_REALTIME");
}
Map<String, String> streamConfigs = new HashMap<>();
streamConfigs.put("stream.kafka.consumer.type", "highLevel");
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index 050d90a..d82f8eb 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -39,10 +39,10 @@ import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.common.utils.config.TagNameUtils;
import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties;
import org.apache.pinot.spi.config.table.FieldConfig;
-import org.apache.pinot.spi.config.table.IngestionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
@@ -290,57 +290,64 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
return getOfflineTableConfig(getTableName());
}
- /**
- * Creates a new REALTIME table config.
- */
- protected TableConfig createRealtimeTableConfig(File sampleAvroFile) {
- Map<String, String> streamConfigs = new HashMap<>();
+ protected Map<String, String> getStreamConfigs() {
+ return getStreamConfigMap();
+ }
+
+ protected Map<String, String> getStreamConfigMap() {
+ Map<String, String> streamConfigMap = new HashMap<>();
String streamType = "kafka";
- streamConfigs.put(StreamConfigProperties.STREAM_TYPE, streamType);
+ streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
boolean useLlc = useLlc();
if (useLlc) {
// LLC
- streamConfigs
+ streamConfigMap
.put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
StreamConfig.ConsumerType.LOWLEVEL.toString());
- streamConfigs.put(KafkaStreamConfigProperties
+ streamConfigMap.put(KafkaStreamConfigProperties
.constructStreamProperty(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BROKER_LIST),
KafkaStarterUtils.DEFAULT_KAFKA_BROKER);
} else {
// HLC
- streamConfigs
+ streamConfigMap
.put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
StreamConfig.ConsumerType.HIGHLEVEL.toString());
- streamConfigs.put(KafkaStreamConfigProperties
+ streamConfigMap.put(KafkaStreamConfigProperties
.constructStreamProperty(KafkaStreamConfigProperties.HighLevelConsumer.KAFKA_HLC_ZK_CONNECTION_STRING),
KafkaStarterUtils.DEFAULT_ZK_STR);
- streamConfigs.put(KafkaStreamConfigProperties
+ streamConfigMap.put(KafkaStreamConfigProperties
.constructStreamProperty(KafkaStreamConfigProperties.HighLevelConsumer.KAFKA_HLC_BOOTSTRAP_SERVER),
KafkaStarterUtils.DEFAULT_KAFKA_BROKER);
}
- streamConfigs.put(StreamConfigProperties
+ streamConfigMap.put(StreamConfigProperties
.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
getStreamConsumerFactoryClassName());
- streamConfigs
+ streamConfigMap
.put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
getKafkaTopic());
- AvroFileSchemaKafkaAvroMessageDecoder.avroFile = sampleAvroFile;
- streamConfigs
+ streamConfigMap
.put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
AvroFileSchemaKafkaAvroMessageDecoder.class.getName());
- streamConfigs
+ streamConfigMap
.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_ROWS, Integer.toString(getRealtimeSegmentFlushSize()));
- streamConfigs.put(StreamConfigProperties
+ streamConfigMap.put(StreamConfigProperties
.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_OFFSET_CRITERIA), "smallest");
+ return streamConfigMap;
+ }
+ /**
+ * Creates a new REALTIME table config.
+ */
+ protected TableConfig createRealtimeTableConfig(File sampleAvroFile) {
+ AvroFileSchemaKafkaAvroMessageDecoder.avroFile = sampleAvroFile;
return new TableConfigBuilder(TableType.REALTIME).setTableName(getTableName()).setSchemaName(getSchemaName())
.setTimeColumnName(getTimeColumnName()).setSortedColumn(getSortedColumn())
.setInvertedIndexColumns(getInvertedIndexColumns()).setNoDictionaryColumns(getNoDictionaryColumns())
.setRangeIndexColumns(getRangeIndexColumns()).setBloomFilterColumns(getBloomFilterColumns())
.setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas()).setSegmentVersion(getSegmentVersion())
.setLoadMode(getLoadMode()).setTaskConfig(getTaskConfig()).setBrokerTenant(getBrokerTenant())
- .setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig()).setLLC(useLlc)
- .setStreamConfigs(streamConfigs).setNullHandlingEnabled(getNullHandlingEnabled()).build();
+ .setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig()).setLLC(useLlc())
+ .setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled()).build();
}
/**
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/IngestionConfigHybridIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/IngestionConfigHybridIntegrationTest.java
index 5fe3807..3537f8f 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/IngestionConfigHybridIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/IngestionConfigHybridIntegrationTest.java
@@ -21,13 +21,12 @@ package org.apache.pinot.integration.tests;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.File;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
-import javax.annotation.Nullable;
-import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import java.util.Map;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
@@ -66,7 +65,15 @@ public class IngestionConfigHybridIntegrationTest extends BaseClusterIntegration
transformConfigs.add(new TransformConfig("AmPm", "Groovy({DepTime < 1200 ? \"AM\": \"PM\"}, DepTime)"));
transformConfigs.add(new TransformConfig("millisSinceEpoch", "fromEpochDays(DaysSinceEpoch)"));
transformConfigs.add(new TransformConfig("lowerCaseDestCityName", "lower(DestCityName)"));
- return new IngestionConfig(filterConfig, transformConfigs);
+
+ List<Map<String, String>> streamConfigMaps = new ArrayList<>();
+ streamConfigMaps.add(getStreamConfigMap());
+ return new IngestionConfig(null, new StreamIngestionConfig(streamConfigMaps), filterConfig, transformConfigs);
+ }
+
+ @Override
+ protected Map<String, String> getStreamConfigs() {
+ return null;
}
@Override
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LuceneRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LuceneRealtimeClusterIntegrationTest.java
index 572737f..f9ab050 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LuceneRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LuceneRealtimeClusterIntegrationTest.java
@@ -35,7 +35,6 @@ import org.apache.commons.io.FileUtils;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.data.TimeGranularitySpec;
import org.apache.pinot.util.TestUtils;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MapTypeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MapTypeClusterIntegrationTest.java
index 79f338e..f6b4a66 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MapTypeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MapTypeClusterIntegrationTest.java
@@ -21,7 +21,6 @@ package org.apache.pinot.integration.tests;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.collect.Lists;
import java.io.File;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -33,12 +32,10 @@ import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.core.util.SchemaUtils;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
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.Schema;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
@@ -87,7 +84,7 @@ public class MapTypeClusterIntegrationTest extends BaseClusterIntegrationTest {
new TransformConfig(STRING_KEY_MAP_STR_FIELD_NAME, "toJsonMapStr(" + STRING_KEY_MAP_FIELD_NAME + ")"),
new TransformConfig(INT_KEY_MAP_STR_FIELD_NAME, "toJsonMapStr(" + INT_KEY_MAP_FIELD_NAME + ")"));
TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(rawTableName)
- .setIngestionConfig(new IngestionConfig(null, transformConfigs)).build();
+ .setIngestionConfig(new IngestionConfig(null, null, null, transformConfigs)).build();
addTableConfig(tableConfig);
// Create and upload segments
diff --git a/pinot-minion/src/test/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutorTest.java b/pinot-minion/src/test/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutorTest.java
index 341f543..62c7f3a 100644
--- a/pinot-minion/src/test/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutorTest.java
+++ b/pinot-minion/src/test/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutorTest.java
@@ -39,7 +39,7 @@ import org.apache.pinot.core.segment.index.metadata.ColumnMetadata;
import org.apache.pinot.core.segment.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.minion.MinionContext;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -101,11 +101,11 @@ public class RealtimeToOfflineSegmentsTaskExecutorTest {
TableConfig tableConfigEpochHours =
new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME_EPOCH_HOURS).setTimeColumnName(T_TRX)
.setSortedColumn(D1).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig(T_TRX, "toEpochHours(t)")))).build();
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig(T_TRX, "toEpochHours(t)")))).build();
TableConfig tableConfigSDF =
new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME_SDF).setTimeColumnName(T_TRX)
.setSortedColumn(D1).setIngestionConfig(
- new IngestionConfig(null, Lists.newArrayList(new TransformConfig(T_TRX, "toDateTime(t, 'yyyyMMddHH')"))))
+ new IngestionConfig(null, null, null, Lists.newArrayList(new TransformConfig(T_TRX, "toDateTime(t, 'yyyyMMddHH')"))))
.build();
Schema schema =
new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension(D1, FieldSpec.DataType.STRING)
diff --git a/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationTaskRunner.java b/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationTaskRunner.java
index c84835a..4d15022 100644
--- a/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationTaskRunner.java
+++ b/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationTaskRunner.java
@@ -20,12 +20,10 @@ package org.apache.pinot.plugin.ingestion.batch.common;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Preconditions;
import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
-import java.util.concurrent.TimeUnit;
import org.apache.pinot.core.indexsegment.generator.SegmentGeneratorConfig;
import org.apache.pinot.core.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.core.segment.name.NormalizedDateSegmentNameGenerator;
@@ -35,13 +33,12 @@ import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DateTimeFormatSpec;
-import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.data.TimeFieldSpec;
import org.apache.pinot.spi.data.readers.RecordReaderConfig;
import org.apache.pinot.spi.ingestion.batch.spec.SegmentGenerationTaskSpec;
import org.apache.pinot.spi.ingestion.batch.spec.SegmentNameGeneratorSpec;
import org.apache.pinot.spi.plugin.PluginManager;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.JsonUtils;
@@ -140,7 +137,8 @@ public class SegmentGenerationTaskRunner implements Serializable {
}
return new NormalizedDateSegmentNameGenerator(tableName, segmentNameGeneratorConfigs.get(SEGMENT_NAME_PREFIX),
Boolean.parseBoolean(segmentNameGeneratorConfigs.get(EXCLUDE_SEQUENCE_ID)),
- validationConfig.getSegmentPushType(), validationConfig.getSegmentPushFrequency(), dateTimeFormatSpec);
+ IngestionConfigUtils.getBatchSegmentPushType(tableConfig),
+ IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig), dateTimeFormatSpec);
default:
throw new UnsupportedOperationException("Unsupported segment name generator type: " + segmentNameGeneratorType);
}
diff --git a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentCreationJob.java b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentCreationJob.java
index b42c92e..2b77efc 100644
--- a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentCreationJob.java
+++ b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentCreationJob.java
@@ -44,6 +44,7 @@ import org.apache.pinot.ingestion.jobs.SegmentCreationJob;
import org.apache.pinot.ingestion.utils.JobPreparationHelper;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
public class HadoopSegmentCreationJob extends SegmentCreationJob {
@@ -142,7 +143,7 @@ public class HadoopSegmentCreationJob extends SegmentCreationJob {
SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
// For APPEND use case, timeColumnName and timeType must be set
- if (APPEND.equalsIgnoreCase(validationConfig.getSegmentPushType())) {
+ if (APPEND.equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentPushType(tableConfig))) {
Preconditions.checkState(validationConfig.getTimeColumnName() != null && validationConfig.getTimeType() != null,
"For APPEND use case, time column and type must be set");
}
diff --git a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java
index 45dca67..9e5e3e1 100644
--- a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java
+++ b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java
@@ -64,11 +64,8 @@ import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableCustomConfig;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
-import org.apache.pinot.spi.data.DateTimeFormatPatternSpec;
import org.apache.pinot.spi.data.DateTimeFormatSpec;
-import org.apache.pinot.spi.data.FieldSpec;
-import org.apache.pinot.spi.data.TimeFieldSpec;
-import org.apache.pinot.spi.data.TimeGranularitySpec;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -371,7 +368,7 @@ public class HadoopSegmentPreprocessingJob extends SegmentPreprocessingJob {
// If the use case is an append use case, check that one time unit is contained in one file. If there is more than one,
// the job should be disabled, as we should not resize for these use cases. Therefore, setting the time column name
// and value
- if (validationConfig.getSegmentPushType().equalsIgnoreCase("APPEND")) {
+ if (IngestionConfigUtils.getBatchSegmentPushType(_tableConfig).equalsIgnoreCase("APPEND")) {
job.getConfiguration().set(InternalConfigConstants.IS_APPEND, "true");
String timeColumnName = validationConfig.getTimeColumnName();
job.getConfiguration().set(InternalConfigConstants.TIME_COLUMN_CONFIG, timeColumnName);
@@ -387,8 +384,8 @@ public class HadoopSegmentPreprocessingJob extends SegmentPreprocessingJob {
.set(InternalConfigConstants.SEGMENT_TIME_SDF_PATTERN, formatSpec.getSDFPattern());
}
}
- job.getConfiguration()
- .set(InternalConfigConstants.SEGMENT_PUSH_FREQUENCY, validationConfig.getSegmentPushFrequency());
+ job.getConfiguration().set(InternalConfigConstants.SEGMENT_PUSH_FREQUENCY,
+ IngestionConfigUtils.getBatchSegmentPushFrequency(_tableConfig));
try (DataFileStream<GenericRecord> dataStreamReader = getAvroReader(path)) {
job.getConfiguration()
.set(InternalConfigConstants.TIME_COLUMN_VALUE, dataStreamReader.next().get(timeColumnName).toString());
diff --git a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java
index fcc5653..9efb7a3 100644
--- a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java
+++ b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java
@@ -52,11 +52,12 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableCustomConfig;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DateTimeFormatSpec;
-import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.IngestionSchemaValidator;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.FileFormat;
import org.apache.pinot.spi.data.readers.RecordReaderConfig;
import org.apache.pinot.spi.utils.DataSizeUtils;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -162,7 +163,8 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
_segmentNameGenerator =
new NormalizedDateSegmentNameGenerator(_rawTableName, _jobConf.get(JobConfigConstants.SEGMENT_NAME_PREFIX),
_jobConf.getBoolean(JobConfigConstants.EXCLUDE_SEQUENCE_ID, false),
- validationConfig.getSegmentPushType(), validationConfig.getSegmentPushFrequency(), dateTimeFormatSpec);
+ IngestionConfigUtils.getBatchSegmentPushType(_tableConfig),
+ IngestionConfigUtils.getBatchSegmentPushFrequency(_tableConfig), dateTimeFormatSpec);
break;
default:
throw new UnsupportedOperationException("Unsupported segment name generator type: " + segmentNameGeneratorType);
diff --git a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationFunction.java b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationFunction.java
index 06c3e19..60d3ef9 100644
--- a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationFunction.java
+++ b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationFunction.java
@@ -50,6 +50,7 @@ import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.FileFormat;
import org.apache.pinot.spi.data.readers.RecordReaderConfig;
import org.apache.pinot.spi.utils.DataSizeUtils;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -130,7 +131,8 @@ public class SparkSegmentCreationFunction implements Serializable {
_segmentNameGenerator =
new NormalizedDateSegmentNameGenerator(_rawTableName, _jobConf.get(JobConfigConstants.SEGMENT_NAME_PREFIX),
_jobConf.getBoolean(JobConfigConstants.EXCLUDE_SEQUENCE_ID, false),
- validationConfig.getSegmentPushType(), validationConfig.getSegmentPushFrequency(), dateTimeFormatSpec);
+ IngestionConfigUtils.getBatchSegmentPushType(_tableConfig),
+ IngestionConfigUtils.getBatchSegmentPushFrequency(_tableConfig), dateTimeFormatSpec);
break;
default:
throw new UnsupportedOperationException("Unsupported segment name generator type: " + segmentNameGeneratorType);
diff --git a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationJob.java b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationJob.java
index fd0b004..d737020 100644
--- a/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationJob.java
+++ b/pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-spark/src/main/java/org/apache/pinot/spark/jobs/SparkSegmentCreationJob.java
@@ -37,6 +37,7 @@ import org.apache.pinot.ingestion.utils.JobPreparationHelper;
import org.apache.pinot.spark.utils.PinotSparkJobPreparationHelper;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
@@ -155,7 +156,7 @@ public class SparkSegmentCreationJob extends SegmentCreationJob {
SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
// For APPEND use case, timeColumnName and timeType must be set
- if (APPEND.equalsIgnoreCase(validationConfig.getSegmentPushType())) {
+ if (APPEND.equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentPushType(tableConfig))) {
Preconditions.checkState(validationConfig.getTimeColumnName() != null && validationConfig.getTimeType() != null,
"For APPEND use case, time column and type must be set");
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java
index 3dd137b..184d00a 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.pinot.spi.config.BaseJsonConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
public class IndexingConfig extends BaseJsonConfig {
@@ -33,6 +34,7 @@ public class IndexingConfig extends BaseJsonConfig {
private List<String> _bloomFilterColumns;
private Map<String, BloomFilterConfig> _bloomFilterConfigs;
private String _loadMode;
+ @Deprecated // Moved to {@link IngestionConfig#getStreamIngestionConfig}
private Map<String, String> _streamConfigs;
private String _segmentFormatVersion;
private String _columnMinMaxValueGeneratorMode;
@@ -123,6 +125,9 @@ public class IndexingConfig extends BaseJsonConfig {
_loadMode = loadMode;
}
+ /**
+ * @deprecated Use <code>List<Map<String, String>> streamConfigs</code> from {@link IngestionConfig#getStreamIngestionConfig()}
+ */
@Nullable
public Map<String, String> getStreamConfigs() {
return _streamConfigs;
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
index 8bc1567..c91fbd4 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import java.util.concurrent.TimeUnit;
import org.apache.pinot.spi.config.BaseJsonConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.utils.TimeUtils;
@@ -29,7 +30,9 @@ import org.apache.pinot.spi.utils.TimeUtils;
public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
private String _retentionTimeUnit;
private String _retentionTimeValue;
+ @Deprecated
private String _segmentPushFrequency; // DO NOT REMOVE, this is used in internal segment generation management
+ @Deprecated
private String _segmentPushType;
private String _replication;
// For high-level consumers, the number of replicas should be same as num server instances
@@ -91,6 +94,9 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
_retentionTimeValue = retentionTimeValue;
}
+ /**
+ * @deprecated Use {@code segmentPushFrequency} from {@link IngestionConfig#getBatchIngestionConfig()}
+ */
public String getSegmentPushFrequency() {
return _segmentPushFrequency;
}
@@ -99,6 +105,9 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
_segmentPushFrequency = segmentPushFrequency;
}
+ /**
+ * @deprecated Use {@code segmentPushType} from {@link IngestionConfig#getBatchIngestionConfig()}
+ */
public String getSegmentPushType() {
return _segmentPushType;
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java
index 2c219e9..f3a6994 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java
@@ -29,6 +29,7 @@ import javax.annotation.Nullable;
import org.apache.pinot.spi.config.BaseJsonConfig;
import org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/BatchIngestionConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/BatchIngestionConfig.java
new file mode 100644
index 0000000..e9b8718
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/BatchIngestionConfig.java
@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.config.table.ingestion;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.config.BaseJsonConfig;
+
+
+/**
+ * Contains all configs related to the batch sources for ingestion.
+ */
+public class BatchIngestionConfig extends BaseJsonConfig {
+
+ @JsonPropertyDescription("Configs for all the batch sources to ingest from")
+ private final List<Map<String, String>> _batchConfigMaps;
+
+ @JsonPropertyDescription("Push type APPEND or REFRESH")
+ private final String _segmentPushType;
+
+ @JsonPropertyDescription("Push frequency HOURLY or DAILY")
+ private final String _segmentPushFrequency;
+
+ @JsonCreator
+ public BatchIngestionConfig(@JsonProperty("batchConfigMaps") @Nullable List<Map<String, String>> batchConfigMaps,
+ @JsonProperty("segmentPushType") String segmentPushType,
+ @JsonProperty("segmentPushFrequency") String segmentPushFrequency) {
+ _batchConfigMaps = batchConfigMaps;
+ _segmentPushType = segmentPushType;
+ _segmentPushFrequency = segmentPushFrequency;
+ }
+
+ @Nullable
+ public List<Map<String, String>> getBatchConfigMaps() {
+ return _batchConfigMaps;
+ }
+
+ public String getSegmentPushType() {
+ return _segmentPushType;
+ }
+
+ public String getSegmentPushFrequency() {
+ return _segmentPushFrequency;
+ }
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IngestionConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/IngestionConfig.java
similarity index 63%
rename from pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IngestionConfig.java
rename to pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/IngestionConfig.java
index aafa5a2..cf6d3e5 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IngestionConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/IngestionConfig.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.spi.config.table;
+package org.apache.pinot.spi.config.table.ingestion;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -24,17 +24,19 @@ import com.fasterxml.jackson.annotation.JsonPropertyDescription;
import java.util.List;
import javax.annotation.Nullable;
import org.apache.pinot.spi.config.BaseJsonConfig;
-import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
-import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
/**
- * Class representing table ingestion configuration. For example, configs needed for
- * decoding, data source configs, ingestion transformations (flattening, filtering, transformations etc.)
- * TODO: Move fields from table config which are ingestion related (e.g. streamConfigs, segment push type, append freq etc)
+ * Class representing table ingestion configuration i.e. all configs related to the data source and the ingestion properties and operations
*/
public class IngestionConfig extends BaseJsonConfig {
+ @JsonPropertyDescription("Config related to the batch data sources")
+ private final BatchIngestionConfig _batchIngestionConfig;
+
+ @JsonPropertyDescription("Config related to the stream data sources")
+ private final StreamIngestionConfig _streamIngestionConfig;
+
@JsonPropertyDescription("Config related to filtering records during ingestion")
private final FilterConfig _filterConfig;
@@ -42,13 +44,27 @@ public class IngestionConfig extends BaseJsonConfig {
private final List<TransformConfig> _transformConfigs;
@JsonCreator
- public IngestionConfig(@JsonProperty("filterConfig") @Nullable FilterConfig filterConfig,
+ public IngestionConfig(@JsonProperty("batchIngestionConfig") @Nullable BatchIngestionConfig batchIngestionConfig,
+ @JsonProperty("streamIngestionConfig") @Nullable StreamIngestionConfig streamIngestionConfig,
+ @JsonProperty("filterConfig") @Nullable FilterConfig filterConfig,
@JsonProperty("transformConfigs") @Nullable List<TransformConfig> transformConfigs) {
+ _batchIngestionConfig = batchIngestionConfig;
+ _streamIngestionConfig = streamIngestionConfig;
_filterConfig = filterConfig;
_transformConfigs = transformConfigs;
}
@Nullable
+ public BatchIngestionConfig getBatchIngestionConfig() {
+ return _batchIngestionConfig;
+ }
+
+ @Nullable
+ public StreamIngestionConfig getStreamIngestionConfig() {
+ return _streamIngestionConfig;
+ }
+
+ @Nullable
public FilterConfig getFilterConfig() {
return _filterConfig;
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/StreamIngestionConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/StreamIngestionConfig.java
new file mode 100644
index 0000000..0ec379d
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/StreamIngestionConfig.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.config.table.ingestion;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.config.BaseJsonConfig;
+
+
+/**
+ * Contains all the configs related to the streams for ingestion
+ */
+public class StreamIngestionConfig extends BaseJsonConfig {
+
+ @JsonPropertyDescription("All configs for the streams from which to ingest")
+ private final List<Map<String, String>> _streamConfigMaps;
+
+ @JsonCreator
+ public StreamIngestionConfig(@JsonProperty("streamConfigMaps") List<Map<String, String>> streamConfigMaps) {
+ _streamConfigMaps = streamConfigMaps;
+ }
+
+ public List<Map<String, String>> getStreamConfigMaps() {
+ return _streamConfigMaps;
+ }
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/BatchConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/BatchConfig.java
new file mode 100644
index 0000000..965eb9a
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/BatchConfig.java
@@ -0,0 +1,133 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.ingestion.batch;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.data.readers.FileFormat;
+
+
+/**
+ * Provides all config related to the batch data source, as configured in the table config's ingestion config
+ */
+public class BatchConfig {
+ private final String _tableNameWithType;
+ private final String _type;
+ private final String _inputDirURI;
+ private final String _outputDirURI;
+ private final String _fsClassName;
+ private final Map<String, String> _fsProps = new HashMap<>();
+ private final FileFormat _inputFormat;
+ private final String _recordReaderClassName;
+ private final String _recordReaderConfigClassName;
+ private final Map<String, String> _recordReaderProps = new HashMap<>();
+
+ private final Map<String, String> _batchConfigMap = new HashMap<>();
+
+ public BatchConfig(String tableNameWithType, Map<String, String> batchConfigsMap) {
+ _tableNameWithType = tableNameWithType;
+
+ _type = batchConfigsMap.get(BatchConfigProperties.BATCH_TYPE);
+ Preconditions.checkState(_type != null, "Property: %s cannot be null", BatchConfigProperties.BATCH_TYPE);
+
+ String inputDirURIKey = BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.INPUT_DIR_URI);
+ _inputDirURI = batchConfigsMap.get(inputDirURIKey);
+ Preconditions.checkState(_inputDirURI != null, "Property: %s cannot be null", inputDirURIKey);
+
+ String outputDirURIKey = BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.OUTPUT_DIR_URI);
+ _outputDirURI = batchConfigsMap.get(outputDirURIKey);
+ Preconditions.checkState(_outputDirURI != null, "Property: %s cannot be null", outputDirURIKey);
+
+ String fsClassNameKey = BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.FS_CLASS);
+ _fsClassName = batchConfigsMap.get(fsClassNameKey);
+ Preconditions.checkState(_fsClassName != null, "Property: %s cannot be null", fsClassNameKey);
+
+ String inputFormatKey = BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.INPUT_FORMAT);
+ String inputFormat = batchConfigsMap.get(inputFormatKey);
+ Preconditions.checkState(inputFormat != null, "Property: %s cannot be null", inputFormat);
+ _inputFormat = FileFormat.valueOf(inputFormat.toUpperCase());
+
+ String recordReaderClassNameKey =
+ BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.RECORD_READER_CLASS);
+ _recordReaderClassName = batchConfigsMap.get(recordReaderClassNameKey);
+ Preconditions.checkState(_recordReaderClassName != null, "Property: %s cannot be null", recordReaderClassNameKey);
+
+ String recordReaderConfigClassNameKey =
+ BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.RECORD_READER_CONFIG_CLASS);
+ _recordReaderConfigClassName = batchConfigsMap.get(recordReaderConfigClassNameKey);
+
+ String fsPropPrefix = BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.FS_PROP_PREFIX);
+ String recordReaderPropPrefix =
+ BatchConfigProperties.constructBatchProperty(_type, BatchConfigProperties.RECORD_READER_PROP_PREFIX);
+ for (Map.Entry<String, String> entry : batchConfigsMap.entrySet()) {
+ String key = entry.getKey();
+ if (key.startsWith(fsPropPrefix)) {
+ _fsProps.put(key, entry.getValue());
+ } else if (key.startsWith(recordReaderPropPrefix)) {
+ _recordReaderProps.put(key, entry.getValue());
+ }
+ _batchConfigMap.put(key, entry.getValue());
+ }
+ }
+
+ public String getTableNameWithType() {
+ return _tableNameWithType;
+ }
+
+ public String getType() {
+ return _type;
+ }
+
+ public String getInputDirURI() {
+ return _inputDirURI;
+ }
+
+ public String getOutputDirURI() {
+ return _outputDirURI;
+ }
+
+ public String getFsClassName() {
+ return _fsClassName;
+ }
+
+ public Map<String, String> getFsProps() {
+ return _fsProps;
+ }
+
+ public FileFormat getInputFormat() {
+ return _inputFormat;
+ }
+
+ public String getRecordReaderClassName() {
+ return _recordReaderClassName;
+ }
+
+ public String getRecordReaderConfigClassName() {
+ return _recordReaderConfigClassName;
+ }
+
+ public Map<String, String> getRecordReaderProps() {
+ return _recordReaderProps;
+ }
+
+ public Map<String, String> getBatchConfigMap() {
+ return _batchConfigMap;
+ }
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/BatchConfigProperties.java b/pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/BatchConfigProperties.java
new file mode 100644
index 0000000..2ebda7b
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/BatchConfigProperties.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.ingestion.batch;
+
+import org.apache.commons.lang3.StringUtils;
+
+
+/**
+ * Defines all the keys used in the batch configs map
+ */
+public class BatchConfigProperties {
+
+ public static final String DOT_SEPARATOR = ".";
+ public static final String BATCH_PREFIX = "batch";
+
+ public static final String BATCH_TYPE = "batchType";
+ public static final String INPUT_DIR_URI = "inputDirURI";
+ public static final String OUTPUT_DIR_URI = "outputDirURI";
+ public static final String FS_CLASS = "fs.className";
+ public static final String FS_PROP_PREFIX = "fs.prop";
+ public static final String INPUT_FORMAT = "inputFormat";
+ public static final String RECORD_READER_CLASS = "recordReader.className";
+ public static final String RECORD_READER_CONFIG_CLASS = "recordReader.config.className";
+ public static final String RECORD_READER_PROP_PREFIX = "recordReader.prop";
+
+ /**
+ * Helper method to create a batch config property
+ */
+ public static String constructBatchProperty(String batchType, String property) {
+ return StringUtils.join(BATCH_PREFIX, batchType, property, DOT_SEPARATOR);
+ }
+
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/IngestionConfigUtils.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/IngestionConfigUtils.java
new file mode 100644
index 0000000..717aab2
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/IngestionConfigUtils.java
@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.utils;
+
+import com.google.common.base.Preconditions;
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig;
+
+
+/**
+ * Helper methods for extracting fields from IngestionConfig in a backward compatible manner
+ */
+public final class IngestionConfigUtils {
+ /**
+ * Fetches the streamConfig from the given realtime table.
+ * First, the ingestionConfigs->stream->streamConfigs will be checked.
+ * If not found, the indexingConfig->streamConfigs will be checked (which is deprecated).
+ * @param tableConfig realtime table config
+ * @return streamConfigs map
+ */
+ public static Map<String, String> getStreamConfigMap(TableConfig tableConfig) {
+ String tableNameWithType = tableConfig.getTableName();
+ Preconditions.checkState(tableConfig.getTableType() == TableType.REALTIME,
+ "Cannot fetch streamConfigs for OFFLINE table: %s", tableNameWithType);
+ Map<String, String> streamConfigMap = null;
+ if (tableConfig.getIngestionConfig() != null && tableConfig.getIngestionConfig().getStreamIngestionConfig() != null) {
+ List<Map<String, String>> streamConfigMaps = tableConfig.getIngestionConfig().getStreamIngestionConfig().getStreamConfigMaps();
+ Preconditions.checkState(streamConfigMaps.size() == 1, "Only 1 stream supported per table");
+ streamConfigMap = streamConfigMaps.get(0);
+ }
+ if (streamConfigMap == null && tableConfig.getIndexingConfig() != null) {
+ streamConfigMap = tableConfig.getIndexingConfig().getStreamConfigs();
+ }
+ if (streamConfigMap == null) {
+ throw new IllegalStateException("Could not find streamConfigs for REALTIME table: " + tableNameWithType);
+ }
+ return streamConfigMap;
+ }
+
+ /**
+ * Fetches the configured segmentPushType (APPEND/REFRESH) from the table config
+ * First checks in the ingestionConfig. If not found, checks in the segmentsConfig (has been deprecated from here in favor of ingestion config)
+ */
+ public static String getBatchSegmentPushType(TableConfig tableConfig) {
+ String segmentPushType = null;
+ if (tableConfig.getIngestionConfig() != null) {
+ BatchIngestionConfig batchIngestionConfig = tableConfig.getIngestionConfig().getBatchIngestionConfig();
+ if (batchIngestionConfig != null) {
+ segmentPushType = batchIngestionConfig.getSegmentPushType();
+ }
+ }
+ if (segmentPushType == null) {
+ segmentPushType = tableConfig.getValidationConfig().getSegmentPushType();
+ }
+ return segmentPushType;
+ }
+
+ /**
+ * Fetches the configured segmentPushFrequency from the table config
+ * First checks in the ingestionConfig. If not found, checks in the segmentsConfig (has been deprecated from here in favor of ingestion config)
+ */
+ public static String getBatchSegmentPushFrequency(TableConfig tableConfig) {
+ String segmentPushFrequency = null;
+ if (tableConfig.getIngestionConfig() != null) {
+ BatchIngestionConfig batchIngestionConfig = tableConfig.getIngestionConfig().getBatchIngestionConfig();
+ if (batchIngestionConfig != null) {
+ segmentPushFrequency = batchIngestionConfig.getSegmentPushFrequency();
+ }
+ }
+ if (segmentPushFrequency == null) {
+ segmentPushFrequency = tableConfig.getValidationConfig().getSegmentPushFrequency();
+ }
+ return segmentPushFrequency;
+ }
+
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
index 0e32a02..3a762a5 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
@@ -25,7 +25,7 @@ import java.util.Map;
import org.apache.pinot.spi.config.table.CompletionConfig;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.IndexingConfig;
-import org.apache.pinot.spi.config.table.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.QueryConfig;
import org.apache.pinot.spi.config.table.QuotaConfig;
import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig;
diff --git a/pinot-spi/src/test/java/org/apache/pinot/spi/ingestion/batch/BatchConfigTest.java b/pinot-spi/src/test/java/org/apache/pinot/spi/ingestion/batch/BatchConfigTest.java
new file mode 100644
index 0000000..bc34f1a
--- /dev/null
+++ b/pinot-spi/src/test/java/org/apache/pinot/spi/ingestion/batch/BatchConfigTest.java
@@ -0,0 +1,171 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.ingestion.batch;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.data.readers.FileFormat;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/**
+ * Tests for BatchConfigs
+ */
+public class BatchConfigTest {
+
+ @Test
+ public void testBatchConfig() {
+ Map<String, String> batchConfigMap = new HashMap<>();
+ String tableName = "foo_REALTIME";
+ String batchType = "s3";
+ String inputDir = "s3://foo/input";
+ String outputDir = "s3://foo/output";
+ String fsClass = "org.apache.S3FS";
+ String region = "us-west";
+ String username = "foo";
+ String inputFormat = "csv";
+ String recordReaderClass = "org.foo.CSVRecordReader";
+ String recordReaderConfigClass = "org.foo.CSVRecordReaderConfig";
+ String separator = "|";
+ batchConfigMap.put(BatchConfigProperties.BATCH_TYPE, batchType);
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.INPUT_DIR_URI), inputDir);
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.OUTPUT_DIR_URI), outputDir);
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.FS_CLASS), fsClass);
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.INPUT_FORMAT), inputFormat);
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.RECORD_READER_CLASS),
+ recordReaderClass);
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.RECORD_READER_CONFIG_CLASS),
+ recordReaderConfigClass);
+ batchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.FS_PROP_PREFIX) + ".region",
+ region);
+ batchConfigMap.put(
+ BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.FS_PROP_PREFIX) + ".username",
+ username);
+ batchConfigMap.put(
+ BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.RECORD_READER_PROP_PREFIX)
+ + ".separator", separator);
+
+ // config with all the right properties
+ BatchConfig batchConfig = new BatchConfig(tableName, batchConfigMap);
+ assertEquals(batchConfig.getType(), batchType);
+ assertEquals(batchConfig.getInputDirURI(), inputDir);
+ assertEquals(batchConfig.getOutputDirURI(), outputDir);
+ assertEquals(batchConfig.getFsClassName(), fsClass);
+ assertEquals(batchConfig.getInputFormat(), FileFormat.CSV);
+ assertEquals(batchConfig.getRecordReaderClassName(), recordReaderClass);
+ assertEquals(batchConfig.getRecordReaderConfigClassName(), recordReaderConfigClass);
+ assertEquals(batchConfig.getFsProps().size(), 2);
+ assertEquals(batchConfig.getFsProps().get(
+ BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.FS_PROP_PREFIX) + ".region"),
+ region);
+ assertEquals(batchConfig.getFsProps().get(
+ BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.FS_PROP_PREFIX) + ".username"),
+ username);
+ assertEquals(batchConfig.getRecordReaderProps().size(), 1);
+ assertEquals(batchConfig.getRecordReaderProps().get(
+ BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.RECORD_READER_PROP_PREFIX)
+ + ".separator"), separator);
+ assertEquals(batchConfig.getTableNameWithType(), tableName);
+
+ // Missing props
+ Map<String, String> testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap.remove(BatchConfigProperties.BATCH_TYPE);
+ try {
+ new BatchConfig(tableName, testBatchConfigMap);
+ Assert.fail("Should fail for missing 'batchType");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap
+ .remove(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.INPUT_DIR_URI));
+ try {
+ new BatchConfig(tableName, testBatchConfigMap);
+ Assert.fail("Should fail for missing 'inputDirURI");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap
+ .remove(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.OUTPUT_DIR_URI));
+ try {
+ new BatchConfig(tableName, testBatchConfigMap);
+ Assert.fail("Should fail for missing 'outputDirURI");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap
+ .remove(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.INPUT_FORMAT));
+ try {
+ new BatchConfig(tableName, testBatchConfigMap);
+ Assert.fail("Should fail for missing 'inputFormat");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap
+ .put(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.INPUT_FORMAT), "moo");
+ try {
+ new BatchConfig(tableName, testBatchConfigMap);
+ Assert.fail("Should fail for incorrect 'inputFormat");
+ } catch (IllegalArgumentException e) {
+ // expected
+ }
+
+ testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap
+ .remove(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.RECORD_READER_CLASS));
+ try {
+ new BatchConfig(tableName, testBatchConfigMap);
+ Assert.fail("Should fail for missing 'recordReaderClassName");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap
+ .remove(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.RECORD_READER_CONFIG_CLASS));
+ new BatchConfig(tableName, testBatchConfigMap);
+
+ testBatchConfigMap = new HashMap<>(batchConfigMap);
+ testBatchConfigMap
+ .remove(BatchConfigProperties.constructBatchProperty(batchType, BatchConfigProperties.FS_CLASS));
+ try {
+ new BatchConfig(tableName, testBatchConfigMap);
+ Assert.fail("Should fail for missing 'fsClassName");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+ }
+}
diff --git a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/IngestionConfigUtilsTest.java b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/IngestionConfigUtilsTest.java
new file mode 100644
index 0000000..dee56bb
--- /dev/null
+++ b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/IngestionConfigUtilsTest.java
@@ -0,0 +1,149 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.utils;
+
+import com.google.common.collect.Lists;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.config.table.IndexingConfig;
+import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Tests for helper methods in {@link IngestionConfigUtils}
+ */
+public class IngestionConfigUtilsTest {
+
+ @Test
+ public void testGetStreamConfigMap() {
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
+ try {
+ IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ Assert.fail("Should fail for OFFLINE table");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ tableConfig =
+ new TableConfigBuilder(TableType.REALTIME).setTableName("myTable").setTimeColumnName("timeColumn").build();
+
+ // get from ingestion config (when not present in indexing config)
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("streamType", "kafka");
+ tableConfig.setIngestionConfig(
+ new IngestionConfig(null, new StreamIngestionConfig(Lists.newArrayList(streamConfigMap)), null, null));
+ Map<String, String> actualStreamConfigsMap = IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ Assert.assertEquals(actualStreamConfigsMap.size(), 1);
+ Assert.assertEquals(actualStreamConfigsMap.get("streamType"), "kafka");
+
+ // get from ingestion config (even if present in indexing config)
+ Map<String, String> deprecatedStreamConfigMap = new HashMap<>();
+ deprecatedStreamConfigMap.put("streamType", "foo");
+ deprecatedStreamConfigMap.put("customProp", "foo");
+ IndexingConfig indexingConfig = new IndexingConfig();
+ indexingConfig.setStreamConfigs(deprecatedStreamConfigMap);
+ tableConfig.setIndexingConfig(indexingConfig);
+ actualStreamConfigsMap = IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ Assert.assertEquals(actualStreamConfigsMap.size(), 1);
+ Assert.assertEquals(actualStreamConfigsMap.get("streamType"), "kafka");
+
+ // fail if multiple found
+ tableConfig.setIngestionConfig(new IngestionConfig(null,
+ new StreamIngestionConfig(Lists.newArrayList(streamConfigMap, deprecatedStreamConfigMap)), null, null));
+ try {
+ IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ Assert.fail("Should fail for multiple stream configs");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ // get from indexing config
+ tableConfig.setIngestionConfig(null);
+ actualStreamConfigsMap = IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ Assert.assertEquals(actualStreamConfigsMap.size(), 2);
+ Assert.assertEquals(actualStreamConfigsMap.get("streamType"), "foo");
+
+ // fail if found nowhere
+ tableConfig.setIndexingConfig(null);
+ try {
+ IngestionConfigUtils.getStreamConfigMap(tableConfig);
+ Assert.fail("Should fail for no stream config found");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testGetPushFrequency() {
+ // get from ingestion config, when not present in segmentsConfig
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
+ tableConfig
+ .setIngestionConfig(new IngestionConfig(new BatchIngestionConfig(null, "APPEND", "HOURLY"), null, null, null));
+ Assert.assertEquals(IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig), "HOURLY");
+
+ // get from ingestion config, even if present in segmentsConfig
+ SegmentsValidationAndRetentionConfig segmentsValidationAndRetentionConfig =
+ new SegmentsValidationAndRetentionConfig();
+ segmentsValidationAndRetentionConfig.setSegmentPushFrequency("DAILY");
+ tableConfig.setValidationConfig(segmentsValidationAndRetentionConfig);
+ Assert.assertEquals(IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig), "HOURLY");
+
+ // get from segmentsConfig
+ tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
+ tableConfig.setValidationConfig(segmentsValidationAndRetentionConfig);
+ Assert.assertEquals(IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig), "DAILY");
+
+ // present nowhere
+ segmentsValidationAndRetentionConfig.setSegmentPushFrequency(null);
+ Assert.assertNull(IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig));
+ }
+
+ @Test
+ public void testGetPushType() {
+ // get from ingestion config, when not present in segmentsConfig
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
+ tableConfig
+ .setIngestionConfig(new IngestionConfig(new BatchIngestionConfig(null, "APPEND", "HOURLY"), null, null, null));
+ Assert.assertEquals(IngestionConfigUtils.getBatchSegmentPushType(tableConfig), "APPEND");
+
+ // get from ingestion config, even if present in segmentsConfig
+ SegmentsValidationAndRetentionConfig segmentsValidationAndRetentionConfig =
+ new SegmentsValidationAndRetentionConfig();
+ segmentsValidationAndRetentionConfig.setSegmentPushType("REFRESH");
+ tableConfig.setValidationConfig(segmentsValidationAndRetentionConfig);
+ Assert.assertEquals(IngestionConfigUtils.getBatchSegmentPushType(tableConfig), "APPEND");
+
+ // get from segmentsConfig
+ tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
+ tableConfig.setValidationConfig(segmentsValidationAndRetentionConfig);
+ Assert.assertEquals(IngestionConfigUtils.getBatchSegmentPushType(tableConfig), "REFRESH");
+
+ // present nowhere
+ segmentsValidationAndRetentionConfig.setSegmentPushType(null);
+ Assert.assertNull(IngestionConfigUtils.getBatchSegmentPushType(tableConfig));
+ }
+}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/segment/converter/SegmentMergeCommand.java b/pinot-tools/src/main/java/org/apache/pinot/tools/segment/converter/SegmentMergeCommand.java
index d7b5c7f..be2196d 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/segment/converter/SegmentMergeCommand.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/segment/converter/SegmentMergeCommand.java
@@ -39,6 +39,7 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DateTimeFormatSpec;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.tools.Command;
@@ -238,8 +239,8 @@ public class SegmentMergeCommand extends AbstractBaseAdminCommand implements Com
// Fetch time related configurations from schema and table config.
SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
- String pushFrequency = validationConfig.getSegmentPushFrequency();
- String pushType = validationConfig.getSegmentPushType();
+ String pushFrequency = IngestionConfigUtils.getBatchSegmentPushFrequency(tableConfig);
+ String pushType = IngestionConfigUtils.getBatchSegmentPushType(tableConfig);
String timeColumnName = validationConfig.getTimeColumnName();
DateTimeFormatSpec dateTimeFormatSpec = null;
if (timeColumnName != null) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org