Vamsi-klu commented on code in PR #19087:
URL: https://github.com/apache/pinot/pull/19087#discussion_r3998761700
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,26 +153,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) {
+ 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());
+ }
+ }
+ }
+ }
+ // 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 != 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.
+ static void addAggregationSourceDataTypes(TableConfig tableConfig, Schema
schema,
Review Comment:
Added on `addAggregationSourceDataTypes`. Left off `infer` / `merge` because
those are production.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,26 +153,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) {
+ Map<String, PinotDataType> dataTypes = new HashMap<>();
+ IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
Review Comment:
Restored at the top of `addSourceFieldDataTypeTransformer`.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,26 +153,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) {
+ 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());
+ }
+ }
+ }
+ }
+ // 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 != 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.
+ static void addAggregationSourceDataTypes(TableConfig tableConfig, Schema
schema,
+ Map<String, PinotDataType> dataTypes) {
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
if (ingestionConfig == null) {
return;
}
Review Comment:
Removed. Caller already returned.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,26 +153,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) {
+ 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());
+ }
+ }
+ }
+ }
+ // 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 != 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.
+ static void addAggregationSourceDataTypes(TableConfig tableConfig, Schema
schema,
+ Map<String, PinotDataType> dataTypes) {
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
if (ingestionConfig == null) {
return;
}
+ List<AggregationConfig> aggregationConfigs =
ingestionConfig.getAggregationConfigs();
+ if (CollectionUtils.isEmpty(aggregationConfigs)) {
+ return;
+ }
+ Set<String> explicitSourceFields =
getExplicitSourceFieldNames(ingestionConfig);
Review Comment:
No. That map is this phase only. `getExplicitSourceFieldNames` walks every
`SourceFieldConfig` so a pre-complex-type name cannot be overwritten after
unnest.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerUtils.java:
##########
@@ -142,26 +153,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) {
+ 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());
+ }
+ }
+ }
+ }
+ // 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 != 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.
+ static void addAggregationSourceDataTypes(TableConfig tableConfig, Schema
schema,
+ Map<String, PinotDataType> dataTypes) {
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
if (ingestionConfig == null) {
return;
}
+ List<AggregationConfig> aggregationConfigs =
ingestionConfig.getAggregationConfigs();
+ if (CollectionUtils.isEmpty(aggregationConfigs)) {
+ return;
+ }
+ Set<String> explicitSourceFields =
getExplicitSourceFieldNames(ingestionConfig);
+ for (AggregationConfig aggregationConfig : aggregationConfigs) {
+ String destColumn = aggregationConfig.getColumnName();
+ String aggregationFunction = aggregationConfig.getAggregationFunction();
+ if (destColumn == null || aggregationFunction == null) {
+ continue;
+ }
Review Comment:
Removed. Table-create validation already rejects those.
##########
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:
Fixed in `f7f074bf`. Wider-type merge. Covered by
`testAggregationSourceConflictingInferredTypesKeepWider`.
--
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]
