Open API: [4/N] Add RCK tests for ReadRestrictions

api/src/main/java/org/apache/iceberg/expressions/BoundApply.java

@@ -0,0 +1,102 @@
+/*
+ * 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.expressions;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.types.Type;
+
+/**
+ * A function applied to bound arguments.
+ *
+ * @param <T> the Java type of the function's result
+ */
+public class BoundApply<T> implements BoundTerm<T> {
+  private final FunctionReference function;
+  private final Object[] arguments;
+  private final Type type;
+
+  BoundApply(FunctionReference function, Object[] arguments, Type type) {
+    this.function = function;
+    this.arguments = arguments;
+    this.type = type;
+  }
+
+  public FunctionReference function() {
+    return function;
+  }
+
+  List<Object> arguments() {
+    return Arrays.asList(arguments);
+  }
+
+  @Override
+  public Type type() {
+    return type;
+  }
+
+  @Override
+  public BoundReference<?> ref() {
+    throw new UnsupportedOperationException("Cannot determine reference for function: " + function);
+  }
+
+  @Override
+  public T eval(StructLike struct) {
+    throw new UnsupportedOperationException("Cannot evaluate " + this);
+  }
+
+  @Override
+  public boolean isEquivalentTo(BoundTerm<?> other) {
+    if (!(other instanceof BoundApply<?> that)) {
+      return false;
+    }
+
+    if (!Objects.equals(function.catalog(), that.function.catalog())
+        || !function.identifier().equals(that.function.identifier())
+        || !type.equals(that.type)
+        || arguments.length != that.arguments.length) {
+      return false;
+    }
+
+    for (int i = 0; i < arguments.length; i += 1) {
+      if (!isEquivalent(arguments[i], that.arguments[i])) {
+        return false;
+      }
+    }
+
+    return true;
+  }
+
+  private static boolean isEquivalent(Object left, Object right) {
+    if (left instanceof BoundTerm<?> term && right instanceof BoundTerm<?> otherTerm) {
+      return term.isEquivalentTo(otherTerm);
+    } else if (left instanceof Expression expr && right instanceof Expression otherExpr) {
+      return expr.isEquivalentTo(otherExpr);
+    }
+
+    return left.equals(right);
+  }
+
+  @Override
+  public String toString() {
+    return function + "(" + Arrays.toString(arguments) + ")";
+  }
+}

api/src/main/java/org/apache/iceberg/expressions/BoundTransform.java

@@ -20,7 +20,6 @@
 
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.transforms.Transform;
-import org.apache.iceberg.types.Type;
 import org.apache.iceberg.util.SerializableFunction;
 
 /**
@@ -29,12 +28,16 @@
  * @param <S> the Java type of values transformed by this function.
  * @param <T> the Java type of values returned by the function.
  */
