[MAPR-30536] Spark SQL queries on Map column fails after upgrade (apache#211)

Author: avovchenko

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
+import org.apache.spark.SparkContext
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.catalog.SimpleCatalogRelation
 import org.apache.spark.sql.catalyst.expressions._
@@ -387,10 +388,13 @@ trait CheckAnalysis extends PredicateHelper {
           // TODO: although map type is not orderable, technically map type should be able to be
           // used in equality comparison, remove this type check once we support it.
           case o if mapColumnInSetOperation(o).isDefined =>
-            val mapCol = mapColumnInSetOperation(o).get
-            failAnalysis("Cannot have map type columns in DataFrame which calls " +
-              s"set operations(intersect, except, etc.), but the type of column ${mapCol.name} " +
-              "is " + mapCol.dataType.simpleString)
+            val conf = SparkContext.getOrCreate().getConf
+            if(!conf.getBoolean("spark.sql.allow.distinct.map", false)) {
+              val mapCol = mapColumnInSetOperation(o).get
+              failAnalysis("Cannot have map type columns in DataFrame which calls " +
+                s"set operations(intersect, except, etc.), but the type of column ${mapCol.name} " +
+                "is " + mapCol.dataType.simpleString)
+            }
 
           case s: SimpleCatalogRelation =>
             failAnalysis(