This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new b724fd87f4f2 [SPARK-58420][SQL] Normalize map values in distinct
aggregates
b724fd87f4f2 is described below
commit b724fd87f4f2d2e0d34a665f84b90edf5da16ef0
Author: Vladimir Golubev <[email protected]>
AuthorDate: Fri Aug 7 09:04:42 2026 +0000
[SPARK-58420][SQL] Normalize map values in distinct aggregates
### What changes were proposed in this pull request?
Normalize map-typed arguments of distinct aggregate expressions with
`MapSort`. Complex distinct
arguments are projected to attributes before normalization, and the
existing grouping aliases are
reused when applicable. The rule retains its plan-wide fast path and has a
default-on internal
killswitch.
### Why are the changes needed?
Map equality is independent of entry order, but distinct aggregation
currently compares the
physical map representation. Equivalent maps with different insertion
orders are therefore treated
as different values.
`InsertMapSortInGroupingExpressions` was introduced by
[SPARK-47430](https://github.com/apache/spark/commit/d57164a5cab92e8821563e776c19a424721adce3)
to normalize `MapType` grouping keys. This PR extends the same
normalization to distinct aggregate
arguments.
### Does this PR introduce _any_ user-facing change?
Yes. `COUNT(DISTINCT map_column)` now treats maps containing the same
entries in different orders as
the same value. For example:
```sql
SELECT count(DISTINCT m)
FROM VALUES
(map('a', 1, 'b', 2)),
(map('b', 2, 'a', 1)),
(map('a', 3)) AS t(m);
```
Before this change, the query returns `3`. After this change, it returns
`2`. Distinct aggregates
that return their argument, such as `COLLECT_LIST(DISTINCT m)`, now return
maps whose entries are
sorted by key.
Normalization is scoped to each use of the map. In a query containing both
`MAX(map_values(m)[0])` and `COUNT(DISTINCT m)`, the non-distinct aggregate
sees the original map
while the distinct aggregate sees the sorted map. Existing grouping-key
behavior is unchanged: a
normalized map grouping key is also the grouped value exposed to the select
list.
Setting `spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled` to
`false` disables the
distinct-specific normalization and restores the previous behavior;
grouping-key normalization is
unchanged.
### How was this patch tested?
Added a dedicated SQL golden test covering `COUNT(DISTINCT map)`, returned
`COLLECT_LIST(DISTINCT map)` values, multiple and multi-argument distinct
aggregates, grouped
aggregation, aggregate filters, `ORDER BY`, `HAVING`, nested maps, null and
empty maps, grouping
sets, and representative cases with the killswitch disabled.
```bash
build/sbt 'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z
distinct-map-aggregates.sql'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #57629 from vladimirg-db/fix-count-distinct-map.
Authored-by: Vladimir Golubev <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit f649423aa3574ec62af637392c713a3bc3daed9b)
Signed-off-by: Uros Bojanic <[email protected]>
---
docs/sql-migration-guide.md | 1 +
.../optimizer/InsertMapSortExpression.scala | 140 ++++++--
.../spark/sql/catalyst/optimizer/Optimizer.scala | 8 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 11 +
.../optimizer/InsertMapSortInAggregateSuite.scala | 117 +++++++
.../distinct-map-aggregates.sql.out | 363 +++++++++++++++++++++
.../sql-tests/inputs/distinct-map-aggregates.sql | 102 ++++++
.../results/distinct-map-aggregates.sql.out | 265 +++++++++++++++
8 files changed, 977 insertions(+), 30 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index b267a5d6fa45..efe999bb036c 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -28,6 +28,7 @@ license: |
## Upgrading from Spark SQL 4.2 to 4.3
+- Since Spark 4.3, map-typed arguments to distinct aggregates are normalized
by key order. As a result, `COUNT(DISTINCT m)` treats maps with the same
entries in different orders as the same value, and aggregates that return
distinct input values, such as `COLLECT_LIST(DISTINCT m)`, return maps sorted
by key.
- Since Spark 4.3, [ASOF JOIN](sql-ref-syntax-qry-select-asof-join.html) is
available as an opt-in SQL feature gated by `spark.sql.join.asofJoin.enabled`
(default `false`). When disabled, `ASOF JOIN` fails at parse time with
`UNSUPPORTED_FEATURE.ASOF_JOIN`.
- Since Spark 4.3, zero-length files are skipped during Parquet schema
inference instead of failing with a `FAILED_READ_FILE.CANNOT_READ_FILE_FOOTER`
error.
- Since Spark 4.3, metrics produced by `Dataset.observe` include values from
only the last successful task attempts instead of aggregating values from all
attempts. To restore the previous behavior, set
`spark.sql.legacy.observeMetricsAggregateAllAttempts` to `true`.
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortExpression.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortExpression.scala
index 9e613c54a49b..e2043e06186f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortExpression.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortExpression.scala
@@ -17,33 +17,49 @@
package org.apache.spark.sql.catalyst.optimizer
+import scala.collection.immutable.VectorMap
import scala.collection.mutable
-import org.apache.spark.sql.catalyst.expressions.{Alias, ArrayTransform,
CreateNamedStruct, Expression, GetStructField, If, IsNull, LambdaFunction,
Literal, MapFromArrays, MapKeys, MapSort, MapValues, NamedExpression,
NamedLambdaVariable}
+import org.apache.spark.sql.catalyst.expressions.{Alias, ArrayTransform,
Attribute, CreateNamedStruct, Expression, GetStructField, If, IsNull,
LambdaFunction, Literal, MapFromArrays, MapKeys, MapSort, MapValues,
NamedExpression, NamedLambdaVariable}
+import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
AggregateFunction}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan,
Project, RepartitionByExpression}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE,
REPARTITION_OPERATION}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
import org.apache.spark.util.ArrayImplicits.SparkArrayOps
/**
- * Adds [[MapSort]] to [[Aggregate]] expressions containing map columns,
- * as the key/value pairs need to be in the correct order before grouping:
+ * Adds [[MapSort]] to grouping expressions and distinct aggregate arguments
that contain maps,
+ * ensuring key/value pairs have a consistent order before aggregation:
*
* SELECT map_column, COUNT(*) FROM TABLE GROUP BY map_column =>
* SELECT _groupingmapsort as map_column, COUNT(*) FROM (
* SELECT map_sort(map_column) as _groupingmapsort FROM TABLE
* ) GROUP BY _groupingmapsort
+ *
+ * SELECT COUNT(DISTINCT map_column) FROM TABLE =>
+ * SELECT COUNT(DISTINCT _distinctmapsort) FROM (
+ * SELECT map_sort(map_column) as _distinctmapsort FROM TABLE
+ * )
+ *
+ * Distinct arguments are normalized here instead of in
[[RewriteDistinctAggregates]] because a
+ * single distinct group without a filter bypasses that rule and is handled by
the physical planner.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInAggregate extends Rule[LogicalPlan] {
import InsertMapSortExpression._
override def apply(plan: LogicalPlan): LogicalPlan = {
if (!plan.containsPattern(AGGREGATE)) {
return plan
}
+ val normalizeDistinctAggregates =
+ conf.getConf(SQLConf.INSERT_MAP_SORT_IN_DISTINCT_AGGREGATES_ENABLED)
val shouldRewrite = plan.exists {
- case agg: Aggregate if
agg.groupingExpressions.exists(mapTypeExistsRecursively) => true
+ case agg: Aggregate =>
+ agg.groupingExpressions.exists(mapTypeExistsRecursively) ||
+ (normalizeDistinctAggregates &&
+
distinctAggregateChildren(agg.aggregateExpressions).exists(mapTypeExistsRecursively))
case _ => false
}
if (!shouldRewrite) {
@@ -52,32 +68,104 @@ object InsertMapSortInGroupingExpressions extends
Rule[LogicalPlan] {
plan transformUpWithNewOutput {
case agg @ Aggregate(groupingExprs, aggregateExpressions, child, hint) =>
- val exprToMapSort = new mutable.HashMap[Expression, NamedExpression]
- val newGroupingKeys = groupingExprs.map { expr =>
- val inserted = insertMapSortRecursively(expr)
- if (expr.ne(inserted)) {
- exprToMapSort.getOrElseUpdate(
- expr.canonicalized,
- Alias(inserted, "_groupingmapsort")()
- ).toAttribute
+ val distinctExpressions = if (normalizeDistinctAggregates) {
+ distinctAggregateChildren(aggregateExpressions)
+ } else {
+ Seq.empty
+ }
+ val expressionsToNormalize = groupingExprs ++ distinctExpressions
+ if (!expressionsToNormalize.exists(mapTypeExistsRecursively)) {
+ agg -> Nil
+ } else {
+ val groupingMapSortAliases = insertMapSortInExpressions(
+ expressions = groupingExprs,
+ aliasName = "_groupingmapsort")
+ val distinctInputAliases =
createDistinctInputAliases(distinctExpressions)
+ val distinctMapSortAliases = insertMapSortInExpressions(
+ expressions = distinctExpressions,
+ aliasName = "_distinctmapsort",
+ reusableAliases = groupingMapSortAliases,
+ inputAliases = distinctInputAliases)
+ val newGroupingKeys = groupingExprs.map { expr =>
+
groupingMapSortAliases.get(expr.canonicalized).map(_.toAttribute).getOrElse(expr)
+ }
+ val newAggregateExprs = aggregateExpressions.map {
+ case named if groupingMapSortAliases.contains(named.canonicalized)
=>
+ // If we replace the top-level named expr, then should add back
the original name
+
groupingMapSortAliases(named.canonicalized).toAttribute.withName(named.name)
+ case other =>
+ // This must be top-down so distinct arguments are normalized
before their children.
+ // Continuing downward also substitutes grouping aliases in
other arguments.
+ other.transformDown {
+ case ae: AggregateExpression if ae.isDistinct =>
+ ae.copy(aggregateFunction =
ae.aggregateFunction.withNewChildren(
+ ae.aggregateFunction.children.map { child =>
+ distinctMapSortAliases.get(child.canonicalized)
+ .map(_.toAttribute).getOrElse(child)
+ }).asInstanceOf[AggregateFunction])
+ case e =>
+
groupingMapSortAliases.get(e.canonicalized).map(_.toAttribute).getOrElse(e)
+ }.asInstanceOf[NamedExpression]
+ }
+ val distinctInput = if (distinctInputAliases.nonEmpty) {
+ // Project complex inputs once because recursive normalization can
reference them
+ // repeatedly. The operator optimization batch later removes these
temporary projection
+ // layers with CollapseProject and ColumnPruning.
+ Project(child.output ++ distinctInputAliases.values, child)
} else {
- expr
+ child
}
+ val newChild = Project(
+ child.output ++ (groupingMapSortAliases ++
distinctMapSortAliases).values,
+ distinctInput)
+ val newAgg = Aggregate(newGroupingKeys, newAggregateExprs, newChild,
hint)
+ newAgg -> agg.output.zip(newAgg.output)
}
- val newAggregateExprs = aggregateExpressions.map {
- case named if exprToMapSort.contains(named.canonicalized) =>
- // If we replace the top-level named expr, then should add back
the original name
- exprToMapSort(named.canonicalized).toAttribute.withName(named.name)
- case other =>
- other.transformUp {
- case e =>
exprToMapSort.get(e.canonicalized).map(_.toAttribute).getOrElse(e)
- }.asInstanceOf[NamedExpression]
- }
- val newChild = Project(child.output ++ exprToMapSort.values, child)
- val newAgg = Aggregate(newGroupingKeys, newAggregateExprs, newChild,
hint)
- newAgg -> agg.output.zip(newAgg.output)
}
}
+
+ private def distinctAggregateChildren(
+ aggregateExpressions: Seq[NamedExpression]): Seq[Expression] = {
+ aggregateExpressions
+ .flatMap(_.collect {
+ case ae: AggregateExpression if ae.isDistinct => ae
+ })
+ .flatMap(_.aggregateFunction.children)
+ }
+
+ private def createDistinctInputAliases(
+ expressions: Seq[Expression]): VectorMap[Expression, NamedExpression] = {
+ val aliases = new mutable.LinkedHashMap[Expression, NamedExpression]
+ expressions.foreach {
+ case _: Attribute =>
+ case expr if mapTypeExistsRecursively(expr) =>
+ aliases.getOrElseUpdate(
+ expr.canonicalized,
+ Alias(expr, "_distinctaggregateexpression")())
+ case _ =>
+ }
+ VectorMap.from(aliases)
+ }
+
+ private def insertMapSortInExpressions(
+ expressions: Seq[Expression],
+ aliasName: String,
+ reusableAliases: Map[Expression, NamedExpression] = Map.empty,
+ inputAliases: Map[Expression, NamedExpression] = Map.empty)
+ : VectorMap[Expression, NamedExpression] = {
+ val aliases = new mutable.LinkedHashMap[Expression, NamedExpression]
+ expressions.foreach { expr =>
+ val canonicalized = expr.canonicalized
+ val input =
inputAliases.get(canonicalized).map(_.toAttribute).getOrElse(expr)
+ val inserted = insertMapSortRecursively(input)
+ if (input.ne(inserted)) {
+ aliases.getOrElseUpdate(
+ canonicalized,
+ reusableAliases.getOrElse(canonicalized, Alias(inserted,
aliasName)()))
+ }
+ }
+ VectorMap.from(aliases)
+ }
}
/**
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 25edd59412e9..b71ee3485de4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -355,10 +355,10 @@ abstract class Optimizer(catalogManager: CatalogManager)
NormalizeFloatingNumbers,
RewriteNonCorrelatedExists,
PullOutGroupingExpressions,
- // Put `InsertMapSortInGroupingExpressions` after
`PullOutGroupingExpressions`,
- // so the grouping keys can only be attribute and literal which makes
- // `InsertMapSortInGroupingExpressions` easy to insert `MapSort`.
- InsertMapSortInGroupingExpressions,
+ // Put `InsertMapSortInAggregate` after `PullOutGroupingExpressions`,
+ // so grouping keys are attributes or literals. The rule also projects
complex distinct
+ // aggregate arguments before inserting `MapSort`.
+ InsertMapSortInAggregate,
InsertMapSortInRepartitionExpressions,
ComputeCurrentTime,
ReplaceCurrentLike(catalogManager),
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 16175a1a8c18..4f1643d4725b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3160,6 +3160,17 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val INSERT_MAP_SORT_IN_DISTINCT_AGGREGATES_ENABLED =
+ buildConf("spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled")
+ .internal()
+ .doc("When true, map-typed arguments of distinct aggregates are
normalized with MapSort. " +
+ "When false, MapSort is not added solely for distinct aggregate
arguments; arguments " +
+ "that are also grouping expressions remain normalized.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
val OPTIMIZE_EXPAND_RATIO =
buildConf("spark.sql.optimizer.optimizeExpandRatio")
.internal()
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInAggregateSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInAggregateSuite.scala
new file mode 100644
index 000000000000..fa2340318cbc
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInAggregateSuite.scala
@@ -0,0 +1,117 @@
+/*
+ * 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.optimizer
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions.{Alias, MapSort}
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical.{
+ Aggregate, LocalRelation, LogicalPlan, Project, Union}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{IntegerType, StringType}
+
+class InsertMapSortInAggregateSuite extends PlanTest {
+ private val input = LocalRelation(Symbol("m").map(StringType, IntegerType))
+ private val mapAttribute = input.output.head
+
+ private def aliasesNamed(plan: LogicalPlan, name: String): Seq[Alias] = {
+ plan.flatMap { node =>
+ node.expressions.flatMap(_.collect {
+ case alias @ Alias(_, aliasName) if aliasName == name => alias
+ })
+ }
+ }
+
+ test("reuse map sort when a grouping key is also a distinct argument") {
+ val plan = Aggregate(
+ Seq(mapAttribute),
+ Seq(mapAttribute, countDistinct(mapAttribute).as("count")),
+ input)
+ val rewritten = InsertMapSortInAggregate(plan)
+ val groupingAliases = aliasesNamed(rewritten, "_groupingmapsort")
+
+ assert(groupingAliases.size == 1)
+ assert(groupingAliases.head.child.isInstanceOf[MapSort])
+ rewritten match {
+ case Aggregate(Seq(groupingExpression), aggregateExpressions, _:
Project, _) =>
+
assert(groupingExpression.semanticEquals(groupingAliases.head.toAttribute))
+ val distinctChildren = aggregateExpressions.flatMap(_.collect {
+ case expression: AggregateExpression if expression.isDistinct =>
+ expression.aggregateFunction.children
+ }).flatten
+ assert(distinctChildren.size == 1)
+
assert(distinctChildren.head.semanticEquals(groupingAliases.head.toAttribute))
+ case other =>
+ fail(s"Unexpected plan:\n$other")
+ }
+ }
+
+ test("project complex distinct arguments only when needed") {
+ val attributePlan = Aggregate(
+ Nil,
+ Seq(countDistinct(mapAttribute).as("count")),
+ input)
+ val complexPlan = Aggregate(
+ Nil,
+ Seq(countDistinct(namedStruct("m", mapAttribute)).as("count")),
+ input)
+
+ val rewrittenAttributePlan = InsertMapSortInAggregate(attributePlan)
+ assert(rewrittenAttributePlan.collect { case _: Project => 1 }.size == 1)
+ assert(aliasesNamed(rewrittenAttributePlan,
"_distinctaggregateexpression").isEmpty)
+ assert(aliasesNamed(rewrittenAttributePlan, "_distinctmapsort").size == 1)
+
+ val rewrittenComplexPlan = InsertMapSortInAggregate(complexPlan)
+ assert(rewrittenComplexPlan.collect { case _: Project => 1 }.size == 2)
+ assert(aliasesNamed(rewrittenComplexPlan,
"_distinctaggregateexpression").size == 1)
+ assert(aliasesNamed(rewrittenComplexPlan, "_distinctmapsort").size == 1)
+ }
+
+ test("skip distinct argument normalization when disabled") {
+ val plan = Aggregate(
+ Nil,
+ Seq(countDistinct(mapAttribute).as("count")),
+ input)
+
+ withSQLConf(SQLConf.INSERT_MAP_SORT_IN_DISTINCT_AGGREGATES_ENABLED.key ->
"false") {
+ comparePlans(InsertMapSortInAggregate(plan), plan)
+ }
+ }
+
+ test("leave map-free aggregates untouched when another aggregate needs
rewriting") {
+ val scalarInput = LocalRelation(Symbol("i").int)
+ val scalarAggregate = Aggregate(
+ Nil,
+ Seq(count(scalarInput.output.head).as("count")),
+ scalarInput)
+ val mapAggregate = Aggregate(
+ Nil,
+ Seq(countDistinct(mapAttribute).as("count")),
+ input)
+
+ InsertMapSortInAggregate(Union(Seq(scalarAggregate, mapAggregate))) match {
+ case Union(Seq(rewrittenScalarAggregate, rewrittenMapAggregate), _, _) =>
+ comparePlans(rewrittenScalarAggregate, scalarAggregate)
+ assert(rewrittenScalarAggregate.collect { case _: Project => 1
}.isEmpty)
+ assert(rewrittenMapAggregate.collect { case _: Project => 1 }.size ==
1)
+ case other =>
+ fail(s"Unexpected plan:\n$other")
+ }
+ }
+}
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/distinct-map-aggregates.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/distinct-map-aggregates.sql.out
new file mode 100644
index 000000000000..50f9153b7e3f
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/distinct-map-aggregates.sql.out
@@ -0,0 +1,363 @@
+-- Automatically generated by SQLQueryTestSuite
+-- !query
+CREATE OR REPLACE TEMPORARY VIEW distinct_map_data AS SELECT * FROM VALUES
+ (2, map('a', 1, 'b', 2), 1, true),
+ (2, map('b', 2, 'a', 1), 1, true),
+ (1, map('a', 1, 'b', 2), 1, true),
+ (1, map('a', 3), 2, false)
+AS distinct_map_data(g, m, id, should_keep)
+-- !query analysis
+CreateViewCommand `distinct_map_data`, SELECT * FROM VALUES
+ (2, map('a', 1, 'b', 2), 1, true),
+ (2, map('b', 2, 'a', 1), 1, true),
+ (1, map('a', 1, 'b', 2), 1, true),
+ (1, map('a', 3), 2, false)
+AS distinct_map_data(g, m, id, should_keep), false, true, LocalTempView,
UNSUPPORTED, true
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT m) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct m#x) AS count(DISTINCT m)#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT SIZE(COLLECT_LIST(DISTINCT m)) FROM distinct_map_data
+-- !query analysis
+Aggregate [size(collect_list(distinct m#x, 0, 0, true), false) AS
size(collect_list(DISTINCT m))#x]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT map_entries(m)
+FROM (
+ SELECT EXPLODE(COLLECT_LIST(DISTINCT m)) AS m
+ FROM distinct_map_data
+) AS collected_maps
+ORDER BY element_at(m, 'a')
+-- !query analysis
+Project [map_entries(m)#x]
++- Sort [element_at(m#x, a, None, true) ASC NULLS FIRST], true
+ +- Project [map_entries(m#x) AS map_entries(m)#x, m#x]
+ +- SubqueryAlias collected_maps
+ +- Project [m#x]
+ +- Generate explode(_gen_input_0#x), false, [m#x]
+ +- Aggregate [collect_list(distinct m#x, 0, 0, true) AS
_gen_input_0#x]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x,
should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as
map<string,int>) AS m#x, cast(id#x as int) AS id#x, cast(should_keep#x as
boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x,
should_keep#x]
+
+
+-- !query
+SELECT map_entries(FIRST(DISTINCT m)), map_entries(LAST(DISTINCT m)),
COUNT(DISTINCT m)
+FROM VALUES (map('b', 2, 'a', 1)) AS single_map_data(m)
+-- !query analysis
+Aggregate [map_entries(first(distinct m#x, false)) AS
map_entries(first(DISTINCT m))#x, map_entries(last(distinct m#x, false)) AS
map_entries(last(DISTINCT m))#x, count(distinct m#x) AS count(DISTINCT m)#xL]
++- SubqueryAlias single_map_data
+ +- LocalRelation [m#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT m, id) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct m#x, id#x) AS count(DISTINCT m, id)#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT m), COUNT(DISTINCT id) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct m#x) AS count(DISTINCT m)#xL, count(distinct id#x)
AS count(DISTINCT id)#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT g, COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY g
+ORDER BY g
+-- !query analysis
+Sort [g#x ASC NULLS FIRST], true
++- Aggregate [g#x], [g#x, count(distinct m#x) AS count(DISTINCT m)#xL]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT m, COUNT(DISTINCT m), COLLECT_LIST(DISTINCT m)
+FROM distinct_map_data
+GROUP BY m
+ORDER BY element_at(m, 'a')
+-- !query analysis
+Sort [element_at(m#x, a, None, true) ASC NULLS FIRST], true
++- Aggregate [m#x], [m#x, count(distinct m#x) AS count(DISTINCT m)#xL,
collect_list(distinct m#x, 0, 0, true) AS collect_list(DISTINCT m)#x]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT m) FILTER (WHERE should_keep) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct m#x) FILTER (WHERE should_keep#x) AS count(DISTINCT
m) FILTER (WHERE should_keep)#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT MAX(map_values(m)[0])
+FROM distinct_map_data
+WHERE id = 1
+-- !query analysis
+Aggregate [max(map_values(m#x)[0]) AS max(map_values(m)[0])#x]
++- Filter (id#x = 1)
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT MAX(map_values(m)[0]), COUNT(DISTINCT m)
+FROM distinct_map_data
+WHERE id = 1
+-- !query analysis
+Aggregate [max(map_values(m#x)[0]) AS max(map_values(m)[0])#x, count(distinct
m#x) AS count(DISTINCT m)#xL]
++- Filter (id#x = 1)
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT g
+FROM distinct_map_data
+GROUP BY g
+ORDER BY COUNT(DISTINCT m), g
+-- !query analysis
+Project [g#x]
++- Sort [count(DISTINCT m)#xL ASC NULLS FIRST, g#x ASC NULLS FIRST], true
+ +- Aggregate [g#x], [g#x, count(distinct m#x) AS count(DISTINCT m)#xL]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>)
AS m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS
should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT g
+FROM distinct_map_data
+GROUP BY g
+HAVING COUNT(DISTINCT m) = 1
+ORDER BY g
+-- !query analysis
+Sort [g#x ASC NULLS FIRST], true
++- Project [g#x]
+ +- Filter (count(DISTINCT m)#xL = cast(1 as bigint))
+ +- Aggregate [g#x], [g#x, count(distinct m#x) AS count(DISTINCT m)#xL]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as
map<string,int>) AS m#x, cast(id#x as int) AS id#x, cast(should_keep#x as
boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT named_struct('m', m)) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct named_struct(m, m#x)) AS count(DISTINCT
named_struct(m, m))#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT array(m)) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct array(m#x)) AS count(DISTINCT array(m))#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT map('m', m)) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct map(m, m#x)) AS count(DISTINCT map(m, m))#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT m), COLLECT_LIST(DISTINCT m)
+FROM VALUES
+ (CAST(map() AS MAP<STRING, INT>)),
+ (CAST(map() AS MAP<STRING, INT>)),
+ (CAST(NULL AS MAP<STRING, INT>))
+AS null_and_empty_map_data(m)
+-- !query analysis
+Aggregate [count(distinct m#x) AS count(DISTINCT m)#xL, collect_list(distinct
m#x, 0, 0, true) AS collect_list(DISTINCT m)#x]
++- SubqueryAlias null_and_empty_map_data
+ +- LocalRelation [m#x]
+
+
+-- !query
+SELECT g, GROUPING(g), COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY GROUPING SETS ((g), ())
+ORDER BY GROUPING(g), g
+-- !query analysis
+Sort [grouping(g)#x ASC NULLS FIRST, g#x ASC NULLS FIRST], true
++- Aggregate [g#x, spark_grouping_id#xL], [g#x,
cast((shiftright(spark_grouping_id#xL, 0) & 1) as tinyint) AS grouping(g)#x,
count(distinct m#x) AS count(DISTINCT m)#xL]
+ +- Expand [[g#x, m#x, id#x, should_keep#x, g#x, 0], [g#x, m#x, id#x,
should_keep#x, null, 1]], [g#x, m#x, id#x, should_keep#x, g#x,
spark_grouping_id#xL]
+ +- Project [g#x, m#x, id#x, should_keep#x, g#x AS g#x]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as
map<string,int>) AS m#x, cast(id#x as int) AS id#x, cast(should_keep#x as
boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT named_struct('m', m, 'n', n))
+FROM VALUES
+ (map('a', 1, 'b', 2), map('x', 1, 'y', 2)),
+ (map('b', 2, 'a', 1), map('y', 2, 'x', 1))
+AS grouped_distinct_map_data(m, n)
+GROUP BY m
+-- !query analysis
+Aggregate [m#x], [count(distinct named_struct(m, m#x, n, n#x)) AS
count(DISTINCT named_struct(m, m, n, n))#xL]
++- SubqueryAlias grouped_distinct_map_data
+ +- LocalRelation [m#x, n#x]
+
+
+-- !query
+SET spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled=false
+-- !query analysis
+SetCommand
(spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled,Some(false))
+
+
+-- !query
+SELECT COUNT(DISTINCT m) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct m#x) AS count(DISTINCT m)#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT map_entries(m)
+FROM (
+ SELECT EXPLODE(COLLECT_LIST(DISTINCT m)) AS m
+ FROM distinct_map_data
+) AS collected_maps
+ORDER BY element_at(m, 'a'), map_entries(m)[0].key
+-- !query analysis
+Project [map_entries(m)#x]
++- Sort [element_at(m#x, a, None, true) ASC NULLS FIRST,
map_entries(m#x)[0].key ASC NULLS FIRST], true
+ +- Project [map_entries(m#x) AS map_entries(m)#x, m#x]
+ +- SubqueryAlias collected_maps
+ +- Project [m#x]
+ +- Generate explode(_gen_input_0#x), false, [m#x]
+ +- Aggregate [collect_list(distinct m#x, 0, 0, true) AS
_gen_input_0#x]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x,
should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as
map<string,int>) AS m#x, cast(id#x as int) AS id#x, cast(should_keep#x as
boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x,
should_keep#x]
+
+
+-- !query
+SELECT COUNT(DISTINCT named_struct('m', m)) FROM distinct_map_data
+-- !query analysis
+Aggregate [count(distinct named_struct(m, m#x)) AS count(DISTINCT
named_struct(m, m))#xL]
++- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SELECT m, COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY m
+ORDER BY element_at(m, 'a')
+-- !query analysis
+Sort [element_at(m#x, a, None, true) ASC NULLS FIRST], true
++- Aggregate [m#x], [m#x, count(distinct m#x) AS count(DISTINCT m)#xL]
+ +- SubqueryAlias distinct_map_data
+ +- View (`distinct_map_data`, [g#x, m#x, id#x, should_keep#x])
+ +- Project [cast(g#x as int) AS g#x, cast(m#x as map<string,int>) AS
m#x, cast(id#x as int) AS id#x, cast(should_keep#x as boolean) AS should_keep#x]
+ +- Project [g#x, m#x, id#x, should_keep#x]
+ +- SubqueryAlias distinct_map_data
+ +- LocalRelation [g#x, m#x, id#x, should_keep#x]
+
+
+-- !query
+SET spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled=true
+-- !query analysis
+SetCommand
(spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled,Some(true))
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/distinct-map-aggregates.sql
b/sql/core/src/test/resources/sql-tests/inputs/distinct-map-aggregates.sql
new file mode 100644
index 000000000000..523ed0caee89
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/inputs/distinct-map-aggregates.sql
@@ -0,0 +1,102 @@
+-- Test DISTINCT aggregates with MapType arguments.
+
+CREATE OR REPLACE TEMPORARY VIEW distinct_map_data AS SELECT * FROM VALUES
+ (2, map('a', 1, 'b', 2), 1, true),
+ (2, map('b', 2, 'a', 1), 1, true),
+ (1, map('a', 1, 'b', 2), 1, true),
+ (1, map('a', 3), 2, false)
+AS distinct_map_data(g, m, id, should_keep);
+
+SELECT COUNT(DISTINCT m) FROM distinct_map_data;
+
+SELECT SIZE(COLLECT_LIST(DISTINCT m)) FROM distinct_map_data;
+
+SELECT map_entries(m)
+FROM (
+ SELECT EXPLODE(COLLECT_LIST(DISTINCT m)) AS m
+ FROM distinct_map_data
+) AS collected_maps
+ORDER BY element_at(m, 'a');
+
+SELECT map_entries(FIRST(DISTINCT m)), map_entries(LAST(DISTINCT m)),
COUNT(DISTINCT m)
+FROM VALUES (map('b', 2, 'a', 1)) AS single_map_data(m);
+
+SELECT COUNT(DISTINCT m, id) FROM distinct_map_data;
+
+SELECT COUNT(DISTINCT m), COUNT(DISTINCT id) FROM distinct_map_data;
+
+SELECT g, COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY g
+ORDER BY g;
+
+SELECT m, COUNT(DISTINCT m), COLLECT_LIST(DISTINCT m)
+FROM distinct_map_data
+GROUP BY m
+ORDER BY element_at(m, 'a');
+
+SELECT COUNT(DISTINCT m) FILTER (WHERE should_keep) FROM distinct_map_data;
+
+SELECT MAX(map_values(m)[0])
+FROM distinct_map_data
+WHERE id = 1;
+
+SELECT MAX(map_values(m)[0]), COUNT(DISTINCT m)
+FROM distinct_map_data
+WHERE id = 1;
+
+SELECT g
+FROM distinct_map_data
+GROUP BY g
+ORDER BY COUNT(DISTINCT m), g;
+
+SELECT g
+FROM distinct_map_data
+GROUP BY g
+HAVING COUNT(DISTINCT m) = 1
+ORDER BY g;
+
+SELECT COUNT(DISTINCT named_struct('m', m)) FROM distinct_map_data;
+
+SELECT COUNT(DISTINCT array(m)) FROM distinct_map_data;
+
+SELECT COUNT(DISTINCT map('m', m)) FROM distinct_map_data;
+
+SELECT COUNT(DISTINCT m), COLLECT_LIST(DISTINCT m)
+FROM VALUES
+ (CAST(map() AS MAP<STRING, INT>)),
+ (CAST(map() AS MAP<STRING, INT>)),
+ (CAST(NULL AS MAP<STRING, INT>))
+AS null_and_empty_map_data(m);
+
+SELECT g, GROUPING(g), COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY GROUPING SETS ((g), ())
+ORDER BY GROUPING(g), g;
+
+SELECT COUNT(DISTINCT named_struct('m', m, 'n', n))
+FROM VALUES
+ (map('a', 1, 'b', 2), map('x', 1, 'y', 2)),
+ (map('b', 2, 'a', 1), map('y', 2, 'x', 1))
+AS grouped_distinct_map_data(m, n)
+GROUP BY m;
+
+SET spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled=false;
+
+SELECT COUNT(DISTINCT m) FROM distinct_map_data;
+
+SELECT map_entries(m)
+FROM (
+ SELECT EXPLODE(COLLECT_LIST(DISTINCT m)) AS m
+ FROM distinct_map_data
+) AS collected_maps
+ORDER BY element_at(m, 'a'), map_entries(m)[0].key;
+
+SELECT COUNT(DISTINCT named_struct('m', m)) FROM distinct_map_data;
+
+SELECT m, COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY m
+ORDER BY element_at(m, 'a');
+
+SET spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled=true;
diff --git
a/sql/core/src/test/resources/sql-tests/results/distinct-map-aggregates.sql.out
b/sql/core/src/test/resources/sql-tests/results/distinct-map-aggregates.sql.out
new file mode 100644
index 000000000000..57f53db7e3f5
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/results/distinct-map-aggregates.sql.out
@@ -0,0 +1,265 @@
+-- Automatically generated by SQLQueryTestSuite
+-- !query
+CREATE OR REPLACE TEMPORARY VIEW distinct_map_data AS SELECT * FROM VALUES
+ (2, map('a', 1, 'b', 2), 1, true),
+ (2, map('b', 2, 'a', 1), 1, true),
+ (1, map('a', 1, 'b', 2), 1, true),
+ (1, map('a', 3), 2, false)
+AS distinct_map_data(g, m, id, should_keep)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT COUNT(DISTINCT m) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT m):bigint>
+-- !query output
+2
+
+
+-- !query
+SELECT SIZE(COLLECT_LIST(DISTINCT m)) FROM distinct_map_data
+-- !query schema
+struct<size(collect_list(DISTINCT m)):int>
+-- !query output
+2
+
+
+-- !query
+SELECT map_entries(m)
+FROM (
+ SELECT EXPLODE(COLLECT_LIST(DISTINCT m)) AS m
+ FROM distinct_map_data
+) AS collected_maps
+ORDER BY element_at(m, 'a')
+-- !query schema
+struct<map_entries(m):array<struct<key:string,value:int>>>
+-- !query output
+[{"key":"a","value":1},{"key":"b","value":2}]
+[{"key":"a","value":3}]
+
+
+-- !query
+SELECT map_entries(FIRST(DISTINCT m)), map_entries(LAST(DISTINCT m)),
COUNT(DISTINCT m)
+FROM VALUES (map('b', 2, 'a', 1)) AS single_map_data(m)
+-- !query schema
+struct<map_entries(first(DISTINCT
m)):array<struct<key:string,value:int>>,map_entries(last(DISTINCT
m)):array<struct<key:string,value:int>>,count(DISTINCT m):bigint>
+-- !query output
+[{"key":"a","value":1},{"key":"b","value":2}]
[{"key":"a","value":1},{"key":"b","value":2}] 1
+
+
+-- !query
+SELECT COUNT(DISTINCT m, id) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT m, id):bigint>
+-- !query output
+2
+
+
+-- !query
+SELECT COUNT(DISTINCT m), COUNT(DISTINCT id) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT m):bigint,count(DISTINCT id):bigint>
+-- !query output
+2 2
+
+
+-- !query
+SELECT g, COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY g
+ORDER BY g
+-- !query schema
+struct<g:int,count(DISTINCT m):bigint>
+-- !query output
+1 2
+2 1
+
+
+-- !query
+SELECT m, COUNT(DISTINCT m), COLLECT_LIST(DISTINCT m)
+FROM distinct_map_data
+GROUP BY m
+ORDER BY element_at(m, 'a')
+-- !query schema
+struct<m:map<string,int>,count(DISTINCT m):bigint,collect_list(DISTINCT
m):array<map<string,int>>>
+-- !query output
+{"a":1,"b":2} 1 [{"a":1,"b":2}]
+{"a":3} 1 [{"a":3}]
+
+
+-- !query
+SELECT COUNT(DISTINCT m) FILTER (WHERE should_keep) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT m) FILTER (WHERE should_keep):bigint>
+-- !query output
+1
+
+
+-- !query
+SELECT MAX(map_values(m)[0])
+FROM distinct_map_data
+WHERE id = 1
+-- !query schema
+struct<max(map_values(m)[0]):int>
+-- !query output
+2
+
+
+-- !query
+SELECT MAX(map_values(m)[0]), COUNT(DISTINCT m)
+FROM distinct_map_data
+WHERE id = 1
+-- !query schema
+struct<max(map_values(m)[0]):int,count(DISTINCT m):bigint>
+-- !query output
+2 1
+
+
+-- !query
+SELECT g
+FROM distinct_map_data
+GROUP BY g
+ORDER BY COUNT(DISTINCT m), g
+-- !query schema
+struct<g:int>
+-- !query output
+2
+1
+
+
+-- !query
+SELECT g
+FROM distinct_map_data
+GROUP BY g
+HAVING COUNT(DISTINCT m) = 1
+ORDER BY g
+-- !query schema
+struct<g:int>
+-- !query output
+2
+
+
+-- !query
+SELECT COUNT(DISTINCT named_struct('m', m)) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT named_struct(m, m)):bigint>
+-- !query output
+2
+
+
+-- !query
+SELECT COUNT(DISTINCT array(m)) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT array(m)):bigint>
+-- !query output
+2
+
+
+-- !query
+SELECT COUNT(DISTINCT map('m', m)) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT map(m, m)):bigint>
+-- !query output
+2
+
+
+-- !query
+SELECT COUNT(DISTINCT m), COLLECT_LIST(DISTINCT m)
+FROM VALUES
+ (CAST(map() AS MAP<STRING, INT>)),
+ (CAST(map() AS MAP<STRING, INT>)),
+ (CAST(NULL AS MAP<STRING, INT>))
+AS null_and_empty_map_data(m)
+-- !query schema
+struct<count(DISTINCT m):bigint,collect_list(DISTINCT
m):array<map<string,int>>>
+-- !query output
+1 [{}]
+
+
+-- !query
+SELECT g, GROUPING(g), COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY GROUPING SETS ((g), ())
+ORDER BY GROUPING(g), g
+-- !query schema
+struct<g:int,grouping(g):tinyint,count(DISTINCT m):bigint>
+-- !query output
+1 0 2
+2 0 1
+NULL 1 2
+
+
+-- !query
+SELECT COUNT(DISTINCT named_struct('m', m, 'n', n))
+FROM VALUES
+ (map('a', 1, 'b', 2), map('x', 1, 'y', 2)),
+ (map('b', 2, 'a', 1), map('y', 2, 'x', 1))
+AS grouped_distinct_map_data(m, n)
+GROUP BY m
+-- !query schema
+struct<count(DISTINCT named_struct(m, m, n, n)):bigint>
+-- !query output
+1
+
+
+-- !query
+SET spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled=false
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled false
+
+
+-- !query
+SELECT COUNT(DISTINCT m) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT m):bigint>
+-- !query output
+3
+
+
+-- !query
+SELECT map_entries(m)
+FROM (
+ SELECT EXPLODE(COLLECT_LIST(DISTINCT m)) AS m
+ FROM distinct_map_data
+) AS collected_maps
+ORDER BY element_at(m, 'a'), map_entries(m)[0].key
+-- !query schema
+struct<map_entries(m):array<struct<key:string,value:int>>>
+-- !query output
+[{"key":"a","value":1},{"key":"b","value":2}]
+[{"key":"b","value":2},{"key":"a","value":1}]
+[{"key":"a","value":3}]
+
+
+-- !query
+SELECT COUNT(DISTINCT named_struct('m', m)) FROM distinct_map_data
+-- !query schema
+struct<count(DISTINCT named_struct(m, m)):bigint>
+-- !query output
+3
+
+
+-- !query
+SELECT m, COUNT(DISTINCT m)
+FROM distinct_map_data
+GROUP BY m
+ORDER BY element_at(m, 'a')
+-- !query schema
+struct<m:map<string,int>,count(DISTINCT m):bigint>
+-- !query output
+{"a":1,"b":2} 1
+{"a":3} 1
+
+
+-- !query
+SET spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled=true
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled true
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
