This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new f6b5fd933ed [SPARK-42733][CONNECT][FOLLOWUP] Write without path or
table
f6b5fd933ed is described below
commit f6b5fd933ed41b09c733ae339118133508dd0870
Author: Zhen Li <zhenli...@users.noreply.github.com>
AuthorDate: Tue Mar 14 18:31:27 2023 +0900
[SPARK-42733][CONNECT][FOLLOWUP] Write without path or table
Fixes `DataFrameWriter.save` to work without path or table parameter.
Added support of jdbc method in the writer as it is one of the impl that
does not contains a path or table.
DataFrameWriter.save should work without path parameter because some data
sources, such as jdbc, noop, works without those parameters.
The follow up fix for scala client of
https://github.com/apache/spark/pull/40356
No
Unit and E2E test
Closes #40358 from zhenlineo/write-without-path-table.
Authored-by: Zhen Li <zhenli...@users.noreply.github.com>
Signed-off-by: Hyukjin Kwon <gurwls...@apache.org>
(cherry picked from commit 93334e295483a0ba66e22d8398512ad970a3ea80)
Signed-off-by: Hyukjin Kwon <gurwls...@apache.org>
---
.../org/apache/spark/sql/DataFrameWriter.scala | 37 ++++++++++++++++++++--
.../org/apache/spark/sql/ClientE2ETestSuite.scala | 21 ++++++++++++
.../scala/org/apache/spark/sql/DatasetSuite.scala | 28 ++++++++++++++++
.../CheckConnectJvmClientCompatibility.scala | 4 +--
.../connect/planner/SparkConnectProtoSuite.scala | 11 ++++---
5 files changed, 92 insertions(+), 9 deletions(-)
diff --git
a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index 8434addec92..b9d1fefb105 100644
---
a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++
b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql
-import java.util.Locale
+import java.util.{Locale, Properties}
import scala.collection.JavaConverters._
@@ -228,7 +228,9 @@ final class DataFrameWriter[T] private[sql] (ds:
Dataset[T]) {
// Set path or table
f(builder)
- require(builder.hasPath != builder.hasTable) // Only one can be set
+
+ // Cannot both be set
+ require(!(builder.hasPath && builder.hasTable))
builder.setMode(mode match {
case SaveMode.Append => proto.WriteOperation.SaveMode.SAVE_MODE_APPEND
@@ -345,6 +347,37 @@ final class DataFrameWriter[T] private[sql] (ds:
Dataset[T]) {
})
}
+ /**
+ * Saves the content of the `DataFrame` to an external database table via
JDBC. In the case the
+ * table already exists in the external database, behavior of this function
depends on the save
+ * mode, specified by the `mode` function (default to throwing an exception).
+ *
+ * Don't create too many partitions in parallel on a large cluster;
otherwise Spark might crash
+ * your external database systems.
+ *
+ * JDBC-specific option and parameter documentation for storing tables via
JDBC in <a
+ *
href="https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html#data-source-option";>
+ * Data Source Option</a> in the version you use.
+ *
+ * @param table
+ * Name of the table in the external database.
+ * @param connectionProperties
+ * JDBC database connection arguments, a list of arbitrary string
tag/value. Normally at least
+ * a "user" and "password" property should be included. "batchsize" can be
used to control the
+ * number of rows per insert. "isolationLevel" can be one of "NONE",
"READ_COMMITTED",
+ * "READ_UNCOMMITTED", "REPEATABLE_READ", or "SERIALIZABLE", corresponding
to standard
+ * transaction isolation levels defined by JDBC's Connection object, with
default of
+ * "READ_UNCOMMITTED".
+ * @since 3.4.0
+ */
+ def jdbc(url: String, table: String, connectionProperties: Properties): Unit
= {
+ // connectionProperties should override settings in extraOptions.
+ this.extraOptions ++= connectionProperties.asScala
+ // explicit url and dbtable should override all
+ this.extraOptions ++= Seq("url" -> url, "dbtable" -> table)
+ format("jdbc").save()
+ }
+
/**
* Saves the content of the `DataFrame` in JSON format (<a
href="http://jsonlines.org/";> JSON
* Lines text format or newline-delimited JSON</a>) at the specified path.
This is equivalent
diff --git
a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala
b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala
index 60bb23516b0..5aa5500116d 100644
---
a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala
+++
b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala
@@ -22,6 +22,7 @@ import java.nio.file.Files
import scala.collection.JavaConverters._
import io.grpc.StatusRuntimeException
+import java.util.Properties
import org.apache.commons.io.FileUtils
import org.apache.commons.io.output.TeeOutputStream
import org.scalactic.TolerantNumerics
@@ -175,6 +176,26 @@ class ClientE2ETestSuite extends RemoteSparkSession with
SQLHelper {
}
}
+ test("write without table or path") {
+ // Should receive no error to write noop
+ spark.range(10).write.format("noop").mode("append").save()
+ }
+
+ test("write jdbc") {
+ val url = "jdbc:derby:memory:1234"
+ val table = "t1"
+ try {
+ spark.range(10).write.jdbc(url = s"$url;create=true", table, new
Properties())
+ val result = spark.read.jdbc(url = url, table, new
Properties()).collect()
+ assert(result.length == 10)
+ } finally {
+ // clean up
+ assertThrows[StatusRuntimeException] {
+ spark.read.jdbc(url = s"$url;drop=true", table, new
Properties()).collect()
+ }
+ }
+ }
+
test("writeTo with create and using") {
// TODO (SPARK-42519): Add more test after we can set configs. See more
WriteTo test cases
// in SparkConnectProtoSuite.
diff --git
a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
index 42376db880b..e5738fe7acd 100644
---
a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
+++
b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
@@ -21,6 +21,7 @@ import java.util.concurrent.atomic.AtomicLong
import io.grpc.Server
import io.grpc.inprocess.{InProcessChannelBuilder, InProcessServerBuilder}
+import java.util.Properties
import org.scalatest.BeforeAndAfterEach
import org.apache.spark.connect.proto
@@ -100,6 +101,33 @@ class DatasetSuite extends ConnectFunSuite with
BeforeAndAfterEach {
assert(actualPlan.equals(expectedPlan))
}
+ test("write jdbc") {
+ val df = ss.newDataFrame(_ => ()).limit(10)
+
+ val builder = proto.WriteOperation.newBuilder()
+ builder
+ .setInput(df.plan.getRoot)
+ .setMode(proto.WriteOperation.SaveMode.SAVE_MODE_ERROR_IF_EXISTS)
+ .setSource("jdbc")
+ .putOptions("a", "b")
+ .putOptions("1", "2")
+ .putOptions("url", "url")
+ .putOptions("dbtable", "table")
+
+ val expectedPlan = proto.Plan
+ .newBuilder()
+ .setCommand(proto.Command.newBuilder().setWriteOperation(builder))
+ .build()
+
+ val connectionProperties = new Properties
+ connectionProperties.put("a", "b")
+ connectionProperties.put("1", "2")
+ df.write.jdbc("url", "table", connectionProperties)
+
+ val actualPlan = service.getAndClearLatestInputPlan()
+ assert(actualPlan.equals(expectedPlan))
+ }
+
test("write V2") {
val df = ss.newDataFrame(_ => ()).limit(10)
diff --git
a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala
b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala
index ae6c6c86fec..97d130421a2 100644
---
a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala
+++
b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala
@@ -130,9 +130,7 @@ object CheckConnectJvmClientCompatibility {
ProblemFilters.exclude[Problem]("org.apache.spark.connect.proto.*"),
// DataFrame Reader & Writer
-
ProblemFilters.exclude[Problem]("org.apache.spark.sql.DataFrameReader.json"),
-
ProblemFilters.exclude[Problem]("org.apache.spark.sql.DataFrameReader.jdbc"),
-
ProblemFilters.exclude[Problem]("org.apache.spark.sql.DataFrameWriter.jdbc"),
+
ProblemFilters.exclude[Problem]("org.apache.spark.sql.DataFrameReader.json"),
// deprecated
// DataFrameNaFunctions
ProblemFilters.exclude[Problem]("org.apache.spark.sql.DataFrameNaFunctions.this"),
diff --git
a/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
b/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
index 9cc714d630b..824ee7aceb4 100644
---
a/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
+++
b/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
@@ -22,7 +22,7 @@ import scala.collection.JavaConverters._
import com.google.protobuf.ByteString
-import org.apache.spark.SparkClassNotFoundException
+import org.apache.spark.{SparkClassNotFoundException,
SparkIllegalArgumentException}
import org.apache.spark.connect.proto
import org.apache.spark.connect.proto.Expression
import org.apache.spark.connect.proto.Join.JoinType
@@ -554,13 +554,16 @@ class SparkConnectProtoSuite extends PlanTest with
SparkConnectPlanTest {
parameters = Map("columnName" -> "`duplicatedcol`"))
}
- // TODO(SPARK-42733): Writes without path or table should work.
- ignore("Writes fails without path or table") {
- assertThrows[UnsupportedOperationException] {
+ test("Writes fails without path or table") {
+ assertThrows[SparkIllegalArgumentException] {
transform(localRelation.write())
}
}
+ test("Writes without path or table") {
+ transform(localRelation.write(format = Some("noop"), mode =
Some("Append")))
+ }
+
test("Write fails with unknown table - AnalysisException") {
val cmd = readRel.write(tableName = Some("dest"))
assertThrows[AnalysisException] {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
