[CARMEL-7225] [CARMEL-3474] BuiltIn UDF java_method() is dangerous (apache#51)

Author: wangyum

common/utils/pom.xml

@@ -32,7 +32,7 @@
   <url>https://spark.apache.org/</url>
   <properties>
     <sbt.project.name>common-utils</sbt.project.name>
-    <cluster>analytics</cluster> <!-- ETL/analytics -->
+    <cluster></cluster> <!-- ETL/analytics -->
   </properties>
 
   <dependencies>

core/src/main/scala/org/apache/spark/SparkConf.scala

@@ -589,6 +589,7 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging with Seria
 private[spark] object SparkConf extends Logging {
 
   private[spark] val isETLCluster = CLUSTER.toLowerCase(Locale.ROOT).equals("etl")
+  private[spark] val isAnalyticsCluster = CLUSTER.toLowerCase(Locale.ROOT).equals("analytics")
 
   /**
    * Maps deprecated config keys to information about the deprecation.

sql/api/src/main/scala/org/apache/spark/sql/catalyst/analysis/noSuchItemsExceptions.scala

@@ -198,6 +198,11 @@ class NoSuchFunctionException private(
 class NoSuchTempFunctionException(func: String)
   extends AnalysisException(errorClass = "ROUTINE_NOT_FOUND", Map("routineName" -> s"`$func`"))
 
+class BannedFunctionException(func: String)
+  extends AnalysisException(
+    s"Banned function: function: '$func' is banned in Carmel SparkSQL to " +
+      s"avoid potential risk, please contact Carmel support for more info.")
+
 // any changes to this class should be backward compatible as it may be used by external connectors
 class NoSuchIndexException private(
     message: String,

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

@@ -2059,6 +2059,12 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
       val externalFunctionNameSet = new mutable.HashSet[Seq[String]]()
 
       plan.resolveExpressionsWithPruning(_.containsAnyPattern(UNRESOLVED_FUNCTION)) {
+        case f @ UnresolvedFunction(Seq(name), _, _, _, _)
+          if FunctionRegistry.bannedFunctionsForAnalytics
+            .contains(normalizeFuncName(f.nameParts).head) =>
+          withPosition(f) {
+            throw new BannedFunctionException(name)
+          }
         case f @ UnresolvedFunction(nameParts, _, _, _, _) =>
           if (ResolveFunctions.lookupBuiltinOrTempFunction(nameParts).isDefined) {
             f

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

@@ -23,6 +23,7 @@ import javax.annotation.concurrent.GuardedBy
 import scala.collection.mutable
 import scala.reflect.ClassTag
 
+import org.apache.spark.SparkConf
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.FunctionIdentifier
@@ -323,6 +324,12 @@ object FunctionRegistry {
 
   val FUNC_ALIAS = TreeNodeTag[String]("functionAliasName")
 
+  private[catalyst] val bannedFunctionsForAnalytics = if (SparkConf.isAnalyticsCluster) {
+    Seq("reflect", "java_method")
+  } else {
+    Seq.empty[String]
+  }
+
   // ==============================================================================================
   //                          The guideline for adding SQL functions
   // ==============================================================================================
@@ -829,7 +836,7 @@ object FunctionRegistry {
     expression[CsvToStructs]("from_csv"),
     expression[SchemaOfCsv]("schema_of_csv"),
     expression[StructsToCsv]("to_csv")
-  )
+  ).filterNot { case (k, _) => bannedFunctionsForAnalytics.contains(k) }
 
   val builtin: SimpleFunctionRegistry = {
     val fr = new SimpleFunctionRegistry