dongjoon-hyun opened a new pull request, #59797: URL: https://github.com/apache/spark/pull/59797
### 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 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
