You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2024/08/21 02:53:49 UTC
(spark) branch master updated: [SPARK-49246][SQL] TableCatalog#loadTable should indicate if it's for writing
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new b6164e6ed717 [SPARK-49246][SQL] TableCatalog#loadTable should indicate if it's for writing
b6164e6ed717 is described below
commit b6164e6ed7174201057cf8f9ad59f59d6f60f089
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Wed Aug 21 10:53:31 2024 +0800
[SPARK-49246][SQL] TableCatalog#loadTable should indicate if it's for writing
### What changes were proposed in this pull request?
For custom catalogs that have access control, read and write permissions can be different. However, currently Spark always call `TableCatalog#loadTable` to look up the table, no matter it's for read or write.
This PR adds a variant of `loadTable`: `loadTableForWrite`, in `TableCatalog`. All the write commands will call this new method to look up tables instead. This new method has a default implementation that just calls `loadTable`, so there is no breaking change.
### Why are the changes needed?
allow more fine-grained access control for custom catalogs.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
new tests
### Was this patch authored or co-authored using generative AI tooling?
no
Closes #47772 from cloud-fan/write.
Lead-authored-by: Wenchen Fan <we...@databricks.com>
Co-authored-by: Wenchen Fan <cl...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/connector/catalog/TableCatalog.java | 20 +++++
.../sql/connector/catalog/TableWritePrivilege.java | 40 ++++++++++
.../spark/sql/catalyst/analysis/Analyzer.scala | 8 +-
.../spark/sql/catalyst/analysis/unresolved.scala | 27 +++++++
.../spark/sql/catalyst/parser/AstBuilder.scala | 56 ++++++++-----
.../sql/catalyst/plans/logical/v2Commands.scala | 15 ++++
.../sql/connector/catalog/CatalogV2Util.scala | 17 +++-
.../spark/sql/catalyst/parser/DDLParserSuite.scala | 29 ++++---
.../sql/catalyst/parser/PlanParserSuite.scala | 93 ++++++++++++----------
.../org/apache/spark/sql/DataFrameWriter.scala | 18 +++--
.../org/apache/spark/sql/DataFrameWriterV2.scala | 11 ++-
.../org/apache/spark/sql/MergeIntoWriter.scala | 3 +-
.../catalyst/analysis/ResolveSessionCatalog.scala | 4 +-
.../datasources/v2/WriteToDataSourceV2Exec.scala | 8 +-
.../sql-tests/analyzer-results/explain-aqe.sql.out | 2 +-
.../sql-tests/analyzer-results/explain.sql.out | 2 +-
.../sql-tests/results/explain-aqe.sql.out | 2 +-
.../resources/sql-tests/results/explain.sql.out | 2 +-
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 84 +++++++++++++++++++
.../command/AlignAssignmentsSuiteBase.scala | 4 +-
.../execution/command/PlanResolutionSuite.scala | 6 +-
21 files changed, 351 insertions(+), 100 deletions(-)
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
index facfc0d774e8..ad4fe743218f 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
@@ -110,6 +110,26 @@ public interface TableCatalog extends CatalogPlugin {
*/
Table loadTable(Identifier ident) throws NoSuchTableException;
+ /**
+ * Load table metadata by {@link Identifier identifier} from the catalog. Spark will write data
+ * into this table later.
+ * <p>
+ * If the catalog supports views and contains a view for the identifier and not a table, this
+ * must throw {@link NoSuchTableException}.
+ *
+ * @param ident a table identifier
+ * @param writePrivileges
+ * @return the table's metadata
+ * @throws NoSuchTableException If the table doesn't exist or is a view
+ *
+ * @since 3.5.3
+ */
+ default Table loadTable(
+ Identifier ident,
+ Set<TableWritePrivilege> writePrivileges) throws NoSuchTableException {
+ return loadTable(ident);
+ }
+
/**
* Load table metadata of a specific version by {@link Identifier identifier} from the catalog.
* <p>
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableWritePrivilege.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableWritePrivilege.java
new file mode 100644
index 000000000000..ca2d4ba9e7b4
--- /dev/null
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableWritePrivilege.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+/**
+ * The table write privileges that will be provided when loading a table.
+ *
+ * @since 3.5.3
+ */
+public enum TableWritePrivilege {
+ /**
+ * The privilege for adding rows to the table.
+ */
+ INSERT,
+
+ /**
+ * The privilege for changing existing rows in th table.
+ */
+ UPDATE,
+
+ /**
+ * The privilege for deleting rows from the table.
+ */
+ DELETE
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 20546c5c5be3..92ab804f0a70 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -1331,8 +1331,12 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
cachedConnectRelation
}.getOrElse(cachedRelation)
}.orElse {
- val table = CatalogV2Util.loadTable(catalog, ident, finalTimeTravelSpec)
- val loaded = createRelation(catalog, ident, table, u.options, u.isStreaming)
+ val writePrivilegesString =
+ Option(u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES))
+ val table = CatalogV2Util.loadTable(
+ catalog, ident, finalTimeTravelSpec, writePrivilegesString)
+ val loaded = createRelation(
+ catalog, ident, table, u.clearWritePrivileges.options, u.isStreaming)
loaded.foreach(AnalysisContext.get.relationCache.update(key, _))
u.getTagValue(LogicalPlan.PLAN_ID_TAG).map { planId =>
loaded.map { loadedRelation =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
index eb718eec4256..60d979e9c7af 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef, LeafNode, Lo
import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.types.{DataType, Metadata, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -134,10 +135,36 @@ case class UnresolvedRelation(
override def name: String = tableName
+ def requireWritePrivileges(privileges: Seq[TableWritePrivilege]): UnresolvedRelation = {
+ if (privileges.nonEmpty) {
+ val newOptions = new java.util.HashMap[String, String]
+ newOptions.putAll(options)
+ newOptions.put(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES, privileges.mkString(","))
+ copy(options = new CaseInsensitiveStringMap(newOptions))
+ } else {
+ this
+ }
+ }
+
+ def clearWritePrivileges: UnresolvedRelation = {
+ if (options.containsKey(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)) {
+ val newOptions = new java.util.HashMap[String, String]
+ newOptions.putAll(options)
+ newOptions.remove(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
+ copy(options = new CaseInsensitiveStringMap(newOptions))
+ } else {
+ this
+ }
+ }
+
final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_RELATION)
}
object UnresolvedRelation {
+ // An internal option of `UnresolvedRelation` to specify the required write privileges when
+ // writing data to this relation.
+ val REQUIRED_WRITE_PRIVILEGES = "__required_write_privileges__"
+
def apply(
tableIdentifier: TableIdentifier,
extraOptions: CaseInsensitiveStringMap,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index 038f15ee1103..b0922542c562 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -45,7 +45,7 @@ import org.apache.spark.sql.catalyst.trees.TreePattern.PARAMETER
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, DateTimeUtils, IntervalUtils}
import org.apache.spark.sql.catalyst.util.DateTimeUtils.{convertSpecialDate, convertSpecialTimestamp, convertSpecialTimestampNTZ, getZoneId, stringToDate, stringToTimestamp, stringToTimestampWithoutTimeZone}
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, SupportsNamespaces, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, SupportsNamespaces, TableCatalog, TableWritePrivilege}
import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition
import org.apache.spark.sql.connector.expressions.{ApplyTransform, BucketTransform, DaysTransform, Expression => V2Expression, FieldReference, HoursTransform, IdentityTransform, LiteralValue, MonthsTransform, Transform, YearsTransform}
import org.apache.spark.sql.errors.{DataTypeErrorsBase, QueryCompilationErrors, QueryParsingErrors, SqlScriptingErrors}
@@ -469,7 +469,7 @@ class AstBuilder extends DataTypeAstBuilder
= visitInsertIntoTable(table)
withIdentClause(relationCtx, Seq(query), (ident, otherPlans) => {
InsertIntoStatement(
- createUnresolvedRelation(relationCtx, ident, options),
+ createUnresolvedRelation(relationCtx, ident, options, Seq(TableWritePrivilege.INSERT)),
partition,
cols,
otherPlans.head,
@@ -482,7 +482,8 @@ class AstBuilder extends DataTypeAstBuilder
= visitInsertOverwriteTable(table)
withIdentClause(relationCtx, Seq(query), (ident, otherPlans) => {
InsertIntoStatement(
- createUnresolvedRelation(relationCtx, ident, options),
+ createUnresolvedRelation(relationCtx, ident, options,
+ Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)),
partition,
cols,
otherPlans.head,
@@ -491,9 +492,11 @@ class AstBuilder extends DataTypeAstBuilder
byName)
})
case ctx: InsertIntoReplaceWhereContext =>
+ val options = Option(ctx.optionsClause())
withIdentClause(ctx.identifierReference, Seq(query), (ident, otherPlans) => {
OverwriteByExpression.byPosition(
- createUnresolvedRelation(ctx.identifierReference, ident, Option(ctx.optionsClause())),
+ createUnresolvedRelation(ctx.identifierReference, ident, options,
+ Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)),
otherPlans.head,
expression(ctx.whereClause().booleanExpression()))
})
@@ -578,7 +581,8 @@ class AstBuilder extends DataTypeAstBuilder
override def visitDeleteFromTable(
ctx: DeleteFromTableContext): LogicalPlan = withOrigin(ctx) {
- val table = createUnresolvedRelation(ctx.identifierReference)
+ val table = createUnresolvedRelation(
+ ctx.identifierReference, writePrivileges = Seq(TableWritePrivilege.DELETE))
val tableAlias = getTableAliasWithoutColumnAlias(ctx.tableAlias(), "DELETE")
val aliasedTable = tableAlias.map(SubqueryAlias(_, table)).getOrElse(table)
val predicate = if (ctx.whereClause() != null) {
@@ -590,7 +594,8 @@ class AstBuilder extends DataTypeAstBuilder
}
override def visitUpdateTable(ctx: UpdateTableContext): LogicalPlan = withOrigin(ctx) {
- val table = createUnresolvedRelation(ctx.identifierReference)
+ val table = createUnresolvedRelation(
+ ctx.identifierReference, writePrivileges = Seq(TableWritePrivilege.UPDATE))
val tableAlias = getTableAliasWithoutColumnAlias(ctx.tableAlias(), "UPDATE")
val aliasedTable = tableAlias.map(SubqueryAlias(_, table)).getOrElse(table)
val assignments = withAssignments(ctx.setClause().assignmentList())
@@ -613,9 +618,6 @@ class AstBuilder extends DataTypeAstBuilder
override def visitMergeIntoTable(ctx: MergeIntoTableContext): LogicalPlan = withOrigin(ctx) {
val withSchemaEvolution = ctx.EVOLUTION() != null
- val targetTable = createUnresolvedRelation(ctx.target)
- val targetTableAlias = getTableAliasWithoutColumnAlias(ctx.targetAlias, "MERGE")
- val aliasedTarget = targetTableAlias.map(SubqueryAlias(_, targetTable)).getOrElse(targetTable)
val sourceTableOrQuery = if (ctx.source != null) {
createUnresolvedRelation(ctx.source)
@@ -646,7 +648,7 @@ class AstBuilder extends DataTypeAstBuilder
s"Unrecognized matched action: ${clause.matchedAction().getText}")
}
}
- }
+ }.toSeq
val notMatchedActions = ctx.notMatchedClause().asScala.map {
clause => {
if (clause.notMatchedAction().INSERT() != null) {
@@ -667,7 +669,7 @@ class AstBuilder extends DataTypeAstBuilder
s"Unrecognized matched action: ${clause.notMatchedAction().getText}")
}
}
- }
+ }.toSeq
val notMatchedBySourceActions = ctx.notMatchedBySourceClause().asScala.map {
clause => {
val notMatchedBySourceAction = clause.notMatchedBySourceAction()
@@ -682,7 +684,7 @@ class AstBuilder extends DataTypeAstBuilder
s"Unrecognized matched action: ${clause.notMatchedBySourceAction().getText}")
}
}
- }
+ }.toSeq
if (matchedActions.isEmpty && notMatchedActions.isEmpty && notMatchedBySourceActions.isEmpty) {
throw QueryParsingErrors.mergeStatementWithoutWhenClauseError(ctx)
}
@@ -701,13 +703,19 @@ class AstBuilder extends DataTypeAstBuilder
throw QueryParsingErrors.nonLastNotMatchedBySourceClauseOmitConditionError(ctx)
}
+ val targetTable = createUnresolvedRelation(
+ ctx.target,
+ writePrivileges = MergeIntoTable.getWritePrivileges(
+ matchedActions, notMatchedActions, notMatchedBySourceActions))
+ val targetTableAlias = getTableAliasWithoutColumnAlias(ctx.targetAlias, "MERGE")
+ val aliasedTarget = targetTableAlias.map(SubqueryAlias(_, targetTable)).getOrElse(targetTable)
MergeIntoTable(
aliasedTarget,
aliasedSource,
mergeCondition,
- matchedActions.toSeq,
- notMatchedActions.toSeq,
- notMatchedBySourceActions.toSeq,
+ matchedActions,
+ notMatchedActions,
+ notMatchedBySourceActions,
withSchemaEvolution)
}
@@ -3130,10 +3138,13 @@ class AstBuilder extends DataTypeAstBuilder
*/
private def createUnresolvedRelation(
ctx: IdentifierReferenceContext,
- optionsClause: Option[OptionsClauseContext] = None): LogicalPlan = withOrigin(ctx) {
+ optionsClause: Option[OptionsClauseContext] = None,
+ writePrivileges: Seq[TableWritePrivilege] = Nil): LogicalPlan = withOrigin(ctx) {
val options = resolveOptions(optionsClause)
- withIdentClause(ctx, parts =>
- new UnresolvedRelation(parts, options, isStreaming = false))
+ withIdentClause(ctx, parts => {
+ val relation = new UnresolvedRelation(parts, options, isStreaming = false)
+ relation.requireWritePrivileges(writePrivileges)
+ })
}
/**
@@ -3142,9 +3153,11 @@ class AstBuilder extends DataTypeAstBuilder
private def createUnresolvedRelation(
ctx: ParserRuleContext,
ident: Seq[String],
- optionsClause: Option[OptionsClauseContext]): UnresolvedRelation = withOrigin(ctx) {
+ optionsClause: Option[OptionsClauseContext],
+ writePrivileges: Seq[TableWritePrivilege]): UnresolvedRelation = withOrigin(ctx) {
val options = resolveOptions(optionsClause)
- new UnresolvedRelation(ident, options, isStreaming = false)
+ val relation = new UnresolvedRelation(ident, options, isStreaming = false)
+ relation.requireWritePrivileges(writePrivileges)
}
private def resolveOptions(
@@ -5020,7 +5033,8 @@ class AstBuilder extends DataTypeAstBuilder
if (query.isDefined) {
CacheTableAsSelect(ident.head, query.get, source(ctx.query()), isLazy, options)
} else {
- CacheTable(createUnresolvedRelation(ctx.identifierReference, ident, None),
+ CacheTable(
+ createUnresolvedRelation(ctx.identifierReference, ident, None, writePrivileges = Nil),
ident, isLazy, options)
}
})
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
index 6339a18796fa..05628d7b1c98 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
@@ -794,6 +794,21 @@ case class MergeIntoTable(
copy(targetTable = newLeft, sourceTable = newRight)
}
+object MergeIntoTable {
+ def getWritePrivileges(
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction],
+ notMatchedBySourceActions: Seq[MergeAction]): Seq[TableWritePrivilege] = {
+ val privileges = scala.collection.mutable.HashSet.empty[TableWritePrivilege]
+ (matchedActions.iterator ++ notMatchedActions ++ notMatchedBySourceActions).foreach {
+ case _: DeleteAction => privileges.add(TableWritePrivilege.DELETE)
+ case _: UpdateAction | _: UpdateStarAction => privileges.add(TableWritePrivilege.UPDATE)
+ case _: InsertAction | _: InsertStarAction => privileges.add(TableWritePrivilege.INSERT)
+ }
+ privileges.toSeq
+ }
+}
+
sealed abstract class MergeAction extends Expression with Unevaluable {
def condition: Option[Expression]
override def nullable: Boolean = false
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
index 283c550c4556..6698f0a02140 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
@@ -403,9 +403,10 @@ private[sql] object CatalogV2Util {
def loadTable(
catalog: CatalogPlugin,
ident: Identifier,
- timeTravelSpec: Option[TimeTravelSpec] = None): Option[Table] =
+ timeTravelSpec: Option[TimeTravelSpec] = None,
+ writePrivilegesString: Option[String] = None): Option[Table] =
try {
- Option(getTable(catalog, ident, timeTravelSpec))
+ Option(getTable(catalog, ident, timeTravelSpec, writePrivilegesString))
} catch {
case _: NoSuchTableException => None
case _: NoSuchDatabaseException => None
@@ -414,8 +415,10 @@ private[sql] object CatalogV2Util {
def getTable(
catalog: CatalogPlugin,
ident: Identifier,
- timeTravelSpec: Option[TimeTravelSpec] = None): Table = {
+ timeTravelSpec: Option[TimeTravelSpec] = None,
+ writePrivilegesString: Option[String] = None): Table = {
if (timeTravelSpec.nonEmpty) {
+ assert(writePrivilegesString.isEmpty, "Should not write to a table with time travel")
timeTravelSpec.get match {
case v: AsOfVersion =>
catalog.asTableCatalog.loadTable(ident, v.version)
@@ -423,7 +426,13 @@ private[sql] object CatalogV2Util {
catalog.asTableCatalog.loadTable(ident, ts.timestamp)
}
} else {
- catalog.asTableCatalog.loadTable(ident)
+ if (writePrivilegesString.isDefined) {
+ val writePrivileges = writePrivilegesString.get.split(",").map(_.trim)
+ .map(TableWritePrivilege.valueOf).toSet.asJava
+ catalog.asTableCatalog.loadTable(ident, writePrivileges)
+ } else {
+ catalog.asTableCatalog.loadTable(ident)
+ }
}
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
index c930292f2793..756ec95c70d2 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
@@ -39,7 +39,16 @@ class DDLParserSuite extends AnalysisTest {
}
private def parseCompare(sql: String, expected: LogicalPlan): Unit = {
- comparePlans(parsePlan(sql), expected, checkAnalysis = false)
+ // We don't care the write privileges in this suite.
+ val parsed = parsePlan(sql).transform {
+ case u: UnresolvedRelation => u.clearWritePrivileges
+ case i: InsertIntoStatement =>
+ i.table match {
+ case u: UnresolvedRelation => i.copy(table = u.clearWritePrivileges)
+ case _ => i
+ }
+ }
+ comparePlans(parsed, expected, checkAnalysis = false)
}
private def internalException(sqlText: String): SparkThrowable = {
@@ -2635,20 +2644,20 @@ class DDLParserSuite extends AnalysisTest {
withSQLConf(
SQLConf.EAGER_EVAL_OF_UNRESOLVED_INLINE_TABLE_ENABLED.key ->
optimizeInsertIntoValues.toString) {
- comparePlans(parsePlan(dateTypeSql), insertPartitionPlan(
+ parseCompare(dateTypeSql, insertPartitionPlan(
"2019-01-02", optimizeInsertIntoValues))
withSQLConf(SQLConf.LEGACY_INTERVAL_ENABLED.key -> "true") {
- comparePlans(parsePlan(intervalTypeSql), insertPartitionPlan(
+ parseCompare(intervalTypeSql, insertPartitionPlan(
interval, optimizeInsertIntoValues))
}
- comparePlans(parsePlan(ymIntervalTypeSql), insertPartitionPlan(
+ parseCompare(ymIntervalTypeSql, insertPartitionPlan(
"INTERVAL '1-2' YEAR TO MONTH", optimizeInsertIntoValues))
- comparePlans(parsePlan(dtIntervalTypeSql),
+ parseCompare(dtIntervalTypeSql,
insertPartitionPlan(
"INTERVAL '1 02:03:04.128462' DAY TO SECOND", optimizeInsertIntoValues))
- comparePlans(parsePlan(timestampTypeSql), insertPartitionPlan(
+ parseCompare(timestampTypeSql, insertPartitionPlan(
timestamp, optimizeInsertIntoValues))
- comparePlans(parsePlan(binaryTypeSql), insertPartitionPlan(
+ parseCompare(binaryTypeSql, insertPartitionPlan(
binaryStr, optimizeInsertIntoValues))
}
}
@@ -2748,12 +2757,12 @@ class DDLParserSuite extends AnalysisTest {
// In each of the following cases, the DEFAULT reference parses as an unresolved attribute
// reference. We can handle these cases after the parsing stage, at later phases of analysis.
- comparePlans(parsePlan("VALUES (1, 2, DEFAULT) AS val"),
+ parseCompare("VALUES (1, 2, DEFAULT) AS val",
SubqueryAlias("val",
UnresolvedInlineTable(Seq("col1", "col2", "col3"), Seq(Seq(Literal(1), Literal(2),
UnresolvedAttribute("DEFAULT"))))))
- comparePlans(parsePlan(
- "INSERT INTO t PARTITION(part = date'2019-01-02') VALUES ('a', DEFAULT)"),
+ parseCompare(
+ "INSERT INTO t PARTITION(part = date'2019-01-02') VALUES ('a', DEFAULT)",
InsertIntoStatement(
UnresolvedRelation(Seq("t")),
Map("part" -> Some("2019-01-02")),
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
index e0217a5637a8..a6a32e87b742 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
@@ -40,7 +40,16 @@ class PlanParserSuite extends AnalysisTest {
import org.apache.spark.sql.catalyst.dsl.plans._
private def assertEqual(sqlCommand: String, plan: LogicalPlan): Unit = {
- comparePlans(parsePlan(sqlCommand), plan, checkAnalysis = false)
+ // We don't care the write privileges in this suite.
+ val parsed = parsePlan(sqlCommand).transform {
+ case u: UnresolvedRelation => u.clearWritePrivileges
+ case i: InsertIntoStatement =>
+ i.table match {
+ case u: UnresolvedRelation => i.copy(table = u.clearWritePrivileges)
+ case _ => i
+ }
+ }
+ comparePlans(parsed, plan, checkAnalysis = false)
}
private def parseException(sqlText: String): SparkThrowable = {
@@ -1048,57 +1057,56 @@ class PlanParserSuite extends AnalysisTest {
errorClass = "PARSE_SYNTAX_ERROR",
parameters = Map("error" -> "'b'", "hint" -> ""))
- comparePlans(
- parsePlan("SELECT /*+ HINT */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ HINT */ * FROM t",
UnresolvedHint("HINT", Seq.empty, table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ BROADCASTJOIN(u) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ BROADCASTJOIN(u) */ * FROM t",
UnresolvedHint("BROADCASTJOIN", Seq($"u"), table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ MAPJOIN(u) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ MAPJOIN(u) */ * FROM t",
UnresolvedHint("MAPJOIN", Seq($"u"), table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ STREAMTABLE(a,b,c) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ STREAMTABLE(a,b,c) */ * FROM t",
UnresolvedHint("STREAMTABLE", Seq($"a", $"b", $"c"), table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ INDEX(t, emp_job_ix) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ INDEX(t, emp_job_ix) */ * FROM t",
UnresolvedHint("INDEX", Seq($"t", $"emp_job_ix"), table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ MAPJOIN(`default.t`) */ * from `default.t`"),
+ assertEqual(
+ "SELECT /*+ MAPJOIN(`default.t`) */ * from `default.t`",
UnresolvedHint("MAPJOIN", Seq(UnresolvedAttribute.quoted("default.t")),
table("default.t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ MAPJOIN(t) */ a from t where true group by a order by a"),
+ assertEqual(
+ "SELECT /*+ MAPJOIN(t) */ a from t where true group by a order by a",
UnresolvedHint("MAPJOIN", Seq($"t"),
table("t").where(Literal(true)).groupBy($"a")($"a")).orderBy($"a".asc))
- comparePlans(
- parsePlan("SELECT /*+ COALESCE(10) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ COALESCE(10) */ * FROM t",
UnresolvedHint("COALESCE", Seq(Literal(10)),
table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ REPARTITION(100) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ REPARTITION(100) */ * FROM t",
UnresolvedHint("REPARTITION", Seq(Literal(100)),
table("t").select(star())))
- comparePlans(
- parsePlan(
- "INSERT INTO s SELECT /*+ REPARTITION(100), COALESCE(500), COALESCE(10) */ * FROM t"),
+ assertEqual(
+ "INSERT INTO s SELECT /*+ REPARTITION(100), COALESCE(500), COALESCE(10) */ * FROM t",
InsertIntoStatement(table("s"), Map.empty, Nil,
UnresolvedHint("REPARTITION", Seq(Literal(100)),
UnresolvedHint("COALESCE", Seq(Literal(500)),
UnresolvedHint("COALESCE", Seq(Literal(10)),
table("t").select(star())))), overwrite = false, ifPartitionNotExists = false))
- comparePlans(
- parsePlan("SELECT /*+ BROADCASTJOIN(u), REPARTITION(100) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ BROADCASTJOIN(u), REPARTITION(100) */ * FROM t",
UnresolvedHint("BROADCASTJOIN", Seq($"u"),
UnresolvedHint("REPARTITION", Seq(Literal(100)),
table("t").select(star()))))
@@ -1109,49 +1117,48 @@ class PlanParserSuite extends AnalysisTest {
errorClass = "PARSE_SYNTAX_ERROR",
parameters = Map("error" -> "'+'", "hint" -> ""))
- comparePlans(
- parsePlan("SELECT /*+ REPARTITION(c) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ REPARTITION(c) */ * FROM t",
UnresolvedHint("REPARTITION", Seq(UnresolvedAttribute("c")),
table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ REPARTITION(100, c) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ REPARTITION(100, c) */ * FROM t",
UnresolvedHint("REPARTITION", Seq(Literal(100), UnresolvedAttribute("c")),
table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ REPARTITION(100, c), COALESCE(50) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ REPARTITION(100, c), COALESCE(50) */ * FROM t",
UnresolvedHint("REPARTITION", Seq(Literal(100), UnresolvedAttribute("c")),
UnresolvedHint("COALESCE", Seq(Literal(50)),
table("t").select(star()))))
- comparePlans(
- parsePlan("SELECT /*+ REPARTITION(100, c), BROADCASTJOIN(u), COALESCE(50) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ REPARTITION(100, c), BROADCASTJOIN(u), COALESCE(50) */ * FROM t",
UnresolvedHint("REPARTITION", Seq(Literal(100), UnresolvedAttribute("c")),
UnresolvedHint("BROADCASTJOIN", Seq($"u"),
UnresolvedHint("COALESCE", Seq(Literal(50)),
table("t").select(star())))))
- comparePlans(
- parsePlan(
- """
- |SELECT
- |/*+ REPARTITION(100, c), BROADCASTJOIN(u), COALESCE(50), REPARTITION(300, c) */
- |* FROM t
- """.stripMargin),
+ assertEqual(
+ """
+ |SELECT
+ |/*+ REPARTITION(100, c), BROADCASTJOIN(u), COALESCE(50), REPARTITION(300, c) */
+ |* FROM t
+ """.stripMargin,
UnresolvedHint("REPARTITION", Seq(Literal(100), UnresolvedAttribute("c")),
UnresolvedHint("BROADCASTJOIN", Seq($"u"),
UnresolvedHint("COALESCE", Seq(Literal(50)),
UnresolvedHint("REPARTITION", Seq(Literal(300), UnresolvedAttribute("c")),
table("t").select(star()))))))
- comparePlans(
- parsePlan("SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t",
UnresolvedHint("REPARTITION_BY_RANGE", Seq(UnresolvedAttribute("c")),
table("t").select(star())))
- comparePlans(
- parsePlan("SELECT /*+ REPARTITION_BY_RANGE(100, c) */ * FROM t"),
+ assertEqual(
+ "SELECT /*+ REPARTITION_BY_RANGE(100, c) */ * FROM t",
UnresolvedHint("REPARTITION_BY_RANGE", Seq(Literal(100), UnresolvedAttribute("c")),
table("t").select(star())))
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index 991487170f17..60734efbf5bb 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -30,12 +30,15 @@ import org.apache.spark.sql.catalyst.plans.logical.{AppendData, CreateTableAsSel
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.connector.catalog.{CatalogPlugin, CatalogV2Implicits, CatalogV2Util, Identifier, SupportsCatalogOptions, Table, TableCatalog, TableProvider, V1Table}
import org.apache.spark.sql.connector.catalog.TableCapability._
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege._
import org.apache.spark.sql.connector.expressions.{ClusterByTransform, FieldReference, IdentityTransform, Transform}
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.execution.command.DDLUtils
import org.apache.spark.sql.execution.datasources.{CreateTable, DataSource, DataSourceUtils, LogicalRelation}
import org.apache.spark.sql.execution.datasources.v2._
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.sql.types.StructType
@@ -473,7 +476,7 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
private def insertInto(catalog: CatalogPlugin, ident: Identifier): Unit = {
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
- val table = catalog.asTableCatalog.loadTable(ident) match {
+ val table = catalog.asTableCatalog.loadTable(ident, getWritePrivileges.toSet.asJava) match {
case _: V1Table =>
return insertInto(TableIdentifier(ident.name(), ident.namespace().headOption))
case t =>
@@ -504,7 +507,7 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
private def insertInto(tableIdent: TableIdentifier): Unit = {
runCommand(df.sparkSession) {
InsertIntoStatement(
- table = UnresolvedRelation(tableIdent),
+ table = UnresolvedRelation(tableIdent).requireWritePrivileges(getWritePrivileges),
partitionSpec = Map.empty[String, Option[String]],
Nil,
query = df.logicalPlan,
@@ -513,6 +516,11 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
}
}
+ private def getWritePrivileges: Seq[TableWritePrivilege] = mode match {
+ case SaveMode.Overwrite => Seq(INSERT, DELETE)
+ case _ => Seq(INSERT)
+ }
+
private def getBucketSpec: Option[BucketSpec] = {
if (sortColumnNames.isDefined && numBuckets.isEmpty) {
throw QueryCompilationErrors.sortByWithoutBucketingError()
@@ -588,7 +596,8 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
val session = df.sparkSession
- val canUseV2 = lookupV2Provider().isDefined
+ val canUseV2 = lookupV2Provider().isDefined ||
+ df.sparkSession.sessionState.conf.getConf(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION).isDefined
session.sessionState.sqlParser.parseMultipartIdentifier(tableName) match {
case nameParts @ NonSessionCatalogAndIdentifier(catalog, ident) =>
@@ -609,7 +618,7 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
private def saveAsTable(
catalog: TableCatalog, ident: Identifier, nameParts: Seq[String]): Unit = {
- val tableOpt = try Option(catalog.loadTable(ident)) catch {
+ val tableOpt = try Option(catalog.loadTable(ident, getWritePrivileges.toSet.asJava)) catch {
case _: NoSuchTableException => None
}
@@ -670,7 +679,6 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
val catalog = df.sparkSession.sessionState.catalog
val qualifiedIdent = catalog.qualifyIdentifier(tableIdent)
val tableExists = catalog.tableExists(qualifiedIdent)
- val tableName = qualifiedIdent.unquotedString
(tableExists, mode) match {
case (true, SaveMode.Ignore) =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala
index d9ad0003a525..576d8276b56e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala
@@ -24,6 +24,7 @@ import org.apache.spark.annotation.Experimental
import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, NoSuchTableException, TableAlreadyExistsException, UnresolvedFunction, UnresolvedIdentifier, UnresolvedRelation}
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, Literal}
import org.apache.spark.sql.catalyst.plans.logical.{AppendData, CreateTableAsSelect, LogicalPlan, OptionList, OverwriteByExpression, OverwritePartitionsDynamic, ReplaceTableAsSelect, UnresolvedTableSpec}
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege._
import org.apache.spark.sql.connector.expressions.{ClusterByTransform, FieldReference, LogicalExpressions, NamedReference, Transform}
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.QueryExecution
@@ -169,7 +170,9 @@ final class DataFrameWriterV2[T] private[sql](table: String, ds: Dataset[T])
*/
@throws(classOf[NoSuchTableException])
def append(): Unit = {
- val append = AppendData.byName(UnresolvedRelation(tableName), logicalPlan, options.toMap)
+ val append = AppendData.byName(
+ UnresolvedRelation(tableName).requireWritePrivileges(Seq(INSERT)),
+ logicalPlan, options.toMap)
runCommand(append)
}
@@ -186,7 +189,8 @@ final class DataFrameWriterV2[T] private[sql](table: String, ds: Dataset[T])
@throws(classOf[NoSuchTableException])
def overwrite(condition: Column): Unit = {
val overwrite = OverwriteByExpression.byName(
- UnresolvedRelation(tableName), logicalPlan, expression(condition), options.toMap)
+ UnresolvedRelation(tableName).requireWritePrivileges(Seq(INSERT, DELETE)),
+ logicalPlan, expression(condition), options.toMap)
runCommand(overwrite)
}
@@ -206,7 +210,8 @@ final class DataFrameWriterV2[T] private[sql](table: String, ds: Dataset[T])
@throws(classOf[NoSuchTableException])
def overwritePartitions(): Unit = {
val dynamicOverwrite = OverwritePartitionsDynamic.byName(
- UnresolvedRelation(tableName), logicalPlan, options.toMap)
+ UnresolvedRelation(tableName).requireWritePrivileges(Seq(INSERT, DELETE)),
+ logicalPlan, options.toMap)
runCommand(dynamicOverwrite)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala
index d8042720577d..6212a7fdb259 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala
@@ -189,7 +189,8 @@ class MergeIntoWriter[T] private[sql] (
}
val merge = MergeIntoTable(
- UnresolvedRelation(tableName),
+ UnresolvedRelation(tableName).requireWritePrivileges(MergeIntoTable.getWritePrivileges(
+ matchedActions, notMatchedActions, notMatchedBySourceActions)),
logicalPlan,
on.expr,
matchedActions,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
index 9053fb9cc73f..20e3b4e980f2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
@@ -683,7 +683,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
}
private def supportsV1Command(catalog: CatalogPlugin): Boolean = {
- catalog.name().equalsIgnoreCase(CatalogManager.SESSION_CATALOG_NAME) &&
- !SQLConf.get.getConf(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION).isDefined
+ isSessionCatalog(catalog) &&
+ SQLConf.get.getConf(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION).isEmpty
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
index 5632595de7cf..89372017257d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.{AppendData, LogicalPlan, TableSpec, UnaryNode}
import org.apache.spark.sql.catalyst.util.{removeInternalMetadata, CharVarcharUtils, WriteDeltaProjections}
import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{DELETE_OPERATION, INSERT_OPERATION, UPDATE_OPERATION}
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, StagedTable, StagingTableCatalog, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, StagedTable, StagingTableCatalog, Table, TableCatalog, TableWritePrivilege}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter, DataWriterFactory, DeltaWrite, DeltaWriter, PhysicalWriteInfoImpl, Write, WriterCommitMessage}
@@ -84,7 +84,8 @@ case class CreateTableAsSelectExec(
}
val table = Option(catalog.createTable(
ident, getV2Columns(query.schema, catalog.useNullableQuerySchema),
- partitioning.toArray, properties.asJava)).getOrElse(catalog.loadTable(ident))
+ partitioning.toArray, properties.asJava)
+ ).getOrElse(catalog.loadTable(ident, Set(TableWritePrivilege.INSERT).asJava))
writeToTable(catalog, table, writeOptions, ident, query)
}
}
@@ -164,7 +165,8 @@ case class ReplaceTableAsSelectExec(
}
val table = Option(catalog.createTable(
ident, getV2Columns(query.schema, catalog.useNullableQuerySchema),
- partitioning.toArray, properties.asJava)).getOrElse(catalog.loadTable(ident))
+ partitioning.toArray, properties.asJava)
+ ).getOrElse(catalog.loadTable(ident, Set(TableWritePrivilege.INSERT).asJava))
writeToTable(catalog, table, writeOptions, ident, query)
}
}
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out
index 3aea86b232cb..f9a282c2b927 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out
@@ -196,7 +196,7 @@ ExplainCommand 'Aggregate ['key], ['key, unresolvedalias('MIN('val))], Formatted
-- !query
EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
-- !query analysis
-ExplainCommand 'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [], false, false, false, false, ExtendedMode
+ExplainCommand 'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false, ExtendedMode
-- !query
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out
index 3aea86b232cb..f9a282c2b927 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out
@@ -196,7 +196,7 @@ ExplainCommand 'Aggregate ['key], ['key, unresolvedalias('MIN('val))], Formatted
-- !query
EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
-- !query analysis
-ExplainCommand 'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [], false, false, false, false, ExtendedMode
+ExplainCommand 'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false, ExtendedMode
-- !query
diff --git a/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out b/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out
index 3830b47ba8a6..16077a78f389 100644
--- a/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out
@@ -1139,7 +1139,7 @@ EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
struct<plan:string>
-- !query output
== Parsed Logical Plan ==
-'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [], false, false, false, false
+'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false
+- 'Project [*]
+- 'UnresolvedRelation [explain_temp4], [], false
diff --git a/sql/core/src/test/resources/sql-tests/results/explain.sql.out b/sql/core/src/test/resources/sql-tests/results/explain.sql.out
index c0dee38e6d07..9d25b829e03f 100644
--- a/sql/core/src/test/resources/sql-tests/results/explain.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/explain.sql.out
@@ -1031,7 +1031,7 @@ EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
struct<plan:string>
-- !query output
== Parsed Logical Plan ==
-'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [], false, false, false, false
+'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false
+- 'Project [*]
+- 'UnresolvedRelation [explain_temp4], [], false
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 4a627727b1ed..a63eaddc2206 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -3703,6 +3703,73 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ test("SPARK-49246: read-only catalog") {
+ def assertPrivilegeError(f: => Unit, privilege: String): Unit = {
+ val e = intercept[RuntimeException](f)
+ assert(e.getMessage.contains(privilege))
+ }
+
+ def checkWriteOperations(catalog: String): Unit = {
+ withSQLConf(s"spark.sql.catalog.$catalog" -> classOf[ReadOnlyCatalog].getName) {
+ val input = sql("SELECT 1")
+ val tbl = s"$catalog.default.t"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (i INT)")
+ val df = sql(s"SELECT * FROM $tbl")
+ assert(df.collect().isEmpty)
+ assert(df.schema == new StructType().add("i", "int"))
+
+ assertPrivilegeError(sql(s"INSERT INTO $tbl SELECT 1"), "INSERT")
+ assertPrivilegeError(
+ sql(s"INSERT INTO $tbl REPLACE WHERE i = 0 SELECT 1"), "DELETE,INSERT")
+ assertPrivilegeError(sql(s"INSERT OVERWRITE $tbl SELECT 1"), "DELETE,INSERT")
+ assertPrivilegeError(sql(s"DELETE FROM $tbl WHERE i = 0"), "DELETE")
+ assertPrivilegeError(sql(s"UPDATE $tbl SET i = 0"), "UPDATE")
+ assertPrivilegeError(
+ sql(s"""
+ |MERGE INTO $tbl USING (SELECT 1 i) AS source
+ |ON source.i = $tbl.i
+ |WHEN MATCHED THEN UPDATE SET *
+ |WHEN NOT MATCHED THEN INSERT *
+ |WHEN NOT MATCHED BY SOURCE THEN DELETE
+ |""".stripMargin),
+ "DELETE,INSERT,UPDATE"
+ )
+
+ assertPrivilegeError(input.write.insertInto(tbl), "INSERT")
+ assertPrivilegeError(input.write.mode("overwrite").insertInto(tbl), "DELETE,INSERT")
+ assertPrivilegeError(input.write.mode("append").saveAsTable(tbl), "INSERT")
+ assertPrivilegeError(input.write.mode("overwrite").saveAsTable(tbl), "DELETE,INSERT")
+ assertPrivilegeError(input.writeTo(tbl).append(), "INSERT")
+ assertPrivilegeError(input.writeTo(tbl).overwrite(df.col("i") === 1), "DELETE,INSERT")
+ assertPrivilegeError(input.writeTo(tbl).overwritePartitions(), "DELETE,INSERT")
+ }
+
+ // Test CTAS
+ withTable(tbl) {
+ assertPrivilegeError(sql(s"CREATE TABLE $tbl AS SELECT 1 i"), "INSERT")
+ }
+ withTable(tbl) {
+ assertPrivilegeError(sql(s"CREATE OR REPLACE TABLE $tbl AS SELECT 1 i"), "INSERT")
+ }
+ withTable(tbl) {
+ assertPrivilegeError(input.write.saveAsTable(tbl), "INSERT")
+ }
+ withTable(tbl) {
+ assertPrivilegeError(input.writeTo(tbl).create(), "INSERT")
+ }
+ withTable(tbl) {
+ assertPrivilegeError(input.writeTo(tbl).createOrReplace(), "INSERT")
+ }
+ }
+ }
+ // Reset CatalogManager to clear the materialized `spark_catalog` instance, so that we can
+ // configure a new implementation.
+ spark.sessionState.catalogManager.reset()
+ checkWriteOperations(SESSION_CATALOG_NAME)
+ checkWriteOperations("read_only_cat")
+ }
+
private def testNotSupportedV2Command(
sqlCommand: String,
sqlParams: String,
@@ -3771,3 +3838,20 @@ class V2CatalogSupportBuiltinDataSource extends InMemoryCatalog {
}
}
+class ReadOnlyCatalog extends InMemoryCatalog {
+ override def createTable(
+ ident: Identifier,
+ columns: Array[ColumnV2],
+ partitions: Array[Transform],
+ properties: util.Map[String, String]): Table = {
+ super.createTable(ident, columns, partitions, properties)
+ null
+ }
+
+ override def loadTable(
+ ident: Identifier,
+ writePrivileges: util.Set[TableWritePrivilege]): Table = {
+ throw new RuntimeException("cannot write with " +
+ writePrivileges.asScala.toSeq.map(_.toString).sorted.mkString(","))
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala
index ebb719a35a8b..75837c59945f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SupportsRowLevelOperations, TableCapability, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SupportsRowLevelOperations, TableCapability, TableCatalog, TableWritePrivilege}
import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources.v2.V2SessionCatalog
@@ -161,6 +161,8 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest {
case name => throw new NoSuchTableException(Seq(name))
}
})
+ when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]()))
+ .thenCallRealMethod()
when(newCatalog.name()).thenReturn("cat")
newCatalog
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
index 13735658e6fd..73bcde1e6e5b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{AlterColumn, AnalysisOnlyCom
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
import org.apache.spark.sql.connector.FakeV2Provider
-import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SupportsDelete, Table, TableCapability, TableCatalog, V1Table}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SupportsDelete, Table, TableCapability, TableCatalog, TableWritePrivilege, V1Table}
import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform}
import org.apache.spark.sql.errors.QueryExecutionErrors
@@ -158,6 +158,8 @@ class PlanResolutionSuite extends AnalysisTest {
case name => throw new NoSuchTableException(Seq(name))
}
})
+ when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]()))
+ .thenCallRealMethod()
when(newCatalog.name()).thenReturn("testcat")
newCatalog
}
@@ -175,6 +177,8 @@ class PlanResolutionSuite extends AnalysisTest {
case name => throw new NoSuchTableException(Seq(name))
}
})
+ when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]()))
+ .thenCallRealMethod()
when(newCatalog.name()).thenReturn(CatalogManager.SESSION_CATALOG_NAME)
newCatalog
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org