johanl-db commented on code in PR #56617:
URL: https://github.com/apache/spark/pull/56617#discussion_r3543694651
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala:
##########
@@ -168,66 +168,90 @@ final class DataFrameWriter[T] private[sql](ds:
Dataset[T]) extends sql.DataFram
import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
val catalogManager = df.sparkSession.sessionState.catalogManager
+
+ def createTableAsSelectCommand(
+ catalog: TableCatalog, ident: Identifier, ignoreIfExists: Boolean):
LogicalPlan = {
+ val tableSpec = UnresolvedTableSpec(
+ properties = Map.empty,
+ provider = Some(source),
+ optionExpression = OptionList(Seq.empty),
+ location = extraOptions.get("path"),
+ comment = extraOptions.get(TableCatalog.PROP_COMMENT),
+ collation = extraOptions.get(TableCatalog.PROP_COLLATION),
+ serde = None,
+ external = false,
+ constraints = Seq.empty)
+ CreateTableAsSelect(
+ UnresolvedIdentifier(
+ catalog.name +: ident.namespace.toImmutableArraySeq :+ ident.name),
+ partitioningAsV2,
+ df.queryExecution.analyzed,
+ tableSpec,
+ finalOptions,
+ ignoreIfExists = ignoreIfExists)
+ }
+
+ def appendOrOverwriteCommand(
Review Comment:
Right, this should be explicitly blocked, I added a guard + test
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala:
##########
@@ -443,3 +494,13 @@ class CatalogSupportingInMemoryTableProvider
}
}
}
+
+/** Opts out of catalog resolution, so load/save fall back to the plain
TableProvider path. */
+class CatalogResolutionOptOutProvider extends
CatalogSupportingInMemoryTableProvider {
+ override def useCatalogResolution(options: CaseInsensitiveStringMap):
Boolean = false
+}
+
+/** Opts out of failing on a missing table, enabling create-on-write for
save(). */
+class CreateOnWriteProvider extends CatalogSupportingInMemoryTableProvider {
+ override def failWriteIfTableDoesNotExist(options:
CaseInsensitiveStringMap): Boolean = false
+}
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala:
##########
@@ -168,66 +168,90 @@ final class DataFrameWriter[T] private[sql](ds:
Dataset[T]) extends sql.DataFram
import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
val catalogManager = df.sparkSession.sessionState.catalogManager
+
+ def createTableAsSelectCommand(
+ catalog: TableCatalog, ident: Identifier, ignoreIfExists: Boolean):
LogicalPlan = {
+ val tableSpec = UnresolvedTableSpec(
+ properties = Map.empty,
+ provider = Some(source),
+ optionExpression = OptionList(Seq.empty),
+ location = extraOptions.get("path"),
+ comment = extraOptions.get(TableCatalog.PROP_COMMENT),
+ collation = extraOptions.get(TableCatalog.PROP_COLLATION),
+ serde = None,
+ external = false,
+ constraints = Seq.empty)
+ CreateTableAsSelect(
+ UnresolvedIdentifier(
+ catalog.name +: ident.namespace.toImmutableArraySeq :+ ident.name),
+ partitioningAsV2,
+ df.queryExecution.analyzed,
+ tableSpec,
+ finalOptions,
+ ignoreIfExists = ignoreIfExists)
+ }
+
+ def appendOrOverwriteCommand(
+ table: Table,
+ catalog: Option[CatalogPlugin],
+ ident: Option[Identifier]): LogicalPlan = {
+ checkPartitioningMatchesV2Table(table)
+ val relation = DataSourceV2Relation.create(table, catalog, ident,
dsOptions)
+ if (curmode == SaveMode.Append) {
+ AppendData.byName(relation, df.logicalPlan, finalOptions,
_withSchemaEvolution)
+ } else {
+ // Truncate the table. TableCapabilityCheck will throw a nice
exception if this
+ // isn't supported
+ OverwriteByExpression.byName(
+ relation, df.logicalPlan, Literal(true), finalOptions,
_withSchemaEvolution)
+ }
+ }
+
curmode match {
case SaveMode.Append | SaveMode.Overwrite =>
- val (table, catalog, ident) = provider match {
- case supportsExtract: SupportsCatalogOptions =>
+ provider match {
+ case supportsExtract: SupportsCatalogOptions
+ if supportsExtract.useCatalogResolution(dsOptions) =>
val ident = supportsExtract.extractIdentifier(dsOptions)
val catalog = CatalogV2Util.getTableProviderCatalog(
supportsExtract, catalogManager, dsOptions)
-
- (catalog.loadTable(ident), Some(catalog), Some(ident))
+ val tableOpt =
+ try Some(catalog.loadTable(ident))
Review Comment:
I *think* we should pass write privileges here but that's unrelated to this
change so I'd rather keep it out, esp. since I don't fully understand the
consequences of passing privileges.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala:
##########
@@ -370,6 +372,55 @@ class SupportsCatalogOptionsSuite extends
SharedSparkSession with BeforeAndAfter
.contains("Cannot specify both version and timestamp when time
travelling the table."))
}
+ test("useCatalogResolution=false: read is resolved via the TableProvider
path, not the catalog") {
+ // The provider opts out of catalog resolution, so load() goes through
getTable instead of
+ // extractIdentifier/extractCatalog. The resulting relation therefore has
no catalog/identifier.
+ val df = spark.read.format(optOutFormat).option("name", "t1").load()
+ val relation = df.logicalPlan.collectFirst {
+ case r: DataSourceV2Relation => r
+ }.getOrElse(fail("Expected a DataSourceV2Relation"))
+ assert(relation.catalog.isEmpty && relation.identifier.isEmpty,
+ "Opting out of catalog resolution should bypass the catalog")
+ }
+
+ test("useCatalogResolution=false: a user-specified schema is allowed (no
catalog check)") {
+ // The schema check only fires on the catalog-resolution path; opting out
skips it.
+ val df = spark.read.format(optOutFormat).option("name", "t1").schema("i
int, j int").load()
+ assert(df.schema.fieldNames === Array("i", "j"))
+ }
+
+ test("failWriteIfTableDoesNotExist=false: append creates a missing table
(create-on-write)") {
+ val df = spark.range(10)
+ // t1 does not exist yet; append should create it from the query instead
of failing.
+ df.write.format(createOnWriteFormat).option("name",
"t1").option("catalog", catalogName)
+ .mode(SaveMode.Append).save()
+ assert(catalog(catalogName).tableExists("t1"), "append should have created
the table")
+ checkAnswer(load("t1", Some(catalogName)), df.toDF())
+ }
+
+ test("failWriteIfTableDoesNotExist=false: overwrite creates a missing table
(create-on-write)") {
+ val df = spark.range(10, 20)
+ df.write.format(createOnWriteFormat).option("name",
"t1").option("catalog", catalogName)
+ .mode(SaveMode.Overwrite).save()
+ assert(catalog(catalogName).tableExists("t1"), "overwrite should have
created the table")
+ checkAnswer(load("t1", Some(catalogName)), df.toDF())
+ }
+
+ test("failWriteIfTableDoesNotExist=false: append to an existing table still
appends") {
+ sql(s"create table $catalogName.t1 (id bigint) using $createOnWriteFormat")
+ spark.range(10).write.format(createOnWriteFormat).option("name", "t1")
+ .option("catalog", catalogName).mode(SaveMode.Append).save()
+ checkAnswer(load("t1", Some(catalogName)), spark.range(10).toDF())
+ }
+
+ test("append to a missing table fails by default
(failWriteIfTableDoesNotExist=true)") {
+ // The default provider keeps the prior behavior: append/overwrite to a
missing table fails.
+ intercept[NoSuchTableException] {
+ spark.range(10).write.format(format).option("name",
"t1").option("catalog", catalogName)
+ .mode(SaveMode.Append).save()
+ }
+ }
+
Review Comment:
Added tests
--
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]
