spark-test commented on code in PR #59894:
URL: https://github.com/apache/spark/pull/59894#discussion_r4407105269
##########
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:
**Blocking (P1):** This picks the catalog table's partition columns on every
start, including restarts of a query whose sink log at the table location
already holds files in another layout. Take a table `PARTITIONED BY (p)` that a
query fed through `toTable` without `partitionBy` before this change: its sink
log lists root-level files. After the restart, `FileStreamSink` appends
`p=<v>/` files to the same `_spark_metadata` log without comparing layouts.
Table reads (`spark.table(...)`) and path reads resolve to
`MetadataLogFileIndex`, because `_spark_metadata` exists. The root directory
parses to no partition values, and every `p=<v>/` directory stops at the same
base path. Partition inference therefore raises no conflict error and returns
the spec `(p)`, and `listFiles` serves only the `p=` directories. Every row
committed before the upgrade silently disappears from reads while its files
stay in the log. At the merge base, the same log held only root files, the
inferred spec was
empty, and every file was listed.
This applies to classic, Spark Connect, and PySpark `toTable` calls. It also
applies to Spark Declarative Pipelines streaming tables declared `PARTITIONED
BY` on V1 file sources, since `StreamingTableWrite` never calls `partitionBy`,
so their next incremental update hides all earlier rows. Fresh tables, tables
created by `toTable`, and path-based `start(path)` writes are unaffected. The
file sink is documented as exactly-once, and the opt-in `retention` TTL is the
only documented way committed files leave reader queries. The PR description
notes that the old root files are no longer returned, but users get no signal,
and recovery means rewriting the data or moving to a new location. The new
tests always start from a fresh table and checkpoint, so they do not reach this
path.
**Recommended change:** On the toTable V1 path, before creating a file
stream sink at the table location, read the layout already committed in that
location's sink log, and fail the query start with a clear error when that
layout differs from the table partitioning the sink is about to use, instead of
silently appending files in a new layout. Add streaming table API tests that
restart a query over a sink log written in another layout.
**Why this works:** When a catalog table is present and its location already
holds a non-empty file stream sink log, the writer derives the partition
columns that the committed entries encode, using the same path-based inference
that table reads apply to that log. It then compares them, under the session's
case sensitivity, with the partition columns selected for the new sink. On a
mismatch, the query start fails before the sink is created, with an error that
names the table, the recorded and the expected partitioning, and the recovery:
a new checkpoint and table location, or rewriting the existing data into the
table's layout. Because the check runs before the first new batch, no file is
ever committed in a conflicting layout, and readers keep seeing every committed
row.
**Scope:** Validate the existing sink-log layout at query start on the
toTable V1 path, with a new error and restart tests.
**Compatibility:** For fresh or already-matching sink locations, the
declared contract still holds: without partitionBy the sink writes with the
table's partitioning, and a given partitionBy must match that partitioning.
**Risks:** Reading the existing sink log at query start adds driver work
proportional to the committed entries. This is comparable to the full log read
that the per-batch table refresh already performs for V1 table sinks. Existing
pipelines over such tables, including incremental SDP updates of partitioned
streaming tables, stop at start until the user moves to a new location,
rewrites the data, or runs an SDP full refresh. The error message and migration
notes must say so.
**Constraints:** Apply the comparison only when the write targets a catalog
table through toTable, so path-based start(path) behavior is unchanged. Compare
column names under the session's case sensitivity, so that case-only
differences, which the read path already maps, do not cause a false failure.
Skip the comparison when the sink log has no committed files, so new tables and
fresh locations keep the declared behavior.
**Exclusions:**
- Repairing or migrating sink logs that already mix layouts is outside this
repair.
- Path-based start(path) writers that change partitionBy between runs keep
their existing behavior.
- Whether previously valid queries may be stopped at all on restart, and
whether a legacy flag or a migration-guide entry is needed, is left to the
author's decision on the restart policy for upgraded queries.
**Success:** Restarting a query on the new code over a partitioned V1 file
table whose sink log holds root-level files fails before writing, with an error
that names the recorded and the expected partitioning. No new file is
committed, and table reads still return every previously committed row. A
restart over a sink log written with a partition column the table does not
declare fails before writing, instead of leaving the table unreadable. Fresh
tables, tables created by toTable, and existing tables whose sink log already
uses the table's partitioning keep starting and writing exactly as the PR
declares. Path-based start(path) writes behave exactly as before.
**Verification:**
- **Regression:** Commit at least one batch into a partitioned V1 file table
through a sink log in the unpartitioned layout, as the previous code wrote it,
then restart from the same checkpoint. Show that the start fails with the new
error, that no file is committed under a partition directory, and that table
reads still return every previously committed row.
- **Regression:** Commit batches into a partitioned V1 file table through a
sink log partitioned by a column the table does not declare, then restart
without that partitioning. Show that the start fails before writing and that
the table stays readable.
- **Behavior:** Starting with no existing sink log, on a table created by
toTable, and over a sink log already partitioned like the table still writes
partition directories that match the table's partitioning.
- **Compatibility:** Path-based start(path) file sink writes keep their
previous partitioning behavior, including when the sink directory already holds
a log.
##########
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:
**Nit (P3):** Every case in this list, and in the test above, uses a table
with at most one partition column. The new guard compares `partitionBy` with
the table's partition columns as an ordered `Seq`, the same rule as a batch
`saveAsTable` append. No test would fail if that comparison regressed to set
equality, or started rejecting an in-order multi-column `partitionBy`.
Multi-column partitioning such as `(year, month, day)` is common, so a
two-column case covering both the accepted table order and a rejected
reordering would give the rule a failure signal.
**Verification:**
- **Behavior:** Against a table partitioned by (a, b), partitionBy('a', 'b')
starts and writes rows under a=<v>/b=<w>/ directories, while partitionBy('b',
'a') fails at start with _LEGACY_ERROR_TEMP_1163 that reports both the
specified and the existing column order.
##########
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:
**Nit (P3):** For an existing v1 table, this sentence is the only
user-facing statement of the new contract, but it reads as a recommendation.
The implementation enforces the contract before the query starts. A
`partitionBy` whose columns differ from the table's partition columns throws
`AnalysisException`: `_LEGACY_ERROR_TEMP_1163`, or
`COLUMN_NOT_DEFINED_IN_TABLE` for an unknown column. That includes the same
columns in another order and a non-empty `partitionBy` on an unpartitioned
table. Without `partitionBy`, a query that does not produce every partition
column of the table also fails at start, with `_LEGACY_ERROR_TEMP_1196`. The
next paragraph says `partitionBy` "will be ignored" for an existing v2 table,
so readers can easily take "should match" as tolerated. Users of existing
queries then learn about the change only from the exception on restart. The
PySpark docstring in `python/pyspark/sql/streaming/readwriter.py` repeats this
text, so both copies need the same change.
**Verification:**
- **Inspection:** The Scaladoc and the PySpark docstring carry the same v1
wording, naming the start-time failure, the ordered match, and the
partition-column requirement, and that wording stays consistent with the v2
paragraph.
--
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]
