[CARMEL-5861] Limit the max size of statement and executePlan (#874)

Author: wakun

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

@@ -1088,6 +1088,13 @@ object SQLConf {
     .intConf
     .createWithDefault(200)
 
+  val THRIFTSERVER_UI_STRING_MAX_LENGTH =
+    buildConf("spark.sql.thriftserver.ui.string.maxLength")
+      .doc("The max length for statement and executePlan in LiveExecutionData")
+      .version("3.0.0")
+      .intConf
+      .createWithDefault(100 * 1024)
+
   val SQL_HIGH_CONSUME_ALERT_ENABLED = buildConf("spark.sql.high.consume.enabled")
     .doc("Enable sql high consume alert or not")
     .booleanConf

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala

@@ -46,6 +46,7 @@ private[thriftserver] class HiveThriftServer2Listener(
     live: Boolean = true,
     queryLogger: Option[QueryLogger] = None) extends SparkListener with Logging {
 
+  val stringMaxLength = sparkConf.get(SQLConf.THRIFTSERVER_UI_STRING_MAX_LENGTH)
   // Spark execution id to Hive operationId map.
   private val execIdToExecutionId = new ConcurrentHashMap[Long, String]()
   private val sessionList = new ConcurrentHashMap[String, LiveSessionData]()
@@ -257,7 +258,9 @@ private[thriftserver] class HiveThriftServer2Listener(
   }
 
   private def onOperationParsed(e: SparkListenerThriftServerOperationParsed): Unit = {
-    executionList.get(e.id).executePlan = e.executionPlan
+    val executionPlan =
+      e.executionPlan.substring(0, Math.min(e.executionPlan.length, stringMaxLength))
+    executionList.get(e.id).executePlan = executionPlan
     executionList.get(e.id).state = ExecutionState.COMPILED
     updateLiveStore(executionList.get(e.id))
   }
@@ -342,12 +345,13 @@ private[thriftserver] class HiveThriftServer2Listener(
   private def getOrCreateExecution(
     execId: String, statement: String,
     sessionId: String, startTimestamp: Long, userName: String): LiveExecutionData = {
+    val subStatement = statement.substring(0, Math.min(statement.length, stringMaxLength))
     queryLogger match {
       case Some(queryLogger) =>
         executionList.computeIfAbsent(execId,
           (_: String) => new LiveExecutionData(
             execId,
-            statement,
+            subStatement,
             sessionId,
             startTimestamp,
             userName,
@@ -357,7 +361,7 @@ private[thriftserver] class HiveThriftServer2Listener(
       executionList.computeIfAbsent(execId,
         (_: String) => new LiveExecutionData(
           execId,
-          statement,
+          subStatement,
           sessionId,
           startTimestamp,
           userName))