-public class BoundTransform<S, T> implements BoundTerm<T> {
+public class BoundTransform<S, T> extends BoundApply<T> {
   private final BoundReference<S> ref;
   private final Transform<S, T> transform;
   private final SerializableFunction<S, T> func;
 
   BoundTransform(BoundReference<S> ref, Transform<S, T> transform) {
+    super(
+        UnboundTransform.functionFor(transform),
+        UnboundTransform.argumentsFor(transform, ref).toArray(),
+        transform.getResultType(ref.type()));
     this.ref = ref;
     this.transform = transform;
     this.func = transform.bind(ref.type());
@@ -61,11 +64,6 @@ public boolean producesNull() {
     return ref.producesNull() || !transform.preservesOrder();
   }
 
-  @Override
-  public Type type() {
-    return transform.getResultType(ref.type());
-  }
-
   @Override
   public boolean isEquivalentTo(BoundTerm<?> other) {
     if (other instanceof BoundTransform) {

api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java

@@ -254,11 +254,28 @@ public static <T> UnboundTerm<T> unbind(BoundTerm<T> term) {
       return Expressions.transform(bound.ref().name(), bound.transform());
     } else if (term instanceof BoundReference) {
       return Expressions.ref(((BoundReference<T>) term).name());
+    } else if (term instanceof BoundApply) {
+      BoundApply<T> bound = (BoundApply<T>) term;
+      return new UnboundApply<>(
+          bound.function(),
+          Lists.transform(bound.arguments(), ExpressionUtil::unbindArgument),
+          bound.type());
     }
 
     throw new UnsupportedOperationException("Cannot unbind unsupported term: " + term);
   }
 
+  private static Object unbindArgument(Object argument) {
+    if (argument instanceof BoundTerm<?> term) {
+      return unbind(term);
+    } else if (argument instanceof Expression) {
+      throw new UnsupportedOperationException(
+          "Cannot unbind unsupported function argument: " + argument);
+    }
+
+    return argument;
+  }
+
   @SuppressWarnings("unchecked")
   public static <T> UnboundTerm<T> unbind(Term term) {
     if (term instanceof UnboundTerm) {

api/src/main/java/org/apache/iceberg/expressions/Expressions.java

@@ -18,6 +18,8 @@
  */
 package org.apache.iceberg.expressions;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.stream.Stream;
 import org.apache.iceberg.expressions.Expression.Operation;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -317,6 +319,39 @@ public static <T> UnboundTerm<T> transform(String name, Transform<?, T> transfor
     return new UnboundTransform<>(ref(name), transform);
   }
 
+  /**
+   * Create a reference to a function by identifier, in the catalog of the referencing object.
+   *
+   * <p>Identifier parts are never split; a part containing dots is a single name.
+   *
+   * @param identifier namespace names followed by the function name
+   * @return a function reference
+   */
+  public static FunctionReference function(String... identifier) {
+    return new FunctionReference(null, Arrays.asList(identifier));
+  }
+
+  /**
+   * Create a reference to a function by identifier, in the catalog of the referencing object.
+   *
+   * @param identifier namespace names followed by the function name
+   * @return a function reference
+   */
+  public static FunctionReference function(List<String> identifier) {
+    return new FunctionReference(null, identifier);
+  }
+
+  /**
+   * Create a reference to a function in a specific catalog.
+   *
+   * @param catalog a catalog name
+   * @param identifier namespace names followed by the function name
+   * @return a function reference
+   */
+  public static FunctionReference function(String catalog, List<String> identifier) {
+    return new FunctionReference(catalog, identifier);
+  }
+
   /**
    * Create a {@link Literal} from an Object.
    *

api/src/main/java/org/apache/iceberg/expressions/FunctionReference.java

@@ -0,0 +1,67 @@
+/*
+ * 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.expressions;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A reference to a function by identifier, optionally qualified with a catalog.
+ *
+ * <p>Created with {@link Expressions#function(String...)}.
+ */
+public class FunctionReference implements Serializable {
+  private final String catalog;
+  private final String[] identifier;
+
+  FunctionReference(String catalog, List<String> identifier) {
+    Preconditions.checkArgument(
+        identifier != null && !identifier.isEmpty(), "Invalid function identifier: %s", identifier);
+    Preconditions.checkArgument(
+        identifier.stream().noneMatch(part -> part == null || part.isEmpty()),
+        "Invalid function identifier (empty or null part): %s",
+        identifier);
+    this.catalog = catalog;
+    this.identifier = identifier.toArray(new String[0]);
+  }
+
+  public String catalog() {
+    return catalog;
+  }
+
+  public List<String> identifier() {
+    return Arrays.asList(identifier);
+  }
+
+  /** Returns the name of the function, without the catalog or namespace. */
+  public String name() {
+    return identifier[identifier.length - 1];
+  }
+
+  @Override
+  public String toString() {
+    if (catalog != null) {
+      return catalog + "." + String.join(".", identifier);
+    }
+
+    return String.join(".", identifier);
+  }
+}

api/src/main/java/org/apache/iceberg/expressions/UnboundApply.java

@@ -0,0 +1,104 @@
+/*
+ * 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.expressions;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+/**
+ * A function applied to unbound arguments.
+ *
+ * @param <T> the Java type of the function's result
+ */
+public class UnboundApply<T> implements UnboundTerm<T> {
+  private final FunctionReference function;
+  private final Object[] arguments;
+  private final Type resultType;
+
+  UnboundApply(FunctionReference function, List<Object> arguments) {
+    this(function, arguments, null);
+  }
+
+  UnboundApply(FunctionReference function, List<Object> arguments, Type resultType) {
+    Preconditions.checkArgument(function != null, "Invalid function: null");
+    this.function = function;
+    this.arguments =
+        arguments == null
+            ? new Object[0]
+            : Lists.transform(arguments, UnboundApply::toArgument).toArray();
+    this.resultType = resultType;
+  }
+
+  private static Object toArgument(Object argument) {
+    Preconditions.checkArgument(argument != null, "Invalid function argument: null");
+    if (argument instanceof Term || argument instanceof Expression) {
+      return argument;
+    }
+
+    return Literals.from(argument);
+  }
+
+  public FunctionReference function() {
+    return function;
+  }
+
+  /** Returns the arguments, each a {@link Term}, an {@link Expression}, or a {@link Literal}. */
+  List<Object> arguments() {
+    return Arrays.asList(arguments);
+  }
+
+  @Override
+  public NamedReference<?> ref() {
+    throw new UnsupportedOperationException("Cannot determine reference for function: " + function);
+  }
+
+  @Override
+  public BoundTerm<T> bind(Types.StructType struct, boolean caseSensitive) {
+    ValidationException.check(
+        resultType != null, "Cannot bind function without a result type: %s", function);
+
+    Object[] boundArguments = new Object[arguments.length];
+    for (int i = 0; i < arguments.length; i += 1) {
+      boundArguments[i] = bindArgument(arguments[i], struct, caseSensitive);
+    }
+
+    return new BoundApply<>(function, boundArguments, resultType);
+  }
+
+  private static Object bindArgument(
+      Object argument, Types.StructType struct, boolean caseSensitive) {
+    if (argument instanceof UnboundTerm<?> term) {
+      return term.bind(struct, caseSensitive);
+    } else if (argument instanceof Expression expr) {
+      return Binder.bind(struct, expr, caseSensitive);
+    }
+
+    return argument;
+  }
+
+  @Override
+  public String toString() {
+    return function + "(" + Arrays.toString(arguments) + ")";
+  }
+}