-
Notifications
You must be signed in to change notification settings - Fork 28.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[SPARK-19851][SQL] Add support for EVERY and ANY (SOME) aggregates #22809
Changes from 6 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -21,6 +21,7 @@ import java.util.Locale | |
|
||
import org.apache.spark.sql.catalyst.InternalRow | ||
import org.apache.spark.sql.catalyst.analysis.{TypeCheckResult, TypeCoercion} | ||
import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate | ||
import org.apache.spark.sql.catalyst.expressions.codegen._ | ||
import org.apache.spark.sql.catalyst.expressions.codegen.Block._ | ||
import org.apache.spark.sql.catalyst.trees.TreeNode | ||
|
@@ -282,6 +283,31 @@ trait RuntimeReplaceable extends UnaryExpression with Unevaluable { | |
override lazy val canonicalized: Expression = child.canonicalized | ||
} | ||
|
||
/** | ||
* An aggregate expression that gets rewritten (currently by the optimizer) into a | ||
* different aggregate expression for evaluation. This is mainly used to provide compatibility | ||
* with other databases. For example, we use this to support every, any/some aggregates by rewriting | ||
* them with Min and Max respectively. | ||
*/ | ||
trait UnevaluableAggregate extends DeclarativeAggregate { | ||
|
||
override def nullable: Boolean = true | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why do we set them always as nullable? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @mgaido91 most of the aggregates are nullable, no ? Did you have an suggestion here ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. shouldn't this be nullable only if the incoming expression is? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @mgaido91 I think for aggregates, its different ? Please see There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ah, right, I was missing that case, sorry, thanks. |
||
|
||
override lazy val aggBufferAttributes = | ||
throw new UnsupportedOperationException(s"Cannot evaluate aggBufferAttributes: $this") | ||
|
||
override lazy val initialValues: Seq[Expression] = | ||
throw new UnsupportedOperationException(s"Cannot evaluate initialValues: $this") | ||
|
||
override lazy val updateExpressions: Seq[Expression] = | ||
throw new UnsupportedOperationException(s"Cannot evaluate updateExpressions: $this") | ||
|
||
override lazy val mergeExpressions: Seq[Expression] = | ||
throw new UnsupportedOperationException(s"Cannot evaluate mergeExpressions: $this") | ||
|
||
override lazy val evaluateExpression: Expression = | ||
throw new UnsupportedOperationException(s"Cannot evaluate evaluateExpression: $this") | ||
} | ||
|
||
/** | ||
* Expressions that don't have SQL representation should extend this trait. Examples are | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,62 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.sql.catalyst.expressions.aggregate | ||
|
||
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult | ||
import org.apache.spark.sql.catalyst.expressions._ | ||
import org.apache.spark.sql.types._ | ||
|
||
abstract class UnevaluableBooleanAggBase(arg: Expression) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why don't we reuse There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @mgaido91 RuntimeReplaceble works for scalar expressions but not for aggregate expressions. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why it doesn't work? Sorry if the question is dumb, but I can't see the problem in using it here, probably I am missing something. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @mgaido91 Actually i tried a few different ideas. They are in the comment of original PR link. I had prepared two branches with two approaches .. [comment] (#22047 (comment)). Could you please take a look ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ok, I see you mentioned that there were issues using There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @mgaido91 I tried it here. I had trouble getting it to work for window expressions. Thats why @cloud-fan suggested to try the current approach in this comment There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @mgaido91 I re-read your comments again and here is what i think you are implying :
Did i understand it correctly ? If so, actually, i wouldn't prefer that way. The reason is, the replacedExpression is hidden from analyzer and so it may not be safe. The way ReplaceExpression framework is nicely designed, the analyzer resolves the rewritten expression normally (as its the child expression). Thats the reason, i have opted for specific/targeted rewrites. If Wenchen and you think otherwise, then i can change. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yes, but in the analyzer you analyze the children of the current expression right? So we would just have something like Anyway, since you have a different opinion, let's see what @cloud-fan thinks on this. Thanks. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We can leave a TODO saying that we should create a framework to replace aggregate functions, but I think the current patch is good enough for these 3 functions, and I'm not aware of more functions like them that we need to deal with. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @cloud-fan @mgaido91 Thank you. I have added a TODO for now. |
||
extends UnevaluableAggregate with ImplicitCastInputTypes { | ||
|
||
override def children: Seq[Expression] = arg :: Nil | ||
|
||
override def dataType: DataType = BooleanType | ||
|
||
override def inputTypes: Seq[AbstractDataType] = Seq(BooleanType) | ||
|
||
override def checkInputDataTypes(): TypeCheckResult = { | ||
arg.dataType match { | ||
case dt if dt != BooleanType => | ||
TypeCheckResult.TypeCheckFailure(s"Input to function '$prettyName' should have been " + | ||
s"${BooleanType.simpleString}, but it's [${arg.dataType.catalogString}].") | ||
case _ => TypeCheckResult.TypeCheckSuccess | ||
} | ||
} | ||
} | ||
|
||
@ExpressionDescription( | ||
usage = "_FUNC_(expr) - Returns true if all values of `expr` are true.", | ||
since = "3.0.0") | ||
case class EveryAgg(arg: Expression) extends UnevaluableBooleanAggBase(arg) { | ||
override def nodeName: String = "Every" | ||
} | ||
|
||
@ExpressionDescription( | ||
usage = "_FUNC_(expr) - Returns true if at least one value of `expr` is true.", | ||
since = "3.0.0") | ||
case class AnyAgg(arg: Expression) extends UnevaluableBooleanAggBase(arg) { | ||
override def nodeName: String = "Any" | ||
} | ||
|
||
@ExpressionDescription( | ||
usage = "_FUNC_(expr) - Returns true if at least one value of `expr` is true.", | ||
since = "3.0.0") | ||
case class SomeAgg(arg: Expression) extends UnevaluableBooleanAggBase(arg) { | ||
override def nodeName: String = "Some" | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -144,6 +144,9 @@ class ExpressionTypeCheckingSuite extends SparkFunSuite { | |
assertSuccess(Sum('stringField)) | ||
assertSuccess(Average('stringField)) | ||
assertSuccess(Min('arrayField)) | ||
assertSuccess(new EveryAgg('booleanField)) | ||
assertSuccess(new AnyAgg('booleanField)) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. shall we add also |
||
assertSuccess(new SomeAgg('booleanField)) | ||
|
||
assertError(Min('mapField), "min does not support ordering on type") | ||
assertError(Max('mapField), "max does not support ordering on type") | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -80,3 +80,69 @@ SELECT 1 FROM range(10) HAVING true; | |
SELECT 1 FROM range(10) HAVING MAX(id) > 0; | ||
|
||
SELECT id FROM range(10) HAVING id > 0; | ||
|
||
-- Test data | ||
CREATE OR REPLACE TEMPORARY VIEW test_agg AS SELECT * FROM VALUES | ||
(1, true), (1, false), | ||
(2, true), | ||
(3, false), (3, null), | ||
(4, null), (4, null), | ||
(5, null), (5, true), (5, false) AS test_agg(k, v); | ||
|
||
-- empty table | ||
SELECT every(v), some(v), any(v) FROM test_agg WHERE 1 = 0; | ||
|
||
-- all null values | ||
SELECT every(v), some(v), any(v) FROM test_agg WHERE k = 4; | ||
|
||
-- aggregates are null Filtering | ||
SELECT every(v), some(v), any(v) FROM test_agg WHERE k = 5; | ||
|
||
-- group by | ||
SELECT k, every(v), some(v), any(v) FROM test_agg GROUP BY k; | ||
|
||
-- having | ||
SELECT k, every(v) FROM test_agg GROUP BY k HAVING every(v) = false; | ||
SELECT k, every(v) FROM test_agg GROUP BY k HAVING every(v) IS NULL; | ||
|
||
-- basic subquery path to make sure rewrite happens in both parent and child plans. | ||
SELECT k, | ||
Every(v) AS every | ||
FROM test_agg | ||
WHERE k = 2 | ||
AND v IN (SELECT Any(v) | ||
FROM test_agg | ||
WHERE k = 1) | ||
GROUP BY k; | ||
|
||
-- basic subquery path to make sure rewrite happens in both parent and child plans. | ||
SELECT k, | ||
Every(v) AS every | ||
FROM test_agg | ||
WHERE k = 2 | ||
AND v IN (SELECT Every(v) | ||
FROM test_agg | ||
WHERE k = 1) | ||
GROUP BY k; | ||
|
||
-- input type checking Int | ||
SELECT every(1); | ||
|
||
-- input type checking Short | ||
SELECT some(1S); | ||
|
||
-- input type checking Long | ||
SELECT any(1L); | ||
|
||
-- input type checking String | ||
SELECT every("true"); | ||
|
||
-- every/some/any aggregates are supported as windows expression. | ||
SELECT k, v, every(v) OVER (PARTITION BY k ORDER BY v) FROM test_agg; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @cloud-fan here are a few window tests. (fyi) |
||
SELECT k, v, some(v) OVER (PARTITION BY k ORDER BY v) FROM test_agg; | ||
SELECT k, v, any(v) OVER (PARTITION BY k ORDER BY v) FROM test_agg; | ||
|
||
-- simple explain of queries having every/some/any agregates. Optimized | ||
-- plan should show the rewritten aggregate expression. | ||
EXPLAIN EXTENDED SELECT k, every(v), some(v), any(v) FROM test_agg GROUP BY k; | ||
|
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.
nit: unneeded newline