[SPARK-58370][SQL] Check table write privileges when the write target is also read in the same statement

[peter-toth]

What changes were proposed in this pull request?

RelationResolution.tryResolvePersistent answers a hit in the per-query AnalysisContext.relationCache before reaching the branch that calls CatalogV2Util.loadTable(catalog, ident, timeTravelSpec, Option(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):

val writePrivileges = u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
val cached = if (writePrivileges == null) relationCache.get(key) else None
cached.map(adaptCachedRelation(_, planId)).orElse { ... }

Loaded relations are still stored in the cache, so a self-MERGE's source keeps sharing the target's Table instance (one snapshot). V2TableReference already treats write targets this way (V2TableReference.WriteTargetContext.cacheable = false).

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 authorize these statements:

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

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 next to "SPARK-49246: read-only catalog" (whose write cases all use a constant source, which is why this gap was never covered), asserting the ReadOnlyCatalog privilege error for every statement above plus the UPDATE / DELETE / MERGE self-reference cases, for both a custom v2 catalog and a ReadOnlyCatalog-backed spark_catalog. It fails on master ("Expected exception java.lang.RuntimeException to be thrown, but no exception was thrown") and passes with the fix.

Also green: 17 suites / 1506 tests - DataSourceV2SQLSuite (V1+V2 filter), DataSourceV2DataFrameSuite, DataSourceV2OptionSuite, PlanResolutionSuite, CachedTableSuite, *ViewTestSuite, group/delta-based MERGE/UPDATE/DELETE suites, InsertSuite. dev/lint-scala clean.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5)

[uros-b]

Thank you @peter-toth!

[dongjoon-hyun]

+1, LGTM.