Merge pull request #588 from mspruc/main

PLUS & MINUS arithmetic for filters in sql-api

Author: juripetersen

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

@@ -20,16 +20,15 @@
 
 import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 
-import org.apache.calcite.rel.core.Aggregate;
 import org.apache.calcite.rel.core.AggregateCall;
 
 import org.apache.wayang.api.sql.calcite.converter.functions.AggregateAddCols;
 import org.apache.wayang.api.sql.calcite.converter.functions.AggregateFunction;
 import org.apache.wayang.api.sql.calcite.converter.functions.AggregateKeyExtractor;
 import org.apache.wayang.api.sql.calcite.converter.functions.AggregateGetResult;
 import org.apache.wayang.api.sql.calcite.rel.WayangAggregate;
+
 import org.apache.wayang.basic.data.Record;
 import org.apache.wayang.basic.operators.GlobalReduceOperator;
 import org.apache.wayang.basic.operators.MapOperator;
@@ -48,10 +47,8 @@ public class WayangAggregateVisitor extends WayangRelNodeVisitor<WayangAggregate
     @Override
     Operator visit(final WayangAggregate wayangRelNode) {
         final Operator childOp = wayangRelConverter.convert(wayangRelNode.getInput(0));
-        Operator aggregateOperator;
 
-        final List<AggregateCall> aggregateCalls = ((Aggregate) wayangRelNode).getAggCallList();
-        final int groupCount = wayangRelNode.getGroupCount();
+        final List<AggregateCall> aggregateCalls = wayangRelNode.getAggCallList();
         final HashSet<Integer> groupingFields = new HashSet<>(wayangRelNode.getGroupSet().asSet());
 
         final MapOperator<Record, Record> mapOperator = new MapOperator<>(
@@ -60,22 +57,15 @@ Operator visit(final WayangAggregate wayangRelNode) {
                 Record.class);
         childOp.connectTo(0, mapOperator, 0);
 
-        if (groupCount > 0) {
-            ReduceByOperator<Record, Object> reduceByOperator;
-            reduceByOperator = new ReduceByOperator<>(
-                    new TransformationDescriptor<>(new AggregateKeyExtractor(groupingFields), Record.class, Object.class),
-                    new ReduceDescriptor<>(new AggregateFunction(aggregateCalls),
-                            DataUnitType.createGrouped(Record.class),
-                            DataUnitType.createBasicUnchecked(Record.class)));
-            aggregateOperator = reduceByOperator;
-        } else {
-            GlobalReduceOperator<Record> globalReduceOperator;
-            globalReduceOperator = new GlobalReduceOperator<>(
-                    new ReduceDescriptor<>(new AggregateFunction(aggregateCalls),
-                            DataUnitType.createGrouped(Record.class),
-                            DataUnitType.createBasicUnchecked(Record.class)));
-            aggregateOperator = globalReduceOperator;
-        }
+        final Operator aggregateOperator = wayangRelNode.getGroupCount() > 0 ? new ReduceByOperator<>(
+                new TransformationDescriptor<>(new AggregateKeyExtractor(groupingFields), Record.class, Object.class),
+                new ReduceDescriptor<>(new AggregateFunction(aggregateCalls),
+                        DataUnitType.createGrouped(Record.class),
+                        DataUnitType.createBasicUnchecked(Record.class)))
+                : new GlobalReduceOperator<>(
+                        new ReduceDescriptor<>(new AggregateFunction(aggregateCalls),
+                                DataUnitType.createGrouped(Record.class),
+                                DataUnitType.createBasicUnchecked(Record.class)));
 
         mapOperator.connectTo(0, aggregateOperator, 0);
 
@@ -85,6 +75,5 @@ Operator visit(final WayangAggregate wayangRelNode) {
                 Record.class);
         aggregateOperator.connectTo(0, mapOperator2, 0);
         return mapOperator2;
-
     }
 }

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

