[MAPR-30536] Spark SQL queries on Map column fails after upgrade (apa…

…che#211)

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.analysis
 
 import org.apache.spark.api.python.PythonEvalType
+import org.apache.spark.SparkContext
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.SubExprUtils._
@@ -403,10 +404,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.catalogString)
+            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.catalogString)
+            }
 
           case o if o.expressions.exists(!_.deterministic) &&
             !o.isInstanceOf[Project] && !o.isInstanceOf[Filter] &&