IGNITE-28844 Calcite. Improve type checking in LIMIT / OFFSET clauses

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

@@ -44,7 +44,9 @@
 import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.calcite.util.mapping.IntPair;
+import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
 import org.apache.ignite.internal.processors.failure.FailureProcessor;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
 import org.apache.ignite.internal.processors.query.QueryUtils;
 import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler.RowFactory;
 import org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactory;
@@ -126,6 +128,8 @@
 import org.apache.ignite.internal.processors.query.calcite.trait.TraitUtils;
 import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
 import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteMath;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteResource;
 import org.apache.ignite.internal.processors.query.calcite.util.RexUtils;
 import org.apache.ignite.internal.util.typedef.F;
 import org.jetbrains.annotations.Nullable;
@@ -567,8 +571,8 @@ private boolean hasExchange(RelNode rel) {
                 ctx,
                 rowType,
                 idxBndRel.first() ? cmp : cmp.reversed(),
-                null,
-                () -> 1
+                0,
+                1
             );
 
             sortNode.register(scanNode);
@@ -630,8 +634,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());
+        long offset = validateAndGetOffset(rel.offset(), LimitNode.OFFSET_DEFAULT);
+        long fetch = validateAndGetFetch(rel.fetch(), LimitNode.FETCH_DEFAULT);
 
         LimitNode<Row> node = new LimitNode<>(ctx, rel.getRowType(), offset, fetch);
 
@@ -646,8 +650,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);
+        long offset = validateAndGetOffset(rel.offset, SortNode.OFFSET_DEFAULT);
+        long fetch = validateAndGetFetch(rel.fetch, SortNode.FETCH_DEFAULT);
 
         SortNode<Row> node = new SortNode<>(ctx, rel.getRowType(), expressionFactory.comparator(collation), offset,
             fetch);
@@ -659,6 +663,16 @@ private boolean hasExchange(RelNode rel) {
         return node;
     }
 
+    /** */
+    private long validateAndGetOffset(RexNode node, long defaultVal) {
+        return node == null ? defaultVal : validateAndGetFetchOffsetParams(node, "offset");
+    }
+
+    /** */
+    private long validateAndGetFetch(RexNode node, long defaultVal) {
+        return node == null ? defaultVal : validateAndGetFetchOffsetParams(node, "fetch");
+    }
+
     /** {@inheritDoc} */
     @Override public Node<Row> visit(IgniteTableSpool rel) {
         TableSpoolNode<Row> node = new TableSpoolNode<>(ctx, rel.getRowType(), rel.readType == Spool.Type.LAZY);
@@ -1050,4 +1064,33 @@ private ScanStorageNode<Row> createStorageScan(
             otherColMapping
         );
     }
+
+    /** */
+    private long validateAndGetFetchOffsetParams(RexNode node, String op) {
+        Supplier<Object> scalar = expressionFactory.execute(node);
+        Object param = scalar.get();
+
+        if (!(param instanceof Number)) {
+            String actual = param == null ? "null" : param.getClass().getSimpleName();
+            throw new IgniteSQLException(IgniteResource.INSTANCE.incorrectDynamicParameterType("BIGINT", actual).str(),
+                IgniteQueryErrorCode.UNEXPECTED_ELEMENT_TYPE);
+        }
+
+        long paramAsLong;
+
+        try {
+            paramAsLong = IgniteMath.convertToLongExact((Number)param);
+        }
+        catch (RuntimeException ex) {
+            throw new IgniteSQLException(IgniteResource.INSTANCE.illegalFetchLimit(op).str(),
+                IgniteQueryErrorCode.UNEXPECTED_ELEMENT_TYPE, ex);
+        }
+
+        if (paramAsLong < 0) {
+            throw new IgniteSQLException(IgniteResource.INSTANCE.illegalFetchLimit(op).str(),
+                IgniteQueryErrorCode.UNEXPECTED_ELEMENT_TYPE);
+        }
+
+        return paramAsLong;
+    }
 }

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

