[SPARK-12504][SQL] [Backport-1.6] Masking credentials in the sql plan explain output for JDBC data sources.

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala

@@ -108,4 +108,9 @@ private[sql] case class JDBCRelation(
       .mode(if (overwrite) SaveMode.Overwrite else SaveMode.Append)
       .jdbc(url, table, properties)
   }
+
+  override def toString: String = {
+    // credentials should not be included in the plan output, table information is sufficient.
+    s"JDBCRelation(${table})"
+  }
 }

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -25,6 +25,8 @@ import org.h2.jdbc.JdbcSQLException
 import org.scalatest.BeforeAndAfter
 
 import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.execution.ExplainCommand
+import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.test.SharedSQLContext
 import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
@@ -499,4 +501,24 @@ class JDBCSuite extends SparkFunSuite with BeforeAndAfter with SharedSQLContext
     assert(rows(0).getAs[java.sql.Timestamp](2)
       === java.sql.Timestamp.valueOf("2002-02-20 11:22:33.543543"))
   }
+
+  test("test credentials in the properties are not in plan output") {
+    val df = sql("SELECT * FROM parts")
+    val explain = ExplainCommand(df.queryExecution.logical, extended = true)
+    sqlContext.executePlan(explain).executedPlan.executeCollect().foreach {
+      r => assert(!List("testPass", "testUser").exists(r.toString.contains))
+    }
+    // test the JdbcRelation toString output
+    df.queryExecution.analyzed.collect {
+      case r: LogicalRelation => assert(r.relation.toString == "JDBCRelation(TEST.PEOPLE)")
+    }
+  }
+
+  test("test credentials in the connection url are not in the plan output") {
+    val df = sqlContext.read.jdbc(urlWithUserAndPass, "TEST.PEOPLE", new Properties)
+    val explain = ExplainCommand(df.queryExecution.logical, extended = true)
+    sqlContext.executePlan(explain).executedPlan.executeCollect().foreach {
+      r => assert(!List("testPass", "testUser").exists(r.toString.contains))
+    }
+  }
 }