Core: Add expression JSON parser

.palantir/revapi.yml

@@ -1,6 +1,11 @@
 versionOverrides:
   org.apache.iceberg:iceberg-api:apache-iceberg-0.14.0: "0.14.0"
 acceptedBreaks:
+  apache-iceberg-0.14.0:
+    org.apache.iceberg:iceberg-api:
+    - code: "java.method.addedToInterface"
+      new: "method java.lang.String org.apache.iceberg.expressions.Reference<T>::name()"
+      justification: "All subclasses implement name"
   release-base-0.13.0:
     org.apache.iceberg:iceberg-api:
     - code: "java.class.defaultSerializationChanged"

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

@@ -26,10 +26,12 @@
 public class BoundReference<T> implements BoundTerm<T>, Reference<T> {
   private final Types.NestedField field;
   private final Accessor<StructLike> accessor;
+  private final String name;
 
-  BoundReference(Types.NestedField field, Accessor<StructLike> accessor) {
+  BoundReference(Types.NestedField field, Accessor<StructLike> accessor, String name) {
     this.field = field;
     this.accessor = accessor;
+    this.name = name;
   }
 
   @Override
@@ -52,6 +54,11 @@ public Type type() {
     return field.type();
   }
 
+  @Override
+  public String name() {
+    return name;
+  }
+
   @Override
   public boolean isEquivalentTo(BoundTerm<?> other) {
     if (other instanceof BoundReference) {

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

@@ -19,6 +19,7 @@
 package org.apache.iceberg.expressions;
 
 import java.util.Set;
+import java.util.function.Supplier;
 import org.apache.iceberg.exceptions.ValidationException;
 
 /** Utils for traversing {@link Expression expressions}. */
@@ -404,4 +405,187 @@ public static Boolean visitEvaluator(Expression expr, ExpressionVisitor<Boolean>
       }
     }
   }
+
+  public abstract static class CustomOrderExpressionVisitor<R> {
+    public R alwaysTrue() {
+      return null;
+    }
+
+    public R alwaysFalse() {
+      return null;
+    }
+
+    public R not(Supplier<R> result) {
+      return null;
+    }
+
+    public R and(Supplier<R> leftResult, Supplier<R> rightResult) {
+      return null;
+    }
+
+    public R or(Supplier<R> leftResult, Supplier<R> rightResult) {
+      return null;
+    }
+
+    public <T> R predicate(UnboundPredicate<T> pred) {
+      throw new UnsupportedOperationException("Not a bound predicate: " + pred);
+    }
+
+    public <T> R predicate(BoundPredicate<T> pred) {
+      if (pred.isLiteralPredicate()) {
+        BoundLiteralPredicate<T> literalPred = pred.asLiteralPredicate();
+        switch (pred.op()) {
+          case LT:
+            return lt(pred.term(), literalPred.literal());
+          case LT_EQ:
+            return ltEq(pred.term(), literalPred.literal());
+          case GT:
+            return gt(pred.term(), literalPred.literal());
+          case GT_EQ:
+            return gtEq(pred.term(), literalPred.literal());
+          case EQ:
+            return eq(pred.term(), literalPred.literal());
+          case NOT_EQ:
+            return notEq(pred.term(), literalPred.literal());
+          case STARTS_WITH:
+            return startsWith(pred.term(), literalPred.literal());
+          case NOT_STARTS_WITH:
+            return notStartsWith(pred.term(), literalPred.literal());
+          default:
+            throw new IllegalStateException(
+                "Invalid operation for BoundLiteralPredicate: " + pred.op());
+        }
+
+      } else if (pred.isUnaryPredicate()) {
+        switch (pred.op()) {
+          case IS_NULL:
+            return isNull(pred.term());
+          case NOT_NULL:
+            return notNull(pred.term());
+          case IS_NAN:
+            return isNaN(pred.term());
+          case NOT_NAN:
+            return notNaN(pred.term());
+          default:
+            throw new IllegalStateException(
+                "Invalid operation for BoundUnaryPredicate: " + pred.op());
+        }
+
+      } else if (pred.isSetPredicate()) {
+        switch (pred.op()) {
+          case IN:
+            return in(pred.term(), pred.asSetPredicate().literalSet());
+          case NOT_IN:
+            return notIn(pred.term(), pred.asSetPredicate().literalSet());
+          default:
+            throw new IllegalStateException(
+                "Invalid operation for BoundSetPredicate: " + pred.op());
+        }
+      }
+
+      throw new IllegalStateException("Unsupported bound predicate: " + pred.getClass().getName());
+    }
+
+    public <T> R isNull(BoundTerm<T> term) {
+      return null;
+    }
+
+    public <T> R notNull(BoundTerm<T> term) {
+      return null;
+    }
+
+    public <T> R isNaN(BoundTerm<T> term) {
+      return null;
+    }
+
+    public <T> R notNaN(BoundTerm<T> term) {
+      return null;
+    }
+
+    public <T> R lt(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+
+    public <T> R ltEq(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+
+    public <T> R gt(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+
+    public <T> R gtEq(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+
+    public <T> R eq(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+
+    public <T> R notEq(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+
+    public <T> R in(BoundTerm<T> term, Set<T> literalSet) {
+      return null;
+    }
+
+    public <T> R notIn(BoundTerm<T> term, Set<T> literalSet) {
+      return null;
+    }
+
+    public <T> R startsWith(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+
+    public <T> R notStartsWith(BoundTerm<T> term, Literal<T> lit) {
+      return null;
+    }
+  }
+
+  /**
+   * Traverses the given {@link Expression expression} with a {@link CustomOrderExpressionVisitor
+   * visitor}.
+   *
+   * <p>This passes a {@link Supplier} to each non-leaf {@link CustomOrderExpressionVisitor visitor}
+   * method. The supplier returns the result of traversing child expressions. Getting the result of
+   * the supplier allows traversing the expression in the desired order.
+   *
+   * @param expr an expression to traverse
+   * @param visitor a visitor that will be called to handle each node in the expression tree
+   * @param <R> the return type produced by the expression visitor
+   * @return the value returned by the visitor for the root expression node
+   */
+  public static <R> R visit(Expression expr, CustomOrderExpressionVisitor<R> visitor) {
+    return visitExpr(expr, visitor).get();
+  }
+
+  private static <R> Supplier<R> visitExpr(
+      Expression expr, CustomOrderExpressionVisitor<R> visitor) {
+    if (expr instanceof Predicate) {
+      if (expr instanceof BoundPredicate) {
+        return () -> visitor.predicate((BoundPredicate<?>) expr);
+      } else {
+        return () -> visitor.predicate((UnboundPredicate<?>) expr);
+      }
+    } else {
+      switch (expr.op()) {
+        case TRUE:
+          return visitor::alwaysTrue;
+        case FALSE:
+          return visitor::alwaysFalse;
+        case NOT:
+          Not not = (Not) expr;
+          return () -> visitor.not(visitExpr(not.child(), visitor));
+        case AND:
+          And and = (And) expr;
+          return () -> visitor.and(visitExpr(and.left(), visitor), visitExpr(and.right(), visitor));
+        case OR:
+          Or or = (Or) expr;
+          return () -> visitor.or(visitExpr(or.left(), visitor), visitExpr(or.right(), visitor));
+        default:
+          throw new UnsupportedOperationException("Unknown operation: " + expr.op());
+      }
+    }
+  }
 }

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

@@ -274,6 +274,10 @@ public static <T> UnboundPredicate<T> predicate(
     return new UnboundPredicate<>(op, expr, values);
   }
 
+  public static <T> UnboundPredicate<T> predicate(Operation op, UnboundTerm<T> expr) {
+    return new UnboundPredicate<>(op, expr);
+  }
+
   public static True alwaysTrue() {
     return True.INSTANCE;
   }

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

@@ -31,6 +31,7 @@ public class NamedReference<T> implements UnboundTerm<T>, Reference<T> {
     this.name = name;
   }
 
+  @Override
   public String name() {
     return name;
   }
@@ -44,7 +45,7 @@ public BoundReference<T> bind(Types.StructType struct, boolean caseSensitive) {
     ValidationException.check(
         field != null, "Cannot find field '%s' in struct: %s", name, schema.asStruct());
 
-    return new BoundReference<>(field, schema.accessorForField(field.fieldId()));
+    return new BoundReference<>(field, schema.accessorForField(field.fieldId()), name);
   }
 
   @Override

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

@@ -24,4 +24,6 @@
  * @see BoundReference
  * @see NamedReference
  */
-public interface Reference<T> extends Term {}
+public interface Reference<T> extends Term {
+  String name();
+}

core/src/main/java/org/apache/iceberg/DefaultValueParser.java → core/src/main/java/org/apache/iceberg/SingleValueParser.java

@@ -41,8 +41,8 @@
 import org.apache.iceberg.util.DateTimeUtil;
 import org.apache.iceberg.util.JsonUtil;
 
-public class DefaultValueParser {
-  private DefaultValueParser() {}
+public class SingleValueParser {
+  private SingleValueParser() {}
 
   private static final String KEYS = "keys";
   private static final String VALUES = "values";