Upgrade fetch first n rows

modules/calcite/pom.xml

@@ -341,6 +341,10 @@
                                     outputRoot="${project.build.directory}/generated-sources/fmpp"
                                     data="tdd(${project.build.directory}/codegen/config.fmpp), default: tdd(${project.build.directory}/codegen/default_config.fmpp)"
                                 />
+                                <patch
+                                    dir="${project.build.directory}/generated-sources/fmpp"
+                                    patchfile="${project.build.directory}/codegen/patches/select-fetch-expression.patch"
+                                />
                             </target>
                         </configuration>
                     </execution>

modules/calcite/src/main/codegen/includes/parserImpls.ftl

@@ -816,3 +816,17 @@ SqlDrop SqlDropView(Span s, boolean replace) :
         return SqlDdlNodes.dropView(s.end(this), ifExists, id);
     }
 }
+
+JAVACODE
+SqlNode FetchCount() {
+    SqlNode e;
+    if (getToken(1).kind == LPAREN) {
+        jj_consume_token(LPAREN);
+        e = Expression(ExprContext.ACCEPT_NON_QUERY);
+        jj_consume_token(RPAREN);
+    }
+    else
+        e = UnsignedNumericLiteralOrParam();
+
+    return e;
+}

modules/calcite/src/main/codegen/patches/select-fetch-expression.patch

@@ -0,0 +1,26 @@
+#
+# 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.
+#
+--- javacc/Parser.jj
++++ javacc/Parser.jj
+@@ -754,7 +754,7 @@
+ {
+     // SQL:2008-style syntax. "OFFSET ... FETCH ...".
+     // If you specify both LIMIT and FETCH, FETCH wins.
+-    <FETCH> ( <FIRST> | <NEXT> ) offsetFetch[1] = UnsignedNumericLiteralOrParam()
++    <FETCH> ( <FIRST> | <NEXT> ) offsetFetch[1] = FetchCount()
+     ( <ROW> | <ROWS> ) <ONLY>
+ }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java

