feat(isthmus): support in-predicate (#204)

core/src/main/java/io/substrait/dsl/SubstraitBuilder.java

@@ -375,6 +375,13 @@ public Expression singleOrList(Expression condition, Expression... options) {
     return SingleOrList.builder().condition(condition).addOptions(options).build();
   }
 
+  public Expression.InPredicate inPredicate(Rel haystack, Expression... needles) {
+    return Expression.InPredicate.builder()
+        .addAllNeedles(Arrays.asList(needles))
+        .haystack(haystack)
+        .build();
+  }
+
   public List<Expression.SortField> sortFields(Rel input, int... indexes) {
     return Arrays.stream(indexes)
         .mapToObj(

isthmus/src/main/java/io/substrait/isthmus/SubstraitRelNodeConverter.java

@@ -110,6 +110,7 @@ public SubstraitRelNodeConverter(
     this.scalarFunctionConverter = scalarFunctionConverter;
     this.aggregateFunctionConverter = aggregateFunctionConverter;
     this.expressionRexConverter = expressionRexConverter;
+    this.expressionRexConverter.setRelNodeConverter(this);
   }
 
   public static RelNode convert(

isthmus/src/main/java/io/substrait/isthmus/expression/ExpressionRexConverter.java

@@ -10,6 +10,7 @@
 import io.substrait.expression.FunctionArg;
 import io.substrait.expression.WindowBound;
 import io.substrait.extension.SimpleExtension;
+import io.substrait.isthmus.SubstraitRelNodeConverter;
 import io.substrait.isthmus.TypeConverter;
 import io.substrait.type.StringTypeVisitor;
 import io.substrait.type.Type;
@@ -22,12 +23,14 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.apache.calcite.avatica.util.ByteString;
+import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexFieldCollation;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexSubQuery;
 import org.apache.calcite.rex.RexWindowBound;
 import org.apache.calcite.rex.RexWindowBounds;
 import org.apache.calcite.sql.SqlAggFunction;
@@ -50,6 +53,7 @@ public class ExpressionRexConverter extends AbstractExpressionVisitor<RexNode, R
   protected final RexBuilder rexBuilder;
   protected final ScalarFunctionConverter scalarFunctionConverter;
   protected final WindowFunctionConverter windowFunctionConverter;
+  protected SubstraitRelNodeConverter relNodeConverter;
 
   private static final SqlIntervalQualifier YEAR_MONTH_INTERVAL =
       new SqlIntervalQualifier(
@@ -79,6 +83,10 @@ public ExpressionRexConverter(
     this.windowFunctionConverter = windowFunctionConverter;
   }
 
+  public void setRelNodeConverter(final SubstraitRelNodeConverter substraitRelNodeConverter) {
+    this.relNodeConverter = substraitRelNodeConverter;
+  }
+
   @Override
   public RexNode visit(Expression.NullLiteral expr) throws RuntimeException {
     return rexBuilder.makeLiteral(null, typeConverter.toCalcite(typeFactory, expr.getType()));
@@ -385,6 +393,14 @@ public RexNode visit(Expression.WindowFunctionInvocation expr) throws RuntimeExc
         ignoreNulls);
   }
 
+  @Override
+  public RexNode visit(Expression.InPredicate expr) throws RuntimeException {
+    List<RexNode> needles =
+        expr.needles().stream().map(e -> e.accept(this)).collect(Collectors.toList());
+    RelNode rel = expr.haystack().accept(relNodeConverter);
+    return RexSubQuery.in(rel, ImmutableList.copyOf(needles));
+  }
+
   static class ToRexWindowBound
       implements WindowBound.WindowBoundVisitor<RexWindowBound, RuntimeException> {
 

isthmus/src/test/java/io/substrait/isthmus/ExpressionConvertabilityTest.java

@@ -57,6 +57,13 @@ public void mapLiteral() throws IOException, SqlParseException {
     assertFullRoundTrip("select MAP[1, 'hello'] from ORDERS");
   }
 
+  @Test
+  public void inPredicate() throws IOException, SqlParseException {
+    assertFullRoundTrip(
+        "select L_PARTKEY from LINEITEM where L_PARTKEY in "
+            + "(SELECT L_SUPPKEY from LINEITEM where L_SUPPKEY < L_ORDERKEY)");
+  }
+
   @Test
   public void singleOrList() {
     Expression singleOrList = b.singleOrList(b.fieldReference(commonTable, 0), b.i32(5), b.i32(10));