[SPARK-34819][SQL] MapType supports comparable semantics

[maropu]

What changes were proposed in this pull request?

This PR proposes to support comparable semantics for map types.

NOTE: This PR is the rework of #31967(@WangGuangxin)/#15970(@hvanhovell).

The approach of the PR is similar to NormalizeFloatingNumbers and it has the same restriction; in the plan optimizing phase, a new rule named NormalizeMaps inserts an expression SortMapKeys to make sure two maps having the same key value pairs but with different key ordering are equal (e.g., Map('a' -> 1, 'b' -> 2) should equal to Map('b' -> 2, 'a' -> 1). As for aggregates, this rule is applied in the physical planning phase because all the grouping exprs are not extracted during the logical phase (This is the same restriction with NormalizeFloatingNumbers).

The major differences from NormalizeFloatingNumbers are as follows;

• The rule covers all the binary comparisons (EqualTo, GreaterThan, ...) and In/InSet in a plan (NormalizeFloatingNumbers is applied only into the EqualTo comparison in a join plan, an equi-join).
• This rule does not apply normalize recursively and just adds a SortMapKeys expr just on each top-level expr (e.g., top-level grouping expr and left/right side expr of binary comparisons).
• This rule additionally handles SortOrders in sort-related plans.

For sorting map entries, I reused the array ordering logic (See: MapType.compare and CodegenContext.genComp) because keys and values in map entries follow the array format; it checks if key arrays in two maps are the same first, an then check if value arrays are the same.

NOTE: Adding duplicate SortMapKeys exprs in a binary comparison tree is a known issue; for example, in a query below, MapType's column, a, is sorted twice;

scala> Seq((Map(1->1), Map(1->2), Map(1->1))).toDF("a", "b", "c").write.saveAsTable("t")
scala> sql("select * from t where a = b and a = c").explain()
== Physical Plan ==
*(1) Filter ((sortmapkeys(a#35) = sortmapkeys(b#36)) AND (sortmapkeys(a#35) = sortmapkeys(c#37)))
+- FileScan parquet default.t[a#35,b#36,c#37] Batched: false, DataFilters: [(sortmapkeys(a#35) = sortmapkeys(b#36)), (sortmapkeys(a#35) = sortmapkeys(c#37))], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/maropu/Repositories/spark/spark-master/spark-warehouse/t], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:map<int,int>,b:map<int,int>,c:map<int,int>>

But, I don't have a smart idea to avoid it in this PR for now. Probably, I think common subexpression elimination in filter plans can solve it, but Spark does not have the optimization now. (Fro more details, see the previous @viirya PR: #30565).

Why are the changes needed?

To improve map usability.

Does this PR introduce any user-facing change?

Yes, a user can use map-typed data in GROUP BY, ORDER BY, and PARTITION BY in WINDOW clauses.

How was this patch tested?

Add unit tests.

[SparkQA]

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43083/

[SparkQA]

Test build #138562 has finished for PR 32552 at commit 33eb7c4.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43091/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43091/

[SparkQA]

Test build #138570 has finished for PR 32552 at commit f3f8019.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43094/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43094/

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43096/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43096/

[SparkQA]

Test build #138573 has finished for PR 32552 at commit 7d6ab65.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #138575 has finished for PR 32552 at commit 539a1e6.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43100/

[SparkQA]

Test build #138579 has finished for PR 32552 at commit d08942f.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43104/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43104/

[maropu]

cc: @hvanhovell @cloud-fan @viirya @WangGuangxin

[SparkQA]

Test build #140384 has finished for PR 32552 at commit 29dd475.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SortMapKeys(child: Expression) extends UnaryExpression with ExpectsInputTypes

[maropu]

retest this please

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44926/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44926/

[SparkQA]

Test build #140411 has finished for PR 32552 at commit 29dd475.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SortMapKeys(child: Expression) extends UnaryExpression with ExpectsInputTypes

[maropu]

retest this please

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45400/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45400/

[SparkQA]

Test build #140889 has finished for PR 32552 at commit 29dd475.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SortMapKeys(child: Expression) extends UnaryExpression with ExpectsInputTypes

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45636/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45636/

[SparkQA]

Test build #141123 has finished for PR 32552 at commit 3be3882.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

retest this please

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45655/

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45656/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45655/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45656/

[SparkQA]

Test build #141144 has finished for PR 32552 at commit 3be3882.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47311/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47311/

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

[rekbun]

Folks, what is the state of this PR? Do we expect to make progress on this?