This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new f6a94aec2502 [SPARK-60665][SS] Respect the partitioning of an existing
V1 table in `DataStreamWriter.toTable`
f6a94aec2502 is described below
commit f6a94aec2502ee8908f26ef54ecdca16bf922501
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Oct 29 08:53:25 2026 -0700
[SPARK-60665][SS] Respect the partitioning of an existing V1 table in
`DataStreamWriter.toTable`
### 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`. An empty `partitionBy()` is
handled as not given, because Spark Connect cannot send the two differently.
- When `partitionBy` is given, it should match the partitioning of the
table including the order of the columns. Otherwise, the query fails to start
with the same error as `DataFrameWriter.saveAsTable` in `append` mode.
- When the file stream sink log in the location of the table already
records committed files with other partition columns, the query fails to start
with the new error condition `STREAMING_FILE_SINK_PARTITIONING_MISMATCH`.
This PR also updates the documentation of `toTable` in Scala and Python
accordingly, and adds an entry to the Structured Streaming migration guide.
### 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].
```
A table with the file stream sink log is read through the log, which infers
the partitions from the paths of the logged files and skips the files outside
of the partition directories. Therefore, writing the files of a table in
another layout than the committed files makes the committed files silently
disappear from reads. The new check turns it into a failure at the start of the
query, which includes the restarts of the existing queries.
### 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 and for
`start(path)`.
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` cannot be restarted after this change. It fails to
start with `STREAMING_FILE_SINK_PARTITIONING_MISMATCH` instead of writing the
new files under the partition directories next to the committed files at the
table root. Such a query should write to a new table location with a new
checkpoint location, or the committed files should be rewritten with the
partition columns of the table. The [...]
### 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
Closes #59894 from dongjoon-hyun/SPARK-60665.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit a28406de92c97979f3fd4b26ebcfb8e7f672595c)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 6 +
docs/streaming/ss-migration-guide.md | 8 +
python/pyspark/sql/streaming/readwriter.py | 7 +-
.../spark/sql/streaming/DataStreamWriter.scala | 7 +-
.../spark/sql/errors/QueryCompilationErrors.scala | 12 ++
.../spark/sql/classic/DataStreamWriter.scala | 18 +-
.../sql/execution/datasources/DataSource.scala | 4 +
.../execution/streaming/sinks/FileStreamSink.scala | 33 +++-
.../streaming/test/DataStreamTableAPISuite.scala | 220 ++++++++++++++++++++-
9 files changed, 306 insertions(+), 9 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 62da39f36371..d4d06cac082f 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -8079,6 +8079,12 @@
],
"sqlState" : "42K03"
},
+ "STREAMING_FILE_SINK_PARTITIONING_MISMATCH" : {
+ "message" : [
+ "The streaming sink of the table <tableName> writes with the partition
columns [<expectedPartCols>], but the files that the sink already committed to
it have the partition columns [<recordedPartCols>]. Appending files in another
layout hides the committed files from reads. Write to a new table location with
a new checkpoint location, or rewrite the committed files with the partition
columns of the table."
+ ],
+ "sqlState" : "55019"
+ },
"STREAMING_OUTPUT_MODE" : {
"message" : [
"Invalid streaming output mode: <outputMode>."
diff --git a/docs/streaming/ss-migration-guide.md
b/docs/streaming/ss-migration-guide.md
index 8450ea048e72..6f2dd1c6367a 100644
--- a/docs/streaming/ss-migration-guide.md
+++ b/docs/streaming/ss-migration-guide.md
@@ -30,6 +30,14 @@ Please refer [Migration Guide: SQL, Datasets and
DataFrame](../sql-migration-gui
unsupported-operation checks. Existing queries can still restart from their
persistent
checkpoints using the compatibility value recorded in the offset log.
+- Since Spark 4.4, `DataStreamWriter.toTable` writes to an existing V1 table
with the partitioning
+ of the table. A query whose `partitionBy` does not match the partitioning of
the table, including
+ the order of the columns, fails to start. A query that already committed
files to the table in
+ another layout, for example without `partitionBy` to a partitioned table,
fails to start with
+ `STREAMING_FILE_SINK_PARTITIONING_MISMATCH`. As a result, such existing
queries fail to restart.
+ To recover, write to a new table location with a new checkpoint location, or
rewrite the
+ committed files with the partition columns of the table.
+
## Upgrading from Structured Streaming 4.2 to 4.3
- Since Spark 4.3, in Spark Connect the DataFrame passed to a Python
`foreachBatch` function is bound to the streaming query's own session, a clone
of the session that started the query, matching classic `foreachBatch`. This
runs the batch under the configuration the streaming engine pins on the clone
rather than the root session's. As a result, the batch DataFrame can no longer
be combined with the root session inside the function: a temporary view created
from it is not visible to the [...]
diff --git a/python/pyspark/sql/streaming/readwriter.py
b/python/pyspark/sql/streaming/readwriter.py
index 4d34ddd1261d..4c6830e4f579 100644
--- a/python/pyspark/sql/streaming/readwriter.py
+++ b/python/pyspark/sql/streaming/readwriter.py
@@ -1891,8 +1891,11 @@ class DataStreamWriter:
-----
This API is evolving.
- 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` must match it, 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.
For v2 table, `partitionBy` will be ignored if the table already
exists. `partitionBy` will
be respected only if the v2 table does not exist. Besides, the v2
table created by this API
diff --git
a/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala
b/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala
index 8f98466d1f17..11afba591fea 100644
---
a/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala
+++
b/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala
@@ -215,8 +215,11 @@ 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` must match it, 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.
*
* For v2 table, `partitionBy` will be ignored if the table already exists.
`partitionBy` will
* be respected only if the v2 table does not exist. Besides, the v2 table
created by this API
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index e869ef819541..39a009060d5f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -2224,6 +2224,18 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
"existingPartCols" -> existingPartCols))
}
+ def fileStreamSinkPartitioningMismatchError(
+ tableName: String,
+ expectedPartCols: Seq[String],
+ recordedPartCols: Seq[String]): Throwable = {
+ new AnalysisException(
+ errorClass = "STREAMING_FILE_SINK_PARTITIONING_MISMATCH",
+ messageParameters = Map(
+ "tableName" -> tableName,
+ "expectedPartCols" -> expectedPartCols.mkString(", "),
+ "recordedPartCols" -> recordedPartCols.mkString(", ")))
+ }
+
def mismatchedTableBucketingError(
tableName: String,
specifiedBucketString: String,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
index bf467761c7d3..2ea4d5d4cc64 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamWriter.scala
@@ -28,7 +28,7 @@ import org.apache.spark.annotation.Evolving
import org.apache.spark.api.java.function.VoidFunction2
import org.apache.spark.sql.{streaming, Dataset => DS, ForeachWriter}
import org.apache.spark.sql.catalyst.analysis.UnresolvedIdentifier
-import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils}
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.catalyst.plans.logical.{ColumnDefinition,
CreateTable, OptionList, UnresolvedTableSpec}
import org.apache.spark.sql.catalyst.streaming.InternalOutputModes
@@ -205,6 +205,17 @@ final class DataStreamWriter[T] private[sql](ds:
Dataset[T]) extends streaming.D
throw
QueryCompilationErrors.inputSourceDiffersFromDataSourceProviderError(
source, tableName, table)
}
+ // Like a batch append, the specified partitioning should match that of
the table. An empty
+ // `partitionBy()` is the same as no `partitionBy`, which is all Spark
Connect can send.
+ partitioningColumns.filter(_.nonEmpty).foreach { cols =>
+ val resolver = ds.sparkSession.sessionState.analyzer.resolver
+ val specifiedPartCols = CatalogUtils.normalizePartCols(
+ table.qualifiedName, table.schema.map(_.name), cols, resolver)
+ if (specifiedPartCols != table.partitionColumnNames) {
+ throw QueryCompilationErrors.mismatchedTablePartitionColumnError(
+ table.qualifiedName, specifiedPartCols,
table.partitionColumnNames.mkString(", "))
+ }
+ }
format(table.provider.get).startInternal(
Some(new Path(table.location).toString), catalogTable = Some(table))
}
@@ -353,12 +364,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")))
+ .orElse(normalizedParCols)
val ds = DataSource(
this.ds.sparkSession,
className = source,
options = optionsWithClusteringColumns,
catalogTable = catalogTable,
- partitionColumns = normalizedParCols.getOrElse(Nil))
+ partitionColumns = partitionColumns.getOrElse(Nil))
ds.createSink(outputMode, this.ds.schema)
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
index 3827604740f0..e0497f6acfc3 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
@@ -357,6 +357,10 @@ case class DataSource(
}
PartitioningUtils.validatePartitionColumn(
schema, partitionColumns, conf.caseSensitiveAnalysis)
+ catalogTable.foreach { table =>
+ FileStreamSink.checkPartitioningOfCommittedFiles(
+ sparkSession, new Path(path), table.qualifiedName,
partitionColumns)
+ }
new FileStreamSink(sparkSession, path, fileFormat, partitionColumns,
caseInsensitiveOptions)
case _ =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sinks/FileStreamSink.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sinks/FileStreamSink.scala
index 978ec99ae9b3..54f0e1e45201 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sinks/FileStreamSink.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sinks/FileStreamSink.scala
@@ -29,10 +29,10 @@ import org.apache.spark.internal.io.FileCommitProtocol
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.classic.ClassicConversions.castToImpl
-import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryExecutionErrors}
import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker,
FileFormat, FileFormatWriter}
import org.apache.spark.sql.execution.streaming.{ManifestFileCommitProtocol,
Sink}
-import org.apache.spark.sql.execution.streaming.runtime.StreamExecution
+import org.apache.spark.sql.execution.streaming.runtime.{MetadataLogFileIndex,
StreamExecution}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.util.{SerializableConfiguration, Utils}
@@ -78,6 +78,35 @@ object FileStreamSink extends Logging {
metadataDir
}
+ /**
+ * Checks that the files committed to the sink log of the table at `path`
have the partition
+ * columns the sink is about to write with. Readers infer the partitioning
from the logged
+ * paths and skip the files outside of it, so appending files in another
layout would hide the
+ * files that are already committed. A log without readable files records no
layout.
+ */
+ def checkPartitioningOfCommittedFiles(
+ sparkSession: SparkSession,
+ path: Path,
+ tableName: String,
+ partitionColumnNames: Seq[String]): Unit = {
+ val sqlConf = sparkSession.sessionState.conf
+ val fs = path.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val logPath = getMetadataLogPath(fs, path, sqlConf)
+ if (!sqlConf.fileStreamSinkMetadataIgnored && fs.exists(logPath)) {
+ val fileIndex = new MetadataLogFileIndex(sparkSession, path, Map.empty,
None)
+ if (fileIndex.allFiles().nonEmpty) {
+ val recordedColumnNames =
fileIndex.partitionSpec().partitionColumns.fieldNames.toSeq
+ if (recordedColumnNames.length != partitionColumnNames.length ||
+ !recordedColumnNames.zip(partitionColumnNames).forall {
+ case (recorded, expected) => sqlConf.resolver(recorded, expected)
+ }) {
+ throw QueryCompilationErrors.fileStreamSinkPartitioningMismatchError(
+ tableName, partitionColumnNames, recordedColumnNames)
+ }
+ }
+ }
+ }
+
def checkEscapedMetadataPath(fs: FileSystem, metadataPath: Path, sqlConf:
SQLConf): Unit = {
if
(sqlConf.getConf(SQLConf.STREAMING_CHECKPOINT_ESCAPED_PATH_CHECK_ENABLED)
&& StreamExecution.containsSpecialCharsInPath(metadataPath)) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
index da41abe76bd9..e8d802a4042f 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
@@ -20,9 +20,10 @@ package org.apache.spark.sql.streaming.test
import java.io.File
import java.util
+import org.apache.hadoop.fs.Path
import org.scalatest.BeforeAndAfter
-import org.apache.spark.sql.{AnalysisException, Row, SaveMode}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
@@ -353,6 +354,223 @@ 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"
+ // An empty `partitionBy()` is the same as no `partitionBy`, which Spark
Connect always sends.
+ Seq(None, Some(Seq.empty[String]), 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(
+ "PARTITIONED BY (p)" -> Seq("i"),
+ "PARTITIONED BY (p)" -> Seq("i", "p"),
+ "" -> Seq("p")).foreach { case (partitionSpec, partitionColumns) =>
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (i INT, p INT) USING parquet
$partitionSpec")
+ checkError(
+ exception = intercept[AnalysisException] {
+ startQuery(df, Some(partitionColumns))
+ },
+ condition = "_LEGACY_ERROR_TEMP_1163",
+ parameters = Map(
+ "tableName" -> s"spark_catalog.default.$tableName",
+ "specifiedPartCols" -> partitionColumns.mkString(", "),
+ "existingPartCols" -> (if (partitionSpec.isEmpty) "" else "p")))
+ }
+ }
+
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (i INT, p INT) USING parquet PARTITIONED
BY (p)")
+ checkError(
+ exception = intercept[AnalysisException] {
+ startQuery(df, Some(Seq("x")))
+ },
+ condition = "COLUMN_NOT_DEFINED_IN_TABLE",
+ parameters = Map(
+ "colType" -> "partition",
+ "colName" -> "`x`",
+ "tableName" -> s"`spark_catalog`.`default`.`$tableName`",
+ "tableCols" -> "`i`, `p`"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ startQuery(df.select("i"), None)
+ },
+ condition = "_LEGACY_ERROR_TEMP_1196",
+ parameters = Map(
+ "columnType" -> "Partition",
+ "columnName" -> "p",
+ "validColumnNames" -> "i"))
+ }
+ }
+
+ test("SPARK-60665: partitionBy of an existing table keeps the order of the
partition columns") {
+ val tableName = "partition_test"
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (i INT, a INT, b INT) USING parquet
PARTITIONED BY (a, b)")
+ withTempDir { checkpointDir =>
+ val input = MemoryStream[Int]
+ val writer = input.toDF().selectExpr("value AS i", "value % 2 AS a",
"value % 3 AS b")
+ .writeStream
+ .format("parquet")
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ checkError(
+ exception = intercept[AnalysisException] {
+ writer.partitionBy("b", "a").toTable(tableName)
+ },
+ condition = "_LEGACY_ERROR_TEMP_1163",
+ parameters = Map(
+ "tableName" -> s"spark_catalog.default.$tableName",
+ "specifiedPartCols" -> "b, a",
+ "existingPartCols" -> "a, b"))
+ val query = writer.partitionBy("a", "b").toTable(tableName)
+ try {
+ input.addData(1, 2, 3)
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+ }
+ checkAnswer(spark.table(tableName), Seq(Row(1, 1, 1), Row(2, 0, 2),
Row(3, 1, 0)))
+ assert(spark.table(tableName).inputFiles.map { file =>
+ val dir = new Path(file).getParent
+ s"${dir.getParent.getName}/${dir.getName}"
+ }.toSet === Set("a=1/b=1", "a=0/b=2", "a=1/b=0"))
+ }
+ }
+
+ test("SPARK-60665: do not write to a table in a layout different from its
committed files") {
+ val tableName = "layout_test"
+ // Writes to the location of the table with the given partitioning like
`toTable` did before.
+ def writeToPath(path: String, partitionColumns: Seq[String]): Unit = {
+ withTempDir { checkpointDir =>
+ val input = MemoryStream[Int]
+ val query = input.toDF().selectExpr("value AS i", "value % 2 AS p")
+ .writeStream
+ .format("parquet")
+ .partitionBy(partitionColumns: _*)
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .start(path)
+ try {
+ input.addData(1, 2)
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+ }
+ }
+
+ Seq(
+ ("PARTITIONED BY (p)", Seq.empty[String], "", "p"),
+ ("PARTITIONED BY (p)", Seq("i"), "i", "p"),
+ ("", Seq("p"), "p", "")).foreach {
+ case (partitionSpec, committedColumns, recordedColumns, expectedColumns)
=>
+ withTempDir { location =>
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (i INT, p INT) USING parquet
$partitionSpec " +
+ s"LOCATION '${location.toURI}'")
+ writeToPath(location.getCanonicalPath, committedColumns)
+ def committedFiles: Set[String] =
+
Utils.recursiveList(location).filter(_.isFile).map(_.getPath).toSet
+ val filesBefore = committedFiles
+
+ withTempDir { checkpointDir =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ MemoryStream[Int].toDF().selectExpr("value AS i", "value % 2
AS p")
+ .writeStream
+ .format("parquet")
+ .option("checkpointLocation",
checkpointDir.getCanonicalPath)
+ .toTable(tableName)
+ },
+ condition = "STREAMING_FILE_SINK_PARTITIONING_MISMATCH",
+ parameters = Map(
+ "tableName" -> s"spark_catalog.default.$tableName",
+ "recordedPartCols" -> recordedColumns,
+ "expectedPartCols" -> expectedColumns))
+ }
+ assert(committedFiles === filesBefore)
+ if (committedColumns.isEmpty) {
+ checkAnswer(spark.table(tableName), Seq(Row(1, 1), Row(2, 0)))
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60665: append to a table with committed files in the layout of
the table") {
+ val tableName = "layout_test"
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (i INT, p INT) USING parquet PARTITIONED
BY (p)")
+ withTempDir { checkpointDir =>
+ val input = MemoryStream[Int]
+ // The first run commits no files, which does not record any layout.
The others restart
+ // the query from the same checkpoint and append to the files
committed by the former.
+ Seq((None, Seq.empty[Int]), (None, Seq(1, 2)), (Some(Seq("P")), Seq(3,
4)))
+ .foreach { case (partitionColumns, values) =>
+ 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 {
+ if (values.nonEmpty) {
+ input.addData(values: _*)
+ }
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+ }
+ }
+ checkAnswer(spark.table(tableName), Seq(Row(1, 1), Row(2, 0), Row(3, 1),
Row(4, 0)))
+ }
+ }
+
test("write: write to new table with clusterBy") {
val tableIdentifier = "testcat.cluster_test"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
