This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new ac389e00e2 [MINOR] fix(spark): enforce PostgreSQL write privileges on
branch-1.3 (#14529)
ac389e00e2 is described below
commit ac389e00e210cc1294959e3af179ad2f2e53318d
Author: Qi Yu <[email protected]>
AuthorDate: Mon Nov 2 19:37:21 2026 +0800
[MINOR] fix(spark): enforce PostgreSQL write privileges on branch-1.3
(#14529)
### What changes were proposed in this pull request?
Override PostgreSQL Spark 3.5's write-aware loadTable on branch-1.3 and
delegate to the existing shared loadTableForWriting authorization path.
Add direct write-privilege cases and real Spark SQL INSERT/INSERT
OVERWRITE planning regressions.
### Why are the changes needed?
Spark's default write-aware overload falls back to the read load.
Require MODIFY_TABLE before physical table access for writes through
this connector.
Fixes #14525
Main already has the common guard from #12414; this PR targets
branch-1.3 only.
### Does this PR introduce _any_ user-facing change?
Normal PostgreSQL Spark 3.5 writes on branch-1.3 now require
MODIFY_TABLE and fail before accessing the physical table when denied.
This preserves the existing trusted-connector authorization model; it
does not add a new server security boundary.
### How was this patch tested?
- Entire Spark 3.5 module test task: 7 cases across 2 classes, no
failures.
- New cases cover all TableWritePrivilege enum values, denial before
delegate access, and actual INSERT/INSERT OVERWRITE SQL planning with
controlled metadata and physical-table boundaries.
- Reverting the override fails both new SQL authorization regressions;
restored code passes.
- Exact-head independent cumulative review. Deployed REST + PostgreSQL +
released-artifact end-to-end verification remains separate.
---
spark-connector/v3.5/spark/build.gradle.kts | 1 +
.../GravitinoPostgreSqlCatalogSpark35.java | 12 ++
.../TestPostgreSqlWriteAuthorization13.java | 221 +++++++++++++++++++++
3 files changed, 234 insertions(+)
diff --git a/spark-connector/v3.5/spark/build.gradle.kts
b/spark-connector/v3.5/spark/build.gradle.kts
index d59088d5bf..6393340ec7 100644
--- a/spark-connector/v3.5/spark/build.gradle.kts
+++ b/spark-connector/v3.5/spark/build.gradle.kts
@@ -49,6 +49,7 @@ if (hasProperty("excludePackagesForSparkConnector")) {
}
dependencies {
+ testImplementation(libs.mockito.core)
implementation(project(":spark-connector:spark-3.4"))
implementation(project(":spark-connector:spark-common"))
compileOnly("org.apache.kyuubi:kyuubi-spark-connector-hive_$scalaVersion:$kyuubiVersion")
diff --git
a/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/postgresql/GravitinoPostgreSqlCatalogSpark35.java
b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/postgresql/GravitinoPostgreSqlCatalogSpark35.java
index 709013786c..58f675d1c9 100644
---
a/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/postgresql/GravitinoPostgreSqlCatalogSpark35.java
+++
b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/postgresql/GravitinoPostgreSqlCatalogSpark35.java
@@ -18,11 +18,23 @@
*/
package org.apache.gravitino.spark.connector.jdbc.postgresql;
+import java.util.Set;
import org.apache.gravitino.spark.connector.PropertiesConverter;
import org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalog;
import org.apache.gravitino.spark.connector.jdbc.JdbcPropertiesConverter;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege;
public class GravitinoPostgreSqlCatalogSpark35 extends GravitinoJdbcCatalog {
+ /** {@inheritDoc} */
+ @Override
+ public Table loadTable(Identifier ident, Set<TableWritePrivilege>
writePrivileges)
+ throws NoSuchTableException {
+ return loadTableForWriting(ident);
+ }
+
@Override
protected PropertiesConverter getPropertiesConverter() {
return JdbcPropertiesConverter.getPGInstance();
diff --git
a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/jdbc/postgresql/TestPostgreSqlWriteAuthorization13.java
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/jdbc/postgresql/TestPostgreSqlWriteAuthorization13.java
new file mode 100644
index 0000000000..037884f1d1
--- /dev/null
+++
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/jdbc/postgresql/TestPostgreSqlWriteAuthorization13.java
@@ -0,0 +1,221 @@
+/*
+ * 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.gravitino.spark.connector.jdbc.postgresql;
+
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.EnumSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.authorization.Privilege;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.exceptions.ForbiddenException;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.gravitino.spark.connector.PropertiesConverter;
+import org.apache.gravitino.spark.connector.SparkTransformConverter;
+import org.apache.gravitino.spark.connector.SparkTypeConverter;
+import org.apache.gravitino.spark.connector.catalog.GravitinoCatalogManager;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.SupportsWrite;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/** Release-branch PostgreSQL writes must use the MODIFY_TABLE metadata load.
*/
+public class TestPostgreSqlWriteAuthorization13 {
+ private static Catalog sqlClient;
+ private static org.apache.spark.sql.connector.catalog.TableCatalog
sqlDelegate;
+ private Catalog client;
+ private TableCatalog metadata;
+ private org.apache.spark.sql.connector.catalog.TableCatalog spark;
+ private TestCatalog catalog;
+ private final Identifier identifier = Identifier.of(new String[] {"schema"},
"table");
+ private final NameIdentifier metadataIdentifier =
NameIdentifier.of("schema", "table");
+
+ @BeforeAll
+ static void initializeManager() {
+ GravitinoCatalogManager.create("test_metalake", () ->
mock(GravitinoClient.class));
+ }
+
+ @AfterAll
+ static void closeManager() {
+ GravitinoCatalogManager.get().close();
+ }
+
+ @BeforeEach
+ void setUp() {
+ client = mock(Catalog.class);
+ metadata = mock(TableCatalog.class);
+ spark = mock(org.apache.spark.sql.connector.catalog.TableCatalog.class);
+ when(client.asTableCatalog()).thenReturn(metadata);
+ catalog = new TestCatalog(client, spark);
+ }
+
+ @ParameterizedTest
+ @EnumSource(TableWritePrivilege.class)
+ void testWriteLoadsRequireModifyPrivilege(TableWritePrivilege privilege)
throws Exception {
+ org.apache.gravitino.rel.Table gravitinoTable =
mock(org.apache.gravitino.rel.Table.class);
+ Table physical = mock(Table.class);
+ when(metadata.loadTable(metadataIdentifier,
Set.of(Privilege.Name.MODIFY_TABLE)))
+ .thenReturn(gravitinoTable);
+ when(spark.loadTable(identifier)).thenReturn(physical);
+ Assertions.assertSame(physical, catalog.loadTable(identifier,
Set.of(privilege)));
+ verify(metadata).loadTable(metadataIdentifier,
Set.of(Privilege.Name.MODIFY_TABLE));
+ verify(metadata, never()).loadTable(metadataIdentifier);
+ }
+
+ @Test
+ void testDeniedWriteDoesNotLoadThePhysicalTable() {
+ when(metadata.loadTable(metadataIdentifier,
Set.of(Privilege.Name.MODIFY_TABLE)))
+ .thenThrow(new ForbiddenException("MODIFY_TABLE is required"));
+ Assertions.assertThrows(
+ ForbiddenException.class,
+ () -> catalog.loadTable(identifier,
Set.of(TableWritePrivilege.INSERT)));
+ verifyNoInteractions(spark);
+ }
+
+ @SuppressWarnings("deprecation")
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "INSERT INTO guarded.schema.table VALUES (1)",
+ "INSERT OVERWRITE guarded.schema.table VALUES (2)"
+ })
+ void testActualSparkSqlRequestsWriteAuthorization(String query) throws
Exception {
+ SupportsWrite physical = mock(SupportsWrite.class);
+ when(physical.name()).thenReturn("schema.table");
+ when(physical.schema()).thenReturn(new StructType().add("id",
DataTypes.IntegerType));
+ when(physical.partitioning()).thenReturn(new Transform[0]);
+ when(physical.properties()).thenReturn(Map.of());
+ when(physical.capabilities())
+ .thenReturn(
+ EnumSet.of(
+ TableCapability.BATCH_WRITE,
+ TableCapability.OVERWRITE_BY_FILTER,
+ TableCapability.TRUNCATE));
+ when(physical.newWriteBuilder(any()))
+ .thenThrow(new AssertionError("Write reached physical table"));
+ when(spark.loadTable(identifier)).thenReturn(physical);
+ when(metadata.loadTable(metadataIdentifier))
+ .thenReturn(mock(org.apache.gravitino.rel.Table.class));
+ when(metadata.loadTable(metadataIdentifier,
Set.of(Privilege.Name.MODIFY_TABLE)))
+ .thenThrow(new ForbiddenException("MODIFY_TABLE is required"));
+ sqlClient = client;
+ sqlDelegate = spark;
+ try (SparkSession session =
+ SparkSession.builder()
+ .master("local[1]")
+ .appName("PostgreSQL write authorization")
+ .config("spark.ui.enabled", "false")
+ .config("spark.driver.host", "127.0.0.1")
+ .config("spark.driver.bindAddress", "127.0.0.1")
+ .config("spark.sql.catalog.guarded", SqlCatalog.class.getName())
+ .getOrCreate()) {
+ Exception failure =
+ Assertions.assertThrows(Exception.class, () ->
session.sql(query).collect());
+ Throwable cause = failure;
+ while (!(cause instanceof ForbiddenException) && cause.getCause() !=
null) {
+ cause = cause.getCause();
+ }
+ Assertions.assertTrue(cause instanceof ForbiddenException,
failure.toString());
+ verify(metadata, atLeastOnce())
+ .loadTable(metadataIdentifier, Set.of(Privilege.Name.MODIFY_TABLE));
+ verify(physical, never()).newWriteBuilder(any());
+ }
+ }
+
+ /** A real Spark catalog plugin using controlled metadata and physical-table
boundaries. */
+ public static class SqlCatalog extends GravitinoPostgreSqlCatalogSpark35 {
+ private String catalogName;
+
+ /** {@inheritDoc} */
+ @Override
+ public void initialize(String name, CaseInsensitiveStringMap options) {
+ catalogName = name;
+ gravitinoCatalogClient = sqlClient;
+ sparkCatalog = sqlDelegate;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String name() {
+ return catalogName;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String[] defaultNamespace() {
+ return new String[] {"schema"};
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected Table createSparkTable(
+ Identifier identifier,
+ org.apache.gravitino.rel.Table gravitinoTable,
+ Table sparkTable,
+ org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog,
+ PropertiesConverter propertiesConverter,
+ SparkTransformConverter sparkTransformConverter,
+ SparkTypeConverter sparkTypeConverter) {
+ return sparkTable;
+ }
+ }
+
+ private static class TestCatalog extends GravitinoPostgreSqlCatalogSpark35 {
+ private TestCatalog(
+ Catalog client, org.apache.spark.sql.connector.catalog.TableCatalog
delegate) {
+ this.gravitinoCatalogClient = client;
+ this.sparkCatalog = delegate;
+ }
+
+ @Override
+ protected Table createSparkTable(
+ Identifier identifier,
+ org.apache.gravitino.rel.Table gravitinoTable,
+ Table sparkTable,
+ org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog,
+ PropertiesConverter propertiesConverter,
+ SparkTransformConverter sparkTransformConverter,
+ SparkTypeConverter sparkTypeConverter) {
+ return sparkTable;
+ }
+ }
+}
