flyrain commented on code in PR #16131: URL: https://github.com/apache/iceberg/pull/16131#discussion_r3777303719
########## core/src/main/java/org/apache/iceberg/rest/BaseRESTTable.java: ########## @@ -0,0 +1,63 @@ +/* + * 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.iceberg.rest; + +import java.util.Optional; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.SupportsReadRestrictions; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.metrics.MetricsReporter; +import org.apache.iceberg.rest.restrictions.ReadRestrictions; + +/** + * BaseTable specialization for tables loaded via a REST catalog. Carries the per-principal {@link + * ReadRestrictions} that the REST server may have attached to the load response and advertises the + * capability via {@link SupportsReadRestrictions}. + * + * <p>Used by {@link RESTSessionCatalog} for REST loadTable paths that do not use server-side scan + * planning. {@link RESTTable} extends this class to add scan-planning. Non-REST catalogs (Hadoop, Review Comment: Could we avoid introducing `BaseRESTTable` here? It appears to exist mainly because the current `RESTTable` is specifically the server-side scan-planning implementation, while the local-planning path still needs to carry read restrictions. The resulting names are a little misleading: `BaseRESTTable` is actually the general REST-loaded table, while `RESTTable` is the specialized one. - option 1, would it be clearer to make `RESTTable` the common implementation that carries restrictions, and move the server-side planning behavior into a subclass such as `RESTScanPlanningTable`? We may have to go through deprecation process if if `RESTTable` has been released. - option 2, a single `RESTTable` could select the scan implementation in `newScan()`. The current hierarchy works, but the extra layer and naming make the distinction between “loaded through REST” and “uses REST scan planning” difficult to understand. ########## data/src/main/java/org/apache/iceberg/data/IcebergGenerics.java: ########## @@ -103,7 +108,16 @@ public ScanBuilder metricsReporter(MetricsReporter reporter) { } public CloseableIterable<Record> build() { - return new TableScanIterable(tableScan, reuseContainers); + Optional<ReadRestrictions> restrictions = TableUtil.readRestrictions(table); + if (restrictions.isPresent() && restrictions.get().rowFilter() != null) { + this.tableScan = tableScan.filter(restrictions.get().rowFilter()); + } + + CloseableIterable<Record> records = new TableScanIterable(tableScan, reuseContainers); + if (restrictions.isPresent()) { + records = ReadRestrictionsApplier.apply(records, restrictions.get(), tableScan.schema()); Review Comment: If `ReadRestrictionsApplier.apply()` throws while binding the restrictions, could this leak the resources already acquired by `TableScanIterable`? For example, callers normally rely on try-with-resources: ```java try (CloseableIterable<Record> records = reader.build()) { // read records } ``` However, `TableScanIterable` calls `scan.planTasks()` before `apply()` runs. If `apply()` throws, `build()` never returns, so the caller never receives `records` and the try-with-resources block cannot close it. Should `build()` close `records` when `apply()` fails, or should restriction binding happen before constructing `TableScanIterable`? ########## core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java: ########## @@ -557,13 +558,22 @@ public Table loadTable(SessionContext context, TableIdentifier identifier) { } List<Credential> credentials = response.credentials(); + ReadRestrictions readRestrictions = response.readRestrictions(); RESTClient tableClient = client.withAuthSession(tableSession); Supplier<BaseTable> tableSupplier = createTableSupplier( - finalIdentifier, tableMetadata, context, tableClient, tableConf, credentials); + finalIdentifier, + tableMetadata, + context, + tableClient, + tableConf, + credentials, + readRestrictions); String eTag = responseHeaders.getOrDefault(HttpHeaders.ETAG, null); - if (eTag != null) { + if (eTag != null && readRestrictions.isEmpty()) { Review Comment: The case which may yield unexpected results: 1. The client loads a table with no restrictions and caches it with ETag `A`. 2. The server later adds read restrictions, but the table metadata does not change. 3. The client sends `If-None-Match: A`. 4. If the ETag represents only table metadata, the server may return `304`, causing the client to reuse the unrestricted cached table without seeing the new policy. I think the cleaner solution may be to strengthen the ETag contract rather than handle this only in the client. This is probably out-of-scope of this PR. Could we require the ETag to cover the complete `LoadTableResponse`, including `read-restrictions`, etc? Then any policy change would produce a different ETag, and `304` would mean that the entire effective response, not just the table metadata, is unchanged. This requirement should probably be made explicit in the spec. ########## core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java: ########## @@ -1009,13 +1030,17 @@ public Table create() { trackFileIO(ops); - RESTTable restTable = restTableForScanPlanning(ops, ident, tableClient, tableConf); + RESTTable restTable = + restTableForScanPlanning(ops, ident, tableClient, tableConf, ReadRestrictions.empty()); Review Comment: Same here. ########## core/src/main/java/org/apache/iceberg/rest/restrictions/ReadRestrictionsParser.java: ########## @@ -0,0 +1,99 @@ +/* + * 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.iceberg.rest.restrictions; + +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonNode; +import java.io.IOException; +import java.util.List; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.ExpressionParser; +import org.apache.iceberg.functions.IcebergFunction; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.util.JsonUtil; + +public class ReadRestrictionsParser { + + private ReadRestrictionsParser() {} + + private static final String REQUIRED_ROW_FILTER = "required-row-filter"; + private static final String REQUIRED_COLUMN_PROJECTIONS = "required-column-projections"; + + public static String toJson(ReadRestrictions restrictions) { + return toJson(restrictions, false); + } + + public static String toJson(ReadRestrictions restrictions, boolean pretty) { + return JsonUtil.generate(gen -> toJson(restrictions, gen), pretty); + } + + public static void toJson(ReadRestrictions restrictions, JsonGenerator generator) + throws IOException { + Preconditions.checkArgument(restrictions != null, "Invalid read restrictions: null"); + + generator.writeStartObject(); + + if (restrictions.rowFilter() != null) { + generator.writeFieldName(REQUIRED_ROW_FILTER); + ExpressionParser.toJson(restrictions.rowFilter(), generator); + } + + if (!restrictions.columnProjections().isEmpty()) { + generator.writeArrayFieldStart(REQUIRED_COLUMN_PROJECTIONS); + for (IcebergFunction<?, ?> action : restrictions.columnProjections()) { + ActionParser.toJson(action, generator); + } + generator.writeEndArray(); + } + + generator.writeEndObject(); + } + + public static ReadRestrictions fromJson(String json) { + return JsonUtil.parse(json, ReadRestrictionsParser::fromJson); + } + + public static ReadRestrictions fromJson(JsonNode node) { + if (node == null || node.isNull()) { + return ReadRestrictions.empty(); + } + Preconditions.checkArgument( + node.isObject(), "Cannot parse read restrictions from non-object value: %s", node); + + Expression rowFilter = null; + if (node.hasNonNull(REQUIRED_ROW_FILTER)) { + rowFilter = ExpressionParser.fromJson(node.get(REQUIRED_ROW_FILTER)); Review Comment: Is the intention to update this implementation to the field-ID-based expression format once #13879 lands? The current code and tests use the existing name-based ExpressionParser representation, which does not appear to support the proposed left/right and ID-reference wire format. I assume the change to support new format should happen in the class `ExpressionParser`. WDYT? ########## core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java: ########## @@ -740,13 +757,17 @@ public Table registerTable( trackFileIO(ops); - RESTTable restTable = restTableForScanPlanning(ops, ident, tableClient, tableConf); + RESTTable restTable = + restTableForScanPlanning(ops, ident, tableClient, tableConf, ReadRestrictions.empty()); Review Comment: Should we also pass the read restrictions from the LoadTableResponse instead of hardcoding them to empty? They are likely empty when registering a table, but passing them through would be more flexible and robust for corner cases. ########## data/src/main/java/org/apache/iceberg/data/ReadRestrictionsApplier.java: ########## @@ -0,0 +1,174 @@ +/* + * 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.iceberg.data; + +import java.security.SecureRandom; +import java.util.List; +import java.util.Map; +import org.apache.iceberg.Schema; +import org.apache.iceberg.expressions.Evaluator; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.functions.IcebergFunction; +import org.apache.iceberg.functions.ReplaceWithNull; +import org.apache.iceberg.functions.SaltedFunction; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.rest.restrictions.ReadRestrictions; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.SerializableFunction; + +/** + * Applies server-provided {@link ReadRestrictions} (row filter + column masks) to a stream of + * {@link Record}s. + * + * <p>The row filter is evaluated per-record against the original column values before any mask is + * applied, as required by the spec: + * + * <blockquote> + * + * Row filters MUST be evaluated against the original, untransformed column values. Required + * projections MUST be applied only after row filters are applied. + * + * </blockquote> + * + * <p>Callers that also push the row filter into {@link org.apache.iceberg.TableScan#filter} get + * partition/stats-level pruning for free; this applier re-evaluates the filter at the row level so + * correctness does not depend on whether the surrounding reader honors residual evaluation. + * + * <p>Currently supports top-level fields only. Masks on nested fieldIds fail closed at bind time so + * unmasked nested data cannot leak. + * + * <p>Projections for columns that are not being read are skipped, as required by the spec: + * + * <blockquote> + * + * A reader must enforce projections on the columns it is actually reading. Projections referencing + * columns that are not being read do not apply. + * + * </blockquote> + */ +class ReadRestrictionsApplier { + + private static final SecureRandom RANDOM = new SecureRandom(); + private static final int SALT_LENGTH = 16; + + private ReadRestrictionsApplier() {} + + static CloseableIterable<Record> apply( + CloseableIterable<Record> records, ReadRestrictions restrictions, Schema projection) { + CloseableIterable<Record> filtered = filterRows(records, restrictions.rowFilter(), projection); Review Comment: Looks good to me now. We could think about pushing down in the future. For example, restriction handling can have two separate phases here: 1. Before scan planning, the row filter should be added to `TableScan` so `planTasks()` can use it for partition, manifest, file-metrics, and reader-level pruning. The row filter and column masks could also be bound and validated at this point, before any scan resources are acquired. 2. During record iteration, the bound row filter should be evaluated again against the original values, followed by the bound column masks. This preserves fail-closed row-level enforcement without depending on residual handling in the underlying reader. Conceptually, the API could look like: ```java BoundReadRestrictions bound = ReadRestrictionsApplier.bind(restrictions, tableScan.schema()); CloseableIterable<Record> records = new TableScanIterable(tableScan, reuseContainers); return bound.apply(records); ``` This keeps filter pushdown before `planTasks()`, while moving the operations that may fail during binding ahead of `TableScanIterable` construction and its resource acquisition. Not a blocker. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
