jerryshao commented on code in PR #13778:
URL: https://github.com/apache/gravitino/pull/13778#discussion_r4333699979
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java:
##########
@@ -343,14 +346,20 @@ public Response loadTable(
LoadTableResponse loadTableResponse =
tableOperationDispatcher.loadTable(context, tableIdentifier);
+ if
(IcebergRESTUtils.SnapshotMode.REFS.getValue().equals(snapshots)) {
+ loadTableResponse = filterSnapshotsByRefs(loadTableResponse);
Review Comment:
[Nit] Now that the refs filter runs first, the fields it drops are dropped
for restricted responses too.
`filterSnapshotsByRefs` rebuilds the response copying only `config` and
`credentials` (lines 655-662), while the new loader carefully carries
`remoteSigningConfig` and `labels` across
(`...ExperimentalIcebergReadRestrictionLoader.java:119-130`). The loss predates
this PR, but this diff moves the filter in front of the loader, so for
`snapshots=refs` the loader now preserves fields on a response that already
lost them. Worth copying the same four fields in `filterSnapshotsByRefs` (or
extracting one shared copy helper) while this code is being touched.
Verified by: comparing the two rebuild sites in this checkout; the loader's
use of `response.remoteSigningConfig()` and `response.labels()` compiles on
this head, so both accessors exist in the Iceberg build in use.
##########
iceberg/iceberg-rest-server-experiment/src/main/java/org/apache/gravitino/iceberg/service/restriction/ExperimentalIcebergReadRestrictionLoader.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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.service.restriction;
+
+import java.security.Principal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.UserGroup;
+import org.apache.gravitino.UserPrincipal;
+import org.apache.gravitino.authorization.AccessControlDispatcher;
+import org.apache.gravitino.exceptions.NoSuchGroupException;
+import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
+import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.ColumnMaskContent;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyDispatcher;
+import org.apache.gravitino.policy.RowFilterContent;
+import org.apache.gravitino.policy.rego.RestrictedRegoExpressionParserFacade;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.ColumnMask;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.RowFilter;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.PrincipalUtils;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.restrictions.ReadRestrictions;
+import org.apache.iceberg.rest.restrictions.ReadRestrictionsParser;
+import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Loads tag-selected Gravitino policies into experimental Iceberg read
restrictions. */
+public final class ExperimentalIcebergReadRestrictionLoader
+ implements IcebergReadRestrictionLoader {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ExperimentalIcebergReadRestrictionLoader.class);
+
+ /** Creates the service-loaded read-restriction loader. */
+ public ExperimentalIcebergReadRestrictionLoader() {}
+
+ @Override
+ public Result load(
+ IcebergRequestContext context, TableIdentifier tableIdentifier,
LoadTableResponse response) {
+ GravitinoEnv environment = GravitinoEnv.getInstance();
+ PolicyDispatcher policyDispatcher = environment.internalPolicyDispatcher();
+ if (policyDispatcher == null) {
+ return Result.of(response, Optional.empty());
+ }
+
+ String metalake = IcebergRESTServerContext.getInstance().metalakeName();
+ NameIdentifier tableName =
+ IcebergIdentifierUtils.toGravitinoTableIdentifier(
+ metalake,
+ context.catalogName(),
+ tableIdentifier,
+ HierarchicalSchemaUtil.schemaSeparator());
+ Schema schema = response.tableMetadata().schema();
+ PolicyEntity[] rowPolicies =
+ policyDispatcher.listPolicyInfosForMetadataObject(
+ metalake, NameIdentifierUtil.toMetadataObject(tableName,
Entity.EntityType.TABLE));
+ Map<Integer, PolicyEntity[]> maskPolicies = new LinkedHashMap<>();
+ for (Types.NestedField field : schema.columns()) {
+ NameIdentifier columnName =
+ NameIdentifierUtil.ofColumn(
+ metalake,
+ context.catalogName(),
+ tableName.namespace().level(2),
+ tableName.name(),
+ field.name());
+ MetadataObject column =
+ NameIdentifierUtil.toMetadataObject(columnName,
Entity.EntityType.COLUMN);
+ maskPolicies.put(
+ field.fieldId(),
policyDispatcher.listPolicyInfosForMetadataObject(metalake, column));
Review Comment:
[Important] This loop makes load-table cost grow with the column count, on a
path that also loses the conditional-GET fast path.
For every top-level field, `listPolicyInfosForMetadataObject` runs
`MetadataObjectUtil.checkMetadataObject` (for a COLUMN this calls
`internalTableDispatcher().tableExists(table)`,
`core/src/main/java/org/apache/gravitino/utils/MetadataObjectUtil.java:260-264`)
and then `ObjectPolicyResolver.resolve`
(`core/src/main/java/org/apache/gravitino/policy/PolicyManager.java:221-227`),
which calls `EffectiveTagResolver.resolve` - another `checkMetadataObject` plus
one `listEntitiesByRelation` per ancestor level (column, table, schema,
catalog, metalake:
`core/src/main/java/org/apache/gravitino/tag/EffectiveTagResolver.java:67-90`),
followed by a batch policy-tag lookup. So a 200-column table means ~400
table-existence checks and ~1000 relation queries per load-table, and because
the experimental loader does not override `enabled()`,
`IcebergTableOperations.java:336` now skips the metadata-location-only
If-None-Match fast path, so every conditional GET pays the full price too.
Suggestion: resolve column policies in one pass - batch the column metadata
objects (the ancestor chain above the table is identical for all of them, and
the table-existence check is repeated), or resolve the table's effective tags
once and only query the columns that carry their own tags. At minimum, please
measure a wide table before this ships, and say in the javadoc that the fast
path is traded away for all tables, including unrestricted ones.
Verified by: following the call chain by hand in this checkout
(PolicyManager -> ObjectPolicyResolver -> EffectiveTagResolver ->
MetadataObjectUtil.getParentMetadataObjects) and the `enabled()` default at
IcebergReadRestrictionLoader.java:40. A Caffeine entity cache exists in
`core/src/main/java/org/apache/gravitino/cache/`, so repeat lookups may be
cheaper than the raw counts - but the per-column fan-out itself is
unconditional.
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java:
##########
@@ -117,6 +119,7 @@ public IcebergTableOperations(
this.icebergMetricsManager = icebergMetricsManager;
this.tableOperationDispatcher = tableOperationDispatcher;
this.icebergObjectMapper = IcebergObjectMapper.getInstance();
+ this.readRestrictionLoader = IcebergReadRestrictionLoader.load();
Review Comment:
[Important] `ServiceLoader` discovery runs once per HTTP request, not once
per server.
This resource class is registered by package scanning
(`iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/RESTService.java:112`)
and carries no `@Singleton`, so JAX-RS instantiates it per request - this
constructor, and therefore `IcebergReadRestrictionLoader.load()` with its
classpath scan and provider instantiation, runs for every table request (list,
create, load, drop, metrics, credentials, plan...). It also means the "multiple
loaders" guard (`IcebergReadRestrictionLoader.java:63-69`) turns a deployment
mistake into a per-request HTTP 500 instead of a startup failure.
Suggestion: resolve the loader once in a static holder (`private static
final IcebergReadRestrictionLoader LOADER =
IcebergReadRestrictionLoader.load();`) or inject it like the other
collaborators, so discovery and the misconfiguration check happen at startup.
Verified by: no `@Singleton` on `IcebergTableOperations` (line 95-98) and no
explicit `register(instance)` for it - `config.packages(...)` is the only
registration, which is JAX-RS per-request lifecycle.
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java:
##########
@@ -343,14 +346,20 @@ public Response loadTable(
LoadTableResponse loadTableResponse =
tableOperationDispatcher.loadTable(context, tableIdentifier);
+ if
(IcebergRESTUtils.SnapshotMode.REFS.getValue().equals(snapshots)) {
+ loadTableResponse = filterSnapshotsByRefs(loadTableResponse);
+ }
+ IcebergReadRestrictionLoader.Result restrictionResult =
+ readRestrictionLoader.load(context, tableIdentifier,
loadTableResponse);
+ loadTableResponse = restrictionResult.response();
Review Comment:
[Important] Restrictions are attached only to load-table, while server-side
scan planning hands out the same table's data without them.
`POST .../tables/{table}/plan` (`IcebergTableOperations.java:517-566`) and
`POST .../tables/{table}/tasks` are guarded by the very same
`LOAD_TABLE_AUTHORIZATION_EXPRESSION` as `loadTable`, and return
`PlanTableScanResponse` / `FetchScanTasksResponse` with file scan tasks (and
optionally vended credentials) untouched by this loader. A client that uses
REST scan planning never sees the row filter or the column masks, so a reader
with plain select privileges gets unrestricted data through a path that needs
no extra privilege. The same is true, less importantly, of the
`LoadTableResponse` returned by `createTable` (line 203) and `updateTable`
(line 246), though those need write privileges.
Given the PR's trust model (clients enforce what the service returns), this
is the gap worth closing or at least documenting: either resolve restrictions
for the scan-planning endpoints as well, or reject plan/tasks requests for a
table that has applicable restrictions, and state the limitation in the
experimental service docs.
Verified by: reading the two endpoint methods and their authorization
annotations in this file at the reviewed commit; neither mentions
`readRestrictionLoader`, and `grep` for `IcebergReadRestrictionLoader` in
`iceberg/` shows `loadTable` as its only call site.
##########
iceberg/iceberg-rest-server-experiment/src/main/java/org/apache/gravitino/iceberg/service/restriction/IcebergReadRestrictionConverter.java:
##########
@@ -0,0 +1,773 @@
+/*
+ * 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.service.restriction;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.format.DateTimeParseException;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.TreeMap;
+import org.apache.gravitino.policy.rego.CanonicalExpression;
+import org.apache.gravitino.policy.rego.Column;
+import org.apache.gravitino.policy.rego.Comparison;
+import org.apache.gravitino.policy.rego.ComparisonOperator;
+import org.apache.gravitino.policy.rego.GroupMembership;
+import org.apache.gravitino.policy.rego.Literal;
+import org.apache.gravitino.policy.rego.LiteralArray;
+import org.apache.gravitino.policy.rego.LiteralType;
+import org.apache.gravitino.policy.rego.LogicalExpression;
+import org.apache.gravitino.policy.rego.LogicalOperator;
+import org.apache.gravitino.policy.rego.Not;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.ColumnMask;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.MaskAction;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.MaskBranch;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.RowFilter;
+import org.apache.gravitino.policy.rego.SessionUser;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionParser;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.functions.IcebergFunction;
+import org.apache.iceberg.functions.IcebergFunctions;
+import org.apache.iceberg.rest.restrictions.ReadRestrictions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+/** Converts validated restricted Rego programs to closed Iceberg read
restrictions. */
+public final class IcebergReadRestrictionConverter {
+
+ private IcebergReadRestrictionConverter() {}
+
+ /** Trusted request context used while binding a restriction. */
+ public interface BindingContext {
+
+ /**
+ * Returns the authenticated effective user name.
+ *
+ * @return non-empty effective user name
+ */
+ String sessionUser();
+
+ /**
+ * Resolves exact flat membership in the current metalake.
+ *
+ * <p>An empty result means that the group is unknown or membership could
not be resolved. The
+ * converter treats that result as an error rather than non-membership.
+ *
+ * @param group exact group name from the policy expression
+ * @return membership when the group and identity snapshot were resolved
+ */
+ Optional<Boolean> groupMembership(String group);
+ }
+
+ /**
+ * Binds parsed policy programs to a request context and table schema.
+ *
+ * <p>The returned restriction contains no symbolic column names or context
functions. Constant
+ * true row filters are omitted, while constant false filters are retained.
+ *
+ * @param rowFilter optional effective row-filter program
+ * @param columnMasks effective column-mask programs keyed by top-level
Iceberg field ID
+ * @param schema concrete Iceberg table schema
+ * @param context trusted request context
+ * @return closed Iceberg read restrictions
+ * @throws IllegalArgumentException if any applicable restriction cannot be
bound exactly
+ */
+ public static ReadRestrictions convert(
+ Optional<RowFilter> rowFilter,
+ Map<Integer, ColumnMask> columnMasks,
+ Schema schema,
+ BindingContext context) {
+ Objects.requireNonNull(rowFilter, "rowFilter cannot be null");
+ Objects.requireNonNull(columnMasks, "columnMasks cannot be null");
+ Objects.requireNonNull(schema, "schema cannot be null");
+ ResolutionContext resolutionContext = new ResolutionContext(context);
+
+ Expression resolvedRowFilter = null;
+ if (rowFilter.isPresent()) {
+ RowFilter filter = rowFilter.get();
+ filter.validate();
+ ResolvedPredicate resolved = resolvePredicate(filter.lower(), schema,
resolutionContext);
+ if (resolved.expression().op() != Expression.Operation.TRUE) {
+ resolvedRowFilter = resolved.expression();
+ }
+ }
+
+ List<Map.Entry<Integer, ColumnMask>> orderedMasks = new
ArrayList<>(columnMasks.entrySet());
+ for (Map.Entry<Integer, ColumnMask> entry : orderedMasks) {
+ if (entry.getKey() == null || entry.getValue() == null) {
+ throw new IllegalArgumentException("column mask field ID and program
cannot be null");
+ }
+ }
+ orderedMasks.sort(Map.Entry.comparingByKey());
+
+ List<IcebergFunction<?, ?>> projections = new ArrayList<>();
+ for (Map.Entry<Integer, ColumnMask> entry : orderedMasks) {
+ Types.NestedField field = topLevelField(schema, entry.getKey());
+ MaskAction action = resolveMaskAction(entry.getValue(), field, schema,
resolutionContext);
+ projections.add(toIcebergFunction(action, field));
+ }
+
+ ReadRestrictions restrictions = ReadRestrictions.of(resolvedRowFilter,
projections);
+ restrictions.validate(Collections.singletonMap(schema.schemaId(), schema));
+ return restrictions;
+ }
+
+ private static ResolvedPredicate resolvePredicate(
+ CanonicalExpression expression, Schema schema, ResolutionContext
context) {
+ expression.validate();
+ if (expression instanceof Literal) {
+ Literal literal = (Literal) expression;
+ if (literal.literalType() != LiteralType.BOOLEAN) {
+ throw new IllegalArgumentException("row-filter literal must be
Boolean");
+ }
+ return constant((Boolean) literal.value());
+ }
+ if (expression instanceof GroupMembership) {
+ return constant(resolveGroup((GroupMembership) expression, context));
+ }
+ if (expression instanceof Comparison) {
+ return resolveComparison((Comparison) expression, schema, context);
+ }
+ if (expression instanceof Not) {
+ ResolvedPredicate child = resolvePredicate(((Not) expression).operand(),
schema, context);
+ if (child.expression().op() == Expression.Operation.TRUE) {
+ return constant(false);
+ }
+ if (child.expression().op() == Expression.Operation.FALSE) {
+ return constant(true);
+ }
+ return resolved(Expressions.not(child.expression()));
+ }
+ if (expression instanceof LogicalExpression) {
+ return resolveLogical((LogicalExpression) expression, schema, context);
+ }
+ throw new IllegalArgumentException(
+ "Unsupported predicate node after restricted Rego parsing: "
+ + expression.getClass().getSimpleName());
+ }
+
+ private static ResolvedPredicate resolveLogical(
+ LogicalExpression logical, Schema schema, ResolutionContext context) {
+ List<ResolvedPredicate> resolvedChildren = new ArrayList<>();
+ for (CanonicalExpression operand : logical.operands()) {
+ resolvedChildren.add(resolvePredicate(operand, schema, context));
+ }
+
+ Map<String, ResolvedPredicate> children = new TreeMap<>();
+ boolean absorbingConstant = false;
+ for (ResolvedPredicate child : resolvedChildren) {
+ if (logical.operator() == LogicalOperator.AND) {
+ if (child.expression().op() == Expression.Operation.FALSE) {
+ absorbingConstant = true;
+ continue;
+ }
+ if (child.expression().op() == Expression.Operation.TRUE) {
+ continue;
+ }
+ } else {
+ if (child.expression().op() == Expression.Operation.TRUE) {
+ absorbingConstant = true;
+ continue;
+ }
+ if (child.expression().op() == Expression.Operation.FALSE) {
+ continue;
+ }
+ }
+ children.put(child.signature(), child);
+ }
+
+ if (absorbingConstant) {
+ return constant(logical.operator() == LogicalOperator.OR);
+ }
+
+ if (children.isEmpty()) {
+ return constant(logical.operator() == LogicalOperator.AND);
+ }
+ if (children.size() == 1) {
+ return children.values().iterator().next();
+ }
+
+ Expression combined = null;
+ for (ResolvedPredicate child : children.values()) {
+ if (combined == null) {
+ combined = child.expression();
+ } else if (logical.operator() == LogicalOperator.AND) {
+ combined = Expressions.and(combined, child.expression());
+ } else {
+ combined = Expressions.or(combined, child.expression());
+ }
+ }
+ return resolved(Objects.requireNonNull(combined, "logical expression
cannot be empty"));
+ }
+
+ private static ResolvedPredicate resolveComparison(
+ Comparison comparison, Schema schema, ResolutionContext context) {
+ CanonicalExpression left = comparison.left();
+ CanonicalExpression right = comparison.right();
+ ComparisonOperator operator = comparison.operator();
+
+ if (left instanceof SessionUser) {
+ return resolveSessionComparison(operator, right, true, schema, context);
+ }
+ if (right instanceof SessionUser) {
+ return resolveSessionComparison(operator, left, false, schema, context);
+ }
+ if (left instanceof Column) {
+ return resolveColumnComparison((Column) left, operator, right, schema,
false);
+ }
+ if (right instanceof Column) {
+ if (operator == ComparisonOperator.IN) {
+ throw new IllegalArgumentException("in requires the column on the
left");
+ }
+ return resolveColumnComparison((Column) right, invert(operator), left,
schema, false);
+ }
+ throw new IllegalArgumentException("comparison must contain a column or
session_user()");
+ }
+
+ private static ResolvedPredicate resolveSessionComparison(
+ ComparisonOperator operator,
+ CanonicalExpression other,
+ boolean sessionOnLeft,
+ Schema schema,
+ ResolutionContext context) {
+ String sessionUser = context.sessionUser();
+ if (other instanceof Column) {
+ ComparisonOperator normalized = sessionOnLeft ? invert(operator) :
operator;
+ return resolveColumnComparison(
+ (Column) other, normalized, SourceValue.sessionUser(sessionUser),
schema, true);
+ }
+ if (other instanceof Literal) {
+ Literal literal = (Literal) other;
+ if (literal.literalType() != LiteralType.STRING
+ || (operator != ComparisonOperator.EQ && operator !=
ComparisonOperator.NEQ)) {
+ throw new IllegalArgumentException(
+ "session_user() supports only equality with a string literal or
column");
+ }
+ boolean equal = sessionUser.equals(literal.value());
+ return constant(operator == ComparisonOperator.EQ ? equal : !equal);
+ }
+ if (other instanceof LiteralArray && sessionOnLeft && operator ==
ComparisonOperator.IN) {
+ LiteralArray values = (LiteralArray) other;
+ if (values.elementType() != LiteralType.STRING) {
+ throw new IllegalArgumentException("session_user() in requires a
string array");
+ }
+ boolean matched = false;
+ for (Literal value : values.values()) {
+ matched |= sessionUser.equals(value.value());
+ }
+ return constant(matched);
+ }
+ throw new IllegalArgumentException(
+ "Unsupported session_user() comparison after restricted Rego parsing");
+ }
+
+ private static ResolvedPredicate resolveColumnComparison(
+ Column column,
+ ComparisonOperator operator,
+ CanonicalExpression other,
+ Schema schema,
+ boolean sessionValue) {
+ if (other instanceof Literal) {
+ return resolveColumnComparison(
+ column, operator, SourceValue.literal((Literal) other), schema,
sessionValue);
+ }
+ if (other instanceof LiteralArray) {
+ if (operator != ComparisonOperator.IN) {
+ throw new IllegalArgumentException("literal arrays are valid only with
in");
+ }
+ return resolveColumnIn(column, (LiteralArray) other, schema);
+ }
+ throw new IllegalArgumentException("column comparison requires a literal
or session_user()");
+ }
+
+ private static ResolvedPredicate resolveColumnComparison(
+ Column column,
+ ComparisonOperator operator,
+ SourceValue sourceValue,
+ Schema schema,
+ boolean sessionValue) {
+ Types.NestedField field = topLevelField(schema, column.name());
+ if (sessionValue && operator != ComparisonOperator.EQ && operator !=
ComparisonOperator.NEQ) {
+ throw new IllegalArgumentException(
+ "session_user() supports only equality with a string column");
+ }
+ if (sourceValue.literalType() == LiteralType.NULL) {
+ if (operator != ComparisonOperator.EQ && operator !=
ComparisonOperator.NEQ) {
+ throw new IllegalArgumentException("null supports only equality
comparisons");
+ }
+ if (field.isRequired()) {
+ throw new IllegalArgumentException(
+ "null comparison requires an optional field: " + field.name());
+ }
+ Expression unbound =
+ operator == ComparisonOperator.EQ
+ ? Expressions.isNull(field.name())
+ : Expressions.notNull(field.name());
+ return resolved(bind(schema, field, unbound));
+ }
+
+ validateOperator(field.type(), operator);
+ Object value = typedValue(sourceValue, field.type());
+ Expression unbound = predicate(field.name(), operator, value);
+ return resolved(bind(schema, field, unbound));
+ }
+
+ private static ResolvedPredicate resolveColumnIn(
+ Column column, LiteralArray array, Schema schema) {
+ Types.NestedField field = topLevelField(schema, column.name());
+ validateOperator(field.type(), ComparisonOperator.IN);
+
+ Map<String, Object> values = new TreeMap<>();
+ for (Literal literal : array.values()) {
+ Object value = typedValue(SourceValue.literal(literal), field.type());
+ values.put(canonicalValue(value), value);
+ }
+ Expression unbound = Expressions.in(field.name(), values.values());
+ return resolved(bind(schema, field, unbound));
+ }
+
+ private static Expression predicate(String fieldName, ComparisonOperator
operator, Object value) {
+ switch (operator) {
+ case EQ:
+ return Expressions.equal(fieldName, value);
+ case NEQ:
+ return Expressions.notEqual(fieldName, value);
+ case LT:
+ return Expressions.lessThan(fieldName, value);
+ case LTE:
+ return Expressions.lessThanOrEqual(fieldName, value);
+ case GT:
+ return Expressions.greaterThan(fieldName, value);
+ case GTE:
+ return Expressions.greaterThanOrEqual(fieldName, value);
+ default:
+ throw new IllegalArgumentException("Unsupported scalar comparison
operator: " + operator);
+ }
+ }
+
+ private static Expression bind(
+ Schema schema, Types.NestedField expectedField, Expression expression) {
+ Types.NestedField resolvedByIceberg =
schema.findField(expectedField.name());
+ if (resolvedByIceberg == null || resolvedByIceberg.fieldId() !=
expectedField.fieldId()) {
+ throw new IllegalArgumentException(
+ "Column name cannot be bound unambiguously as a top-level field: "
+ + expectedField.name());
+ }
+ Expression bound = Binder.bind(schema.asStruct(), expression, true);
+ if (!Binder.boundReferences(schema.asStruct(), bound, true)
+ .equals(Collections.singleton(expectedField.fieldId()))) {
+ throw new IllegalArgumentException(
+ "Bound predicate did not preserve field ID " +
expectedField.fieldId());
+ }
+ return bound;
+ }
+
+ private static void validateOperator(Type type, ComparisonOperator operator)
{
+ switch (type.typeId()) {
+ case BOOLEAN:
+ if (operator != ComparisonOperator.EQ
+ && operator != ComparisonOperator.NEQ
+ && operator != ComparisonOperator.IN) {
+ throw new IllegalArgumentException("Boolean fields do not support
ordering predicates");
+ }
+ return;
+ case INTEGER:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case DECIMAL:
+ case DATE:
+ case TIME:
+ case STRING:
+ return;
+ case TIMESTAMP:
+ if (((Types.TimestampType) type).shouldAdjustToUTC()) {
+ throw new IllegalArgumentException(
+ "Timestamp-with-timezone fields are not supported by read
restrictions");
+ }
+ return;
+ default:
+ throw new IllegalArgumentException("Unsupported read-restriction field
type: " + type);
+ }
+ }
+
+ private static Object typedValue(SourceValue source, Type type) {
+ switch (type.typeId()) {
+ case BOOLEAN:
+ requireLiteralType(source, LiteralType.BOOLEAN, type);
+ return source.value();
+ case INTEGER:
+ requireLiteralType(source, LiteralType.NUMBER, type);
+ try {
+ return ((BigDecimal) source.value()).intValueExact();
+ } catch (ArithmeticException exception) {
+ throw invalidLiteral(source, type, exception);
+ }
+ case LONG:
+ requireLiteralType(source, LiteralType.NUMBER, type);
+ try {
+ return ((BigDecimal) source.value()).longValueExact();
+ } catch (ArithmeticException exception) {
+ throw invalidLiteral(source, type, exception);
+ }
+ case FLOAT:
+ requireLiteralType(source, LiteralType.NUMBER, type);
+ return exactFloat(source);
+ case DOUBLE:
+ requireLiteralType(source, LiteralType.NUMBER, type);
+ return exactDouble(source);
+ case DECIMAL:
+ requireLiteralType(source, LiteralType.NUMBER, type);
+ return exactDecimal(source, (Types.DecimalType) type);
+ case STRING:
+ requireLiteralType(source, LiteralType.STRING, type);
+ return source.value();
+ case DATE:
+ requireLiteralType(source, LiteralType.STRING, type);
+ validateDate((String) source.value(), type);
+ return source.value();
+ case TIME:
+ requireLiteralType(source, LiteralType.STRING, type);
+ validateTime((String) source.value(), type);
+ return source.value();
+ case TIMESTAMP:
+ if (((Types.TimestampType) type).shouldAdjustToUTC()) {
+ throw new IllegalArgumentException(
+ "Timestamp-with-timezone fields are not supported by read
restrictions");
+ }
+ requireLiteralType(source, LiteralType.STRING, type);
+ validateTimestamp((String) source.value(), type);
+ return source.value();
+ default:
+ throw new IllegalArgumentException("Unsupported read-restriction field
type: " + type);
+ }
+ }
+
+ private static Float exactFloat(SourceValue source) {
+ BigDecimal decimal = (BigDecimal) source.value();
+ float value = decimal.floatValue();
+ if (!Float.isFinite(value) || new BigDecimal((double)
value).compareTo(decimal) != 0) {
+ throw invalidLiteral(source, Types.FloatType.get(), null);
+ }
+ return source.negativeZero() ? -0.0F : value;
+ }
+
+ private static Double exactDouble(SourceValue source) {
+ BigDecimal decimal = (BigDecimal) source.value();
+ double value = decimal.doubleValue();
+ if (!Double.isFinite(value) || new BigDecimal(value).compareTo(decimal) !=
0) {
+ throw invalidLiteral(source, Types.DoubleType.get(), null);
+ }
+ return source.negativeZero() ? -0.0D : value;
+ }
+
+ private static BigDecimal exactDecimal(SourceValue source, Types.DecimalType
type) {
+ BigDecimal value;
+ try {
+ value = ((BigDecimal) source.value()).setScale(type.scale(),
RoundingMode.UNNECESSARY);
+ } catch (ArithmeticException exception) {
+ throw invalidLiteral(source, type, exception);
+ }
+ if (value.precision() > type.precision()) {
+ throw invalidLiteral(source, type, null);
+ }
+ return value;
+ }
+
+ private static void validateDate(String value, Type type) {
+ try {
+ LocalDate.parse(value);
+ } catch (DateTimeParseException exception) {
+ throw invalidLiteral(SourceValue.string(value), type, exception);
+ }
+ }
+
+ private static void validateTime(String value, Type type) {
+ try {
+ LocalTime parsed = LocalTime.parse(value);
+ if (!parsed.equals(parsed.truncatedTo(ChronoUnit.MICROS))) {
+ throw invalidLiteral(SourceValue.string(value), type, null);
+ }
+ } catch (DateTimeParseException exception) {
+ throw invalidLiteral(SourceValue.string(value), type, exception);
+ }
+ }
+
+ private static void validateTimestamp(String value, Type type) {
+ try {
+ LocalDateTime parsed = LocalDateTime.parse(value);
+ if (!parsed.equals(parsed.truncatedTo(ChronoUnit.MICROS))) {
+ throw invalidLiteral(SourceValue.string(value), type, null);
+ }
+ } catch (DateTimeParseException exception) {
+ throw invalidLiteral(SourceValue.string(value), type, exception);
+ }
+ }
+
+ private static void requireLiteralType(SourceValue source, LiteralType
expected, Type fieldType) {
+ if (source.literalType() != expected) {
+ throw invalidLiteral(source, fieldType, null);
+ }
+ }
+
+ private static IllegalArgumentException invalidLiteral(
+ SourceValue source, Type type, Exception cause) {
+ String message =
+ String.format(
+ "Literal %s cannot be represented exactly as Iceberg %s",
source.value(), type);
+ return cause == null
+ ? new IllegalArgumentException(message)
+ : new IllegalArgumentException(message, cause);
+ }
+
+ private static boolean resolveGroup(GroupMembership membership,
ResolutionContext context) {
+ return context.groupMembership(membership.group());
+ }
+
+ private static MaskAction resolveMaskAction(
+ ColumnMask mask, Types.NestedField field, Schema schema,
ResolutionContext context) {
+ mask.validate();
+ List<Boolean> branchMatches = new ArrayList<>();
+ for (MaskBranch branch : mask.branches()) {
+ toIcebergFunction(branch.action(), field);
+ ResolvedPredicate condition = resolvePredicate(branch.condition(),
schema, context);
+ if (condition.expression().op() != Expression.Operation.TRUE
+ && condition.expression().op() != Expression.Operation.FALSE) {
+ throw new IllegalArgumentException("column-mask condition did not
resolve to a constant");
+ }
+ branchMatches.add(condition.expression().op() ==
Expression.Operation.TRUE);
+ }
+ toIcebergFunction(mask.fallback(), field);
+
+ for (int index = 0; index < mask.branches().size(); index++) {
+ if (branchMatches.get(index)) {
+ return mask.branches().get(index).action();
+ }
+ }
+ return mask.fallback();
+ }
+
+ private static IcebergFunction<?, ?> toIcebergFunction(
+ MaskAction action, Types.NestedField field) {
+ IcebergFunction<?, ?> function;
+ switch (action) {
+ case MASK_ALPHANUM:
+ function = IcebergFunctions.maskAlphanum(field.fieldId());
+ break;
+ case MASK_TO_FIXED_VALUE:
+ function = IcebergFunctions.maskToFixedValue(field.fieldId());
+ break;
+ case REPLACE_WITH_NULL:
+ if (field.isRequired()) {
+ throw new IllegalArgumentException(
+ "replace-with-null is invalid for required field " +
field.fieldId());
+ }
Review Comment:
[Important] A column-mask policy attached through a tag on the *table* (or
schema) is inherited by every column, so one incompatible column makes the
whole table unloadable.
`Policy.BuiltInType.COLUMN_MASK` is documented as applying to "tagged tables
or tagged columns"
(`api/src/main/java/org/apache/gravitino/policy/Policy.java:62-63`). For each
column, `ExperimentalIcebergReadRestrictionLoader.load` asks for the column's
effective policies
(`...ExperimentalIcebergReadRestrictionLoader.java:102-103`), and
`EffectiveTagResolver.resolve` walks the column *and its ancestors*
(`core/src/main/java/org/apache/gravitino/tag/EffectiveTagResolver.java:67-90`,
inherited tags at line 76); `ObjectPolicyResolver.resolve` does not filter by
the object type the policy was meant for. So a table-level mask tag yields the
same `ColumnMask` program for all top-level fields.
Concrete failure: a table-level tag carrying `mask :=
action("replace-with-null")` on a schema like the one in the new unit test
(`id` is `required`) hits this line and throws, and `resolveForLoad` turns it
into `IllegalStateException`
(`...ExperimentalIcebergReadRestrictionLoader.java:154-158`), which
`IcebergExceptionMapper.getErrorCode` does not map, so load-table answers HTTP
500 for every reader of that table - not just for the offending column. The
`canBind` guard at line 616-621 has the same shape (for example an alphanumeric
or show-last-4 mask inherited onto an `int` column).
Suggestion: decide per field. Either skip (and log) a mask that cannot bind
to that field's type, or restrict table/schema-level mask inheritance to
compatible fields, so one unsuitable column cannot take the whole table
offline. If failing closed for the entire table really is intended, it deserves
an explicit comment plus a test, because the current behaviour turns a
plausible admin action into a total read outage.
Verified by: reading the inheritance chain in
EffectiveTagResolver/ObjectPolicyResolver/MetadataObjectUtil.getParentMetadataObjects
(COLUMN -> TABLE -> SCHEMA -> CATALOG -> METALAKE) and the error mapping table
in IcebergExceptionMapper.java:60-88; no test in this PR covers a table-level
mask tag.
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java:
##########
@@ -343,14 +346,20 @@ public Response loadTable(
LoadTableResponse loadTableResponse =
tableOperationDispatcher.loadTable(context, tableIdentifier);
+ if
(IcebergRESTUtils.SnapshotMode.REFS.getValue().equals(snapshots)) {
+ loadTableResponse = filterSnapshotsByRefs(loadTableResponse);
+ }
+ IcebergReadRestrictionLoader.Result restrictionResult =
+ readRestrictionLoader.load(context, tableIdentifier,
loadTableResponse);
+ loadTableResponse = restrictionResult.response();
Optional<EntityTag> etag =
-
generateETag(loadTableResponse.tableMetadata().metadataFileLocation(),
snapshots);
+ generateETag(
+ loadTableResponse.tableMetadata().metadataFileLocation(),
+ snapshots,
+ restrictionResult.etagSeed());
Review Comment:
[Nit] The documented "ETags from create/update/register match the default
loadTable ETag" invariant no longer holds for restricted users.
`IcebergRESTUtils.buildResponseWithETag(LoadTableResponse)` states exactly
that invariant and computes the ETag without a representation seed
(`iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java:346-358`),
which `createTable` (line 203) and `updateTable` (line 246) use. With a seed
added here, a restricted client that keeps the ETag from a create or update
response and sends it as `If-None-Match` on load-table always gets a full 200 -
correct, but a silently lost cache hit. Either fix up that javadoc or seed
those responses the same way.
Verified by: reading both ETag construction paths at the reviewed commit;
`generateETag` has no other non-test caller (grep for `generateETag` outside
tests).
##########
iceberg/iceberg-rest-server-experiment/src/main/java/org/apache/gravitino/iceberg/service/restriction/ExperimentalIcebergReadRestrictionLoader.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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.service.restriction;
+
+import java.security.Principal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.UserGroup;
+import org.apache.gravitino.UserPrincipal;
+import org.apache.gravitino.authorization.AccessControlDispatcher;
+import org.apache.gravitino.exceptions.NoSuchGroupException;
+import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
+import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.ColumnMaskContent;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyDispatcher;
+import org.apache.gravitino.policy.RowFilterContent;
+import org.apache.gravitino.policy.rego.RestrictedRegoExpressionParserFacade;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.ColumnMask;
+import org.apache.gravitino.policy.rego.RestrictedRegoProgram.RowFilter;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.PrincipalUtils;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.restrictions.ReadRestrictions;
+import org.apache.iceberg.rest.restrictions.ReadRestrictionsParser;
+import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Loads tag-selected Gravitino policies into experimental Iceberg read
restrictions. */
+public final class ExperimentalIcebergReadRestrictionLoader
+ implements IcebergReadRestrictionLoader {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ExperimentalIcebergReadRestrictionLoader.class);
+
+ /** Creates the service-loaded read-restriction loader. */
+ public ExperimentalIcebergReadRestrictionLoader() {}
+
+ @Override
+ public Result load(
+ IcebergRequestContext context, TableIdentifier tableIdentifier,
LoadTableResponse response) {
+ GravitinoEnv environment = GravitinoEnv.getInstance();
+ PolicyDispatcher policyDispatcher = environment.internalPolicyDispatcher();
+ if (policyDispatcher == null) {
+ return Result.of(response, Optional.empty());
+ }
+
+ String metalake = IcebergRESTServerContext.getInstance().metalakeName();
+ NameIdentifier tableName =
+ IcebergIdentifierUtils.toGravitinoTableIdentifier(
+ metalake,
+ context.catalogName(),
+ tableIdentifier,
+ HierarchicalSchemaUtil.schemaSeparator());
+ Schema schema = response.tableMetadata().schema();
+ PolicyEntity[] rowPolicies =
+ policyDispatcher.listPolicyInfosForMetadataObject(
+ metalake, NameIdentifierUtil.toMetadataObject(tableName,
Entity.EntityType.TABLE));
+ Map<Integer, PolicyEntity[]> maskPolicies = new LinkedHashMap<>();
+ for (Types.NestedField field : schema.columns()) {
+ NameIdentifier columnName =
+ NameIdentifierUtil.ofColumn(
+ metalake,
+ context.catalogName(),
+ tableName.namespace().level(2),
+ tableName.name(),
+ field.name());
+ MetadataObject column =
+ NameIdentifierUtil.toMetadataObject(columnName,
Entity.EntityType.COLUMN);
+ maskPolicies.put(
+ field.fieldId(),
policyDispatcher.listPolicyInfosForMetadataObject(metalake, column));
+ }
+
+ IdentityBindingContext bindingContext =
+ new IdentityBindingContext(
+ context.userName(), metalake,
environment.internalAccessControlDispatcher());
+ ReadRestrictions restrictions =
+ resolveForLoad(rowPolicies, maskPolicies, schema, bindingContext);
+ if (restrictions.isEmpty()) {
+ return Result.of(response, Optional.empty());
+ }
+
+ LoadTableResponse.Builder builder =
+ LoadTableResponse.builder()
+ .withTableMetadata(response.tableMetadata())
+ .withReadRestrictions(restrictions);
+ if (response.config() != null) {
+ builder.addAllConfig(response.config());
+ }
+ if (response.credentials() != null) {
+ builder.addAllCredentials(response.credentials());
+ }
+ if (response.remoteSigningConfig() != null) {
+ builder.withRemoteSigningConfig(response.remoteSigningConfig());
+ }
+ if (response.labels() != null) {
+ builder.withLabels(response.labels());
+ }
+
+ String signature = ReadRestrictionsParser.toJson(restrictions);
+ return Result.of(builder.build(), Optional.of(signature));
+ }
+
+ static ReadRestrictions resolve(
+ PolicyEntity[] tablePolicies,
+ Map<Integer, PolicyEntity[]> columnPolicies,
+ Schema schema,
+ IcebergReadRestrictionConverter.BindingContext bindingContext) {
+ Optional<RowFilter> rowFilter = resolveRowFilter(tablePolicies, schema,
bindingContext);
+ Map<Integer, ColumnMask> columnMasks =
+ resolveColumnMasks(columnPolicies, schema, bindingContext);
+ return IcebergReadRestrictionConverter.convert(rowFilter, columnMasks,
schema, bindingContext);
+ }
+
+ static ReadRestrictions resolveForLoad(
+ PolicyEntity[] tablePolicies,
+ Map<Integer, PolicyEntity[]> columnPolicies,
+ Schema schema,
+ IcebergReadRestrictionConverter.BindingContext bindingContext) {
+ try {
+ return resolve(tablePolicies, columnPolicies, schema, bindingContext);
+ } catch (RuntimeException exception) {
+ LOG.error("Failed to resolve Iceberg read restrictions", exception);
+ throw new IllegalStateException(
+ "Unable to resolve read restrictions for the requested table");
+ }
+ }
+
+ private static Optional<RowFilter> resolveRowFilter(
+ PolicyEntity[] policies,
+ Schema schema,
+ IcebergReadRestrictionConverter.BindingContext bindingContext) {
+ Map<String, RowFilter> distinct = new LinkedHashMap<>();
+ for (PolicyEntity policy : policies) {
+ if (policy.policyType() != Policy.BuiltInType.ROW_FILTER) {
+ continue;
+ }
+ RowFilter program =
+ RestrictedRegoExpressionParserFacade.parseRowFilter(
+ ((RowFilterContent) policy.content()).expression());
+ ReadRestrictions resolved =
+ IcebergReadRestrictionConverter.convert(
+ Optional.of(program), Collections.emptyMap(), schema,
bindingContext);
+ distinct.putIfAbsent(ReadRestrictionsParser.toJson(resolved), program);
+ }
+ if (distinct.size() > 1) {
+ throw new IllegalArgumentException(
+ "Conflicting row-filter policies apply to the requested table");
+ }
Review Comment:
[Question] Two row-filter policies that reach the same table make it
unreadable for everyone - is a hard failure the intended end state?
Because effective tags are inherited from the schema, catalog and metalake,
a broad tag (say a metalake-wide `filter := true`) landing on a table that also
has its own filter is enough to trip this, which the new test documents as
intended (`...TestExperimentalIcebergReadRestrictionLoader.java:138-152`). The
result is `IllegalStateException` from `resolveForLoad` -> HTTP 500 on every
load of that table, with only "Unable to resolve read restrictions for the
requested table" on the wire. Most row-filter systems combine instead (AND for
defence in depth, OR for grants).
If the conflict is deliberate for the first iteration, could the permissive
case at least be excluded (a filter that resolves to constant TRUE adds no
restriction, so it need not conflict with anything), and the failure be
surfaced as a distinguishable status rather than a bare 500? Either way it is
worth calling out in the docs, since the blast radius is "table cannot be read
by anyone".
Verified by: reading `resolveRowFilter` plus `resolveForLoad` (lines
147-183) and the error-code table in `IcebergExceptionMapper.java:60-88`
(`IllegalStateException` is unmapped, so 500); behaviour matches the new unit
tests.
--
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]
