This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new bf794730ee4b [SPARK-57274][CONNECT] Support fetch/type accessors and
getMoreResults for SparkConnectStatement
bf794730ee4b is described below
commit bf794730ee4b576f47e18d7f1ffa9589be41fb3a
Author: Jiwon Park <[email protected]>
AuthorDate: Tue Jun 9 17:24:49 2026 +0800
[SPARK-57274][CONNECT] Support fetch/type accessors and getMoreResults for
SparkConnectStatement
### What changes were proposed in this pull request?
Implement the `SparkConnectStatement` accessors that previously threw
`SQLFeatureNotSupportedException`
(`setFetchSize`/`getFetchSize`,`setFetchDirection`/`getFetchDirection`,
`getResultSetType`, `setQueryTimeout`/`getQueryTimeout`), fix `getMoreResults`
(which threw, breaking JDBC result-drain loops), and implement the
`createStatement` type/concurrency overloads on `SparkConnectConnection`.
Follow-up to SPARK-54108 / SPARK-54014.
Fetch size / query timeout are stored as hints (Spark Connect is
forward-only and server-paginated); `createStatement` accepts forward-only /
scroll-insensitive and rejects updatable / scroll-sensitive, mirroring the
Spark Thrift Server's Hive JDBC policy. See SPARK-57274 for details.
### Why are the changes needed?
JDBC client tools (e.g. DataGrip) call these methods around every query, so
the throws abort the query path. `getMoreResults` throwing in particular makes
the standard drain loop `while (getMoreResults() || getUpdateCount() != -1)`
error out or spin forever; DataGrip hangs on a result-less command such as `USE
<db>`.
### Does this PR introduce _any_ user-facing change?
Yes, in the unreleased master only: methods that previously threw
`SQLFeatureNotSupportedException` are now implemented. No existing behavior is
removed.
### How was this patch tested?
New cases in `SparkConnectStatementSuite`: accessor defaults/validation,
drain-loop termination, and the typed `createStatement` overloads.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #56341 from j1wonpark/SPARK-57274.
Authored-by: Jiwon Park <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../client/jdbc/SparkConnectStatement.scala | 64 +++++++++++++++++-----
.../client/jdbc/SparkConnectStatementSuite.scala | 57 +++++++++++++++++++
2 files changed, 106 insertions(+), 15 deletions(-)
diff --git
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala
index d1947ae93a40..245832087268 100644
---
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala
+++
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.connect.client.jdbc
import java.sql.{Array => _, _}
import org.apache.spark.sql.connect.client.SparkResult
+import org.apache.spark.sql.connect.client.jdbc.util.JdbcErrorUtils
class SparkConnectStatement(conn: SparkConnectConnection) extends Statement {
@@ -28,6 +29,8 @@ class SparkConnectStatement(conn: SparkConnectConnection)
extends Statement {
private var maxRows: Int = 0
+ private var resultsExhausted: Boolean = false
+
@volatile private var closed: Boolean = false
override def isClosed: Boolean = closed
@@ -94,6 +97,7 @@ class SparkConnectStatement(conn: SparkConnectConnection)
extends Statement {
// reset before executing new query
operationId = null
resultSet = null
+ resultsExhausted = false
var df = conn.spark.sql(sql)
if (maxRows > 0) {
@@ -143,8 +147,13 @@ class SparkConnectStatement(conn: SparkConnectConnection)
extends Statement {
0
}
- override def setQueryTimeout(seconds: Int): Unit =
- throw new SQLFeatureNotSupportedException
+ // This driver does not apply a query timeout; validate and silently drop
the value.
+ override def setQueryTimeout(seconds: Int): Unit = {
+ checkOpen()
+ if (seconds < 0) {
+ throw new SQLException("Query timeout must be zero or a positive
integer.")
+ }
+ }
override def cancel(): Unit = {
checkOpen()
@@ -164,35 +173,60 @@ class SparkConnectStatement(conn: SparkConnectConnection)
extends Statement {
override def getUpdateCount: Int = {
checkOpen()
- if (resultSet != null) {
+ if (resultsExhausted || resultSet != null) {
-1
} else {
0 // always return 0 because affected rows is not supported yet
}
}
- override def getMoreResults: Boolean =
- throw new SQLFeatureNotSupportedException
+ // a single result per execute(), so there is no next one: close the current
+ // ResultSet and mark exhausted, flipping getUpdateCount() to -1 so drain
loops end
+ override def getMoreResults: Boolean = {
+ checkOpen()
+ if (resultSet != null) {
+ resultSet.close()
+ resultSet = null
+ }
+ resultsExhausted = true
+ false
+ }
- override def setFetchDirection(direction: Int): Unit =
- throw new SQLFeatureNotSupportedException
+ override def setFetchDirection(direction: Int): Unit = {
+ checkOpen()
+ if (direction != ResultSet.FETCH_FORWARD) {
+ throw new SQLException(
+ s"Fetch direction ${JdbcErrorUtils.stringifyFetchDirection(direction)}
is not supported.")
+ }
+ }
- override def getFetchDirection: Int =
- throw new SQLFeatureNotSupportedException
+ override def getFetchDirection: Int = {
+ checkOpen()
+ ResultSet.FETCH_FORWARD
+ }
- override def setFetchSize(rows: Int): Unit =
- throw new SQLFeatureNotSupportedException
+ // This driver does not apply a fetch size hint; validate and silently drop
the value.
+ override def setFetchSize(rows: Int): Unit = {
+ checkOpen()
+ if (rows < 0) {
+ throw new SQLException("Fetch size must be zero or a positive integer.")
+ }
+ }
- override def getFetchSize: Int =
- throw new SQLFeatureNotSupportedException
+ override def getFetchSize: Int = {
+ checkOpen()
+ 0
+ }
override def getResultSetConcurrency: Int = {
checkOpen()
ResultSet.CONCUR_READ_ONLY
}
- override def getResultSetType: Int =
- throw new SQLFeatureNotSupportedException
+ override def getResultSetType: Int = {
+ checkOpen()
+ ResultSet.TYPE_FORWARD_ONLY
+ }
override def addBatch(sql: String): Unit =
throw new SQLFeatureNotSupportedException
diff --git
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala
index fa9df3f1247f..6ace1fa9cf78 100644
---
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala
+++
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala
@@ -116,4 +116,61 @@ class SparkConnectStatementSuite extends ConnectFunSuite
with RemoteSparkSession
}
}
}
+
+ test("fetch size, fetch direction, result set type and query timeout
accessors") {
+ withStatement { stmt =>
+ // fetch size: validated then silently dropped, always reads back as 0
+ assert(stmt.getFetchSize === 0)
+ stmt.setFetchSize(42)
+ assert(stmt.getFetchSize === 0)
+ val se1 = intercept[SQLException] {
+ stmt.setFetchSize(-1)
+ }
+ assert(se1.getMessage === "Fetch size must be zero or a positive
integer.")
+
+ // fetch direction: only FETCH_FORWARD is supported
+ stmt.setFetchDirection(ResultSet.FETCH_FORWARD)
+ assert(stmt.getFetchDirection === ResultSet.FETCH_FORWARD)
+ intercept[SQLException] {
+ stmt.setFetchDirection(ResultSet.FETCH_REVERSE)
+ }
+
+ // result set type is forward-only
+ assert(stmt.getResultSetType === ResultSet.TYPE_FORWARD_ONLY)
+
+ // query timeout: validated then silently dropped, always reads back as 0
+ assert(stmt.getQueryTimeout === 0)
+ stmt.setQueryTimeout(30)
+ assert(stmt.getQueryTimeout === 0)
+ val se2 = intercept[SQLException] {
+ stmt.setQueryTimeout(-1)
+ }
+ assert(se2.getMessage === "Query timeout must be zero or a positive
integer.")
+ }
+ }
+
+ test("getMoreResults terminates JDBC drain loops") {
+ // A typical JDBC result-draining loop. With getMoreResults throwing (or
not
+ // flipping getUpdateCount to -1) this would spin forever; assert it
returns.
+ def drain(stmt: Statement): Unit = {
+ while (stmt.getMoreResults || stmt.getUpdateCount != -1) {}
+ }
+
+ withTable("t_drain") {
+ withStatement { stmt =>
+ // result-bearing command
+ assert(stmt.execute("SELECT id FROM range(3)"))
+ assert(stmt.getUpdateCount === -1)
+ drain(stmt)
+ assert(stmt.getResultSet === null)
+ assert(stmt.getUpdateCount === -1)
+
+ // result-less command
+ assert(!stmt.execute("CREATE TABLE t_drain (id INT) USING Parquet"))
+ assert(stmt.getUpdateCount === 0)
+ drain(stmt)
+ assert(stmt.getUpdateCount === -1)
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