@@ -626,8 +626,8 @@ private boolean hasExchange(RelNode rel) {
 
     /** {@inheritDoc} */
     @Override public Node<Row> visit(IgniteLimit rel) {
-        Supplier<Integer> offset = (rel.offset() == null) ? null : expressionFactory.execute(rel.offset());
-        Supplier<Integer> fetch = (rel.fetch() == null) ? null : expressionFactory.execute(rel.fetch());
+        Supplier<Number> offset = (rel.offset() == null) ? null : expressionFactory.execute(rel.offset());
+        Supplier<Number> fetch = (rel.fetch() == null) ? null : expressionFactory.execute(rel.fetch());
 
         LimitNode<Row> node = new LimitNode<>(ctx, rel.getRowType(), offset, fetch);
 
@@ -642,8 +642,8 @@ private boolean hasExchange(RelNode rel) {
     @Override public Node<Row> visit(IgniteSort rel) {
         RelCollation collation = rel.getCollation();
 
-        Supplier<Integer> offset = (rel.offset == null) ? null : expressionFactory.execute(rel.offset);
-        Supplier<Integer> fetch = (rel.fetch == null) ? null : expressionFactory.execute(rel.fetch);
+        Supplier<Number> offset = (rel.offset == null) ? null : expressionFactory.execute(rel.offset);
+        Supplier<Number> fetch = (rel.fetch == null) ? null : expressionFactory.execute(rel.fetch);
 
         SortNode<Row> node = new SortNode<>(ctx, rel.getRowType(), expressionFactory.comparator(collation), offset,
             fetch);

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java

@@ -35,7 +35,7 @@ public class LimitNode<Row> extends AbstractNode<Row> implements SingleNode<Row>
     private int rowsProcessed;
 
     /** Fetch can be unset, in this case we need all rows. */
-    private @Nullable Supplier<Integer> fetchNode;
+    private final @Nullable Supplier<Number> fetchNode;
 
     /** Waiting results counter. */
     private int waiting;
@@ -49,13 +49,13 @@ public class LimitNode<Row> extends AbstractNode<Row> implements SingleNode<Row>
     public LimitNode(
         ExecutionContext<Row> ctx,
         RelDataType rowType,
-        Supplier<Integer> offsetNode,
-        Supplier<Integer> fetchNode
+        @Nullable Supplier<Number> offsetNode,
+        @Nullable Supplier<Number> fetchNode
     ) {
         super(ctx, rowType);
 
-        offset = offsetNode == null ? 0 : offsetNode.get();
-        fetch = fetchNode == null ? 0 : fetchNode.get();
+        offset = RelNodeUtils.limitValueWithCheck(offsetNode, "OFFSET");
+        fetch = RelNodeUtils.limitValueWithCheck(fetchNode, "FETCH");
         this.fetchNode = fetchNode;
     }
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RelNodeUtils.java

@@ -0,0 +1,56 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.exec.rel;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.function.Supplier;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.jetbrains.annotations.Nullable;
+
+/** Utility class for rel nodes containing useful methods and constants. */
+class RelNodeUtils {
+    /** Decimal of Integer.MAX_VALUE for fetch/offset bounding. */
+    private static final BigDecimal DEC_INT_MAX = BigDecimal.valueOf(Integer.MAX_VALUE);
+
+    /** */
+    static int limitValueWithCheck(@Nullable Supplier<Number> s, String name) {
+        if (s == null)
+            return 0;
+
+        Number n = s.get();
+
+        if (n == null)
+            throw new IgniteSQLException(name + " must not be null");
+        else if (n instanceof Double || n instanceof Float) {
+            double v = n.doubleValue();
+
+            if (!Double.isFinite(v))
+                throw new IgniteSQLException(name + " must be an finite number");
+        }
+
+        BigDecimal v = new BigDecimal(n.toString()).setScale(0, RoundingMode.DOWN);
+
+        if (v.compareTo(BigDecimal.ZERO) < 0)
+            throw new IgniteSQLException(name + " must not be negative");
+        else if (v.compareTo(DEC_INT_MAX) > 0)
+            throw new IgniteSQLException(name + " must not be greater than " + DEC_INT_MAX);
+
+        return v.intValue();
+    }
+}

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java

@@ -58,15 +58,13 @@ public class SortNode<Row> extends MemoryTrackingNode<Row> implements SingleNode
     public SortNode(
         ExecutionContext<Row> ctx, RelDataType rowType,
         Comparator<Row> comp,
-        @Nullable Supplier<Integer> offset,
-        @Nullable Supplier<Integer> fetch
+        @Nullable Supplier<Number> offset,
+        @Nullable Supplier<Number> fetch
     ) {
         super(ctx, rowType);
 
-        assert fetch == null || fetch.get() >= 0;
-        assert offset == null || offset.get() >= 0;
-
-        limit = fetch == null ? -1 : fetch.get() + (offset == null ? 0 : offset.get());
+        limit = fetch == null ?
+            -1 : RelNodeUtils.limitValueWithCheck(fetch, "FETCH") + RelNodeUtils.limitValueWithCheck(offset, "OFFSET");
 
         if (limit < 0)
             rows = new PriorityQueue<>(comp);

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlToRelConvertor.java

@@ -19,22 +19,30 @@
 
 import java.util.ArrayDeque;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Deque;
 import java.util.List;
+import java.util.Set;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.prepare.Prepare;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelRoot;
 import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rel.logical.LogicalJoin;
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.calcite.rel.logical.LogicalTableModify;
 import org.apache.calcite.rel.logical.LogicalValues;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexDynamicParam;
 import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.SqlCall;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlInsert;
@@ -46,11 +54,15 @@
 import org.apache.calcite.sql.validate.SqlValidator;
 import org.apache.calcite.sql.validate.SqlValidatorScope;
 import org.apache.calcite.sql.validate.SqlValidatorUtil;
+import org.apache.calcite.sql2rel.RelFieldTrimmer;
 import org.apache.calcite.sql2rel.SqlRexConvertletTable;
 import org.apache.calcite.sql2rel.SqlToRelConverter;
 import org.apache.calcite.tools.RelBuilder;
 import org.apache.calcite.util.ControlFlowException;
+import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.calcite.util.Pair;
+import org.apache.calcite.util.mapping.Mapping;
+import org.apache.calcite.util.mapping.Mappings;
 import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
 import org.jetbrains.annotations.Nullable;
 
@@ -87,6 +99,11 @@ public IgniteSqlToRelConvertor(
             return super.convertQueryRecursive(qry, top, targetRowType);
     }
 
+    /** {@inheritDoc} */
+    @Override protected RelFieldTrimmer newFieldTrimmer() {
+        return new IgniteRelFieldTrimmer(validator, relBuilder);
+    }
+
     /** {@inheritDoc} */
     @Override protected RelNode convertInsert(SqlInsert call) {
         datasetStack.push(call);
@@ -199,6 +216,58 @@ private boolean hasDefaults(SqlCall call) {
         }
     }
 
+    /** Field trimmer that preserves expression-based OFFSET/FETCH nodes. */
+    public static class IgniteRelFieldTrimmer extends RelFieldTrimmer {
+        /** */
+        private IgniteRelFieldTrimmer(@Nullable SqlValidator validator, RelBuilder relBuilder) {
+            super(validator, relBuilder);
+        }
+
+        /** {@inheritDoc} */
+        @Override public TrimResult trimFields(
+            Sort sort,
+            ImmutableBitSet fieldsUsed,
+            Set<RelDataTypeField> extraFields
+        ) {
+            if (supportedByRelBuilder(sort.offset) && supportedByRelBuilder(sort.fetch))
+                return super.trimFields(sort, fieldsUsed, extraFields);
+
+            RelCollation collation = sort.getCollation();
+            RelNode input = sort.getInput();
+            int fieldCnt = sort.getRowType().getFieldCount();
+
+            ImmutableBitSet.Builder inputFieldsUsed = fieldsUsed.rebuild();
+
+            for (RelFieldCollation field : collation.getFieldCollations())
+                inputFieldsUsed.set(field.getFieldIndex());
+
+            TrimResult trimRes = trimChild(sort, input, inputFieldsUsed.build(), Collections.emptySet());
+            RelNode newInput = trimRes.left;
+            Mapping inputMapping = trimRes.right;
+
+            if (newInput == input && inputMapping.isIdentity() && fieldsUsed.cardinality() == fieldCnt)
+                return result(sort, Mappings.createIdentity(fieldCnt));
+
+            RelNode newSort = sort.copy(
+                sort.getTraitSet(),
+                newInput,
+                RexUtil.apply(inputMapping, collation),
+                sort.offset,
+                sort.fetch
+            );
+
+            return result(newSort, inputMapping, sort);
+        }
+
+        /**
+         * @param node Rex node.
+         * @return {@code true} if Calcite RelBuilder accepts the node for OFFSET/FETCH.
+         */
+        private static boolean supportedByRelBuilder(@Nullable RexNode node) {
+            return node == null || node instanceof RexLiteral || node instanceof RexDynamicParam;
+        }
+    }
+
     /**
      * This method was copy-pasted from super-method except this changes:
      * - For updateCall we require all columns in the project and should not skip anything.