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 0000000000000..d1930d137bdf9 --- /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 1e93eecaf7c29..0ec1781f6f563 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.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.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 implements IgniteRelVisitor> { /** */ private final ExecutionContext ctx; + /** FETCH/OFFSET rounding policy used by the Calcite engine. */ + private final FetchOffsetRoundingPolicy fetchOffsetRoundingPolicy; + /** */ private final AffinityService affSrvc; @@ -176,6 +183,7 @@ public LogicalRelImplementor( this.ctx = ctx; expressionFactory = ctx.expressionFactory(); + fetchOffsetRoundingPolicy = getFetchOffsetRoundingPolicy(ctx); } /** {@inheritDoc} */ @@ -568,7 +576,7 @@ private boolean hasExchange(RelNode rel) { rowType, idxBndRel.first() ? cmp : cmp.reversed(), null, - () -> 1 + () -> BigDecimal.ONE ); sortNode.register(scanNode); @@ -630,8 +638,8 @@ private boolean hasExchange(RelNode rel) { /** {@inheritDoc} */ @Override public Node visit(IgniteLimit rel) { - Supplier offset = (rel.offset() == null) ? null : expressionFactory.execute(rel.offset()); - Supplier fetch = (rel.fetch() == null) ? null : expressionFactory.execute(rel.fetch()); + Supplier offset = fetchOffsetSupplier(rel.offset(), "OFFSET"); + Supplier fetch = fetchOffsetSupplier(rel.fetch(), "FETCH"); LimitNode node = new LimitNode<>(ctx, rel.getRowType(), offset, fetch); @@ -646,8 +654,8 @@ private boolean hasExchange(RelNode rel) { @Override public Node visit(IgniteSort rel) { RelCollation collation = rel.getCollation(); - Supplier offset = (rel.offset == null) ? null : expressionFactory.execute(rel.offset); - Supplier fetch = (rel.fetch == null) ? null : expressionFactory.execute(rel.fetch); + Supplier offset = fetchOffsetSupplier(rel.offset, "OFFSET"); + Supplier fetch = fetchOffsetSupplier(rel.fetch, "FETCH"); SortNode node = new SortNode<>(ctx, rel.getRowType(), expressionFactory.comparator(collation), offset, fetch); @@ -1050,4 +1058,34 @@ private ScanStorageNode createStorageScan( 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 fetchOffsetSupplier(@Nullable RexNode node, String kind) { + if (node == null) + return null; + + Supplier 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 9fa9d14090cd8..74fdac07e9360 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 @@ /** Offset, fetch|limit support node. */ public class LimitNode extends AbstractNode implements SingleNode, Downstream { /** 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 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 extends AbstractNode implements SingleNode public LimitNode( ExecutionContext ctx, RelDataType rowType, - Supplier offsetNode, - Supplier fetchNode + @Nullable Supplier offsetNode, + @Nullable Supplier 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 LimitNode( 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 LimitNode( /** {@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 09376b210f601..d68bee3cc293b 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 extends MemoryTrackingNode implements SingleNode /** Rows buffer. */ private final PriorityQueue rows; - /** SQL select limit. Negative if disabled. */ - private final int limit; - /** Reverse-ordered rows in case of limited sort. */ - private List reversed; + private ArrayList reversed; /** * @param ctx Execution context. @@ -58,21 +55,25 @@ public class SortNode extends MemoryTrackingNode implements SingleNode public SortNode( ExecutionContext ctx, RelDataType rowType, Comparator comp, - @Nullable Supplier offset, - @Nullable Supplier fetch + @Nullable Supplier offset, + @Nullable Supplier 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)Comparator.reverseOrder() - : comp.reversed()); + rows = new GridBoundedPriorityQueue<>(rowsToKeep.intValueExact(), + comp == null ? (Comparator)Comparator.reverseOrder() : comp.reversed()); + + reversed = new ArrayList<>(); } } @@ -106,8 +107,8 @@ public SortNode(ExecutionContext ctx, RelDataType rowType, Comparator /** {@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 @@ else if (!inLoop) /** {@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 @@ else if (!inLoop) /** {@inheritDoc} */ @Override public void end() throws Exception { assert downstream() != null; - assert waiting > 0; + assert waiting > 0 : waiting; checkState(); @@ -160,16 +161,15 @@ private void flush() throws Exception { 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 1c7f00da26537..d09946450cd77 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.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 @@ /** 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 @@ private void validateTableModify(SqlNode table) { /** {@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 @@ private void validateTableModify(SqlNode table) { * @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 @@ else if (n instanceof SqlDynamicParam) { 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 df74d56a91638..7e0df675a0a37 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 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 correctIntegerLimit(String a0); + @Resources.BaseMessage("Illegal value of {0}. The value must be positive") + Resources.ExInst 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/LogicalRelImplementorTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementorTest.java index 06a00e8622f8d..e0fe01ce86d74 100644 --- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementorTest.java +++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementorTest.java @@ -60,6 +60,7 @@ 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.RexUtils; +import org.apache.ignite.internal.util.GridBoundedPriorityQueue; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; @@ -196,7 +197,9 @@ private void checkIndexFirstOrLastRewriter(boolean first) { node = relImplementor.visit(idxScan); assertTrue(node instanceof SortNode); - assertEquals(1, (int)U.field(node, "limit")); + Object rows = U.field(node, "rows"); + assertTrue(rows instanceof GridBoundedPriorityQueue); + assertEquals(1, (int)U.field(rows, "maxCap")); assertTrue(node.sources() != null && node.sources().size() == 1); assertTrue(node.sources().get(0) instanceof ScanNode); } 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 6dd3c7b36cfdd..874cf68700875 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 void testSortLimit() { checkLimitSort(2000, 3000); } + /** Tests sort limit values greater than the integer range. */ + @Test + public void testBigDecimalSortLimit() { + ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0); + IgniteTypeFactory tf = ctx.getTypeFactory(); + RelDataType rowType = TypeUtils.createRowType(tf, int.class); + + RootNode rootNode = new RootNode<>(ctx, rowType); + SortNode sortNode = new SortNode<>(ctx, rowType, F::compareArrays, null, + () -> BigDecimal.valueOf(Integer.MAX_VALUE).add(BigDecimal.ONE)); + + List 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 ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0); + IgniteTypeFactory tf = ctx.getTypeFactory(); + RelDataType rowType = TypeUtils.createRowType(tf, int.class); + + RootNode rootNode = new RootNode<>(ctx, rowType); + SortNode sortNode = new SortNode<>(ctx, rowType, F::compareArrays); + LimitNode limitNode = new LimitNode<>(ctx, rowType, null, + () -> BigDecimal.valueOf(IN_BUFFER_SIZE * 2L)); + List 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 @@ private void checkLimitSort(int offset, int fetch) { RootNode rootNode = new RootNode<>(ctx, rowType); - SortNode sortNode = new SortNode<>(ctx, rowType, F::compareArrays, () -> offset, - fetch == 0 ? null : () -> fetch); + SortNode sortNode = new SortNode<>(ctx, rowType, F::compareArrays, + () -> BigDecimal.valueOf(offset), fetch == 0 ? null : () -> BigDecimal.valueOf(fetch)); List data = IntStream.range(0, SourceNode.IN_BUFFER_SIZE + fetch + offset).boxed() .map(i -> new Object[] {i}).collect(Collectors.toList()); @@ -109,7 +163,8 @@ private void checkLimit(int offset, int fetch) { RelDataType rowType = TypeUtils.createRowType(tf, int.class); RootNode rootNode = new RootNode<>(ctx, rowType); - LimitNode limitNode = new LimitNode<>(ctx, rowType, () -> offset, fetch == 0 ? null : () -> fetch); + LimitNode 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 428d4d273d9e2..a6bcb4c48b012 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.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 void testNestedLimitOffsetWithUnion() { ).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"); + if (sqlTxMode != SqlTransactionMode.NONE) + startTransaction(client); + + 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 void testInvalidLimitOffset() { 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 36ee4dc647a06..e0c28fde7d506 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 void testExplain() throws Exception { ).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);