Copilot commented on code in PR #19087:
URL: https://github.com/apache/pinot/pull/19087#discussion_r3762499762
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,26 +152,142 @@ private static void addIfNotNoOp(List<RecordTransformer>
transformers, @Nullable
}
}
- private static void addSourceFieldDataTypeTransformer(TableConfig
tableConfig, List<RecordTransformer> transformers,
- boolean preComplexTypeTransform) {
+ private static void addSourceFieldDataTypeTransformer(TableConfig
tableConfig, @Nullable Schema schema,
+ List<RecordTransformer> transformers, boolean preComplexTypeTransform) {
+ Map<String, PinotDataType> dataTypes = new HashMap<>();
+ IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
+ if (ingestionConfig != null) {
+ List<SourceFieldConfig> sourceFieldConfigs =
ingestionConfig.getSourceFieldConfigs();
+ if (CollectionUtils.isNotEmpty(sourceFieldConfigs)) {
+ for (SourceFieldConfig sourceFieldConfig : sourceFieldConfigs) {
+ // If pre-ComplexType transformers are requested, add only
pre-ComplexType source fields. Similarly, if
+ // non pre-ComplexType transformers are requested, add only non
pre-ComplexType source fields.
+ if (sourceFieldConfig.isPreComplexTypeTransform() ==
preComplexTypeTransform) {
+ dataTypes.put(sourceFieldConfig.getName(),
sourceFieldConfig.getDataType());
+ }
+ }
+ }
+ }
+ // Auto-register aggregation source columns not in the schema so mistyped
JSON/Avro string numbers are converted
+ // before MutableSegmentImpl indexes them. Explicit SourceFieldConfig wins
(already in dataTypes). Only runs in the
+ // post-complex-type phase so flattened/unnested fields are available.
Auto-derived columns are converted lazily
+ // (only when the incoming value is incompatible with the aggregators) to
avoid per-record conversion overhead on
+ // the common correctly-typed path.
+ Set<String> lazyColumns = new HashSet<>();
+ if (!preComplexTypeTransform && schema != null) {
+ addAggregationSourceDataTypes(tableConfig, schema, dataTypes,
lazyColumns);
+ }
+ if (!dataTypes.isEmpty()) {
+ transformers.add(new DataTypeTransformer(tableConfig, dataTypes,
lazyColumns));
+ }
+ }
+
+ /// Derives [PinotDataType]s for ingestion-aggregation source columns that
are absent from the schema (and not already
+ /// covered by an explicit [SourceFieldConfig]). Types are inferred from the
aggregation function and destination
+ /// metric, and recorded in `lazyColumns` so that [DataTypeTransformer] only
converts values the aggregators cannot
+ /// consume directly. Sketch/HLL/COUNT sources are left unconverted so
offering semantics (e.g. hashing a string vs a
+ /// number) are preserved.
[org.apache.pinot.segment.local.aggregator.ValueAggregatorUtils#toDouble]
remains a safety
+ /// net.
+ static void addAggregationSourceDataTypes(TableConfig tableConfig, Schema
schema,
+ Map<String, PinotDataType> dataTypes, Set<String> lazyColumns) {
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
if (ingestionConfig == null) {
return;
}
- List<SourceFieldConfig> sourceFieldConfigs =
ingestionConfig.getSourceFieldConfigs();
- if (CollectionUtils.isEmpty(sourceFieldConfigs)) {
+ List<AggregationConfig> aggregationConfigs =
ingestionConfig.getAggregationConfigs();
+ if (CollectionUtils.isEmpty(aggregationConfigs)) {
return;
}
- Map<String, PinotDataType> dataTypes = new HashMap<>();
- for (SourceFieldConfig sourceFieldConfig : sourceFieldConfigs) {
- // If pre-ComplexType transformers are requested, add only
pre-ComplexType source fields. Similarly, if
- // non pre-ComplexType transformers are requested, add only non
pre-ComplexType source fields.
- if (sourceFieldConfig.isPreComplexTypeTransform() ==
preComplexTypeTransform) {
- dataTypes.put(sourceFieldConfig.getName(),
sourceFieldConfig.getDataType());
+ for (AggregationConfig aggregationConfig : aggregationConfigs) {
+ String destColumn = aggregationConfig.getColumnName();
+ String aggregationFunction = aggregationConfig.getAggregationFunction();
+ if (destColumn == null || aggregationFunction == null) {
+ continue;
+ }
+ ExpressionContext expressionContext;
+ try {
+ expressionContext =
RequestContextUtils.getExpression(aggregationFunction);
+ } catch (Exception e) {
+ // Invalid configs are rejected at table-create validation time; skip
here to keep transformer build resilient.
+ continue;
+ }
+ if (expressionContext.getType() != ExpressionContext.Type.FUNCTION) {
+ continue;
+ }
+ FunctionContext functionContext = expressionContext.getFunction();
+ AggregationFunctionType functionType;
+ try {
+ functionType =
AggregationFunctionType.getAggregationFunctionType(functionContext.getFunctionName());
+ } catch (Exception e) {
+ continue;
+ }
+ List<ExpressionContext> arguments = functionContext.getArguments();
+ if (arguments.isEmpty()) {
+ continue;
+ }
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() != ExpressionContext.Type.IDENTIFIER) {
+ continue;
+ }
+ String sourceColumn = firstArgument.getIdentifier();
+ if (AggregationFunctionColumnPair.STAR.equals(sourceColumn) ||
schema.hasColumn(sourceColumn)
+ || dataTypes.containsKey(sourceColumn)) {
Review Comment:
An explicit `SourceFieldConfig` with `preComplexTypeTransform=true` is not
present in this post-phase `dataTypes` map, so the same source is
auto-registered again and its explicitly selected type can be overwritten after
complex transformation. Check all configured source fields here, not only those
selected for this phase, so the documented “explicit config wins” rule holds.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformer.java:
##########
@@ -144,6 +161,17 @@ public void transform(GenericRow record) {
}
}
+ /// Returns true when a lazily-converted aggregation source value can be
consumed directly by the
+ /// [org.apache.pinot.segment.local.aggregator.ValueAggregator]s without
conversion: null, any [Number] box
+ /// (`ValueAggregatorUtils.toDouble` and `SumPrecisionValueAggregator`
accept them), or a multi-value array that is
+ /// already of the target array type.
+ private static boolean isAggregatorCompatible(@Nullable Object value,
PinotDataType targetType) {
+ if (value == null || value instanceof Number) {
+ return true;
+ }
+ return targetType == PinotDataType.DOUBLE_ARRAY && value instanceof
Double[];
Review Comment:
This compatibility check excludes input forms that the existing aggregators
deliberately consume. For example,
`AvgValueAggregator`/`MinMaxRangeValueAggregator` merge serialized `byte[]`,
while SUM/MIN/MAX/AVG and related aggregators accept multi-element `Object[]`;
these now enter scalar conversion and fail (`BYTES`→`DOUBLE` or array→single
value) before indexing. Preserve aggregator-specific accepted forms or make
conversion shape-aware.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,26 +152,142 @@ private static void addIfNotNoOp(List<RecordTransformer>
transformers, @Nullable
}
}
- private static void addSourceFieldDataTypeTransformer(TableConfig
tableConfig, List<RecordTransformer> transformers,
- boolean preComplexTypeTransform) {
+ private static void addSourceFieldDataTypeTransformer(TableConfig
tableConfig, @Nullable Schema schema,
+ List<RecordTransformer> transformers, boolean preComplexTypeTransform) {
+ Map<String, PinotDataType> dataTypes = new HashMap<>();
+ IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
+ if (ingestionConfig != null) {
+ List<SourceFieldConfig> sourceFieldConfigs =
ingestionConfig.getSourceFieldConfigs();
+ if (CollectionUtils.isNotEmpty(sourceFieldConfigs)) {
+ for (SourceFieldConfig sourceFieldConfig : sourceFieldConfigs) {
+ // If pre-ComplexType transformers are requested, add only
pre-ComplexType source fields. Similarly, if
+ // non pre-ComplexType transformers are requested, add only non
pre-ComplexType source fields.
+ if (sourceFieldConfig.isPreComplexTypeTransform() ==
preComplexTypeTransform) {
+ dataTypes.put(sourceFieldConfig.getName(),
sourceFieldConfig.getDataType());
+ }
+ }
+ }
+ }
+ // Auto-register aggregation source columns not in the schema so mistyped
JSON/Avro string numbers are converted
+ // before MutableSegmentImpl indexes them. Explicit SourceFieldConfig wins
(already in dataTypes). Only runs in the
+ // post-complex-type phase so flattened/unnested fields are available.
Auto-derived columns are converted lazily
+ // (only when the incoming value is incompatible with the aggregators) to
avoid per-record conversion overhead on
+ // the common correctly-typed path.
+ Set<String> lazyColumns = new HashSet<>();
+ if (!preComplexTypeTransform && schema != null) {
+ addAggregationSourceDataTypes(tableConfig, schema, dataTypes,
lazyColumns);
+ }
+ if (!dataTypes.isEmpty()) {
+ transformers.add(new DataTypeTransformer(tableConfig, dataTypes,
lazyColumns));
+ }
+ }
+
+ /// Derives [PinotDataType]s for ingestion-aggregation source columns that
are absent from the schema (and not already
+ /// covered by an explicit [SourceFieldConfig]). Types are inferred from the
aggregation function and destination
+ /// metric, and recorded in `lazyColumns` so that [DataTypeTransformer] only
converts values the aggregators cannot
+ /// consume directly. Sketch/HLL/COUNT sources are left unconverted so
offering semantics (e.g. hashing a string vs a
+ /// number) are preserved.
[org.apache.pinot.segment.local.aggregator.ValueAggregatorUtils#toDouble]
remains a safety
+ /// net.
+ static void addAggregationSourceDataTypes(TableConfig tableConfig, Schema
schema,
+ Map<String, PinotDataType> dataTypes, Set<String> lazyColumns) {
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
if (ingestionConfig == null) {
return;
}
- List<SourceFieldConfig> sourceFieldConfigs =
ingestionConfig.getSourceFieldConfigs();
- if (CollectionUtils.isEmpty(sourceFieldConfigs)) {
+ List<AggregationConfig> aggregationConfigs =
ingestionConfig.getAggregationConfigs();
+ if (CollectionUtils.isEmpty(aggregationConfigs)) {
return;
}
- Map<String, PinotDataType> dataTypes = new HashMap<>();
- for (SourceFieldConfig sourceFieldConfig : sourceFieldConfigs) {
- // If pre-ComplexType transformers are requested, add only
pre-ComplexType source fields. Similarly, if
- // non pre-ComplexType transformers are requested, add only non
pre-ComplexType source fields.
- if (sourceFieldConfig.isPreComplexTypeTransform() ==
preComplexTypeTransform) {
- dataTypes.put(sourceFieldConfig.getName(),
sourceFieldConfig.getDataType());
+ for (AggregationConfig aggregationConfig : aggregationConfigs) {
+ String destColumn = aggregationConfig.getColumnName();
+ String aggregationFunction = aggregationConfig.getAggregationFunction();
+ if (destColumn == null || aggregationFunction == null) {
+ continue;
+ }
+ ExpressionContext expressionContext;
+ try {
+ expressionContext =
RequestContextUtils.getExpression(aggregationFunction);
+ } catch (Exception e) {
+ // Invalid configs are rejected at table-create validation time; skip
here to keep transformer build resilient.
+ continue;
+ }
+ if (expressionContext.getType() != ExpressionContext.Type.FUNCTION) {
+ continue;
+ }
+ FunctionContext functionContext = expressionContext.getFunction();
+ AggregationFunctionType functionType;
+ try {
+ functionType =
AggregationFunctionType.getAggregationFunctionType(functionContext.getFunctionName());
+ } catch (Exception e) {
+ continue;
+ }
+ List<ExpressionContext> arguments = functionContext.getArguments();
+ if (arguments.isEmpty()) {
+ continue;
+ }
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() != ExpressionContext.Type.IDENTIFIER) {
+ continue;
+ }
+ String sourceColumn = firstArgument.getIdentifier();
+ if (AggregationFunctionColumnPair.STAR.equals(sourceColumn) ||
schema.hasColumn(sourceColumn)
+ || dataTypes.containsKey(sourceColumn)) {
+ // Explicit SourceFieldConfig or schema column already covers
conversion; COUNT(*) has no source value.
+ continue;
+ }
+ FieldSpec destFieldSpec = schema.getFieldSpecFor(destColumn);
+ PinotDataType inferredType =
inferAggregationSourceDataType(functionType, destFieldSpec);
+ if (inferredType != null) {
+ dataTypes.put(sourceColumn, inferredType);
Review Comment:
This overwrite makes conversion depend on aggregation-config order when one
raw source feeds multiple destination metrics. For example, a valid
`SUM_PRECISION(metric, 38)` plus `AVG(metric)` configuration can let the later
`DOUBLE` inference replace `BIG_DECIMAL`, causing a large decimal string to
lose digits before `SumPrecisionValueAggregator` sees it. Merge compatible
requirements deterministically or reject conflicting inferred types.
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerTest.java:
##########
@@ -615,6 +616,186 @@ public void testSourceFieldDataTypeConversion() {
assertNull(nullRecord.getValue("srcLong"));
}
+ @Test
+ public void testAggregationSourceAutoDataTypeConversion() {
+ // Aggregation source "metric" is not in the schema; TransformPipeline
should still convert string "123" before
+ // indexing (issue #16317).
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("aggSrcSchema")
+ .addSingleValueDimension("dim", DataType.STRING)
+ .addMetric("sumMetric", DataType.DOUBLE)
+ .addMetric("minMetric", DataType.DOUBLE)
+ .addMetric("maxMetric", DataType.DOUBLE)
+ .addDateTime("ts", DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setAggregationConfigs(List.of(
+ new AggregationConfig("sumMetric", "SUM(metric)"),
+ new AggregationConfig("minMetric", "MIN(metric)"),
+ new AggregationConfig("maxMetric", "MAX(metric)")));
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName("aggSrcTable")
+ .setTimeColumnName("ts")
+ .setNoDictionaryColumns(List.of("sumMetric", "minMetric", "maxMetric"))
+ .setIngestionConfig(ingestionConfig)
+ .build();
+
+ List<RecordTransformer> transformers =
RecordTransformerUtils.getDefaultTransformers(tableConfig, schema);
+ // First transformer is the post-complex-type source-field
DataTypeTransformer covering "metric".
+ assertTrue(transformers.get(0) instanceof DataTypeTransformer);
+ assertEquals(transformers.get(0).getInputColumns(), Set.of("metric"));
+
+ TransformPipeline pipeline = new TransformPipeline(tableConfig, schema);
+ GenericRow row = new GenericRow();
+ row.putValue("dim", "a");
+ row.putValue("ts", 1L);
+ row.putValue("metric", "123");
+ TransformPipeline.Result result = pipeline.processRow(row);
+ assertEquals(result.getTransformedRows().size(), 1);
+ assertEquals(result.getTransformedRows().get(0).getValue("metric"), 123.0);
+
+ // Lazy conversion: an already-typed value passes through as the same
object reference (no conversion, no
+ // allocation).
+ Double typedValue = 42.5;
+ GenericRow typedRow = new GenericRow();
+ typedRow.putValue("dim", "a");
+ typedRow.putValue("ts", 1L);
+ typedRow.putValue("metric", typedValue);
+
assertSame(pipeline.processRow(typedRow).getTransformedRows().get(0).getValue("metric"),
typedValue);
+
+ // Lazy conversion: a compatible Number of a different box (Integer for
DOUBLE target) is also passed through
+ // untouched; ValueAggregatorUtils.toDouble accepts any Number.
+ Integer intBoxValue = 7;
+ GenericRow intBoxRow = new GenericRow();
+ intBoxRow.putValue("dim", "a");
+ intBoxRow.putValue("ts", 1L);
+ intBoxRow.putValue("metric", intBoxValue);
+
assertSame(pipeline.processRow(intBoxRow).getTransformedRows().get(0).getValue("metric"),
intBoxValue);
+
+ // Non-numeric string fails before indexing when continueOnError is false.
+ GenericRow bad = new GenericRow();
+ bad.putValue("dim", "a");
+ bad.putValue("ts", 1L);
+ bad.putValue("metric", "abc");
+ try {
+ pipeline.processRow(bad);
+ fail("Expected data type conversion failure for non-numeric aggregation
source");
+ } catch (Exception e) {
+ // expected
+ }
+
+ // With continueOnError, unparsable source becomes null and the row is
marked incomplete.
+ ingestionConfig.setContinueOnError(true);
+ TransformPipeline continuePipeline = new TransformPipeline(tableConfig,
schema);
+ GenericRow badContinue = new GenericRow();
+ badContinue.putValue("dim", "a");
+ badContinue.putValue("ts", 1L);
+ badContinue.putValue("metric", "abc");
+ TransformPipeline.Result continueResult =
continuePipeline.processRow(badContinue);
+ assertEquals(continueResult.getTransformedRows().size(), 1);
+ assertNull(continueResult.getTransformedRows().get(0).getValue("metric"));
+ assertTrue(continueResult.getTransformedRows().get(0).isIncomplete());
+ assertEquals(continueResult.getIncompleteRowCount(), 1);
+ }
+
+ @Test
+ public void testAggregationSourceMultiValueAutoDataTypeConversion() {
+ // SUMMV source is multi-value: string elements must convert to Double[]
(not throw on multi-element arrays), and
+ // an already-typed Double[] must pass through as the same object
reference.
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("aggMvSrcSchema")
+ .addSingleValueDimension("dim", DataType.STRING)
+ .addMetric("summvMetric", DataType.DOUBLE)
+ .addDateTime("ts", DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setAggregationConfigs(List.of(new
AggregationConfig("summvMetric", "SUMMV(metricMv)")));
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName("aggMvSrcTable")
+ .setTimeColumnName("ts")
+ .setNoDictionaryColumns(List.of("summvMetric"))
+ .setIngestionConfig(ingestionConfig)
+ .build();
+
+ TransformPipeline pipeline = new TransformPipeline(tableConfig, schema);
+ GenericRow row = new GenericRow();
+ row.putValue("dim", "a");
+ row.putValue("ts", 1L);
+ row.putValue("metricMv", new Object[]{"1.5", "2.5"});
+ Object converted =
pipeline.processRow(row).getTransformedRows().get(0).getValue("metricMv");
+ assertEquals(converted, new Double[]{1.5, 2.5});
+
+ Double[] typedValues = new Double[]{3.5, 4.5};
+ GenericRow typedRow = new GenericRow();
+ typedRow.putValue("dim", "a");
+ typedRow.putValue("ts", 1L);
+ typedRow.putValue("metricMv", typedValues);
+
assertSame(pipeline.processRow(typedRow).getTransformedRows().get(0).getValue("metricMv"),
typedValues);
+
+ // Unparsable element fails in the transformer, before MutableSegmentImpl
mutates the row.
+ GenericRow bad = new GenericRow();
+ bad.putValue("dim", "a");
+ bad.putValue("ts", 1L);
+ bad.putValue("metricMv", new Object[]{"1.5", "abc"});
+ try {
+ pipeline.processRow(bad);
+ fail("Expected data type conversion failure for non-numeric multi-value
aggregation source");
+ } catch (Exception e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testAggregationSourceExplicitSourceFieldConfigWins() {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("aggSrcSchema")
+ .addSingleValueDimension("dim", DataType.STRING)
+ .addMetric("sumMetric", DataType.LONG)
+ .addDateTime("ts", DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setAggregationConfigs(List.of(new
AggregationConfig("sumMetric", "SUM(metric)")));
+ // Explicit LONG conversion wins over auto DOUBLE inference from
destination LONG... destination is LONG so auto
+ // would also be LONG; use INT destination type via explicit override to
prove precedence.
Review Comment:
This explanation reverses the configured types: the
destination/auto-inferred type is `LONG`, while the explicit override is `INT`.
As written, the comment makes the precedence test misleading.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
