dongjoon-hyun commented on code in PR #59894:
URL: https://github.com/apache/spark/pull/59894#discussion_r4407935069
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala:
##########
@@ -348,12 +358,15 @@ final class DataStreamWriter[T] private[sql](ds:
Dataset[T]) extends streaming.D
DataSourceUtils.encodePartitioningColumns(cols))
case None => optionsWithoutClusteringKey
}
+ // A table is written with its partitioning, which `toTable` checks
against `partitionBy`.
+ val partitionColumns =
catalogTable.map(_.partitionColumnNames.map(normalize(_, "Partition")))
Review Comment:
Thanks for the detailed analysis. Fixed in 450d0bd8529.
`toTable` now checks the sink log at the table location before it creates
the sink (`FileStreamSink.checkPartitioningOfCommittedFiles`, called from
`DataSource.createSink` when a catalog table is given). It infers the partition
columns of the committed files with `MetadataLogFileIndex`, the same way reads
do, and compares them in order, under the case sensitivity of the session, with
the partition columns the sink is about to write with. On a mismatch, the query
fails to start with the new `STREAMING_FILE_SINK_PARTITIONING_MISMATCH`, which
names the table, both column lists, and the recovery. A log without readable
files and `spark.sql.streaming.fileStreamSink.metadata.ignored` skip the check,
and `start(path)` is unchanged.
The new tests in `DataStreamTableAPISuite` cover the following. Three tables
fail to start and add no file: a partitioned table whose committed files are at
the table root (its committed rows are still returned), a partitioned table
whose files are partitioned by an undeclared column, and an unpartitioned table
with `p=` directories. A restart from the same checkpoint with the matching
layout, and a start over a log without files, still append.
One thing I noticed while writing the tests: starting a new checkpoint over
an existing log restarts the batch ids from 0, and the sink skips the batches
up to the latest committed id. So the error message names both a new table
location and a new checkpoint location as the recovery.
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala:
##########
@@ -215,8 +215,9 @@ abstract class DataStreamWriter[T] extends
WriteConfigMethods[DataStreamWriter[T
* [[org.apache.spark.sql.streaming.StreamingQuery]] object can be used to
interact with the
* stream.
*
- * For v1 table, partitioning columns provided by `partitionBy` will be
respected no matter the
- * table exists or not. A new table will be created if the table not exists.
+ * For v1 table, a new table will be created with the partitioning columns
provided by
+ * `partitionBy` if the table does not exist. If the table exists, the
partitioning of the table
+ * will be respected, and `partitionBy` should match it if provided.
Review Comment:
Updated both the Scaladoc and the PySpark docstring in 450d0bd8529:
`partitionBy` must match the partitioning of the table, including the order of
the columns, if provided. Otherwise, or if the files already committed to the
table have other partition columns, the query fails to start.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala:
##########
@@ -353,6 +354,98 @@ class DataStreamTableAPISuite extends StreamTest with
BeforeAndAfter {
}
}
+ test("SPARK-60665: write to a partitioned table with the partitioning of the
table") {
+ def writeAndCheck(tableName: String, partitionColumns:
Option[Seq[String]]): Unit = {
+ withTempDir { checkpointDir =>
+ val input = MemoryStream[Int]
+ val writer = input.toDF().selectExpr("value AS i", "value % 2 AS p")
+ .writeStream
+ .format("parquet")
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ val query = partitionColumns.map(cols => writer.partitionBy(cols:
_*)).getOrElse(writer)
+ .toTable(tableName)
+ try {
+ input.addData(1, 2)
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+ }
+ checkAnswer(spark.table(tableName), Seq(Row(1, 1), Row(2, 0)))
+ assert(spark.table(tableName).inputFiles.map(new
Path(_).getParent.getName).toSet ===
+ Set("p=0", "p=1"))
+ }
+
+ val tableName = "partition_test"
+ Seq(None, Some(Seq("P"))).foreach { partitionColumns =>
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (i INT, p INT) USING parquet PARTITIONED
BY (p)")
+ writeAndCheck(tableName, partitionColumns)
+ }
+ }
+ withTable(tableName) {
+ writeAndCheck(tableName, Some(Seq("p")))
+
assert(spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName))
+ .partitionColumnNames === Seq("p"))
+ }
+ }
+
+ test("SPARK-60665: write to an existing table with mismatched partitioning")
{
+ val tableName = "partition_test"
+ val df = MemoryStream[Int].toDF().selectExpr("value AS i", "value % 2 AS
p")
+ def startQuery(data: DataFrame, partitionColumns: Option[Seq[String]]):
Unit = {
+ withTempDir { checkpointDir =>
+ val writer = data.writeStream
+ .format("parquet")
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ partitionColumns.map(cols => writer.partitionBy(cols:
_*)).getOrElse(writer)
+ .toTable(tableName)
+ }
+ }
+
+ Seq(
Review Comment:
Added `partitionBy of an existing table keeps the order of the partition
columns` in 450d0bd8529. For a table partitioned by `(a, b)`, `partitionBy("b",
"a")` fails with `_LEGACY_ERROR_TEMP_1163` that reports both orders, and
`partitionBy("a", "b")` writes `a=<v>/b=<w>/` directories.
--
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]
