fetch, offset & limit for java platforms in sql-api

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

@@ -22,12 +22,14 @@
 
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelFieldCollation.Direction;
+import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.rex.RexNode;
 
+import org.apache.wayang.api.sql.calcite.converter.functions.SortFilter;
 import org.apache.wayang.api.sql.calcite.converter.functions.SortKeyExtractor;
 import org.apache.wayang.api.sql.calcite.rel.WayangSort;
 import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.operators.FilterOperator;
 import org.apache.wayang.basic.operators.SortOperator;
 import org.apache.wayang.core.function.TransformationDescriptor;
 import org.apache.wayang.core.plan.wayangplan.Operator;
@@ -45,12 +47,14 @@ Operator visit(final WayangSort wayangRelNode) {
 
         final Operator childOp = wayangRelConverter.convert(wayangRelNode.getInput());
 
-        //TODO: implement fetch & offset for java
-        final RexNode fetch = wayangRelNode.fetch;
-        final RexLiteral offset = (RexLiteral) wayangRelNode.offset;
+        // TODO: implement fetch & offset for java
+        final RexLiteral fetch = (RexLiteral) wayangRelNode.fetch;
+        final RexInputRef offset =  (RexInputRef) wayangRelNode.offset;
+
+        // if (fetch != null || offset != null) throw new
+        // UnsupportedOperationException("Offset and fetch currently not supported,
+        // these appear via LIMIT statements in SQL");
 
-        if (fetch != null || offset != null) throw new UnsupportedOperationException("Offset and fetch currently not supported, these appear via LIMIT statements in SQL");
-
         final RelCollation collation = wayangRelNode.getCollation();
 
         final List<Direction> collationDirections = collation.getFieldCollations().stream()
@@ -71,7 +75,16 @@ Operator visit(final WayangSort wayangRelNode) {
 
         childOp.connectTo(0, sort, 0);
 
-        return sort;
+
+        final SortFilter sortFilter = new SortFilter(
+                fetch != null ? RexLiteral.intValue(fetch) : Integer.MAX_VALUE,
+                offset != null ? RexLiteral.intValue(offset) : 0);
+
+        final FilterOperator<Record> filter = new FilterOperator<Record>(sortFilter, Record.class);
+
+        sort.connectTo(0, filter, 0);
+
+        return filter;
     }
 
 }

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

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.sql.calcite.converter.functions;
+
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.core.function.FunctionDescriptor;
+
+public class SortFilter implements FunctionDescriptor.SerializablePredicate<Record> {
+    final int offset;
+    final int fetch;
+    int increment;
+
+    /**
+     * The filter for a calcite/sql sort operator
+     * usually triggered by "LIMIT x", "OFFSET x", "FETCH x" statements
+     * @param offset amount of records ignored before accepting
+     * @param fetch amount of records accepted
+     */
+    public SortFilter(final int fetch, final int offset) {
+        this.fetch = fetch;
+        this.offset = offset;
+    }
+
+    @Override
+    public boolean test(final Record record) {
+        final boolean test = increment >= offset && increment <= fetch;
+        increment++;
+
+        return test;
+    }
+}

wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/rules/WayangSortRule.java

@@ -54,7 +54,7 @@ public RelNode convert(final RelNode rel) {
                 sort.getHints(),
                 newInput,
                 sort.collation,
-                sort.fetch,
-                sort.offset);
+                sort.offset,
+                sort.fetch);
     }
 }

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

@@ -328,7 +328,7 @@ public void filterWithLike() throws Exception {
                 assert (result.stream().anyMatch(rec -> rec.equals(new Record("test1", "test1", "test2"))));
         }
 
-        //@Test
+        @Test
         public void javaLimit() throws Exception {
                 final SqlContext sqlContext = createSqlContext("/data/exampleSort.csv");