Jackie-Jiang commented on code in PR #19087:
URL: https://github.com/apache/pinot/pull/19087#discussion_r3984900107
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,29 +154,193 @@ 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) {
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
if (ingestionConfig == null) {
return;
}
- List<SourceFieldConfig> sourceFieldConfigs =
ingestionConfig.getSourceFieldConfigs();
- if (CollectionUtils.isEmpty(sourceFieldConfigs)) {
- 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());
+ 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());
+ }
}
}
+ // Opt-in: convert aggregation source columns that are not in the schema
(and not already covered by an explicit
+ // SourceFieldConfig) so mistyped JSON/Avro string numbers are converted
before MutableSegmentImpl indexes them.
+ // Off by default; uses the stock DataTypeTransformer (no lazy
compatibility short-circuit).
+ if (!preComplexTypeTransform && schema != null &&
ingestionConfig.isConvertAggregationSourceTypes()) {
+ addAggregationSourceDataTypes(tableConfig, schema, dataTypes);
+ }
if (!dataTypes.isEmpty()) {
transformers.add(new DataTypeTransformer(tableConfig, dataTypes));
}
}
+ /// Derives [PinotDataType]s for ingestion-aggregation source columns that
are absent from the schema (and not already
+ /// covered by an explicit [SourceFieldConfig] in either phase). Types are
inferred from the aggregation function and
+ /// destination metric. When one source feeds multiple aggregations,
inferred numeric types are merged by keeping the
+ /// wider type so config order cannot drop precision. 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.
+ @VisibleForTesting
+ static void addAggregationSourceDataTypes(TableConfig tableConfig, Schema
schema,
+ Map<String, PinotDataType> dataTypes) {
+ IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
+ List<AggregationConfig> aggregationConfigs =
ingestionConfig.getAggregationConfigs();
+ if (CollectionUtils.isEmpty(aggregationConfigs)) {
+ return;
+ }
+ // dataTypes only has this phase's SourceFieldConfigs. Pre-complex-type
names are absent from the post-phase map
+ // and must still skip inference so an explicit type is not overwritten.
+ Set<String> explicitSourceFields =
getExplicitSourceFieldNames(ingestionConfig);
+ for (AggregationConfig aggregationConfig : aggregationConfigs) {
+ String destColumn = aggregationConfig.getColumnName();
+ String aggregationFunction = aggregationConfig.getAggregationFunction();
+ 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)
+ || explicitSourceFields.contains(sourceColumn)) {
+ // Any explicit SourceFieldConfig (including pre-complex-type) 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) {
+ PinotDataType existing = dataTypes.get(sourceColumn);
+ dataTypes.put(sourceColumn,
+ existing == null ? inferredType :
mergeInferredAggregationSourceTypes(existing, inferredType));
Review Comment:
[P1] Preserve raw input when numeric and HLL aggregations share a source
Skipping inference for an HLL aggregation does not protect its source if
another aggregation registers a conversion for the same column. For example,
configure `SUM(metric)` into a DOUBLE metric and `DISTINCTCOUNTHLL(metric, 12)`
into a BYTES metric, with `metric` absent from the schema and no explicit
SourceFieldConfig. Ingest `"01"` and `"1"` into the same group.
I reproduced this through TransformPipeline and MutableSegmentImpl on this
commit:
- Flag off: SUM = 2.0, HLL distinct count = 2.
- Flag on: SUM = 2.0, HLL distinct count = 1.
SUM itself remains correct. Its inferred conversion rewrites both shared
input values to `Double(1.0)` before HLL receives them, so HLL loses the
original string identity. Returning null for the HLL inference cannot undo the
conversion registered by SUM, regardless of configuration order.
Please preserve the raw value for identity-sensitive consumers and add a
regression test for this mixed-aggregation case. The source's safe conversion
cannot be inferred from one aggregation in isolation.
--
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]
