This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch ignite-28896
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/ignite-28896 by this push:
new ac25617f0a8 IGNITE-28896 Wip
ac25617f0a8 is described below
commit ac25617f0a817cc8bbad20c70b8423e06598ac77
Author: Kirill Tkalenko <[email protected]>
AuthorDate: Wed Jul 15 15:35:44 2026 +0300
IGNITE-28896 Wip
---
.../enumerable/FetchOffsetRoundingPolicy.java | 36 +++++++++
.../query/calcite/exec/LogicalRelImplementor.java | 48 ++++++++++--
.../query/calcite/exec/rel/LimitNode.java | 90 +++++++++++++---------
.../query/calcite/exec/rel/SortNode.java | 44 +++++------
.../query/calcite/prepare/IgniteSqlValidator.java | 33 +++++---
.../query/calcite/util/IgniteResource.java | 5 +-
.../query/calcite/exec/rel/LimitExecutionTest.java | 61 ++++++++++++++-
.../integration/LimitOffsetIntegrationTest.java | 81 ++++++++++++++++---
.../integration/QueryMetadataIntegrationTest.java | 22 ++++++
9 files changed, 332 insertions(+), 88 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/calcite/adapter/enumerable/FetchOffsetRoundingPolicy.java
b/modules/calcite/src/main/java/org/apache/calcite/adapter/enumerable/FetchOffsetRoundingPolicy.java
new file mode 100644
index 00000000000..d1930d137bd
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/calcite/adapter/enumerable/FetchOffsetRoundingPolicy.java
@@ -0,0 +1,36 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import java.math.BigDecimal;
+
+/** Defines how enumerable FETCH and OFFSET values are rounded. */
+// TODO https://issues.apache.org/jira/browse/CALCITE-7624
+// Check and remove after update to Calcite 1.43.
+public interface FetchOffsetRoundingPolicy {
+ /** Default policy that preserves the value produced by validation or
parameter binding. */
+ FetchOffsetRoundingPolicy NONE = value -> value;
+
+ /**
+ * Rounds a FETCH or OFFSET value.
+ *
+ * @param value Value to round.
+ * @return Rounded value.
+ */
+ BigDecimal round(BigDecimal value);
+}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
index 1e93eecaf7c..0ec1781f6f5 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.processors.query.calcite.exec;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
@@ -28,6 +30,7 @@ import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import com.google.common.collect.ImmutableList;
+import org.apache.calcite.adapter.enumerable.FetchOffsetRoundingPolicy;
import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
@@ -126,6 +129,7 @@ import
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribut
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.RexUtils;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.Nullable;
@@ -144,6 +148,9 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
/** */
private final ExecutionContext<Row> ctx;
+ /** FETCH/OFFSET rounding policy used by the Calcite engine. */
+ private final FetchOffsetRoundingPolicy fetchOffsetRoundingPolicy;
+
/** */
private final AffinityService affSrvc;
@@ -176,6 +183,7 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
this.ctx = ctx;
expressionFactory = ctx.expressionFactory();
+ fetchOffsetRoundingPolicy = getFetchOffsetRoundingPolicy(ctx);
}
/** {@inheritDoc} */
@@ -568,7 +576,7 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
rowType,
idxBndRel.first() ? cmp : cmp.reversed(),
null,
- () -> 1
+ () -> BigDecimal.ONE
);
sortNode.register(scanNode);
@@ -630,8 +638,8 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
/** {@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<BigDecimal> offset = fetchOffsetSupplier(rel.offset(),
"OFFSET");
+ Supplier<BigDecimal> fetch = fetchOffsetSupplier(rel.fetch(), "FETCH");
LimitNode<Row> node = new LimitNode<>(ctx, rel.getRowType(), offset,
fetch);
@@ -646,8 +654,8 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
@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<BigDecimal> offset = fetchOffsetSupplier(rel.offset,
"OFFSET");
+ Supplier<BigDecimal> fetch = fetchOffsetSupplier(rel.fetch, "FETCH");
SortNode<Row> node = new SortNode<>(ctx, rel.getRowType(),
expressionFactory.comparator(collation), offset,
fetch);
@@ -1050,4 +1058,34 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
otherColMapping
);
}
+
+ /** */
+ private static FetchOffsetRoundingPolicy
getFetchOffsetRoundingPolicy(ExecutionContext<?> ctx) {
+ FetchOffsetRoundingPolicy roundingPlc =
ctx.unwrap(FetchOffsetRoundingPolicy.class);
+
+ return roundingPlc == null ? FetchOffsetRoundingPolicy.NONE :
roundingPlc;
+ }
+
+ /** Converts a FETCH/OFFSET expression to the row count expected by Ignite
execution nodes. */
+ private @Nullable Supplier<BigDecimal> fetchOffsetSupplier(@Nullable
RexNode node, String kind) {
+ if (node == null)
+ return null;
+
+ Supplier<Number> val = expressionFactory.execute(node);
+
+ return () -> fetchOffsetValue(val.get(), kind);
+ }
+
+ /** Converts a FETCH/OFFSET runtime value to a row count. */
+ private BigDecimal fetchOffsetValue(Object val, String kind) {
+ if (!(val instanceof Number))
+ throw new IllegalArgumentException(kind + " must be a number");
+
+ BigDecimal decimal = IgniteMath.convertToBigDecimal((Number)val);
+
+ if (decimal.signum() < 0)
+ throw new IllegalArgumentException(kind + " must not be negative");
+
+ return fetchOffsetRoundingPolicy.round(decimal).setScale(0,
RoundingMode.CEILING);
+ }
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java
index 9fa9d14090c..74fdac07e93 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+import java.math.BigDecimal;
import java.util.function.Supplier;
import org.apache.calcite.rel.type.RelDataType;
import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
@@ -26,19 +27,16 @@ 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;
+ private final BigDecimal offset;
- /** Fetch if its present, otherwise 0. */
- private final int fetch;
+ /** Fetch if its present, otherwise all rows. */
+ private final @Nullable BigDecimal fetch;
/** Already processed (pushed to upstream) rows count. */
- private int rowsProcessed;
+ private BigDecimal rowsProcessed = BigDecimal.ZERO;
- /** Fetch can be unset, in this case we need all rows. */
- private @Nullable Supplier<Integer> fetchNode;
-
- /** Waiting results counter. */
- private int waiting;
+ /** Number of upstream rows that remain to be processed for the current
downstream request. */
+ private BigDecimal waiting = BigDecimal.ZERO;
/**
* Constructor.
@@ -49,20 +47,19 @@ 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<BigDecimal> offsetNode,
+ @Nullable Supplier<BigDecimal> fetchNode
) {
super(ctx, rowType);
- offset = offsetNode == null ? 0 : offsetNode.get();
- fetch = fetchNode == null ? 0 : fetchNode.get();
- this.fetchNode = fetchNode;
+ offset = offsetNode == null ? BigDecimal.ZERO : offsetNode.get();
+ fetch = fetchNode == null ? null : fetchNode.get();
}
/** {@inheritDoc} */
@Override public void request(int rowsCnt) throws Exception {
assert !F.isEmpty(sources()) && sources().size() == 1;
- assert rowsCnt > 0;
+ assert rowsCnt > 0 : rowsCnt;
if (fetchNone()) {
end();
@@ -70,54 +67,56 @@ public class LimitNode<Row> extends AbstractNode<Row>
implements SingleNode<Row>
return;
}
- if (offset > 0 && rowsProcessed == 0)
- rowsCnt = offset + rowsCnt;
-
- waiting = rowsCnt;
+ waiting = BigDecimal.valueOf(rowsCnt);
- if (fetch > 0)
- rowsCnt = Math.min(rowsCnt, (fetch + offset) - rowsProcessed);
+ if (rowsProcessed.compareTo(offset) < 0)
+ waiting = waiting.add(offset.subtract(rowsProcessed));
- checkState();
-
- source().request(rowsCnt);
+ requestNextBatch();
}
/** {@inheritDoc} */
@Override public void push(Row row) throws Exception {
- if (waiting == -1)
+ if (waiting.signum() < 0)
return;
- ++rowsProcessed;
+ rowsProcessed = rowsProcessed.add(BigDecimal.ONE);
- --waiting;
+ waiting = waiting.subtract(BigDecimal.ONE);
checkState();
- if (rowsProcessed > offset) {
- if (fetchNode == null || (fetchNode != null && rowsProcessed <=
fetch + offset))
- downstream().push(row);
+ boolean endAfterRow = fetchNone() && waiting.signum() > 0;
+ boolean reqNextAfterRow = !endAfterRow && waiting.signum() > 0
+ &&
rowsProcessed.remainder(BigDecimal.valueOf(IN_BUFFER_SIZE)).signum() == 0;
+
+ if (rowsProcessed.compareTo(offset) > 0
+ && (fetch == null || rowsProcessed.compareTo(offset.add(fetch)) <=
0)) {
+ downstream().push(row);
}
- if (fetch > 0 && rowsProcessed == fetch + offset && waiting > 0)
+ if (endAfterRow)
end();
+ else if (reqNextAfterRow)
+ requestNextBatch();
}
/** {@inheritDoc} */
@Override public void end() throws Exception {
- if (waiting == -1)
+ if (waiting.signum() < 0)
return;
assert downstream() != null;
- waiting = -1;
+ waiting = BigDecimal.ONE.negate();
downstream().end();
}
/** {@inheritDoc} */
@Override protected void rewindInternal() {
- rowsProcessed = 0;
+ rowsProcessed = BigDecimal.ZERO;
+ waiting = BigDecimal.ZERO;
}
/** {@inheritDoc} */
@@ -130,6 +129,27 @@ public class LimitNode<Row> extends AbstractNode<Row>
implements SingleNode<Row>
/** {@code True} if requested 0 results, or all already processed. */
private boolean fetchNone() {
- return (fetchNode != null && fetch == 0) || (fetch > 0 &&
rowsProcessed == fetch + offset);
+ return fetch != null && (fetch.signum() == 0 ||
rowsProcessed.compareTo(offset.add(fetch)) >= 0);
+ }
+
+ /** Requests the next upstream batch, taking large decimal offsets into
account. */
+ private void requestNextBatch() throws Exception {
+ BigDecimal bufSize = BigDecimal.valueOf(IN_BUFFER_SIZE);
+ BigDecimal rowsAvailable = waiting;
+
+ if (fetch != null)
+ rowsAvailable =
rowsAvailable.min(offset.add(fetch).subtract(rowsProcessed));
+
+ BigDecimal rowsToReq =
bufSize.subtract(rowsProcessed.remainder(bufSize)).min(rowsAvailable);
+
+ if (rowsToReq.signum() == 0) {
+ end();
+
+ return;
+ }
+
+ checkState();
+
+ source().request(rowsToReq.intValueExact());
}
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
index 09376b210f6..d68bee3cc29 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
@@ -16,9 +16,9 @@
*/
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Comparator;
-import java.util.List;
import java.util.PriorityQueue;
import java.util.function.Supplier;
import org.apache.calcite.rel.type.RelDataType;
@@ -43,11 +43,8 @@ public class SortNode<Row> extends MemoryTrackingNode<Row>
implements SingleNode
/** Rows buffer. */
private final PriorityQueue<Row> rows;
- /** SQL select limit. Negative if disabled. */
- private final int limit;
-
/** Reverse-ordered rows in case of limited sort. */
- private List<Row> reversed;
+ private ArrayList<Row> reversed;
/**
* @param ctx Execution context.
@@ -58,21 +55,25 @@ 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<BigDecimal> offset,
+ @Nullable Supplier<BigDecimal> fetch
) {
super(ctx, rowType);
- assert fetch == null || fetch.get() >= 0;
- assert offset == null || offset.get() >= 0;
+ BigDecimal offsetVal = offset == null ? BigDecimal.ZERO : offset.get();
+ BigDecimal fetchVal = fetch == null ? null : fetch.get();
- limit = fetch == null ? -1 : fetch.get() + (offset == null ? 0 :
offset.get());
+ BigDecimal rowsToKeep = fetchVal == null ? null :
fetchVal.add(offsetVal);
- if (limit < 0)
+ if (rowsToKeep == null || rowsToKeep.signum() == 0
+ || rowsToKeep.compareTo(BigDecimal.valueOf(Integer.MAX_VALUE)) >
0) {
rows = new PriorityQueue<>(comp);
+ }
else {
- rows = new GridBoundedPriorityQueue<>(limit, comp == null ?
(Comparator<Row>)Comparator.reverseOrder()
- : comp.reversed());
+ rows = new GridBoundedPriorityQueue<>(rowsToKeep.intValueExact(),
+ comp == null ? (Comparator<Row>)Comparator.reverseOrder() :
comp.reversed());
+
+ reversed = new ArrayList<>();
}
}
@@ -106,8 +107,8 @@ public class SortNode<Row> extends MemoryTrackingNode<Row>
implements SingleNode
/** {@inheritDoc} */
@Override public void request(int rowsCnt) throws Exception {
assert !F.isEmpty(sources()) && sources().size() == 1;
- assert rowsCnt > 0 && requested == 0;
- assert waiting <= 0;
+ assert rowsCnt > 0 && requested == 0 : "rowsCnt=" + rowsCnt + ",
requested=" + requested;
+ assert waiting <= 0 : waiting;
checkState();
@@ -122,8 +123,8 @@ public class SortNode<Row> extends MemoryTrackingNode<Row>
implements SingleNode
/** {@inheritDoc} */
@Override public void push(Row row) throws Exception {
assert downstream() != null;
- assert waiting > 0;
- assert reversed == null || reversed.isEmpty();
+ assert waiting > 0 : waiting;
+ assert reversed == null || reversed.isEmpty() : reversed.size();
checkState();
@@ -146,7 +147,7 @@ public class SortNode<Row> extends MemoryTrackingNode<Row>
implements SingleNode
/** {@inheritDoc} */
@Override public void end() throws Exception {
assert downstream() != null;
- assert waiting > 0;
+ assert waiting > 0 : waiting;
checkState();
@@ -160,16 +161,15 @@ public class SortNode<Row> extends
MemoryTrackingNode<Row> implements SingleNode
if (isClosed())
return;
- assert waiting == -1;
+ assert waiting == -1 : waiting;
int processed = 0;
inLoop = true;
try {
// Prepare final order (reversed).
- if (limit > 0 && !rows.isEmpty()) {
- if (reversed == null)
- reversed = new ArrayList<>(rows.size());
+ if (reversed != null && !rows.isEmpty()) {
+ reversed.ensureCapacity(rows.size());
while (!rows.isEmpty()) {
reversed.add(rows.poll());
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
index 1c7f00da265..d09946450cd 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
@@ -74,6 +74,7 @@ import
org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
import
org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlDecimalLiteral;
import
org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
import org.apache.ignite.internal.processors.query.calcite.type.OtherType;
+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.util.typedef.F;
import org.immutables.value.Value;
@@ -84,9 +85,6 @@ import static org.apache.calcite.util.Static.RESOURCE;
/** Validator. */
@Value.Enclosing
public class IgniteSqlValidator extends SqlValidatorImpl {
- /** Decimal of Integer.MAX_VALUE for fetch/offset bounding. */
- private static final BigDecimal DEC_INT_MAX =
BigDecimal.valueOf(Integer.MAX_VALUE);
-
/** **/
private static final int MAX_LENGTH_OF_ALIASES = 256;
@@ -241,10 +239,21 @@ public class IgniteSqlValidator extends SqlValidatorImpl {
/** {@inheritDoc} */
@Override protected void validateSelect(SqlSelect select, RelDataType
targetRowType) {
- checkIntegerLimit(select.getFetch(), "fetch / limit");
- checkIntegerLimit(select.getOffset(), "offset");
+ checkLimit(select.getFetch(), "fetch / limit");
+ checkLimit(select.getOffset(), "offset");
super.validateSelect(select, targetRowType);
+
+ setFetchOffsetType(select.getFetch());
+ setFetchOffsetType(select.getOffset());
+ }
+
+ /** */
+ // TODO https://issues.apache.org/jira/browse/CALCITE-7624
+ // Check SqlValidatorImpl#handleOffsetFetch and remove after update to
Calcite 1.43.
+ private void setFetchOffsetType(@Nullable SqlNode node) {
+ if (node instanceof SqlDynamicParam)
+ setValidatedNodeType(node,
typeFactory.createSqlType(SqlTypeName.DECIMAL));
}
/** {@inheritDoc} */
@@ -265,12 +274,12 @@ public class IgniteSqlValidator extends SqlValidatorImpl {
* @param n Node to check limit.
* @param nodeName Node name.
*/
- private void checkIntegerLimit(SqlNode n, String nodeName) {
+ private void checkLimit(SqlNode n, String nodeName) {
if (n instanceof SqlLiteral) {
BigDecimal offFetchLimit = ((SqlLiteral)n).bigDecimalValue();
- if (offFetchLimit.compareTo(DEC_INT_MAX) > 0 ||
offFetchLimit.compareTo(BigDecimal.ZERO) < 0)
- throw newValidationError(n,
IgniteResource.INSTANCE.correctIntegerLimit(nodeName));
+ if (offFetchLimit.compareTo(BigDecimal.ZERO) < 0)
+ throw newValidationError(n,
IgniteResource.INSTANCE.illegalLimit(nodeName));
}
else if (n instanceof SqlDynamicParam) {
// will fail in params check.
@@ -281,9 +290,11 @@ public class IgniteSqlValidator extends SqlValidatorImpl {
if (idx < parameters.length) {
Object param = parameters[idx];
- if (parameters[idx] instanceof Integer) {
- if ((Integer)param < 0)
- throw newValidationError(n,
IgniteResource.INSTANCE.correctIntegerLimit(nodeName));
+ if (param instanceof Number) {
+ BigDecimal val =
IgniteMath.convertToBigDecimal((Number)param);
+
+ if (val.compareTo(BigDecimal.ZERO) < 0)
+ throw newValidationError(n,
IgniteResource.INSTANCE.illegalLimit(nodeName));
}
}
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java
index df74d56a916..7e0df675a0a 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java
@@ -40,9 +40,8 @@ public interface IgniteResource {
Resources.ExInst<SqlValidatorException>
unsupportedAggregationFunction(String a0);
/** */
- @Resources.BaseMessage("Illegal value of {0}. The value must be positive
and less than Integer.MAX_VALUE " +
- "(" + Integer.MAX_VALUE + ")." )
- Resources.ExInst<SqlValidatorException> correctIntegerLimit(String a0);
+ @Resources.BaseMessage("Illegal value of {0}. The value must be positive")
+ Resources.ExInst<SqlValidatorException> illegalLimit(String a0);
/** */
@Resources.BaseMessage("Option ''{0}'' has already been defined")
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitExecutionTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitExecutionTest.java
index 6dd3c7b36cf..874cf687008 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitExecutionTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitExecutionTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+import java.math.BigDecimal;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
@@ -64,6 +65,59 @@ public class LimitExecutionTest extends
AbstractExecutionTest {
checkLimitSort(2000, 3000);
}
+ /** Tests sort limit values greater than the integer range. */
+ @Test
+ public void testBigDecimalSortLimit() {
+ ExecutionContext<Object[]> ctx = executionContext(F.first(nodes()),
UUID.randomUUID(), 0);
+ IgniteTypeFactory tf = ctx.getTypeFactory();
+ RelDataType rowType = TypeUtils.createRowType(tf, int.class);
+
+ RootNode<Object[]> rootNode = new RootNode<>(ctx, rowType);
+ SortNode<Object[]> sortNode = new SortNode<>(ctx, rowType,
F::compareArrays, null,
+ () -> BigDecimal.valueOf(Integer.MAX_VALUE).add(BigDecimal.ONE));
+
+ List<Object[]> data = IntStream.range(0, 10).boxed().map(i -> new
Object[] {i}).collect(Collectors.toList());
+
+ Collections.reverse(data);
+
+ rootNode.register(sortNode);
+ sortNode.register(new ScanNode<>(ctx, rowType, data));
+
+ for (int i = 0; i < data.size(); i++) {
+ assertTrue(rootNode.hasNext());
+ assertEquals(i, rootNode.next()[0]);
+ }
+
+ assertFalse(rootNode.hasNext());
+ }
+
+ /** Tests a reentrant downstream request at an input buffer boundary. */
+ @Test
+ public void testReentrantRequestToLimit() {
+ ExecutionContext<Object[]> ctx = executionContext(F.first(nodes()),
UUID.randomUUID(), 0);
+ IgniteTypeFactory tf = ctx.getTypeFactory();
+ RelDataType rowType = TypeUtils.createRowType(tf, int.class);
+
+ RootNode<Object[]> rootNode = new RootNode<>(ctx, rowType);
+ SortNode<Object[]> sortNode = new SortNode<>(ctx, rowType,
F::compareArrays);
+ LimitNode<Object[]> limitNode = new LimitNode<>(ctx, rowType, null,
+ () -> BigDecimal.valueOf(IN_BUFFER_SIZE * 2L));
+ List<Object[]> data = IntStream.range(0, IN_BUFFER_SIZE * 2).boxed()
+ .map(i -> new Object[] {i})
+ .collect(Collectors.toList());
+
+ rootNode.register(sortNode);
+ sortNode.register(limitNode);
+ limitNode.register(new ScanNode<>(ctx, rowType, data));
+
+ for (int i = 0; i < data.size(); i++) {
+ assertTrue(rootNode.hasNext());
+ assertEquals(i, rootNode.next()[0]);
+ }
+
+ assertFalse(rootNode.hasNext());
+ }
+
/**
* @param offset Rows offset.
* @param fetch Fetch rows count (zero means unlimited).
@@ -78,8 +132,8 @@ public class LimitExecutionTest extends
AbstractExecutionTest {
RootNode<Object[]> rootNode = new RootNode<>(ctx, rowType);
- SortNode<Object[]> sortNode = new SortNode<>(ctx, rowType,
F::compareArrays, () -> offset,
- fetch == 0 ? null : () -> fetch);
+ SortNode<Object[]> sortNode = new SortNode<>(ctx, rowType,
F::compareArrays,
+ () -> BigDecimal.valueOf(offset), fetch == 0 ? null : () ->
BigDecimal.valueOf(fetch));
List<Object[]> data = IntStream.range(0, SourceNode.IN_BUFFER_SIZE +
fetch + offset).boxed()
.map(i -> new Object[] {i}).collect(Collectors.toList());
@@ -109,7 +163,8 @@ public class LimitExecutionTest extends
AbstractExecutionTest {
RelDataType rowType = TypeUtils.createRowType(tf, int.class);
RootNode<Object[]> rootNode = new RootNode<>(ctx, rowType);
- LimitNode<Object[]> limitNode = new LimitNode<>(ctx, rowType, () ->
offset, fetch == 0 ? null : () -> fetch);
+ LimitNode<Object[]> limitNode = new LimitNode<>(ctx, rowType, () ->
BigDecimal.valueOf(offset),
+ fetch == 0 ? null : () -> BigDecimal.valueOf(fetch));
SourceNode srcNode = new SourceNode(ctx, rowType);
rootNode.register(limitNode);
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java
index 428d4d273d9..af841b9c698 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java
@@ -18,9 +18,14 @@
package org.apache.ignite.internal.processors.query.calcite.integration;
import java.math.BigDecimal;
+import java.math.RoundingMode;
import java.util.Arrays;
import java.util.List;
+import org.apache.calcite.adapter.enumerable.FetchOffsetRoundingPolicy;
+import org.apache.calcite.plan.Contexts;
import org.apache.calcite.sql.validate.SqlValidatorException;
+import org.apache.calcite.tools.FrameworkConfig;
+import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
@@ -33,6 +38,7 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.junit.Test;
import static java.util.Collections.singletonList;
+import static
org.apache.ignite.internal.processors.query.calcite.CalciteQueryProcessor.FRAMEWORK_CONFIG;
/**
* Limit / offset tests.
@@ -54,7 +60,8 @@ public class LimitOffsetIntegrationTest extends
AbstractBasicIntegrationTransact
/** {@inheritDoc} */
@Override protected void afterTest() throws Exception {
- // Override method to keep caches after tests.
+ // Keep caches between tests, but do not leak an active transaction
into the next test.
+ clearTransaction();
}
/** {@inheritDoc} */
@@ -94,6 +101,8 @@ public class LimitOffsetIntegrationTest extends
AbstractBasicIntegrationTransact
/** */
@Test
public void testNestedLimitOffsetWithUnion() {
+ cacheRepl.clear();
+
sql("INSERT into TEST_REPL VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4,
'd')");
assertQuery("(SELECT id FROM TEST_REPL WHERE id = 2) UNION ALL " +
@@ -101,20 +110,71 @@ public class LimitOffsetIntegrationTest extends
AbstractBasicIntegrationTransact
).returns(2).returns(4).check();
}
- /** Tests correctness of fetch / offset params. */
+ /** */
@Test
- public void testInvalidLimitOffset() {
+ public void testFractionalLimitOffset() {
+ cacheRepl.clear();
+
+ sql("INSERT into TEST_REPL VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4,
'd')");
+
+ assertQuery("SELECT id FROM TEST_REPL ORDER BY id LIMIT 1.2")
+ .returns(1)
+ .returns(2)
+ .check();
+
+ assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET 1.1 ROWS
FETCH FIRST 1.1 ROWS ONLY")
+ .returns(3)
+ .returns(4)
+ .check();
+
+ assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET ? ROWS FETCH
FIRST ? ROWS ONLY")
+ .withParams(new BigDecimal("1.1"), new BigDecimal("1.2"))
+ .returns(3)
+ .returns(4)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testFetchOffsetRoundingPolicy() {
+ FetchOffsetRoundingPolicy floorPlc = value -> value.setScale(0,
RoundingMode.FLOOR);
+ FrameworkConfig floorPlcCfg =
Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
+ .context(Contexts.of(floorPlc))
+ .build();
+
+ if (sqlTxMode != SqlTransactionMode.NONE)
+ startTransaction(client);
+
+ assertQuery("SELECT * FROM (VALUES (1), (2), (3), (4)) LIMIT 1.1")
+ .withFrameworkConfig(floorPlcCfg)
+ .returns(1)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testBigDecimalLimitOffset() {
String bigInt = BigDecimal.valueOf(10000000000L).toString();
- assertThrows("SELECT * FROM TEST_REPL OFFSET " + bigInt + " ROWS",
- SqlValidatorException.class, "Illegal value of offset");
+ // QueryChecker expects an active transaction for transactional test
modes.
+ sql("SELECT 1");
+
+ assertQuery("SELECT * FROM (VALUES (1)) OFFSET " + bigInt + " ROWS")
+ .resultSize(0)
+ .check();
- assertThrows("SELECT * FROM TEST_REPL FETCH FIRST " + bigInt + " ROWS
ONLY",
- SqlValidatorException.class, "Illegal value of fetch / limit");
+ assertQuery("SELECT * FROM (VALUES (1)) FETCH FIRST " + bigInt + "
ROWS ONLY")
+ .returns(1)
+ .check();
- assertThrows("SELECT * FROM TEST_REPL LIMIT " + bigInt,
- SqlValidatorException.class, "Illegal value of fetch / limit");
+ assertQuery("SELECT * FROM (VALUES (1)) LIMIT " + bigInt)
+ .returns(1)
+ .check();
+ }
+ /** Tests correctness of fetch / offset params. */
+ @Test
+ public void testInvalidLimitOffset() {
assertThrows("SELECT * FROM TEST_REPL OFFSET -1 ROWS FETCH FIRST -1
ROWS ONLY",
IgniteSQLException.class, null);
@@ -133,6 +193,9 @@ public class LimitOffsetIntegrationTest extends
AbstractBasicIntegrationTransact
assertThrows("SELECT * FROM TEST_REPL FETCH FIRST ? ROWS ONLY",
SqlValidatorException.class, "Illegal value of fetch / limit", -1);
+
+ assertThrows("SELECT * FROM TEST_REPL OFFSET ? ROWS",
+ SqlValidatorException.class, "Illegal value of offset", new
BigDecimal("-1.5"));
}
/**
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/QueryMetadataIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/QueryMetadataIntegrationTest.java
index 36ee4dc647a..e0c28fde7d5 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/QueryMetadataIntegrationTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/QueryMetadataIntegrationTest.java
@@ -177,6 +177,28 @@ public class QueryMetadataIntegrationTest extends
AbstractBasicIntegrationTest {
).check();
}
+ /** */
+ @Test
+ public void testFetchOffsetParameters() throws Exception {
+ executeSql("CREATE TABLE tbl (id BIGINT, PRIMARY KEY(id))");
+
+ checker("SELECT * FROM tbl OFFSET ? ROWS FETCH FIRST ? ROWS ONLY")
+ .addMeta(
+ builder -> builder.add("PUBLIC", "TBL", Long.class, "ID", 19,
0, true),
+ builder -> builder
+ .add(null, null, BigDecimal.class, "?0", 32767, 0, false)
+ .add(null, null, BigDecimal.class, "?1", 32767, 0, false)
+ )
+ .check();
+
+ checker("SELECT * FROM tbl LIMIT ?")
+ .addMeta(
+ builder -> builder.add("PUBLIC", "TBL", Long.class, "ID", 19,
0, true),
+ builder -> builder.add(null, null, BigDecimal.class, "?0",
32767, 0, false)
+ )
+ .check();
+ }
+
/** */
private MetadataChecker checker(String sql) {
return new MetadataChecker(queryEngine(client), sql);
