jerryshao commented on code in PR #14331:
URL: https://github.com/apache/gravitino/pull/14331#discussion_r4413188173
##########
iceberg/iceberg-rest-server/build.gradle.kts:
##########
@@ -269,44 +200,35 @@ tasks {
}
}
-val experimentalTest by tasks.registering(Test::class) {
- description = "Runs Iceberg REST tests that require the experimental Iceberg
runtime"
- 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"
+// Share the authorization IT harness with the experimental service module.
+val testJar by tasks.registering(Jar::class) {
+ archiveClassifier.set("tests")
+ from(sourceSets.test.get().output) {
+ include(
+
"org/apache/gravitino/iceberg/integration/test/IcebergAuthorizationIT*.class",
+
"org/apache/gravitino/iceberg/integration/test/IcebergTableAuthorizationIT*.class"
)
}
- 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"
- )
+}
+
+configurations.create("testArtifacts") {
+ isCanBeResolved = false
+ extendsFrom(configurations.testRuntimeClasspath.get())
Review Comment:
[Important] `extendsFrom(configurations.testRuntimeClasspath.get())` makes
the consumable `testArtifacts` configuration re-export this module's entire
test runtime closure -- Spark, the Iceberg bundles, testcontainers,
integration-test-common -- and
`testImplementation(project(":iceberg:iceberg-rest-server", "testArtifacts"))`
in the experiment module
(iceberg/iceberg-rest-server-experiment/build.gradle.kts:130) puts all of it on
that module's test **compile and runtime** classpaths.
The side effect is visible in this same PR: because the experiment module's
`testRuntimeClasspath` now drags in Spark, the Jackson `force(...)` had to be
switched off for every configuration whose name contains `test`
(iceberg/iceberg-rest-server-experiment/build.gradle.kts:46). So the module's
three pure unit-test classes (`TestIcebergReadRestrictionConverter`,
`TestExperimentalIcebergReadRestrictionLoader`, `TestSqlPredicateResult`) now
resolve Jackson from whatever Spark brings rather than the pinned
`libs.versions.jackson`, while the `main` classpath they exercise is still
pinned. That is exactly the kind of skew those unit tests are meant to catch.
Every other producer of `testArtifacts` in the repo uses the plain form with
no `extendsFrom` and lets the consumer declare what it needs:
`integration-test-common/build.gradle.kts:81-83`,
`catalogs/catalog-jdbc-common/build.gradle.kts:32-34`,
`catalogs/hive-metastore-common/build.gradle.kts:152`,
`authorizations/authorization-ranger/build.gradle.kts:168`. Suggest following
that: drop `extendsFrom`, declare the handful of deps the experimental suite
actually needs (`project(":integration-test-common", "testArtifacts")`, the
Spark/Iceberg test bundles, PostgreSQL testcontainers) in the experiment
module, and then the Jackson pin at line 46 can stay unconditional.
Verified by: read both build files in full on this branch, compared against
the four other `testArtifacts` producers listed above, and confirmed
`IcebergAuthorizationIT` imports `org.apache.spark.sql.SparkSession` (line 58)
so Spark really is in the re-exported closure.
##########
iceberg/iceberg-rest-server-experiment/src/test/java/org/apache/gravitino/server/authentication/ReadRestrictionGroupAuthenticator.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.server.authentication;
+
+import java.security.Principal;
+import java.util.Collections;
+import java.util.Optional;
+import org.apache.gravitino.UserGroup;
+import org.apache.gravitino.UserPrincipal;
+
+/** Supplies a group-bearing principal for Iceberg read-restriction
integration tests. */
+public class ReadRestrictionGroupAuthenticator extends SimpleAuthenticator {
Review Comment:
[Important] `SimpleAuthenticator` is package-private: `class
SimpleAuthenticator implements Authenticator`
(server-common/src/main/java/org/apache/gravitino/server/authentication/SimpleAuthenticator.java:36).
Subclassing it from here only compiles because this test class is planted into
`org.apache.gravitino.server.authentication`, creating a split package between
server-common's `main` jar and an Iceberg module's `test` source set.
That works today only because every path happens to load both halves through
the same loader: in embedded mode both come off the test classpath, and in
deploy mode the fixture jar lands on `GRAVITINO_CLASSPATH` while server-common
sits in the distribution libs, both on the app loader, and
`IsolatedClassLoader.CustomURLClassLoader` delegates parent-first
(core/src/main/java/org/apache/gravitino/utils/IsolatedClassLoader.java:218-233).
Package-private access is checked against the *runtime* package, i.e. name
plus defining loader, so the day the auxiliary service's classpath also carries
server-common and resolution goes child-first -- or anything moves onto the
module path -- this turns into an `IllegalAccessError` at server startup,
surfacing as an opaque authenticator construction failure through
`AuthenticatorFactory` (server-common/.../AuthenticatorFactory.java:59-66).
Two cleaner options: make `SimpleAuthenticator` public (it is already
reachable by name through `gravitino.authenticators`, so it is public API in
all but the modifier), or implement the public `Authenticator` interface here
directly -- the basic-token parsing this fixture depends on is ~15 lines.
Either removes the split package and lets the fixture live in its own package.
Verified by: read `SimpleAuthenticator` (package-private at line 36),
`AuthenticatorFactory.createAuthenticators`, and
`IsolatedClassLoader.CustomURLClassLoader.doLoadClass` on this branch.
##########
iceberg/iceberg-rest-server-experiment/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java:
##########
@@ -0,0 +1,839 @@
+/*
+ * 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.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.server.authentication.ReadRestrictionGroupAuthenticator;
+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.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";
+
+ @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<>());
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected String authenticator() {
+ return ReadRestrictionGroupAuthenticator.class.getCanonicalName() +
",simple";
Review Comment:
[Question] Is the trailing `,simple` meant to be a fallback? It cannot be
reached. `ReadRestrictionGroupAuthenticator` inherits
`SimpleAuthenticator.supportsToken`, which returns true for any Basic header
and for a missing one (server-common/.../SimpleAuthenticator.java:74-79), and
`AuthenticationFilter.resolvePrincipal` stops at the first authenticator that
both supports the token and is `isDataFromToken()` -- "If token is supported by
multiple authenticators, use the first by default"
(server-common/.../AuthenticationFilter.java:193-199). Since this class is
listed first and `super.authenticateToken` never returns null (worst case
`ANONYMOUS_PRINCIPAL`), the `simple` entry is dead config.
Not harmful, but it reads as if non-`normal` users were handled by a
different authenticator, when in fact the same class handles them and just
passes the principal through. Dropping `,simple` would make that clearer;
keeping it deserves a one-line comment saying why.
Verified by: read `SimpleAuthenticator.supportsToken`/`authenticateToken`
and `AuthenticationFilter.resolvePrincipal` on this branch.
##########
iceberg/iceberg-rest-server-experiment/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java:
##########
@@ -0,0 +1,839 @@
+/*
+ * 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.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.server.authentication.ReadRestrictionGroupAuthenticator;
+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.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";
+
+ @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<>());
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected String authenticator() {
+ return ReadRestrictionGroupAuthenticator.class.getCanonicalName() +
",simple";
+ }
+
+ @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));
+ }
+
+ @Test
+ void testRowFilterFollowsPolicyAndTagChanges() {
+ 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);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 1");
+ Table table = loadTable(tableName);
+ table.supportsTags().associateTags(new String[] {tag}, null);
+ assertRows(tableName, new Object[] {1, 10});
+
+ String policy = policyNameForTag(tag);
+ 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});
+ }
+
+ @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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
tableName));
+ }
+
+ @Test
+ void testDuplicateRowFilterIsAppliedOnce() {
+ 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);
+
+ String firstTag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 1");
+ String secondTag = "duplicate_tag_" +
UUID.randomUUID().toString().replace("-", "");
+ metalakeClientWithAllPrivilege.createTag(secondTag, null, new HashMap<>());
+ String policy = policyNameForTag(firstTag);
+ metalakeClientWithAllPrivilege.addPolicyForTag(secondTag, policy);
+ loadTable(tableName).supportsTags().associateTags(new String[] {firstTag,
secondTag}, null);
+
+ 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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
tableName));
+
+ table.supportsTags().associateTags(null, new String[] {secondTag});
+ assertRows(tableName, new Object[] {1, 10});
+ }
+
+ @Test
+ void testInheritedTagValueSelectsPolicyAndTableValueOverridesIt() {
+ 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);
+
+ 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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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.assertEquals(1, rows.get(0)[0]);
+ String account = (String) rows.get(0)[1];
+ String note = (String) rows.get(0)[2];
+ Assertions.assertNotEquals("12345678", account);
+ Assertions.assertNotEquals("abcdefgh", note);
+ Assertions.assertTrue(account.endsWith("5678"));
+ Assertions.assertTrue(note.startsWith("abcd"));
+ }
+
+ @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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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) {
+ 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 tag;
+ }
+
+ 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 String policyNameForTag(String tag) {
+ return "restriction_policy_" + tag.substring("restriction_tag_".length());
+ }
+
+ 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 String loadRestrictionsAs(String tableName, String userName) throws
Exception {
+ String uri =
+ getIcebergRestServiceUri()
+ + "v1/"
+ + GRAVITINO_CATALOG_NAME
Review Comment:
[Nit] The method takes a `tableName` but hardcodes `SCHEMA_NAME` in the URI,
so it silently only works for tables in the default schema. This PR already
introduces two other schemas -- `inherited_restrictions` (line 409) and
`tag_value_precedence` (line 461) -- so a future caller passing a table from
one of those gets a 404 that surfaces as the opaque `assertEquals(200, ...)`
failure at line 813 rather than as a wrong-schema mistake. Taking the schema as
a parameter (with a one-arg overload defaulting to `SCHEMA_NAME`, mirroring
`loadTable`) costs nothing.
Verified by: checked all three call sites (lines 217-218, 263, 593-594); all
use tables in `SCHEMA_NAME`, so there is no failure today.
##########
iceberg/iceberg-rest-server/build.gradle.kts:
##########
@@ -269,44 +200,35 @@ tasks {
}
}
-val experimentalTest by tasks.registering(Test::class) {
- description = "Runs Iceberg REST tests that require the experimental Iceberg
runtime"
- 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"
+// Share the authorization IT harness with the experimental service module.
+val testJar by tasks.registering(Jar::class) {
+ archiveClassifier.set("tests")
+ from(sourceSets.test.get().output) {
+ include(
+
"org/apache/gravitino/iceberg/integration/test/IcebergAuthorizationIT*.class",
Review Comment:
[Nit] This `include` list makes the shared harness jar two class-name
patterns wide and drops test resources entirely. Two consequences:
1. `src/test/resources/log4j2.properties` is not in the jar, which is why
the PR adds a byte-identical 73-line copy at
`iceberg/iceberg-rest-server-experiment/src/test/resources/log4j2.properties`
(I diffed them -- identical). Logging config for the experimental suite will
now silently diverge from the one it was moved out of.
2. A helper class added later to
`org.apache.gravitino.iceberg.integration.test` and used by the harness
compiles fine here but is missing from the consumer's runtime classpath, so it
fails as a `NoClassDefFoundError` inside `experimentalTest` rather than at
compile time.
`from(sourceSets.test.get().output)` with no `include` -- the form used by
`integration-test-common/build.gradle.kts:76-79` and
`catalogs/catalog-jdbc-common/build.gradle.kts:27-30` -- fixes both and lets
the duplicated `log4j2.properties` go away.
Verified by: checked all 28 other classes in that package for references
from `IcebergAuthorizationIT`/`IcebergTableAuthorizationIT` (none, so the list
is sufficient today), and `diff`ed the two `log4j2.properties` files.
##########
iceberg/iceberg-rest-server-experiment/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergReadRestrictionIT.java:
##########
@@ -0,0 +1,839 @@
+/*
+ * 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.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.server.authentication.ReadRestrictionGroupAuthenticator;
+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.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";
+
+ @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<>());
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected String authenticator() {
+ return ReadRestrictionGroupAuthenticator.class.getCanonicalName() +
",simple";
+ }
+
+ @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));
+ }
+
+ @Test
+ void testRowFilterFollowsPolicyAndTagChanges() {
+ 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);
+
+ String tag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 1");
+ Table table = loadTable(tableName);
+ table.supportsTags().associateTags(new String[] {tag}, null);
+ assertRows(tableName, new Object[] {1, 10});
+
+ String policy = policyNameForTag(tag);
+ 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});
+ }
+
+ @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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
tableName));
+ }
+
+ @Test
+ void testDuplicateRowFilterIsAppliedOnce() {
+ 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);
+
+ String firstTag =
+ createPolicyTag(
+ Policy.BuiltInType.ROW_FILTER, "(col_1 INTEGER) RETURNS BOOLEAN ->
col_1 = 1");
+ String secondTag = "duplicate_tag_" +
UUID.randomUUID().toString().replace("-", "");
+ metalakeClientWithAllPrivilege.createTag(secondTag, null, new HashMap<>());
+ String policy = policyNameForTag(firstTag);
+ metalakeClientWithAllPrivilege.addPolicyForTag(secondTag, policy);
+ loadTable(tableName).supportsTags().associateTags(new String[] {firstTag,
secondTag}, null);
+
+ 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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
tableName));
+
+ table.supportsTags().associateTags(null, new String[] {secondTag});
+ assertRows(tableName, new Object[] {1, 10});
+ }
+
+ @Test
+ void testInheritedTagValueSelectsPolicyAndTableValueOverridesIt() {
+ 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);
+
+ 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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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.assertEquals(1, rows.get(0)[0]);
+ String account = (String) rows.get(0)[1];
+ String note = (String) rows.get(0)[2];
+ Assertions.assertNotEquals("12345678", account);
+ Assertions.assertNotEquals("abcdefgh", note);
+ Assertions.assertTrue(account.endsWith("5678"));
+ Assertions.assertTrue(note.startsWith("abcd"));
+ }
+
+ @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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> sql("SELECT col_1, col_2 FROM %s",
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) {
+ 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 tag;
+ }
+
+ 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 String policyNameForTag(String tag) {
+ return "restriction_policy_" + tag.substring("restriction_tag_".length());
Review Comment:
[Nit] Recovering the policy name by slicing the tag name couples two helpers
through a string convention that holds only for tags built by `createPolicyTag`
(line 755). This PR already creates a tag outside that helper: `secondTag =
"duplicate_tag_" + ...` at line 333. Passing that one here would not throw --
it is long enough to slice -- it would just produce
`restriction_policy_tag_<hex>` and fail later as a confusing
`NoSuchPolicyException` from `disablePolicy`/`enablePolicy`.
Having `createPolicyTag` return both names (a tiny record, or a second
`createPolicy`+`createTag` pair the caller holds onto) removes the coupling and
the failure mode.
Verified by: traced both current call sites (lines 160 and 335); both pass
tags produced by `createPolicyTag`, so there is no bug today.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
