You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/07/28 17:33:50 UTC
(spark) branch branch-3.5 updated: [SPARK-58370][3.5][SQL] Check table write privileges when the write target is also read in the same statement
This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new a025e49a98a5 [SPARK-58370][3.5][SQL] Check table write privileges when the write target is also read in the same statement
a025e49a98a5 is described below
commit a025e49a98a5ed957a329186b636c03fdcd9ae43
Author: Peter Toth <pe...@gmail.com>
AuthorDate: Tue Jul 28 19:33:39 2026 +0200
[SPARK-58370][3.5][SQL] Check table write privileges when the write target is also read in the same statement
### What changes were proposed in this pull request?
Backport of #57563 to `branch-3.5`.
`ResolveRelations.resolveRelation` in `Analyzer.scala` answers a hit in the per-query `AnalysisContext.relationCache` before reaching the branch that calls `CatalogV2Util.loadTable(catalog, ident, timeTravelSpec, writePrivileges)`. This PR stops serving a reference that carries `REQUIRED_WRITE_PRIVILEGES` from that cache, so a write target always goes through `TableCatalog.loadTable(ident, writePrivileges)`:
```scala
val writePrivilegesString =
Option(u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES))
val cached = if (writePrivilegesString.isEmpty) {
AnalysisContext.get.relationCache.get(key)
} else {
None
}
cached.map { cache => ... }.orElse { ... }
```
Loaded relations are still *stored* in the cache, so a self-MERGE's source keeps sharing the target's `Table` instance (one snapshot).
### Backport notes
The cherry-pick of `a1c3f310a8c` does not apply cleanly, so the change was re-applied by hand:
- `branch-3.5` has no `RelationResolution.scala` at all - relation resolution still lives in the `ResolveRelations` rule in `Analyzer.scala`, so that is where the fix goes. The cache-then-load sequence, and therefore the bug, is the same.
- The local is `writePrivilegesString: Option[String]` here, not a raw `String`, so the guard reads `if (writePrivilegesString.isEmpty)` rather than `if (writePrivileges == null)`.
- The comment block is wrapped for this file's indentation.
The test hunk applies unchanged.
### Why are the changes needed?
`TableCatalog.loadTable(Identifier, Set<TableWritePrivilege>)` (`since 3.5.3`) is where a catalog authorizes a write; the only other caller in the tree is `ResolveSchemaEvolution`. A write target hits the relation cache whenever another reference to the same table was resolved earlier in the same analyzer run, which is exactly the case for `InsertIntoStatement` and `V2WriteCommand` - their target is resolved at the command node *after* its query. So the catalog is never asked to autho [...]
```sql
INSERT INTO t SELECT * FROM t
INSERT OVERWRITE t SELECT * FROM t WHERE 1 = 0 -- empties the table
INSERT INTO t REPLACE WHERE i = 0 SELECT * FROM t
INSERT INTO t SELECT i FROM (SELECT i FROM t) x
WITH c AS (SELECT i FROM t) INSERT INTO t SELECT i FROM c
```
```scala
spark.table("t").writeTo("t").append() // DataFrameWriterV2 passes the unanalyzed plan
```
`INSERT INTO t SELECT 1` on the same table is authorized normally, and `UPDATE` / `DELETE` / `MERGE` are unaffected (their target is a plain child, resolved before the source, so it is a cache miss), as is `DataFrameWriter.insertInto` (it passes the already-analyzed plan).
### Does this PR introduce _any_ user-facing change?
Yes. On a catalog that enforces privileges in `loadTable(ident, writePrivileges)`, the statements above are now authorized like every other write, so a user without the required privilege gets the catalog's error instead of silently writing. Catalogs that ignore the privileges argument (including Spark's built-in `V2SessionCatalog`) are unaffected. The write target of a self-referencing INSERT is now loaded once more, which is what a non-self-referencing INSERT already does.
### How was this patch tested?
New test in `DataSourceV2SQLSuite`, identical to the one on `master`. It fails on `branch-3.5` without the fix ("Expected exception java.lang.RuntimeException to be thrown, but no exception was thrown") and passes with it.
Also green on this branch: 15 suites / 794 tests - `DataSourceV2SQLSuite` (V1+V2 filter), `DataSourceV2DataFrameSuite`, `PlanResolutionSuite`, `CachedTableSuite`, `*ViewTestSuite`, group/delta-based MERGE/UPDATE/DELETE suites, `InsertSuite`. `DataSourceV2OptionSuite` and `MergeIntoDataFrameSuite` do not exist on this branch. `dev/lint-scala` clean. Built and tested on JDK 17.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
Closes #57601 from peter-toth/SPARK-58370-write-privileges-self-reference-3.5.
Authored-by: Peter Toth <pe...@gmail.com>
Signed-off-by: Peter Toth <pe...@gmail.com>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 18 +++++--
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 59 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 3 deletions(-)
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 a7ebde57b863..a4d9455de4a8 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
@@ -1300,7 +1300,21 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
expandIdentifier(u.multipartIdentifier) match {
case CatalogAndIdentifier(catalog, ident) =>
val key = ((catalog.name +: ident.namespace :+ ident.name).toSeq, timeTravelSpec)
- AnalysisContext.get.relationCache.get(key).map { cache =>
+ val writePrivilegesString =
+ Option(u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES))
+ // A reference that requires write privileges is never served from the per-query
+ // relation cache. The catalog authorizes the write in
+ // `loadTable(ident, writePrivileges)` below, and a cache hit would skip that call
+ // entirely. The hit happens whenever the write target is also read in the same
+ // statement -- the target is resolved after its query (see `ResolveRelations`), so it
+ // finds the relation the query already put in the cache, e.g. for
+ // `INSERT INTO t SELECT * FROM t`.
+ val cached = if (writePrivilegesString.isEmpty) {
+ AnalysisContext.get.relationCache.get(key)
+ } else {
+ None
+ }
+ cached.map { cache =>
val cachedRelation = cache.transform {
case multi: MultiInstanceRelation =>
val newRelation = multi.newInstance()
@@ -1313,8 +1327,6 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
cachedConnectRelation
}.getOrElse(cachedRelation)
}.orElse {
- val writePrivilegesString =
- Option(u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES))
val table = CatalogV2Util.loadTable(
catalog, ident, timeTravelSpec, writePrivilegesString)
val loaded = createRelation(
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 9df4b0932f25..c5add84c3894 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
@@ -3517,6 +3517,65 @@ class DataSourceV2SQLSuiteV1Filter
checkWriteOperations("read_only_cat")
}
+ test("SPARK-58370: read-only catalog, write whose query reads the target table") {
+ def assertPrivilegeError(f: => Unit, privileges: String): Unit = {
+ val e = intercept[RuntimeException](f)
+ assert(e.getMessage == s"cannot write with $privileges")
+ }
+
+ def checkWritesReadingTheTarget(catalog: String): Unit = {
+ withSQLConf(s"spark.sql.catalog.$catalog" -> classOf[ReadOnlyCatalog].getName) {
+ val tbl = s"$catalog.default.t"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (i INT)")
+
+ // The write target is resolved after its query, so it used to find the query's
+ // relation in the per-query relation cache and skip `loadTable(ident, writePrivileges)`.
+ assertPrivilegeError(sql(s"INSERT INTO $tbl SELECT * FROM $tbl"), "INSERT")
+ assertPrivilegeError(
+ sql(s"INSERT OVERWRITE $tbl SELECT * FROM $tbl"), "DELETE,INSERT")
+ assertPrivilegeError(
+ sql(s"INSERT OVERWRITE $tbl SELECT * FROM $tbl WHERE 1 = 0"), "DELETE,INSERT")
+ assertPrivilegeError(
+ sql(s"INSERT INTO $tbl REPLACE WHERE i = 0 SELECT * FROM $tbl"), "DELETE,INSERT")
+ // The reference to the target can be anywhere in the query.
+ assertPrivilegeError(
+ sql(s"INSERT INTO $tbl SELECT i FROM (SELECT i FROM $tbl) x"), "INSERT")
+ assertPrivilegeError(
+ sql(s"WITH c AS (SELECT i FROM $tbl) INSERT INTO $tbl SELECT i FROM c"), "INSERT")
+
+ // `DataFrameWriterV2` passes the unanalyzed query plan, so its target is resolved in the
+ // same analyzer run as the query's reference to the same table.
+ assertPrivilegeError(sql(s"SELECT * FROM $tbl").writeTo(tbl).append(), "INSERT")
+ assertPrivilegeError(
+ sql(s"SELECT * FROM $tbl").writeTo(tbl).overwritePartitions(), "DELETE,INSERT")
+ assertPrivilegeError(sql(s"SELECT * FROM $tbl").write.insertInto(tbl), "INSERT")
+
+ // Row-level operations resolve the target before the source / subquery, so these were
+ // already checked; they are here to keep both orders covered.
+ assertPrivilegeError(
+ sql(s"UPDATE $tbl SET i = 0 WHERE i IN (SELECT i FROM $tbl)"), "UPDATE")
+ assertPrivilegeError(
+ sql(s"DELETE FROM $tbl WHERE i IN (SELECT i FROM $tbl)"), "DELETE")
+ assertPrivilegeError(
+ sql(
+ s"""
+ |MERGE INTO $tbl USING (SELECT i FROM $tbl) AS source
+ |ON source.i = $tbl.i
+ |WHEN MATCHED THEN UPDATE SET *
+ |""".stripMargin),
+ "UPDATE")
+ }
+ }
+ }
+
+ // Reset CatalogManager to clear the materialized `spark_catalog` instance, so that we can
+ // configure a new implementation.
+ spark.sessionState.catalogManager.reset()
+ checkWritesReadingTheTarget(SESSION_CATALOG_NAME)
+ checkWritesReadingTheTarget("read_only_self_ref_cat")
+ }
+
test("StagingTableCatalog without atomic support") {
withSQLConf("spark.sql.catalog.fakeStagedCat" -> classOf[FakeStagedTableCatalog].getName) {
withTable("fakeStagedCat.t") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org