@@ -38,8 +38,7 @@ public class WayangFilterVisitor extends WayangRelNodeVisitor<WayangFilter> {
     @Override
     Operator visit(final WayangFilter wayangRelNode) {
         final Operator childOp = wayangRelConverter.convert(wayangRelNode.getInput(0));
-
-        final RexNode condition = ((Filter) wayangRelNode).getCondition();
+        final RexNode condition = wayangRelNode.getCondition();
 
         final FilterOperator<Record> filter = new FilterOperator<>(
                 new FilterPredicateImpl(condition),

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

@@ -18,7 +18,11 @@
 
 package org.apache.wayang.api.sql.calcite.converter.functions;
 
+import java.util.Date;
 import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.runtime.SqlFunctions;
 import org.apache.calcite.sql.SqlKind;
@@ -56,31 +60,54 @@ public SerializableFunction<List<Object>, Object> deriveOperation(final SqlKind
                     isLessThan(input.get(0), input.get(1)) || isEqualTo(input.get(0), input.get(1));
                 case AND -> input.stream().map(Boolean.class::cast).allMatch(Boolean::booleanValue);
                 case OR -> input.stream().map(Boolean.class::cast).anyMatch(Boolean::booleanValue);
+                case MINUS -> widenToDouble.apply(input.get(0)) - widenToDouble.apply(input.get(1));
+                case PLUS -> widenToDouble.apply(input.get(0)) + widenToDouble.apply(input.get(1));
                 default -> throw new UnsupportedOperationException("Kind not supported: " + kind);
             };
         }
     }
 
     /**
-     * Widening conversions
+     * Widens number types to optional double 
+     * see also {@link #widenToDouble} 
+     * @return Optional.empty if no conversion available, Optional.of(double) otherwise
+     */
+    final SerializableFunction<Object, Optional<Double>> widenToOptionalDouble = 
+                field -> field instanceof final Number number ? Optional.of(number.doubleValue())    :
+                         field instanceof final Date date     ? Optional.of((double) date.getTime()) :  
+                         Optional.empty();
+    
+
+    /**
+     * Consumes the option from {@link #widenToOptionalDouble()}, and eagerly provides the underlying double.
+     * @throws UnsupportedOperationException if conversion was not possible
      */
-    final SerializableFunction<Object, Comparable> ensureComparable = (a) -> a instanceof Integer val ? val.longValue() : (Comparable<?>) a;
+    final SerializableFunction<Object, Double> widenToDouble = field -> widenToOptionalDouble
+            .andThen(option -> option.orElseThrow(() -> new UnsupportedOperationException("Could not convert: " + option + " to double.")))
+            .apply(field);
+
+    /**
+     * Widening conversions, all numbers to double
+     */
+    final SerializableFunction<Object, Comparable> ensureComparable = field -> 
+        field instanceof Number || field instanceof Date ? widenToDouble.apply(field) : 
+        Comparable.class.cast(field);
+    
 
     /**
      * Java equivalent of SQL like clauses
+     * 
      * @param s1
      * @param s2
      * @return true if {@code s1} like {@code s2}
      */
     private boolean like(final String s1, final String s2) {
-        final SqlFunctions.LikeFunction likeFunction = new SqlFunctions.LikeFunction();
-        final boolean isMatch = likeFunction.like(s1, s2);
-
-        return isMatch;
+        return new SqlFunctions.LikeFunction().like(s1, s2);
     }
 
     /**
      * Java equivalent of sql greater than clauses
+     * 
      * @param o1
      * @param o2
      * @return true if {@code o1 > o2}
@@ -91,6 +118,7 @@ private boolean isGreaterThan(final Object o1, final Object o2) {
 
     /**
      * Java equivalent of sql less than clauses
+     * 
      * @param o1
      * @param o2
      * @return true if {@code o1 < o2}
@@ -101,11 +129,12 @@ private boolean isLessThan(final Object o1, final Object o2) {
 
     /**
      * Java equivalent of SQL equals clauses
+     * 
      * @param o1
      * @param o2
      * @return true if {@code o1 == o2}
      */
     private boolean isEqualTo(final Object o1, final Object o2) {
-        return ensureComparable.apply(o1).equals(ensureComparable.apply(o2));
+        return Objects.equals(ensureComparable.apply(o1), ensureComparable.apply(o2));
     }
 }