diff options
author | Tathagata Das <tathagata.das1565@gmail.com> | 2016-12-15 11:54:35 -0800 |
---|---|---|
committer | Tathagata Das <tathagata.das1565@gmail.com> | 2016-12-15 11:54:35 -0800 |
commit | 4f7292c87512a7da3542998d0e5aa21c27a511e9 (patch) | |
tree | e1c5184f76024c3c106fae7c85c484022463e1d7 /sql/catalyst/src/test | |
parent | 01e14bf303e61a5726f3b1418357a50c1bf8b16f (diff) | |
download | spark-4f7292c87512a7da3542998d0e5aa21c27a511e9.tar.gz spark-4f7292c87512a7da3542998d0e5aa21c27a511e9.tar.bz2 spark-4f7292c87512a7da3542998d0e5aa21c27a511e9.zip |
[SPARK-18870] Disallowed Distinct Aggregations on Streaming Datasets
## What changes were proposed in this pull request?
Check whether Aggregation operators on a streaming subplan have aggregate expressions with isDistinct = true.
## How was this patch tested?
Added unit test
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #16289 from tdas/SPARK-18870.
Diffstat (limited to 'sql/catalyst/src/test')
-rw-r--r-- | sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala | 13 |
1 files changed, 13 insertions, 0 deletions
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala index ff1bb126f4..34e94c7142 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala @@ -98,6 +98,19 @@ class UnsupportedOperationsSuite extends SparkFunSuite { outputMode = Update, expectedMsgs = Seq("multiple streaming aggregations")) + // Aggregation: Distinct aggregates not supported on streaming relation + val distinctAggExprs = Seq(Count("*").toAggregateExpression(isDistinct = true).as("c")) + assertSupportedInStreamingPlan( + "distinct aggregate - aggregate on batch relation", + Aggregate(Nil, distinctAggExprs, batchRelation), + outputMode = Append) + + assertNotSupportedInStreamingPlan( + "distinct aggregate - aggregate on streaming relation", + Aggregate(Nil, distinctAggExprs, streamRelation), + outputMode = Complete, + expectedMsgs = Seq("distinct aggregation")) + // Inner joins: Stream-stream not supported testBinaryOperationInStreamingPlan( "inner join", |