This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new b84bc05c6c60 [SPARK-60563][SS] Check `INTERVAL` columns and empty
schemas when starting a file stream sink query
b84bc05c6c60 is described below
commit b84bc05c6c60e8a153584ce9dd5cd8d58f18b0ac
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Oct 26 16:37:54 2026 -0700
[SPARK-60563][SS] Check `INTERVAL` columns and empty schemas when starting
a file stream sink query
### What changes were proposed in this pull request?
This PR aims to check `INTERVAL` columns and empty schemas when a streaming
query with a file stream sink starts, like batch file writes.
`DataSource.createSink` now checks the schema of the streaming query before
creating `FileStreamSink`, in the same way as the corresponding batch writes.
| Streaming write | Batch write | `INTERVAL` columns | Empty schemas |
| --- | --- | --- | --- |
| `DataStreamWriter.start(path)` | `DataFrameWriter.save(path)`
(`DataSource.planForWriting`) | Rejected | Rejected |
| `DataStreamWriter.toTable(tableName)` | `DataFrameWriter.saveAsTable` and
`CREATE TABLE AS SELECT` (`DataSource.writeAndRead`) | Rejected | Not checked |
To tell them apart, `DataStreamWriter` passes the target `CatalogTable` of
`toTable` to `DataSource`. Like batch writes, ANSI interval types are allowed,
and `spark.sql.legacy.allowEmptySchemaWrite` disables the empty schema check.
Spark Connect is also covered because its server uses `DataStreamWriter`.
### Why are the changes needed?
Batch file writes do the above checks during analysis. For example,
SPARK-23372 added the empty schema check in Apache Spark 2.4.0
([v2.4.0](https://github.com/apache/spark/releases/tag/v2.4.0), 2018-10-29).
However, the file stream sink has never done these checks when a query starts.
As a result, streaming queries fail late, or write data which cannot be read
back correctly.
- A query with `INTERVAL` columns starts and creates `_spark_metadata` in
the output path, and then fails at the first micro-batch with data.
```
org.apache.spark.sql.streaming.StreamingQueryException: [STREAM_FAILED]
Query [id = ..., runId = ...] terminated with exception:
[UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE] The Parquet datasource doesn't support
the column `d` of the type "INTERVAL". SQLSTATE: 0A000 SQLSTATE: XXKST
```
- Empty struct columns fail at tasks with Parquet (`InvalidSchemaException:
Cannot write a schema with an empty group`) and at the first micro-batch with
CSV. JSON writes them as `{}`, and they disappear when the output is read back.
ORC writes them although batch writes reject them.
```scala
scala> spark.readStream.format("rate").load().selectExpr("value AS i",
"struct() AS s").writeStream.format("json").option("checkpointLocation",
"/tmp/checkpoint").start("/tmp/output")
scala> spark.read.json("/tmp/output").printSchema()
root
|-- i: long (nullable = true)
```
- Empty schemas without any column are written as `{}` lines with JSON and
as empty files with CSV, and ORC writes files which cannot be read back
(`UNABLE_TO_INFER_SCHEMA`).
### Does this PR introduce _any_ user-facing change?
Yes. A streaming query with a file stream sink fails at start with the same
errors as the corresponding batch writes.
```
org.apache.spark.sql.AnalysisException:
[UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE] The Parquet datasource doesn't support
the column `d` of the type "INTERVAL". SQLSTATE: 0A000
```
```
org.apache.spark.sql.AnalysisException:
[EMPTY_SCHEMA_NOT_SUPPORTED_FOR_DATASOURCE] The JSON datasource does not
support writing empty or nested empty schemas. Please make sure the data schema
has at least one or more column(s). SQLSTATE: 0A000
```
Note that existing streaming queries which write empty or nested empty
schemas to a path, e.g., with JSON or ORC, fail to restart.
`spark.sql.legacy.allowEmptySchemaWrite` restores the previous behavior.
### How was this patch tested?
Pass the CIs with the newly added test case in `FileStreamSinkSuite`, which
fails without this fix. It checks the following with `start(path)`.
- `INTERVAL` columns and empty struct columns are rejected at start with
Parquet, ORC, JSON, and CSV.
- An empty struct column is accepted with JSON when
`spark.sql.legacy.allowEmptySchemaWrite` is enabled.
It also checks that `toTable` does not check empty struct columns of an
existing table, like `saveAsTable`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59797 from dongjoon-hyun/SPARK-60563.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 97bf1e15020f8861d63fa2dcc21946f95eec55ff)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/classic/DataStreamWriter.scala | 9 ++--
.../sql/execution/datasources/DataSource.scala | 5 ++
.../spark/sql/streaming/FileStreamSinkSuite.scala | 62 ++++++++++++++++++++++
3 files changed, 73 insertions(+), 3 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
index 62849aab56e1..56f010b7966b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
@@ -292,10 +292,10 @@ final class DataStreamWriter[T] private[sql](ds:
Dataset[T]) extends streaming.D
table match {
case table: SupportsWrite if table.supports(STREAMING_WRITE) =>
table
- case _ => createV1Sink(optionsWithPath)
+ case _ => createV1Sink(optionsWithPath, catalogTable)
}
} else {
- createV1Sink(optionsWithPath)
+ createV1Sink(optionsWithPath, catalogTable)
}
startQuery(sink, optionsWithPath, catalogTable = catalogTable)
@@ -333,7 +333,9 @@ final class DataStreamWriter[T] private[sql](ds:
Dataset[T]) extends streaming.D
catalogTable = catalogTable)
}
- private def createV1Sink(optionsWithPath: CaseInsensitiveMap[String]): Sink
= {
+ private def createV1Sink(
+ optionsWithPath: CaseInsensitiveMap[String],
+ catalogTable: Option[CatalogTable]): Sink = {
// Do not allow the user to specify clustering columns in the options.
Ignoring this option is
// consistent with the behavior of DataFrameWriter on non Path-based
tables and with the
// behavior of DataStreamWriter on partitioning columns specified in
options.
@@ -350,6 +352,7 @@ final class DataStreamWriter[T] private[sql](ds:
Dataset[T]) extends streaming.D
this.ds.sparkSession,
className = source,
options = optionsWithClusteringColumns,
+ catalogTable = catalogTable,
partitionColumns = normalizedParCols.getOrElse(Nil))
ds.createSink(outputMode, this.ds.schema)
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
index 93711d583907..e95a17898ca8 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
@@ -343,6 +343,11 @@ case class DataSource(
s.createSink(sparkSession.sqlContext, caseInsensitiveOptions,
partitionColumns, outputMode)
case fileFormat: FileFormat =>
+ disallowWritingIntervals(schema, fileFormat.toString,
forbidAnsiIntervals = false)
+ // Like batch writes, check empty schemas only when writing to a path,
not to a table.
+ if (catalogTable.isEmpty) {
+ DataSource.validateSchema(fileFormat.toString, schema, conf)
+ }
val path = makeQualified(new
Path(caseInsensitiveOptions.getOrElse("path", {
throw QueryExecutionErrors.dataPathNotSpecifiedError()
}))).toString
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala
index 8bc5037f86c0..c244d9a28030 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala
@@ -293,6 +293,68 @@ abstract class FileStreamSinkSuite extends StreamTest {
}
}
+ test("SPARK-60563: Check INTERVAL and empty struct columns when starting a
query") {
+ val df = MemoryStream[Int].toDF()
+ val intervalDf = df.selectExpr("value AS i", "make_interval(0, 0, 0,
value) AS d")
+ val emptyStructDf = df.selectExpr("value AS i", "struct() AS s")
+
+ val ansiIntervalDf = df.selectExpr(
+ "value AS i", "make_ym_interval(0, value) AS ym", "make_dt_interval(0,
0, 0, value) AS dt")
+
+ def startQuery(data: DataFrame, format: String): StreamingQuery = {
+ data.writeStream
+ .option("checkpointLocation",
+ Utils.createTempDir(namePrefix =
"stream.checkpoint").getCanonicalPath)
+ .format(format)
+ .start(Utils.createTempDir(namePrefix =
"stream.output").getCanonicalPath)
+ }
+
+ def startTableQuery(data: DataFrame, table: String): StreamingQuery = {
+ data.writeStream
+ .option("checkpointLocation",
+ Utils.createTempDir(namePrefix =
"stream.checkpoint").getCanonicalPath)
+ .format("json")
+ .toTable(table)
+ }
+
+ Seq("parquet" -> "Parquet", "orc" -> "ORC", "json" -> "JSON", "csv" ->
"CSV").foreach {
+ case (format, formatName) =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ startQuery(intervalDf, format)
+ },
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`d`", "columnType" -> "\"INTERVAL\"", "format" ->
formatName))
+ checkError(
+ exception = intercept[AnalysisException] {
+ startQuery(emptyStructDf, format)
+ },
+ condition = "EMPTY_SCHEMA_NOT_SUPPORTED_FOR_DATASOURCE",
+ parameters = Map("format" -> formatName))
+ // Like batch writes, ANSI interval types are allowed.
+ startQuery(ansiIntervalDf, format).stop()
+ }
+
+ withSQLConf(SQLConf.ALLOW_EMPTY_SCHEMAS_FOR_WRITES.key -> "true") {
+ startQuery(emptyStructDf, "json").stop()
+ }
+
+ withTable("t") {
+ sql("CREATE TABLE t (i INT, s STRUCT<>) USING json")
+ // Like `DataFrameWriter.saveAsTable`, writing to a table does not check
empty structs.
+ startTableQuery(emptyStructDf, "t").stop()
+ // However, it checks INTERVAL columns.
+ checkError(
+ exception = intercept[AnalysisException] {
+ startTableQuery(intervalDf, "t")
+ },
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`d`", "columnType" -> "\"INTERVAL\"", "format" ->
"JSON"))
+ }
+ }
+
test("parquet") {
testFormat(None) // should not throw error as default format parquet when
not specified
testFormat(Some("parquet"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
