fix(function): Support Spark legacy behavior for central moments functions and change the input type #12566
+208
−60
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -887,6 +887,11 @@ Spark-specific Configuration | |
| Joda date formatter performs strict checking of its input and uses different pattern string. | ||
| For example, the 2015-07-22 10:00:00 timestamp cannot be parsed if pattern is yyyy-MM-dd because the parser does not consume whole input. | ||
| Another example is that the 'W' pattern, which means week in month, is not supported. For more differences, see :issue:`10354`. | ||
| * - spark.legacy_statistical_aggregate | ||
rui-mo marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| - bool | ||
| - false | ||
| - if true, statistical aggregation function includes skewness, kurtosis will return std::numeric_limits<double>::quiet_NaN() | ||
|
There was a problem hiding this comment. Please highlight that this config is partially honored, there is still some functions should honor this config but not such as stddev There was a problem hiding this comment. Perhaps rename it to statistical_agg_null_on_divide_by_zero and update all related functions in this PR as well. There was a problem hiding this comment. @zhli1142015 @jinchengchenghh There was a problem hiding this comment. +1 for improving the documentation. We'd better make it clear which functions will depend on this config when adding it. There was a problem hiding this comment.
@rui-mo You are right. Updated. Thanks! There was a problem hiding this comment.
@zhli1142015 I thought it would be clear to align with Spark. |
||
| - instead of NULL when DivideByZero occurs during expression evaluation. | ||
|
|
||
| Tracing | ||
| -------- | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,40 +15,51 @@ | |
| */ | ||
|
|
||
| #include "velox/functions/sparksql/aggregates/CentralMomentsAggregate.h" | ||
| #include <limits> | ||
| #include "velox/functions/lib/aggregates/CentralMomentsAggregatesBase.h" | ||
|
|
||
| namespace facebook::velox::functions::aggregate::sparksql { | ||
|
|
||
| namespace { | ||
| template <bool nullOnDivideByZero> | ||
|
There was a problem hiding this comment. Perhaps document this variable. |
||
| struct SkewnessResultAccessor { | ||
| static bool hasResult(const CentralMomentsAccumulator& accumulator) { | ||
| if constexpr (nullOnDivideByZero) { | ||
| return accumulator.count() >= 1 && accumulator.m2() != 0; | ||
| } | ||
| return accumulator.count() >= 1; | ||
| } | ||
|
|
||
| static double result(const CentralMomentsAccumulator& accumulator) { | ||
| if (accumulator.m2() == 0) { | ||
| return std::numeric_limits<double>::quiet_NaN(); | ||
|
There was a problem hiding this comment. Perhaps check |
||
| } | ||
| return std::sqrt(accumulator.count()) * accumulator.m3() / | ||
| std::pow(accumulator.m2(), 1.5); | ||
| } | ||
| }; | ||
|
|
||
| template <bool nullOnDivideByZero> | ||
| struct KurtosisResultAccessor { | ||
| static bool hasResult(const CentralMomentsAccumulator& accumulator) { | ||
| if constexpr (nullOnDivideByZero) { | ||
| return accumulator.count() >= 1 && accumulator.m2() != 0; | ||
| } | ||
| return accumulator.count() >= 1; | ||
| } | ||
|
|
||
| static double result(const CentralMomentsAccumulator& accumulator) { | ||
| if (accumulator.m2() == 0) { | ||
| return std::numeric_limits<double>::quiet_NaN(); | ||
| } | ||
| double count = accumulator.count(); | ||
| double m2 = accumulator.m2(); | ||
| double m4 = accumulator.m4(); | ||
| return count * m4 / (m2 * m2) - 3.0; | ||
| } | ||
| }; | ||
|
|
||
| std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> getSignatures() { | ||
| std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures; | ||
| std::vector<std::string> inputTypes = { | ||
| "smallint", "integer", "bigint", "real", "double"}; | ||
|
|
@@ -60,6 +71,115 @@ exec::AggregateRegistrationResult registerCentralMoments( | |
| .argumentType(inputType) | ||
| .build()); | ||
| } | ||
| return signatures; | ||
| } | ||
|
|
||
| exec::AggregateRegistrationResult registerSkewness( | ||
| const std::string& name, | ||
| bool withCompanionFunctions, | ||
| bool overwrite) { | ||
| std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures = | ||
| getSignatures(); | ||
|
|
||
| return exec::registerAggregateFunction( | ||
| name, | ||
| std::move(signatures), | ||
| [name]( | ||
| core::AggregationNode::Step step, | ||
| const std::vector<TypePtr>& argTypes, | ||
| const TypePtr& resultType, | ||
| const core::QueryConfig& config) -> std::unique_ptr<exec::Aggregate> { | ||
| VELOX_CHECK_LE( | ||
|
There was a problem hiding this comment. I think this function should require exactly one argument, rather than allowing at most one. |
||
| argTypes.size(), 1, "{} takes at most one argument", name); | ||
| const auto& inputType = argTypes[0]; | ||
| if (config.sparkLegacyStatisticalAggregate()) { | ||
| if (exec::isRawInput(step)) { | ||
| switch (inputType->kind()) { | ||
|
There was a problem hiding this comment. We might don't need a switch clause if only one valid case. |
||
| case TypeKind::SMALLINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int16_t, | ||
| SkewnessResultAccessor<false>>>(resultType); | ||
| case TypeKind::INTEGER: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int32_t, | ||
| SkewnessResultAccessor<false>>>(resultType); | ||
| case TypeKind::BIGINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t, | ||
| SkewnessResultAccessor<false>>>(resultType); | ||
| case TypeKind::DOUBLE: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| double, | ||
| SkewnessResultAccessor<false>>>(resultType); | ||
| case TypeKind::REAL: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| float, | ||
| SkewnessResultAccessor<false>>>(resultType); | ||
| default: | ||
| VELOX_UNSUPPORTED( | ||
| "Unsupported input type: {}. " | ||
| "Expected SMALLINT, INTEGER, BIGINT, DOUBLE or REAL.", | ||
| inputType->toString()); | ||
| } | ||
| } else { | ||
| checkAccumulatorRowType( | ||
| inputType, | ||
| "Input type for final aggregation must be " | ||
| "(count:bigint, m1:double, m2:double, m3:double, m4:double) struct"); | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t /*unused*/, | ||
| SkewnessResultAccessor<false>>>(resultType); | ||
| } | ||
| } else { | ||
| if (exec::isRawInput(step)) { | ||
| switch (inputType->kind()) { | ||
| case TypeKind::SMALLINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int16_t, | ||
| SkewnessResultAccessor<true>>>(resultType); | ||
| case TypeKind::INTEGER: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int32_t, | ||
| SkewnessResultAccessor<true>>>(resultType); | ||
| case TypeKind::BIGINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t, | ||
| SkewnessResultAccessor<true>>>(resultType); | ||
| case TypeKind::DOUBLE: | ||
|
There was a problem hiding this comment. I think we only need to register this function with double type raw input. There was a problem hiding this comment. Do you mean this function should only support double type input? If so, I agree with you, as Spark only supports double type input, see https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/CentralMomentAgg.scala#L58C16-L58C24 |
||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| double, | ||
| SkewnessResultAccessor<true>>>(resultType); | ||
| case TypeKind::REAL: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| float, | ||
| SkewnessResultAccessor<true>>>(resultType); | ||
| default: | ||
| VELOX_UNSUPPORTED( | ||
| "Unsupported input type: {}. " | ||
| "Expected SMALLINT, INTEGER, BIGINT, DOUBLE or REAL.", | ||
| inputType->toString()); | ||
| } | ||
| } else { | ||
| checkAccumulatorRowType( | ||
| inputType, | ||
| "Input type for final aggregation must be " | ||
| "(count:bigint, m1:double, m2:double, m3:double, m4:double) struct"); | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t /*unused*/, | ||
| SkewnessResultAccessor<true>>>(resultType); | ||
| } | ||
| } | ||
| }, | ||
| withCompanionFunctions, | ||
| overwrite); | ||
| } | ||
|
|
||
| exec::AggregateRegistrationResult registerKurtosis( | ||
| const std::string& name, | ||
| bool withCompanionFunctions, | ||
| bool overwrite) { | ||
| std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures = | ||
| getSignatures(); | ||
|
|
||
| return exec::registerAggregateFunction( | ||
| name, | ||
|
|
@@ -68,47 +188,86 @@ exec::AggregateRegistrationResult registerCentralMoments( | |
| core::AggregationNode::Step step, | ||
| const std::vector<TypePtr>& argTypes, | ||
| const TypePtr& resultType, | ||
| const core::QueryConfig& config) -> std::unique_ptr<exec::Aggregate> { | ||
| VELOX_CHECK_LE( | ||
| argTypes.size(), 1, "{} takes at most one argument", name); | ||
| const auto& inputType = argTypes[0]; | ||
| if (config.sparkLegacyStatisticalAggregate()) { | ||
| if (exec::isRawInput(step)) { | ||
| switch (inputType->kind()) { | ||
| case TypeKind::SMALLINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int16_t, | ||
| KurtosisResultAccessor<false>>>(resultType); | ||
| case TypeKind::INTEGER: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int32_t, | ||
| KurtosisResultAccessor<false>>>(resultType); | ||
| case TypeKind::BIGINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t, | ||
| KurtosisResultAccessor<false>>>(resultType); | ||
| case TypeKind::DOUBLE: | ||
|
There was a problem hiding this comment. same for this function. |
||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| double, | ||
| KurtosisResultAccessor<false>>>(resultType); | ||
| case TypeKind::REAL: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| float, | ||
| KurtosisResultAccessor<false>>>(resultType); | ||
| default: | ||
| VELOX_UNSUPPORTED( | ||
| "Unsupported input type: {}. " | ||
| "Expected SMALLINT, INTEGER, BIGINT, DOUBLE or REAL.", | ||
| inputType->toString()); | ||
| } | ||
| } else { | ||
| checkAccumulatorRowType( | ||
| inputType, | ||
| "Input type for final aggregation must be " | ||
| "(count:bigint, m1:double, m2:double, m3:double, m4:double) struct"); | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t /*unused*/, | ||
| KurtosisResultAccessor<false>>>(resultType); | ||
| } | ||
| } else { | ||
| if (exec::isRawInput(step)) { | ||
| switch (inputType->kind()) { | ||
| case TypeKind::SMALLINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int16_t, | ||
| KurtosisResultAccessor<true>>>(resultType); | ||
| case TypeKind::INTEGER: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int32_t, | ||
| KurtosisResultAccessor<true>>>(resultType); | ||
| case TypeKind::BIGINT: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t, | ||
| KurtosisResultAccessor<true>>>(resultType); | ||
| case TypeKind::DOUBLE: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| double, | ||
| KurtosisResultAccessor<true>>>(resultType); | ||
| case TypeKind::REAL: | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| float, | ||
| KurtosisResultAccessor<true>>>(resultType); | ||
| default: | ||
| VELOX_UNSUPPORTED( | ||
| "Unsupported input type: {}. " | ||
| "Expected SMALLINT, INTEGER, BIGINT, DOUBLE or REAL.", | ||
| inputType->toString()); | ||
| } | ||
| } else { | ||
| checkAccumulatorRowType( | ||
| inputType, | ||
| "Input type for final aggregation must be " | ||
| "(count:bigint, m1:double, m2:double, m3:double, m4:double) struct"); | ||
| return std::make_unique<CentralMomentsAggregatesBase< | ||
| int64_t /*unused*/, | ||
| KurtosisResultAccessor<true>>>(resultType); | ||
| } | ||
| } | ||
| }, | ||
| withCompanionFunctions, | ||
|
|
@@ -120,10 +279,8 @@ void registerCentralMomentsAggregate( | |
| const std::string& prefix, | ||
| bool withCompanionFunctions, | ||
| bool overwrite) { | ||
| registerSkewness(prefix + "skewness", withCompanionFunctions, overwrite); | ||
| registerKurtosis(prefix + "kurtosis", withCompanionFunctions, overwrite); | ||
| } | ||
|
|
||
| } // namespace facebook::velox::functions::aggregate::sparksql | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If true, the first letter should be uppercase, so as the document
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks.