flyrain commented on code in PR #16131:
URL: https://github.com/apache/iceberg/pull/16131#discussion_r3770965088
##########
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);
Review Comment:
[P1] Enforcement currently exists only in `IcebergGenerics`. Spark, Flink,
and callers that use `table.newScan()` directly never consult
`TableUtil.readRestrictions`, while `BaseRESTTable` still exposes a normal
unrestricted scan. A REST response with restrictions can therefore return raw
rows through those readers. The spec says restrictions apply to every read
performed using the response, so please either integrate enforcement at each
supported engine/scan boundary or make unsupported scan paths fail closed.
##########
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:
[P1] This guard only prevents caching the response after restrictions have
been observed. An earlier unrestricted entry is still used to send
`If-None-Match` at the start of `loadTable`; if the server later adds
restrictions without changing table metadata/ETag, it can return 304 and the
method returns the unrestricted cached table before this line is reached.
Please make conditional loading policy-aware (or disable it for this feature)
before the cached ETag is sent, unless the protocol requires ETags to cover
`read-restrictions`.
##########
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:
[P1] This discards `response.readRestrictions()` and attaches
`ReadRestrictions.empty()` to the returned table. `Builder.create()` has the
same pattern. Because both endpoints deserialize `LoadTableResponse`, a server
can legally include restrictions and these paths will silently return an
unrestricted table. Please pass the response restrictions through all
`LoadTableResponse` paths (and cover create/register with end-to-end tests).
##########
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:
[P1] The current spec proposal encodes row-filter predicates with
`left`/`right` operands and an ID reference such as `{ "type": "reference",
"id": 14 }`. The existing `ExpressionParser` instead expects/emits name-based
`term`/`value` JSON. As a result, a response conforming to #13879 cannot be
parsed here, and the current tests pin the obsolete name-based representation.
This needs an ID-reference-aware parser (or a compatible extension to
`ExpressionParser`) and wire-level tests using the proposed schema.
##########
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:
[P2] `TableScanIterable` has already called `scan.planTasks()` before
restriction binding happens here. If `ReadRestrictionsApplier.apply` throws
while binding an unknown/unsupported/nested action or row filter, the planned
task iterable is never closed. Please bind before constructing the iterable, or
close `records` in an exception path before rethrowing.
--
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]
