Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zstan wants to merge 4 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
02859d4
IGNITE-28844 Calcite. Improve type checking in LIMIT / OFFSET clauses
zstan Jun 30, 2026
d066d4a
fix test
zstan Jul 7, 2026
6201e62
remove checker class
zstan Jul 18, 2026
1d8b892
fix
zstan Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 39 additions & 6 deletions .../java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -567,8 +571,8 @@ private boolean hasExchange(RelNode rel) {
ctx,
rowType,
idxBndRel.first() ? cmp : cmp.reversed(),
null,
() -> 1
0,
1
);

sortNode.register(scanNode);
Expand Down Expand Up @@ -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 = rel.offset() == null ? 0 : validateAndGetFetchOffsetParams(rel.offset(), "offset");
long fetch = rel.fetch() == null ? -1 : validateAndGetFetchOffsetParams(rel.fetch(), "fetch");

LimitNode<Row> node = new LimitNode<>(ctx, rel.getRowType(), offset, fetch);

Expand All @@ -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 = rel.offset == null ? 0 : validateAndGetFetchOffsetParams(rel.offset, "offset");
long fetch = rel.fetch == null ? -1 : validateAndGetFetchOffsetParams(rel.fetch, "fetch");

SortNode<Row> node = new SortNode<>(ctx, rel.getRowType(), expressionFactory.comparator(collation), offset,
fetch);
Expand Down Expand Up @@ -1050,4 +1054,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;
}
}
3 changes: 3 additions & 0 deletions .../main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,9 @@
* Abstract node of execution tree.
*/
public abstract class AbstractNode<Row> implements Node<Row> {
/** Special value to highlight that all rows were received and we do not expect more. */
static final int NOT_WAITING = -1;

/** */
protected static final int IN_BUFFER_SIZE = IgniteSystemProperties.getInteger(IGNITE_CALCITE_EXEC_IN_BUFFER_SIZE, 512);

Expand Down
81 changes: 47 additions & 34 deletions ...src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,29 +17,31 @@

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;
/** Offset param. */
private final long offset;

/** Fetch if its present, otherwise 0. */
private final int fetch;
/** Fetch param. */
private final long fetch;

/** Fetch can be unset. */
private final boolean fetchUndefined;

@alex-plekhanov alex-plekhanov Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's add long limit = fetch == -1 ? Long.MAX_VALUE : IgniteMath.addExact(fetch, offset) instead. To execute IgniteMath.addExact(fetch, offset) only once (and not for each request and each push), to simplify hasMoreData (rowsProcessed < limit).


/** Already processed (pushed to upstream) rows count. */
private int rowsProcessed;

@alex-plekhanov alex-plekhanov Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comparing int with long, can overflow


/** Fetch can be unset, in this case we need all rows. */
private @Nullable Supplier<Integer> fetchNode;

/** Waiting results counter. */
private int waiting;

/** Upper requested rows. */
private int requested;

/**
* Constructor.
*
Expand All @@ -49,34 +51,37 @@ public class LimitNode<Row> extends AbstractNode<Row> implements SingleNode<Row>
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;
fetchUndefined = fetch == -1;
this.fetch = fetch == -1 ? 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 = IgniteMath.addExact(fetch, offset) - rowsProcessed;

if (fetch > 0)
rowsCnt = Math.min(rowsCnt, (fetch + offset) - rowsProcessed);
rowsCnt = remain > rowsCnt ? rowsCnt : (int)remain;
}

waiting = rowsCnt;

checkState();

Expand All @@ -85,38 +90,46 @@ 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.

@alex-plekhanov alex-plekhanov Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment should be started with an uppercase letter.

--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;

downstream().end();
}

/** {@inheritDoc} */
@Override protected void rewindInternal() {
waiting = 0;
requested = 0;
rowsProcessed = 0;
}

Expand All @@ -128,8 +141,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 fetch is undefined, or current rows processed is less than required. */
private boolean hasMoreData() {
return fetchUndefined || rowsProcessed < IgniteMath.addExact(fetch, offset);
}
}
27 changes: 13 additions & 14 deletions .../src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,12 +20,11 @@
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;
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.GridBoundedPriorityQueue;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.Nullable;

/**
* Sort node.
Expand All @@ -44,7 +43,7 @@ public class SortNode<Row> extends MemoryTrackingNode<Row> implements SingleNode
private final PriorityQueue<Row> rows;

/** SQL select limit. Negative if disabled. */
private final int limit;
private final long limit;

/** Reverse-ordered rows in case of limited sort. */
private List<Row> reversed;
Expand All @@ -58,21 +57,21 @@ 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
long offset,
long fetch
) {
super(ctx, rowType);

assert fetch == null || fetch.get() >= 0;
assert offset == null || offset.get() >= 0;
assert fetch == -1 || fetch >= 0;
assert offset >= 0;

limit = fetch == null ? -1 : fetch.get() + (offset == null ? 0 : offset.get());
limit = fetch == -1 ? -1 : (fetch > Long.MAX_VALUE - offset ? -1 : fetch + offset);

if (limit < 0)
if (limit < 1 || limit > Integer.MAX_VALUE)

@alex-plekhanov alex-plekhanov Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we rich this line with limit == 0? If yes, it's more correct to use `limit <= 1' or add other logic to return empty result

rows = new PriorityQueue<>(comp);
else {
rows = new GridBoundedPriorityQueue<>(limit, comp == null ? (Comparator<Row>)Comparator.reverseOrder()
: comp.reversed());
rows = new GridBoundedPriorityQueue<>(IgniteMath.convertToIntExact(limit), comp == null ?
(Comparator<Row>)Comparator.reverseOrder() : comp.reversed());
}
}

Expand All @@ -81,7 +80,7 @@ public SortNode(
* @param comp Rows comparator.
*/
public SortNode(ExecutionContext<Row> ctx, RelDataType rowType, Comparator<Row> comp) {
this(ctx, rowType, comp, null, null);
this(ctx, rowType, comp, 0, -1);
}

/** {@inheritDoc} */
Expand Down Expand Up @@ -150,7 +149,7 @@ else if (!inLoop)

checkState();

waiting = -1;
waiting = NOT_WAITING;

flush();
}
Expand All @@ -160,7 +159,7 @@ private void flush() throws Exception {
if (isClosed())
return;

assert waiting == -1;
assert waiting == NOT_WAITING;

int processed = 0;

Expand Down
Loading
Loading