This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new c6552066091 Support no-dictionary aggregation-key columns in consuming
segments and consolidate metrics-aggregation validation (#18955)
c6552066091 is described below
commit c6552066091a259513e0503a95a1e704a7f14543
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Thu Jul 9 23:47:57 2026 -0700
Support no-dictionary aggregation-key columns in consuming segments and
consolidate metrics-aggregation validation (#18955)
---
.../indexsegment/mutable/MutableSegmentImpl.java | 82 ++-
.../segment/local/utils/TableConfigUtils.java | 626 +++++++++++----------
.../MutableSegmentImplAggregateMetricsTest.java | 67 +++
.../segment/local/utils/TableConfigUtilsTest.java | 126 ++++-
4 files changed, 540 insertions(+), 361 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
index 3eac1210a2a..dbf8f935311 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
@@ -575,20 +575,17 @@ public class MutableSegmentImpl implements MutableSegment
{
if (indexConfigs.getConfig(StandardIndexes.dictionary()).isEnabled()) {
return false;
}
- // Earlier we didn't support noDict in consuming segments for STRING and
BYTES columns.
- // So even if the user had the column in noDictionaryColumns set in table
config, we still
- // created dictionary in consuming segments.
- // Later on we added this support. There is a particular impact of this
change on the use cases
- // that have set noDict on their STRING dimension columns for other
performance
- // reasons and also want metricsAggregation. These use cases don't get to
- // aggregateMetrics because the new implementation is able to honor their
table config setting
- // of noDict on STRING/BYTES. Without metrics aggregation, memory pressure
increases.
- // So to continue aggregating metrics for such cases, we will create
dictionary even
- // if the column is part of noDictionary set from table config
- if (fieldSpec instanceof DimensionFieldSpec && isAggregateMetricsEnabled()
&& (dataType == STRING
- || dataType == BYTES)) {
- _logger.info("Aggregate metrics is enabled. Will create dictionary in
consuming segment for column {} of type {}",
- column, dataType);
+ // Metrics aggregation keys each row on the dictionary ids of the
dimension and time columns (see
+ // getOrCreateDocId), so those columns must be dictionary encoded in the
consuming segment even when the table
+ // config marks them as no-dictionary. The consuming-segment dictionary is
a transient structure that only exists
+ // to drive the in-memory rollup; the committed segment is rebuilt from
the table config (see
+ // RealtimeSegmentConverter), so the no-dictionary setting is still
honored there. Metric columns are excluded:
+ // aggregated values are mutated in place in the raw forward index and
must stay no-dictionary.
+ FieldSpec.FieldType fieldType = fieldSpec.getFieldType();
+ if (isAggregateMetricsEnabled() && (fieldType ==
FieldSpec.FieldType.DIMENSION
+ || fieldType == FieldSpec.FieldType.DATE_TIME || fieldType ==
FieldSpec.FieldType.TIME)) {
+ _logger.info("Metrics aggregation is enabled. Will create dictionary in
consuming segment for key column: {} of "
+ + "type: {}", column, dataType);
return false;
}
// So don't create dictionary if the column (1) is member of noDictionary,
and (2) is single-value or multi-value
@@ -1473,6 +1470,9 @@ public class MutableSegmentImpl implements MutableSegment
{
}
int i = 0;
+ // Dimension and time columns form the aggregation key. They are always
dictionary encoded in the consuming
+ // segment (isNoDictionaryColumn forces a dictionary on them when
aggregation is enabled), so the _dictId read
+ // below is always valid. Keep this set of columns in sync with the field
types forced there.
int[] dictIds = new int[_numKeyColumns]; // dimensions + date time columns
+ time column.
// FIXME: this for loop breaks for multi value dimensions.
https://github.com/apache/pinot/issues/3867
@@ -1485,22 +1485,21 @@ public class MutableSegmentImpl implements
MutableSegment {
return _recordIdMap.put(new FixedIntArray(dictIds));
}
- /**
- * Helper method to enable/initialize aggregation of metrics, based on
following conditions:
- * <ul>
- * <li> Config to enable aggregation of metrics is specified. </li>
- * <li> All dimensions and time are dictionary encoded. This is because an
integer array containing dictionary id's
- * is used as key for dimensions to record Id map. </li>
- * <li> None of the metrics are dictionary encoded. </li>
- * <li> All columns should be single-valued (see
https://github.com/apache/pinot/issues/3867)</li>
- * </ul>
- *
- * TODO: Eliminate the requirement on dictionary encoding for dimension and
metric columns.
- *
- * @param config Segment config.
- *
- * @return Map from dictionary id array to doc id, null if metrics
aggregation cannot be enabled.
- */
+ /// Enables and initializes metrics aggregation for the consuming segment
when configured and feasible.
+ ///
+ /// Aggregation is enabled when all of the following hold:
+ /// - The `aggregateMetrics` flag or ingestion `aggregationConfigs` is
specified.
+ /// - No metric column is dictionary encoded. Aggregated values are mutated
in place in the raw forward index, so
+ /// metrics must stay no-dictionary.
+ /// - All metric and dimension columns are single-valued (see
https://github.com/apache/pinot/issues/3867).
+ ///
+ /// Dimension and time columns form the aggregation key via their dictionary
ids (see [#getOrCreateDocId]), so they
+ /// must be dictionary encoded. This is not required from the caller:
[#isNoDictionaryColumn] forces a dictionary on
+ /// those columns in the consuming segment whenever aggregation is enabled,
even when the table config marks them as
+ /// no-dictionary. The committed segment is rebuilt from the table config,
so the no-dictionary setting is still
+ /// honored there.
+ ///
+ /// Returns the map from dictionary id array to doc id, or `null` if metrics
aggregation cannot be enabled.
private IdMap<FixedIntArray>
enableMetricsAggregationIfPossible(RealtimeSegmentConfig config) {
Set<String> noDictionaryColumns =
FieldIndexConfigsUtil.columnsWithIndexDisabled(StandardIndexes.dictionary(),
config.getIndexConfigByCol());
@@ -1526,29 +1525,12 @@ public class MutableSegmentImpl implements
MutableSegment {
}
}
- // All dimension columns should be dictionary encoded.
- // All dimension columns must be single value
+ // All dimension columns must be single value. No-dictionary dimensions
are supported: isNoDictionaryColumn()
+ // forces a dictionary on them in the consuming segment so they can be
used as the aggregation key.
for (FieldSpec fieldSpec : _physicalDimensionFieldSpecs) {
- String dimension = fieldSpec.getName();
- if (noDictionaryColumns.contains(dimension)) {
- _logger.warn("Metrics aggregation cannot be turned ON in presence of
no-dictionary dimensions, eg: {}",
- dimension);
- return null;
- }
-
if (!fieldSpec.isSingleValueField()) {
_logger.warn("Metrics aggregation cannot be turned ON in presence of
multi-value dimension columns, eg: {}",
- dimension);
- return null;
- }
- }
-
- // Time columns should be dictionary encoded.
- for (String timeColumnName : _physicalTimeColumnNames) {
- if (noDictionaryColumns.contains(timeColumnName)) {
- _logger.warn(
- "Metrics aggregation cannot be turned ON in presence of
no-dictionary datetime/time columns, eg: {}",
- timeColumnName);
+ fieldSpec.getName());
return null;
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 9106f7fe56b..f5330eafbde 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -443,325 +443,371 @@ public final class TableConfigUtils {
||
CommonConstants.HTTPS_PROTOCOL.equalsIgnoreCase(peerSegmentDownloadScheme);
}
- /**
- * Validates the following:
- * 1. validity of filter function
- * 2. checks for duplicate transform configs
- * 3. checks for null column name or transform function in transform config
- * 4. validity of transform function string
- * 5. checks for source fields used in destination columns
- * 6. ingestion type for dimension tables
- */
+ /// Validates the table's [IngestionConfig] together with the closely
related metrics-aggregation config, covering:
+ /// - Metrics aggregation (both the `aggregateMetrics` flag and ingestion
`aggregationConfigs`), delegated to
+ /// [#validateMetricsAggregation].
+ /// - Batch ingestion: each batch config map is well-formed, and a dimension
table has batch ingestion configured
+ /// with `REFRESH` segment ingestion type.
+ /// - Stream ingestion: streams are declared in only one place, at least one
stream is present, and pauseless
+ /// consumption has a valid `peerSegmentDownloadScheme`.
+ /// - Filter config: the filter function is valid and not a disabled Groovy
expression.
+ /// - Source field configs: no source field is duplicated within the same
complex-type phase.
+ /// - Enrichment configs: each config is valid.
+ /// - Transform configs: non-null column and function, no duplicate
destination, and each destination is a schema
+ /// column, an intermediate consumed by another transform, or an
aggregation source column; the function is valid
+ /// and does not reference its own destination.
+ /// - Complex-type config: no schema field collides with a
`prefixesToRename` prefix.
+ /// - Schema-conforming transformer config.
@VisibleForTesting
public static void validateIngestionConfig(TableConfig tableConfig, Schema
schema) {
+ // All metrics-aggregation validation lives here; it returns the columns
referenced as aggregation sources, which
+ // a transform config is allowed to target as its destination (see the
transform validation below).
+ Set<String> aggregationSourceColumns =
validateMetricsAggregation(tableConfig, schema);
+
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
+ if (ingestionConfig == null) {
+ return;
+ }
- if (ingestionConfig != null) {
- String tableNameWithType = tableConfig.getTableName();
+ String tableNameWithType = tableConfig.getTableName();
- // Batch
- if (ingestionConfig.getBatchIngestionConfig() != null) {
- BatchIngestionConfig cfg = ingestionConfig.getBatchIngestionConfig();
- List<Map<String, String>> batchConfigMaps = cfg.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);
- }
- if (tableConfig.isDimTable()) {
-
Preconditions.checkState(cfg.getSegmentIngestionType().equalsIgnoreCase("REFRESH"),
- "Dimension tables must have segment ingestion type REFRESH");
+ // Batch
+ if (ingestionConfig.getBatchIngestionConfig() != null) {
+ BatchIngestionConfig cfg = ingestionConfig.getBatchIngestionConfig();
+ List<Map<String, String>> batchConfigMaps = cfg.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);
}
if (tableConfig.isDimTable()) {
- Preconditions.checkState(ingestionConfig.getBatchIngestionConfig() !=
null,
- "Dimension tables must have batch ingestion configuration");
- }
-
- // Stream
- // stream config map can either be in ingestion config or indexing
config. cannot be in both places
- 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");
- StreamIngestionConfig streamIngestionConfig =
ingestionConfig.getStreamIngestionConfig();
- List<Map<String, String>> streamConfigMaps =
streamIngestionConfig.getStreamConfigMaps();
- Preconditions.checkState(!streamConfigMaps.isEmpty(), "Must have at
least 1 stream in REALTIME table");
- // TODO: for multiple stream configs, validate them
-
- boolean isPauselessEnabled =
streamIngestionConfig.isPauselessConsumptionEnabled();
- if (isPauselessEnabled) {
- int replication = tableConfig.getReplication();
- // We are checking for this only when replication is greater than 1
because in test environments
- // users still prefer to create pauseless tables with replication 1
- if (replication > 1) {
- String peerSegmentDownloadScheme =
tableConfig.getValidationConfig().getPeerSegmentDownloadScheme();
-
Preconditions.checkState(StringUtils.isNotEmpty(peerSegmentDownloadScheme) &&
isValidPeerDownloadScheme(
- peerSegmentDownloadScheme),
- "Must have a valid peerSegmentDownloadScheme set in validation
config for pauseless consumption");
- } else {
- LOGGER.warn("It's not recommended to create pauseless tables with
replication 1 for stability reasons.");
- }
- }
+
Preconditions.checkState(cfg.getSegmentIngestionType().equalsIgnoreCase("REFRESH"),
+ "Dimension tables must have segment ingestion type REFRESH");
}
+ }
+ if (tableConfig.isDimTable()) {
+ Preconditions.checkState(ingestionConfig.getBatchIngestionConfig() !=
null,
+ "Dimension tables must have batch ingestion configuration");
+ }
- // Filter config
- FilterConfig filterConfig = ingestionConfig.getFilterConfig();
- if (filterConfig != null) {
- String filterFunction = filterConfig.getFilterFunction();
- if (filterFunction != null) {
- if (_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(filterFunction)) {
- throw new IllegalStateException(
- "Groovy filter functions are disabled for table config. Found
'" + filterFunction + "'");
- }
- try {
- FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction);
- } catch (Exception e) {
- throw new IllegalStateException(
- "Invalid filter function '" + filterFunction + "', exception:
" + e.getMessage(), e);
- }
+ // Stream
+ // stream config map can either be in ingestion config or indexing config.
cannot be in both places
+ 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");
+ StreamIngestionConfig streamIngestionConfig =
ingestionConfig.getStreamIngestionConfig();
+ List<Map<String, String>> streamConfigMaps =
streamIngestionConfig.getStreamConfigMaps();
+ Preconditions.checkState(!streamConfigMaps.isEmpty(), "Must have at
least 1 stream in REALTIME table");
+ // TODO: for multiple stream configs, validate them
+
+ boolean isPauselessEnabled =
streamIngestionConfig.isPauselessConsumptionEnabled();
+ if (isPauselessEnabled) {
+ int replication = tableConfig.getReplication();
+ // We are checking for this only when replication is greater than 1
because in test environments
+ // users still prefer to create pauseless tables with replication 1
+ if (replication > 1) {
+ String peerSegmentDownloadScheme =
tableConfig.getValidationConfig().getPeerSegmentDownloadScheme();
+ Preconditions.checkState(
+ StringUtils.isNotEmpty(peerSegmentDownloadScheme) &&
isValidPeerDownloadScheme(peerSegmentDownloadScheme),
+ "Must have a valid peerSegmentDownloadScheme set in validation
config for pauseless consumption");
+ } else {
+ LOGGER.warn("It's not recommended to create pauseless tables with
replication 1 for stability reasons.");
}
}
+ }
- // Aggregation configs
- List<AggregationConfig> aggregationConfigs =
ingestionConfig.getAggregationConfigs();
- Set<String> aggregationSourceColumns = new HashSet<>();
- if (CollectionUtils.isNotEmpty(aggregationConfigs)) {
-
Preconditions.checkState(!tableConfig.getIndexingConfig().isAggregateMetrics(),
- "aggregateMetrics cannot be set with AggregationConfig");
- Set<String> aggregationColumns = new HashSet<>();
- for (AggregationConfig aggregationConfig : aggregationConfigs) {
- String columnName = aggregationConfig.getColumnName();
- String aggregationFunction =
aggregationConfig.getAggregationFunction();
- if (columnName == null || aggregationFunction == null) {
- throw new IllegalStateException(
- "columnName/aggregationFunction cannot be null in
AggregationConfig " + aggregationConfig);
- }
-
- FieldSpec fieldSpec = schema.getFieldSpecFor(columnName);
- Preconditions.checkState(fieldSpec != null,
- "The destination column '" + columnName + "' of the aggregation
function must be present in the schema");
- Preconditions.checkState(fieldSpec.getFieldType() ==
FieldSpec.FieldType.METRIC,
- "The destination column '" + columnName + "' of the aggregation
function must be a metric column");
- DataType dataType = fieldSpec.getDataType();
-
- if (!aggregationColumns.add(columnName)) {
- throw new IllegalStateException("Duplicate aggregation config
found for column '" + columnName + "'");
- }
- ExpressionContext expressionContext;
- try {
- expressionContext =
RequestContextUtils.getExpression(aggregationConfig.getAggregationFunction());
- } catch (Exception e) {
- throw new IllegalStateException(
- "Invalid aggregation function '" + aggregationFunction + "'
for column '" + columnName + "'", e);
- }
- Preconditions.checkState(expressionContext.getType() ==
ExpressionContext.Type.FUNCTION,
- "aggregation function must be a function for: %s",
aggregationConfig);
-
- FunctionContext functionContext = expressionContext.getFunction();
- AggregationFunctionType functionType =
-
AggregationFunctionType.getAggregationFunctionType(functionContext.getFunctionName());
- List<ExpressionContext> arguments = functionContext.getArguments();
- int numArguments = arguments.size();
- if (functionType == DISTINCTCOUNTHLL) {
- Preconditions.checkState(numArguments >= 1 && numArguments <= 2,
- "DISTINCT_COUNT_HLL can have at most two arguments: %s",
aggregationConfig);
- if (numArguments == 2) {
- ExpressionContext secondArgument = arguments.get(1);
- Preconditions.checkState(secondArgument.getType() ==
ExpressionContext.Type.LITERAL,
- "Second argument of DISTINCT_COUNT_HLL must be literal: %s",
aggregationConfig);
- String literal = secondArgument.getLiteral().getStringValue();
- Preconditions.checkState(StringUtils.isNumeric(literal),
- "Second argument of DISTINCT_COUNT_HLL must be a number:
%s", aggregationConfig);
- }
- Preconditions.checkState(dataType == DataType.BYTES, "Result type
for DISTINCT_COUNT_HLL must be BYTES: %s",
- aggregationConfig);
- } else if (functionType == DISTINCTCOUNTHLLPLUS) {
- Preconditions.checkState(numArguments >= 1 && numArguments <= 3,
- "DISTINCT_COUNT_HLL_PLUS can have at most three arguments:
%s", aggregationConfig);
- if (numArguments == 2) {
- ExpressionContext secondArgument = arguments.get(1);
- Preconditions.checkState(secondArgument.getType() ==
ExpressionContext.Type.LITERAL,
- "Second argument of DISTINCT_COUNT_HLL_PLUS must be literal:
%s", aggregationConfig);
- String literal = secondArgument.getLiteral().getStringValue();
- Preconditions.checkState(StringUtils.isNumeric(literal),
- "Second argument of DISTINCT_COUNT_HLL_PLUS must be a
number: %s", aggregationConfig);
- }
- if (numArguments == 3) {
- ExpressionContext thirdArgument = arguments.get(2);
- Preconditions.checkState(thirdArgument.getType() ==
ExpressionContext.Type.LITERAL,
- "Third argument of DISTINCT_COUNT_HLL_PLUS must be literal:
%s", aggregationConfig);
- String literal = thirdArgument.getLiteral().getStringValue();
- Preconditions.checkState(StringUtils.isNumeric(literal),
- "Third argument of DISTINCT_COUNT_HLL_PLUS must be a number:
%s", aggregationConfig);
- }
- Preconditions.checkState(dataType == DataType.BYTES,
- "Result type for DISTINCT_COUNT_HLL_PLUS must be BYTES: %s",
aggregationConfig);
- } else if (functionType == SUMPRECISION) {
- Preconditions.checkState(numArguments >= 2 && numArguments <= 3,
- "SUM_PRECISION must specify precision (required), scale
(optional): %s", aggregationConfig);
- ExpressionContext secondArgument = arguments.get(1);
- Preconditions.checkState(secondArgument.getType() ==
ExpressionContext.Type.LITERAL,
- "Second argument of SUM_PRECISION must be literal: %s",
aggregationConfig);
- String literal = secondArgument.getLiteral().getStringValue();
- Preconditions.checkState(StringUtils.isNumeric(literal),
- "Second argument of SUM_PRECISION must be a number: %s",
aggregationConfig);
- Preconditions.checkState(dataType == DataType.BIG_DECIMAL ||
dataType == DataType.BYTES,
- "Result type for SUM_PRECISION must be BIG_DECIMAL or BYTES:
%s", aggregationConfig);
- } else {
- Preconditions.checkState(numArguments == 1, "%s can only have one
argument: %s", functionType,
- aggregationConfig);
- }
- ExpressionContext firstArgument = arguments.get(0);
- Preconditions.checkState(firstArgument.getType() ==
ExpressionContext.Type.IDENTIFIER,
- "First argument of aggregation function: %s must be identifier,
got: %s", functionType,
- firstArgument.getType());
- // Create a ValueAggregator for the aggregation function and check
if it is supported for ingestion (fixed
- // size aggregated value).
- ValueAggregator<?, ?> valueAggregator;
- try {
- valueAggregator =
- ValueAggregatorFactory.getValueAggregator(functionType,
arguments.subList(1, numArguments));
- } catch (Exception e) {
- throw new IllegalStateException(
- "Caught exception while creating ValueAggregator for
aggregation function: " + aggregationFunction, e);
- }
-
Preconditions.checkState(valueAggregator.isAggregatedValueFixedSize(),
- "Aggregation function: %s must have fixed size aggregated
value", aggregationFunction);
-
- aggregationSourceColumns.add(firstArgument.getIdentifier());
+ // Filter config
+ FilterConfig filterConfig = ingestionConfig.getFilterConfig();
+ if (filterConfig != null) {
+ String filterFunction = filterConfig.getFilterFunction();
+ if (filterFunction != null) {
+ if (_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(filterFunction)) {
+ throw new IllegalStateException(
+ "Groovy filter functions are disabled for table config. Found '"
+ filterFunction + "'");
}
- Preconditions.checkState(new
HashSet<>(schema.getMetricNames()).equals(aggregationColumns),
- "all metric columns must be aggregated");
-
- // This is required by
MutableSegmentImpl.enableMetricsAggregationIfPossible().
- // That code will disable ingestion aggregation if all metrics aren't
noDictionaryColumns.
- // But if you do that after the table is already created, all future
aggregations will
- // just be the default value.
- Map<String, DictionaryIndexConfig> configPerCol =
StandardIndexes.dictionary().getConfig(tableConfig, schema);
- aggregationColumns.forEach(column -> {
- DictionaryIndexConfig dictConfig = configPerCol.get(column);
- Preconditions.checkState(dictConfig != null &&
dictConfig.isDisabled(),
- "Aggregated column: %s must be a no-dictionary column", column);
- });
- }
-
- // Source field configs
- List<SourceFieldConfig> sourceFieldConfigs =
ingestionConfig.getSourceFieldConfigs();
- if (sourceFieldConfigs != null) {
- // A source field can be configured once per phase (pre- and
post-complex-type), but not twice within the same
- // phase, because that would yield two DataTypeTransformers targeting
it in the same phase, which is ambiguous.
- Set<String> preComplexTypeFieldNames = new HashSet<>();
- Set<String> postComplexTypeFieldNames = new HashSet<>();
- for (SourceFieldConfig sourceFieldConfig : sourceFieldConfigs) {
- String name = sourceFieldConfig.getName();
- boolean preComplexTypeTransform =
sourceFieldConfig.isPreComplexTypeTransform();
- Set<String> fieldNames = preComplexTypeTransform ?
preComplexTypeFieldNames : postComplexTypeFieldNames;
- Preconditions.checkState(fieldNames.add(name),
- "Duplicate SourceFieldConfig found for source field: %s
(preComplexTypeTransform: %s)", name,
- preComplexTypeTransform);
+ try {
+ FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction);
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Invalid filter function '" + filterFunction + "', exception: "
+ e.getMessage(), e);
}
}
+ }
- // Enrichment configs
- List<EnrichmentConfig> enrichmentConfigs =
ingestionConfig.getEnrichmentConfigs();
- if (enrichmentConfigs != null) {
- for (EnrichmentConfig enrichmentConfig : enrichmentConfigs) {
- RecordEnricherRegistry.validateEnrichmentConfig(enrichmentConfig,
- new RecordEnricherValidationConfig(_disableGroovy));
+ // Source field configs
+ List<SourceFieldConfig> sourceFieldConfigs =
ingestionConfig.getSourceFieldConfigs();
+ if (sourceFieldConfigs != null) {
+ // A source field can be configured once per phase (pre- and
post-complex-type), but not twice within the same
+ // phase, because that would yield two DataTypeTransformers targeting it
in the same phase, which is ambiguous.
+ Set<String> preComplexTypeFieldNames = new HashSet<>();
+ Set<String> postComplexTypeFieldNames = new HashSet<>();
+ for (SourceFieldConfig sourceFieldConfig : sourceFieldConfigs) {
+ String name = sourceFieldConfig.getName();
+ boolean preComplexTypeTransform =
sourceFieldConfig.isPreComplexTypeTransform();
+ Set<String> fieldNames = preComplexTypeTransform ?
preComplexTypeFieldNames : postComplexTypeFieldNames;
+ Preconditions.checkState(fieldNames.add(name),
+ "Duplicate SourceFieldConfig found for source field: %s
(preComplexTypeTransform: %s)", name,
+ preComplexTypeTransform);
+ }
+ }
+
+ // Enrichment configs
+ List<EnrichmentConfig> enrichmentConfigs =
ingestionConfig.getEnrichmentConfigs();
+ if (enrichmentConfigs != null) {
+ for (EnrichmentConfig enrichmentConfig : enrichmentConfigs) {
+ RecordEnricherRegistry.validateEnrichmentConfig(enrichmentConfig,
+ new RecordEnricherValidationConfig(_disableGroovy));
+ }
+ }
+
+ // Transform configs
+ List<TransformConfig> transformConfigs =
ingestionConfig.getTransformConfigs();
+ if (transformConfigs != null) {
+ // Pre-pass: collect every column referenced as a transform-function
argument. A transform whose destination
+ // is not in the schema is still valid when another transform consumes
it as an input - i.e. it is an
+ // intermediate ("derived") column. This enables chained / parse-once
transforms, e.g.
+ // message_obj = jsonExtractObject(message) //
intermediate, not in the schema
+ // level = JSONPATHSTRING(message_obj, '$.level') // consumes
the intermediate
+ // The intermediate is materialized in the record during transformation
and dropped before indexing (only
+ // schema columns are indexed). Unreferenced non-schema destinations
still fail below (typo protection).
+ Set<String> transformInputColumns = new HashSet<>();
+ for (TransformConfig transformConfig : transformConfigs) {
+ String transformFunction = transformConfig.getTransformFunction();
+ // Skip Groovy expressions when Groovy is disabled: do not compile
them just to collect arguments (the main
+ // loop below rejects Groovy without compiling). Such a config is
rejected anyway, so these columns are not
+ // needed as valid intermediate targets.
+ if (transformFunction != null && !(_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(
+ transformFunction))) {
+ try {
+ transformInputColumns.addAll(
+
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction).getArguments());
+ } catch (Exception ignore) {
+ // Invalid functions are reported with a descriptive error in the
main loop below.
+ }
}
}
-
- // Transform configs
- List<TransformConfig> transformConfigs =
ingestionConfig.getTransformConfigs();
- if (transformConfigs != null) {
- // Pre-pass: collect every column referenced as a transform-function
argument. A transform whose destination
- // is not in the schema is still valid when another transform consumes
it as an input - i.e. it is an
- // intermediate ("derived") column. This enables chained / parse-once
transforms, e.g.
- // message_obj = jsonExtractObject(message) //
intermediate, not in the schema
- // level = JSONPATHSTRING(message_obj, '$.level') // consumes
the intermediate
- // The intermediate is materialized in the record during
transformation and dropped before indexing (only
- // schema columns are indexed). Unreferenced non-schema destinations
still fail below (typo protection).
- Set<String> transformInputColumns = new HashSet<>();
- for (TransformConfig transformConfig : transformConfigs) {
- String transformFunction = transformConfig.getTransformFunction();
- // Skip Groovy expressions when Groovy is disabled: do not compile
them just to collect arguments (the main
- // loop below rejects Groovy without compiling). Such a config is
rejected anyway, so these columns are not
- // needed as valid intermediate targets.
- if (transformFunction != null
- && !(_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(transformFunction))) {
- try {
- transformInputColumns.addAll(
-
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction).getArguments());
- } catch (Exception ignore) {
- // Invalid functions are reported with a descriptive error in
the main loop below.
- }
- }
+ Set<String> transformColumns = new HashSet<>();
+ for (TransformConfig transformConfig : transformConfigs) {
+ String columnName = transformConfig.getColumnName();
+ String transformFunction = transformConfig.getTransformFunction();
+ if (columnName == null || transformFunction == null) {
+ throw new IllegalStateException(
+ "columnName/transformFunction cannot be null in TransformConfig
" + transformConfig);
}
- Set<String> transformColumns = new HashSet<>();
- for (TransformConfig transformConfig : transformConfigs) {
- String columnName = transformConfig.getColumnName();
- String transformFunction = transformConfig.getTransformFunction();
- if (columnName == null || transformFunction == null) {
- throw new IllegalStateException(
- "columnName/transformFunction cannot be null in
TransformConfig " + transformConfig);
- }
- if (!transformColumns.add(columnName)) {
- throw new IllegalStateException("Duplicate transform config found
for column '" + columnName + "'");
- }
- Preconditions.checkState(
- schema.hasColumn(columnName) ||
aggregationSourceColumns.contains(columnName)
- || transformInputColumns.contains(columnName),
- "The destination column '" + columnName
- + "' of the transform function must be present in the
schema, be consumed as the input of another "
- + "transform function, or be a source column for
aggregations");
- FunctionEvaluator expressionEvaluator;
- if (_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
- throw new IllegalStateException(
- "Groovy transform functions are disabled for table config.
Found '" + transformFunction
- + "' for column '" + columnName + "'");
- }
- try {
- expressionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
- } catch (Exception e) {
- throw new IllegalStateException(
- "Invalid transform function '" + transformFunction + "' for
column '" + columnName + "', exception: "
- + e.getMessage(), e);
- }
- List<String> arguments = expressionEvaluator.getArguments();
- if (arguments.contains(columnName)) {
- throw new IllegalStateException(
- "Arguments of a transform function '" + arguments + "' cannot
contain the destination column '"
- + columnName + "'");
- }
+ if (!transformColumns.add(columnName)) {
+ throw new IllegalStateException("Duplicate transform config found
for column '" + columnName + "'");
+ }
+ Preconditions.checkState(schema.hasColumn(columnName) ||
aggregationSourceColumns.contains(columnName)
+ || transformInputColumns.contains(columnName), "The destination
column '" + columnName
+ + "' of the transform function must be present in the schema, be
consumed as the input of another "
+ + "transform function, or be a source column for aggregations");
+ FunctionEvaluator expressionEvaluator;
+ if (_disableGroovy &&
FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
+ throw new IllegalStateException(
+ "Groovy transform functions are disabled for table config. Found
'" + transformFunction + "' for column '"
+ + columnName + "'");
+ }
+ try {
+ expressionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Invalid transform function '" + transformFunction + "' for
column '" + columnName + "', exception: "
+ + e.getMessage(), e);
+ }
+ List<String> arguments = expressionEvaluator.getArguments();
+ if (arguments.contains(columnName)) {
+ throw new IllegalStateException(
+ "Arguments of a transform function '" + arguments + "' cannot
contain the destination column '"
+ + columnName + "'");
}
}
+ }
- // Complex configs
- ComplexTypeConfig complexTypeConfig =
ingestionConfig.getComplexTypeConfig();
- if (complexTypeConfig != null) {
- Map<String, String> prefixesToRename =
complexTypeConfig.getPrefixesToRename();
- if (MapUtils.isNotEmpty(prefixesToRename)) {
- Set<String> fieldNames = schema.getColumnNames();
- for (String prefix : prefixesToRename.keySet()) {
- for (String field : fieldNames) {
- Preconditions.checkState(!field.startsWith(prefix),
- "Fields in the schema may not begin with any prefix
specified in the prefixesToRename"
- + " config. Name conflict with field: " + field + " and
prefix: " + prefix);
- }
+ // Complex configs
+ ComplexTypeConfig complexTypeConfig =
ingestionConfig.getComplexTypeConfig();
+ if (complexTypeConfig != null) {
+ Map<String, String> prefixesToRename =
complexTypeConfig.getPrefixesToRename();
+ if (MapUtils.isNotEmpty(prefixesToRename)) {
+ Set<String> fieldNames = schema.getColumnNames();
+ for (String prefix : prefixesToRename.keySet()) {
+ for (String field : fieldNames) {
+ Preconditions.checkState(!field.startsWith(prefix),
+ "Fields in the schema may not begin with any prefix specified
in the prefixesToRename"
+ + " config. Name conflict with field: " + field + " and
prefix: " + prefix);
}
}
}
+ }
+
+ SchemaConformingTransformerConfig schemaConformingTransformerConfig =
+ ingestionConfig.getSchemaConformingTransformerConfig();
+ if (schemaConformingTransformerConfig != null) {
+ SchemaConformingTransformer.validateSchema(schema,
schemaConformingTransformerConfig);
+ }
+ }
- SchemaConformingTransformerConfig schemaConformingTransformerConfig =
- ingestionConfig.getSchemaConformingTransformerConfig();
- if (schemaConformingTransformerConfig != null) {
- SchemaConformingTransformer.validateSchema(schema,
schemaConformingTransformerConfig);
+ /// Validates all metrics-aggregation configuration for both mechanisms (the
`aggregateMetrics` flag and ingestion
+ /// `aggregationConfigs`) in one place, and returns the set of source
columns referenced by the aggregation functions
+ /// (empty when aggregation is disabled). Consuming-segment rollup keys each
row on the dictionary ids of the
+ /// dimension and time columns and mutates the aggregated value in place in
the metric's raw forward index, so:
+ /// - The two mechanisms are mutually exclusive, and aggregation is
incompatible with upsert / dedup.
+ /// - The schema must not contain a COMPLEX column (neither a valid key nor
an aggregatable metric).
+ /// - Every dimension column must be single-valued; the rollup key holds a
single dictionary id per dimension, so it
+ /// cannot represent a row with multiple values for that column (issue
#3867). No-dictionary dimensions are allowed
+ /// — the consuming segment force-creates a dictionary for them.
+ /// - Every metric column must be single-valued and no-dictionary
(aggregated values are mutated in place; a
+ /// dictionary-encoded metric would silently disable aggregation once the
table is created).
+ /// - For `aggregationConfigs`, every metric must be covered by a valid
aggregation function whose aggregated value
+ /// is fixed size.
+ private static Set<String> validateMetricsAggregation(TableConfig
tableConfig, Schema schema) {
+ IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
+ boolean aggregateMetrics =
tableConfig.getIndexingConfig().isAggregateMetrics();
+ List<AggregationConfig> aggregationConfigs =
+ ingestionConfig != null ? ingestionConfig.getAggregationConfigs() :
null;
+ boolean hasAggregationConfigs =
CollectionUtils.isNotEmpty(aggregationConfigs);
+
+ Preconditions.checkState(!(aggregateMetrics && hasAggregationConfigs),
+ "Metrics aggregation cannot be enabled in the Indexing Config and
Ingestion Config at the same time");
+
+ if (!aggregateMetrics && !hasAggregationConfigs) {
+ return Set.of();
+ }
+
+ Preconditions.checkState(!tableConfig.isUpsertEnabled(),
+ "Metrics aggregation and upsert cannot be enabled together");
+ Preconditions.checkState(!tableConfig.isDedupEnabled(), "Metrics
aggregation and dedup cannot be enabled together");
+
+
Preconditions.checkState(CollectionUtils.isEmpty(schema.getComplexFieldSpecs()),
+ "Metrics aggregation cannot be enabled when the schema contains
COMPLEX columns");
+
+ for (String dimension : schema.getDimensionNames()) {
+
Preconditions.checkState(schema.getFieldSpecFor(dimension).isSingleValueField(),
+ "Metrics aggregation cannot be enabled with multi-value dimension
column: %s", dimension);
+ }
+
+ Map<String, DictionaryIndexConfig> dictConfigByCol =
StandardIndexes.dictionary().getConfig(tableConfig, schema);
+ for (String metric : schema.getMetricNames()) {
+
Preconditions.checkState(schema.getFieldSpecFor(metric).isSingleValueField(),
+ "Metrics aggregation cannot be enabled with multi-value metric
column: %s", metric);
+ DictionaryIndexConfig dictConfig = dictConfigByCol.get(metric);
+ Preconditions.checkState(dictConfig != null && dictConfig.isDisabled(),
+ "Metric column: %s must be a no-dictionary column when metrics
aggregation is enabled", metric);
+ }
+
+ // The aggregateMetrics flag implicitly SUMs every metric; there are no
per-config functions to validate.
+ if (!hasAggregationConfigs) {
+ return Set.of();
+ }
+
+ Set<String> aggregationSourceColumns = new HashSet<>();
+ Set<String> aggregationColumns = new HashSet<>();
+ for (AggregationConfig aggregationConfig : aggregationConfigs) {
+ String columnName = aggregationConfig.getColumnName();
+ String aggregationFunction = aggregationConfig.getAggregationFunction();
+ if (columnName == null || aggregationFunction == null) {
+ throw new IllegalStateException(
+ "columnName/aggregationFunction cannot be null in
AggregationConfig " + aggregationConfig);
}
+
+ FieldSpec fieldSpec = schema.getFieldSpecFor(columnName);
+ Preconditions.checkState(fieldSpec != null,
+ "The destination column '" + columnName + "' of the aggregation
function must be present in the schema");
+ Preconditions.checkState(fieldSpec.getFieldType() ==
FieldSpec.FieldType.METRIC,
+ "The destination column '" + columnName + "' of the aggregation
function must be a metric column");
+ DataType dataType = fieldSpec.getDataType();
+
+ if (!aggregationColumns.add(columnName)) {
+ throw new IllegalStateException("Duplicate aggregation config found
for column '" + columnName + "'");
+ }
+ ExpressionContext expressionContext;
+ try {
+ expressionContext =
RequestContextUtils.getExpression(aggregationConfig.getAggregationFunction());
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Invalid aggregation function '" + aggregationFunction + "' for
column '" + columnName + "'", e);
+ }
+ Preconditions.checkState(expressionContext.getType() ==
ExpressionContext.Type.FUNCTION,
+ "aggregation function must be a function for: %s",
aggregationConfig);
+
+ FunctionContext functionContext = expressionContext.getFunction();
+ AggregationFunctionType functionType =
+
AggregationFunctionType.getAggregationFunctionType(functionContext.getFunctionName());
+ List<ExpressionContext> arguments = functionContext.getArguments();
+ int numArguments = arguments.size();
+ if (functionType == DISTINCTCOUNTHLL) {
+ Preconditions.checkState(numArguments >= 1 && numArguments <= 2,
+ "DISTINCT_COUNT_HLL can have at most two arguments: %s",
aggregationConfig);
+ if (numArguments == 2) {
+ ExpressionContext secondArgument = arguments.get(1);
+ Preconditions.checkState(secondArgument.getType() ==
ExpressionContext.Type.LITERAL,
+ "Second argument of DISTINCT_COUNT_HLL must be literal: %s",
aggregationConfig);
+ String literal = secondArgument.getLiteral().getStringValue();
+ Preconditions.checkState(StringUtils.isNumeric(literal),
+ "Second argument of DISTINCT_COUNT_HLL must be a number: %s",
aggregationConfig);
+ }
+ Preconditions.checkState(dataType == DataType.BYTES, "Result type for
DISTINCT_COUNT_HLL must be BYTES: %s",
+ aggregationConfig);
+ } else if (functionType == DISTINCTCOUNTHLLPLUS) {
+ Preconditions.checkState(numArguments >= 1 && numArguments <= 3,
+ "DISTINCT_COUNT_HLL_PLUS can have at most three arguments: %s",
aggregationConfig);
+ if (numArguments == 2) {
+ ExpressionContext secondArgument = arguments.get(1);
+ Preconditions.checkState(secondArgument.getType() ==
ExpressionContext.Type.LITERAL,
+ "Second argument of DISTINCT_COUNT_HLL_PLUS must be literal:
%s", aggregationConfig);
+ String literal = secondArgument.getLiteral().getStringValue();
+ Preconditions.checkState(StringUtils.isNumeric(literal),
+ "Second argument of DISTINCT_COUNT_HLL_PLUS must be a number:
%s", aggregationConfig);
+ }
+ if (numArguments == 3) {
+ ExpressionContext thirdArgument = arguments.get(2);
+ Preconditions.checkState(thirdArgument.getType() ==
ExpressionContext.Type.LITERAL,
+ "Third argument of DISTINCT_COUNT_HLL_PLUS must be literal: %s",
aggregationConfig);
+ String literal = thirdArgument.getLiteral().getStringValue();
+ Preconditions.checkState(StringUtils.isNumeric(literal),
+ "Third argument of DISTINCT_COUNT_HLL_PLUS must be a number:
%s", aggregationConfig);
+ }
+ Preconditions.checkState(dataType == DataType.BYTES,
+ "Result type for DISTINCT_COUNT_HLL_PLUS must be BYTES: %s",
aggregationConfig);
+ } else if (functionType == SUMPRECISION) {
+ Preconditions.checkState(numArguments >= 2 && numArguments <= 3,
+ "SUM_PRECISION must specify precision (required), scale
(optional): %s", aggregationConfig);
+ ExpressionContext secondArgument = arguments.get(1);
+ Preconditions.checkState(secondArgument.getType() ==
ExpressionContext.Type.LITERAL,
+ "Second argument of SUM_PRECISION must be literal: %s",
aggregationConfig);
+ String literal = secondArgument.getLiteral().getStringValue();
+ Preconditions.checkState(StringUtils.isNumeric(literal),
+ "Second argument of SUM_PRECISION must be a number: %s",
aggregationConfig);
+ Preconditions.checkState(dataType == DataType.BIG_DECIMAL || dataType
== DataType.BYTES,
+ "Result type for SUM_PRECISION must be BIG_DECIMAL or BYTES: %s",
aggregationConfig);
+ } else {
+ Preconditions.checkState(numArguments == 1, "%s can only have one
argument: %s", functionType,
+ aggregationConfig);
+ }
+ ExpressionContext firstArgument = arguments.get(0);
+ Preconditions.checkState(firstArgument.getType() ==
ExpressionContext.Type.IDENTIFIER,
+ "First argument of aggregation function: %s must be identifier, got:
%s", functionType,
+ firstArgument.getType());
+ // Create a ValueAggregator for the aggregation function and check if it
is supported for ingestion (fixed
+ // size aggregated value).
+ ValueAggregator<?, ?> valueAggregator;
+ try {
+ valueAggregator =
ValueAggregatorFactory.getValueAggregator(functionType, arguments.subList(1,
numArguments));
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Caught exception while creating ValueAggregator for aggregation
function: " + aggregationFunction, e);
+ }
+ Preconditions.checkState(valueAggregator.isAggregatedValueFixedSize(),
+ "Aggregation function: %s must have fixed size aggregated value",
aggregationFunction);
+
+ aggregationSourceColumns.add(firstArgument.getIdentifier());
}
+ Preconditions.checkState(new
HashSet<>(schema.getMetricNames()).equals(aggregationColumns),
+ "all metric columns must be aggregated");
+ return aggregationSourceColumns;
}
private static void validateStreamConfigMaps(TableConfig tableConfig) {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplAggregateMetricsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplAggregateMetricsTest.java
index 10cfddcd6cc..6282a960c75 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplAggregateMetricsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplAggregateMetricsTest.java
@@ -90,6 +90,73 @@ public class MutableSegmentImplAggregateMetricsTest {
mutableSegmentImpl.destroy();
}
+ @Test
+ public void testAggregateMetricsWithNoDictionaryKeyColumns()
+ throws Exception {
+ // The dimension and time columns are marked no-dictionary in the table
config. Aggregation keys each row on the
+ // dictionary ids of these columns, so the consuming segment must still
create dictionaries for them (the committed
+ // segment is rebuilt from the table config and honors the no-dictionary
setting). Metrics stay no-dictionary.
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("testSchema")
+ .addSingleValueDimension(DIMENSION_1, FieldSpec.DataType.INT)
+ .addSingleValueDimension(DIMENSION_2, FieldSpec.DataType.STRING)
+ .addMetric(METRIC, FieldSpec.DataType.LONG)
+ .addMetric(METRIC_2, FieldSpec.DataType.FLOAT)
+ .addDateTime(TIME_COLUMN1, FieldSpec.DataType.INT, "1:DAYS:EPOCH",
"1:DAYS")
+ .addDateTime(TIME_COLUMN2, FieldSpec.DataType.INT, "1:HOURS:EPOCH",
"1:HOURS")
+ .build();
+ Set<String> noDictionaryColumns = Set.of(DIMENSION_1, DIMENSION_2,
TIME_COLUMN1, TIME_COLUMN2, METRIC, METRIC_2);
+ MutableSegmentImpl mutableSegmentImpl =
+ MutableSegmentImplTestUtils.createMutableSegmentImpl(schema,
noDictionaryColumns, Set.of(), Set.of(), true);
+
+ testAggregateMetrics(mutableSegmentImpl);
+
+ // Key columns must have a dictionary even though they are configured as
no-dictionary.
+
Assert.assertNotNull(mutableSegmentImpl.getDataSourceNullable(DIMENSION_1).getDictionary());
+
Assert.assertNotNull(mutableSegmentImpl.getDataSourceNullable(DIMENSION_2).getDictionary());
+
Assert.assertNotNull(mutableSegmentImpl.getDataSourceNullable(TIME_COLUMN1).getDictionary());
+
Assert.assertNotNull(mutableSegmentImpl.getDataSourceNullable(TIME_COLUMN2).getDictionary());
+ // Metrics must stay no-dictionary so their values can be aggregated in
place.
+
Assert.assertNull(mutableSegmentImpl.getDataSourceNullable(METRIC).getDictionary());
+
Assert.assertNull(mutableSegmentImpl.getDataSourceNullable(METRIC_2).getDictionary());
+
+ mutableSegmentImpl.destroy();
+ }
+
+ @Test
+ public void testMultiValueDimensionDisablesAggregation()
+ throws Exception {
+ // A multi-value dimension cannot be used as an aggregation key (issue
#3867), so aggregation stays disabled and no
+ // rollup happens even though aggregateMetrics is set. This guards against
over-broadening the aggregation-enabling
+ // and dictionary-forcing logic to columns that cannot support it.
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("testSchema")
+ .addSingleValueDimension(DIMENSION_1, FieldSpec.DataType.INT)
+ .addMultiValueDimension(DIMENSION_2, FieldSpec.DataType.STRING)
+ .addMetric(METRIC, FieldSpec.DataType.LONG)
+ .addMetric(METRIC_2, FieldSpec.DataType.FLOAT)
+ .addDateTime(TIME_COLUMN1, FieldSpec.DataType.INT, "1:DAYS:EPOCH",
"1:DAYS")
+ .addDateTime(TIME_COLUMN2, FieldSpec.DataType.INT, "1:HOURS:EPOCH",
"1:HOURS")
+ .build();
+ MutableSegmentImpl mutableSegmentImpl =
+ MutableSegmentImplTestUtils.createMutableSegmentImpl(schema,
Set.of(METRIC, METRIC_2), Set.of(), Set.of(),
+ true);
+
+ Random random = new Random();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue(DIMENSION_1, random.nextInt(10));
+ row.putValue(DIMENSION_2, new Object[]{"a", "b"});
+ row.putValue(TIME_COLUMN1, random.nextInt(5));
+ row.putValue(TIME_COLUMN2, random.nextInt(10));
+ row.putValue(METRIC, (long) random.nextInt());
+ row.putValue(METRIC_2, random.nextFloat());
+ mutableSegmentImpl.index(row, METADATA);
+ }
+
+ // Aggregation is disabled, so every row is stored as its own document (no
rollup).
+ Assert.assertEquals(mutableSegmentImpl.getNumDocsIndexed(), NUM_ROWS);
+ mutableSegmentImpl.destroy();
+ }
+
private void testAggregateMetrics(MutableSegmentImpl mutableSegmentImpl)
throws Exception {
String[] stringValues = new String[10];
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 1a8abf60d59..11aefa5d3c2 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -635,6 +635,12 @@ public class TableConfigUtilsTest {
}
schema.addField(new MetricFieldSpec("m1", FieldSpec.DataType.DOUBLE));
+ // Mark the metric as no-dictionary up front (a requirement validated by
validateMetricsAggregation, covered in
+ // metricsAggregationValidationTest) so the steps below exercise the
per-aggregation-config function validation.
+ IndexingConfig indexingConfig = new IndexingConfig();
+ indexingConfig.setNoDictionaryColumns(List.of("m1"));
+ tableConfig.setIndexingConfig(indexingConfig);
+
ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig(null,
null)));
try {
TableConfigUtils.validateIngestionConfig(tableConfig, schema);
@@ -668,27 +674,6 @@ public class TableConfigUtilsTest {
// expected
}
- ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1",
"SUM(m1)")));
- try {
- TableConfigUtils.validateIngestionConfig(tableConfig, schema);
- fail("Should fail due to noDictionaryColumns being null");
- } catch (IllegalStateException e) {
- // expected
- }
-
- IndexingConfig indexingConfig = new IndexingConfig();
- indexingConfig.setNoDictionaryColumns(List.of());
- tableConfig.setIndexingConfig(indexingConfig);
-
- try {
- TableConfigUtils.validateIngestionConfig(tableConfig, schema);
- fail("Should fail due to noDictionaryColumns not containing m1");
- } catch (IllegalStateException e) {
- // expected
- }
-
- indexingConfig.setNoDictionaryColumns(List.of("m1"));
-
ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1",
"SUM(m1)")));
TableConfigUtils.validateIngestionConfig(tableConfig, schema);
@@ -696,6 +681,7 @@ public class TableConfigUtilsTest {
TableConfigUtils.validateIngestionConfig(tableConfig, schema);
schema.addField(new MetricFieldSpec("m2", FieldSpec.DataType.DOUBLE));
+ indexingConfig.setNoDictionaryColumns(List.of("m1", "m2"));
try {
TableConfigUtils.validateIngestionConfig(tableConfig, schema);
fail("Should fail due to one metric column not being aggregated");
@@ -842,6 +828,104 @@ public class TableConfigUtilsTest {
}
}
+ @Test
+ public void metricsAggregationValidationTest() {
+ // A COMPLEX column is neither a valid aggregation key nor an aggregatable
metric, so metrics aggregation (via
+ // either the aggregateMetrics flag or ingestion aggregationConfigs)
cannot be enabled when the schema contains one.
+ Schema schemaWithComplex = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addMetric("m1", FieldSpec.DataType.LONG)
+ .addComplex("complexCol", FieldSpec.DataType.MAP, Map.of())
+ .build();
+
+ TableConfig aggregateMetricsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setAggregateMetrics(true)
+ .setNoDictionaryColumns(List.of("m1"))
+ .build();
+ assertMetricsAggregationValidationFails(aggregateMetricsConfig,
schemaWithComplex, "COMPLEX columns");
+
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setAggregationConfigs(List.of(new AggregationConfig("m1",
"SUM(s1)")));
+ TableConfig aggregationConfigsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setIngestionConfig(ingestionConfig)
+ .setNoDictionaryColumns(List.of("m1"))
+ .build();
+ assertMetricsAggregationValidationFails(aggregationConfigsConfig,
schemaWithComplex, "COMPLEX columns");
+
+ // A multi-value dimension cannot be used as an aggregation key.
+ Schema schemaWithMvDimension = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addMultiValueDimension("mvDim", FieldSpec.DataType.INT)
+ .addMetric("m1", FieldSpec.DataType.LONG)
+ .build();
+ assertMetricsAggregationValidationFails(aggregationConfigsConfig,
schemaWithMvDimension,
+ "multi-value dimension column");
+
+ // Metric columns must be no-dictionary, for both mechanisms.
+ Schema schemaWithSvColumns = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addDateTime("timeColumn", FieldSpec.DataType.TIMESTAMP,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addSingleValueDimension("d1", FieldSpec.DataType.INT)
+ .addMetric("m1", FieldSpec.DataType.LONG)
+ .build();
+ TableConfig dictMetricAggregateMetricsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setAggregateMetrics(true)
+ .build();
+ assertMetricsAggregationValidationFails(dictMetricAggregateMetricsConfig,
schemaWithSvColumns,
+ "must be a no-dictionary column when metrics aggregation is enabled");
+
+ IngestionConfig dictMetricIngestionConfig = new IngestionConfig();
+ dictMetricIngestionConfig.setAggregationConfigs(List.of(new
AggregationConfig("m1", "SUM(s1)")));
+ TableConfig dictMetricAggregationConfigsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setIngestionConfig(dictMetricIngestionConfig)
+ .build();
+
assertMetricsAggregationValidationFails(dictMetricAggregationConfigsConfig,
schemaWithSvColumns,
+ "must be a no-dictionary column when metrics aggregation is enabled");
+
+ // Metrics aggregation is incompatible with dedup, for both mechanisms.
+ TableConfig dedupAggregateMetricsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setAggregateMetrics(true)
+ .setNoDictionaryColumns(List.of("m1"))
+ .setDedupConfig(new DedupConfig())
+ .build();
+ assertMetricsAggregationValidationFails(dedupAggregateMetricsConfig,
schemaWithSvColumns,
+ "Metrics aggregation and dedup cannot be enabled together");
+
+ IngestionConfig dedupIngestionConfig = new IngestionConfig();
+ dedupIngestionConfig.setAggregationConfigs(List.of(new
AggregationConfig("m1", "SUM(s1)")));
+ TableConfig dedupAggregationConfigsConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setIngestionConfig(dedupIngestionConfig)
+ .setNoDictionaryColumns(List.of("m1"))
+ .setDedupConfig(new DedupConfig())
+ .build();
+ assertMetricsAggregationValidationFails(dedupAggregationConfigsConfig,
schemaWithSvColumns,
+ "Metrics aggregation and dedup cannot be enabled together");
+
+ // Valid: single-value dimensions, a no-dictionary metric, and no COMPLEX
column.
+ TableConfig validConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setAggregateMetrics(true)
+ .setNoDictionaryColumns(List.of("m1"))
+ .build();
+ TableConfigUtils.validateIngestionConfig(validConfig, schemaWithSvColumns);
+ }
+
+ private void assertMetricsAggregationValidationFails(TableConfig
tableConfig, Schema schema,
+ String expectedMessageSubstring) {
+ try {
+ TableConfigUtils.validateIngestionConfig(tableConfig, schema);
+ fail("Should fail with message containing: " + expectedMessageSubstring);
+ } catch (IllegalStateException e) {
+ assertTrue(e.getMessage().contains(expectedMessageSubstring),
+ "Expected message containing '" + expectedMessageSubstring + "' but
got: " + e.getMessage());
+ }
+ }
+
@Test
public void ingestionStreamConfigsTest() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
