laskoviymishka commented on code in PR #16131:
URL: https://github.com/apache/iceberg/pull/16131#discussion_r3684943741
##########
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:
This is the one I think questions the approach itself, so worth surfacing
for the spec discussion. Both the masks and the row filter bind against
`tableScan.schema()` — the projected output schema after the user's `.select()`
— not the full table schema.
If the server masks field 2 (`email`) but the user did `.select("id")`,
`bindMasks` can't find field 2 and throws `unknown field id: 2`, so a query
that never touches the restricted column fails hard. The row filter has the
sharper version: `tableScan.filter(rowFilter())` binds against the projected
schema too, so a filter over a projected-away column either throws or,
depending on residual handling upstream, passes every row unfiltered — a silent
leak on a security filter.
The underlying question for #13879 is *where* enforcement binds — full table
schema vs projected. Binding the filter/masks against `table.schema()` and
skipping masks for projected-away ids seems right to me, but it's a real design
fork the spec should nail down. wdyt?
##########
api/src/main/java/org/apache/iceberg/functions/MaskToFixedValue.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.functions;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.DateTimeUtil;
+import org.apache.iceberg.util.SerializableFunction;
+import org.apache.iceberg.variants.Variant;
+
+/** Returns a spec-defined fixed value for the column's type. */
+public final class MaskToFixedValue extends
IcebergFunction.BaseFunction<Object, Object> {
+
+ private static final Integer INT_DEFAULT = 0;
+ private static final Long LONG_DEFAULT = 0L;
+ private static final Float FLOAT_DEFAULT = 0.0f;
+ private static final Double DOUBLE_DEFAULT = 0.0d;
+ // Per spec: string uses "XXXXXXXX" (not "") so masked strings stay visually
distinct from
+ // legitimately empty strings. All other types use the zero value of their
representation.
+ private static final String STRING_DEFAULT = "XXXXXXXX";
+ private static final Integer DATE_DEFAULT =
DateTimeUtil.daysFromDate(LocalDate.of(1970, 1, 1));
+ private static final Long TIME_DEFAULT_MICROS = 0L;
+ private static final Long TIMESTAMP_DEFAULT_MICROS =
+ DateTimeUtil.microsFromTimestamp(LocalDateTime.of(1970, 1, 1, 0, 0));
+ private static final Long TIMESTAMP_DEFAULT_NANOS =
+ DateTimeUtil.nanosFromTimestamp(LocalDateTime.of(1970, 1, 1, 0, 0));
+ private static final UUID UUID_DEFAULT =
UUID.fromString("00000000-0000-0000-0000-000000000000");
+ private static final ByteBuffer EMPTY_BUFFER =
ByteBuffer.allocate(0).asReadOnlyBuffer();
+
+ // Empty Variant: V1 metadata with no entries + empty object value.
+ private static final Variant EMPTY_VARIANT =
+ Variant.from(
+ ByteBuffer.wrap(new byte[] {0x01, 0x00, 0x00, 0x02, 0x00, 0x00})
+ .order(ByteOrder.LITTLE_ENDIAN));
+
+ public MaskToFixedValue(int fieldId) {
+ super(fieldId);
+ }
+
+ @Override
+ public String name() {
+ return MASK_TO_FIXED_VALUE;
+ }
+
+ @Override
+ public boolean canBind(Type type) {
+ switch (type.typeId()) {
+ case BOOLEAN:
+ case INTEGER:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case STRING:
+ case DATE:
+ case TIME:
+ case TIMESTAMP:
+ case TIMESTAMP_NANO:
+ case UUID:
+ case FIXED:
+ case BINARY:
+ case DECIMAL:
+ case VARIANT:
+ case LIST:
+ case MAP:
+ case STRUCT:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ @Override
+ public SerializableFunction<Object, Object> bind(Type type) {
+ Preconditions.checkArgument(
+ canBind(type), "mask-to-fixed-value is not supported for type: %s",
type);
+ Object defaultValue = defaultValueFor(type);
+ return defaultValue instanceof ByteBuffer
+ ? new ConstantByteBufferFn((ByteBuffer) defaultValue)
+ : new ConstantFn(defaultValue);
+ }
+
+ private static Object defaultValueFor(Type type) {
+ switch (type.typeId()) {
+ case BOOLEAN:
+ return Boolean.FALSE;
+ case INTEGER:
+ return INT_DEFAULT;
+ case LONG:
+ return LONG_DEFAULT;
+ case FLOAT:
+ return FLOAT_DEFAULT;
+ case DOUBLE:
+ return DOUBLE_DEFAULT;
+ case STRING:
+ return STRING_DEFAULT;
+ case DATE:
+ return DATE_DEFAULT;
+ case TIME:
+ return TIME_DEFAULT_MICROS;
+ case TIMESTAMP:
+ return TIMESTAMP_DEFAULT_MICROS;
+ case TIMESTAMP_NANO:
+ return TIMESTAMP_DEFAULT_NANOS;
+ case UUID:
+ return UUID_DEFAULT;
+ case FIXED:
+ return ByteBuffer.allocate(((Types.FixedType)
type).length()).asReadOnlyBuffer();
+ case BINARY:
+ return EMPTY_BUFFER;
+ case DECIMAL:
+ return new BigDecimal(BigInteger.ZERO, ((Types.DecimalType)
type).scale());
+ case VARIANT:
+ return EMPTY_VARIANT;
+ case LIST:
+ return Collections.emptyList();
+ case MAP:
+ return Collections.emptyMap();
+ case STRUCT:
+ return defaultStruct(type.asStructType());
+ default:
+ throw new IllegalStateException("unreachable: canBind should have
rejected " + type);
+ }
+ }
+
+ private static StructLike defaultStruct(Types.StructType structType) {
+ List<Types.NestedField> fields = structType.fields();
+ Object[] values = new Object[fields.size()];
+ for (int i = 0; i < fields.size(); i++) {
+ values[i] = defaultValueFor(fields.get(i).type());
+ }
+ return new DefaultStruct(values);
+ }
+
+ private static final class ConstantFn implements
SerializableFunction<Object, Object> {
+ private final Object constant;
+
+ ConstantFn(Object constant) {
+ this.constant = constant;
+ }
+
+ @Override
+ public Object apply(Object value) {
+ return constant;
Review Comment:
`MaskToFixedValue` is the only function that doesn't preserve null —
`NullSafeFunction` documents "if the input column value is NULL, the output
MUST be NULL," every other function extends it, and `ConstantFn` returns the
constant regardless, so `apply(null)` gives `0` / `"XXXXXXXX"`.
`testMaskToFixedValueNullReturnsFixedValue` pins that.
Worth flagging because it's a semantics question for the spec, not just a
code detail: is `null → fixed value` intended, or should mask-to-fixed-value
preserve null like the rest? There's a small tell that it matters — `null → 0`
is distinguishable from "never had a value," so it leaks that the original was
non-null. Whichever way #13879 lands, I'd get the javadoc, the code, and the
test to agree. Which is intended 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);
Review Comment:
`ExpressionParser` serializes column refs by name, so `country = 'US'` is
stored as the name `country` — after a rename to `region` the filter silently
stops binding, or binds to the wrong column.
#13879 already calls for field-id refs precisely for schema-evolution
stability, so this is mostly a note that the prototype currently diverges from
where the spec is heading. Good thing to reconcile as the spec firms up —
field-id-based refs, or a name↔id translation on parse.
##########
data/src/main/java/org/apache/iceberg/data/ReadRestrictionsApplier.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.expressions.Expressions;
+import org.apache.iceberg.functions.IcebergFunction;
+import org.apache.iceberg.functions.SaltedFunction;
+import org.apache.iceberg.io.CloseableIterable;
+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.
+ */
+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);
+ return maskColumns(filtered, restrictions.columnProjections(), projection);
+ }
+
+ private static CloseableIterable<Record> filterRows(
+ CloseableIterable<Record> records, Expression rowFilter, Schema
projection) {
+ if (rowFilter == null || rowFilter == Expressions.alwaysTrue()) {
+ return records;
+ }
+
+ Types.StructType struct = projection.asStruct();
+ Evaluator evaluator = new Evaluator(struct, rowFilter, true);
+ InternalRecordWrapper wrapper = new InternalRecordWrapper(struct);
+ return CloseableIterable.filter(records, record ->
evaluator.eval(wrapper.wrap(record)));
+ }
+
+ private static CloseableIterable<Record> maskColumns(
+ CloseableIterable<Record> records, List<IcebergFunction<?, ?>> actions,
Schema projection) {
+ if (actions.isEmpty()) {
+ return records;
+ }
+
+ Map<String, SerializableFunction<Object, Object>> masksByName =
bindMasks(actions, projection);
+ return CloseableIterable.transform(records, record -> mask(record,
masksByName));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<String, SerializableFunction<Object, Object>> bindMasks(
+ List<IcebergFunction<?, ?>> actions, Schema projection) {
+ ImmutableMap.Builder<String, SerializableFunction<Object, Object>> builder
=
+ ImmutableMap.builder();
+ byte[] querySalt = null;
+ List<Types.NestedField> topLevelFields = projection.asStruct().fields();
+
+ for (IcebergFunction<?, ?> action : actions) {
+ int fieldId = action.fieldId();
+ Types.NestedField field = findTopLevel(topLevelFields, fieldId);
+ if (field == null) {
+ // Fail closed: nested masks, unknown fieldIds, or fields projected
away all reach here.
+ // Skipping them silently would either leak unmasked values (nested
case) or surprise the
+ // caller (typo case). The latter is acceptable noise since this
surfaces at bind time.
+ String path = projection.findColumnName(fieldId);
+ if (path == null) {
+ throw new IllegalStateException(
+ "ReadRestrictions references unknown field id: " + fieldId);
+ }
+ throw new IllegalStateException(
+ "ReadRestrictions on nested fields are not yet supported "
+ + "(fieldId="
+ + fieldId
+ + ", path='"
+ + path
+ + "')");
+ }
+
+ SerializableFunction<Object, Object> bound;
+ if (action instanceof SaltedFunction) {
+ if (querySalt == null) {
+ querySalt = new byte[SALT_LENGTH];
+ RANDOM.nextBytes(querySalt);
+ }
+ bound =
+ (SerializableFunction<Object, Object>)
+ ((SaltedFunction<?, ?>) action).bind(field.type(), querySalt);
+ } else {
+ bound = (SerializableFunction<Object, Object>)
action.bind(field.type());
+ }
+ builder.put(field.name(), bound);
Review Comment:
Two projections for the same field id (valid JSON, nothing dedups) make
`ImmutableMap.Builder.build()` throw `Multiple entries with same key` — an
internal-looking error rather than a policy one. I'd reject duplicate field ids
in `ReadRestrictions.of()`, or define last-write-wins explicitly.
##########
data/src/main/java/org/apache/iceberg/data/ReadRestrictionsApplier.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.expressions.Expressions;
+import org.apache.iceberg.functions.IcebergFunction;
+import org.apache.iceberg.functions.SaltedFunction;
+import org.apache.iceberg.io.CloseableIterable;
+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.
+ */
+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);
+ return maskColumns(filtered, restrictions.columnProjections(), projection);
+ }
+
+ private static CloseableIterable<Record> filterRows(
+ CloseableIterable<Record> records, Expression rowFilter, Schema
projection) {
+ if (rowFilter == null || rowFilter == Expressions.alwaysTrue()) {
+ return records;
+ }
+
+ Types.StructType struct = projection.asStruct();
+ Evaluator evaluator = new Evaluator(struct, rowFilter, true);
+ InternalRecordWrapper wrapper = new InternalRecordWrapper(struct);
+ return CloseableIterable.filter(records, record ->
evaluator.eval(wrapper.wrap(record)));
+ }
+
+ private static CloseableIterable<Record> maskColumns(
+ CloseableIterable<Record> records, List<IcebergFunction<?, ?>> actions,
Schema projection) {
+ if (actions.isEmpty()) {
+ return records;
+ }
+
+ Map<String, SerializableFunction<Object, Object>> masksByName =
bindMasks(actions, projection);
+ return CloseableIterable.transform(records, record -> mask(record,
masksByName));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<String, SerializableFunction<Object, Object>> bindMasks(
+ List<IcebergFunction<?, ?>> actions, Schema projection) {
+ ImmutableMap.Builder<String, SerializableFunction<Object, Object>> builder
=
+ ImmutableMap.builder();
+ byte[] querySalt = null;
+ List<Types.NestedField> topLevelFields = projection.asStruct().fields();
+
+ for (IcebergFunction<?, ?> action : actions) {
+ int fieldId = action.fieldId();
+ Types.NestedField field = findTopLevel(topLevelFields, fieldId);
+ if (field == null) {
+ // Fail closed: nested masks, unknown fieldIds, or fields projected
away all reach here.
+ // Skipping them silently would either leak unmasked values (nested
case) or surprise the
+ // caller (typo case). The latter is acceptable noise since this
surfaces at bind time.
+ String path = projection.findColumnName(fieldId);
+ if (path == null) {
+ throw new IllegalStateException(
+ "ReadRestrictions references unknown field id: " + fieldId);
+ }
+ throw new IllegalStateException(
+ "ReadRestrictions on nested fields are not yet supported "
+ + "(fieldId="
+ + fieldId
+ + ", path='"
+ + path
+ + "')");
+ }
+
+ SerializableFunction<Object, Object> bound;
Review Comment:
`ReplaceWithNull`'s javadoc says it's only valid for optional fields and
delegates the check to the caller, but `bindMasks` is the only caller and
doesn't check — a `replace-with-null` on a required field binds and produces
null where the schema says non-null, which blows up the writer downstream with
a confusing error. `findTopLevel` already returns the `NestedField`, so a quick
`!field.isOptional()` guard here would do it.
##########
data/src/test/java/org/apache/iceberg/data/TestReadRestrictionsApplier.java:
##########
@@ -0,0 +1,203 @@
+/*
+ * 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 static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.List;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.functions.MaskAlphanum;
+import org.apache.iceberg.functions.ShowLast4;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.rest.restrictions.ReadRestrictions;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+public class TestReadRestrictionsApplier {
+
Review Comment:
The coverage is all unit-level on synthetic records — good for the applier
in isolation, but nothing exercises the seam that matters: `LoadTableResponse`
with restrictions → parse → load through `RESTSessionCatalog` → read via
`IcebergGenerics`, asserting masks and filter are enforced.
Since every failure mode here is fail-open, an end-to-end test against a
mock REST server would be the highest-value thing to add whenever this moves
past prototype (and would be a nice demonstration for the spec too).
##########
core/src/main/java/org/apache/iceberg/rest/responses/LoadTableResponseParser.java:
##########
@@ -70,6 +72,11 @@ public static void toJson(LoadTableResponse response,
JsonGenerator gen) throws
gen.writeEndArray();
}
+ if (!response.readRestrictions().isEmpty()) {
Review Comment:
Not asking for a spec YAML change in a prototype — that's expected to trail
the discussion. But this is the concrete spot where the fail-open behavior
lives: the field is purely additive, so a client that doesn't read it returns
unmasked data silently.
The question for #13879 is whether there needs to be a mandatory-enforcement
signal — some flag a conforming client must honor and refuse to return data if
it can't enforce — so a masking feature doesn't default to open across clients.
Feels like a core thing for the spec to take a position on. wdyt?
##########
core/src/test/java/org/apache/iceberg/rest/restrictions/TestReadRestrictionsParser.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.List;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.functions.IcebergFunction;
+import org.apache.iceberg.functions.MaskAlphanum;
+import org.apache.iceberg.functions.MaskToFixedValue;
+import org.apache.iceberg.functions.ReplaceWithNull;
+import org.apache.iceberg.functions.Sha256Global;
+import org.apache.iceberg.functions.Sha256QueryLocal;
+import org.apache.iceberg.functions.ShowFirst4;
+import org.apache.iceberg.functions.ShowLast4;
+import org.apache.iceberg.functions.TruncateToMonth;
+import org.apache.iceberg.functions.TruncateToYear;
+import org.apache.iceberg.functions.UnknownFunction;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Test;
+
+public class TestReadRestrictionsParser {
+
+ @Test
+ public void emptyRestrictionsRoundTrip() {
+ String json = ReadRestrictionsParser.toJson(ReadRestrictions.empty());
+ ReadRestrictions parsed = ReadRestrictionsParser.fromJson(json);
+ assertThat(parsed.isEmpty()).isTrue();
+ }
+
+ @Test
+ public void emptyObjectParsesAsEmpty() {
+ ReadRestrictions parsed = ReadRestrictionsParser.fromJson("{}");
+ assertThat(parsed.isEmpty()).isTrue();
+ }
+
+ @Test
+ public void rowFilterRoundTrip() {
+ Expression filter = Expressions.equal("country", "US");
+ ReadRestrictions restrictions = ReadRestrictions.of(filter,
ImmutableList.of());
+ String json = ReadRestrictionsParser.toJson(restrictions);
+
+ assertThat(json).contains("required-row-filter");
+ assertThat(json).doesNotContain("required-column-projections");
+
+ ReadRestrictions parsed = ReadRestrictionsParser.fromJson(json);
+ assertThat(parsed.rowFilter()).isNotNull();
Review Comment:
`rowFilterRoundTrip` only checks non-null, so a parser regression that
round-tripped `country = 'US'` into always-true would pass while quietly
disabling the filter. Asserting semantic equality —
`assertThat(ExpressionParser.toJson(parsed.rowFilter())).isEqualTo(ExpressionParser.toJson(filter))`,
or an `Evaluator` check on a known struct — makes it actually exercise the
round-trip.
##########
api/src/main/java/org/apache/iceberg/functions/IcebergFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.functions;
+
+import java.io.Serializable;
+import java.util.Objects;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.util.SerializableFunction;
+
+/**
+ * A named, type-aware function that can be bound to an Iceberg {@link Type}.
+ *
+ * <p>{@link #bind(Type)} returns a {@link SerializableFunction} that applies
this function's logic
+ * to values of the bound type.
+ *
+ * @param <S> input value type
+ * @param <T> output value type
+ */
+public interface IcebergFunction<S, T> extends Serializable {
Review Comment:
Everything here rides in Spark task closures and none of the `Serializable`
types declare a `serialVersionUID`, so any refactor risks
`InvalidClassException` across mixed-version clusters. Worth adding
`serialVersionUID = 1L` to the concrete types and `readResolve()` to the
singleton `*Fn` instances before this is load-bearing.
##########
core/src/main/java/org/apache/iceberg/rest/RESTTable.java:
##########
@@ -51,8 +51,9 @@ class RESTTable extends BaseTable implements
SupportsDistributedScanPlanning {
ResourcePaths resourcePaths,
Set<Endpoint> supportedEndpoints,
Map<String, String> catalogProperties,
- Object hadoopConf) {
- super(ops, name, reporter);
+ Object hadoopConf,
+ ReadRestrictions readRestrictions) {
+ super(ops, name, reporter, readRestrictions);
Review Comment:
Heads-up — this looks like it's on a pre-labels base (the `super(...)` calls
here don't carry `Labels`). A rebase on current main would let reviewers see it
against the labels work rather than diffing it away; no real impact on the
prototype itself.
##########
data/src/main/java/org/apache/iceberg/data/ReadRestrictionsApplier.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.expressions.Expressions;
+import org.apache.iceberg.functions.IcebergFunction;
+import org.apache.iceberg.functions.SaltedFunction;
+import org.apache.iceberg.io.CloseableIterable;
+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.
+ */
+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);
+ return maskColumns(filtered, restrictions.columnProjections(), projection);
+ }
+
+ private static CloseableIterable<Record> filterRows(
+ CloseableIterable<Record> records, Expression rowFilter, Schema
projection) {
+ if (rowFilter == null || rowFilter == Expressions.alwaysTrue()) {
Review Comment:
Reference equality against `True.INSTANCE` works today but breaks quietly if
the filter is ever rebuilt through a different path; `rowFilter.op() ==
Expression.Operation.TRUE` is the sturdier idiom. Relatedly, returning `null`
from `rowFilter()` for "no filter" is what forces the null-checks everywhere —
`Expressions.alwaysTrue()` would tidy that up.
##########
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:
Two edges here for later: restricted tables are never cached, so conditional
GET is permanently off for them; and empty-restriction tables still get cached,
so a server that later attaches restrictions under the same ETag gets bypassed.
Folding a policy version into the cache key resolves both — a productionization
detail, not a prototype concern.
--
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]
