dongjoon-hyun opened a new pull request, #59894: URL: https://github.com/apache/spark/pull/59894
### What changes were proposed in this pull request? This PR aims to make `DataStreamWriter.toTable` respect the partitioning of an existing V1 table like batch writes. - When `partitionBy` is not given, the query writes with the partitioning of the table like `DataFrameWriter.insertInto`. - When `partitionBy` is given, it should match the partitioning of the table. Otherwise, the query fails to start with the same error as `DataFrameWriter.saveAsTable` in `append` mode. This PR also updates the documentation of `toTable` in Scala and Python accordingly. ### Why are the changes needed? Since Apache Spark 3.1.1 ([v3.1.1](https://github.com/apache/spark/releases/tag/v3.1.1), 2021-02-22), `toTable` supports V1 tables (SPARK-33577), but it creates the file stream sink only with the `partitionBy` of the writer and ignores the partitioning of an existing table. SPARK-33659 documented this as the current behavior after a discussion about whether to respect the partitioning of an existing table or to pass `partitionBy` as it is (https://github.com/apache/spark/pull/30521). As a result, the layout of the output does not match the table. For example, for a table partitioned by `p`, a query without `partitionBy` writes the files at the table root, and a query with `partitionBy("i")` writes `i=1` and `i=2` directories. ```scala sql("CREATE TABLE t (i INT, p INT) USING parquet PARTITIONED BY (p)") val input = MemoryStream[Int] input.toDF().selectExpr("value AS i", "value % 2 AS p") .writeStream .format("parquet") .option("checkpointLocation", checkpointDir) .toTable("t") // writes `part-*` files at the table root instead of `p=0` and `p=1` directories ``` On the other hand, batch writes do not allow such layouts. `insertInto` writes with the partitioning of the table, and `saveAsTable` in `append` mode fails when the partitioning does not match. ```scala scala> Seq((1, 1)).toDF("i", "p").write.mode("append").partitionBy("i").saveAsTable("t") org.apache.spark.sql.AnalysisException: Specified partitioning does not match that of the existing table spark_catalog.default.t. Specified partition columns: [i]. Existing partition columns: [p]. ``` ### Does this PR introduce _any_ user-facing change? Yes. For an existing V1 table, `toTable` behaves like the following. | `partitionBy` | Before | After | | --- | --- | --- | | Not given | Writes without partitioning | Writes with the partitioning of the table | | Matches the table | Writes with the partitioning of the table | Writes with the partitioning of the table | | Does not match the table | Writes with `partitionBy` | Fails to start with `_LEGACY_ERROR_TEMP_1163` | In addition, a query without the partition columns of the table fails to start. There is no change for a table which `toTable` creates. Since Spark Declarative Pipelines writes a streaming table via `toTable` without `partitionBy`, a streaming table with `PARTITIONED BY` in a V1 file source table is also written with its partitioning after this change. Note that an existing streaming query which has written to a partitioned table without `partitionBy` starts to write the new files under the partition directories after this change. Since the table is read through the metadata log of the file stream sink, which infers the partitions from the paths of the logged files, the previous files at the table root are not returned anymore. ### How was this patch tested? Pass the CIs with the newly added test cases. ### 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]
