Vamsi-klu commented on code in PR #19087:
URL: https://github.com/apache/pinot/pull/19087#discussion_r3994755000
##########
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:
Thanks for the reproduction. Agreed this is a correctness bug, not an opt-in
caveat.
Skipping inference on DISTINCTCOUNTHLL only skips that one aggregation.
SUM(metric) still registers metric to DOUBLE, and DataTypeTransformer rewrites
the shared field before either aggregator runs. "01" and "1" both become
Double(1.0), so HLL cardinality drops 2 to 1. Config order does not matter.
We should not infer a source type from one aggregation and rewrite the
field. Plan:
1. Per-source compatibility: register a rewrite only if every aggregation
that reads that source is conversion-safe. Any COUNT / HLL / sketch / bitmap
consumer vetoes the rewrite.
2. Keep convertAggregationSourceTypes (default false) and the stock
DataTypeTransformer for numeric-only sources.
3. Leave explicit sourceFieldConfigs as the override, and document that they
are unsafe when an identity-sensitive aggregation shares the column.
4. Add a TransformPipeline + MutableSegmentImpl regression for SUM(metric) +
DISTINCTCOUNTHLL(metric, 12) with "01" and "1". Flag on and off both expect
SUM=2.0 and HLL cardinality=2.
If we also need fail-before-mutate for mixed tables, that would be a
follow-up: convert a local copy at the numeric aggregator input and never
putValue on the shared field. I would rather not move that into
MutableSegmentImpl unless you want it in this PR.
Does the compatibility veto match what you want?
--
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]
