peter-toth commented on code in PR #57865:
URL: https://github.com/apache/spark/pull/57865#discussion_r3779147324
##########
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 2.** Base here was `catalog.loadTable(ident)` -- no write
privileges at all. This now asks the catalog to authorize `INSERT` (append) or
`INSERT, DELETE` (overwrite). I think that's the right fix: this was the last
DataFrame write path that skipped the `TableWritePrivilege` hook SPARK-58370
hardened. But it's a semantic change well beyond forwarding options -- a
catalog that authorizes in `loadTable(ident, writePrivileges)` can now reject a
```scala
df.write.format("...").option("catalog", "cat").option("name",
"t").mode("overwrite").save()
```
that succeeded before, and the failure surfaces as an authorization error
from the connector rather than anything Spark can explain.
Worth noting the earlier version of this same fix (my finding 5 on #57582,
which was pulled back out before merge to keep that PR read-only) passed
options only -- `CatalogV2Util.getTable(catalog, ident, options = dsOptions)`
-- so the privileges are new here and no one has reviewed them yet.
Please call it out in the "What changes were proposed" bullets and in the
user-facing-change section, so connector authors reading the release notes know
an unauthorized `save()` on this path can now fail.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -507,7 +507,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 `V2TableReference` re-resolution path that #57582
explicitly declared out of scope ("still loads via the single-arg `loadTable`
... forwarding options there is better done as a separate change"). Landing it
here is fine by me, but it isn't a write path: `WriteTargetContext` is excluded
just above, so the only contexts that reach this branch are
`TemporaryViewContext` (DataFrame temp views, `views.scala:761`) and
`TransactionContext` (`UnresolveRelationsInTransaction.scala:64`) -- both reads.
The observable change: a connector that overrides `loadTable(Identifier,
TableContext, CaseInsensitiveStringMap)` is now invoked for those reloads,
where before the analyzer called the single-arg `loadTable(Identifier)`
directly and the override never ran. That is the same class of change as the
"one `loadTable` call per distinct option bag" note #57582 put in its
user-facing section.
Under a title that says "for writes", and a description whose bullets list
writer entry points plus "internal target reloads (the post-schema-evolution
reload after `catalog.alterTable` and staging/non-staging CTAS/RTAS fallback
loads)", this is invisible. Please add it to the bullet list and to the
user-facing-change section -- or split it out if you'd rather keep this PR
strictly to writes.
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriterV2.scala:
##########
@@ -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)
Review Comment:
**Finding 3.** Putting the writer's options on the `UnresolvedRelation` also
feeds them to the *option form* of time travel.
`RelationResolution.resolveRelation:163` runs
`TimeTravelSpec.fromOptions(u.options, ...)` with the default keys
`versionAsOf` / `timestampAsOf` (`spark.sql.timeTravelVersionKey` /
`spark.sql.timeTravelTimestampKey`), and `tryResolvePersistent:240` then throws
because the write-privilege marker is present.
I ran it to be sure. On this head:
```
org.apache.spark.sql.AnalysisException: [UNSUPPORTED_FEATURE.TIME_TRAVEL]
The feature is not
supported: Time travel on the relation: `cat`.`ns1`.`test_table`. SQLSTATE:
0A000
```
for `spark.table(t).writeTo(t).option("versionAsOf", "0").append()`; with
only this file reverted to base, the same call completes normally. So it's a
hard break of code that used to work, and #57582 documented exactly this shape
for `INSERT INTO t WITH ('versionAsOf' = ...)`.
Two things make it more than a doc line. Reusing one option map for a read
and a write is a common pattern. And the behaviour is now inconsistent between
the two DataFrame writers: `insertInto` and `saveAsTable` build the
`DataSourceV2Relation` directly (`DataFrameWriter.scala:360`, `:492`) and never
parse the options, so `df.write.option("versionAsOf", "0").insertInto(t)` stays
silently ignored. If rejecting it is the intent,
`CatalogV2Util.getTableForWrite` is the one place that would cover all three.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala:
##########
@@ -19,14 +19,41 @@ 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,
TableWritePrivilege, Txn}
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
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 (_, options) => options.get(targetLoadOption) == targetLoadValue
+ }
+ 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(targetWriteOption) === targetWriteValue)
+ }
+
+ assert(table.lastWriteInfo != null, "the V2 table did not receive
LogicalWriteInfo")
+ assert(table.lastWriteInfo.options().get(targetLoadOption) ===
targetLoadValue)
Review Comment:
**Finding 5.** `CaseInsensitiveStringMap.get` is case-insensitive, so these
pass whether or not the key casing survived. This suite is the only place with
a mixed-case fixture key (`targetLoadOption`); `DataSourceV2OptionSuite`,
`SupportsCatalogOptionsSuite` and the row-level suites all use lowercase
`load-option` / `write-option`. So the only test that actually pins casing is
the new `CatalogV2UtilSuite` one, and it stops at `UnresolvedRelation` --
nothing covers the rest of the path (`clearWritePrivileges` -> relation options
-> `V2Writes.mergeOptions`'s `asCaseSensitiveMap` -> `LogicalWriteInfo`), which
is exactly where the user-facing guarantee lives.
Two lines here close it, and they fail on base for this suite's SQL cases
(`clearWritePrivileges` lowercased, so the connector saw `targetloadoption`):
```scala
assert(table.lastWriteInfo.options().asCaseSensitiveMap().containsKey(targetLoadOption))
```
plus the catalog side, inside the `targetLoads.foreach` above:
```scala
assert(options.asCaseSensitiveMap().containsKey(targetLoadOption))
```
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala:
##########
@@ -196,6 +198,14 @@ class TxnTableCatalog(delegate:
InMemoryRowLevelOperationTableCatalog) extends T
})
}
+ override def loadTable(
+ ident: Identifier,
+ context: TableContext,
+ options: CaseInsensitiveStringMap): Table = {
+ loadTableCalls += ((context, options))
+ loadTable(ident)
Review Comment:
**Finding 6.** This drops `context` on the floor.
`BasicInMemoryTableCatalog` records and then calls `super.loadTable(ident,
context, options)`, with a comment saying why -- "defers to the default
dispatch in TableCatalog (rather than reimplementing it here)". As written the
fixture models the anti-pattern the `loadTable` javadoc warns about ("An
override replaces that dispatch and must honor `context` itself"), and it is
what keeps finding 4 invisible: a time-travel load inside a transaction quietly
returns the latest pinned table instead of failing or honoring the version.
One caveat on the obvious fix, which I hit when trying it: delegating alone
turns a time-travel read inside a transaction into `TABLE_OR_VIEW_NOT_FOUND`,
because `TxnTableCatalog` doesn't implement `loadTable(ident, version)` /
`(ident, timestamp)` either, so the default dispatch reaches `TableCatalog`'s
throwing overload. Loud beats silent, and no existing test in the suite
regresses (17/17 pass), but the complete fix is to delegate *and* serve the
version overloads from the pinned delegate:
```scala
override def loadTable(
ident: Identifier,
context: TableContext,
options: CaseInsensitiveStringMap): Table = {
loadTableCalls += ((context, options))
super.loadTable(ident, context, options)
}
```
If you'd rather keep the fixture as-is, a comment saying it deliberately
ignores time travel would at least stop the next reader from trusting it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -507,7 +507,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.
Review Comment:
**Finding 4.** Not caused by this PR -- I checked base -- but it lives in
the method you're changing, so flagging it while we're here. Take it or file it
separately, your call.
`V2TableReference` doesn't retain the time-travel spec, and
`UnresolveRelationsInTransaction.scala:63` has no `r.timeTravelSpec.isEmpty`
guard (unlike `prepareTemporaryViewPlan`, `views.scala:759`). So a
time-travelled relation that is already resolved when a transactional write is
built -- which is any `Dataset` whose `logicalPlan` is the analyzed plan, e.g.
after a `select` -- gets unresolved, comes back through here with the spec
gone, and silently reloads the latest snapshot. `validateNoChanges` won't catch
it: same `tableId`, same columns.
Measured in `AppendDataTransactionSuite` (pin version `0`, append one more
row, then use the pinned read as the source of a transactional append into a
second table):
```
outside-txn-source-rows=1
rows-written=2 sink=[1,100,hr],[2,200,software]
```
Same result with `RelationResolution.scala` reverted to base, so the wrong
data is pre-existing. What this PR adds is a second call site with the same
mismatch: `getTable` is called with `timeTravel = None` while `ref.options`
still says `versionAsOf = 0`, and the javadoc you just updated tells the
connector to trust the context. (Base already had one such call from
`V2TableRefreshUtil`, on the same spec-stripped relation.)
The guard fixes it -- I applied it and got `rows-written=1`, with the other
17 tests in the suite still passing:
```scala
case r: DataSourceV2Relation
if isLoadedFromCatalog(r, catalog) && r.timeTravelSpec.isEmpty =>
V2TableReference.createForTransaction(r)
```
Trade-off worth naming: the guard also stops the transaction from tracking
that read, which is what the rule exists for. Carrying the spec on
`V2TableReference` and passing it into `getTable` keeps both.
--
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]
