[SPARK-18538] [SQL] Fix Concurrent Table Fetching Using DataFrameReader JDBC APIs by gatorsmile · Pull Request #15975 · apache/spark

gatorsmile · 2016-11-22T06:01:58Z

What changes were proposed in this pull request?

The following two DataFrameReader JDBC APIs ignore the user-specified parameters of parallelism degree.

  def jdbc(
      url: String,
      table: String,
      columnName: String,
      lowerBound: Long,
      upperBound: Long,
      numPartitions: Int,
      connectionProperties: Properties): DataFrame

  def jdbc(
      url: String,
      table: String,
      predicates: Array[String],
      connectionProperties: Properties): DataFrame

This PR is to fix the issues. To verify the behavior correctness, we improve the plan output of EXPLAIN command by adding numPartitions in the JDBCRelation node.

Before the fix,

== Physical Plan ==
*Scan JDBCRelation(TEST.PEOPLE) [NAME#1896,THEID#1897] ReadSchema: struct<NAME:string,THEID:int>

After the fix,

== Physical Plan ==
*Scan JDBCRelation(TEST.PEOPLE) [numPartitions=3] [NAME#1896,THEID#1897] ReadSchema: struct<NAME:string,THEID:int>

How was this patch tested?

Added the verification logics on all the test cases for JDBC concurrent fetching.

SparkQA · 2016-11-22T06:02:38Z

Test build #68981 has started for PR 15975 at commit bcc86c0.

viirya · 2016-11-22T10:39:26Z

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

   */
  def jdbc(url: String, table: String, properties: Properties): DataFrame = {
-    jdbc(url, table, JDBCRelation.columnPartition(null), properties)
+    // connectionProperties should override settings in extraOptions.


nit: no connectionProperties here but only properties.

viirya · 2016-11-22T10:46:17Z

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

+    val partitioningInfo = if (parts.nonEmpty) s" [numPartitions=${parts.length}]"
    // credentials should not be included in the plan output, table information is sufficient.
-    s"JDBCRelation(${jdbcOptions.table})"
+    s"JDBCRelation(${jdbcOptions.table})" + partitioningInfo


If parts is empty, this string looks weird like "JDBCRelation(.....)()" as I tried locally.

Fixed. Thanks!

dongjoon-hyun · 2016-11-22T18:36:32Z

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

@@ -404,6 +425,7 @@ class JDBCSuite extends SparkFunSuite
      numPartitions = 0,


Oh, is this 0 numPartitions valid?

In the read path, YES.

Will fix this after your PR #15966 is merged

it's merged, has it been fixed?

: ) Let me move the value check to the write path, and then, we can keep the existing behavior of numPartitions in the read path.

dongjoon-hyun · 2016-11-22T18:42:44Z

Hi, @gatorsmile .
Can we can add a testcase for writing partition here? (According to the PR title, it's beyond of the scope.)

gatorsmile · 2016-11-22T19:22:21Z

Sure, will add the test cases for the write path. @dongjoon-hyun

gatorsmile · 2016-11-24T07:25:37Z

The test cases of the write path will be added after #15966 is merged.

SparkQA · 2016-11-24T07:27:32Z

Test build #69114 has started for PR 15975 at commit 5c5b3ca.

gatorsmile · 2016-11-24T18:39:49Z

retest this please

SparkQA · 2016-11-24T20:30:44Z

Test build #69139 has finished for PR 15975 at commit 5c5b3ca.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

rxin · 2016-11-25T23:52:10Z

@gatorsmile did you update this?

gatorsmile · 2016-11-26T03:21:14Z

Will update it tonight.

gatorsmile · 2016-11-26T05:00:56Z

@dongjoon-hyun Will not add test cases for the write path in this PR, because it requires code changes on the source codes. Will submit a separate PR after this is merged.

dongjoon-hyun · 2016-11-26T06:41:52Z

@gatorsmile NP. Thank you for informing that.

SparkQA · 2016-11-26T07:10:30Z

Test build #69180 has finished for PR 15975 at commit 1b0caea.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2016-11-26T22:29:45Z

Any more comment? : )

Also cc @cloud-fan @hvanhovell

cloud-fan · 2016-11-27T05:02:44Z

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

-  private def jdbc(
-      url: String,
-      table: String,
-      parts: Array[Partition],


this parameter is never used, when did we introduce it?

Introduced in the PR #15499 which was merged to 2.1 only

cloud-fan · 2016-11-27T07:16:30Z

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

-    this.extraOptions += ("url" -> url, "dbtable" -> table)
-    format("jdbc").load()
+    val relation = JDBCRelation(parts, options)(sparkSession)
+    sparkSession.baseRelationToDataFrame(relation)


so we have 2 code path for jdbc? The API with predicates parameter is kind of an advanced API that allows users to customize the partitions, right?

Yeah. The predicate-based API is very useful for the advanced JDBC users.

cloud-fan · 2016-11-28T06:11:36Z

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

+    // properties should override settings in extraOptions.
+    this.extraOptions = this.extraOptions ++ properties.asScala
+    // explicit url and dbtable should override all
+    this.extraOptions += ("url" -> url, "dbtable" -> table)


please use the constant JDBCOptions.xxx

cloud-fan · 2016-11-28T06:13:20Z

LGTM except https://github.com/apache/spark/pull/15975/files#r89722356, what's the status of it?

SparkQA · 2016-11-28T20:49:23Z

Test build #69253 has finished for PR 15975 at commit c29199c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2016-11-29T06:00:23Z

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

-        df
-      }
+    val repartitionedDF = options.numPartitions match {
+      case Some(n) if n <= 0 => throw new IllegalArgumentException(


so this check is only in write path now?

cloud-fan · 2016-11-29T06:06:56Z

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

+  def checkNumPartitions(df: DataFrame, expectedNumPartitions: Int): Unit = {
+    val explain = ExplainCommand(df.queryExecution.logical, extended = true)
+    val plans = spark.sessionState.executePlan(explain).executedPlan
+    val expectedMsg = s"${JDBCOptions.JDBC_NUM_PARTITIONS}=$expectedNumPartitions"


instead of matching string, can we call collect on the logical plan and get the JDBCRelation to check the partitions directly?

cloud-fan · 2016-11-29T06:08:37Z

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

@@ -76,9 +76,6 @@ class JDBCOptions(

  // the number of partitions
  val numPartitions = parameters.get(JDBC_NUM_PARTITIONS).map(_.toInt)


what's the behaviour of this config is set to 0 or negative for read path?

Reading the table using a single partition.

have we documented this behaviour?

Not yet. : ) Will try to document it in the jdbc API of DataFrameReader.scala

cloud-fan · 2016-11-30T07:51:24Z

retest this please

SparkQA · 2016-11-30T07:52:41Z

Test build #69392 has started for PR 15975 at commit 404aa22.

cloud-fan · 2016-11-30T07:54:02Z

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


+  // Check whether the tables are fetched in the expected degree of parallelism
+  def checkNumPartitions(df: DataFrame, expectedNumPartitions: Int): Unit = {
+    df.queryExecution.analyzed.collectFirst {


nit:

val jdbcRelations = df.queryExecution....collect { case LogicalRelation(r: JDBCRelation, _, _) => r } assert(jdbcRelations.length == 1) assert(jdbcRelations.head.parts.length == expected, "error message")

SparkQA · 2016-11-30T23:17:22Z

Test build #69430 has finished for PR 15975 at commit 728c103.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2016-12-01T02:23:46Z

retest this please

SparkQA · 2016-12-01T04:51:27Z

Test build #69444 has finished for PR 15975 at commit 728c103.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2016-12-01T07:48:48Z

thanks, merging to master!

Since #15868 is not backported to 2.1, this PR conflicts with 2.1, @gatorsmile can you send a backport PR? thanks.

gatorsmile · 2016-12-01T16:16:28Z

Sure, will do it soon.

…DataFrameReader JDBC APIs ### What changes were proposed in this pull request? #### This PR is to backport #15975 to Branch 2.1 --- The following two `DataFrameReader` JDBC APIs ignore the user-specified parameters of parallelism degree. ```Scala def jdbc( url: String, table: String, columnName: String, lowerBound: Long, upperBound: Long, numPartitions: Int, connectionProperties: Properties): DataFrame ``` ```Scala def jdbc( url: String, table: String, predicates: Array[String], connectionProperties: Properties): DataFrame ``` This PR is to fix the issues. To verify the behavior correctness, we improve the plan output of `EXPLAIN` command by adding `numPartitions` in the `JDBCRelation` node. Before the fix, ``` == Physical Plan == *Scan JDBCRelation(TEST.PEOPLE) [NAME#1896,THEID#1897] ReadSchema: struct<NAME:string,THEID:int> ``` After the fix, ``` == Physical Plan == *Scan JDBCRelation(TEST.PEOPLE) [numPartitions=3] [NAME#1896,THEID#1897] ReadSchema: struct<NAME:string,THEID:int> ``` ### How was this patch tested? Added the verification logics on all the test cases for JDBC concurrent fetching. Author: gatorsmile <gatorsmile@gmail.com> Closes #16111 from gatorsmile/jdbcFix2.1.

…r JDBC APIs ### What changes were proposed in this pull request? The following two `DataFrameReader` JDBC APIs ignore the user-specified parameters of parallelism degree. ```Scala def jdbc( url: String, table: String, columnName: String, lowerBound: Long, upperBound: Long, numPartitions: Int, connectionProperties: Properties): DataFrame ``` ```Scala def jdbc( url: String, table: String, predicates: Array[String], connectionProperties: Properties): DataFrame ``` This PR is to fix the issues. To verify the behavior correctness, we improve the plan output of `EXPLAIN` command by adding `numPartitions` in the `JDBCRelation` node. Before the fix, ``` == Physical Plan == *Scan JDBCRelation(TEST.PEOPLE) [NAME#1896,THEID#1897] ReadSchema: struct<NAME:string,THEID:int> ``` After the fix, ``` == Physical Plan == *Scan JDBCRelation(TEST.PEOPLE) [numPartitions=3] [NAME#1896,THEID#1897] ReadSchema: struct<NAME:string,THEID:int> ``` ### How was this patch tested? Added the verification logics on all the test cases for JDBC concurrent fetching. Author: gatorsmile <gatorsmile@gmail.com> Closes apache#15975 from gatorsmile/jdbc.

fix.

bcc86c0

gatorsmile changed the title ~~Fix Concurrent Table Fetching Using DataFrameReader JDBC APIs~~ [SPARK-18538] [SQL] Fix Concurrent Table Fetching Using DataFrameReader JDBC APIs Nov 22, 2016

gatorsmile mentioned this pull request Nov 22, 2016

[SPARK-18413][SQL][FOLLOW-UP] Use numPartitions instead of maxConnections #15966

Closed

srowen approved these changes Nov 22, 2016

View reviewed changes

viirya reviewed Nov 22, 2016

View reviewed changes

dongjoon-hyun reviewed Nov 22, 2016

View reviewed changes

address comments

5c5b3ca

gatorsmile added 3 commits November 25, 2016 19:24

Merge remote-tracking branch 'upstream/master' into jdbc

483c2a4

fix test failure introduced by code merge.

fdb7392

nit

1b0caea

cloud-fan reviewed Nov 27, 2016

View reviewed changes

cloud-fan reviewed Nov 28, 2016

View reviewed changes

fix.

c29199c

cloud-fan reviewed Nov 29, 2016

View reviewed changes

gatorsmile added 2 commits November 29, 2016 21:07

Merge remote-tracking branch 'upstream/master' into jdbc

57e72f5

address comments.

404aa22

cloud-fan reviewed Nov 30, 2016

View reviewed changes

gatorsmile added 2 commits November 30, 2016 09:44

Merge remote-tracking branch 'upstream/master' into jdbc

1987e77

address comments.

728c103

asfgit closed this in b28fe4a Dec 1, 2016

gatorsmile mentioned this pull request Dec 1, 2016

[SPARK-18538] [SQL] [Backport-2.1] Fix Concurrent Table Fetching Using DataFrameReader JDBC APIs #16111

Closed

gatorsmile mentioned this pull request Nov 10, 2017

[SPARK-22479][SQL] Exclude credentials from SaveintoDataSourceCommand.simpleString #19708

Closed

		@@ -404,6 +425,7 @@ class JDBCSuite extends SparkFunSuite
		numPartitions = 0,

		@@ -76,9 +76,6 @@ class JDBCOptions(

		// the number of partitions
		val numPartitions = parameters.get(JDBC_NUM_PARTITIONS).map(_.toInt)

Conversation

gatorsmile commented Nov 22, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Nov 22, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun commented Nov 22, 2016

Uh oh!

gatorsmile commented Nov 22, 2016

Uh oh!

gatorsmile commented Nov 24, 2016

Uh oh!

SparkQA commented Nov 24, 2016

Uh oh!

gatorsmile commented Nov 24, 2016

Uh oh!

SparkQA commented Nov 24, 2016

Uh oh!

rxin commented Nov 25, 2016

Uh oh!

gatorsmile commented Nov 26, 2016

Uh oh!

gatorsmile commented Nov 26, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dongjoon-hyun commented Nov 26, 2016

Uh oh!

SparkQA commented Nov 26, 2016

Uh oh!

gatorsmile commented Nov 26, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Nov 28, 2016

Uh oh!

SparkQA commented Nov 28, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gatorsmile commented Nov 26, 2016 •

edited

Loading