peter-toth commented on code in PR #57865:
URL: https://github.com/apache/spark/pull/57865#discussion_r3811400888
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala:
##########
@@ -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(
Review Comment:
**Finding 14.** Worth weighing rather than a request, and a follow-up given
the PR is approved.
The rejection is now enforced at five places: here (covering `save()`, the
schema-evolution reload and the four CTAS/RTAS fallbacks), `ResolveCatalogs`'
two new cases, `RelationResolution:250`, `DataFrameWriter:367` and
`DataFrameWriter:508`. Three of those exist only because the check has to be
*positioned* relative to something else — after the `V1Table` dispatch on the
two `DataFrameWriter` paths, and before `TimeTravelSpec.fromOptions` in
`RelationResolution` — and the copy inside `loadTableForV2Write` is already
redundant for `loadTableForInsert`, since `ResolveCatalogs` has rejected every
reachable CTAS/RTAS long before execution.
A single check over the analyzed plan collapses all five, and the
positioning falls out for free because the V1 fallbacks are not V2 write
commands:
```scala
// CheckAnalysis, or a small rule after ResolveCatalogs
case w: V2WriteCommand => w.table match {
case DataSourceV2Relation(_, _, Some(catalog), Some(ident), options, _) =>
CatalogV2Util.rejectTimeTravelOptionsForWrite(catalog, ident, options)
case _ =>
}
case c: V2CreateTableAsSelectPlan => c.name match {
case ResolvedIdentifier(catalog, ident) =>
CatalogV2Util.rejectTimeTravelOptionsForWrite(
catalog, ident, new CaseInsensitiveStringMap(c.writeOptions.asJava))
case _ =>
}
```
`insertInto` / `saveAsTable` on a V1 target produce `InsertIntoStatement` /
`SaveAsV1TableCommand`, so they are skipped structurally instead of by call
ordering, and a write entry point added later cannot forget the check — which
is the failure mode findings 3 and 9 were both instances of. It would also
settle the divergence @szehon-ho noted above: `insertInto` now reports
`TABLE_OR_VIEW_NOT_FOUND` on a missing target while `saveAsTable`, `save()` and
`DataFrameWriterV2` report `UNSUPPORTED_FEATURE.TIME_TRAVEL`, and a plan-level
check makes all four report the missing table.
The counter-arguments, honestly:
1. It is not literally one case. DELETE / UPDATE / MERGE arrive as
`DeleteFromTable` and the row-level plans, which are not `V2WriteCommand`s, so
the match grows and the "can't forget it" property weakens.
2. The error moves from relation resolution to after analysis, so what wins
changes: `writeTo("temp_view").option("versionAsOf", …)` would report the
temp-view error rather than `UNSUPPORTED_FEATURE.TIME_TRAVEL`, and the
temp-view case in `time travel is rejected for resolved and newly created V2
write targets` would have to change with it.
3. The check can only fire once analysis has completed, so on the `save()`
path the connector's `loadTable` — with `INSERT` / `INSERT, DELETE` requested —
has already run by the time we reject. Today it rejects first and the catalog
is never called. For a catalog that authorizes, audit-logs or pins state in
`loadTable`, that is a new side effect on a query that fails anyway.
Not asking you to rework it here. The thing I would like on the record is
that five sites now have to stay in sync, and the two that are order-sensitive
are not obviously so from the code.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala:
##########
@@ -98,6 +98,25 @@ 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")
+ df.write
+ .format("csv")
Review Comment:
**Finding 13.** The `insertInto` half of this test does execute the branch
it is meant to cover. The `saveAsTable` half never enters
`saveAsTableCommand(catalog, …)` at all.
`format("csv")` makes `lookupV2Provider()` return `None` —
`DataSource.lookupDataSourceV2:782` short-circuits on
`useV1Sources.contains(d.shortName())` and csv is in
`spark.sql.sources.useV1SourceList` — and `InMemoryTableSessionCatalog` is a
`DelegatingCatalogExtension`, i.e. a `CatalogExtension`. So `canUseV2` at
`DataFrameWriter.scala:468` is false and `saveAsTableCommand(tableName)` falls
to the `AsTableIdentifier` branch, straight into `saveAsV1TableCommand`. The
new `case Some(_: V1Table) => return` at `:502` and the reject-after-V1
ordering never run.
Measured on this head. Body of the four-arg `saveAsTableCommand` replaced by
`throw new IllegalStateException("PROBE: …")`, test still green:
```
[info] - SPARK-58389: time travel options are ignored for V1 table writes (2
seconds, 436 ms)
[info] Tests: succeeded 1, failed 0, canceled 0, ignored 0, pending 0
```
Same probe with `.format("csv")` swapped for `.format(v2Format)` — that is
the path in:
```
[info] - SPARK-58389: time travel options are ignored for V1 table writes
*** FAILED ***
[info] java.lang.IllegalStateException: PROBE: catalog saveAsTableCommand
reached
```
For contrast, a probe on `insertIntoCommand`'s `case _: V1Table` (`:364`)
does fire, so that half is effective as written:
```
[info] java.lang.IllegalStateException: PROBE: insertInto V1Table branch
reached
```
The catch is that `format(v2Format)` is not a working fix by itself — I ran
both modes without the probe and each fails for an unrelated fixture reason:
```
mode(Append) org.apache.spark.sql.AnalysisException: The format of the
existing table
spark_catalog.default.t is `CSVDataSourceV2`. It doesn't
match the specified
format `FakeV2ProviderWithCustomSchema`.
(PreprocessTableCreation, rules.scala:186)
mode(Overwrite) org.apache.spark.SparkException: [INTERNAL_ERROR]
org.apache.spark.sql.connector.FakeV2ProviderWithCustomSchema does not allow
create table as select.
(DataSource.writeAndRead:557)
```
Reaching the branch needs a format that `lookupDataSourceV2` returns *and*
whose V1 write path works against an existing csv table, which this suite has
no fixture for. `SPARK-49246: saveAsTable with v1 format` just above has the
same gap for the same reason, so this is not a regression you introduced. But
the ordering fix is specifically about `saveAsTable`, and both the
migration-guide bullet ("catalog-backed Data Source V2 writes") and the
scaladoc clause ("for V1 tables the options are ignored") lean on it, so one
test that actually executes it would be worth having. The Hive session-catalog
suites are the other place `canUseV2` is true with a V1 target. If it can't be
built, dropping the `saveAsTable` lines and naming the test after `insertInto`
would at least stop it reading as coverage it does not provide — the same
clarification you just made to the matrix test above.
##########
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("transaction catalog honors time travel context") {
Review Comment:
**Finding 16.** Seven of the nine new tests carry the ticket id
(`SPARK-58389: …`); this one and `explicit time travel specs on internal write
targets use qualified names`
(`sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala:1180`)
don't. Prefixing both with `SPARK-58389: ` keeps grep-by-ticket working across
the whole change.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala:
##########
@@ -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(
Review Comment:
**Finding 15.** The two loaders differ only in whether they validate, but
the names say "V2 write" versus "write" — both load a table for a V2 write, so
at a call site you cannot tell which one skips the check without reading both
scaladocs. The lenient one also has the stricter-sounding name of the pair.
Naming them for the actual difference would read at the call site, e.g.
`loadTableForWrite` (validating) plus
`loadTableForWriteSkippingOptionValidation`, or one method with an explicit
flag. Alternatively drop `loadTableForV2Write` and have its three callers do
what `DataFrameWriter` already does — call `rejectTimeTravelOptionsForWrite`
next to the load — which makes the validation visible at every site instead of
hidden inside one of two near-identical helpers.
Separately, while you are in here: this method is now a near-duplicate of
`getTable:507`. Same body (`new TableContext`, `extractTableStateOptions`,
`loadTable(ident, context, stateOptions)`), differing only in taking
`Set[TableWritePrivilege]` instead of the comma-joined marker string. If
`getTable` took the set and `parseWritePrivileges` moved to its single
`RelationResolution` caller, there would be one loader instead of three.
--
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]
