szehon-ho commented on code in PR #57865:
URL: https://github.com/apache/spark/pull/57865#discussion_r3788083696
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala:
##########
@@ -354,13 +358,13 @@ 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.getTableForWrite(
Review Comment:
Non-blocking, and I have traced this rather than run it, so worth a check
before acting on it.
`getTableForWrite` rejects the time-travel options as its first step, before
the returned table is inspected, so the reject lands ahead of the `V1Table`
dispatch on the line below. `saveAsTableCommand` has the same ordering --
`getTableForWrite` at `:495`, `case (_, Some(_: V1Table))` at `:501`.
That matters for `saveAsTable` in particular, because `canUseV2` there is
also true when a custom V2 session catalog is configured, regardless of format.
So `df.write.option("versionAsOf", "1").saveAsTable("hive_table")` against a V1
target would now raise `UNSUPPORTED_FEATURE.TIME_TRAVEL`, where previously
every option including that one was simply dropped on the V1 path.
Two knock-on effects if it holds. The migration-guide bullet scopes the
change to "catalog-backed Data Source V2 writes", which would not cover this.
And the `insertInto` scaladoc clause added for finding 12 says "for V1 tables
the options are ignored", which becomes slightly optimistic once one option can
raise an error instead.
Either widening the wording or moving the reject after the `V1Table`
dispatch would resolve it; the latter also keeps the V1 path behaviourally
untouched, which seems closer to the PR's stated scope.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -236,15 +246,18 @@ 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))
Review Comment:
Non-blocking, and a tail of @peter-toth's finding 10 rather than a new issue.
The option-form paths all agree now, but this check and the one three lines
above it disagree on qualification. `rejectTimeTravelOptionsForWrite` at `:250`
renders `toSQLId(ident.toQualifiedNameParts(catalog))`, while this line renders
`toSQLId(identifier)` -- the name as written. Both are checks on a persistent
write target, in the same method, with `catalog` and `ident` in scope for
either. (The third site, `resolveTempView:505`, is legitimately as-written: a
temp view has no catalog to qualify against, and the new test correctly expects
`` `temp_view` `` there.)
Worth noting the divergence came out of the fix suggested on finding 10, not
from anything that was wrong before.
Separately, I could not find a path that reaches this branch at all for a
write target. `RelationTimeTravel` is only built from the relation-primary
temporal clause (`AstBuilder:2901`, `:2994`), and the INSERT / MERGE / DELETE /
UPDATE targets plus both DataFrame writers all resolve their names through
`parseMultipartIdentifier`, so `finalTimeTravelSpec` should always be empty
when `writePrivileges != null`. If that holds, deleting the branch is cleaner
than aligning it, since `getTableForWrite` and the `:250` check now cover
everything reachable. Either way, leaving two renderings three lines apart
invites the next reader to preserve both.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala:
##########
@@ -60,9 +68,44 @@ 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 WriteStateAwareInMemoryCatalog extends InMemoryCatalog {
+ override def tableStateOptionKeys(): util.Set[String] =
util.Set.of("load-option")
+}
+
+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._
+ override protected def testCatalogClass: Class[_ <: InMemoryCatalog] =
Review Comment:
Non-blocking, fixture wiring. There are now two ways to declare state option
keys for the in-memory catalogs, and this suite uses the less flexible one.
`BasicInMemoryTableCatalog` gained a `tableStateOptionKeys` init option in
this PR, and `SupportsCatalogOptionsSuite` drives it through conf:
```scala
spark.conf.set(s"spark.sql.catalog.$catalogName.tableStateOptionKeys",
"load-option")
```
Here the same thing is done by subclassing
(`WriteStateAwareInMemoryCatalog`, `:78`) and hardcoding the override, which
also means the init option is silently ignored for `testcat` in this suite.
Two consequences worth weighing. `testCatalogClass` swaps `testcat` for the
whole file, so `load-option` is a declared state key for every test here,
including the read tests that predate this PR -- they pass because they use
`split-size`, but the next read test that happens to pick `load-option` will
behave differently for a non-obvious reason. And `registerCatalog` plus the
conf would give the same coverage without a new class or the suite-wide swap,
keeping both suites on one mechanism.
If the subclass is preferred for another reason, a one-line comment noting
that `testcat` deliberately declares `load-option` for the whole suite would
save the next reader the lookup.
##########
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:
Following up here -- the widening covers the `writeOptions` channel, and
SQL's CTAS fills the other one, so the reworded description still reads as
broader than the check.
`CreateTableAsSelect` has two option channels and the two front ends fill
opposite ones. `DataFrameWriterV2.createCommand` puts `.option()` calls in
`writeOptions` and leaves the spec's list empty (`buildTableSpec`:
`optionExpression = OptionList(Seq.empty)`), while the parser does the reverse
-- `AstBuilder:6262` passes `Map.empty` for `writeOptions` and routes `OPTIONS
(...)` into `tableSpec`. Since the new `ResolveCatalogs` cases destructure only
`writeOptions`, `writeTo(t).option("versionAsOf", ...).create()` is rejected
and `CREATE TABLE t OPTIONS ('versionAsOf' = '1') AS SELECT ...` is not. Same
for `REPLACE TABLE ... AS SELECT`.
I think keeping the boundary at the write-option channel is defensible:
`tableSpec.options` is not a per-write directive, it becomes persisted table
metadata (`CatalogV2Util.convertTableProperties` stores it both bare and under
`option.`), so policing names there would be a new policy rather than an
extension of this one. But then "including table creation and replacement" in
the migration guide reads as covering SQL CTAS. Either scope the wording to the
DataFrame writer APIs, or add a case to the new `DataSourceV2OptionSuite` test
pinning that SQL CTAS is deliberately out of scope -- as it stands that test
enumerates seven entry points and reads as an exhaustive matrix.
##########
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:
This landed in `sql/api`, but the override in `sql/core` carries the same
scaladoc and did not get the clause.
`sql/api/.../DataFrameWriter.scala:270`
> ...all options are forwarded to the write; for V1 tables the options are
ignored.
`sql/core/.../classic/DataFrameWriter.scala:320`
> ...all options are forwarded to the write.
Both carry the full doc rather than `@inheritdoc`, so the concrete class
still leaves a V1 reader without a statement. Worth adding there too -- and
while you are in it, "the options are ignored" is now slightly optimistic for
V1 as well, since the time-travel reject in `getTableForWrite` runs ahead of
the `V1Table` dispatch (I left a note on `DataFrameWriter.scala:362`), so one
option can raise an error rather than be 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]
