[SC-5628] Make sure commit protocol confs can be set via spark.*

## What changes were proposed in this pull request?

Spark confs need to be prefixed by spark.* in order to be set via the config UI.

## How was this patch tested?

Tested manually.

Author: Eric Liang <[email protected]>

Closes apache#177 from ericl/fix-conf.

Author: ericl

sql/core/src/main/scala/com/databricks/sql/transaction/DatabricksAtomicReadProtocol.scala

@@ -51,7 +51,7 @@ object DatabricksAtomicReadProtocol extends Logging {
   def filterDirectoryListing(
       fs: FileSystem, dir: Path, initialFiles: Seq[FileStatus]): Seq[FileStatus] = {
     // we use SparkEnv for this escape-hatch flag since this may be called on executors
-    if (!SparkEnv.get.conf.getBoolean("com.databricks.sql.enableFilterUncommitted", true)) {
+    if (!SparkEnv.get.conf.getBoolean("spark.databricks.sql.enableFilterUncommitted", true)) {
       return initialFiles
     }
 
@@ -151,7 +151,7 @@ object DatabricksAtomicReadProtocol extends Logging {
 
     // Optimization: can assume the list request was atomic if the files have not changed recently.
     val horizonMillis = SparkEnv.get.conf.getLong(
-      "com.databricks.sql.writeReorderingHorizonMillis", 5 * 60 * 1000)
+      "spark.databricks.sql.writeReorderingHorizonMillis", 5 * 60 * 1000)
 
     if ((state.missingMarkers.nonEmpty || state.missingDataFiles.nonEmpty) &&
           state.lastModified > clock.getTimeMillis - horizonMillis) {
@@ -235,7 +235,7 @@ object DatabricksAtomicReadProtocol extends Logging {
             case NonFatal(e) =>
               // we use SparkEnv for this escape-hatch flag since this may be called on executors
               if (SparkEnv.get.conf.getBoolean(
-                  "com.databricks.sql.ignoreCorruptCommitMarkers", false)) {
+                  "spark.databricks.sql.ignoreCorruptCommitMarkers", false)) {
                 logWarning("Failed to read job commit marker: " + stat, e)
                 corruptCommitMarkers += txnId
               } else {

sql/core/src/test/scala/com/databricks/sql/transaction/DatabricksAtomicCommitProtocolSuite.scala

@@ -37,11 +37,11 @@ class DatabricksAtomicCommitProtocolSuite extends QueryTest with SharedSQLContex
       create(dir, "part-r-00001-tid-77777-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb-0_00003.csv")
       create(dir, "part-r-00001-tid-12345-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb-0_00003.csv")
       try {
-        SparkEnv.get.conf.set("com.databricks.sql.enableFilterUncommitted", "false")
+        SparkEnv.get.conf.set("spark.databricks.sql.enableFilterUncommitted", "false")
         assert(spark.read.csv(dir.getAbsolutePath).count == 2)
         assert(spark.read.csv(dir.getAbsolutePath).inputFiles.length == 2)
       } finally {
-        SparkEnv.get.conf.remove("com.databricks.sql.enableFilterUncommitted")
+        SparkEnv.get.conf.remove("spark.databricks.sql.enableFilterUncommitted")
       }
     }
   }
@@ -171,10 +171,10 @@ class DatabricksAtomicCommitProtocolSuite extends QueryTest with SharedSQLContex
       }
       assert(error.getMessage.contains("Failed to read job commit marker"))
       try {
-        SparkEnv.get.conf.set("com.databricks.sql.ignoreCorruptCommitMarkers", "true")
+        SparkEnv.get.conf.set("spark.databricks.sql.ignoreCorruptCommitMarkers", "true")
         assert(spark.read.csv(dir.getAbsolutePath).count == 1)
       } finally {
-        SparkEnv.get.conf.remove("com.databricks.sql.ignoreCorruptCommitMarkers")
+        SparkEnv.get.conf.remove("spark.databricks.sql.ignoreCorruptCommitMarkers")
       }
     }
   }