add filters w. where clauses for java platforms in sql-api

Author: mspruc

wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/functions/FilterEvaluateCondition.java

@@ -29,13 +29,13 @@
 import org.apache.wayang.api.sql.calcite.converter.WayangFilterVisitor;
 import org.apache.wayang.basic.data.Record;
 
-public class FilterEvaluateCondition extends RexVisitorImpl<Boolean>  {
+public class FilterEvaluateCondition extends RexVisitorImpl<Boolean> {
     final Record record;
 
     protected FilterEvaluateCondition(final boolean deep, final Record record) {
-            super(deep);
-            this.record = record;
-        }
+        super(deep);
+        this.record = record;
+    }
 
     @Override
     public Boolean visitCall(final RexCall call) {
@@ -64,7 +64,6 @@ public Boolean visitCall(final RexCall call) {
 
     public boolean eval(final Record record, final SqlKind kind, final RexNode leftOperand,
             final RexNode rightOperand) {
-
         if (leftOperand instanceof RexInputRef && rightOperand instanceof RexLiteral) {
             final RexInputRef rexInputRef = (RexInputRef) leftOperand;
             final int index = rexInputRef.getIndex();
@@ -85,13 +84,25 @@ public boolean eval(final Record record, final SqlKind kind, final RexNode leftO
                 case LESS_THAN_OR_EQUAL:
                     return isLessThan(field, rexLiteral) || isEqualTo(field, rexLiteral);
                 default:
-                    throw new IllegalStateException("Predicate not supported yet");
+                    throw new IllegalStateException("Predicate not supported yet: " + kind);
             }
+        } else if (leftOperand instanceof RexInputRef && rightOperand instanceof RexInputRef) {
+            final RexInputRef leftRexInputRef = (RexInputRef) leftOperand;
+            final int leftIndex = leftRexInputRef.getIndex();
+            final RexInputRef righRexInputRef = (RexInputRef) rightOperand;
+            final int rightIndex = righRexInputRef.getIndex();
 
+            switch (kind) {
+                case EQUALS:
+                    return isEqualTo(record.getField(leftIndex), record.getField(rightIndex));
+                default:
+                    throw new IllegalStateException("Predicate not supported yet, kind: " + kind + " left field: "
+                            + record.getField(leftIndex) + " right field: " + record.getField(rightIndex));
+            }
         } else {
-            throw new IllegalStateException("Predicate not supported yet");
+            throw new IllegalStateException("Predicate not supported with types yet, predicate: " + kind + ", type1: "
+                    + leftOperand.getClass() + ", type2: " + rightOperand.getClass());
         }
-
     }
 
     private boolean isGreaterThan(final Object o, final RexLiteral rexLiteral) {
@@ -111,4 +122,14 @@ private boolean isEqualTo(final Object o, final RexLiteral rexLiteral) {
             throw new IllegalStateException("Predicate not supported yet");
         }
     }
+
+    private boolean isEqualTo(final Object o1, final Object o2) {
+        System.out.println("comparing: " + o1 + " with " + o2);
+        System.out.println("true: " + ((Comparable) o1).compareTo(o2));
+        try {
+            return ((Comparable) o1).compareTo(o2) == 0;
+        } catch (final Exception e) {
+            throw new IllegalStateException("Predicate not supported yet");
+        }
+    }
 }

wayang-api/wayang-api-sql/src/test/java/org/apache/wayang/api/sql/SqlToWayangRelTest.java

@@ -323,6 +323,7 @@ public void joinWithLargeLeftTableIndexCorrect() throws Exception {
         final Tuple2<Collection<Record>, WayangPlan> t = this.buildCollectorAndWayangPlan(sqlContext,
                 "SELECT * FROM fs.largeLeftTableIndex AS na INNER JOIN fs.largeLeftTableIndex AS nb ON na.NAMEB = nb.NAMEA " //
         );
+
         final Collection<Record> result = t.field0;
         final WayangPlan wayangPlan = t.field1;
         sqlContext.execute(wayangPlan);
@@ -340,16 +341,19 @@ public void joinWithLargeLeftTableIndexMirrorAlias() throws Exception {
         sqlContext.execute(wayangPlan);
     }
 
-    // @Test
+    @Test
     public void exampleFilterTableRefToTableRef() throws Exception {
         final SqlContext sqlContext = createSqlContext("/data/exampleRefToRef.csv");
 
         final Tuple2<Collection<Record>, WayangPlan> t = this.buildCollectorAndWayangPlan(sqlContext,
                 "SELECT * FROM fs.exampleRefToRef WHERE exampleRefToRef.NAMEA = exampleRefToRef.NAMEB" //
         );
+
         final Collection<Record> result = t.field0;
         final WayangPlan wayangPlan = t.field1;
         sqlContext.execute(wayangPlan);
+
+        assert (result.stream().findFirst().get().equals(new Record("test1","test1")));
     }
 
     @Test