[LIVY-988] Support for escaping backtick from spark-submit arguments

server/src/main/scala/org/apache/livy/LivyConf.scala

@@ -267,6 +267,9 @@ object LivyConf {
 
   val SESSION_ALLOW_CUSTOM_CLASSPATH = Entry("livy.server.session.allow-custom-classpath", false)
 
+  // LIVY-988: Escape backtick from spark-submit arguments
+  val SPARK_SUBMIT_ESCAPE_BACKTICKS = Entry("livy.server.escapeBackTicks", false)
+
   val SPARK_MASTER = "spark.master"
   val SPARK_DEPLOY_MODE = "spark.submit.deployMode"
   val SPARK_JARS = "spark.jars"

server/src/main/scala/org/apache/livy/utils/SparkProcessBuilder.scala

@@ -154,6 +154,16 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
     this
   }
 
+  /**
+   * Return the arg after escaping backticks, it also unescape the backticks first
+   * then escape it again to avoid double escaping
+   * @param arg
+   */
+  def escapeBackTicks(arg: String): String = {
+    val unescapedArg = arg.replace("\\`", "`")
+    unescapedArg.replace("`", "\\`")
+  }
+
   def start(file: Option[String], args: Traversable[String]): LineBufferedProcess = {
     var arguments = ArrayBuffer(_executable)
 
@@ -195,6 +205,11 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
     arguments += file.getOrElse("spark-internal")
     arguments ++= args
 
+    if (livyConf.getBoolean(LivyConf.SPARK_SUBMIT_ESCAPE_BACKTICKS)) {
+      debug("Escaping backticks if present in spark-submit arguments")
+      arguments = arguments.map(escapeBackTicks(_))
+    }
+
     val argsString = arguments
       .map("'" + _.replace("'", "\\'") + "'")
       .mkString(" ")

server/src/test/scala/org/apache/livy/utils/SparkProcessBuilderSuite.scala

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.livy.utils
+
+import org.apache.livy.{LivyBaseUnitTestSuite, LivyConf}
+
+import org.scalatest.{BeforeAndAfter, FunSpec}
+
+class SparkProcessBuildSuite
+  extends FunSpec
+    with BeforeAndAfter
+    with org.scalatest.Matchers
+    with LivyBaseUnitTestSuite {
+
+  // LIVY-998: Escape backtick from spark-submit arguments
+  it("should only escape unescaped backticks") {
+    val livyConf = new LivyConf
+    livyConf.set(LivyConf.SPARK_SUBMIT_ESCAPE_BACKTICKS, true)
+
+    val builder = new SparkProcessBuilder(livyConf)
+
+    assert(builder.escapeBackTicks("test_db.test_table") == "test_db.test_table")
+    assert(builder.escapeBackTicks("test_db.`test_table`") == "test_db.\\`test_table\\`")
+    assert(builder.escapeBackTicks("test_db.\\`test_table\\`") == "test_db.\\`test_table\\`")
+  }
+}