This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new d67467d164f6 [SPARK-57225][SQL] Reject batch writes for DSv2 sources
without write capability
d67467d164f6 is described below
commit d67467d164f6704dd475d328bbc5c7846025e118
Author: Shrirang Mhalgi <[email protected]>
AuthorDate: Tue Aug 4 14:45:12 2026 +0800
[SPARK-57225][SQL] Reject batch writes for DSv2 sources without write
capability
### What changes were proposed in this pull request?
When `DataFrameWriter.saveCommand` encounters a DSv2 `TableProvider` whose
table declares neither `BATCH_WRITE` nor `V1_BATCH_WRITE`, it now throws a
clear `UNSUPPORTED_FEATURE.TABLE_OPERATION` error with operation "batch write"
instead of silently falling through to the V1 write path and producing a
confusing `INTERNAL_ERROR`.
The V1 fallback is preserved when:
- The table declares `V1_BATCH_WRITE` (JDBC pattern), OR
- The provider implements `CreatableRelationProvider` (backward
compatibility with sources that predate the `V1_BATCH_WRITE` capability)
### Why are the changes needed?
Read-only DSv2 sources like `StateDataSource` (statestore) and
`StateMetadataSource` (state-metadata) only declare `BATCH_READ`. Previously,
`df.write.format("statestore").save()` fell through to the V1 write path, hit
the `case _` branch in `DataSource.planForWriting`, and threw:
```
[INTERNAL_ERROR] The operation "planForWriting" is not supported by the
data source
org.apache.spark.sql.execution.datasources.v2.state.StateDataSource does
not allow
create table as select.
```
### Does this PR introduce _any_ user-facing change?
Yes. Write attempts on read-only DSv2 sources now produce:
```
[UNSUPPORTED_FEATURE.TABLE_OPERATION] The feature is not supported:
Table `read_only_table` does not support batch write.
```
### How was this patch tested?
- Added test: read-only DSv2 source (BATCH_READ only) throws
`UNSUPPORTED_FEATURE.TABLE_OPERATION` on both Append mode and default
(ErrorIfExists) mode
- Added test: DSv2 source with `V1_BATCH_WRITE` still falls back to V1 path
without error
- Verified existing `V1WriteFallbackSuite`,
`V1WriteFallbackSessionCatalogSuite`, `TableCapabilityCheckSuite`,
`DataFrameReaderWriterSuite` all pass.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Using Claude-Opus 4.6
Closes #56319 from shrirangmhalgi/SPARK-57225-state-datasource-readonly.
Authored-by: Shrirang Mhalgi <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit ac1d66260046a3d2bf74c8f503fd9fe4b1578d12)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/errors/QueryCompilationErrors.scala | 4 ++
.../apache/spark/sql/classic/DataFrameWriter.scala | 25 ++++---
.../spark/sql/connector/DataSourceV2Suite.scala | 84 +++++++++++++++++++++-
3 files changed, 102 insertions(+), 11 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index e98c7966b894..b618399388e6 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -1240,6 +1240,10 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
unsupportedTableOperationError(table.name(), "batch scan")
}
+ def unsupportedBatchWriteError(table: Table): Throwable = {
+ unsupportedTableOperationError(table.name(), "batch write")
+ }
+
def unsupportedStreamingScanError(table: Table): Throwable = {
unsupportedTableOperationError(table.name(), "either micro-batch or
continuous scan")
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala
index a9f16ffa87be..96714a419511 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala
@@ -41,6 +41,7 @@ import
org.apache.spark.sql.execution.datasources.{DataSource, DataSourceUtils}
import org.apache.spark.sql.execution.datasources.v2._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
+import org.apache.spark.sql.sources.CreatableRelationProvider
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.ArrayImplicits._
@@ -181,11 +182,15 @@ final class DataFrameWriter[T] private[sql](ds:
Dataset[T]) extends sql.DataFram
val t = getTable
if (t.supports(BATCH_WRITE)) {
(t, None, None)
- } else {
- // Streaming also uses the data source V2 API. So it may be
that the data source
- // implements v2, but has no v2 implementation for batch
writes. In that case, we
- // fall back to saving as though it's a V1 source.
+ } else if (t.supports(V1_BATCH_WRITE) ||
+ provider.isInstanceOf[CreatableRelationProvider]) {
+ // Fall back to V1 write path. V1_BATCH_WRITE is the explicit
capability
+ // for DSv2 tables that opt into the V1 write path. We also
check
+ // CreatableRelationProvider for backward compatibility with
sources that
+ // predate the V1_BATCH_WRITE capability (SPARK-28334).
return saveToV1SourceCommand(path)
+ } else {
+ throw QueryCompilationErrors.unsupportedBatchWriteError(t)
}
}
@@ -229,15 +234,17 @@ final class DataFrameWriter[T] private[sql](ds:
Dataset[T]) extends sql.DataFram
finalOptions,
ignoreIfExists = createMode == SaveMode.Ignore)
case _: TableProvider =>
- if (getTable.supports(BATCH_WRITE)) {
+ val t = getTable
+ if (t.supports(BATCH_WRITE)) {
throw
QueryCompilationErrors.writeWithSaveModeUnsupportedBySourceError(
source, createMode.name())
- } else {
- // Streaming also uses the data source V2 API. So it may be
that the data source
- // implements v2, but has no v2 implementation for batch
writes. In that case, we
- // fallback to saving as though it's a V1 source.
+ } else if (t.supports(V1_BATCH_WRITE) ||
+ provider.isInstanceOf[CreatableRelationProvider]) {
+ // Fall back to V1 write path (see comment in Append/Overwrite
branch above).
assertSchemaEvolutionNotEnabledForV1Write()
saveToV1SourceCommand(path)
+ } else {
+ throw QueryCompilationErrors.unsupportedBatchWriteError(t)
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index 1ae10fbf9679..08afcf79f3e7 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -26,7 +26,7 @@ import scala.jdk.CollectionConverters._
import test.org.apache.spark.sql.connector._
-import org.apache.spark.SparkUnsupportedOperationException
+import org.apache.spark.{SparkException, SparkUnsupportedOperationException}
import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SQLContext}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{
@@ -53,7 +53,7 @@ import
org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters
+import org.apache.spark.sql.internal.connector.{SimpleTableProvider,
SupportsPushDownCatalystFilters}
import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThan,
TableScan}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
@@ -1687,6 +1687,48 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-57225: DSv2 source without batch write capability throws clear
error") {
+ val cls = classOf[ReadOnlyV2DataSource].getName
+ val df = spark.range(1).toDF("i")
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.write.format(cls).mode("append").save()
+ },
+ condition = "UNSUPPORTED_FEATURE.TABLE_OPERATION",
+ parameters = Map(
+ "tableName" -> "`read_only_v2_test`",
+ "operation" -> "batch write"
+ )
+ )
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.write.format(cls).save()
+ },
+ condition = "UNSUPPORTED_FEATURE.TABLE_OPERATION",
+ parameters = Map(
+ "tableName" -> "`read_only_v2_test`",
+ "operation" -> "batch write"
+ )
+ )
+ }
+
+ test("SPARK-57225: DSv2 source with V1_BATCH_WRITE still falls back to V1
path") {
+ val cls = classOf[V1BatchWriteV2DataSource].getName
+ val df = spark.range(1).toDF("i")
+ // V1_BATCH_WRITE sources should fall through to saveToV1SourceCommand,
which calls
+ // DataSource.planForWriting. Since our test source doesn't implement
+ // CreatableRelationProvider or FileFormat, planForWriting hits the case _
branch
+ // and throws INTERNAL_ERROR with "does not allow create table as select".
+ // This proves V1_BATCH_WRITE reached the V1 write path.
+ val ex = intercept[SparkException] {
+ df.write.format(cls).mode("append").save()
+ }
+ assert(ex.getCondition == "INTERNAL_ERROR",
+ "V1_BATCH_WRITE source should reach DataSource.planForWriting (V1 path)")
+ assert(ex.getMessage.contains("does not allow create table as select"))
+ }
+
}
case class RangeInputPartition(start: Int, end: Int) extends InputPartition
@@ -2545,3 +2587,41 @@ class InvalidDataSource extends TestingV2Source {
override def getTable(options: CaseInsensitiveStringMap): Table = null
}
+
+/**
+ * A read-only DSv2 data source that only declares BATCH_READ capability.
+ * Used to test that write attempts produce a clear error instead of falling
through to V1.
+ */
+class ReadOnlyV2DataSource extends SimpleTableProvider {
+ override def getTable(options: CaseInsensitiveStringMap): Table = {
+ new ReadOnlyV2Table
+ }
+}
+
+class ReadOnlyV2Table extends Table with SupportsRead {
+ override def name(): String = "read_only_v2_test"
+ override def schema(): StructType = new StructType().add("i", "long")
+ override def capabilities(): java.util.Set[TableCapability] =
+ java.util.EnumSet.of(TableCapability.BATCH_READ)
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder =
+ throw new UnsupportedOperationException("scan not needed for write tests")
+}
+
+/**
+ * A DSv2 data source that declares V1_BATCH_WRITE, simulating the JDBC
pattern.
+ * Used to verify that V1 fallback is preserved for sources that opt into it.
+ */
+class V1BatchWriteV2DataSource extends SimpleTableProvider {
+ override def getTable(options: CaseInsensitiveStringMap): Table = {
+ new V1BatchWriteV2Table
+ }
+}
+
+class V1BatchWriteV2Table extends Table with SupportsRead {
+ override def name(): String = "v1_batch_write_test"
+ override def schema(): StructType = new StructType().add("i", "long")
+ override def capabilities(): java.util.Set[TableCapability] =
+ java.util.EnumSet.of(TableCapability.BATCH_READ,
TableCapability.V1_BATCH_WRITE)
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder =
+ throw new UnsupportedOperationException("scan not needed for write tests")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
