Merge branch 'main' into migrate_to_junit5

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

Author: strangelookingnerd

wayang-api/wayang-api-sql/pom.xml

@@ -26,22 +26,26 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
+    <properties>
+        <calcite.version>1.39.0</calcite.version>       
+    </properties>
+    
     <artifactId>wayang-api-sql</artifactId>
     <dependencies>
         <dependency>
             <groupId>org.apache.calcite</groupId>
             <artifactId>calcite-core</artifactId>
-            <version>1.32.0</version>
+            <version>${calcite.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.calcite</groupId>
             <artifactId>calcite-linq4j</artifactId>
-            <version>1.32.0</version>
+            <version>${calcite.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.calcite</groupId>
             <artifactId>calcite-file</artifactId>
-            <version>1.29.0</version>
+            <version>${calcite.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.wayang</groupId>

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

@@ -18,6 +18,7 @@
 
 package org.apache.wayang.api.sql.calcite.converter;
 
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 
@@ -51,7 +52,7 @@ Operator visit(final WayangAggregate wayangRelNode) {
 
         final List<AggregateCall> aggregateCalls = ((Aggregate) wayangRelNode).getAggCallList();
         final int groupCount = wayangRelNode.getGroupCount();
-        final Set<Integer> groupingFields = wayangRelNode.getGroupSet().asSet();
+        final HashSet<Integer> groupingFields = new HashSet<>(wayangRelNode.getGroupSet().asSet());
 
         final MapOperator<Record, Record> mapOperator = new MapOperator<>(
                 new AggregateAddCols(aggregateCalls),

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

@@ -18,37 +18,42 @@
 package org.apache.wayang.api.sql.calcite.converter.functions;
 
 import java.util.List;
+import java.util.stream.Collectors;
 
 import org.apache.calcite.rel.core.AggregateCall;
-
+import org.apache.calcite.sql.SqlKind;
 import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.data.Tuple2;
 import org.apache.wayang.core.function.FunctionDescriptor;
 
 public class AggregateAddCols implements FunctionDescriptor.SerializableFunction<Record, Record> {
-    final List<AggregateCall> aggregateCalls;
-    
-    public AggregateAddCols(final List<AggregateCall> aggregateCalls){
-        this.aggregateCalls = aggregateCalls;
+    final List<Tuple2<SqlKind, List<Integer>>> aggregateCalls;
+
+    public AggregateAddCols(final List<AggregateCall> aggregateCalls) {
+        this.aggregateCalls = aggregateCalls.stream().map(call -> new Tuple2<>(call.getAggregation().getKind(), call.getArgList())).collect(Collectors.toList());
     }
 
     @Override
     public Record apply(final Record record) {
         final int l = record.size();
         final int newRecordSize = l + aggregateCalls.size() + 1;
         final Object[] resValues = new Object[newRecordSize];
-        
+
         for (int i = 0; i < l; i++) {
             resValues[i] = record.getField(i);
         }
 
         int i = l;
-        for (final AggregateCall aggregateCall : aggregateCalls) {
-            switch (aggregateCall.getAggregation().kind) {
+        for (final Tuple2<SqlKind, List<Integer>> aggregateCall : aggregateCalls) {
+            final SqlKind kind = aggregateCall.field0;
+            final List<Integer> argList = aggregateCall.field1;
+
+            switch (kind) {
                 case COUNT:
                     resValues[i] = 1;
                     break;
                 default:
-                    resValues[i] = record.getField(aggregateCall.getArgList().get(0));
+                    resValues[i] = record.getField(argList.get(0));
             }
             i++;
         }

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

@@ -17,40 +17,44 @@
  */
 package org.apache.wayang.api.sql.calcite.converter.functions;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.BiFunction;
+import java.util.stream.Collectors;
 
 import org.apache.calcite.rel.core.AggregateCall;
 import org.apache.calcite.runtime.SqlFunctions;
+import org.apache.calcite.sql.SqlKind;
 import org.apache.wayang.basic.data.Record;
 import org.apache.wayang.core.function.FunctionDescriptor;
 
 public class AggregateFunction
         implements FunctionDescriptor.SerializableBinaryOperator<Record> {
-
-    final List<AggregateCall> aggregateCalls;
+    final List<SqlKind> aggregateKinds;
 
     public AggregateFunction(final List<AggregateCall> aggregateCalls) {
-        this.aggregateCalls = aggregateCalls;
+        this.aggregateKinds = aggregateCalls.stream()
+            .map(call -> call.getAggregation().getKind())
+            .collect(Collectors.toList());
     }
 
     @Override
     public Record apply(final Record record1, final Record record2) {
         final int l = record1.size();
         final Object[] resValues = new Object[l];
-        final boolean countDone = false;
+        boolean countDone = false;
 
-        for (int i = 0; i < l - aggregateCalls.size() - 1; i++) {
+        for (int i = 0; i < l - aggregateKinds.size() - 1; i++) {
             resValues[i] = record1.getField(i);
         }
 
-        int counter = l - aggregateCalls.size() - 1;
-        for (final AggregateCall aggregateCall : aggregateCalls) {
+        int counter = l - aggregateKinds.size() - 1;
+        for (final SqlKind kind : aggregateKinds) {
             final Object field1 = record1.getField(counter);
             final Object field2 = record2.getField(counter);
 
-            switch (aggregateCall.getAggregation().kind) {
+            switch (kind) {
                 case SUM:
                     resValues[counter] = this.castAndMap(field1, field2, null, Long::sum, Integer::sum, Double::sum);
                     break;
@@ -61,25 +65,33 @@ public Record apply(final Record record1, final Record record2) {
                 case MAX:
                     resValues[counter] = this.castAndMap(field1, field2, SqlFunctions::greatest, SqlFunctions::greatest,
                             SqlFunctions::greatest, SqlFunctions::greatest);
+                    break;
                 case COUNT:
                     // since aggregates inject an extra column for counting before,
                     // see AggregateAddCols. the column we operate on are integer counts,
                     // which means we can eagerly get the fields as integers and simply sum
                     assert (field1 instanceof Integer && field2 instanceof Integer)
                             : "Expected to find integers for count but found: " + field1 + " and " + field2;
-                    Object obj = Integer.class.cast(field1) + Integer.class.cast(field2);
-                    resValues[counter] = obj;
+                    final Object count = Integer.class.cast(field1) + Integer.class.cast(field2);
+                    resValues[counter] = count;
                     break;
                 case AVG:
-                    throw new UnsupportedOperationException("Averages not currently supported");
-                // resValues[counter] = this.castAndMap(field1, field2, null, null, null, null);
-                // break;
+                    assert (field1 instanceof Integer && field2 instanceof Integer)
+                            : "Expected to find integers for count but found: " + field1 + " and " + field2;
+                    final Object avg = Integer.class.cast(field1) + Integer.class.cast(field2);
+
+                    resValues[counter] = avg;
+
+                    if (!countDone) {
+                        resValues[l - 1] = record1.getInt(l - 1) + record2.getInt(l - 1);
+                        countDone = true;
+                    }
+                    break;
                 default:
-                    throw new IllegalStateException("Unsupported operation: " + aggregateCall.getAggregation().kind);
+                    throw new IllegalStateException("Unsupported operation: " + kind);
             }
             counter++;
         }
-
         return new Record(resValues);
     }
 

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

@@ -18,49 +18,48 @@
 
 package org.apache.wayang.api.sql.calcite.converter.functions;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
 
 import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.sql.SqlKind;
 import org.apache.wayang.basic.data.Record;
 import org.apache.wayang.core.function.FunctionDescriptor;
 
 public class AggregateGetResult implements FunctionDescriptor.SerializableFunction<Record, Record> {
-    private final List<AggregateCall> aggregateCallList;
-    private final Set<Integer> groupingfields;
+        private final List<SqlKind> aggregateKindList;
+        private final Set<Integer> groupingfields;
 
-    public AggregateGetResult(final List<AggregateCall> aggregateCalls, final Set<Integer> groupingfields) {
-        this.aggregateCallList = aggregateCalls;
-        this.groupingfields = groupingfields;
-    }
+        public AggregateGetResult(final List<AggregateCall> aggregateCalls, final Set<Integer> groupingfields) {
+                this.aggregateKindList = aggregateCalls.stream()
+                                .map(call -> call.getAggregation().getKind())
+                                .collect(Collectors.toList());
+                this.groupingfields = groupingfields;
+        }
 
-    @Override
-    public Record apply(final Record record) {
-        final int l = record.size();
-        final int outputRecordSize = aggregateCallList.size() + groupingfields.size();
-        final Object[] resValues = new Object[outputRecordSize];
+        @Override
+        public Record apply(final Record record) {
+                final int recordSize = record.size();
+                final int aggregateCallOffset = recordSize - aggregateKindList.size() - 1;
 
-        int i = 0;
-        int j = 0;
-        for (i = 0; j < groupingfields.size(); i++) {
-            if (groupingfields.contains(i)) {
-                resValues[j] = record.getField(i);
-                j++;
-            }
-        }
+                final Object[] fields = groupingfields.stream()
+                                .map(record::getField)
+                                .toArray();
 
-        i = l - aggregateCallList.size() - 1;
-        for (final AggregateCall aggregateCall : aggregateCallList) {
-            final String name = aggregateCall.getAggregation().getName();
-            if (name.equals("AVG")) {
-                resValues[j] = record.getDouble(i) / record.getDouble(l - 1);
-            } else {
-                resValues[j] = record.getField(i);
-            }
-            j++;
-            i++;
-        }
+                final Object[] aggregateCallFields = IntStream.range(0, aggregateKindList.size())
+                                .mapToObj(i -> aggregateKindList.get(i).equals(SqlKind.AVG)
+                                                ? record.getDouble(i + aggregateCallOffset)
+                                                                / record.getDouble(recordSize - 1)
+                                                : record.getField(i + aggregateCallOffset))
+                                .toArray();
 
-        return new Record(resValues);
-    }
+                final Object[] combinedFields = Stream.concat(Arrays.stream(fields), Arrays.stream(aggregateCallFields))
+                                .toArray();
+
+                return new Record(combinedFields);
+        }
 }

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

@@ -19,16 +19,16 @@
 package org.apache.wayang.api.sql.calcite.converter.functions;
 
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 
 import org.apache.wayang.basic.data.Record;
 import org.apache.wayang.core.function.FunctionDescriptor;
 
 public class AggregateKeyExtractor implements FunctionDescriptor.SerializableFunction<Record, Object> {
-    private final Set<Integer> indexSet;
+    private final HashSet<Integer> indexSet;
 
-    public AggregateKeyExtractor(final Set<Integer> indexSet) {
+    public AggregateKeyExtractor(final HashSet<Integer> indexSet) {
         this.indexSet = indexSet;
     }
 

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

@@ -78,7 +78,7 @@ public boolean eval(final Record record, final SqlKind kind, final RexNode leftO
 
             switch (kind) {
                 case LIKE:
-                    return SqlFunctions.like(field.toString(), rexLiteral.toString().replace("'", ""));
+                    return like(field.toString(), rexLiteral.toString().replace("'", ""));
                 case GREATER_THAN:
                     return isGreaterThan(field, rexLiteral);
                 case LESS_THAN:
@@ -127,6 +127,13 @@ public boolean eval(final Record record, final SqlKind kind, final RexNode leftO
         }
     }
 
+    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;
+    }
+
     private boolean isGreaterThan(final Object o, final RexLiteral rexLiteral) {
         // return rexLiteral.getValue().compareTo(o)< 0;
         return ((Comparable) o).compareTo(rexLiteral.getValueAs(o.getClass())) > 0;

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

@@ -39,9 +39,9 @@ public SortFilter(final int fetch, final int offset) {
 
     @Override
     public boolean test(final Record record) {
-        final boolean test = increment >= offset && increment <= fetch;
         increment++;
-
+        final boolean test = increment >= offset && increment <= fetch;
+        
         return test;
     }
 }