Jackie-Jiang commented on code in PR #19087:
URL: https://github.com/apache/pinot/pull/19087#discussion_r3739345305
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -144,26 +152,132 @@ 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.
+ if (!preComplexTypeTransform && schema != null) {
+ 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]). Types are inferred from the
aggregation function and destination
+ /// metric. 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) {
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);
}
}
- if (!dataTypes.isEmpty()) {
- transformers.add(new DataTypeTransformer(tableConfig, dataTypes));
+ }
+
+ /// Returns the target type for converting an aggregation source column, or
{@code null} when no conversion should be
+ /// applied (COUNT, HLL, sketches — keep raw offering values).
+ @Nullable
+ static PinotDataType inferAggregationSourceDataType(AggregationFunctionType
functionType,
Review Comment:
Currently this will do too many conversions (performance overhead). We
should only return non-null type when the type is incompatible
--
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]
