initial implementation of startsWith

api/src/main/java/com/netflix/iceberg/expressions/Evaluator.java

@@ -109,6 +109,11 @@ public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
       return cmp.compare(ref.get(struct), lit.value()) > 0;
     }
 
+    @Override
+    public Boolean startsWith(BoundReference<String> ref, Literal<String> lit) {
+      return ref.get(struct).startsWith(lit.value());
+    }
+
     @Override
     public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
       Comparator<T> cmp = lit.comparator();

api/src/main/java/com/netflix/iceberg/expressions/Expression.java

@@ -37,7 +37,8 @@ enum Operation {
     NOT_IN,
     NOT,
     AND,
-    OR;
+    OR,
+    STARTS_WITH;
 
     /**
      * @return the operation used when this is negated

api/src/main/java/com/netflix/iceberg/expressions/ExpressionVisitors.java

@@ -91,6 +91,10 @@ public <T> R notIn(BoundReference<T> ref, Literal<T> lit) {
       return null;
     }
 
+    public R startsWith(BoundReference<String> ref, Literal<String> lit) {
+      return null;
+    }
+
     public <T> R predicate(BoundPredicate<T> pred) {
       switch (pred.op()) {
         case IS_NULL:
@@ -113,6 +117,11 @@ public <T> R predicate(BoundPredicate<T> pred) {
           return in(pred.ref(), pred.literal());
         case NOT_IN:
           return notIn(pred.ref(), pred.literal());
+        case STARTS_WITH:
+//          due to the fact that startWith function in Expressions accepts only string types casting is a must here.
+//          in  Unbound#bind function we added a check for that
+
+          return startsWith((BoundReference<String>) pred.ref(), pred.literal().to(pred.ref().type()));
         default:
           throw new UnsupportedOperationException(
               "Unknown operation for predicate: " + pred.op());

api/src/main/java/com/netflix/iceberg/expressions/Expressions.java

@@ -96,6 +96,10 @@ public static <T> UnboundPredicate<T> notEqual(String name, T value) {
     return new UnboundPredicate<>(Expression.Operation.NOT_EQ, ref(name), value);
   }
 
+  public static <T> UnboundPredicate<T> startWith(String name, T value) {
+    return new UnboundPredicate<>(Operation.STARTS_WITH, ref(name), value);
+  }
+
   public static <T> UnboundPredicate<T> predicate(Operation op, String name, T value) {
     Preconditions.checkArgument(op != Operation.IS_NULL && op != Operation.NOT_NULL,
         "Cannot create %s predicate inclusive a value", op);

api/src/main/java/com/netflix/iceberg/expressions/Predicate.java

@@ -59,6 +59,8 @@ public String toString() {
         return String.valueOf(ref()) + " == " + literal();
       case NOT_EQ:
         return String.valueOf(ref()) + " != " + literal();
+      case STARTS_WITH:
+        return "startsWith(" + ref() + "," + literal() + ")";
 //      case IN:
 //        break;
 //      case NOT_IN:

api/src/main/java/com/netflix/iceberg/expressions/UnboundPredicate.java

@@ -21,6 +21,7 @@
 
 import static com.netflix.iceberg.expressions.Expression.Operation.IS_NULL;
 import static com.netflix.iceberg.expressions.Expression.Operation.NOT_NULL;
+import static com.netflix.iceberg.expressions.Expression.Operation.STARTS_WITH;
 
 public class UnboundPredicate<T> extends Predicate<T, NamedReference> {
 
@@ -63,6 +64,10 @@ public Expression bind(Types.StructType struct) {
       }
     }
 
+    if(op() == STARTS_WITH && field.type() != Types.StringType.get()) {
+      throw new ValidationException("Operation startsWith accepts only strings");
+    }
+
     Literal<T> lit = literal().to(field.type());
     if (lit == null) {
       throw new ValidationException(String.format(

api/src/test/java/com/netflix/iceberg/expressions/TestEvaluatior.java

@@ -17,6 +17,7 @@
 package com.netflix.iceberg.expressions;
 
 import com.netflix.iceberg.TestHelpers;
+import com.netflix.iceberg.exceptions.ValidationException;
 import com.netflix.iceberg.types.Types;
 import com.netflix.iceberg.types.Types.StructType;
 import org.apache.avro.util.Utf8;
@@ -26,6 +27,7 @@
 import static com.netflix.iceberg.expressions.Expressions.alwaysFalse;
 import static com.netflix.iceberg.expressions.Expressions.alwaysTrue;
 import static com.netflix.iceberg.expressions.Expressions.and;
+import static com.netflix.iceberg.expressions.Expressions.startWith;
 import static com.netflix.iceberg.expressions.Expressions.equal;
 import static com.netflix.iceberg.expressions.Expressions.greaterThan;
 import static com.netflix.iceberg.expressions.Expressions.greaterThanOrEqual;
@@ -151,4 +153,24 @@ public void testCharSeqValue() {
     Assert.assertFalse("string(abc) == utf8(abcd) => false",
         evaluator.eval(TestHelpers.Row.of(new Utf8("abcd"))));
   }
+
+  @Test
+  public void testStartWith() {
+    StructType struct = StructType.of(required(3, "s", Types.StringType.get()));
+    Evaluator evaluator = new Evaluator(struct, startWith("s", "abc"));
+    Assert.assertTrue("startWith(abcdddd, abc)  => true",
+        evaluator.eval(TestHelpers.Row.of(("abcdddd"))));
+
+    Assert.assertFalse("startWith(xyzffff, abc)  => false",
+        evaluator.eval(TestHelpers.Row.of(("xyzffff"))));
+
+  }
+
+  @Test(expected = ValidationException.class)
+  public void testStartsWithThrowsOnNotString() {
+    StructType struct = StructType.of(required(3, "s", Types.IntegerType.get()));
+    Evaluator evaluator = new Evaluator(struct, startWith("s", 112));
+    evaluator.eval(TestHelpers.Row.of(("xyzffff")));
+
+  }
 }