You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sz...@apache.org on 2026/08/20 03:12:24 UTC
(spark) branch master updated: [SPARK-58389][SQL][FOLLOWUP] Pass table-state options while loading write targets
This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7c2f2c4ce950 [SPARK-58389][SQL][FOLLOWUP] Pass table-state options while loading write targets
7c2f2c4ce950 is described below
commit 7c2f2c4ce9505dd2c3329facd42f7c4b2e7e0a21
Author: yyanyy <yy...@gmail.com>
AuthorDate: Wed Aug 19 20:12:11 2026 -0700
[SPARK-58389][SQL][FOLLOWUP] Pass table-state options while loading write targets
### What changes were proposed in this pull request?
This PR follows up on #57582 and has been rebased on #57799's `TableCatalog.tableStateOptionKeys()` contract.
It centralizes catalog-backed batch-write target loads in `CatalogV2Util` write-loading helpers and applies that contract consistently:
* `TableCatalog.loadTable(Identifier, TableContext, CaseInsensitiveStringMap)` receives the required write privileges plus only the user options declared by the catalog as table-state options;
* the resolved `DataSourceV2Relation`, logical write command, and `LogicalWriteInfo` retain the complete user option bag for actual write planning;
* the behavior covers `DataFrameWriter` (`SupportsCatalogOptions`, `insertInto`, and `saveAsTable`), existing-table `DataFrameWriterV2` operations (`append`, `overwrite`, and `overwritePartitions`), schema-evolution reloads, staging and non-staging CTAS/RTAS fallback loads, transaction re-resolution, and row-level DELETE/UPDATE/MERGE target reloads;
* the `SupportsCatalogOptions` save path now requests `INSERT` or `INSERT, DELETE` privileges when loading its write target;
* Spark-recognized time-travel options are rejected consistently on catalog-backed Data Source V2 writes, including creation and replacement through the DataFrame writer APIs; SQL CTAS/RTAS `OPTIONS` remain table properties;
* mixed-case option keys retain their original casing when Spark adds or removes its internal write-privilege marker; and
* the `TableCatalog`, `StagingTableCatalog`, and `DataFrameWriter.insertInto` documentation describes the load/write option split.
#### Streaming scope
Streaming writes are intentionally unchanged. `DataStreamWriter.toTable` and transactional streaming write-target re-resolution require a follow-up that propagates options and write privileges together.
### Why are the changes needed?
Batch-write paths need user options both while selecting/authorizing a catalog table and while planning the write. After #57799, however, options that do not select table state must not affect `loadTable`: passing the complete write option bag would violate the catalog contract and could create inconsistent table identities or cache behavior.
This change therefore projects catalog-declared table-state options for every batch target load while preserving all user options on the write relation and command. It also preserves option-key casing so the command and target relation agree for mixed-case options.
### Does this PR introduce _any_ user-facing change?
Yes. For catalog-backed DataSource V2 batch writes, only catalog-declared table-state options are passed to the target `loadTable` call, together with write privileges. All accepted user-provided options still reach the actual V2 write. Spark-recognized `versionAsOf` and `timestampAsOf` options (or their configured equivalents) are rejected on catalog-backed Data Source V2 writes, including creation and replacement through `DataFrameWriter` and `DataFrameWriterV2`, rather than being t [...]
### How was this patch tested?
Tests use distinct table-state and write-only options to verify that target loads receive only declared state options and expected privileges, while actual V2 writes receive the complete option bag. Coverage includes `DataFrameWriter`, `DataFrameWriterV2`, `SupportsCatalogOptions`, SQL INSERT/overwrite/DELETE/UPDATE/MERGE, transaction re-resolution and source/target isolation, schema-evolution reloads, staging/non-staging CTAS/RTAS fallback loads, V1 fallback behavior, qualified error [...]
Targeted commands:
* `build/mvn -pl sql/catalyst -am -Dtest=none -DwildcardSuites=org.apache.spark.sql.connector.catalog.CatalogV2UtilSuite test`
* `build/mvn -pl sql/core -am -Dtest=none -DwildcardSuites=<batch-write and all DELETE/UPDATE/MERGE suite variants> test`
Post-review focused commands:
* `build/mvn -pl sql/catalyst -am -Dtest=none -DwildcardSuites=org.apache.spark.sql.connector.catalog.CatalogV2UtilSuite test`
* `build/mvn -pl sql/core -am -Dtest=none -DwildcardSuites=org.apache.spark.sql.connector.DataSourceV2OptionSuite test`
* `build/mvn -pl sql/core -am -Dtest=none -DwildcardSuites=org.apache.spark.sql.connector.DataSourceV2DataFrameSessionCatalogSuite test`
* `build/mvn -pl sql/core -am -Dtest=none -DwildcardSuites=org.apache.spark.sql.connector.SupportsCatalogOptionsSuite test`
* `build/mvn -pl sql/core -am -Dtest=none -DwildcardSuites=org.apache.spark.sql.connector.AppendDataTransactionSuite test`
The original targeted matrix ran 1,089 Scala tests with 0 failures and 0 errors. After the latest review fixes, the five focused suites above ran 128 tests with 0 failures and 0 errors. Each run selected only the listed `wildcardSuites` suite.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex CLI 0.145.0
Closes #57865 from yyanyy/spark-write-path-coverage-20260804.
Authored-by: yyanyy <yy...@gmail.com>
Signed-off-by: Szehon Ho <sz...@gmail.com>
---
docs/sql-migration-guide.md | 2 +
.../org/apache/spark/sql/DataFrameWriter.scala | 4 +-
.../sql/connector/catalog/StagingTableCatalog.java | 12 +-
.../spark/sql/connector/catalog/TableCatalog.java | 8 +-
.../sql/catalyst/analysis/RelationResolution.scala | 47 +-
.../sql/catalyst/analysis/ResolveCatalogs.scala | 22 +
.../catalyst/analysis/ResolveSchemaEvolution.scala | 6 +-
.../spark/sql/catalyst/analysis/unresolved.scala | 7 +-
.../sql/connector/catalog/CatalogV2Util.scala | 53 ++-
.../sql/connector/catalog/CatalogV2UtilSuite.scala | 62 ++-
.../connector/catalog/InMemoryTableCatalog.scala | 6 +
.../apache/spark/sql/connector/catalog/txns.scala | 20 +
.../apache/spark/sql/classic/DataFrameWriter.scala | 36 +-
.../spark/sql/classic/DataFrameWriterV2.scala | 16 +-
.../sql/execution/datasources/v2/V2Writes.scala | 1 -
.../datasources/v2/WriteToDataSourceV2Exec.scala | 23 +-
.../sql/connector/AppendDataTransactionSuite.scala | 110 ++++-
.../DataSourceV2DataFrameSessionCatalogSuite.scala | 23 +
.../sql/connector/DataSourceV2OptionSuite.scala | 495 ++++++++++++++++++---
.../sql/connector/DeleteFromTableSuiteBase.scala | 53 ++-
.../sql/connector/MergeIntoTableSuiteBase.scala | 15 +-
.../sql/connector/RowLevelOperationSuiteBase.scala | 36 +-
.../connector/SupportsCatalogOptionsSuite.scala | 78 +++-
.../spark/sql/connector/UpdateTableSuiteBase.scala | 7 +-
24 files changed, 1002 insertions(+), 140 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index be38201fa3b3..c07768cd3c99 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -44,6 +44,8 @@ license: |
- Since Spark 4.3, `unix_seconds`, `unix_millis`, and `unix_micros` accept `TIMESTAMP_NTZ` and the nanosecond-precision timestamp types directly, reading them with no time-zone shift. Previously these functions accepted only `TIMESTAMP_LTZ`; a `TIMESTAMP_NTZ` or nanosecond-timestamp argument was rejected with a `DATATYPE_MISMATCH` error. This is a new capability and does not change the result of any query that previously succeeded.
- Since Spark 4.3, `hash()` and `xxhash64()` include the `days` field of `CalendarInterval` when computing the hash, so their output for interval values differs from earlier releases. Previously the codegen path dropped `days`, disagreeing with interpreted evaluation.
- Since Spark 4.3, when a `SELECT` or `INSERT` statement references the same table more than once with different `WITH (...)` options (for example a self-join, or `INSERT INTO t WITH (...) SELECT * FROM t WITH (...)`), each reference now uses its own options instead of the second reference silently inheriting the first reference's options via the analyzer's relation cache.
+- Since Spark 4.3, `DataFrameWriter.save()` on a `SupportsCatalogOptions` source loads its write target with the required `TableWritePrivilege`s (`INSERT`, or `INSERT` and `DELETE` for `SaveMode.Overwrite`), matching `insertInto` and `saveAsTable`. A catalog that enforces these privileges may now reject an unauthorized `save()` that previously bypassed this authorization check.
+- Since Spark 4.3, the Spark-recognized time-travel options (`versionAsOf` and `timestampAsOf`, or the keys configured by `spark.sql.timeTravelVersionKey` and `spark.sql.timeTravelTimestampKey`) are rejected with `UNSUPPORTED_FEATURE.TIME_TRAVEL` on catalog-backed Data Source V2 writes, including table creation and replacement through `DataFrameWriterV2`, because writes must target the current table state rather than a historical version. Previously, Spark passed them to the connector as [...]
- Since Spark 4.3, `HAVING` is evaluated before window functions when the `SELECT` list also contains generator functions such as `explode`. Previously, window functions could include groups removed by `HAVING` and produce incorrect results.
- Since Spark 4.3, the Spark Connect session errors `INVALID_HANDLE.SESSION_CHANGED`/`SESSION_CLOSED`/`SESSION_NOT_FOUND` carry SQLSTATE `08003` instead of `HY000`; the condition names are unchanged. Code matching these errors on SQLSTATE should match `08003` or class `08`.
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/api/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index 71daf86c9404..d9a86424f419 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -267,7 +267,9 @@ abstract class DataFrameWriter[T] {
* +---+---+
* }}}
*
- * Because it inserts data to an existing table, format or options will be ignored.
+ * Because it inserts data to an existing table, the format is ignored. For data source V2
+ * tables, catalog-declared table-state options are forwarded to the table load and all options
+ * are forwarded to the write; for V1 tables the options are ignored.
* @since 1.4.0
*/
def insertInto(tableName: String): Unit
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/StagingTableCatalog.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/StagingTableCatalog.java
index 6811ea380b3a..bd8a34cb758e 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/StagingTableCatalog.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/StagingTableCatalog.java
@@ -99,7 +99,9 @@ public interface StagingTableCatalog extends TableCatalog {
* @param ident a table identifier
* @param tableInfo information about the table
* @return metadata for the new table. This can be null if the catalog does not support atomic
- * creation for this table. Spark will call {@link #loadTable(Identifier)} later.
+ * creation for this table. Spark will call
+ * {@link #loadTable(Identifier, TableContext, CaseInsensitiveStringMap)} later,
+ * forwarding the catalog-declared table-state options and required privileges.
* @throws TableAlreadyExistsException If a table or view already exists for the identifier
* @throws UnsupportedOperationException If a requested partition transform is not supported
* @throws NoSuchNamespaceException If the identifier namespace does not exist (optional)
@@ -163,7 +165,9 @@ public interface StagingTableCatalog extends TableCatalog {
* @param ident a table identifier
* @param tableInfo information about the table
* @return metadata for the new table. This can be null if the catalog does not support atomic
- * creation for this table. Spark will call {@link #loadTable(Identifier)} later.
+ * creation for this table. Spark will call
+ * {@link #loadTable(Identifier, TableContext, CaseInsensitiveStringMap)} later,
+ * forwarding the catalog-declared table-state options and required privileges.
* @throws UnsupportedOperationException If a requested partition transform is not supported
* @throws NoSuchNamespaceException If the identifier namespace does not exist (optional)
* @throws NoSuchTableException If the table does not exist
@@ -226,7 +230,9 @@ public interface StagingTableCatalog extends TableCatalog {
* @param ident a table identifier
* @param tableInfo information about the table
* @return metadata for the new table. This can be null if the catalog does not support atomic
- * creation for this table. Spark will call {@link #loadTable(Identifier)} later.
+ * creation for this table. Spark will call
+ * {@link #loadTable(Identifier, TableContext, CaseInsensitiveStringMap)} later,
+ * forwarding the catalog-declared table-state options and required privileges.
* @throws UnsupportedOperationException If a requested partition transform is not supported
* @throws NoSuchNamespaceException If the identifier namespace does not exist (optional)
*/
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
index 5d6e9d01f782..36916f907160 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
@@ -221,7 +221,8 @@ public interface TableCatalog extends CatalogPlugin {
* <p>
* The default implementation ignores {@code stateOptions} and delegates to the existing
* {@code loadTable} overloads based on {@code context}. Catalogs that want to receive the user
- * options while reading a table must override {@link #tableStateOptionKeys()} and this method.
+ * options while loading a table for a read or write must override
+ * {@link #tableStateOptionKeys()} and this method.
* Spark passes only the options declared by {@link #tableStateOptionKeys()}. Spark retains the
* complete user option map on the resolved relation for subsequent scan and write planning.
* <p>
@@ -345,7 +346,10 @@ public interface TableCatalog extends CatalogPlugin {
* @param ident a table identifier
* @param tableInfo information about the table
* @return metadata for the new table. This can be null if getting the metadata for the new table
- * is expensive. Spark will call {@link #loadTable(Identifier)} if needed (e.g. CTAS).
+ * is expensive. Spark will call
+ * {@link #loadTable(Identifier, TableContext, CaseInsensitiveStringMap)} if needed
+ * (e.g. CTAS), forwarding the catalog-declared table-state options and required
+ * privileges.
*
* @throws TableAlreadyExistsException If a table already exists for the identifier
* @throws UnsupportedOperationException If a requested partition transform is not supported
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
index 130ee920cf49..9a61138cd180 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
@@ -161,16 +161,26 @@ class RelationResolution(
def resolveRelation(
u: UnresolvedRelation,
timeTravelSpec: Option[TimeTravelSpec] = None): Option[LogicalPlan] = {
- val timeTravelSpecFromOptions = TimeTravelSpec.fromOptions(
- u.options,
- conf.getConf(SQLConf.TIME_TRAVEL_TIMESTAMP_KEY),
- conf.getConf(SQLConf.TIME_TRAVEL_VERSION_KEY),
- conf.sessionLocalTimeZone
- )
+ val isWriteTarget = u.options.containsKey(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
+ val hasTimeTravelWriteOptions =
+ isWriteTarget && CatalogV2Util.containsTimeTravelOptions(u.options)
+ val timeTravelSpecFromOptions = if (hasTimeTravelWriteOptions) {
+ // Time travel applies to reads only. Defer the option check until the identifier is resolved
+ // so every write API reports the same qualified relation ID, without parsing the options as
+ // a read time-travel specification first.
+ None
+ } else {
+ TimeTravelSpec.fromOptions(
+ u.options,
+ conf.getConf(SQLConf.TIME_TRAVEL_TIMESTAMP_KEY),
+ conf.getConf(SQLConf.TIME_TRAVEL_VERSION_KEY),
+ conf.sessionLocalTimeZone)
+ }
if (timeTravelSpec.nonEmpty && timeTravelSpecFromOptions.nonEmpty) {
throw new AnalysisException("MULTIPLE_TIME_TRAVEL_SPEC", Map.empty[String, String])
}
val finalTimeTravelSpec = timeTravelSpec.orElse(timeTravelSpecFromOptions)
+ val isTimeTravel = finalTimeTravelSpec.isDefined || hasTimeTravelWriteOptions
val identifier = u.multipartIdentifier
// system.session.v (3 parts): only local temp view by name; same as SessionCatalog matching.
@@ -179,7 +189,7 @@ class RelationResolution(
return resolveTempView(
normalized,
u.isStreaming,
- finalTimeTravelSpec.isDefined
+ isTimeTravel
)
}
@@ -189,7 +199,7 @@ class RelationResolution(
identifier.head.equalsIgnoreCase(CatalogManager.SESSION_NAMESPACE)) {
val viewNameOnly = Seq(identifier.last)
val tempSession = () =>
- resolveTempView(viewNameOnly, u.isStreaming, finalTimeTravelSpec.isDefined)
+ resolveTempView(viewNameOnly, u.isStreaming, isTimeTravel)
val persistentSessionDb = () =>
tryResolvePersistent(u, identifier, finalTimeTravelSpec)
return if (conf.prioritizeSystemCatalog) {
@@ -205,7 +215,7 @@ class RelationResolution(
return resolveTempView(
identifier,
u.isStreaming,
- finalTimeTravelSpec.isDefined
+ isTimeTravel
).orElse(tryResolvePersistent(u, identifier, finalTimeTravelSpec))
}
@@ -215,7 +225,7 @@ class RelationResolution(
for (step <- steps) {
val result = step match {
case SessionScopeStep =>
- resolveTempView(identifier, u.isStreaming, finalTimeTravelSpec.isDefined)
+ resolveTempView(identifier, u.isStreaming, isTimeTravel)
case PersistentCatalogStep(prefix) =>
tryResolvePersistent(u, prefix ++ identifier, finalTimeTravelSpec)
}
@@ -236,15 +246,19 @@ class RelationResolution(
val planId = u.getTagValue(LogicalPlan.PLAN_ID_TAG)
val writePrivileges = u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
val finalOptions = u.clearWritePrivileges.options
- // Time travel applies to reads only; reject it on a write target (reachable via the option
- // form, e.g. `INSERT INTO t WITH ('versionAsOf' = ...)`) with a user-facing error.
+ if (writePrivileges != null) {
+ CatalogV2Util.rejectTimeTravelOptionsForWrite(catalog, ident, finalOptions)
+ }
+ // Time travel applies to reads only; reject an explicit time-travel specification on a
+ // write target with a user-facing error.
if (finalTimeTravelSpec.nonEmpty && writePrivileges != null) {
- throw QueryCompilationErrors.timeTravelUnsupportedError(toSQLId(identifier))
+ throw QueryCompilationErrors.timeTravelUnsupportedError(
+ toSQLId(ident.toQualifiedNameParts(catalog)))
}
val key = toCacheKey(catalog, ident, finalTimeTravelSpec, finalOptions)
// A reference that requires write privileges is never served from the per-query relation
- // cache. The catalog authorizes the write in `loadTable(ident, writePrivileges)` below, and
- // a cache hit would skip that call entirely. The hit happens whenever the write target is
+ // cache. The catalog authorizes the write during the uncached `loadTable` below, and a
+ // cache hit would skip that call entirely. The hit happens whenever the write target is
// also read in the same statement -- the target is resolved after its query (see
// `ResolveRelations`), so it finds the relation the query already put in the cache, e.g.
// for `INSERT INTO t SELECT * FROM t`.
@@ -548,6 +562,9 @@ class RelationResolution(
*/
private def loadRelation(ref: V2TableReference): LogicalPlan = {
val resolvedCatalog = catalogManager.catalog(ref.catalog.name).asTableCatalog
+ // Only WriteTargetContext gets here (the sole non-cacheable context); it is currently used by
+ // transactional streaming writes and does not retain required write privileges. Keep the
+ // legacy load unchanged; options and privileges must be handled together in a follow-up.
val table = resolvedCatalog.loadTable(ref.identifier)
createRelation(ref, resolvedCatalog, table)
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala
index aa119b4d5b1e..6fc196774a04 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala
@@ -30,6 +30,7 @@ import org.apache.spark.sql.catalyst.util.SparkCharVarcharUtils.replaceCharVarch
import org.apache.spark.sql.connector.catalog._
import org.apache.spark.sql.errors.DataTypeErrors.toSQLId
import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.ArrayImplicits._
/**
@@ -127,6 +128,18 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
val resolvedIdentifier = resolveIdentifier(nameParts, allowTemp, Nil)
c.copy(name = resolvedIdentifier)
+ case c @ CreateTableAsSelect(
+ UnresolvedIdentifier(nameParts, allowTemp), _, _, _, writeOptions, _, _) =>
+ val resolvedIdentifier = resolveIdentifier(nameParts, allowTemp, Nil)
+ rejectTimeTravelOptionsForWrite(resolvedIdentifier, writeOptions)
+ c.copy(name = resolvedIdentifier)
+
+ case r @ ReplaceTableAsSelect(
+ UnresolvedIdentifier(nameParts, allowTemp), _, _, _, writeOptions, _, _) =>
+ val resolvedIdentifier = resolveIdentifier(nameParts, allowTemp, Nil)
+ rejectTimeTravelOptionsForWrite(resolvedIdentifier, writeOptions)
+ r.copy(name = resolvedIdentifier)
+
case UnresolvedIdentifier(nameParts, allowTemp) =>
resolveIdentifier(nameParts, allowTemp, Nil)
@@ -159,6 +172,15 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
}
}
+ private def rejectTimeTravelOptionsForWrite(
+ resolvedIdentifier: ResolvedIdentifier,
+ writeOptions: Map[String, String]): Unit = {
+ CatalogV2Util.rejectTimeTravelOptionsForWrite(
+ resolvedIdentifier.catalog,
+ resolvedIdentifier.identifier,
+ new CaseInsensitiveStringMap(writeOptions.asJava))
+ }
+
private def isSystemBuiltinName(nameParts: Seq[String]): Boolean = {
nameParts.length == 3 &&
nameParts(0).equalsIgnoreCase(CatalogManager.SYSTEM_CATALOG_NAME) &&
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala
index 09f22c16bfd1..08237128358a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.jdk.CollectionConverters._
import scala.util.control.NonFatal
import org.apache.spark.SparkThrowable
@@ -27,7 +26,7 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-import org.apache.spark.sql.connector.catalog.{Identifier, SupportsSchemaEvolution, Table, TableCatalog, TableChange}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, SupportsSchemaEvolution, Table, TableCatalog, TableChange}
import org.apache.spark.sql.connector.catalog.TableChange.ColumnChange
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, ExtractV2CatalogAndIdentifier, ExtractV2Table}
@@ -51,7 +50,8 @@ object ResolveSchemaEvolution extends Rule[LogicalPlan] {
write.table match {
case relation @ ExtractV2CatalogAndIdentifier(catalog, ident) =>
evolveSchema(catalog, ident, write.pendingSchemaChanges)
- val newTable = catalog.loadTable(ident, write.writePrivileges.asJava)
+ val newTable = CatalogV2Util.loadTableForV2Write(
+ catalog, ident, write.writePrivileges, relation.options)
val writeWithNewTarget = replaceWriteTarget(write, relation, newTable)
val remainingChanges = writeWithNewTarget.pendingSchemaChanges
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
index fc51dd8c72d4..50b2970544dc 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
@@ -144,7 +144,9 @@ case class UnresolvedRelation(
def requireWritePrivileges(privileges: Set[TableWritePrivilege]): UnresolvedRelation = {
if (privileges.nonEmpty) {
val newOptions = new java.util.HashMap[String, String]
- newOptions.putAll(options)
+ // CaseInsensitiveStringMap's Map view exposes lowercase keys. Copy the original map to
+ // preserve user-provided key casing when adding the internal marker.
+ newOptions.putAll(options.asCaseSensitiveMap())
newOptions.put(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES, privileges.mkString(","))
copy(options = new CaseInsensitiveStringMap(newOptions))
} else {
@@ -155,7 +157,8 @@ case class UnresolvedRelation(
def clearWritePrivileges: UnresolvedRelation = {
if (options.containsKey(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)) {
val newOptions = new java.util.HashMap[String, String]
- newOptions.putAll(options)
+ // Preserve user-provided key casing while removing the internal marker as well.
+ newOptions.putAll(options.asCaseSensitiveMap())
newOptions.remove(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
copy(options = new CaseInsensitiveStringMap(newOptions))
} else {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
index 8a9e77693824..9660ede92121 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
@@ -36,6 +36,8 @@ import org.apache.spark.sql.connector.catalog.TableChange._
import org.apache.spark.sql.connector.catalog.constraints.Constraint
import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
import org.apache.spark.sql.connector.expressions.{ClusterByTransform, LiteralValue, Transform}
+import org.apache.spark.sql.errors.DataTypeErrors.toSQLId
+import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{ArrayType, MapType, Metadata, MetadataBuilder, StructField, StructType}
@@ -491,9 +493,8 @@ private[sql] object CatalogV2Util {
val stateKeys = catalog.asTableCatalog.tableStateOptionKeys.asScala
.map(_.toLowerCase(Locale.ROOT))
.toSet
- val projected = options.entrySet.asScala.collect {
- case entry if stateKeys.contains(entry.getKey.toLowerCase(Locale.ROOT)) =>
- entry.getKey -> entry.getValue
+ val projected = options.asCaseSensitiveMap().asScala.collect {
+ case (key, value) if stateKeys.contains(key.toLowerCase(Locale.ROOT)) => key -> value
}.toMap
new CaseInsensitiveStringMap(projected.asJava)
}
@@ -519,6 +520,52 @@ private[sql] object CatalogV2Util {
catalog.asTableCatalog.loadTable(ident, context, stateOptions)
}
+ /**
+ * Loads a table for a write, forwarding the required privileges and only the write options that
+ * the catalog declares may affect table state. The complete option map remains on the write
+ * relation for write planning.
+ */
+ def loadTableForV2Write(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ writePrivileges: Set[TableWritePrivilege],
+ options: CaseInsensitiveStringMap): Table = {
+ rejectTimeTravelOptionsForWrite(catalog, ident, options)
+ loadTableForWrite(catalog, ident, writePrivileges, options)
+ }
+
+ /**
+ * Loads a table for a write without validating the complete write option map. This is used by
+ * callers that must inspect whether the loaded table falls back to V1 before applying V2-only
+ * option validation.
+ */
+ def loadTableForWrite(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ writePrivileges: Set[TableWritePrivilege],
+ options: CaseInsensitiveStringMap): Table = {
+ val context = new TableContext(null, writePrivileges.asJava)
+ val stateOptions = extractTableStateOptions(catalog, options)
+ catalog.asTableCatalog.loadTable(ident, context, stateOptions)
+ }
+
+ def rejectTimeTravelOptionsForWrite(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ options: CaseInsensitiveStringMap): Unit = {
+ if (containsTimeTravelOptions(options)) {
+ throw QueryCompilationErrors.timeTravelUnsupportedError(
+ toSQLId(ident.toQualifiedNameParts(catalog)))
+ }
+ }
+
+ def containsTimeTravelOptions(options: CaseInsensitiveStringMap): Boolean = {
+ val conf = SQLConf.get
+ Seq(
+ conf.getConf(SQLConf.TIME_TRAVEL_TIMESTAMP_KEY),
+ conf.getConf(SQLConf.TIME_TRAVEL_VERSION_KEY)).exists(options.containsKey)
+ }
+
/**
* Parses the comma-separated write-privileges string (as carried in the internal
* [[org.apache.spark.sql.catalyst.analysis.UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES]]
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala
index f8919a8730f2..e128adebdff8 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala
@@ -17,12 +17,16 @@
package org.apache.spark.sql.connector.catalog
+import org.mockito.ArgumentCaptor
import org.mockito.ArgumentMatchers.{any, eq => mockEq}
import org.mockito.Mockito.{mock, verify, when}
import org.apache.spark.{SparkFunSuite, SparkIllegalArgumentException}
-import org.apache.spark.sql.catalyst.analysis.{AsOfTimestamp, AsOfVersion, TimeTravelSpec}
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.{
+ AsOfTimestamp, AsOfVersion, TimeTravelSpec, UnresolvedRelation}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -103,6 +107,62 @@ class CatalogV2UtilSuite extends SparkFunSuite {
assert(e.getMessage.contains("Cannot set both time travel and write privileges"))
}
+ test("loadTableForV2Write forwards write privileges and only table-state options") {
+ val testCatalog = mock(classOf[TableCatalog])
+ when(testCatalog.tableStateOptionKeys()).thenReturn(java.util.Set.of("state"))
+ val ident = mock(classOf[Identifier])
+ val options = new CaseInsensitiveStringMap(
+ java.util.Map.of("state", "branch", "custom", "value"))
+ val contextCaptor = ArgumentCaptor.forClass(classOf[TableContext])
+
+ CatalogV2Util.loadTableForV2Write(
+ testCatalog, ident, Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE), options)
+
+ val expectedStateOptions =
+ new CaseInsensitiveStringMap(java.util.Map.of("state", "branch"))
+ verify(testCatalog).loadTable(
+ mockEq(ident), contextCaptor.capture(), mockEq(expectedStateOptions))
+ assert(contextCaptor.getValue.timeTravel().isEmpty)
+ assert(contextCaptor.getValue.writePrivileges() ===
+ java.util.Set.of(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
+ }
+
+ test("loadTableForV2Write rejects configured time travel options") {
+ val testCatalog = mock(classOf[TableCatalog])
+ when(testCatalog.name()).thenReturn("testcat")
+ val ident = Identifier.of(Array("ns"), "table")
+ val conf = new SQLConf
+ conf.setConf(SQLConf.TIME_TRAVEL_VERSION_KEY, "customVersion")
+ conf.setConf(SQLConf.TIME_TRAVEL_TIMESTAMP_KEY, "customTimestamp")
+
+ SQLConf.withExistingConf(conf) {
+ Seq("customVersion", "customTimestamp").foreach { key =>
+ val options = new CaseInsensitiveStringMap(java.util.Map.of(key, "value"))
+ val e = intercept[AnalysisException] {
+ CatalogV2Util.loadTableForV2Write(
+ testCatalog, ident, Set(TableWritePrivilege.INSERT), options)
+ }
+ assert(e.getCondition === "UNSUPPORTED_FEATURE.TIME_TRAVEL")
+ }
+ }
+ }
+
+ test("UnresolvedRelation preserves option key case while updating write privileges") {
+ val options = new CaseInsensitiveStringMap(java.util.Map.of(
+ "targetLoadOption", "loadValue",
+ "targetWriteOption", "writeValue"))
+ val relation = UnresolvedRelation(Seq("catalog", "table"), options)
+
+ val withPrivileges = relation.requireWritePrivileges(Set(TableWritePrivilege.INSERT))
+ assert(withPrivileges.options.asCaseSensitiveMap().containsKey("targetLoadOption"))
+ assert(withPrivileges.options.asCaseSensitiveMap().containsKey("targetWriteOption"))
+ assert(!withPrivileges.options.asCaseSensitiveMap().containsKey("targetloadoption"))
+ assert(withPrivileges.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES) === "INSERT")
+
+ val cleared = withPrivileges.clearWritePrivileges
+ assert(cleared.options.asCaseSensitiveMap() === options.asCaseSensitiveMap())
+ }
+
test("TableContext normalizes null time travel and null write privileges to empty") {
val context = new TableContext(null, null)
assert(context.timeTravel().isEmpty)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
index c46f96493ea7..96f18e8b901c 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
@@ -48,6 +48,7 @@ class BasicInMemoryTableCatalog extends TableCatalog {
private var _name: Option[String] = None
private var copyOnLoad: Boolean = false
+ private var stateOptionKeys: util.Set[String] = util.Set.of()
// Records every (TableContext, table-state options) pair passed to the options-aware
// loadTable(), in call order, so tests can verify that the analyzer / DataFrame API correctly
@@ -65,8 +66,13 @@ class BasicInMemoryTableCatalog extends TableCatalog {
override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {
_name = Some(name)
copyOnLoad = options.getBoolean("copyOnLoad", false)
+ stateOptionKeys = Option(options.get("tableStateOptionKeys"))
+ .map(_.split(",").iterator.map(_.trim).filter(_.nonEmpty).toSet.asJava)
+ .getOrElse(util.Set.of())
}
+ override def tableStateOptionKeys(): util.Set[String] = stateOptionKeys
+
override def name: String = _name.get
override def listTables(namespace: Array[String]): Array[Identifier] = {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala
index f2b16b731f41..f48f99994a8c 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala
@@ -136,6 +136,7 @@ class TxnTable(
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
catalog.writeTarget = this
+ lastWriteInfo = info
super.newWriteBuilder(info)
}
@@ -171,6 +172,7 @@ class TxnTable(
class TxnTableCatalog(delegate: InMemoryRowLevelOperationTableCatalog) extends TableCatalog {
private val tables: util.Map[Identifier, TxnTable] = new ConcurrentHashMap[Identifier, TxnTable]()
+ val loadTableCalls: ArrayBuffer[(TableContext, CaseInsensitiveStringMap)] = ArrayBuffer.empty
var writeTarget: TxnTable = _
@@ -178,6 +180,8 @@ class TxnTableCatalog(delegate: InMemoryRowLevelOperationTableCatalog) extends T
override def capabilities: java.util.Set[TableCatalogCapability] = delegate.capabilities
+ override def tableStateOptionKeys(): util.Set[String] = delegate.tableStateOptionKeys()
+
override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {}
override def listTables(namespace: Array[String]): Array[Identifier] = {
@@ -196,6 +200,22 @@ class TxnTableCatalog(delegate: InMemoryRowLevelOperationTableCatalog) extends T
})
}
+ override def loadTable(
+ ident: Identifier,
+ context: TableContext,
+ options: CaseInsensitiveStringMap): Table = {
+ loadTableCalls += ((context, options))
+ super.loadTable(ident, context, options)
+ }
+
+ override def loadTable(ident: Identifier, version: String): Table = {
+ delegate.loadTable(ident, version)
+ }
+
+ override def loadTable(ident: Identifier, timestamp: Long): Table = {
+ delegate.loadTable(ident, timestamp)
+ }
+
override def alterTable(ident: Identifier, changes: TableChange*): Table = {
// AlterTable may be called by ResolveSchemaEvolution when schema evolution is enabled. Thus,
// it needs to be transactional. The schema changes are only propagated to the delegate at
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala
index 96714a419511..3d0036f86522 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala
@@ -177,7 +177,9 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) extends sql.DataFram
val catalog = CatalogV2Util.getTableProviderCatalog(
supportsExtract, catalogManager, dsOptions)
- (catalog.loadTable(ident), Some(catalog), Some(ident))
+ val table = CatalogV2Util.loadTableForV2Write(
+ catalog, ident, getWritePrivileges, dsOptions)
+ (table, Some(catalog), Some(ident))
case _: TableProvider =>
val t = getTable
if (t.supports(BATCH_WRITE)) {
@@ -315,7 +317,9 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) extends sql.DataFram
* +---+---+
* }}}
*
- * Because it inserts data to an existing table, format or options will be ignored.
+ * Because it inserts data to an existing table, the format is ignored. For data source V2
+ * tables, catalog-declared table-state options are forwarded to the table load and all options
+ * are forwarded to the write; for V1 tables the options are ignored.
*
* @since 1.4.0
*/
@@ -354,13 +358,14 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) extends sql.DataFram
}
private def insertIntoCommand(catalog: CatalogPlugin, ident: Identifier): LogicalPlan = {
- import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
-
- val table = catalog.asTableCatalog.loadTable(ident, getWritePrivileges.toSet.asJava) match {
+ val tableOptions = new CaseInsensitiveStringMap(extraOptions.toMap.asJava)
+ val table = CatalogV2Util.loadTableForWrite(
+ catalog, ident, getWritePrivileges, tableOptions) match {
case _: V1Table =>
return insertIntoCommand(TableIdentifier(ident.name(), ident.namespace().headOption))
case t =>
- DataSourceV2Relation.create(t, Some(catalog), Some(ident))
+ CatalogV2Util.rejectTimeTravelOptionsForWrite(catalog, ident, tableOptions)
+ DataSourceV2Relation.create(t, Some(catalog), Some(ident), tableOptions)
}
curmode match {
@@ -486,18 +491,27 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) extends sql.DataFram
v2ProviderOpt: Option[TableProvider],
ident: Identifier,
nameParts: Seq[String]): LogicalPlan = {
- val tableOpt = try Option(catalog.loadTable(ident, getWritePrivileges.toSet.asJava)) catch {
+ val tableOptions = new CaseInsensitiveStringMap(extraOptions.toMap.asJava)
+ val tableOpt = try {
+ Option(CatalogV2Util.loadTableForWrite(catalog, ident, getWritePrivileges, tableOptions))
+ } catch {
case _: NoSuchTableException => None
}
- (curmode, tableOpt) match {
- case (_, Some(_: V1Table)) =>
+ tableOpt match {
+ case Some(_: V1Table) =>
assertSchemaEvolutionNotEnabledForV1Write()
- saveAsV1TableCommand(TableIdentifier(ident.name(), ident.namespace().headOption))
+ return saveAsV1TableCommand(TableIdentifier(ident.name(), ident.namespace().headOption))
+ case _ => ()
+ }
+
+ CatalogV2Util.rejectTimeTravelOptionsForWrite(catalog, ident, tableOptions)
+ (curmode, tableOpt) match {
case (SaveMode.Append, Some(table)) =>
checkPartitioningMatchesV2Table(table)
- val v2Relation = DataSourceV2Relation.create(table, Some(catalog), Some(ident))
+ val v2Relation =
+ DataSourceV2Relation.create(table, Some(catalog), Some(ident), tableOptions)
AppendData.byName(v2Relation, df.logicalPlan, extraOptions.toMap, _withSchemaEvolution)
case (SaveMode.Overwrite, _) =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriterV2.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriterV2.scala
index 0da60e98cdbe..a8cc7c79f7fc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriterV2.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriterV2.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.classic
import java.util
import scala.collection.mutable
-import scala.jdk.CollectionConverters.MapHasAsScala
+import scala.jdk.CollectionConverters._
import org.apache.spark.annotation.Experimental
import org.apache.spark.sql
@@ -29,12 +29,14 @@ import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, UnresolvedF
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, Literal}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege
import org.apache.spark.sql.connector.catalog.TableWritePrivilege._
import org.apache.spark.sql.connector.expressions._
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
/**
* Interface used to write a [[org.apache.spark.sql.classic.Dataset]] to external storage using
@@ -197,7 +199,7 @@ final class DataFrameWriterV2[T] private[sql](table: String, ds: Dataset[T])
private[sql] def appendCommand(): LogicalPlan = {
AppendData.byName(
- UnresolvedRelation(tableName).requireWritePrivileges(Set(INSERT)),
+ createUnresolvedWriteTarget(Set(INSERT)),
logicalPlan, options.toMap, withSchemaEvolution = _withSchemaEvolution)
}
@@ -209,7 +211,7 @@ final class DataFrameWriterV2[T] private[sql](table: String, ds: Dataset[T])
private[sql] def overwriteCommand(condition: Column): LogicalPlan = {
OverwriteByExpression.byName(
- UnresolvedRelation(tableName).requireWritePrivileges(Set(INSERT, DELETE)),
+ createUnresolvedWriteTarget(Set(INSERT, DELETE)),
logicalPlan, expression(condition), options.toMap, _withSchemaEvolution)
}
@@ -221,10 +223,16 @@ final class DataFrameWriterV2[T] private[sql](table: String, ds: Dataset[T])
private[sql] def overwritePartitionsCommand(): LogicalPlan = {
OverwritePartitionsDynamic.byName(
- UnresolvedRelation(tableName).requireWritePrivileges(Set(INSERT, DELETE)),
+ createUnresolvedWriteTarget(Set(INSERT, DELETE)),
logicalPlan, options.toMap, _withSchemaEvolution)
}
+ private def createUnresolvedWriteTarget(
+ privileges: Set[TableWritePrivilege]): UnresolvedRelation = {
+ val tableOptions = new CaseInsensitiveStringMap(options.toMap.asJava)
+ UnresolvedRelation(tableName, tableOptions).requireWritePrivileges(privileges)
+ }
+
/**
* Wrap an action to track the QueryExecution and time cost, then report to the user-registered
* callback functions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala
index be8e96e8034d..5340937115d9 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala
@@ -129,7 +129,6 @@ object V2Writes extends Rule[LogicalPlan] with PredicateHelper {
commandOptions: Map[String, String],
dsOptions: Map[String, String]): Map[String, String] = {
// for DataFrame API cases, same options are carried by both Command and DataSourceV2Relation
- // for DataFrameV2 API cases, options are only carried by Command
// for SQL cases, options are only carried by DataSourceV2Relation
assert(commandOptions == dsOptions || commandOptions.isEmpty || dsOptions.isEmpty)
commandOptions ++ dsOptions
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
index d280076622b0..b9d01153e5ee 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
@@ -40,6 +40,7 @@ import org.apache.spark.sql.execution.{EmptyRDDWithPartitions, QueryExecution, S
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.metric.{CustomMetrics, SQLLastAttemptMetric, SQLLastAttemptMetrics, SQLMetric, SQLMetrics}
import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.sql.util.SchemaValidationMode.PROHIBIT_CHANGES
import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.Utils
@@ -98,7 +99,7 @@ case class CreateTableAsSelectExec(
.withProperties(properties.asJava)
.build()
val table = Option(catalog.createTable(ident, tableInfo))
- .getOrElse(catalog.loadTable(ident, Set(TableWritePrivilege.INSERT).asJava))
+ .getOrElse(loadTableForInsert(catalog, ident, writeOptions))
val result = writeToTable(catalog, table, writeOptions, ident, query, overwrite = false)
transaction.foreach(TransactionUtils.commit)
result
@@ -142,7 +143,7 @@ case class AtomicCreateTableAsSelectExec(
.withProperties(properties.asJava)
.build()
val stagedTable = Option(catalog.stageCreate(ident, tableInfo)
- ).getOrElse(catalog.loadTable(ident, Set(TableWritePrivilege.INSERT).asJava))
+ ).getOrElse(loadTableForInsert(catalog, ident, writeOptions))
writeToTable(catalog, stagedTable, writeOptions, ident, query, overwrite = false)
}
}
@@ -205,7 +206,7 @@ case class ReplaceTableAsSelectExec(
.withProperties(properties.asJava)
.build()
val table = Option(catalog.createTable(ident, tableInfo))
- .getOrElse(catalog.loadTable(ident, Set(TableWritePrivilege.INSERT).asJava))
+ .getOrElse(loadTableForInsert(catalog, ident, writeOptions))
val result = writeToTable(
catalog, table, writeOptions, ident, refreshedQuery,
overwrite = true, refreshPhaseEnabled = false)
@@ -274,8 +275,7 @@ case class AtomicReplaceTableAsSelectExec(
ident,
CatalogV2Util.searchPathForTableIdentifier(catalog, ident))
}
- val table = Option(staged).getOrElse(
- catalog.loadTable(ident, Set(TableWritePrivilege.INSERT).asJava))
+ val table = Option(staged).getOrElse(loadTableForInsert(catalog, ident, writeOptions))
writeToTable(catalog, table, writeOptions, ident, query, overwrite = true)
}
}
@@ -964,6 +964,15 @@ case class DeltaWithMetadataWritingSparkTask(
private[v2] trait V2CreateTableAsSelectBaseExec extends LeafV2CommandExec {
override def output: Seq[Attribute] = Nil
+ protected def loadTableForInsert(
+ catalog: TableCatalog,
+ ident: Identifier,
+ writeOptions: Map[String, String]): Table = {
+ val options = new CaseInsensitiveStringMap(writeOptions.asJava)
+ CatalogV2Util.loadTableForV2Write(
+ catalog, ident, Set(TableWritePrivilege.INSERT), options)
+ }
+
protected def getV2Columns(schema: StructType, forceNullable: Boolean): Array[Column] = {
val rawSchema = CharVarcharUtils.getRawSchema(removeInternalMetadata(schema), conf)
val tableSchema = if (forceNullable) rawSchema.asNullable else rawSchema
@@ -979,7 +988,9 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends LeafV2CommandExec {
overwrite: Boolean,
refreshPhaseEnabled: Boolean = true): Seq[InternalRow] = {
Utils.tryWithSafeFinallyAndFailureCallbacks({
- val relation = DataSourceV2Relation.create(table, Some(catalog), Some(ident))
+ val tableOptions = new CaseInsensitiveStringMap(writeOptions.asJava)
+ val relation =
+ DataSourceV2Relation.create(table, Some(catalog), Some(ident), tableOptions)
val writeCommand = if (overwrite) {
OverwriteByExpression.byPosition(relation, query, Literal.TrueLiteral, writeOptions)
} else {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
index aef9c65550fc..e48d99061671 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
@@ -19,14 +19,70 @@ package org.apache.spark.sql.connector
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.Row
-import org.apache.spark.sql.connector.catalog.{Aborted, Committed}
+import org.apache.spark.sql.connector.catalog.{
+ Aborted,
+ Committed,
+ TableContext,
+ TableWritePrivilege,
+ TimeTravel,
+ Txn,
+ TxnTableCatalog}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
import org.apache.spark.sql.sources
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
+ private val targetLoadOption = "targetLoadOption"
+ private val targetLoadValue = "loadValue"
+ private val targetWriteOption = "targetWriteOption"
+ private val targetWriteValue = "writeValue"
+ private val targetOptionsClause =
+ s"WITH (`$targetLoadOption` = '$targetLoadValue', " +
+ s"`$targetWriteOption` = '$targetWriteValue')"
+
+ private def assertTargetLoadAndWriteOptions(
+ txn: Txn,
+ expectedPrivileges: java.util.Set[TableWritePrivilege],
+ minTargetLoads: Int = 1): Unit = {
+ val targetLoads = txn.catalog.loadTableCalls.filter {
+ case (context, _) => context.writePrivileges() == expectedPrivileges
+ }
+ assert(targetLoads.size >= minTargetLoads,
+ s"expected at least $minTargetLoads target loads with write options")
+ targetLoads.foreach { case (context, options) =>
+ assert(context.writePrivileges() === expectedPrivileges)
+ assert(options.get(targetLoadOption) === targetLoadValue)
+ assert(options.asCaseSensitiveMap().containsKey(targetLoadOption))
+ assert(options.get(targetWriteOption) === null)
+ assert(options.size() === 1)
+ }
+
+ assert(table.lastWriteInfo != null, "the V2 table did not receive LogicalWriteInfo")
+ assert(table.lastWriteInfo.options().get(targetLoadOption) === targetLoadValue)
+ assert(table.lastWriteInfo.options().asCaseSensitiveMap().containsKey(targetLoadOption))
+ assert(table.lastWriteInfo.options().get(targetWriteOption) === targetWriteValue)
+ }
+
+ test("SPARK-58389: transaction catalog honors time travel context") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }""")
+ val pinnedVersion = catalog.loadTable(ident).version()
+ catalog.pinTable(ident, "pinned")
+ append("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 2, "salary": 200, "dep": "software" }""")
+ assert(catalog.loadTable(ident).version() !== pinnedVersion)
+
+ val txnCatalog = new TxnTableCatalog(catalog)
+ val context = new TableContext(
+ new TimeTravel.AsOfVersion("pinned"), java.util.Set.of[TableWritePrivilege]())
+ val loaded = txnCatalog.loadTable(ident, context, CaseInsensitiveStringMap.empty())
+
+ assert(loaded.version() === pinnedVersion)
+ }
+
test("writeTo append with transactional checks") {
// create table with initial data
createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
@@ -35,14 +91,17 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
|""".stripMargin)
// create a source on top of itself that will be fully resolved and analyzed
- val sourceDF = spark.table(tableNameAsString)
+ val sourceDF = spark.read.option("customReadOption", "customValue").table(tableNameAsString)
.where("pk == 1")
.select(col("pk") + 10 as "pk", col("salary"), col("dep"))
sourceDF.queryExecution.assertAnalyzed()
// append data using the DataFrame API
val (txn, txnTables) = executeTransaction {
- sourceDF.writeTo(tableNameAsString).append()
+ sourceDF.writeTo(tableNameAsString)
+ .option(targetLoadOption, targetLoadValue)
+ .option(targetWriteOption, targetWriteValue)
+ .append()
}
// check txn was properly committed and closed
@@ -50,6 +109,22 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
assert(txn.isClosed)
assert(txnTables.size === 1)
assert(table.version() === "2")
+ val sourceLoads = txn.catalog.loadTableCalls.filter {
+ case (context, _) => context.writePrivileges().isEmpty
+ }
+ assert(sourceLoads.nonEmpty, "transaction re-resolution did not reload the source")
+ sourceLoads.foreach { case (context, options) =>
+ assert(context.timeTravel().isEmpty)
+ assert(context.writePrivileges().isEmpty)
+ assert(options.isEmpty)
+ }
+ assertTargetLoadAndWriteOptions(txn, java.util.Set.of(TableWritePrivilege.INSERT))
+ txn.catalog.loadTableCalls.filter {
+ case (context, _) => !context.writePrivileges().isEmpty
+ }.foreach { case (_, options) =>
+ assert(options.get("customReadOption") === null)
+ }
+ assert(table.lastWriteInfo.options().get("customReadOption") === null)
// check the source scan was tracked via the transaction catalog
val targetTxnTable = txnTables(tableNameAsString)
@@ -77,7 +152,8 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
// SQL INSERT INTO using VALUES
val (txn, txnTables) = executeTransaction {
- sql(s"INSERT INTO $tableNameAsString VALUES (3, 300, 'hr'), (4, 400, 'finance')")
+ sql(s"INSERT INTO $tableNameAsString $targetOptionsClause " +
+ "VALUES (3, 300, 'hr'), (4, 400, 'finance')")
}
// check txn was properly committed and closed
@@ -87,6 +163,7 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
// VALUES literal - No catalog tables were scanned
assert(txnTables.isEmpty)
+ assertTargetLoadAndWriteOptions(txn, java.util.Set.of(TableWritePrivilege.INSERT))
// check data was inserted correctly
checkAnswer(
@@ -109,12 +186,12 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
val insertOverwrite = if (isDynamic) {
// OverwritePartitionsDynamic
- s"""INSERT OVERWRITE $tableNameAsString
+ s"""INSERT OVERWRITE $tableNameAsString $targetOptionsClause
|SELECT pk + 10, salary, dep FROM $tableNameAsString WHERE dep = 'hr'
|""".stripMargin
} else {
// OverwriteByExpression
- s"""INSERT OVERWRITE $tableNameAsString
+ s"""INSERT OVERWRITE $tableNameAsString $targetOptionsClause
|PARTITION (dep = 'hr')
|SELECT pk + 10, salary FROM $tableNameAsString WHERE dep = 'hr'
|""".stripMargin
@@ -137,6 +214,8 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
case sources.EqualTo("dep", "hr") => true
case _ => false
})
+ assertTargetLoadAndWriteOptions(
+ txn, java.util.Set.of(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
@@ -159,7 +238,10 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
toDF("pk", "salary", "dep")
val (txn, txnTables) = executeTransaction {
- sourceDF.writeTo(tableNameAsString).overwrite(col("dep") === "hr")
+ sourceDF.writeTo(tableNameAsString)
+ .option(targetLoadOption, targetLoadValue)
+ .option(targetWriteOption, targetWriteValue)
+ .overwrite(col("dep") === "hr")
}
assert(txn.currentState === Committed)
@@ -168,6 +250,8 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
// literal DataFrame source - no catalog tables were scanned
assert(txnTables.isEmpty)
+ assertTargetLoadAndWriteOptions(
+ txn, java.util.Set.of(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
@@ -190,7 +274,10 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
toDF("pk", "salary", "dep")
val (txn, txnTables) = executeTransaction {
- sourceDF.writeTo(tableNameAsString).overwritePartitions()
+ sourceDF.writeTo(tableNameAsString)
+ .option(targetLoadOption, targetLoadValue)
+ .option(targetWriteOption, targetWriteValue)
+ .overwritePartitions()
}
assert(txn.currentState === Committed)
@@ -199,6 +286,8 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
// literal DataFrame source - no catalog tables were scanned
assert(txnTables.isEmpty)
+ assertTargetLoadAndWriteOptions(
+ txn, java.util.Set.of(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
@@ -412,7 +501,8 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
sql(s"INSERT INTO $sourceNameAsString VALUES (3, 300, 'hr', true), (4, 400, 'software', false)")
val (txn, txnTables) = executeTransaction {
- sql(s"INSERT WITH SCHEMA EVOLUTION INTO $tableNameAsString SELECT * FROM $sourceNameAsString")
+ sql(s"INSERT WITH SCHEMA EVOLUTION INTO $tableNameAsString $targetOptionsClause " +
+ s"SELECT * FROM $sourceNameAsString")
}
assert(txn.currentState === Committed)
@@ -420,6 +510,8 @@ class AppendDataTransactionSuite extends RowLevelOperationSuiteBase {
// the new column must be visible in the committed delegate's schema
assert(table.schema.fieldNames.toSeq === Seq("pk", "salary", "dep", "active"))
+ assertTargetLoadAndWriteOptions(
+ txn, java.util.Set.of(TableWritePrivilege.INSERT), minTargetLoads = 2)
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala
index 97cdebe2d32d..d10460181329 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala
@@ -98,6 +98,29 @@ class DataSourceV2DataFrameSessionCatalogSuite
verifyTable("t", df)
}
}
+
+ test("SPARK-58389: time travel options are ignored for V1 table writes") {
+ withTable("t") {
+ sql("CREATE TABLE t(c BIGINT) USING csv")
+ val df = spark.range(1).toDF("c")
+
+ df.write
+ .format(v2Format)
+ .option("versionAsOf", "1")
+ .insertInto("t")
+ verifyTable("t", df)
+
+ // SaveMode.Ignore leaves the existing V1 table unchanged, but the write must still
+ // reach the V1 fallback before Spark rejects the time-travel option.
+ df.write
+ .format(v2Format)
+ .option("versionAsOf", "1")
+ .mode(SaveMode.Ignore)
+ .saveAsTable("t")
+
+ verifyTable("t", df)
+ }
+ }
}
class InMemoryTableSessionCatalog extends TestV2SessionCatalogBase[InMemoryTable] {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
index 82b5938ae813..1490871e4440 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
@@ -20,10 +20,13 @@ package org.apache.spark.sql.connector
import java.util
import java.util.concurrent.atomic.AtomicInteger
+import scala.jdk.CollectionConverters._
+
import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.analysis.{
AnalysisContext,
+ AsOfVersion,
RelationCache,
RelationResolution,
UnresolvedRelation,
@@ -35,13 +38,19 @@ import org.apache.spark.sql.connector.catalog.{
InMemoryBaseTable,
InMemoryCatalog,
InMemoryRowLevelOperationTableCatalog,
+ InMemoryTableCatalog,
+ StagedTable,
+ StagingTableCatalog,
Table,
TableChange,
+ TableInfo,
TableWritePrivilege,
TimeTravel}
-import org.apache.spark.sql.execution.CommandResultExec
+import org.apache.spark.sql.connector.write.Write
+import org.apache.spark.sql.execution.{CommandResultExec, QueryExecution, SparkPlan}
import org.apache.spark.sql.execution.datasources.v2._
import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.util.CaseInsensitiveStringMap
class LoadCountingInMemoryCatalog extends InMemoryCatalog {
@@ -60,6 +69,34 @@ class StateAwareInMemoryCatalog extends LoadCountingInMemoryCatalog {
util.Set.of("snapshot", UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
}
+class NullReturningInMemoryCatalog extends InMemoryCatalog {
+ override def createTable(ident: Identifier, tableInfo: TableInfo): Table = {
+ super.createTable(ident, tableInfo)
+ null
+ }
+}
+
+class NullReturningStagingInMemoryCatalog extends InMemoryCatalog with StagingTableCatalog {
+ override def stageCreate(ident: Identifier, tableInfo: TableInfo): StagedTable = {
+ createTable(ident, tableInfo)
+ null
+ }
+
+ override def stageReplace(ident: Identifier, tableInfo: TableInfo): StagedTable = {
+ dropTable(ident)
+ createTable(ident, tableInfo)
+ null
+ }
+
+ override def stageCreateOrReplace(ident: Identifier, tableInfo: TableInfo): StagedTable = {
+ if (tableExists(ident)) {
+ dropTable(ident)
+ }
+ createTable(ident, tableInfo)
+ null
+ }
+}
+
class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
import testImplicits._
@@ -92,6 +129,71 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}, s"expected only snapshot=$expectedSnapshot to be forwarded, got: $loadOptions")
}
+ private val loadOption = "load-Option"
+ private val loadOptionValue = "load-value"
+ private val writeOption = "write-option"
+ private val writeOptionValue = "write-value"
+
+ private def testWithLoadOptionAsTableState(testName: String)(f: => Unit): Unit = {
+ test(testName) {
+ withSQLConf("spark.sql.catalog.testcat.tableStateOptionKeys" -> loadOption)(f)
+ }
+ }
+
+ private def assertTargetOptions(
+ options: CaseInsensitiveStringMap): org.scalatest.Assertion = {
+ assert(options.get(loadOption) === loadOptionValue)
+ assert(options.get(writeOption) === writeOptionValue)
+ }
+
+ private def assertWriteLoad(
+ tableCatalog: InMemoryTableCatalog,
+ expectedPrivileges: Set[TableWritePrivilege]): Unit = {
+ val matchingCalls = tableCatalog.loadTableCalls.filter {
+ case (context, _) => context.writePrivileges() == expectedPrivileges.asJava
+ }
+ assert(matchingCalls.nonEmpty, "loadTable did not receive the expected write privileges")
+ matchingCalls.foreach { case (context, options) =>
+ assert(context.writePrivileges() === expectedPrivileges.asJava)
+ assert(options.get(loadOption) === loadOptionValue)
+ assert(options.asCaseSensitiveMap().containsKey(loadOption))
+ assert(options.get(writeOption) === null)
+ assert(options.size() === 1)
+ }
+ }
+
+ private def assertWriteLoad(expectedPrivileges: Set[TableWritePrivilege]): Unit = {
+ assertWriteLoad(inMemoryCatalog, expectedPrivileges)
+ }
+
+ private def inMemoryWriteOptions(write: Write): CaseInsensitiveStringMap = {
+ write.toBatch match {
+ case append: InMemoryBaseTable#Append => append.info.options
+ case overwrite: InMemoryBaseTable#TruncateAndAppend => overwrite.info.options
+ case dynamic: InMemoryBaseTable#DynamicOverwrite => dynamic.info.options
+ case other => fail(s"expected a V2 in-memory batch write, got ${other.getClass.getName}")
+ }
+ }
+
+ private def collectInMemoryWriteOptions(plan: SparkPlan): Seq[CaseInsensitiveStringMap] = {
+ val direct = plan.collect {
+ case AppendDataExec(_, _, write, _, _) => inMemoryWriteOptions(write)
+ case OverwriteByExpressionExec(_, _, write, _, _) => inMemoryWriteOptions(write)
+ case OverwritePartitionsDynamicExec(_, _, write, _, _) => inMemoryWriteOptions(write)
+ }
+ val commandResults = plan.collect {
+ case CommandResultExec(_, commandPhysicalPlan, _) =>
+ collectInMemoryWriteOptions(commandPhysicalPlan)
+ }.flatten
+ direct ++ commandResults
+ }
+
+ private def assertV2Write(captured: Seq[QueryExecution]): Unit = {
+ val writeOptions = captured.flatMap(qe => collectInMemoryWriteOptions(qe.executedPlan))
+ assert(writeOptions.nonEmpty, "expected a V2 in-memory batch write")
+ writeOptions.foreach(assertTargetOptions)
+ }
+
test("SPARK-36680: Supports Dynamic Table Options for SQL Select") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
@@ -157,15 +259,19 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
- test("SPARK-49098, SPARK-50286: Supports Dynamic Table Options for SQL Insert") {
+ testWithLoadOptionAsTableState(
+ "SPARK-49098, SPARK-50286: Supports Dynamic Table Options for SQL Insert") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
- val df = sql(s"INSERT INTO $t1 WITH (`write.split-size` = 10) VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
+ val df = sql(s"INSERT INTO $t1 WITH (`$loadOption` = '$loadOptionValue', " +
+ s"`$writeOption` = '$writeOptionValue') VALUES (1, 'a'), (2, 'b')")
var collected = df.queryExecution.optimizedPlan.collect {
case CommandResult(_, AppendData(relation: DataSourceV2Relation, _, _, _, _, _, _), _, _) =>
- assert(relation.options.get("write.split-size") == "10")
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assertTargetOptions(relation.options)
}
assert (collected.size == 1)
@@ -174,9 +280,10 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
_, AppendDataExec(_, _, write, _, _),
_) =>
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#Append]
- assert(append.info.options.get("write.split-size") === "10")
+ assertTargetOptions(append.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT))
val insertResult = sql(s"SELECT * FROM $t1")
checkAnswer(insertResult, Seq(Row(1, "a"), Row(2, "b")))
@@ -284,74 +391,87 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
- test("SPARK-50286: Propagate options for DataFrameWriter Append") {
+ testWithLoadOptionAsTableState("SPARK-50286: Propagate options for DataFrameWriter Append") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ inMemoryCatalog.resetLoadTableCalls()
val captured = withQueryExecutionsCaptured(spark) {
Seq(1 -> "a", 2 -> "b").toDF("id", "data")
.write
- .option("write.split-size", "10")
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
.mode("append")
.insertInto(t1)
}
assert(captured.size === 1)
val qe = captured.head
var collected = qe.optimizedPlan.collect {
- case AppendData(_: DataSourceV2Relation, _, writeOptions, _, _, _, _) =>
- assert(writeOptions("write.split-size") == "10")
+ case AppendData(relation: DataSourceV2Relation, _, writeOptions, _, _, _, _) =>
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assert(writeOptions(loadOption) === loadOptionValue)
+ assert(writeOptions(writeOption) === writeOptionValue)
}
assert (collected.size == 1)
collected = qe.executedPlan.collect {
case AppendDataExec(_, _, write, _, _) =>
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#Append]
- assert(append.info.options.get("write.split-size") === "10")
+ assertTargetOptions(append.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT))
}
}
- test("SPARK-50286: Propagate options for DataFrameWriterV2 Append") {
+ testWithLoadOptionAsTableState("SPARK-50286: Propagate options for DataFrameWriterV2 Append") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ inMemoryCatalog.resetLoadTableCalls()
val captured = withQueryExecutionsCaptured(spark) {
Seq(1 -> "a", 2 -> "b").toDF("id", "data")
.writeTo(t1)
- .option("write.split-size", "10")
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
.append()
}
assert(captured.size === 1)
val qe = captured.head
var collected = qe.optimizedPlan.collect {
- case AppendData(_: DataSourceV2Relation, _, writeOptions, _, _, _, _) =>
- assert(writeOptions("write.split-size") == "10")
+ case AppendData(relation: DataSourceV2Relation, _, writeOptions, _, _, _, _) =>
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assert(writeOptions(loadOption) === loadOptionValue)
+ assert(writeOptions(writeOption) === writeOptionValue)
}
assert (collected.size == 1)
collected = qe.executedPlan.collect {
case AppendDataExec(_, _, write, _, _) =>
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#Append]
- assert(append.info.options.get("write.split-size") === "10")
+ assertTargetOptions(append.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT))
}
}
- test("SPARK-49098, SPARK-50286: Supports Dynamic Table Options for SQL Insert Overwrite") {
+ testWithLoadOptionAsTableState(
+ "SPARK-49098, SPARK-50286: Supports Dynamic Table Options for SQL Insert Overwrite") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
- val df = sql(s"INSERT OVERWRITE $t1 WITH (`write.split-size` = 10) " +
- s"VALUES (3, 'c'), (4, 'd')")
+ val df = sql(s"INSERT OVERWRITE $t1 WITH (`$loadOption` = '$loadOptionValue', " +
+ s"`$writeOption` = '$writeOptionValue') VALUES (3, 'c'), (4, 'd')")
var collected = df.queryExecution.optimizedPlan.collect {
case CommandResult(_,
OverwriteByExpression(relation: DataSourceV2Relation, _, _, _, _, _, _, _),
_, _) =>
- assert(relation.options.get("write.split-size") === "10")
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assertTargetOptions(relation.options)
}
assert (collected.size == 1)
@@ -360,58 +480,69 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
_, OverwriteByExpressionExec(_, _, write, _, _),
_) =>
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#TruncateAndAppend]
- assert(append.info.options.get("write.split-size") === "10")
+ assertTargetOptions(append.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
val insertResult = sql(s"SELECT * FROM $t1")
checkAnswer(insertResult, Seq(Row(3, "c"), Row(4, "d")))
}
}
- test("SPARK-50286: Propagate options for DataFrameWriterV2 OverwritePartitions") {
+ testWithLoadOptionAsTableState(
+ "SPARK-50286: Propagate options for DataFrameWriterV2 OverwritePartitions") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
val captured = withQueryExecutionsCaptured(spark) {
Seq(3 -> "c", 4 -> "d").toDF("id", "data")
.writeTo(t1)
- .option("write.split-size", "10")
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
.overwritePartitions()
}
assert(captured.size === 1)
val qe = captured.head
var collected = qe.optimizedPlan.collect {
- case OverwritePartitionsDynamic(_: DataSourceV2Relation, _, writeOptions, _, _, _) =>
- assert(writeOptions("write.split-size") === "10")
+ case OverwritePartitionsDynamic(
+ relation: DataSourceV2Relation, _, writeOptions, _, _, _) =>
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assert(writeOptions(loadOption) === loadOptionValue)
+ assert(writeOptions(writeOption) === writeOptionValue)
}
assert (collected.size == 1)
collected = qe.executedPlan.collect {
case OverwritePartitionsDynamicExec(_, _, write, _, _) =>
val dynOverwrite = write.toBatch.asInstanceOf[InMemoryBaseTable#DynamicOverwrite]
- assert(dynOverwrite.info.options.get("write.split-size") === "10")
+ assertTargetOptions(dynOverwrite.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
}
}
- test("SPARK-49098, SPARK-50286: Supports Dynamic Table Options for SQL Insert Replace") {
+ testWithLoadOptionAsTableState(
+ "SPARK-49098, SPARK-50286: Supports Dynamic Table Options for SQL Insert Replace") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
- val df = sql(s"INSERT INTO $t1 WITH (`write.split-size` = 10) " +
- s"REPLACE WHERE TRUE " +
+ val df = sql(s"INSERT INTO $t1 WITH (`$loadOption` = '$loadOptionValue', " +
+ s"`$writeOption` = '$writeOptionValue') REPLACE WHERE TRUE " +
s"VALUES (3, 'c'), (4, 'd')")
var collected = df.queryExecution.optimizedPlan.collect {
case CommandResult(_,
OverwriteByExpression(relation: DataSourceV2Relation, _, _, _, _, _, _, _),
_, _) =>
- assert(relation.options.get("write.split-size") == "10")
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assertTargetOptions(relation.options)
}
assert (collected.size == 1)
@@ -420,23 +551,26 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
_, OverwriteByExpressionExec(_, _, write, _, _),
_) =>
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#TruncateAndAppend]
- assert(append.info.options.get("write.split-size") === "10")
+ assertTargetOptions(append.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
val insertResult = sql(s"SELECT * FROM $t1")
checkAnswer(insertResult, Seq(Row(3, "c"), Row(4, "d")))
}
}
- test("SPARK-50286: Propagate options for DataFrameWriter Overwrite") {
+ testWithLoadOptionAsTableState("SPARK-50286: Propagate options for DataFrameWriter Overwrite") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ inMemoryCatalog.resetLoadTableCalls()
val captured = withQueryExecutionsCaptured(spark) {
Seq(1 -> "a", 2 -> "b").toDF("id", "data")
.write
- .option("write.split-size", "10")
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
.mode("overwrite")
.insertInto(t1)
}
@@ -444,47 +578,201 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
val qe = captured.head
var collected = qe.optimizedPlan.collect {
- case OverwriteByExpression(_: DataSourceV2Relation, _, _, writeOptions, _, _, _, _) =>
- assert(writeOptions("write.split-size") === "10")
+ case OverwriteByExpression(
+ relation: DataSourceV2Relation, _, _, writeOptions, _, _, _, _) =>
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assert(writeOptions(loadOption) === loadOptionValue)
+ assert(writeOptions(writeOption) === writeOptionValue)
}
assert (collected.size == 1)
collected = qe.executedPlan.collect {
case OverwriteByExpressionExec(_, _, write, _, _) =>
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#TruncateAndAppend]
- assert(append.info.options.get("write.split-size") === "10")
+ assertTargetOptions(append.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
+ }
+ }
+
+ testWithLoadOptionAsTableState(
+ "SPARK-58389: dynamic partition overwrite separates load and write options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) PARTITIONED BY (id)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
+
+ val captured = withSQLConf(
+ SQLConf.PARTITION_OVERWRITE_MODE.key ->
+ SQLConf.PartitionOverwriteMode.DYNAMIC.toString) {
+ withQueryExecutionsCaptured(spark) {
+ Seq(2 -> "updated", 3 -> "new").toDF("id", "data")
+ .write
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
+ .mode("overwrite")
+ .insertInto(t1)
+ }
+ }
+ assert(captured.size === 1)
+
+ val qe = captured.head
+ val logicalWrites = qe.optimizedPlan.collect {
+ case OverwritePartitionsDynamic(
+ relation: DataSourceV2Relation, _, writeOptions, _, _, _) =>
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assert(writeOptions(loadOption) === loadOptionValue)
+ assert(writeOptions(writeOption) === writeOptionValue)
+ }
+ assert(logicalWrites.size === 1)
+
+ val physicalWrites = qe.executedPlan.collect {
+ case OverwritePartitionsDynamicExec(_, _, write, _, _) =>
+ val dynamicOverwrite = write.toBatch.asInstanceOf[InMemoryBaseTable#DynamicOverwrite]
+ assertTargetOptions(dynamicOverwrite.info.options)
+ }
+ assert(physicalWrites.size === 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
+ checkAnswer(sql(s"SELECT * FROM $t1"),
+ Seq(Row(1, "a"), Row(2, "updated"), Row(3, "new")))
}
}
- test("SPARK-50286: Propagate options for DataFrameWriterV2 Overwrite") {
+ testWithLoadOptionAsTableState(
+ "SPARK-50286: Propagate options for DataFrameWriterV2 Overwrite") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
val captured = withQueryExecutionsCaptured(spark) {
Seq(3 -> "c", 4 -> "d").toDF("id", "data")
.writeTo(t1)
- .option("write.split-size", "10")
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
.overwrite(lit(true))
}
assert(captured.size === 1)
val qe = captured.head
var collected = qe.optimizedPlan.collect {
- case OverwriteByExpression(_: DataSourceV2Relation, _, _, writeOptions, _, _, _, _) =>
- assert(writeOptions("write.split-size") === "10")
+ case OverwriteByExpression(
+ relation: DataSourceV2Relation, _, _, writeOptions, _, _, _, _) =>
+ assert(relation.table.isInstanceOf[InMemoryBaseTable])
+ assert(writeOptions(loadOption) === loadOptionValue)
+ assert(writeOptions(writeOption) === writeOptionValue)
}
assert (collected.size == 1)
collected = qe.executedPlan.collect {
case OverwriteByExpressionExec(_, _, write, _, _) =>
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#TruncateAndAppend]
- assert(append.info.options.get("write.split-size") === "10")
+ assertTargetOptions(append.info.options)
}
assert (collected.size == 1)
+ assertWriteLoad(Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
+ }
+ }
+
+ testWithLoadOptionAsTableState(
+ "SPARK-58389: DataFrameWriter saveAsTable separates load and write options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ Seq(
+ ("append", Set(TableWritePrivilege.INSERT)),
+ ("overwrite", Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
+ ).foreach { case (mode, expectedPrivileges) =>
+ inMemoryCatalog.resetLoadTableCalls()
+
+ val captured = withQueryExecutionsCaptured(spark) {
+ Seq(1 -> "a").toDF("id", "data")
+ .write
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
+ .mode(mode)
+ .saveAsTable(t1)
+ }
+
+ assertWriteLoad(expectedPrivileges)
+ assertV2Write(captured)
+ }
+ }
+ }
+
+ testWithLoadOptionAsTableState(
+ "SPARK-58389: schema evolution reload separates table-state and write options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint)")
+ inMemoryCatalog.resetLoadTableCalls()
+
+ val captured = withQueryExecutionsCaptured(spark) {
+ Seq(1L -> "a").toDF("id", "data")
+ .writeTo(t1)
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
+ .withSchemaEvolution()
+ .append()
+ }
+
+ val matchingCalls = inMemoryCatalog.loadTableCalls.filter {
+ case (_, options) => options.get(loadOption) == loadOptionValue
+ }
+ assert(matchingCalls.size >= 2, "expected initial target load and post-evolution reload")
+ matchingCalls.foreach { case (context, options) =>
+ assert(context.writePrivileges() === java.util.Set.of(TableWritePrivilege.INSERT))
+ assert(options.get(writeOption) === null)
+ assert(options.size() === 1)
+ }
+ assertV2Write(captured)
+ }
+ }
+
+ Seq(
+ "non-staging" -> classOf[NullReturningInMemoryCatalog],
+ "staging" -> classOf[NullReturningStagingInMemoryCatalog]
+ ).foreach { case (catalogType, catalogClass) =>
+ test(s"SPARK-58389: $catalogType CTAS/RTAS fallback separates load and write options") {
+ val catalogName = s"${catalogType.replace('-', '_')}_null_catalog"
+ registerCatalog(catalogName, catalogClass)
+ spark.conf.set(s"spark.sql.catalog.$catalogName.tableStateOptionKeys", loadOption)
+ val fallbackCatalog = catalog(catalogName).asInstanceOf[InMemoryCatalog]
+ val t1 = s"$catalogName.table"
+ withTable(t1) {
+ fallbackCatalog.resetLoadTableCalls()
+ val createExecutions = withQueryExecutionsCaptured(spark) {
+ spark.range(1).writeTo(t1)
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
+ .create()
+ }
+ assertWriteLoad(fallbackCatalog, Set(TableWritePrivilege.INSERT))
+ assertV2Write(createExecutions)
+
+ fallbackCatalog.resetLoadTableCalls()
+ val replaceExecutions = withQueryExecutionsCaptured(spark) {
+ spark.range(1, 2).writeTo(t1)
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
+ .replace()
+ }
+ assertWriteLoad(fallbackCatalog, Set(TableWritePrivilege.INSERT))
+ assertV2Write(replaceExecutions)
+
+ fallbackCatalog.resetLoadTableCalls()
+ val createOrReplaceExecutions = withQueryExecutionsCaptured(spark) {
+ spark.range(2, 3).writeTo(t1)
+ .option(loadOption, loadOptionValue)
+ .option(writeOption, writeOptionValue)
+ .createOrReplace()
+ }
+ assertWriteLoad(fallbackCatalog, Set(TableWritePrivilege.INSERT))
+ assertV2Write(createOrReplaceExecutions)
+ }
}
}
@@ -889,6 +1177,32 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
+ test("SPARK-58389: explicit time travel specs on internal write targets use qualified names") {
+ withStateAwareTable { (_, tableName) =>
+ val resolver = new RelationResolution(
+ spark.sessionState.catalogManager,
+ RelationCache.empty)
+ val write = UnresolvedRelation(Seq("ns", "table"))
+ .requireWritePrivileges(Set(TableWritePrivilege.INSERT))
+ val previousCatalog = spark.catalog.currentCatalog()
+
+ try {
+ spark.catalog.setCurrentCatalog("statecat")
+ val e = AnalysisContext.withNewAnalysisContext {
+ intercept[AnalysisException] {
+ resolver.resolveRelation(write, Some(AsOfVersion("v1")))
+ }
+ }
+ checkError(
+ exception = e,
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ parameters = Map("relationId" -> "`statecat`.`ns`.`table`"))
+ } finally {
+ spark.catalog.setCurrentCatalog(previousCatalog)
+ }
+ }
+ }
+
test("persistent write targets establish table pins for subsequent reads") {
withStateAwareTable { (stateCatalog, tableName) =>
stateCatalog.resetLoadTableCalls()
@@ -1123,20 +1437,93 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
- test("SPARK-58389: time travel option on a write target is rejected with a user-facing error") {
+ test("SPARK-58389: time travel is rejected for resolved and newly created V2 write targets") {
val t1 = s"${catalogAndNamespace}table"
- withTable(t1) {
+ val newTable = s"${catalogAndNamespace}new_table"
+ withTable(t1, newTable) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
- // A time-travel option on a write target is reachable via the option form (the `AS OF`
- // syntax is blocked earlier by the parser). It must surface as a user-facing analysis error,
- // not the internal TableContext mutual-exclusion guard (which would report INTERNAL_ERROR).
- checkError(
- exception = intercept[AnalysisException] {
- sql(s"INSERT INTO $t1 WITH ('versionAsOf' = 'v1') VALUES (1, 'a')")
- },
- condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
- parameters = Map("relationId" -> "`testcat`.`ns1`.`ns2`.`table`"))
+ val input = Seq(1L -> "a").toDF("id", "data")
+ val existingRelationId = "`testcat`.`ns1`.`ns2`.`table`"
+ val newRelationId = "`testcat`.`ns1`.`ns2`.`new_table`"
+ Seq(
+ existingRelationId -> (() => sql(s"INSERT INTO $t1 WITH " +
+ "('versionAsOf' = 'v1', 'timestampAsOf' = '2021-01-01') VALUES (1, 'a')")),
+ existingRelationId -> (() => input.writeTo(t1)
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .append()),
+ existingRelationId -> (() => input.write
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .insertInto(t1)),
+ existingRelationId -> (() => input.write
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .mode("append")
+ .saveAsTable(t1)),
+ newRelationId -> (() => input.writeTo(newTable)
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .create()),
+ existingRelationId -> (() => input.writeTo(t1)
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .replace()),
+ existingRelationId -> (() => input.writeTo(t1)
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .createOrReplace())
+ ).foreach { case (relationId, writeToTimeTravelTarget) =>
+ checkError(
+ exception = intercept[AnalysisException](writeToTimeTravelTarget()),
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ parameters = Map("relationId" -> relationId))
+ }
+
+ withTempView("temp_view") {
+ input.createOrReplaceTempView("temp_view")
+ checkError(
+ exception = intercept[AnalysisException] {
+ input.writeTo("temp_view")
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .append()
+ },
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ parameters = Map("relationId" -> "`temp_view`"))
+ }
+ }
+ }
+
+ test("SPARK-58389: a missing insertInto target reports table not found before time travel") {
+ val missingTable = s"${catalogAndNamespace}missing"
+ val input = Seq(1L -> "a").toDF("id", "data")
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ input.write.option("versionAsOf", "v1").insertInto(missingTable)
+ },
+ condition = "TABLE_OR_VIEW_NOT_FOUND",
+ parameters = Map("relationName" -> "`ns1`.`ns2`.`missing`"))
+ }
+
+ test("SPARK-58389: SQL CTAS and RTAS options remain table properties") {
+ val t1 = s"${catalogAndNamespace}table"
+ val ident = Identifier.of(Array("ns1", "ns2"), "table")
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 USING foo OPTIONS ('versionAsOf' = 'v1') " +
+ "AS SELECT 1L AS id, 'a' AS data")
+ val createdProperties = inMemoryCatalog.loadTable(ident).properties()
+ assert(createdProperties.get("versionAsOf") === "v1")
+ assert(createdProperties.get("option.versionAsOf") === "v1")
+
+ sql(s"REPLACE TABLE $t1 USING foo OPTIONS ('timestampAsOf' = '2021-01-01') " +
+ "AS SELECT 2L AS id, 'b' AS data")
+ val replacedProperties = inMemoryCatalog.loadTable(ident).properties()
+ assert(replacedProperties.get("timestampAsOf") === "2021-01-01")
+ assert(replacedProperties.get("option.timestampAsOf") === "2021-01-01")
+ checkAnswer(spark.table(t1), Row(2L, "b"))
}
}
@@ -1278,6 +1665,7 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
// options. A later option-free reference to the same table must not inherit them.
withTempView("v") {
spark.read.option("split-size", "5").table(t1).createOrReplaceTempView("v")
+ inMemoryCatalog.resetLoadTableCalls()
val df = sql(s"SELECT v.id FROM v JOIN $t1 b ON v.id = b.id")
val splitSizes = df.queryExecution.analyzed.collect {
@@ -1288,6 +1676,11 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
s"option leaked to the option-free reference, got: $splitSizes")
assert(splitSizes.contains(None),
s"expected an option-free reference, got: $splitSizes")
+ assert(inMemoryCatalog.loadTableCalls.nonEmpty,
+ "V2TableReference temp-view reload did not load the table")
+ assert(inMemoryCatalog.loadTableCalls.forall(_._2.isEmpty),
+ s"expected temp-view reloads to filter scan options, got: " +
+ inMemoryCatalog.loadTableCalls)
}
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
index b75c0fc1c947..f8cc82eb8a89 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
@@ -22,7 +22,7 @@ import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.expressions.CheckInvariant
import org.apache.spark.sql.catalyst.plans.logical.Filter
-import org.apache.spark.sql.connector.catalog.{Aborted, Committed, InMemoryRowLevelOperationTable, InMemoryTable, InMemoryTruncatableOnlyTable, InMemoryTruncatableOnlyTableCatalog, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{Aborted, Committed, InMemoryRowLevelOperationTable, InMemoryTable, InMemoryTruncatableOnlyTable, InMemoryTruncatableOnlyTableCatalog, TableWritePrivilege}
import org.apache.spark.sql.connector.write.DeleteSummary
import org.apache.spark.sql.execution.datasources.v2.{DeleteFromTableExec, ReplaceDataExec, TruncateTableExec, WriteDeltaExec}
import org.apache.spark.sql.internal.SQLConf
@@ -1047,8 +1047,10 @@ abstract class DeleteFromTableSuiteBase extends RowLevelOperationSuiteBase {
// replace the row-level plan with a filter-only deleteWhere call via
// OptimizeMetadataOnlyDeleteFromTable (canDeleteWhere returns false for LessThan).
checkRowLevelOperationOptions(
- sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) WHERE salary < 200"),
- "write.split-size" -> "10")
+ sql(s"DELETE FROM $tableNameAsString WITH " +
+ s"(`load-option` = 'load-value', `write-option` = 'write-value') WHERE salary < 200"),
+ "load-option" -> "load-value",
+ "write-option" -> "write-value")
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
@@ -1101,16 +1103,20 @@ abstract class DeleteFromTableSuiteBase extends RowLevelOperationSuiteBase {
// dep = 'hr' is an EqualTo on the partition column. OptimizeMetadataOnlyDeleteFromTable
// converts the row-level plan to a deleteWhere call. Verify options flow into the exec.
val Seq(qe) = withQueryExecutionsCaptured(spark) {
- sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) WHERE dep = 'hr'")
+ sql(s"DELETE FROM $tableNameAsString WITH " +
+ s"(`load-option` = 'load-value', `write-option` = 'write-value') WHERE dep = 'hr'")
}
val exec = qe.executedPlan.collectFirst {
case e: DeleteFromTableExec => e
}.getOrElse(fail("expected DeleteFromTableExec for the metadata-only deleteWhere path"))
- assert(exec.options.get("write.split-size") === "10",
- "options must reach DeleteFromTableExec on the deleteWhere path")
- assert(exec.table.asInstanceOf[InMemoryRowLevelOperationTable].lastDeleteOptions
- .get("write.split-size") === "10",
- "options must be forwarded to SupportsDeleteV2.deleteWhere(predicates, options)")
+ assert(exec.options.get("load-option") === "load-value")
+ assert(exec.options.get("write-option") === "write-value")
+ val v2Table = exec.table.asInstanceOf[InMemoryRowLevelOperationTable]
+ assert(v2Table.lastDeleteOptions.get("load-option") === "load-value")
+ assert(v2Table.lastDeleteOptions.get("write-option") === "write-value")
+ assertLastTransactionWriteLoadOptions(
+ "load-option" -> "load-value",
+ "write-option" -> "write-value")
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
@@ -1146,27 +1152,40 @@ abstract class DeleteFromTableSuiteBase extends RowLevelOperationSuiteBase {
// is planned as TruncateTableExec (not DeleteFromTableExec).
withSQLConf(
"spark.sql.catalog.trunccat" ->
- classOf[InMemoryTruncatableOnlyTableCatalog].getName) {
+ classOf[InMemoryTruncatableOnlyTableCatalog].getName,
+ "spark.sql.catalog.trunccat.tableStateOptionKeys" -> "load-option") {
withTable("trunccat.ns.tbl") {
sql("CREATE TABLE trunccat.ns.tbl (pk INT NOT NULL, dep STRING) USING foo")
sql("INSERT INTO trunccat.ns.tbl VALUES (1, 'hr'), (2, 'software')")
+ val truncCatalog = spark.sessionState.catalogManager
+ .catalog("trunccat").asInstanceOf[InMemoryTruncatableOnlyTableCatalog]
+ truncCatalog.resetLoadTableCalls()
val Seq(qe) = withQueryExecutionsCaptured(spark) {
- sql("DELETE FROM trunccat.ns.tbl WITH (`write.split-size` = 10)")
+ sql("DELETE FROM trunccat.ns.tbl WITH " +
+ "(`load-option` = 'load-value', `write-option` = 'write-value')")
}
val exec = qe.executedPlan.collectFirst {
case e: TruncateTableExec => e
}.getOrElse(fail("expected TruncateTableExec for the truncate path"))
- assert(exec.options.get("write.split-size") === "10",
- "options must reach TruncateTableExec")
+ assert(exec.options.get("load-option") === "load-value")
+ assert(exec.options.get("write-option") === "write-value")
+
+ val targetLoads = truncCatalog.loadTableCalls.filter {
+ case (context, options) =>
+ context.writePrivileges() === java.util.Set.of(TableWritePrivilege.DELETE) &&
+ options.get("load-option") == "load-value" &&
+ options.get("write-option") == null &&
+ options.size() == 1
+ }
+ assert(targetLoads.nonEmpty, "target loadTable did not receive truncate state options")
- val truncTable = spark.sessionState.catalogManager
- .catalog("trunccat").asInstanceOf[TableCatalog]
+ val truncTable = truncCatalog
.loadTable(org.apache.spark.sql.connector.catalog.Identifier.of(
Array("ns"), "tbl"))
.asInstanceOf[InMemoryTruncatableOnlyTable]
- assert(truncTable.lastTruncateOptions.get("write.split-size") === "10",
- "options must be forwarded to TruncatableTable.truncateTable(options)")
+ assert(truncTable.lastTruncateOptions.get("load-option") === "load-value")
+ assert(truncTable.lastTruncateOptions.get("write-option") === "write-value")
checkAnswer(spark.table("trunccat.ns.tbl"), Nil)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
index de6c45006911..ea15f13c225c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
@@ -2866,13 +2866,15 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
// DataSourceV2Relation, the RowLevelOperationInfo, and the write builder's LogicalWriteInfo
checkRowLevelOperationOptions(
sql(
- s"""MERGE INTO $tableNameAsString t WITH (`write.split-size` = 10)
+ s"""MERGE INTO $tableNameAsString t WITH
+ | (`load-option` = 'load-value', `write-option` = 'write-value')
|USING $sourceNameAsString s
|ON t.pk = s.pk
|WHEN MATCHED THEN UPDATE SET t.salary = s.salary
|WHEN NOT MATCHED THEN INSERT *
|""".stripMargin),
- "write.split-size" -> "10")
+ "load-option" -> "load-value",
+ "write-option" -> "write-value")
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
@@ -3016,7 +3018,8 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
// write from the target's options at a different site than the row-level rewrite path.
val executedPlan = executeAndKeepPlan {
sql(
- s"""MERGE INTO $tableNameAsString t WITH (`write.split-size` = 10)
+ s"""MERGE INTO $tableNameAsString t WITH
+ | (`load-option` = 'load-value', `write-option` = 'write-value')
|USING $sourceNameAsString s
|ON t.pk = s.pk
|WHEN NOT MATCHED THEN INSERT *
@@ -3026,7 +3029,11 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
case e: InsertOnlyMergeExec => e.write
}.getOrElse(fail("expected an InsertOnlyMergeExec in the executed plan"))
val append = write.toBatch.asInstanceOf[InMemoryBaseTable#Append]
- assert(append.info.options.get("write.split-size") === "10")
+ assert(append.info.options.get("load-option") === "load-value")
+ assert(append.info.options.get("write-option") === "write-value")
+ assertLastTransactionWriteLoadOptions(
+ "load-option" -> "load-value",
+ "write-option" -> "write-value")
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
index 5d52325da818..21e27e9b6c1c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
@@ -19,6 +19,8 @@ package org.apache.spark.sql.connector
import java.util.Collections
+import scala.jdk.CollectionConverters._
+
import org.scalatest.BeforeAndAfter
import org.apache.spark.sql.{DataFrame, Encoders, Row}
@@ -48,12 +50,15 @@ abstract class RowLevelOperationSuiteBase
before {
spark.conf.set("spark.sql.catalog.cat", classOf[InMemoryRowLevelOperationTableCatalog].getName)
+ spark.conf.set(
+ "spark.sql.catalog.cat.tableStateOptionKeys", "load-option,targetLoadOption")
}
after {
catalog.nextTxnRejectRegisteredScansAttempt = false
spark.sessionState.catalogManager.reset()
spark.sessionState.conf.unsetConf("spark.sql.catalog.cat")
+ spark.sessionState.conf.unsetConf("spark.sql.catalog.cat.tableStateOptionKeys")
}
protected final val PK_FIELD = StructField("pk", IntegerType, nullable = false)
@@ -179,9 +184,9 @@ abstract class RowLevelOperationSuiteBase
}.getOrElse(fail("couldn't find row-level operation in optimized plan"))
}
- // asserts the given SQL options reached every layer that should carry them: the rewritten
- // DataSourceV2Relation, the RowLevelOperationInfo passed to the operation builder, and the
- // write builder's LogicalWriteInfo
+ // Asserts the given SQL options reached every V2 layer that should carry them: the target
+ // catalog load, the rewritten DataSourceV2Relation, the RowLevelOperationInfo passed to the
+ // operation builder, and the write builder's LogicalWriteInfo.
protected def checkRowLevelOperationOptions(
func: => Unit,
expectedOptions: (String, String)*): Unit = {
@@ -191,8 +196,12 @@ abstract class RowLevelOperationSuiteBase
case wd: WriteDelta => wd.table
}.getOrElse(fail("couldn't find row-level operation in optimized plan"))
.asInstanceOf[DataSourceV2Relation]
+ assert(writeRelation.catalog.nonEmpty, "expected a catalog-backed V2 write relation")
+ assert(writeRelation.identifier.nonEmpty, "expected a catalog-backed V2 write identifier")
val operation = writeRelation.table.asInstanceOf[RowLevelOperationTable].operation
.asInstanceOf[RowLevelOperationWithOptions]
+ assertLastTransactionWriteLoadOptions(expectedOptions: _*)
+
expectedOptions.foreach { case (key, value) =>
assert(writeRelation.options.get(key) === value, s"relation option '$key'")
assert(operation.options.get(key) === value, s"row-level operation option '$key'")
@@ -200,6 +209,27 @@ abstract class RowLevelOperationSuiteBase
}
}
+ protected def assertLastTransactionWriteLoadOptions(
+ expectedOptions: (String, String)*): Unit = {
+ val stateKeys = catalog.tableStateOptionKeys().asScala
+ val expectedStateOptions = expectedOptions.filter { case (key, _) =>
+ stateKeys.exists(_.equalsIgnoreCase(key))
+ }
+ val targetLoads = catalog.lastTransaction.catalog.loadTableCalls.filter {
+ case (context, _) => !context.writePrivileges().isEmpty
+ }
+ assert(targetLoads.nonEmpty, "target loadTable did not receive write privileges")
+ targetLoads.foreach { case (_, options) =>
+ assert(options.size() === expectedStateOptions.size)
+ expectedStateOptions.foreach { case (key, value) =>
+ assert(options.get(key) === value, s"table-state option '$key'")
+ }
+ expectedOptions.diff(expectedStateOptions).foreach { case (key, _) =>
+ assert(options.get(key) === null, s"non-state load option '$key'")
+ }
+ }
+ }
+
protected def assertNoScanPlanning(plan: LogicalPlan): Unit = {
val relations = plan.collect { case r: DataSourceV2Relation => r }
assert(relations.nonEmpty, "plan must contain relations")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala
index 891bc0490a89..1c293bfa268d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala
@@ -27,14 +27,15 @@ import org.scalatest.BeforeAndAfter
import org.apache.spark.SparkException
import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, SaveMode}
+import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.analysis.{AsOfTimestamp, AsOfVersion, NoSuchTableException, TableAlreadyExistsException, TimeTravelSpec}
import org.apache.spark.sql.catalyst.plans.logical.{AppendData, LogicalPlan, OverwriteByExpression}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, InMemoryTableCatalog, SupportsCatalogOptions, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, SupportsCatalogOptions, TableCatalog, TableWritePrivilege}
import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTransform}
import org.apache.spark.sql.execution.QueryExecution
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{AppendDataExec, DataSourceV2Relation, OverwriteByExpressionExec}
import org.apache.spark.sql.internal.SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.LongType
@@ -65,6 +66,7 @@ class SupportsCatalogOptionsSuite extends SharedSparkSession with BeforeAndAfter
V2_SESSION_CATALOG_IMPLEMENTATION.key, classOf[InMemoryTableSessionCatalog].getName)
spark.conf.set(
s"spark.sql.catalog.$catalogName", classOf[InMemoryTableCatalog].getName)
+ spark.conf.set(s"spark.sql.catalog.$catalogName.tableStateOptionKeys", "load-option")
}
override def afterEach(): Unit = {
@@ -75,6 +77,7 @@ class SupportsCatalogOptionsSuite extends SharedSparkSession with BeforeAndAfter
catalog(catalogName).dropTable(_))
spark.conf.unset(V2_SESSION_CATALOG_IMPLEMENTATION.key)
spark.conf.unset(s"spark.sql.catalog.$catalogName")
+ spark.conf.unset(s"spark.sql.catalog.$catalogName.tableStateOptionKeys")
}
private def testCreateAndRead(
@@ -402,6 +405,77 @@ class SupportsCatalogOptionsSuite extends SharedSparkSession with BeforeAndAfter
assert(opts.get.isEmpty)
}
+ test("SPARK-58389: SupportsCatalogOptions separates table-state and write options") {
+ sql(s"create table $catalogName.t1 (id bigint) using $format")
+ val cat = catalog(catalogName).asInstanceOf[InMemoryTableCatalog]
+ val loadOption = "load-Option"
+ val loadValue = "load-value"
+ val writeOption = "write-option"
+
+ Seq(
+ (SaveMode.Append, "append", java.util.Set.of(TableWritePrivilege.INSERT)),
+ (SaveMode.Overwrite, "overwrite",
+ java.util.Set.of(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
+ ).foreach { case (mode, optionValue, expectedPrivileges) =>
+ cat.resetLoadTableCalls()
+ val Seq(qe) = withQueryExecutionsCaptured(spark) {
+ spark.range(1).write
+ .format(format)
+ .option("name", "t1")
+ .option("catalog", catalogName)
+ .option(loadOption, loadValue)
+ .option(writeOption, optionValue)
+ .mode(mode)
+ .save()
+ }
+
+ val matchingCalls = cat.loadTableCalls.filter {
+ case (_, options) => options.get(loadOption) == loadValue
+ }
+ assert(matchingCalls.nonEmpty, "loadTable(context, options) was not invoked for the write")
+ matchingCalls.foreach { case (context, options) =>
+ assert(context.writePrivileges() === expectedPrivileges)
+ assert(options.size() === 1)
+ assert(options.asCaseSensitiveMap().containsKey(loadOption))
+ assert(options.get(writeOption) === null)
+ }
+
+ val actualWriteOptions = mode match {
+ case SaveMode.Append =>
+ qe.executedPlan.collectFirst {
+ case AppendDataExec(_, _, write, _, _) =>
+ write.toBatch.asInstanceOf[InMemoryBaseTable#Append].info.options
+ }
+ case SaveMode.Overwrite =>
+ qe.executedPlan.collectFirst {
+ case OverwriteByExpressionExec(_, _, write, _, _) =>
+ write.toBatch.asInstanceOf[InMemoryBaseTable#TruncateAndAppend].info.options
+ }
+ case other => fail(s"unexpected save mode: $other")
+ }
+ assert(actualWriteOptions.isDefined, "expected a V2 in-memory batch write")
+ assert(actualWriteOptions.get.get(loadOption) === loadValue)
+ assert(actualWriteOptions.get.get(writeOption) === optionValue)
+ assert(actualWriteOptions.get.get("name") === "t1")
+ assert(actualWriteOptions.get.get("catalog") === catalogName)
+ }
+ }
+
+ test("SPARK-58389: SupportsCatalogOptions rejects time travel options for table creation") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.range(1).write
+ .format(format)
+ .option("name", "t1")
+ .option("catalog", catalogName)
+ .option("versionAsOf", "v1")
+ .option("timestampAsOf", "2021-01-01")
+ .save()
+ },
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ parameters = Map("relationId" -> "`testcat`.`t1`"))
+ }
+
private def load(
name: String,
catalogOpt: Option[String],
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
index b54445466d7e..c122713c1873 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
@@ -1243,8 +1243,11 @@ abstract class UpdateTableSuiteBase extends RowLevelOperationSuiteBase {
|""".stripMargin)
checkRowLevelOperationOptions(
- sql(s"UPDATE $tableNameAsString WITH (`write.split-size` = 10) SET salary = -1 WHERE pk = 1"),
- "write.split-size" -> "10")
+ sql(s"UPDATE $tableNameAsString WITH " +
+ s"(`load-option` = 'load-value', `write-option` = 'write-value') " +
+ s"SET salary = -1 WHERE pk = 1"),
+ "load-option" -> "load-value",
+ "write-option" -> "write-value")
checkAnswer(
sql(s"SELECT * FROM $tableNameAsString"),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org