dongjoon-hyun opened a new pull request, #59759: URL: https://github.com/apache/spark/pull/59759
### What changes were proposed in this pull request? This PR aims to validate the partition columns of a file stream sink when a streaming query starts, like batch file writes. `DataSource.createSink` now receives the schema of the streaming query from `DataStreamWriter` and calls `PartitioningUtils.validatePartitionColumn` before creating `FileStreamSink`, as `DataSource.planForWritingFileFormat` does for batch writes. This covers `DataStreamWriter.start` and `DataStreamWriter.toTable` with an existing file source table. Spark Connect is also covered because its server uses `DataStreamWriter`. Note that `DataStreamWriter.toTable` with a new table already validates the partition columns when it creates the table. ### Why are the changes needed? Apache Spark 2.0 validated the partition columns of a file stream sink in `FileStreamSinkWriter`. However, SPARK-18025 removed `FileStreamSinkWriter` to reuse `FileFormatWriter` in Apache Spark 2.1.0 ([v2.1.0](https://github.com/apache/spark/releases/tag/v2.1.0), 2016-12-15), and this validation has been missing since then. As a result, the file stream sink accepts the partition columns which batch writes reject, and fails at runtime or writes data which cannot be read back. - `VARIANT` partition columns fail at the first micro-batch with a compilation error of the generated code. ``` org.codehaus.commons.compiler.CompileException: File 'generated.java', Line 20, Column 20: Failed to compile: org.codehaus.commons.compiler.CompileException: File 'generated.java', Line 20, Column 20: A method named "compare" is not declared in any enclosing class nor any supertype, nor through a static import ``` - `ARRAY`, `MAP`, and `STRUCT` partition columns are written as directories like `p=%5B1, 2%5D`, `p=%7B1 -> 2}`, and `p=%7B1}`, which cannot be read back as the original values. ```scala scala> spark.readStream.format("rate").load().selectExpr("value AS i", "array(1, 2) AS p").writeStream.format("parquet").partitionBy("p").option("checkpointLocation", "/tmp/checkpoint").start("/tmp/output") ``` ``` $ ls /tmp/output _spark_metadata p=%5B1, 2%5D ``` - Non-default collated string partition columns are partitioned by their binary values. For example, `'a'` and `'A'` of `STRING COLLATE UTF8_LCASE` are written as different partitions. - Duplicate partition columns, e.g., `partitionBy("p", "p")`, are written as nested directories like `p=0/p=0`, which fail to be read with `COLUMN_ALREADY_EXISTS`. - Partitioning by all columns fails at the first micro-batch for Parquet and Text, and writes data which cannot be read back correctly for ORC (`UNABLE_TO_INFER_SCHEMA`) and CSV (no rows). ### 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 batch writes if its partition columns are invalid. ``` org.apache.spark.sql.AnalysisException: [INVALID_PARTITION_COLUMN_DATA_TYPE] Cannot use "ARRAY<INT>" for partition column. SQLSTATE: 0A000 ``` ``` org.apache.spark.sql.AnalysisException: [COLUMN_ALREADY_EXISTS] The column `p` already exists. Choose another name or rename the existing column. SQLSTATE: 42711 ``` ``` org.apache.spark.sql.AnalysisException: [ALL_PARTITION_COLUMNS_NOT_ALLOWED] Cannot use all columns for partition columns. SQLSTATE: KD005 ``` Note that existing streaming queries with such partition columns fail to restart. This includes partitioning by all columns with JSON format, whose output can be read back, but which batch writes have rejected since Apache Spark 2.0.0 (SPARK-15743). ### How was this patch tested? Pass the CIs with the newly added test case in `FileStreamSinkSuite` and the updated test case in `DataStreamReaderWriterSuite`, which fail without this fix. Previously, the test case `prevent all column partitioning` in `DataStreamReaderWriterSuite` passed regardless of the validation, because `spark.range(10).writeStream` fails before starting a query since `spark.range(10)` is not a streaming `Dataset`. This PR updates it to use a streaming `Dataset` and to check the error condition. ### 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]
