This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 17f7000ad9 [#13719] test(iceberg): Expand SQL read restriction
integration coverage (#14331)
17f7000ad9 is described below
commit 17f7000ad9980d3ef71730d0d68710f83e132466
Author: roryqi <[email protected]>
AuthorDate: Thu Nov 5 15:00:25 2026 +0800
[#13719] test(iceberg): Expand SQL read restriction integration coverage
(#14331)
### What changes were proposed in this pull request?
Expand SQL read-restriction coverage to 32 integration cases across 25
test methods: 23 ordinary methods and two parameterized methods with 7
and 2 cases. Cover users/groups, policy/tag lifecycle and inheritance,
conflicts, conditional masks, and SQL NULL/CASE semantics. Include
response-field absence and tag-deletion coverage; revoke user roles
before each case. Other tags and policies remain in the suite's shared
metalake until teardown.
Keep both experimental Spark IT suites in
`iceberg-rest-server-experiment`, using a dedicated `experimentalTest`
source set with explicit dependencies. Reuse the server's complete
authorization test harness and resources without exporting its
dependency closure. Ordinary unit tests retain the production Jackson
pins and do not load Spark.
Implement the group-authentication fixture through the public
`Authenticator` interface in its own package, with a small dedicated JAR
for standalone deploy tests. Retain the existing CI task as a forwarding
entry point.
### Why are the changes needed?
Verify the SQL contract introduced in #14237 through REST responses and
Spark reads. Fix deploy-mode startup, where the standalone server could
not load the test authenticator, and isolate experimental Spark
dependencies from ordinary unit tests.
Related to #13719.
### Does this PR introduce _any_ user-facing change?
No. Test coverage and infrastructure only.
### How was this patch tested?
- 48 experimental integration cases passed in embedded mode: 32
read-restriction cases and 16 authorization cases.
- 48 experimental integration cases passed in deploy/PostgreSQL mode: 32
read-restriction cases and 16 authorization cases, through the existing
CI entry point with a freshly rebuilt distribution.
- Earlier deploy/MySQL validation at `386b6c7` passed all 46 cases,
including both authorization and SQL read-restriction suites.
- 68 ordinary unit tests passed: 54 converter/loader/predicate-result
cases and 14 authenticator cases.
- Verified that production and ordinary test compile/runtime classpaths
contain no Spark and resolve Jackson databind 2.15.2.
- Module Spotless and `git diff --check` passed.
```bash
./gradlew :iceberg:iceberg-rest-server-experiment:experimentalTest \
-PskipTests -PskipDockerTests=false
./gradlew :iceberg:iceberg-rest-server:experimentalTest \
-PskipTests -PtestMode=deploy -PjdbcBackend=postgresql
-PskipDockerTests=false
./gradlew :iceberg:iceberg-rest-server-experiment:test -PskipITs
```
---
.../workflows/backend-integration-test-action.yml | 1 +
gradle/libs.versions.toml | 1 +
.../build.gradle.kts | 132 ++-
.../test/IcebergExperimentalAuthorizationIT.java | 0
.../integration/test/IcebergReadRestrictionIT.java | 931 +++++++++++++++++++++
.../TestReadRestrictionGroupAuthenticator.java | 90 ++
.../auth/ReadRestrictionGroupAuthenticator.java | 88 ++
iceberg/iceberg-rest-server/build.gradle.kts | 122 +--
.../integration/test/IcebergAuthorizationIT.java | 7 +-
.../integration/test/IcebergReadRestrictionIT.java | 132 ---
10 files changed, 1267 insertions(+), 237 deletions(-)
diff --git a/.github/workflows/backend-integration-test-action.yml
b/.github/workflows/backend-integration-test-action.yml
index 54c635fa58..dd9c8018fd 100644
--- a/.github/workflows/backend-integration-test-action.yml
+++ b/.github/workflows/backend-integration-test-action.yml
@@ -126,6 +126,7 @@ jobs:
path: |
build/reports
iceberg/iceberg-rest-server/build/*.log
+ iceberg/iceberg-rest-server-experiment/build/*.log
distribution/package/logs/*.out
distribution/package/logs/*.log
catalogs/**/*.log
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index d274a17618..1a3f3c6914 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -40,6 +40,7 @@ lance-namespace = "0.7.5"
delta-kernel = "3.3.0"
jetty = "9.4.58.v20250814"
jersey = "2.41"
+hk2 = "2.6.1"
mockito = "4.11.0"
airlift-json = "237"
airlift-log = "237"
diff --git a/iceberg/iceberg-rest-server-experiment/build.gradle.kts
b/iceberg/iceberg-rest-server-experiment/build.gradle.kts
index 49937bfb29..7303921cb4 100644
--- a/iceberg/iceberg-rest-server-experiment/build.gradle.kts
+++ b/iceberg/iceberg-rest-server-experiment/build.gradle.kts
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
description = "iceberg-rest-server-experiment"
plugins {
@@ -27,11 +28,19 @@ plugins {
val datastratoIcebergVersion: String = libs.versions.datastrato.iceberg.get()
val caffeineVersion: String = libs.versions.caffeine.get()
val jacksonVersion: String = libs.versions.jackson.get()
+val experimentalScalaVersion = "2.13"
+val experimentalSparkVersion: String = libs.versions.spark41.get()
+val experimentalSparkMajorVersion: String =
experimentalSparkVersion.substringBeforeLast(".")
+val datastratoIcebergGitCommit: String =
libs.versions.experimental.iceberg.git.commit.get()
+val jerseyVersion: String = libs.versions.jersey.get()
+val javaxWsRsVersion: String = libs.versions.javax.ws.rs.api.get()
+val hk2Version: String = libs.versions.hk2.get()
+val jakartaValidationVersion: String = libs.versions.jakarta.validation.get()
configurations.configureEach {
resolutionStrategy {
+ force("com.github.ben-manes.caffeine:caffeine:$caffeineVersion")
force(
- "com.github.ben-manes.caffeine:caffeine:$caffeineVersion",
"com.fasterxml.jackson.core:jackson-annotations:$jacksonVersion",
"com.fasterxml.jackson.core:jackson-core:$jacksonVersion",
"com.fasterxml.jackson.core:jackson-databind:$jacksonVersion",
@@ -44,12 +53,58 @@ configurations.configureEach {
(requested.group == "org.apache.iceberg" || requested.group ==
"com.datastrato")
) {
useTarget("com.datastrato:experimental-${requested.name}:$datastratoIcebergVersion")
- because("Use the Datastrato Iceberg distribution in the experimental
auxiliary service")
+ because("Run experimental Iceberg integration tests with the
Datastrato distribution")
}
}
}
}
+// Keep the Spark runtime isolated from the ordinary unit tests and production
classpaths.
+val testFixturesSources = sourceSets.create("testFixtures") {
+ compileClasspath += sourceSets.main.get().output
+ runtimeClasspath += sourceSets.main.get().output
+}
+val experimentalTestSources = sourceSets.create("experimentalTest") {
+ compileClasspath += sourceSets.main.get().output
+ runtimeClasspath += sourceSets.main.get().output
+}
+configurations.named(testFixturesSources.implementationConfigurationName) {
+ extendsFrom(configurations.implementation.get())
+}
+configurations.named(experimentalTestSources.implementationConfigurationName) {
+ extendsFrom(configurations.implementation.get())
+}
+listOf(
+ experimentalTestSources.compileClasspathConfigurationName,
+ experimentalTestSources.runtimeClasspathConfigurationName
+).forEach { configurationName ->
+ configurations.named(configurationName) {
+ resolutionStrategy {
+ // The configureEach block above must apply the default Jackson pins
before this override.
+ // Only Spark ITs use Spark's matching Jackson modules.
+ setForcedModules(
+ *forcedModules.filterNot {
it.group.startsWith("com.fasterxml.jackson.") }.toTypedArray()
+ )
+ force(
+ "javax.ws.rs:javax.ws.rs-api:$javaxWsRsVersion",
+ "org.glassfish.jersey.core:jersey-server:$jerseyVersion",
+ "org.glassfish.jersey.core:jersey-common:$jerseyVersion",
+ "org.glassfish.jersey.core:jersey-client:$jerseyVersion",
+
"org.glassfish.jersey.containers:jersey-container-servlet:$jerseyVersion",
+
"org.glassfish.jersey.containers:jersey-container-servlet-core:$jerseyVersion",
+
"org.glassfish.jersey.containers:jersey-container-jetty-http:$jerseyVersion",
+ "org.glassfish.jersey.inject:jersey-hk2:$jerseyVersion",
+ "org.glassfish.hk2:hk2-api:$hk2Version",
+ "org.glassfish.hk2:hk2-locator:$hk2Version",
+ "org.glassfish.hk2:hk2-utils:$hk2Version",
+ "org.glassfish.hk2.external:aopalliance-repackaged:$hk2Version",
+ "org.glassfish.hk2.external:jakarta.inject:$hk2Version",
+ "jakarta.validation:jakarta.validation-api:$jakartaValidationVersion"
+ )
+ }
+ }
+}
+
dependencies {
compileOnly(libs.jsr305)
implementation(project(":api"))
@@ -65,10 +120,49 @@ dependencies {
implementation(project(":iceberg:iceberg-rest-server"))
implementation(libs.bundles.iceberg)
+ testImplementation(testFixturesSources.output)
+ testImplementation(project(":server-common"))
+ testImplementation(libs.junit.jupiter.params)
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.mockito.core)
testCompileOnly(libs.jsr305)
testRuntimeOnly(libs.junit.jupiter.engine)
+
+ add(testFixturesSources.compileOnlyConfigurationName, libs.jsr305)
+ add(testFixturesSources.implementationConfigurationName,
project(":server-common"))
+ val experimentalImplementation =
experimentalTestSources.implementationConfigurationName
+ // The server uses unshaded Iceberg classes before the Spark runtime's
bundled copies.
+ add(experimentalImplementation, libs.bundles.iceberg)
+ add(experimentalImplementation, testFixturesSources.output)
+ add(experimentalImplementation, project(":iceberg:iceberg-rest-server",
"testArtifacts"))
+ add(experimentalImplementation, project(":integration-test-common",
"testArtifacts"))
+ add(experimentalImplementation, project(":server"))
+ add(experimentalImplementation, project(":server-common"))
+ add(experimentalImplementation, project(":clients:client-java"))
+ add(experimentalImplementation,
"org.apache.spark:spark-sql_$experimentalScalaVersion:$experimentalSparkVersion")
{
+ exclude(group = "org.apache.avro")
+ exclude(group = "org.apache.hadoop")
+ exclude(group = "org.apache.zookeeper")
+ exclude(group = "io.dropwizard.metrics")
+ exclude(group = "org.rocksdb")
+ }
+ add(experimentalImplementation,
"com.datastrato:experimental-iceberg-spark-runtime-${experimentalSparkMajorVersion}_$experimentalScalaVersion:$datastratoIcebergVersion")
+ add(experimentalImplementation,
"org.scala-lang.modules:scala-collection-compat_$experimentalScalaVersion:${libs.versions.scala.collection.compat.get()}")
+ add(experimentalImplementation, libs.hadoop3.client.api)
+ add(experimentalImplementation, libs.hadoop3.client.runtime)
+ add(experimentalImplementation, libs.javax.ws.rs.api)
+ add(experimentalImplementation, libs.awaitility)
+ add(experimentalImplementation, libs.commons.lang3)
+ add(experimentalImplementation, libs.commons.io)
+ add(experimentalImplementation, libs.guava)
+ add(experimentalImplementation, libs.slf4j.api)
+ add(experimentalImplementation, libs.postgresql.driver)
+ add(experimentalImplementation, libs.testcontainers)
+ add(experimentalImplementation, libs.testcontainers.postgresql)
+ add(experimentalImplementation, libs.junit.jupiter.api)
+ add(experimentalImplementation, libs.junit.jupiter.params)
+ add(experimentalTestSources.runtimeOnlyConfigurationName, libs.mysql.driver)
+ add(experimentalTestSources.runtimeOnlyConfigurationName,
libs.junit.jupiter.engine)
}
tasks {
@@ -111,3 +205,37 @@ tasks {
dependsOn(copyLibs, copyConfigs)
}
}
+
+val readRestrictionAuthenticatorJar by tasks.registering(Jar::class) {
+ archiveClassifier.set("read-restriction-test-authenticator")
+ destinationDirectory.set(layout.buildDirectory.dir("test-fixtures"))
+ from(testFixturesSources.output)
+}
+
+val experimentalTest by tasks.registering(Test::class) {
+ description = "Runs Iceberg REST tests that require the experimental Iceberg
runtime"
+ group = "verification"
+ testClassesDirs = experimentalTestSources.output.classesDirs
+ classpath = experimentalTestSources.runtimeClasspath
+ systemProperty("datastratoIcebergGitCommit", datastratoIcebergGitCommit)
+ systemProperty("spark.ui.enabled", "false")
+ // Deploy mode starts a separate server process; expose only the test
authenticator fixture.
+ environment(
+ "GRAVITINO_CLASSPATH",
+ readRestrictionAuthenticatorJar.flatMap { it.archiveFile
}.get().asFile.absolutePath
+ )
+ setIncludes(
+ setOf(
+ "**/IcebergExperimentalAuthorizationIT.class",
+ "**/IcebergReadRestrictionIT.class"
+ )
+ )
+ mustRunAfter(tasks.test, ":iceberg:iceberg-rest-server:test")
+ dependsOn(
+ tasks.named(experimentalTestSources.classesTaskName),
+ tasks.jar,
+ ":catalogs:catalog-lakehouse-iceberg:jar",
+ tasks.named("copyDepends"),
+ readRestrictionAuthenticatorJar
+ )
+}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergExperimentalAuthorizationIT.java
b/iceberg/iceberg-rest-server-experiment/src/experimentalTest/java/org/apache/gravitino/iceberg/integration/test/IcebergExperimentalAuthorizationIT.java
similarity index 100%
rename from
iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergExperimentalAuthorizationIT.java
rename to
iceberg/iceberg-rest-server-experiment/src/experimentalTest/java/org/apache/gravitino/iceberg/integration/test/IcebergExperimentalAuthorizationIT.java
diff --git
a/iceberg/iceberg-rest-server-experiment/src/experimentalTest/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java
b/iceberg/iceberg-rest-server-experiment/src/experimentalTest/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java
new file mode 100644
index 0000000000..f42c6de1e8
--- /dev/null
+++
b/iceberg/iceberg-rest-server-experiment/src/experimentalTest/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java
@@ -0,0 +1,931 @@
+/*
+ * 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.iceberg.integration.test;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Stream;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.authorization.Privilege;
+import org.apache.gravitino.authorization.Privileges;
+import org.apache.gravitino.authorization.SecurableObject;
+import org.apache.gravitino.authorization.SecurableObjects;
+import org.apache.gravitino.client.GravitinoAdminClient;
+import
org.apache.gravitino.iceberg.integration.test.auth.ReadRestrictionGroupAuthenticator;
+import org.apache.gravitino.integration.test.util.TestDatabaseName;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.policy.TagValueSelector;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.types.Types;
+import org.apache.gravitino.tag.TagValue;
+import org.apache.gravitino.tag.TagValueConstraint;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/** Verifies that Spark enforces read restrictions returned by the
experimental Iceberg service. */
+@Tag("gravitino-docker-test")
+public class IcebergReadRestrictionIT extends IcebergAuthorizationIT {
+
+ private static final String EXPERIMENTAL_SERVER_DIRECTORY =
"iceberg-rest-server-experiment";
+ private static final String SCHEMA_NAME = "schema";
+
+ private final HttpClient httpClient = HttpClient.newHttpClient();
+
+ @BeforeAll
+ @Override
+ public void startIntegrationTest() throws Exception {
+ String rootDirectory = System.getenv("GRAVITINO_ROOT_DIR");
+ String experimentalServerHome =
+ Paths.get(rootDirectory, "iceberg",
EXPERIMENTAL_SERVER_DIRECTORY).toString();
+ customConfigs.put(
+ "gravitino.iceberg-rest.classpath",
+ String.join(
+ ",",
+ Paths.get(experimentalServerHome, "build", "libs").toString(),
+ Paths.get(experimentalServerHome, "build",
"dependencies").toString()));
+ super.startIntegrationTest();
+ // Limit retries when table loads are expected to fail.
+ sql("SET `spark.sql.catalog.%s.rest.client.max-retries`=1",
SPARK_CATALOG_NAME);
+
metalakeClientWithAllPrivilege.addGroup(ReadRestrictionGroupAuthenticator.AUDITORS_GROUP);
+ catalogClientWithAllPrivilege.asSchemas().createSchema(SCHEMA_NAME, null,
new HashMap<>());
+ }
+
+ /**
+ * Starts the server and authenticates setup requests as the service
administrator.
+ *
+ * @throws Exception if server startup or client initialization fails
+ */
+ @Override
+ protected void startServer() throws Exception {
+ super.startServer();
+ client.close();
+ client =
GravitinoAdminClient.builder(serverUri).withSimpleAuth(SUPER_USER).build();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected String authenticator() {
+ return ReadRestrictionGroupAuthenticator.class.getCanonicalName();
+ }
+
+ @Override
+ protected TestDatabaseName getTestDatabaseName() {
+ return TestDatabaseName.PG_ICEBERG_READ_RESTRICTION_IT;
+ }
+
+ @BeforeEach
+ void resetUserRoles() {
+ revokeUserRoles();
+ }
+
+ @Test
+ void testSparkEnforcesReadRestrictions() throws Exception {
+ String tableName = "read_restrictions";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+ List<Object[]> unrestrictedRows = sql("SELECT col_1, col_2 FROM %s ORDER
BY col_1", tableName);
+ Assertions.assertEquals(2, unrestrictedRows.size());
+
+ assertNoReadRestrictions(SCHEMA_NAME, tableName);
+
+ String suffix = UUID.randomUUID().toString().replace("-", "");
+ String rowFilterTag = "row_filter_tag_" + suffix;
+ String rowFilterPolicy = "row_filter_policy_" + suffix;
+ metalakeClientWithAllPrivilege.createTag(rowFilterTag, null, new
HashMap<>());
+ metalakeClientWithAllPrivilege.createPolicy(
+ rowFilterPolicy,
+ Policy.BuiltInType.ROW_FILTER.policyType(),
+ null,
+ true,
+ PolicyContents.rowFilter("(col_1 INTEGER) RETURNS BOOLEAN -> col_1 =
1"));
+ metalakeClientWithAllPrivilege.addPolicyForTag(rowFilterTag,
rowFilterPolicy);
+
+ String columnMaskTag = "column_mask_tag_" + suffix;
+ String columnMaskPolicy = "column_mask_policy_" + suffix;
+ metalakeClientWithAllPrivilege.createTag(columnMaskTag, null, new
HashMap<>());
+ metalakeClientWithAllPrivilege.createPolicy(
+ columnMaskPolicy,
+ Policy.BuiltInType.COLUMN_MASK.policyType(),
+ null,
+ true,
+ PolicyContents.columnMask("(value INTEGER) RETURNS INTEGER ->
replace_with_null(value)"));
+ metalakeClientWithAllPrivilege.addPolicyForTag(columnMaskTag,
columnMaskPolicy);
+
+ Table gravitinoTable =
+ catalogClientWithAllPrivilege
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of(SCHEMA_NAME, tableName));
+ gravitinoTable.supportsTags().associateTags(new String[] {rowFilterTag},
null);
+ gravitinoTable.columns()[1].supportsTags().associateTags(new String[]
{columnMaskTag}, null);
+
+ List<Object[]> restrictedRows = sql("SELECT col_1, col_2 FROM %s ORDER BY
col_1", tableName);
+ Assertions.assertEquals(1, restrictedRows.size());
+ Assertions.assertArrayEquals(new Object[] {1, null},
restrictedRows.get(0));
+ }
+
+ @Test
+ void testTrueRowFilterOmitsReadRestrictions() throws Exception {
+ String tableName = "true_row_filter";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow());
+
+ String tag = createPolicyTag(Policy.BuiltInType.ROW_FILTER, "() RETURNS
BOOLEAN -> TRUE");
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertNoReadRestrictions(SCHEMA_NAME, tableName);
+ }
+
+ @Test
+ void testIdentityMaskOmitsReadRestrictions() throws Exception {
+ String tableName = "identity_column_mask";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow());
+
+ String tag =
+ createPolicyTag(Policy.BuiltInType.COLUMN_MASK, "(value INTEGER)
RETURNS INTEGER -> value");
+ loadTable(tableName).columns()[1].supportsTags().associateTags(new
String[] {tag}, null);
+
+ assertNoReadRestrictions(SCHEMA_NAME, tableName);
+ }
+
+ @Test
+ void testRowFilterFollowsPolicyAndTagChanges() throws Exception {
+ String tableName = "row_filter_lifecycle";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ PolicyTag policyTag =
+ createPolicyAndTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 1");
+ String tag = policyTag.tagName;
+ Table table = loadTable(tableName);
+ table.supportsTags().associateTags(new String[] {tag}, null);
+ assertRows(tableName, new Object[] {1, 10});
+
+ String policy = policyTag.policyName;
+ metalakeClientWithAllPrivilege.disablePolicy(policy);
+ assertRows(tableName, new Object[] {1, 10}, new Object[] {2, 20});
+
+ metalakeClientWithAllPrivilege.enablePolicy(policy);
+ assertRows(tableName, new Object[] {1, 10});
+
+ table.supportsTags().associateTags(null, new String[] {tag});
+ assertRows(tableName, new Object[] {1, 10}, new Object[] {2, 20});
+ assertNoReadRestrictions(SCHEMA_NAME, tableName);
+ table.supportsTags().associateTags(new String[] {tag}, null);
+ assertRows(tableName, new Object[] {1, 10});
+ Assertions.assertTrue(metalakeClientWithAllPrivilege.deleteTag(tag));
+ assertRows(tableName, new Object[] {1, 10}, new Object[] {2, 20});
+ assertNoReadRestrictions(SCHEMA_NAME, tableName);
+ }
+
+ @Test
+ void testRowFilterBindsAuthenticatedUser() {
+ String tableName = "row_filter_user";
+ catalogClientWithAllPrivilege
+ .asTableCatalog()
+ .createTable(
+ NameIdentifier.of(SCHEMA_NAME, tableName),
+ new Column[] {
+ Column.of("owner", Types.StringType.get(), "owner"),
+ Column.of("value", Types.IntegerType.get(), "value")
+ },
+ "table_comment",
+ new HashMap<>());
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES ('normal', 10), ('other', 20)", tableName);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER,
+ "(owner STRING) RETURNS BOOLEAN -> owner = CURRENT_USER()");
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ List<Object[]> rows = sql("SELECT owner, value FROM %s ORDER BY owner",
tableName);
+ Assertions.assertEquals(1, rows.size());
+ Assertions.assertArrayEquals(new Object[] {NORMAL_USER, 10}, rows.get(0));
+ }
+
+ @Test
+ void testReadRestrictionsBindEachRequestUser() throws Exception {
+ String tableName = "request_user_binding";
+ catalogClientWithAllPrivilege
+ .asTableCatalog()
+ .createTable(
+ NameIdentifier.of(SCHEMA_NAME, tableName),
+ new Column[] {Column.of("owner", Types.StringType.get(), "owner")},
+ "table_comment",
+ new HashMap<>());
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow());
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER,
+ "(owner STRING) RETURNS BOOLEAN -> owner = CURRENT_USER()");
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ String normalRestrictions = loadRestrictionsAs(tableName, NORMAL_USER);
+ String superRestrictions = loadRestrictionsAs(tableName, SUPER_USER);
+ Assertions.assertTrue(normalRestrictions.contains(NORMAL_USER));
+ Assertions.assertTrue(superRestrictions.contains(SUPER_USER));
+ Assertions.assertNotEquals(normalRestrictions, superRestrictions);
+ }
+
+ @Test
+ void testKnownGroupNonmemberSelectsFallback() {
+ String tableName = "known_group_nonmember";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ String group = "restriction_group_" +
UUID.randomUUID().toString().replace("-", "");
+ metalakeClientWithAllPrivilege.addGroup(group);
+ String expression =
+ "(col_1 INTEGER) RETURNS BOOLEAN -> CASE WHEN is_group_member('"
+ + group
+ + "') THEN col_1 = 1 ELSE col_1 = 2 END";
+ String tag = createPolicyTag(Policy.BuiltInType.ROW_FILTER, expression);
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertRows(tableName, new Object[] {2, 20});
+ }
+
+ @Test
+ void testGroupMemberSelectsProtectedBranch() throws Exception {
+ String tableName = "group_member_filter";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ String expression =
+ "(col_1 INTEGER) RETURNS BOOLEAN -> CASE WHEN is_group_member('"
+ + ReadRestrictionGroupAuthenticator.AUDITORS_GROUP
+ + "') THEN col_1 = 1 ELSE col_1 = 2 END";
+ String tag = createPolicyTag(Policy.BuiltInType.ROW_FILTER, expression);
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertRows(tableName, new Object[] {1, 10});
+ Assertions.assertNotEquals(
+ loadRestrictionsAs(tableName, NORMAL_USER),
loadRestrictionsAs(tableName, SUPER_USER));
+ }
+
+ @Test
+ void testComplexRowFilterExpression() {
+ String tableName = "complex_row_filter";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20), (3, 30), (4, 40)", tableName);
+
+ String expression =
+ "(col_1 INTEGER, col_2 INTEGER) RETURNS BOOLEAN -> "
+ + "(col_1 IN (1, 3) AND col_2 >= 20) OR (col_1 = 4 AND NOT (col_2
= 10))";
+ String tag = createPolicyTag(Policy.BuiltInType.ROW_FILTER, expression);
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertRows(tableName, new Object[] {3, 30}, new Object[] {4, 40});
+ }
+
+ @Test
+ void testConditionalRowFilterSelectsMatchingBranch() {
+ String tableName = "conditional_row_filter";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20), (3, 30)", tableName);
+
+ String expression =
+ "(col_1 INTEGER) RETURNS BOOLEAN -> CASE "
+ + "WHEN CURRENT_USER() = 'super' THEN col_1 = 1 "
+ + "WHEN CURRENT_USER() = 'normal' THEN col_1 = 2 ELSE FALSE END";
+ String tag = createPolicyTag(Policy.BuiltInType.ROW_FILTER, expression);
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertRows(tableName, new Object[] {2, 20});
+ }
+
+ @Test
+ void testInvalidUnselectedBranchFailsClosed() {
+ String tableName = "invalid_unselected_branch";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+
+ String expression =
+ "(missing_column INTEGER) RETURNS BOOLEAN -> CASE "
+ + "WHEN CURRENT_USER() = 'normal' THEN TRUE ELSE missing_column =
1 END";
+ String tag = createPolicyTag(Policy.BuiltInType.ROW_FILTER, expression);
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertReadRestrictionLoadFails(tableName);
+ }
+
+ @Test
+ void testDuplicateRowFilterPreservesEffectiveRestrictions() throws Exception
{
+ String tableName = "duplicate_row_filter";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ PolicyTag policyTag =
+ createPolicyAndTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 1");
+ String firstTag = policyTag.tagName;
+ String secondTag = "duplicate_tag_" +
UUID.randomUUID().toString().replace("-", "");
+ metalakeClientWithAllPrivilege.createTag(secondTag, null, new HashMap<>());
+ String policy = policyTag.policyName;
+ metalakeClientWithAllPrivilege.addPolicyForTag(secondTag, policy);
+ Table table = loadTable(tableName);
+ table.supportsTags().associateTags(new String[] {firstTag}, null);
+ JsonNode singleTagRestrictions =
+ new ObjectMapper().readTree(loadRestrictionsAs(tableName,
NORMAL_USER));
+
+ table.supportsTags().associateTags(new String[] {secondTag}, null);
+ JsonNode duplicateTagRestrictions =
+ new ObjectMapper().readTree(loadRestrictionsAs(tableName,
NORMAL_USER));
+ Assertions.assertEquals(singleTagRestrictions, duplicateTagRestrictions);
+
+ assertRows(tableName, new Object[] {1, 10});
+ }
+
+ @Test
+ void testEquivalentComplexRowFiltersDoNotConflict() {
+ String tableName = "equivalent_row_filters";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20), (3, 30)", tableName);
+
+ String firstTag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER,
+ "(col_1 INTEGER, col_2 INTEGER) RETURNS BOOLEAN -> col_1 IN (2, 1,
2) AND col_2 >= 10");
+ String secondTag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER,
+ "(col_1 INTEGER, col_2 INTEGER) RETURNS BOOLEAN -> col_2 >= 10 AND
col_1 IN (1, 2)");
+ loadTable(tableName).supportsTags().associateTags(new String[] {firstTag,
secondTag}, null);
+
+ assertRows(tableName, new Object[] {1, 10}, new Object[] {2, 20});
+ }
+
+ @Test
+ void testUnknownGroupInConditionFailsClosed() {
+ String tableName = "unknown_group_filter";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+
+ String expression =
+ "(col_1 INTEGER) RETURNS BOOLEAN -> CASE WHEN
is_group_member('missing_group') "
+ + "THEN col_1 = 1 ELSE col_1 = 2 END";
+ String tag = createPolicyTag(Policy.BuiltInType.ROW_FILTER, expression);
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertReadRestrictionLoadFails(tableName);
+ }
+
+ @Test
+ void testConflictingRowFiltersFailClosed() {
+ String tableName = "conflicting_row_filters";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ String firstTag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 1");
+ String secondTag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 2");
+ Table table = loadTable(tableName);
+ table.supportsTags().associateTags(new String[] {firstTag, secondTag},
null);
+
+ assertReadRestrictionLoadFails(tableName);
+
+ table.supportsTags().associateTags(null, new String[] {secondTag});
+ assertRows(tableName, new Object[] {1, 10});
+ }
+
+ @Test
+ void testInheritedTagValueSelectsPolicyAndTableValueOverridesIt() throws
Exception {
+ String schemaName = "inherited_restrictions";
+ String inheritedTable = "inherited_filter";
+ String overriddenTable = "overridden_filter";
+ catalogClientWithAllPrivilege.asSchemas().createSchema(schemaName, null,
new HashMap<>());
+ createTable(schemaName, inheritedTable);
+ createTable(schemaName, overriddenTable);
+ grantTablePrivileges(
+ schemaName, inheritedTable, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ grantTablePrivileges(
+ schemaName,
+ overriddenTable,
+ Privileges.SelectTable.allow(),
+ Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", schemaName);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", inheritedTable);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", overriddenTable);
+
+ String suffix = UUID.randomUUID().toString().replace("-", "");
+ String tag = "inherited_tag_" + suffix;
+ String policy = "inherited_policy_" + suffix;
+ metalakeClientWithAllPrivilege.createTag(
+ tag, null, new HashMap<>(),
TagValueConstraint.ofAllowedValues("finance", "risk"));
+ metalakeClientWithAllPrivilege.createPolicy(
+ policy,
+ Policy.BuiltInType.ROW_FILTER.policyType(),
+ null,
+ true,
+ PolicyContents.rowFilter("(col_1 INTEGER) RETURNS BOOLEAN -> col_1 =
1"));
+ metalakeClientWithAllPrivilege.addPolicyForTag(tag, policy,
TagValueSelector.of("finance"));
+ catalogClientWithAllPrivilege
+ .asSchemas()
+ .loadSchema(schemaName)
+ .supportsTags()
+ .associateTags(new TagValue[] {TagValue.of(tag, "finance")}, null);
+ loadTable(schemaName, overriddenTable)
+ .supportsTags()
+ .associateTags(new TagValue[] {TagValue.of(tag, "risk")}, null);
+
+ Assertions.assertFalse(loadRestrictionsAs(schemaName, inheritedTable,
NORMAL_USER).isEmpty());
+ assertNoReadRestrictions(schemaName, overriddenTable);
+ assertRows(inheritedTable, new Object[] {1, 10});
+ assertRows(overriddenTable, new Object[] {1, 10}, new Object[] {2, 20});
+
+ loadTable(schemaName, overriddenTable)
+ .supportsTags()
+ .associateTags(
+ new TagValue[] {TagValue.of(tag, "finance")},
+ new TagValue[] {TagValue.of(tag, "risk")});
+ assertRows(overriddenTable, new Object[] {1, 10});
+ }
+
+ @Test
+ void testCatalogSchemaAndTableTagValuesFallBackInOrder() {
+ String schemaName = "tag_value_precedence";
+ String tableName = "tag_value_precedence_table";
+ catalogClientWithAllPrivilege.asSchemas().createSchema(schemaName, null,
new HashMap<>());
+ createTable(schemaName, tableName);
+ grantTablePrivileges(
+ schemaName, tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", schemaName);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ String suffix = UUID.randomUUID().toString().replace("-", "");
+ String tag = "precedence_tag_" + suffix;
+ String policy = "precedence_policy_" + suffix;
+ metalakeClientWithAllPrivilege.createTag(
+ tag, null, new HashMap<>(),
TagValueConstraint.ofAllowedValues("finance", "risk"));
+ metalakeClientWithAllPrivilege.createPolicy(
+ policy,
+ Policy.BuiltInType.ROW_FILTER.policyType(),
+ null,
+ true,
+ PolicyContents.rowFilter("(col_1 INTEGER) RETURNS BOOLEAN -> col_1 =
1"));
+ metalakeClientWithAllPrivilege.addPolicyForTag(tag, policy,
TagValueSelector.of("finance"));
+ TagValue finance = TagValue.of(tag, "finance");
+ TagValue risk = TagValue.of(tag, "risk");
+ catalogClientWithAllPrivilege.supportsTags().associateTags(new TagValue[]
{risk}, null);
+ try {
+ assertRows(tableName, new Object[] {1, 10}, new Object[] {2, 20});
+
+ catalogClientWithAllPrivilege
+ .asSchemas()
+ .loadSchema(schemaName)
+ .supportsTags()
+ .associateTags(new TagValue[] {finance}, null);
+ assertRows(tableName, new Object[] {1, 10});
+
+ loadTable(schemaName, tableName).supportsTags().associateTags(new
TagValue[] {risk}, null);
+ assertRows(tableName, new Object[] {1, 10}, new Object[] {2, 20});
+
+ loadTable(schemaName, tableName).supportsTags().associateTags(null, new
TagValue[] {risk});
+ assertRows(tableName, new Object[] {1, 10});
+
+ catalogClientWithAllPrivilege
+ .asSchemas()
+ .loadSchema(schemaName)
+ .supportsTags()
+ .associateTags(null, new TagValue[] {finance});
+ assertRows(tableName, new Object[] {1, 10}, new Object[] {2, 20});
+ } finally {
+ catalogClientWithAllPrivilege.supportsTags().associateTags(null, new
TagValue[] {risk});
+ }
+ }
+
+ @Test
+ void testColumnMaskFollowsTagChanges() {
+ String tableName = "column_mask_lifecycle";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10)", tableName);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK,
+ "(value INTEGER) RETURNS INTEGER -> replace_with_null(value)");
+ Table table = loadTable(tableName);
+ table.columns()[1].supportsTags().associateTags(new String[] {tag}, null);
+ assertRows(tableName, new Object[] {1, null});
+
+ table.columns()[1].supportsTags().associateTags(null, new String[] {tag});
+ assertRows(tableName, new Object[] {1, 10});
+ }
+
+ @Test
+ void testConflictingColumnMasksFailClosed() {
+ String tableName = "conflicting_column_masks";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+
+ String nullTag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK,
+ "(value INTEGER) RETURNS INTEGER -> replace_with_null(value)");
+ String fixedTag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK,
+ "(value INTEGER) RETURNS INTEGER -> mask_to_fixed_value(value)");
+ Table table = loadTable(tableName);
+ table.columns()[1].supportsTags().associateTags(new String[] {nullTag,
fixedTag}, null);
+
+ assertReadRestrictionLoadFails(tableName);
+ }
+
+ @Test
+ void testConditionalMasksApplyToMultipleColumns() throws Exception {
+ String tableName = "conditional_column_masks";
+ catalogClientWithAllPrivilege
+ .asTableCatalog()
+ .createTable(
+ NameIdentifier.of(SCHEMA_NAME, tableName),
+ new Column[] {
+ Column.of("id", Types.IntegerType.get(), "id"),
+ Column.of("account", Types.StringType.get(), "account"),
+ Column.of("note", Types.StringType.get(), "note")
+ },
+ "table_comment",
+ new HashMap<>());
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, '12345678', 'abcdefgh')", tableName);
+
+ String accountTag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK,
+ "(value STRING) RETURNS STRING -> CASE WHEN CURRENT_USER() =
'normal' "
+ + "THEN replace_with_null(value) ELSE show_last_4(value) END");
+ String noteTag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK,
+ "(value STRING) RETURNS STRING -> replace_with_null(value)");
+ Table table = loadTable(tableName);
+ table.columns()[1].supportsTags().associateTags(new String[] {accountTag},
null);
+ table.columns()[2].supportsTags().associateTags(new String[] {noteTag},
null);
+
+ List<Object[]> rows = sql("SELECT id, account, note FROM %s", tableName);
+ Assertions.assertEquals(1, rows.size());
+ Assertions.assertArrayEquals(new Object[] {1, null, null}, rows.get(0));
+
+ String normalRestrictions = loadRestrictionsAs(tableName, NORMAL_USER);
+ String superRestrictions = loadRestrictionsAs(tableName, SUPER_USER);
+ Assertions.assertFalse(normalRestrictions.contains("show-last-4"));
+ Assertions.assertTrue(superRestrictions.contains("show-last-4"));
+ }
+
+ @Test
+ void testSparkAppliesFirstAndLastFourMasks() {
+ String tableName = "string_mask_actions";
+ catalogClientWithAllPrivilege
+ .asTableCatalog()
+ .createTable(
+ NameIdentifier.of(SCHEMA_NAME, tableName),
+ new Column[] {
+ Column.of("id", Types.IntegerType.get(), "id"),
+ Column.of("account", Types.StringType.get(), "account"),
+ Column.of("note", Types.StringType.get(), "note")
+ },
+ "table_comment",
+ new HashMap<>());
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, '12345678', 'abcdefgh')", tableName);
+
+ String lastTag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK, "(value STRING) RETURNS STRING ->
show_last_4(value)");
+ String firstTag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK, "(value STRING) RETURNS STRING ->
show_first_4(value)");
+ Table table = loadTable(tableName);
+ table.columns()[1].supportsTags().associateTags(new String[] {lastTag},
null);
+ table.columns()[2].supportsTags().associateTags(new String[] {firstTag},
null);
+
+ List<Object[]> rows = sql("SELECT id, account, note FROM %s", tableName);
+ Assertions.assertEquals(1, rows.size());
+ Assertions.assertArrayEquals(new Object[] {1, "nnnn5678", "abcdxxxx"},
rows.get(0));
+ }
+
+ @ParameterizedTest
+ @MethodSource("sqlNullPredicates")
+ void testSparkEnforcesSqlNullAndCaseSemantics(String predicate,
List<Integer> expectedIds) {
+ String tableName = "sql_nulls_" +
UUID.randomUUID().toString().replace("-", "");
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20), (3, NULL), (4, 40)",
tableName);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_2 INTEGER) RETURNS BOOLEAN ->
" + predicate);
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ List<Object[]> rows = sql("SELECT col_1 FROM %s ORDER BY col_1",
tableName);
+ List<Integer> actualIds = new ArrayList<>();
+ for (Object[] row : rows) {
+ actualIds.add((Integer) row[0]);
+ }
+ Assertions.assertEquals(expectedIds, actualIds, predicate);
+ }
+
+ @Test
+ void testMatchedCaseSkipsLaterGroupLookup() {
+ String tableName = "case_skips_group";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER,
+ "(col_1 INTEGER) RETURNS BOOLEAN -> CASE "
+ + "WHEN CURRENT_USER() = 'normal' THEN col_1 = 1 "
+ + "WHEN is_group_member('missing_group') THEN col_1 = 2 ELSE
FALSE END");
+ loadTable(tableName).supportsTags().associateTags(new String[] {tag},
null);
+
+ assertRows(tableName, new Object[] {1, 10});
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "CASE WHEN CURRENT_USER() = 'normal' THEN value ELSE NULL END",
+ "CASE WHEN CURRENT_USER() = 'normal' THEN value END"
+ })
+ void testNullableCaseMaskFailsForRequiredColumn(String expression) {
+ String tableName = "required_mask_" +
UUID.randomUUID().toString().replace("-", "");
+ catalogClientWithAllPrivilege
+ .asTableCatalog()
+ .createTable(
+ NameIdentifier.of(SCHEMA_NAME, tableName),
+ new Column[] {
+ Column.of("col_1", Types.IntegerType.get()),
+ Column.of(
+ "col_2",
+ Types.IntegerType.get(),
+ null,
+ false,
+ false,
+ Column.DEFAULT_VALUE_NOT_SET)
+ },
+ "table_comment",
+ new HashMap<>());
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10)", tableName);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK, "(value INTEGER) RETURNS INTEGER
-> " + expression);
+ Table table = loadTable(tableName);
+ Assertions.assertFalse(table.columns()[1].nullable());
+ table.columns()[1].supportsTags().associateTags(new String[] {tag}, null);
+
+ assertReadRestrictionLoadFails(tableName);
+ }
+
+ @Test
+ void testRowDependentMaskSelectionFailsClosed() {
+ String tableName = "row_dependent_mask";
+ createTable(SCHEMA_NAME, tableName);
+ grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
+ sql("USE %s", SPARK_CATALOG_NAME);
+ sql("USE %s", SCHEMA_NAME);
+ sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.COLUMN_MASK,
+ "(value INTEGER) RETURNS INTEGER -> CASE WHEN value = 10 "
+ + "THEN mask_to_fixed_value(value) ELSE value END");
+ loadTable(tableName).columns()[1].supportsTags().associateTags(new
String[] {tag}, null);
+
+ assertReadRestrictionLoadFails(tableName);
+ }
+
+ private static Stream<Arguments> sqlNullPredicates() {
+ return Stream.of(
+ Arguments.of("NOT (col_2 = 20)", ImmutableList.of(1, 4)),
+ Arguments.of("col_2 NOT IN (20, NULL)", ImmutableList.of()),
+ Arguments.of("col_2 IN (20, NULL)", ImmutableList.of(2)),
+ Arguments.of("col_2 IS NULL OR col_2 NOT BETWEEN 20 AND 30",
ImmutableList.of(1, 3, 4)),
+ Arguments.of(
+ "CASE WHEN col_2 = NULL THEN TRUE WHEN col_2 = 20 THEN TRUE ELSE
FALSE END",
+ ImmutableList.of(2)),
+ Arguments.of("CASE col_2 WHEN NULL THEN TRUE WHEN 20 THEN TRUE END",
ImmutableList.of(2)),
+ Arguments.of("CASE WHEN col_2 = 20 THEN TRUE END",
ImmutableList.of(2)));
+ }
+
+ private String createPolicyTag(Policy.BuiltInType type, String expression) {
+ return createPolicyAndTag(type, expression).tagName;
+ }
+
+ private PolicyTag createPolicyAndTag(Policy.BuiltInType type, String
expression) {
+ String suffix = UUID.randomUUID().toString().replace("-", "");
+ String tag = "restriction_tag_" + suffix;
+ String policy = "restriction_policy_" + suffix;
+ metalakeClientWithAllPrivilege.createTag(tag, null, new HashMap<>());
+ metalakeClientWithAllPrivilege.createPolicy(
+ policy,
+ type.policyType(),
+ null,
+ true,
+ type == Policy.BuiltInType.ROW_FILTER
+ ? PolicyContents.rowFilter(expression)
+ : PolicyContents.columnMask(expression));
+ metalakeClientWithAllPrivilege.addPolicyForTag(tag, policy);
+ return new PolicyTag(tag, policy);
+ }
+
+ private Table loadTable(String tableName) {
+ return loadTable(SCHEMA_NAME, tableName);
+ }
+
+ private Table loadTable(String schemaName, String tableName) {
+ return catalogClientWithAllPrivilege
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of(schemaName, tableName));
+ }
+
+ private void assertRows(String tableName, Object[]... expectedRows) {
+ List<Object[]> rows = sql("SELECT col_1, col_2 FROM %s ORDER BY col_1",
tableName);
+ Assertions.assertEquals(expectedRows.length, rows.size());
+ for (int index = 0; index < expectedRows.length; index++) {
+ Assertions.assertArrayEquals(expectedRows[index], rows.get(index));
+ }
+ }
+
+ private void assertNoReadRestrictions(String schemaName, String tableName)
throws Exception {
+ JsonNode response = loadTableResponseAs(schemaName, tableName,
NORMAL_USER);
+ Assertions.assertTrue(response.path("read-restrictions").isMissingNode(),
response::toString);
+ }
+
+ private void assertReadRestrictionLoadFails(String tableName) {
+ RuntimeException failure =
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
tableName));
+ for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
+ if (cause.getMessage() != null
+ && cause.getMessage().contains("Unable to resolve read
restrictions")) {
+ return;
+ }
+ }
+ Assertions.fail("Expected a read-restriction loader failure in the cause
chain", failure);
+ }
+
+ private String loadRestrictionsAs(String tableName, String userName) throws
Exception {
+ return loadRestrictionsAs(SCHEMA_NAME, tableName, userName);
+ }
+
+ private String loadRestrictionsAs(String schemaName, String tableName,
String userName)
+ throws Exception {
+ JsonNode response = loadTableResponseAs(schemaName, tableName, userName);
+ JsonNode restrictions = response.path("read-restrictions");
+ Assertions.assertFalse(restrictions.isMissingNode(), response.toString());
+ return restrictions.toString();
+ }
+
+ private JsonNode loadTableResponseAs(String schemaName, String tableName,
String userName)
+ throws Exception {
+ String uri =
+ getIcebergRestServiceUri()
+ + "v1/"
+ + GRAVITINO_CATALOG_NAME
+ + "/namespaces/"
+ + schemaName
+ + "/tables/"
+ + tableName;
+ String credentials = userName + ":mock";
+ String authorization =
+ "Basic " +
Base64.getEncoder().encodeToString(credentials.getBytes(StandardCharsets.UTF_8));
+ HttpRequest request =
+ HttpRequest.newBuilder(URI.create(uri))
+ .header("Authorization", authorization)
+ .GET()
+ .build();
+ HttpResponse<String> response = httpClient.send(request,
HttpResponse.BodyHandlers.ofString());
+ Assertions.assertEquals(200, response.statusCode(), response.body());
+ return new ObjectMapper().readTree(response.body());
+ }
+
+ private void grantTablePrivileges(String tableName, Privilege... privileges)
{
+ grantTablePrivileges(SCHEMA_NAME, tableName, privileges);
+ }
+
+ private void grantTablePrivileges(String schemaName, String tableName,
Privilege... privileges) {
+ String roleName = "readRestrictionRole_" + UUID.randomUUID();
+ List<SecurableObject> securableObjects = new ArrayList<>();
+ SecurableObject catalogObject =
+ SecurableObjects.ofCatalog(
+ GRAVITINO_CATALOG_NAME,
ImmutableList.of(Privileges.UseCatalog.allow()));
+ securableObjects.add(catalogObject);
+ SecurableObject schemaObject =
+ SecurableObjects.ofSchema(
+ catalogObject, schemaName,
ImmutableList.of(Privileges.UseSchema.allow()));
+ securableObjects.add(schemaObject);
+ securableObjects.add(
+ SecurableObjects.ofTable(schemaObject, tableName,
ImmutableList.copyOf(privileges)));
+ metalakeClientWithAllPrivilege.createRole(roleName, new HashMap<>(),
securableObjects);
+
metalakeClientWithAllPrivilege.grantRolesToUser(ImmutableList.of(roleName),
NORMAL_USER);
+ }
+
+ private static final class PolicyTag {
+ private final String tagName;
+ private final String policyName;
+
+ private PolicyTag(String tagName, String policyName) {
+ this.tagName = tagName;
+ this.policyName = policyName;
+ }
+ }
+}
diff --git
a/iceberg/iceberg-rest-server-experiment/src/test/java/org/apache/gravitino/iceberg/auth/TestReadRestrictionGroupAuthenticator.java
b/iceberg/iceberg-rest-server-experiment/src/test/java/org/apache/gravitino/iceberg/auth/TestReadRestrictionGroupAuthenticator.java
new file mode 100644
index 0000000000..fb377451b8
--- /dev/null
+++
b/iceberg/iceberg-rest-server-experiment/src/test/java/org/apache/gravitino/iceberg/auth/TestReadRestrictionGroupAuthenticator.java
@@ -0,0 +1,90 @@
+/*
+ * 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.iceberg.auth;
+
+import static org.apache.gravitino.utils.PrincipalUtils.ANONYMOUS_PRINCIPAL;
+
+import java.nio.charset.StandardCharsets;
+import java.security.Principal;
+import java.util.Base64;
+import org.apache.gravitino.UserPrincipal;
+import
org.apache.gravitino.iceberg.integration.test.auth.ReadRestrictionGroupAuthenticator;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestReadRestrictionGroupAuthenticator {
+
+ private final ReadRestrictionGroupAuthenticator authenticator =
+ new ReadRestrictionGroupAuthenticator();
+
+ @Test
+ void testNormalUserReceivesAuditorsGroupAndOriginalToken() {
+ String authorization = basic("normal:mock");
+ UserPrincipal principal =
+ (UserPrincipal)
+
authenticator.authenticateToken(authorization.getBytes(StandardCharsets.UTF_8));
+ Assertions.assertEquals("normal", principal.getName());
+ Assertions.assertEquals(1, principal.getGroups().size());
+ Assertions.assertEquals(
+ ReadRestrictionGroupAuthenticator.AUDITORS_GROUP,
+ principal.getGroups().get(0).getGroupName());
+ Assertions.assertEquals(authorization,
principal.getAccessToken().orElseThrow());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"super:mock", "other:mock", "other",
"other:password:with:colons"})
+ void testOtherUsersHaveNoGroups(String credential) {
+ String authorization = basic(credential);
+ UserPrincipal principal =
+ (UserPrincipal)
+
authenticator.authenticateToken(authorization.getBytes(StandardCharsets.UTF_8));
+ Assertions.assertEquals(credential.split(":")[0], principal.getName());
+ Assertions.assertTrue(principal.getGroups().isEmpty());
+ Assertions.assertEquals(authorization,
principal.getAccessToken().orElseThrow());
+ }
+
+ @ParameterizedTest
+ @NullSource
+ @ValueSource(
+ strings = {"", " ", "Bearer token", "Basic ", "Basic !!!", "Basic Og==",
"Basic Om1vY2s="})
+ void testInvalidOrMissingCredentialsAreAnonymous(String authorization) {
+ byte[] token = authorization == null ? null :
authorization.getBytes(StandardCharsets.UTF_8);
+ Principal principal = authenticator.authenticateToken(token);
+ Assertions.assertSame(ANONYMOUS_PRINCIPAL, principal);
+ }
+
+ @Test
+ void testSupportedTokens() {
+ Assertions.assertTrue(authenticator.isDataFromToken());
+ Assertions.assertTrue(authenticator.supportsToken(null));
+ Assertions.assertTrue(
+
authenticator.supportsToken(basic("normal:mock").getBytes(StandardCharsets.UTF_8)));
+ Assertions.assertFalse(
+ authenticator.supportsToken("Bearer
token".getBytes(StandardCharsets.UTF_8)));
+ Assertions.assertFalse(authenticator.supportsToken(new byte[0]));
+ }
+
+ private static String basic(String credential) {
+ return "Basic "
+ +
Base64.getEncoder().encodeToString(credential.getBytes(StandardCharsets.UTF_8));
+ }
+}
diff --git
a/iceberg/iceberg-rest-server-experiment/src/testFixtures/java/org/apache/gravitino/iceberg/integration/test/auth/ReadRestrictionGroupAuthenticator.java
b/iceberg/iceberg-rest-server-experiment/src/testFixtures/java/org/apache/gravitino/iceberg/integration/test/auth/ReadRestrictionGroupAuthenticator.java
new file mode 100644
index 0000000000..9bbb31efc9
--- /dev/null
+++
b/iceberg/iceberg-rest-server-experiment/src/testFixtures/java/org/apache/gravitino/iceberg/integration/test/auth/ReadRestrictionGroupAuthenticator.java
@@ -0,0 +1,88 @@
+/*
+ * 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.iceberg.integration.test.auth;
+
+import static org.apache.gravitino.utils.PrincipalUtils.ANONYMOUS_PRINCIPAL;
+
+import java.nio.charset.StandardCharsets;
+import java.security.Principal;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import javax.annotation.Nullable;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.UserGroup;
+import org.apache.gravitino.UserPrincipal;
+import org.apache.gravitino.auth.AuthConstants;
+import org.apache.gravitino.server.authentication.Authenticator;
+
+/** Basic authentication fixture that gives the normal test user a known group
membership. */
+public class ReadRestrictionGroupAuthenticator implements Authenticator {
+
+ /** Group used by the read restriction integration tests. */
+ public static final String AUDITORS_GROUP = "read_restriction_auditors";
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean isDataFromToken() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Principal authenticateToken(@Nullable byte[] tokenData) {
+ if (tokenData == null || !supportsToken(tokenData)) {
+ return ANONYMOUS_PRINCIPAL;
+ }
+ String authData = new String(tokenData, StandardCharsets.UTF_8);
+ try {
+ String credential =
authData.substring(AuthConstants.AUTHORIZATION_BASIC_HEADER.length());
+ String userName =
+ new String(Base64.getDecoder().decode(credential),
StandardCharsets.UTF_8)
+ .split(":", -1)[0];
+ if (userName.isEmpty()) {
+ return ANONYMOUS_PRINCIPAL;
+ }
+ // Keep "normal" in sync with IcebergAuthorizationIT.NORMAL_USER; this
fixture JAR is
+ // separate.
+ List<UserGroup> groups =
+ "normal".equals(userName)
+ ? Collections.singletonList(new UserGroup(Optional.empty(),
AUDITORS_GROUP))
+ : Collections.emptyList();
+ return new UserPrincipal(userName, groups, authData);
+ } catch (IllegalArgumentException e) {
+ return ANONYMOUS_PRINCIPAL;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void initialize(Config config) {
+ // The fixture does not need configuration.
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean supportsToken(@Nullable byte[] tokenData) {
+ return tokenData == null
+ || new String(tokenData, StandardCharsets.UTF_8)
+ .startsWith(AuthConstants.AUTHORIZATION_BASIC_HEADER);
+ }
+}
diff --git a/iceberg/iceberg-rest-server/build.gradle.kts
b/iceberg/iceberg-rest-server/build.gradle.kts
index e6bb8b9d3d..7fe437c182 100644
--- a/iceberg/iceberg-rest-server/build.gradle.kts
+++ b/iceberg/iceberg-rest-server/build.gradle.kts
@@ -16,8 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-import org.gradle.api.attributes.Attribute
-
description = "iceberg-rest-service"
plugins {
@@ -29,75 +27,8 @@ plugins {
val scalaVersion: String = project.properties["scalaVersion"] as? String ?:
extra["defaultScalaVersion"].toString()
val sparkVersion: String = libs.versions.spark34.get()
val sparkMajorVersion: String = sparkVersion.substringBeforeLast(".")
-val experimentalScalaVersion = "2.13"
-val experimentalSparkVersion: String = libs.versions.spark41.get()
-val experimentalSparkMajorVersion: String =
experimentalSparkVersion.substringBeforeLast(".")
val icebergVersion: String = libs.versions.iceberg.get()
-val datastratoIcebergVersion: String = libs.versions.datastrato.iceberg.get()
-val datastratoIcebergGitCommit: String =
libs.versions.experimental.iceberg.git.commit.get()
val scalaCollectionCompatVersion: String =
libs.versions.scala.collection.compat.get()
-val caffeineVersion: String = libs.versions.caffeine.get()
-val jerseyVersion: String = libs.versions.jersey.get()
-val javaxWsRsVersion: String = libs.versions.javax.ws.rs.api.get()
-val hk2Version = "2.6.1"
-val jakartaValidationVersion = "2.0.2"
-
-val testRuntimeClasspath = configurations.testRuntimeClasspath.get()
-val experimentalTestRuntimeClasspath by configurations.creating {
- isCanBeConsumed = false
- isCanBeResolved = true
- extendsFrom(testRuntimeClasspath)
- attributes {
- testRuntimeClasspath.attributes.keySet().forEach { key ->
- @Suppress("UNCHECKED_CAST")
- val typedKey = key as Attribute<Any>
- attribute(typedKey,
testRuntimeClasspath.attributes.getAttribute(typedKey)!!)
- }
- }
- resolutionStrategy {
- force(
- "com.github.ben-manes.caffeine:caffeine:$caffeineVersion",
- "javax.ws.rs:javax.ws.rs-api:$javaxWsRsVersion",
- "org.glassfish.jersey.core:jersey-server:$jerseyVersion",
- "org.glassfish.jersey.core:jersey-common:$jerseyVersion",
- "org.glassfish.jersey.core:jersey-client:$jerseyVersion",
-
"org.glassfish.jersey.containers:jersey-container-servlet:$jerseyVersion",
-
"org.glassfish.jersey.containers:jersey-container-servlet-core:$jerseyVersion",
-
"org.glassfish.jersey.containers:jersey-container-jetty-http:$jerseyVersion",
- "org.glassfish.jersey.inject:jersey-hk2:$jerseyVersion",
- "org.glassfish.hk2:hk2-api:$hk2Version",
- "org.glassfish.hk2:hk2-locator:$hk2Version",
- "org.glassfish.hk2:hk2-utils:$hk2Version",
- "org.glassfish.hk2.external:aopalliance-repackaged:$hk2Version",
- "org.glassfish.hk2.external:jakarta.inject:$hk2Version",
- "jakarta.validation:jakarta.validation-api:$jakartaValidationVersion"
- )
- eachDependency {
- if (requested.group == "org.apache.spark") {
- val experimentalSparkModule = requested.name.replace("_$scalaVersion",
"_$experimentalScalaVersion")
-
useTarget("${requested.group}:$experimentalSparkModule:$experimentalSparkVersion")
- because("The experimental Iceberg Spark runtime requires Spark 4.1
with Scala 2.13")
- } else if (
- requested.group == "org.scala-lang.modules" &&
requested.name.endsWith("_$scalaVersion")
- ) {
- val experimentalScalaModule = requested.name.replace("_$scalaVersion",
"_$experimentalScalaVersion")
-
useTarget("${requested.group}:$experimentalScalaModule:${requested.version}")
- because("Spark 4.1 uses Scala 2.13")
- } else if (requested.name.startsWith("iceberg-spark-runtime-")) {
- useTarget(
-
"com.datastrato:experimental-iceberg-spark-runtime-${experimentalSparkMajorVersion}_$experimentalScalaVersion:$datastratoIcebergVersion"
- )
- because("Run experimental Iceberg integration tests with the matching
Spark runtime")
- } else if (
- requested.name.startsWith("iceberg-") &&
- (requested.group == "org.apache.iceberg" || requested.group ==
"com.datastrato")
- ) {
-
useTarget("com.datastrato:experimental-${requested.name}:$datastratoIcebergVersion")
- because("Run experimental Iceberg integration tests with the
Datastrato distribution")
- }
- }
- }
-}
dependencies {
implementation(project(":api"))
@@ -269,44 +200,31 @@ tasks {
}
}
-val experimentalTest by tasks.registering(Test::class) {
- description = "Runs Iceberg REST tests that require the experimental Iceberg
runtime"
+// Share the authorization IT harness with the experimental service module.
+val testJar by tasks.registering(Jar::class) {
+ archiveClassifier.set("tests")
+ // Shared test resources must not inherit the production JAR's
logging/resource exclusions.
+ setExcludes(emptyList())
+ from(sourceSets.test.get().output)
+}
+
+configurations.create("testArtifacts") {
+ isCanBeResolved = false
+}
+
+artifacts {
+ add("testArtifacts", testJar)
+}
+
+// Retain the existing CI entry point while the experimental module owns the
tests and runtime.
+tasks.register("experimentalTest") {
+ description = "Runs experimental Iceberg REST integration tests"
group = "verification"
- testClassesDirs = sourceSets.test.get().output.classesDirs
- classpath =
- sourceSets.main.get().output +
- sourceSets.test.get().output +
- experimentalTestRuntimeClasspath
- systemProperty("datastratoIcebergGitCommit", datastratoIcebergGitCommit)
- systemProperty("spark.ui.enabled", "false")
- setIncludes(
- setOf(
- "**/IcebergExperimentalAuthorizationIT.class",
- "**/IcebergReadRestrictionIT.class"
- )
- )
- filter {
- includeTestsMatching(
-
"org.apache.gravitino.iceberg.integration.test.IcebergExperimentalAuthorizationIT"
- )
- includeTestsMatching(
- "org.apache.gravitino.iceberg.integration.test.IcebergReadRestrictionIT"
- )
- }
- mustRunAfter(tasks.test)
- dependsOn(
- tasks.testClasses,
- tasks.jar,
- ":catalogs:catalog-lakehouse-iceberg:jar",
- ":iceberg:iceberg-rest-server-experiment:copyDepends",
- ":iceberg:iceberg-rest-server-experiment:jar"
- )
+ dependsOn(":iceberg:iceberg-rest-server-experiment:experimentalTest")
}
tasks.test {
val skipITs = project.hasProperty("skipITs")
- exclude("**/IcebergExperimentalAuthorizationIT.class")
- exclude("**/IcebergReadRestrictionIT.class")
if (skipITs) {
// Exclude integration tests
exclude("**/integration/test/**")
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergAuthorizationIT.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergAuthorizationIT.java
index aba2d127aa..8318311696 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergAuthorizationIT.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergAuthorizationIT.java
@@ -178,7 +178,7 @@ public class IcebergAuthorizationIT extends BaseIT {
"gravitino.authorization.serviceAdmins",
SUPER_USER,
"gravitino.authenticators",
- "simple",
+ authenticator(),
// BaseIT will use SimpleAuthUserName to create Gravitino client
"SimpleAuthUserName",
SUPER_USER,
@@ -207,6 +207,11 @@ public class IcebergAuthorizationIT extends BaseIT {
return Maps.newHashMap();
}
+ /** Returns the authenticator used by the Gravitino server in this test
suite. */
+ protected String authenticator() {
+ return "simple";
+ }
+
protected boolean supportsCredentialVending() {
return false;
}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java
deleted file mode 100644
index 9936ebfaf0..0000000000
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * 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.iceberg.integration.test;
-
-import com.google.common.collect.ImmutableList;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.UUID;
-import org.apache.gravitino.NameIdentifier;
-import org.apache.gravitino.authorization.Privilege;
-import org.apache.gravitino.authorization.Privileges;
-import org.apache.gravitino.authorization.SecurableObject;
-import org.apache.gravitino.authorization.SecurableObjects;
-import org.apache.gravitino.integration.test.util.TestDatabaseName;
-import org.apache.gravitino.policy.Policy;
-import org.apache.gravitino.policy.PolicyContents;
-import org.apache.gravitino.rel.Table;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Tag;
-import org.junit.jupiter.api.Test;
-
-/** Verifies that Spark enforces read restrictions returned by the
experimental Iceberg service. */
-@Tag("gravitino-docker-test")
-public class IcebergReadRestrictionIT extends IcebergAuthorizationIT {
-
- private static final String EXPERIMENTAL_SERVER_DIRECTORY =
"iceberg-rest-server-experiment";
- private static final String SCHEMA_NAME = "schema";
-
- @BeforeAll
- @Override
- public void startIntegrationTest() throws Exception {
- String rootDirectory = System.getenv("GRAVITINO_ROOT_DIR");
- String experimentalServerHome =
- Paths.get(rootDirectory, "iceberg",
EXPERIMENTAL_SERVER_DIRECTORY).toString();
- customConfigs.put(
- "gravitino.iceberg-rest.classpath",
- String.join(
- ",",
- Paths.get(experimentalServerHome, "build", "libs").toString(),
- Paths.get(experimentalServerHome, "build",
"dependencies").toString()));
- super.startIntegrationTest();
- catalogClientWithAllPrivilege.asSchemas().createSchema(SCHEMA_NAME, null,
new HashMap<>());
- }
-
- @Override
- protected TestDatabaseName getTestDatabaseName() {
- return TestDatabaseName.PG_ICEBERG_READ_RESTRICTION_IT;
- }
-
- @Test
- void testSparkEnforcesReadRestrictions() {
- String tableName = "read_restrictions";
- createTable(SCHEMA_NAME, tableName);
- grantTablePrivileges(tableName, Privileges.SelectTable.allow(),
Privileges.ModifyTable.allow());
- sql("USE %s", SPARK_CATALOG_NAME);
- sql("USE %s", SCHEMA_NAME);
-
- sql("INSERT INTO %s VALUES (1, 10), (2, 20)", tableName);
- List<Object[]> unrestrictedRows = sql("SELECT col_1, col_2 FROM %s ORDER
BY col_1", tableName);
- Assertions.assertEquals(2, unrestrictedRows.size());
-
- String suffix = UUID.randomUUID().toString().replace("-", "");
- String rowFilterTag = "row_filter_tag_" + suffix;
- String rowFilterPolicy = "row_filter_policy_" + suffix;
- metalakeClientWithAllPrivilege.createTag(rowFilterTag, null, new
HashMap<>());
- metalakeClientWithAllPrivilege.createPolicy(
- rowFilterPolicy,
- Policy.BuiltInType.ROW_FILTER.policyType(),
- null,
- true,
- PolicyContents.rowFilter("(col_1 INTEGER) RETURNS BOOLEAN -> col_1 =
1"));
- metalakeClientWithAllPrivilege.addPolicyForTag(rowFilterTag,
rowFilterPolicy);
-
- String columnMaskTag = "column_mask_tag_" + suffix;
- String columnMaskPolicy = "column_mask_policy_" + suffix;
- metalakeClientWithAllPrivilege.createTag(columnMaskTag, null, new
HashMap<>());
- metalakeClientWithAllPrivilege.createPolicy(
- columnMaskPolicy,
- Policy.BuiltInType.COLUMN_MASK.policyType(),
- null,
- true,
- PolicyContents.columnMask("(value INTEGER) RETURNS INTEGER ->
replace_with_null(value)"));
- metalakeClientWithAllPrivilege.addPolicyForTag(columnMaskTag,
columnMaskPolicy);
-
- Table gravitinoTable =
- catalogClientWithAllPrivilege
- .asTableCatalog()
- .loadTable(NameIdentifier.of(SCHEMA_NAME, tableName));
- gravitinoTable.supportsTags().associateTags(new String[] {rowFilterTag},
null);
- gravitinoTable.columns()[1].supportsTags().associateTags(new String[]
{columnMaskTag}, null);
-
- List<Object[]> restrictedRows = sql("SELECT col_1, col_2 FROM %s ORDER BY
col_1", tableName);
- Assertions.assertEquals(1, restrictedRows.size());
- Assertions.assertArrayEquals(new Object[] {1, null},
restrictedRows.get(0));
- }
-
- private void grantTablePrivileges(String tableName, Privilege... privileges)
{
- String roleName = "readRestrictionRole_" + UUID.randomUUID();
- List<SecurableObject> securableObjects = new ArrayList<>();
- SecurableObject catalogObject =
- SecurableObjects.ofCatalog(
- GRAVITINO_CATALOG_NAME,
ImmutableList.of(Privileges.UseCatalog.allow()));
- securableObjects.add(catalogObject);
- SecurableObject schemaObject =
- SecurableObjects.ofSchema(
- catalogObject, SCHEMA_NAME,
ImmutableList.of(Privileges.UseSchema.allow()));
- securableObjects.add(schemaObject);
- securableObjects.add(
- SecurableObjects.ofTable(schemaObject, tableName,
ImmutableList.copyOf(privileges)));
- metalakeClientWithAllPrivilege.createRole(roleName, new HashMap<>(),
securableObjects);
-
metalakeClientWithAllPrivilege.grantRolesToUser(ImmutableList.of(roleName),
NORMAL_USER);
- }
-}
