[CARMEL-7531][CARMEL-5095] Add limitation of user sql size (apache#204)

Author: wakun

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -4827,6 +4827,12 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val THRIFTSERVER_SQLSIZE_LIMIT =
+    buildConf("spark.sql.thriftserver.sqlSize.limit")
+      .doc("limitation of thrift server sql size")
+      .longConf
+      .createWithDefault(Long.MaxValue)
+
   val CONTAINS_JOIN_ENABLED = buildConf("spark.sql.planner.containsJoin.enabled")
     .internal()
     .doc(s"When true and '${ADAPTIVE_EXECUTION_ENABLED.key}' is true, " +

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/server/SparkSQLOperationManager.scala

@@ -30,6 +30,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.hive.HiveUtils
 import org.apache.spark.sql.hive.thriftserver._
+import org.apache.spark.sql.internal.SQLConf
 
 /**
  * Executes queries using Spark SQL, and maintains a list of handles to active queries.
@@ -55,6 +56,10 @@ private[thriftserver] class SparkSQLOperationManager()
     sqlContext.sparkContext.setLocalProperty(SparkHadoopUtil.HIVE_SESSION_ID,
       parentSession.getSessionHandle.getHandleIdentifier.toString)
     val conf = sqlContext.sessionState.conf
+    val sqlSizeLimit = conf.getConf(SQLConf.THRIFTSERVER_SQLSIZE_LIMIT)
+    require(statement.length < sqlSizeLimit,
+      s"User query size ${statement.length} exceeds threshold $sqlSizeLimit")
+
     val runInBackground = async && conf.getConf(HiveUtils.HIVE_THRIFT_SERVER_ASYNC)
     val operation = new SparkExecuteStatementOperation(
       sqlContext, parentSession, statement, confOverlay, runInBackground, queryTimeout)