Jackie-Jiang commented on code in PR #19087:
URL: https://github.com/apache/pinot/pull/19087#discussion_r3858294097
##########
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:
(minor) Annotate it with `@VisibleForTesting`, same for other
package-private methods if only used for testing purpose
##########
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:
Is this available form the passed in `dataTypes`?
##########
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:
(minor) Shouldn't be possible
##########
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:
(minor) You can keep the original short-circuit of `ingestionConfig == null`
##########
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:
(minor) This check is redundant
--
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]
