You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by li...@apache.org on 2017/08/11 19:03:41 UTC
spark git commit: [SPARK-21519][SQL] Add an option to the JDBC data
source to initialize the target DB environment
Repository: spark
Updated Branches:
refs/heads/master 2387f1e31 -> 0377338bf
[SPARK-21519][SQL] Add an option to the JDBC data source to initialize the target DB environment
Add an option to the JDBC data source to initialize the environment of the remote database session
## What changes were proposed in this pull request?
This proposes an option to the JDBC datasource, tentatively called " sessionInitStatement" to implement the functionality of session initialization present for example in the Sqoop connector for Oracle (see https://sqoop.apache.org/docs/1.4.6/SqoopUserGuide.html#_oraoop_oracle_session_initialization_statements ) . After each database session is opened to the remote DB, and before starting to read data, this option executes a custom SQL statement (or a PL/SQL block in the case of Oracle).
See also https://issues.apache.org/jira/browse/SPARK-21519
## How was this patch tested?
Manually tested using Spark SQL data source and Oracle JDBC
Author: LucaCanali <lu...@cern.ch>
Closes #18724 from LucaCanali/JDBC_datasource_sessionInitStatement.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0377338b
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0377338b
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0377338b
Branch: refs/heads/master
Commit: 0377338bf76c0d55963a338dd73076113a922de9
Parents: 2387f1e
Author: LucaCanali <lu...@cern.ch>
Authored: Fri Aug 11 12:03:37 2017 -0700
Committer: gatorsmile <ga...@gmail.com>
Committed: Fri Aug 11 12:03:37 2017 -0700
----------------------------------------------------------------------
docs/sql-programming-guide.md | 7 +++++
.../datasources/jdbc/JDBCOptions.scala | 3 ++
.../execution/datasources/jdbc/JDBCRDD.scala | 15 ++++++++++
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 31 ++++++++++++++++++++
4 files changed, 56 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/0377338b/docs/sql-programming-guide.md
----------------------------------------------------------------------
diff --git a/docs/sql-programming-guide.md b/docs/sql-programming-guide.md
index 2ac2383..ee231a9 100644
--- a/docs/sql-programming-guide.md
+++ b/docs/sql-programming-guide.md
@@ -1309,6 +1309,13 @@ the following case-insensitive options:
</tr>
<tr>
+ <td><code>sessionInitStatement</code></td>
+ <td>
+ After each database session is opened to the remote DB and before starting to read data, this option executes a custom SQL statement (or a PL/SQL block). Use this to implement session initialization code. Example: <code>option("sessionInitStatement", """BEGIN execute immediate 'alter session set "_serial_direct_read"=true'; END;""")</code>
+ </td>
+ </tr>
+
+ <tr>
<td><code>truncate</code></td>
<td>
This is a JDBC writer related option. When <code>SaveMode.Overwrite</code> is enabled, this option causes Spark to truncate an existing table instead of dropping and recreating it. This can be more efficient, and prevents the table metadata (e.g., indices) from being removed. However, it will not work in some cases, such as when the new data has a different schema. It defaults to <code>false</code>. This option applies only to writing.
http://git-wip-us.apache.org/repos/asf/spark/blob/0377338b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
index 96a8a51..05b0005 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
@@ -138,6 +138,8 @@ class JDBCOptions(
case "REPEATABLE_READ" => Connection.TRANSACTION_REPEATABLE_READ
case "SERIALIZABLE" => Connection.TRANSACTION_SERIALIZABLE
}
+ // An option to execute custom SQL before fetching data from the remote DB
+ val sessionInitStatement = parameters.get(JDBC_SESSION_INIT_STATEMENT)
}
object JDBCOptions {
@@ -161,4 +163,5 @@ object JDBCOptions {
val JDBC_CREATE_TABLE_COLUMN_TYPES = newOption("createTableColumnTypes")
val JDBC_BATCH_INSERT_SIZE = newOption("batchsize")
val JDBC_TXN_ISOLATION_LEVEL = newOption("isolationLevel")
+ val JDBC_SESSION_INIT_STATEMENT = newOption("sessionInitStatement")
}
http://git-wip-us.apache.org/repos/asf/spark/blob/0377338b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
index 24e1369..3274be9 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
@@ -273,6 +273,21 @@ private[jdbc] class JDBCRDD(
import scala.collection.JavaConverters._
dialect.beforeFetch(conn, options.asProperties.asScala.toMap)
+ // This executes a generic SQL statement (or PL/SQL block) before reading
+ // the table/query via JDBC. Use this feature to initialize the database
+ // session environment, e.g. for optimizations and/or troubleshooting.
+ options.sessionInitStatement match {
+ case Some(sql) =>
+ val statement = conn.prepareStatement(sql)
+ logInfo(s"Executing sessionInitStatement: $sql")
+ try {
+ statement.execute()
+ } finally {
+ statement.close()
+ }
+ case None =>
+ }
+
// H2's JDBC driver does not support the setSchema() method. We pass a
// fully-qualified table name in the SELECT statement. I don't know how to
// talk about a table in a completely portable way.
http://git-wip-us.apache.org/repos/asf/spark/blob/0377338b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index 4c43646..8dc11d8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
@@ -1044,4 +1044,35 @@ class JDBCSuite extends SparkFunSuite
assert(sql("select * from people_view").count() == 3)
}
}
+
+ test("SPARK-21519: option sessionInitStatement, run SQL to initialize the database session.") {
+ val initSQL1 = "SET @MYTESTVAR 21519"
+ val df1 = spark.read.format("jdbc")
+ .option("url", urlWithUserAndPass)
+ .option("dbtable", "(SELECT NVL(@MYTESTVAR, -1))")
+ .option("sessionInitStatement", initSQL1)
+ .load()
+ assert(df1.collect() === Array(Row(21519)))
+
+ val initSQL2 = "SET SCHEMA DUMMY"
+ val df2 = spark.read.format("jdbc")
+ .option("url", urlWithUserAndPass)
+ .option("dbtable", "TEST.PEOPLE")
+ .option("sessionInitStatement", initSQL2)
+ .load()
+ val e = intercept[SparkException] {df2.collect()}.getMessage
+ assert(e.contains("""Schema "DUMMY" not found"""))
+
+ sql(
+ s"""
+ |CREATE OR REPLACE TEMPORARY VIEW test_sessionInitStatement
+ |USING org.apache.spark.sql.jdbc
+ |OPTIONS (url '$urlWithUserAndPass',
+ |dbtable '(SELECT NVL(@MYTESTVAR1, -1), NVL(@MYTESTVAR2, -1))',
+ |sessionInitStatement 'SET @MYTESTVAR1 21519; SET @MYTESTVAR2 1234')
+ """.stripMargin)
+
+ val df3 = sql("SELECT * FROM test_sessionInitStatement")
+ assert(df3.collect() === Array(Row(21519, 1234)))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org