@@ -37,6 +37,9 @@
  * Abstract node of execution tree.
  */
 public abstract class AbstractNode<Row> implements Node<Row> {
+    /** Special flag which marks that all the rows are received. */
+    static final int NOT_WAITING = -1;
+
     /** */
     public static final int IN_BUFFER_SIZE = IgniteSystemProperties.getInteger(IGNITE_CALCITE_EXEC_IN_BUFFER_SIZE, 512);
 

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

@@ -29,9 +29,6 @@ public abstract class AbstractRightMaterializedJoinNode<Row> extends MemoryTrack
     /** */
     protected static final int HALF_BUF_SIZE = IN_BUFFER_SIZE >> 1;
 
-    /** Special flag which marks that all the rows are received. */
-    protected static final int NOT_WAITING = -1;
-
     /** */
     protected final Deque<Row> leftInBuf = new ArrayDeque<>(IN_BUFFER_SIZE);
 

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

@@ -17,66 +17,79 @@
 
 package org.apache.ignite.internal.processors.query.calcite.exec.rel;
 
-import java.util.function.Supplier;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteMath;
 import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.Nullable;
 
 /** Offset, fetch|limit support node. */
 public class LimitNode<Row> extends AbstractNode<Row> implements SingleNode<Row>, Downstream<Row> {
-    /** Offset if its present, otherwise 0. */
-    private final int offset;
+    /** */
+    public static final long FETCH_DEFAULT = -1;
 
-    /** Fetch if its present, otherwise 0. */
-    private final int fetch;
+    /** */
+    public static final long OFFSET_DEFAULT = 0;
 
-    /** Already processed (pushed to upstream) rows count. */
-    private int rowsProcessed;
+    /** Offset param. */
+    private final long offset;
 
-    /** Fetch can be unset, in this case we need all rows. */
-    private @Nullable Supplier<Integer> fetchNode;
+    /** How many rows need to be processed, if {@code 0} it depends on {@link #rowsSummary}. */
+    private final long fetch;
+
+    /** Summary rows to process. */
+    private final long rowsSummary;
+
+    /** Already processed (pushed to downstream) rows count. */
+    private long rowsProcessed;
 
     /** Waiting results counter. */
     private int waiting;
 
+    /** Upper requested rows. */
+    private int requested;
+
     /**
      * Constructor.
      *
      * @param ctx Execution context.
      * @param rowType Row type.
+     * @param offset How many rows need to be skipped.
+     * @param fetch How many rows need to be processed, {@link #FETCH_DEFAULT} if param is undefined.
      */
     public LimitNode(
         ExecutionContext<Row> ctx,
         RelDataType rowType,
-        Supplier<Integer> offsetNode,
-        Supplier<Integer> fetchNode
+        long offset,
+        long fetch
     ) {
         super(ctx, rowType);
 
-        offset = offsetNode == null ? 0 : offsetNode.get();
-        fetch = fetchNode == null ? 0 : fetchNode.get();
-        this.fetchNode = fetchNode;
+        this.offset = offset;
+        rowsSummary = fetch == FETCH_DEFAULT ? Long.MAX_VALUE : IgniteMath.addExact(fetch, offset);
+        this.fetch = fetch == FETCH_DEFAULT ? 0 : fetch;
     }
 
     /** {@inheritDoc} */
     @Override public void request(int rowsCnt) throws Exception {
         assert !F.isEmpty(sources()) && sources().size() == 1;
         assert rowsCnt > 0;
 
-        if (fetchNone()) {
+        if (!hasMoreData()) {
             end();
 
             return;
         }
 
-        if (offset > 0 && rowsProcessed == 0)
-            rowsCnt = offset + rowsCnt;
+        assert requested == 0 : requested;
+        requested = rowsCnt;
 
-        waiting = rowsCnt;
+        if (fetch > 0) {
+            long remain = rowsSummary - rowsProcessed;
 
-        if (fetch > 0)
-            rowsCnt = Math.min(rowsCnt, (fetch + offset) - rowsProcessed);
+            rowsCnt = remain > rowsCnt ? rowsCnt : (int)remain;
+        }
+
+        waiting = rowsCnt;
 
         checkState();
 
@@ -85,38 +98,49 @@ public LimitNode(
 
     /** {@inheritDoc} */
     @Override public void push(Row row) throws Exception {
-        if (waiting == -1)
+        if (waiting == NOT_WAITING)
             return;
 
-        ++rowsProcessed;
-
         --waiting;
 
-        checkState();
-
-        if (rowsProcessed > offset) {
-            if (fetchNode == null || (fetchNode != null && rowsProcessed <= fetch + offset))
-                downstream().push(row);
+        if (rowsProcessed >= offset && hasMoreData()) {
+            // This two rows can`t be swapped, cause if all requested rows have been pushed it will trigger further request call.
+            --requested;
+            downstream().push(row);
         }
 
-        if (fetch > 0 && rowsProcessed == fetch + offset && waiting > 0)
+        ++rowsProcessed;
+
+        // There several cases are possible:
+        //  1) requested = 512, limit = 1, offset = not defined: need to pass 1 row and call end()
+        //  2) requested = 512, limit = 512, offset = not defined: just need to pass all rows without end() call
+        //  3) requested = 512, limit = 512, offset = 1: need to request initially 512 and further 1 row
+        if (!hasMoreData() && requested > 0)
             end();
+
+        if (waiting == 0 && requested > 0)
+            source().request(waiting = requested);
     }
 
     /** {@inheritDoc} */
     @Override public void end() throws Exception {
-        if (waiting == -1)
+        if (waiting == NOT_WAITING)
             return;
 
         assert downstream() != null;
 
-        waiting = -1;
+        waiting = NOT_WAITING;
+
+        if (requested > 0)
+            requested = 0;
 
         downstream().end();
     }
 
     /** {@inheritDoc} */
     @Override protected void rewindInternal() {
+        waiting = 0;
+        requested = 0;
         rowsProcessed = 0;
     }
 
@@ -128,8 +152,8 @@ public LimitNode(
         return this;
     }
 
-    /** {@code True} if requested 0 results, or all already processed. */
-    private boolean fetchNone() {
-        return (fetchNode != null && fetch == 0) || (fetch > 0 && rowsProcessed == fetch + offset);
+    /** {@code True} If current rows processed is less than required or undefined. */
+    private boolean hasMoreData() {
+        return rowsProcessed < rowsSummary;
     }
 }

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

@@ -35,9 +35,6 @@ public abstract class MergeJoinNode<Row> extends AbstractNode<Row> {
     /** */
     private static final int HALF_BUF_SIZE = IN_BUFFER_SIZE >> 1;
 
-    /** Special value to highlights that all row were received and we are not waiting any more. */
-    protected static final int NOT_WAITING = -1;
-
     /** */
     protected final Comparator<Row> comp;