peter-toth commented on code in PR #57865:
URL: https://github.com/apache/spark/pull/57865#discussion_r3780813492
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -548,7 +548,15 @@ class RelationResolution(
*/
private def loadRelation(ref: V2TableReference): LogicalPlan = {
val resolvedCatalog =
catalogManager.catalog(ref.catalog.name).asTableCatalog
- val table = resolvedCatalog.loadTable(ref.identifier)
+ val table = ref.context match {
+ case V2TableReference.WriteTargetContext =>
+ // WriteTargetContext 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.
+ resolvedCatalog.loadTable(ref.identifier)
+ case _ =>
+ CatalogV2Util.getTable(resolvedCatalog, ref.identifier, options =
ref.options)
Review Comment:
**Finding 1.** This is the same site as my round-1 comment, but the rebase
onto #57799 turned it around: the branch can't be reached, and the description
now claims a change the PR doesn't make.
`resolveReference:499` picks `loadRelation` only when
`!ref.context.cacheable`, and `Context` is sealed with three implementations
(`V2TableReference.scala:88-112`) — `cacheable = false` only for
`WriteTargetContext`. `TemporaryViewContext` and `TransactionContext` go to
`getOrLoadRelation:508`, which on master already calls
`CatalogV2Util.getTable(catalog, ref.identifier, options = ref.options)` at
`:520`. So `case _` is dead.
Measured: with the branch body replaced by a throw,
`DataSourceV2OptionSuite`, `AppendDataTransactionSuite`,
`StreamingTransactionSuite` and `DataSourceV2SQLSuiteV1Filter` are all green
(453 tests, 1 ignored).
Two consequences. The `match` reads as if `loadRelation` served several
contexts when only one reaches it, which is exactly the kind of thing the next
reader will keep alive. And these two description lines are now false and
should go:
* "DataFrame temporary-view and transaction-source re-resolution now invoke
the context/options-aware `loadTable` overload, forwarding only
catalog-declared table-state options"
* the matching user-facing sentence, "DataFrame temporary-view and
transaction-source re-resolution now invoke the context/options-aware
`loadTable` overload."
Back to the base shape, with the comment kept and the reachability spelled
out:
```scala
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)
}
```
##########
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.getTableForWrite(
+ catalog, ident, getWritePrivileges, dsOptions)
Review Comment:
**Finding 8.** Thanks for putting this in the description — that closes my
round-1 finding 2. What's still missing is the release-notes surface:
`docs/sql-migration-guide.md` has no entry, and this PR changes two behaviours
that have already shipped.
1. This line asks the catalog to authorize `INSERT` / `INSERT, DELETE` on
the `SupportsCatalogOptions` `save()` path, which passed no privileges before,
so a connector that authorizes can reject a `save()` that used to work.
2. `getTableForWrite` now rejects `versionAsOf` / `timestampAsOf` (or their
configured equivalents) on `insertInto`, `saveAsTable`, `save()` and, via the
`UnresolvedRelation`, `DataFrameWriterV2.append` / `overwrite` /
`overwritePartitions`. Measured on this head: `df.write.option("versionAsOf",
"1").insertInto(t)` throws `UNSUPPORTED_FEATURE.TIME_TRAVEL`. Master has no
check on that path at all — `insertIntoCommand` and `saveAsTableCommand` never
build an `UnresolvedRelation`, so `RelationResolution`'s check never ran and
the option was passed to the connector as an ordinary write option.
The precedent is in the same section: "Upgrading from Spark SQL 4.2 to 4.3"
already carries #57508's bullet ("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"), which is the read-side twin of
this work.
Rough bullets:
```markdown
- 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 authorizes writes
in `TableCatalog.loadTable` may now reject a `save()` that previously
succeeded.
- Since Spark 4.3, the Spark-recognized time-travel options (`versionAsOf`
and `timestampAsOf`, or
the keys configured by `spark.sql.timeTravelVersionKey` /
`spark.sql.timeTravelTimestampKey`) are
rejected with `UNSUPPORTED_FEATURE.TIME_TRAVEL` when set on a DataSource
V2 write target.
Previously they were passed to the connector as ordinary write options.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala:
##########
@@ -486,7 +490,10 @@ 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.getTableForWrite(catalog, ident,
getWritePrivileges, tableOptions))
Review Comment:
**Finding 9.** `rejectTimeTravelOptionsForWrite` runs before the load, so it
throws here before the `NoSuchTableException` catch can tell us whether there
is a write target at all. Two measurements on this head:
```
df.write.option("versionAsOf",
"1").mode("append").saveAsTable("testcat.ns1.new_tbl")
-> [UNSUPPORTED_FEATURE.TIME_TRAVEL] ... testcat.ns1.new_tbl (table
does not exist)
df.writeTo("testcat.ns1.new_tbl2").option("versionAsOf", "1").create()
-> succeeds, option passed straight through as a write option
```
So "rejected consistently on existing write targets across the DataFrame
writer APIs" doesn't hold either way round: it fires where the command is a
`CreateTableAsSelect` and no target state is read, and it never fires on
`DataFrameWriterV2`'s `create` / `replace` / `createOrReplace`.
`SaveMode.Overwrite` here is the same shape — it goes to `ReplaceTableAsSelect`
and discards the loaded table apart from the `V1Table` test at `:501`.
Either narrow it to loads of an existing target, which is what the
description promises:
```scala
val tableOpt = try {
Option(catalog.loadTable(ident, new TableContext(null,
getWritePrivileges.asJava),
CatalogV2Util.extractTableStateOptions(catalog, tableOptions)))
} catch {
case _: NoSuchTableException => None
}
tableOpt.foreach(_ =>
CatalogV2Util.rejectTimeTravelOptionsForWrite(catalog, ident, tableOptions))
```
or apply it to the `CreateTableAsSelect` / `ReplaceTableAsSelect` write
options too and reword the description to "on all catalog-backed V2 write
paths". I'd take the second — the option is meaningless on a create as well —
as long as the wording matches.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala:
##########
@@ -519,6 +519,36 @@ 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 getTableForWrite(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ writePrivileges: Set[TableWritePrivilege],
+ options: CaseInsensitiveStringMap): Table = {
+ rejectTimeTravelOptionsForWrite(catalog, ident, options)
+ val context = new TableContext(null, writePrivileges.asJava)
+ val stateOptions = extractTableStateOptions(catalog, options)
+ catalog.asTableCatalog.loadTable(ident, context, stateOptions)
+ }
+
+ private def rejectTimeTravelOptionsForWrite(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ options: CaseInsensitiveStringMap): Unit = {
+ val conf = SQLConf.get
+ val containsTimeTravelOption = Seq(
+ conf.getConf(SQLConf.TIME_TRAVEL_TIMESTAMP_KEY),
+
conf.getConf(SQLConf.TIME_TRAVEL_VERSION_KEY)).exists(options.containsKey)
+ if (containsTimeTravelOption) {
+ throw QueryCompilationErrors.timeTravelUnsupportedError(
Review Comment:
**Finding 10.** Same user mistake, different error depending on which API
you used.
`.quoted` backticks only when a part needs it, while the pre-existing check
at `RelationResolution.scala:242` uses `toSQLId`, which always backticks. The
new test encodes both forms side by side — `` `testcat`.`ns1`.`ns2`.`table` ``
for the SQL and `writeTo` cases, `testcat.ns1.ns2.table` for `insertInto` and
`saveAsTable`. Measured: `Time travel on the relation:
testcat.ns1.probe_new_tbl`.
The error *class* diverges too, because on the `UnresolvedRelation` paths
`TimeTravelSpec.fromOptions:98` throws before `tryResolvePersistent` gets to
its check:
```
df.write.option("versionAsOf","1").option("timestampAsOf","2021-01-01").insertInto(t)
-> [UNSUPPORTED_FEATURE.TIME_TRAVEL]
same options via writeTo(t).append()
-> [INVALID_TIME_TRAVEL_SPEC] Cannot specify both version and timestamp ...
```
`toSQLId` is already in scope in this file (`:407`), so the format half is
one token:
```scala
throw QueryCompilationErrors.timeTravelUnsupportedError(
toSQLId(ident.toQualifiedNameParts(catalog)))
```
For the class half, move the presence check ahead of the parse in
`resolveRelation:163` when the write-privilege marker is set, so every path
answers the same way for the same input:
```scala
val timeTravelSpecFromOptions =
if
(u.options.containsKey(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)) {
// Time travel applies to reads only; reject it before parsing so a
malformed or duplicated
// spec on a write target reports TIME_TRAVEL rather than a
parse-level error.
rejectTimeTravelOptionsForWrite(...)
None
} else {
TimeTravelSpec.fromOptions(u.options, ...)
}
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala:
##########
@@ -463,28 +639,131 @@ class DataSourceV2OptionSuite extends
DatasourceV2SQLBase {
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))
+ }
+ }
+
+ test("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)
Review Comment:
**Finding 11.** Round-1 finding 5 is closed for the `UnresolvedRelation`
paths, thanks. The three writer paths that don't go through it are still
unpinned: `insertInto` (`DataFrameWriter.scala:361`), `saveAsTable` (`:493`)
and `save()` (`:152`) each build `new
CaseInsensitiveStringMap(extraOptions.toMap.asJava)`, so casing survives only
because `CaseInsensitiveMap.toMap` is overridden to return `originalMap` —
`iterator` is lowercased, so passing `extraOptions` itself would silently
lowercase every key on all three paths and no test here would fail.
`loadOption` / `writeOption` are lowercase, and `assertTargetOptions` reads
through the case-insensitive `get`, so nothing catches it. One mixed-case key
plus one assert in `assertWriteLoad` covers all three:
```scala
private val loadOption = "load-Option"
```
and in `assertWriteLoad`, next to the existing `options.get(loadOption)`
check:
```scala
assert(options.asCaseSensitiveMap().containsKey(loadOption))
```
##########
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
Review Comment:
**Finding 12.** The sentence you replaced ("format or options will be
ignored") covered V1 as well, and that half is still true: for a V1 target
`insertInto` builds `InsertIntoStatement` with no options at all
(`DataFrameWriter.scala:392`), so the options are dropped. The new text only
says what happens for V2, which leaves a V1 user with no statement either way.
```scala
* 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.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
