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..adb9943cbc3b4 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 @@ -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 visit(IgniteLimit rel) { - Supplier offset = (rel.offset() == null) ? null : expressionFactory.execute(rel.offset()); - Supplier 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 node = new LimitNode<>(ctx, rel.getRowType(), offset, fetch); @@ -646,8 +650,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); + long offset = rel.offset == null ? 0 : validateAndGetFetchOffsetParams(rel.offset, "offset"); + long fetch = rel.fetch == null ? -1 : validateAndGetFetchOffsetParams(rel.fetch, "fetch"); SortNode node = new SortNode<>(ctx, rel.getRowType(), expressionFactory.comparator(collation), offset, fetch); @@ -1050,4 +1054,33 @@ private ScanStorageNode createStorageScan( otherColMapping ); } + + /** */ + private long validateAndGetFetchOffsetParams(RexNode node, String op) { + Supplier 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; + } } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java index ef372b62f2ad2..d6c89a7b74777 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java +++ b/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 implements Node { + /** 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); 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..4872424823345 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,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 extends AbstractNode implements SingleNode, Downstream { - /** 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; /** Already processed (pushed to upstream) rows count. */ private int rowsProcessed; - /** Fetch can be unset, in this case we need all rows. */ - private @Nullable Supplier fetchNode; - /** Waiting results counter. */ private int waiting; + /** Upper requested rows. */ + private int requested; + /** * Constructor. * @@ -49,14 +51,14 @@ public class LimitNode extends AbstractNode implements SingleNode public LimitNode( ExecutionContext ctx, RelDataType rowType, - Supplier offsetNode, - Supplier 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} */ @@ -64,19 +66,22 @@ public LimitNode( 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(); @@ -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. + --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; } @@ -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); } } 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..f10afb606d7c2 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 @@ -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. @@ -44,7 +43,7 @@ public class SortNode extends MemoryTrackingNode implements SingleNode private final PriorityQueue 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 reversed; @@ -58,21 +57,21 @@ public class SortNode extends MemoryTrackingNode implements SingleNode public SortNode( ExecutionContext ctx, RelDataType rowType, Comparator comp, - @Nullable Supplier offset, - @Nullable Supplier 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) rows = new PriorityQueue<>(comp); else { - rows = new GridBoundedPriorityQueue<>(limit, comp == null ? (Comparator)Comparator.reverseOrder() - : comp.reversed()); + rows = new GridBoundedPriorityQueue<>(IgniteMath.convertToIntExact(limit), comp == null ? + (Comparator)Comparator.reverseOrder() : comp.reversed()); } } @@ -81,7 +80,7 @@ public SortNode( * @param comp Rows comparator. */ public SortNode(ExecutionContext ctx, RelDataType rowType, Comparator comp) { - this(ctx, rowType, comp, null, null); + this(ctx, rowType, comp, 0, -1); } /** {@inheritDoc} */ @@ -150,7 +149,7 @@ else if (!inLoop) checkState(); - waiting = -1; + waiting = NOT_WAITING; flush(); } @@ -160,7 +159,7 @@ private void flush() throws Exception { if (isClosed()) return; - assert waiting == -1; + assert waiting == NOT_WAITING; int processed = 0; 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..7b3bd13a8d6dc 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 @@ -29,6 +29,7 @@ import org.apache.calcite.prepare.Prepare; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.runtime.Resources; import org.apache.calcite.sql.JoinConditionType; import org.apache.calcite.sql.JoinType; import org.apache.calcite.sql.SqlAggFunction; @@ -61,6 +62,7 @@ import org.apache.calcite.sql.validate.SelectScope; import org.apache.calcite.sql.validate.SqlQualified; import org.apache.calcite.sql.validate.SqlValidator; +import org.apache.calcite.sql.validate.SqlValidatorException; import org.apache.calcite.sql.validate.SqlValidatorImpl; import org.apache.calcite.sql.validate.SqlValidatorNamespace; import org.apache.calcite.sql.validate.SqlValidatorScope; @@ -74,6 +76,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 +87,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,54 +241,81 @@ private void validateTableModify(SqlNode table) { /** {@inheritDoc} */ @Override protected void validateSelect(SqlSelect select, RelDataType targetRowType) { - checkIntegerLimit(select.getFetch(), "fetch / limit"); - checkIntegerLimit(select.getOffset(), "offset"); - super.validateSelect(select, targetRowType); - } - /** {@inheritDoc} */ - @Override protected void validateNamespace(SqlValidatorNamespace namespace, RelDataType targetRowType) { - SqlValidatorTable table = namespace.getTable(); - - if (table != null) { - IgniteCacheTable igniteTable = table.unwrap(IgniteCacheTable.class); - - if (igniteTable != null) - igniteTable.ensureCacheStarted(); - } - - super.validateNamespace(namespace, targetRowType); + invalidateFetchOffset(select.getFetch(), "fetch / limit"); + invalidateFetchOffset(select.getOffset(), "offset"); } /** - * @param n Node to check limit. + * Invalidate fetch/offset params restrictions. + * + * @param n Node to check limit. * @param nodeName Node name. */ - private void checkIntegerLimit(SqlNode n, String nodeName) { + private void invalidateFetchOffset(@Nullable SqlNode n, String nodeName) { + if (n == null) { + return; + } + if (n instanceof SqlLiteral) { - BigDecimal offFetchLimit = ((SqlLiteral)n).bigDecimalValue(); + BigDecimal offsetFetchLimit = ((SqlLiteral)n).bigDecimalValue(); - if (offFetchLimit.compareTo(DEC_INT_MAX) > 0 || offFetchLimit.compareTo(BigDecimal.ZERO) < 0) - throw newValidationError(n, IgniteResource.INSTANCE.correctIntegerLimit(nodeName)); + checkLimitOffset(offsetFetchLimit, n, nodeName); } - else if (n instanceof SqlDynamicParam) { - // will fail in params check. + else if (n instanceof SqlDynamicParam dynamicParam) { if (F.isEmpty(parameters)) return; - int idx = ((SqlDynamicParam)n).getIndex(); + if (dynamicParam.getIndex() < parameters.length) { + Object param = parameters[dynamicParam.getIndex()]; + + if (!(param instanceof Number)) { + SqlTypeName expectType = SqlTypeName.BIGINT; + Resources.ExInst err; + + if (param == null) + err = IgniteResource.INSTANCE.incorrectDynamicParameterType(expectType.toString(), "null"); + else { + SqlTypeName paramType = typeFactory().createType(param.getClass()).getSqlTypeName(); + err = IgniteResource.INSTANCE.incorrectDynamicParameterType(expectType.toString(), paramType.getName()); + } - if (idx < parameters.length) { - Object param = parameters[idx]; - if (parameters[idx] instanceof Integer) { - if ((Integer)param < 0) - throw newValidationError(n, IgniteResource.INSTANCE.correctIntegerLimit(nodeName)); + throw newValidationError(n, err); } + else + checkLimitOffset((Number)param, n, nodeName); } } } + /** */ + private void checkLimitOffset(Number offsetFetchLimit, @Nullable SqlNode n, String nodeName) { + try { + long res = IgniteMath.convertToLongExact(offsetFetchLimit); + + if (res < 0) + throw newValidationError(n, IgniteResource.INSTANCE.illegalFetchLimit(nodeName)); + } + catch (ArithmeticException e) { + throw newValidationError(n, IgniteResource.INSTANCE.illegalFetchLimit(nodeName)); + } + } + + /** {@inheritDoc} */ + @Override protected void validateNamespace(SqlValidatorNamespace namespace, RelDataType targetRowType) { + SqlValidatorTable table = namespace.getTable(); + + if (table != null) { + IgniteCacheTable igniteTable = table.unwrap(IgniteCacheTable.class); + + if (igniteTable != null) + igniteTable.ensureCacheStarted(); + } + + super.validateNamespace(namespace, targetRowType); + } + /** {@inheritDoc} */ @Override public void validateCall(SqlCall call, SqlValidatorScope scope) { if (call.getKind() == SqlKind.AS) { 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..8dff42046dcd7 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 @@ -39,11 +39,6 @@ public interface IgniteResource { @Resources.BaseMessage("Illegal aggregate function. {0} is unsupported at the moment.") 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("Option ''{0}'' has already been defined") Resources.ExInst optionAlreadyDefined(String optName); @@ -84,4 +79,12 @@ public interface IgniteResource { /** */ @Resources.BaseMessage("Operator ''CAST'' supports only the parameters: value and target type.") Resources.ExInst invalidCastParameters(); + + /** */ + @Resources.BaseMessage("Illegal value of {0}. The value must be non-negative and less than or equal to " + Long.MAX_VALUE) + Resources.ExInst illegalFetchLimit(String a0); + + /** */ + @Resources.BaseMessage("Incorrect type of a dynamic parameter. Expected <{0}> but got <{1}>") + Resources.ExInst incorrectDynamicParameterType(String expected, String actual); } 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..3e87461ea0bdf 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 @@ -196,7 +196,7 @@ private void checkIndexFirstOrLastRewriter(boolean first) { node = relImplementor.visit(idxScan); assertTrue(node instanceof SortNode); - assertEquals(1, (int)U.field(node, "limit")); + assertEquals(1L, (long)U.field(node, "limit")); 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..ef22b0947275e 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 @@ -20,7 +20,6 @@ import java.util.Collections; import java.util.List; import java.util.UUID; -import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.IntStream; import org.apache.calcite.rel.type.RelDataType; @@ -78,10 +77,10 @@ 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, offset, + fetch == 0 ? -1 : fetch); - List data = IntStream.range(0, SourceNode.IN_BUFFER_SIZE + fetch + offset).boxed() + List data = IntStream.range(0, IN_BUFFER_SIZE + fetch + offset).boxed() .map(i -> new Object[] {i}).collect(Collectors.toList()); Collections.shuffle(data); @@ -109,56 +108,21 @@ 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); - SourceNode srcNode = new SourceNode(ctx, rowType); + LimitNode limitNode = new LimitNode<>(ctx, rowType, offset, fetch == 0 ? -1 : fetch); + + List data = IntStream.range(0, IN_BUFFER_SIZE + fetch + offset).boxed() + .map(i -> new Object[] {i}).collect(Collectors.toList()); + + ScanNode srcNode = new ScanNode<>(ctx, rowType, data); rootNode.register(limitNode); limitNode.register(srcNode); - if (fetch > 0) { - for (int i = offset; i < offset + fetch; i++) { - assertTrue(rootNode.hasNext()); - assertEquals(i, rootNode.next()[0]); - } - - assertFalse(rootNode.hasNext()); - assertEquals(srcNode.requested.get(), offset + fetch); - } - else { + for (int i = offset; i < offset + fetch; i++) { assertTrue(rootNode.hasNext()); - assertEquals(offset, rootNode.next()[0]); - assertTrue(srcNode.requested.get() > offset); - } - } - - /** */ - private static class SourceNode extends AbstractNode { - /** */ - AtomicInteger requested = new AtomicInteger(); - - /** */ - public SourceNode(ExecutionContext ctx, RelDataType rowType) { - super(ctx, rowType); - } - - /** {@inheritDoc} */ - @Override protected void rewindInternal() { - // No-op. - } - - /** {@inheritDoc} */ - @Override protected Downstream requestDownstream(int idx) { - return null; + assertEquals(i, rootNode.next()[0]); } - /** {@inheritDoc} */ - @Override public void request(int rowsCnt) { - int r = requested.getAndAdd(rowsCnt); - - context().execute(() -> { - for (int i = 0; i < rowsCnt; i++) - downstream().push(new Object[] {r + i}); - }, this::onError); - } + assertEquals(fetch == 0, rootNode.hasNext()); } } diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DynamicParametersIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DynamicParametersIntegrationTest.java index daa39e374a018..ae807f96f0f34 100644 --- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DynamicParametersIntegrationTest.java +++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DynamicParametersIntegrationTest.java @@ -150,6 +150,9 @@ public void testDynamicParameters() { assertQuery("SELECT name LIKE '%' || ? || '%' FROM person where name is not null").withParams("go") .returns(true).returns(false).returns(false).returns(false).check(); + assertQuery("SELECT id FROM person ORDER BY id LIMIT ?").withParams(1).returns(0).check(); + assertQuery("SELECT id FROM person ORDER BY id LIMIT ?").withParams(new BigDecimal(1)).returns(0).check(); + assertQuery("SELECT id FROM person WHERE name LIKE ? ORDER BY id LIMIT ?").withParams("I%", 1) .returns(0).check(); 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..4aec497847200 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 @@ -17,15 +17,12 @@ package org.apache.ignite.internal.processors.query.calcite.integration; -import java.math.BigDecimal; import java.util.Arrays; import java.util.List; -import org.apache.calcite.sql.validate.SqlValidatorException; import org.apache.ignite.IgniteCache; import org.apache.ignite.cache.CacheMode; import org.apache.ignite.cache.QueryEntity; import org.apache.ignite.configuration.IgniteConfiguration; -import org.apache.ignite.internal.processors.query.IgniteSQLException; import org.apache.ignite.internal.processors.query.calcite.exec.rel.AbstractNode; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.X; @@ -101,40 +98,6 @@ public void testNestedLimitOffsetWithUnion() { ).returns(2).returns(4).check(); } - /** Tests correctness of fetch / offset params. */ - @Test - public void testInvalidLimitOffset() { - String bigInt = BigDecimal.valueOf(10000000000L).toString(); - - assertThrows("SELECT * FROM TEST_REPL OFFSET " + bigInt + " ROWS", - SqlValidatorException.class, "Illegal value of offset"); - - assertThrows("SELECT * FROM TEST_REPL FETCH FIRST " + bigInt + " ROWS ONLY", - SqlValidatorException.class, "Illegal value of fetch / limit"); - - assertThrows("SELECT * FROM TEST_REPL LIMIT " + bigInt, - SqlValidatorException.class, "Illegal value of fetch / limit"); - - assertThrows("SELECT * FROM TEST_REPL OFFSET -1 ROWS FETCH FIRST -1 ROWS ONLY", - IgniteSQLException.class, null); - - assertThrows("SELECT * FROM TEST_REPL OFFSET -1 ROWS", - IgniteSQLException.class, null); - - assertThrows("SELECT * FROM TEST_REPL OFFSET 2+1 ROWS", - IgniteSQLException.class, null); - - // Check with parameters - assertThrows("SELECT * FROM TEST_REPL OFFSET ? ROWS FETCH FIRST ? ROWS ONLY", - SqlValidatorException.class, "Illegal value of fetch / limit", -1, -1); - - assertThrows("SELECT * FROM TEST_REPL OFFSET ? ROWS", - SqlValidatorException.class, "Illegal value of offset", -1); - - assertThrows("SELECT * FROM TEST_REPL FETCH FIRST ? ROWS ONLY", - SqlValidatorException.class, "Illegal value of fetch / limit", -1); - } - /** * */ diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/AbstractPlannerTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/AbstractPlannerTest.java index b49af08b5c700..f05f4e1c70a29 100644 --- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/AbstractPlannerTest.java +++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/AbstractPlannerTest.java @@ -75,6 +75,7 @@ import org.apache.ignite.internal.processors.query.calcite.schema.IgniteSchema; import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution; import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory; +import org.apache.ignite.internal.util.lang.RunnableX; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.plugin.extensions.communication.Message; import org.apache.ignite.testframework.GridTestUtils; @@ -220,7 +221,33 @@ protected PlanningContext plannerCtx( @Nullable RelOptListener planLsnr, String... disabledRules ) { - return plannerCtx(sql, Collections.singleton(publicSchema), planLsnr, disabledRules); + return plannerCtx(sql, Collections.singleton(publicSchema), planLsnr, null, disabledRules); + } + + /** */ + private PlanningContext plannerCtx( + String sql, + Collection schemas, + @Nullable RelOptListener planLsnr, + Collection params, + Collection disabledRules + ) { + PlanningContext.Builder ctxBuilder = PlanningContext.builder() + .parentContext(Contexts.of(baseQueryContext(schemas), planLsnr)) + .query(sql); + + if (params != null) + ctxBuilder.parameters(params.toArray(Object[]::new)); + + PlanningContext ctx = ctxBuilder.build(); + + IgnitePlanner planner = ctx.planner(); + + assertNotNull(planner); + + planner.addDisabledRules(disabledRules); + + return ctx; } /** */ @@ -228,12 +255,17 @@ protected PlanningContext plannerCtx( String sql, Collection schemas, @Nullable RelOptListener planLsnr, + @Nullable List params, String... disabledRules ) { - PlanningContext ctx = PlanningContext.builder() + PlanningContext.Builder ctxBuilder = PlanningContext.builder() .parentContext(Contexts.of(baseQueryContext(schemas), planLsnr)) - .query(sql) - .build(); + .query(sql); + + if (params != null) + ctxBuilder.parameters(params.toArray(Object[]::new)); + + PlanningContext ctx = ctxBuilder.build(); IgnitePlanner planner = ctx.planner(); @@ -438,6 +470,21 @@ protected static TestTable createTable(IgniteSchema schema, String name, RelData return table; } + /** */ + protected void assertPlan( + TestPlanningContextBuilder ctxBuilder + ) throws Exception { + assertPlan(ctxBuilder, rel -> true); + } + + /** */ + protected void assertPlan( + TestPlanningContextBuilder ctxBuilder, + Predicate predicate + ) throws Exception { + invalidatePlan(ctxBuilder, predicate); + } + /** */ protected void assertPlan( String sql, @@ -445,7 +492,9 @@ protected void assertPlan( Predicate predicate, String... disabledRules ) throws Exception { - assertPlan(sql, schema, null, predicate, disabledRules); + TestPlanningContextBuilder builder = contextBuilder().query(sql).schema(schema).disabledRules(disabledRules); + + assertPlan(builder, predicate); } /** */ @@ -455,20 +504,34 @@ protected void assertPlan( Predicate predicate, String... disabledRules ) throws Exception { - assertPlan(sql, schemas, null, predicate, disabledRules); + TestPlanningContextBuilder builder = contextBuilder().query(sql).schemas(schemas).disabledRules(disabledRules); + + assertPlan(builder, predicate); } /** */ protected void assertPlan( String sql, - Collection schemas, - @Nullable RelOptListener planLsnr, + IgniteSchema schema, + RelOptListener planLsnr, Predicate predicate, String... disabledRules ) throws Exception { - IgniteRel plan = physicalPlan(plannerCtx(sql, schemas, planLsnr, disabledRules)); + TestPlanningContextBuilder builder = contextBuilder().query(sql).schema(schema).disabledRules(disabledRules) + .planListener(planLsnr); + + assertPlan(builder, predicate); + } + + /** */ + private void invalidatePlan( + TestPlanningContextBuilder ctxBuilder, + Predicate predicate + ) throws Exception { + IgniteRel plan = physicalPlan(plannerCtx(ctxBuilder.query, ctxBuilder.schemas, ctxBuilder.planListener, + ctxBuilder.params, ctxBuilder.disabledRules)); - checkSplitAndSerialization(plan, schemas); + checkSplitAndSerialization(plan, ctxBuilder.schemas); if (!predicate.test((T)plan)) { String invalidPlanMsg = "Invalid plan (" + lastErrorMsg + "):\n" + @@ -478,17 +541,6 @@ protected void assertPlan( } } - /** */ - protected void assertPlan( - String sql, - IgniteSchema schema, - @Nullable RelOptListener planLsnr, - Predicate predicate, - String... disabledRules - ) throws Exception { - assertPlan(sql, Collections.singletonList(schema), planLsnr, predicate, disabledRules); - } - /** * Predicate builder for "Instance of class" condition. */ @@ -694,7 +746,7 @@ protected Predicate hasColumns(String... cols) { * E.g. {@code createTable("MY_TABLE", distribution, "ID", Integer.class, "VAL", String.class)}. * @return Instance of the {@link TestTable}. */ - protected static TestTable createTable(String name, IgniteDistribution distr, Object... fields) { + static TestTable createTable(String name, IgniteDistribution distr, Object... fields) { return createTable(name, DEFAULT_TBL_SIZE, distr, fields); } @@ -805,4 +857,79 @@ class TestFailureProcessor extends FailureProcessor { return true; } } + + /** Test planning context builder. */ + public static class TestPlanningContextBuilder { + /** */ + private String query; + + /** */ + private Collection schemas; + + /** */ + private Collection params = List.of(); + + /** */ + private Collection disabledRules = List.of(); + + /** */ + @Nullable private RelOptListener planListener; + + /** */ + public TestPlanningContextBuilder query(String qry) { + query = qry; + return this; + } + + /** */ + public TestPlanningContextBuilder schema(IgniteSchema schemas) { + this.schemas = List.of(schemas); + return this; + } + + /** */ + public TestPlanningContextBuilder schemas(Collection schemas) { + this.schemas = List.copyOf(schemas); + return this; + } + + /** */ + public TestPlanningContextBuilder params(Collection params) { + this.params = List.copyOf(params); + return this; + } + + /** */ + public TestPlanningContextBuilder params(Object... params) { + this.params = Arrays.asList(params); + return this; + } + + /** */ + public TestPlanningContextBuilder disabledRules(String... rules) { + disabledRules = List.of(rules); + return this; + } + + /** */ + public TestPlanningContextBuilder planListener(@Nullable RelOptListener planListener) { + this.planListener = planListener; + return this; + } + } + + /** */ + public static TestPlanningContextBuilder contextBuilder() { + return new TestPlanningContextBuilder(); + } + + /** */ + @SuppressWarnings("ThrowableNotThrown") + static void assertThrows( + RunnableX run, + Class cls, + @Nullable String msg + ) { + GridTestUtils.assertThrows(null, run, cls, msg); + } } diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/DynamicParametersPlannerTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/DynamicParametersPlannerTest.java new file mode 100644 index 0000000000000..5fe33a4315038 --- /dev/null +++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/DynamicParametersPlannerTest.java @@ -0,0 +1,85 @@ +/* + * 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.planner; + +import java.math.BigInteger; +import org.apache.ignite.IgniteException; +import org.apache.ignite.internal.processors.query.calcite.schema.IgniteSchema; +import org.junit.Test; + +import static org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions.single; + +/** */ +public class DynamicParametersPlannerTest extends AbstractPlannerTest { + /** Dynamic parameters in LIMIT / OFFSET. */ + @Test + public void testLimitOffset() throws Exception { + IgniteSchema schema = createSchema(createTable("T1", single(), "c1", Integer.class)); + + TestPlanningContextBuilder builder = contextBuilder().query("SELECT * FROM t1 LIMIT ?").schema(schema); + + assertPlan(builder.params(Long.MAX_VALUE)); + + // Count of dynamic parameters need to be invalidated, remove it after: IGNITE-28906 + assertPlan(builder.params(Long.MAX_VALUE, -1)); + + assertThrows(() -> assertPlan(builder.params("a")), IgniteException.class, + "Incorrect type of a dynamic parameter. Expected but got "); + + BigInteger moreThanMaxLong = BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.ONE); + + assertThrows(() -> assertPlan(builder.params(moreThanMaxLong)), IgniteException.class, + "Illegal value of fetch / limit"); + + assertThrows(() -> assertPlan(builder.params(-1)), IgniteException.class, + "Illegal value of fetch / limit"); + + assertThrows(() -> assertPlan(builder.params((Object)null)), IgniteException.class, + "Incorrect type of a dynamic parameter. Expected but got "); + + // OFFSET. + builder.query("SELECT * FROM t1 OFFSET ?"); + + assertThrows(() -> assertPlan(builder.params(moreThanMaxLong)), IgniteException.class, + "Illegal value of offset"); + + assertThrows(() -> assertPlan(builder.params(-1)), IgniteException.class, + "Illegal value of offset"); + + assertThrows(() -> assertPlan(builder.params((Object)null)), IgniteException.class, + "Incorrect type of a dynamic parameter. Expected but got "); + + // OFFSET Alternate syntax. + builder.query("SELECT * FROM t1 OFFSET ? ROWS"); + + assertThrows(() -> assertPlan(builder.params(moreThanMaxLong)), IgniteException.class, + "Illegal value of offset"); + + assertThrows(() -> assertPlan(builder.params(-1)), IgniteException.class, + "Illegal value of offset"); + + assertThrows(() -> assertPlan(builder.params((Object)null)), IgniteException.class, + "Incorrect type of a dynamic parameter. Expected but got "); + + // Expression. + builder.query("SELECT * FROM TEST_REPL OFFSET 2+? ROWS"); + + assertThrows(() -> assertPlan(builder), IgniteException.class, + "Encountered \" \"+\""); + } +} diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/HashJoinPlannerTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/HashJoinPlannerTest.java index 97a8689535e98..55140639755bd 100644 --- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/HashJoinPlannerTest.java +++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/HashJoinPlannerTest.java @@ -28,10 +28,10 @@ import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.testframework.GridTestUtils; import org.junit.Test; import static org.apache.calcite.rel.RelFieldCollation.Direction.ASCENDING; -import static org.apache.ignite.testframework.GridTestUtils.assertThrows; /** */ public class HashJoinPlannerTest extends AbstractPlannerTest { @@ -148,7 +148,7 @@ public void testHashJoinApplied() throws Exception { if (canBePlanned) assertPlan(sql0, schema, nodeOrAnyChild(isInstanceOf(IgniteHashJoin.class)), DISABLED_RULES); else { - assertThrows(null, () -> physicalPlan(sql0, schema, DISABLED_RULES), CannotPlanException.class, + GridTestUtils.assertThrows(null, () -> physicalPlan(sql0, schema, DISABLED_RULES), CannotPlanException.class, "There are not enough rules"); } } diff --git a/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java b/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java index 7daf90205ddda..be2d0283c15c9 100644 --- a/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java +++ b/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java @@ -23,6 +23,7 @@ import org.apache.ignite.internal.processors.query.calcite.planner.CorrelatedNestedLoopJoinPlannerTest; import org.apache.ignite.internal.processors.query.calcite.planner.CorrelatedSubqueryPlannerTest; import org.apache.ignite.internal.processors.query.calcite.planner.DataTypesPlannerTest; +import org.apache.ignite.internal.processors.query.calcite.planner.DynamicParametersPlannerTest; import org.apache.ignite.internal.processors.query.calcite.planner.HashAggregatePlannerTest; import org.apache.ignite.internal.processors.query.calcite.planner.HashIndexSpoolPlannerTest; import org.apache.ignite.internal.processors.query.calcite.planner.HashJoinPlannerTest; @@ -96,6 +97,7 @@ AbstractPlannerUtilityTest.class, HintsTestSuite.class, + DynamicParametersPlannerTest.class, }) public class PlannerTestSuite { } diff --git a/modules/calcite/src/test/sql/order/test_limit.test b/modules/calcite/src/test/sql/order/test_limit.test index 2c2c3d34db9a6..5d611a7529963 100644 --- a/modules/calcite/src/test/sql/order/test_limit.test +++ b/modules/calcite/src/test/sql/order/test_limit.test @@ -17,9 +17,47 @@ SELECT a FROM test ORDER BY a LIMIT 1 ---- 11 -# LIMIT with non-scalar should fail +# decimal limit +query I +SELECT a FROM test ORDER BY a FETCH FIRST 0 ROWS ONLY +---- + +# decimal limit +query I +SELECT a FROM test ORDER BY a LIMIT 1.2 +---- +11 + +# decimal limit +query I +SELECT a FROM test ORDER BY a FETCH FIRST 1.2 ROWS ONLY +---- +11 + +# decimal offset/limit +query I +SELECT a FROM test ORDER BY a OFFSET 1.1 ROWS FETCH FIRST 1.1 ROWS ONLY +---- +12 + +# Unexpected literal +statement error +SELECT a FROM test ORDER BY a FETCH FIRST '1' + +# More than max of big integer +statement error +SELECT a FROM test LIMIT 9223372036854775808 + +# More than max of big integer +statement error +SELECT a FROM test ORDER BY a FETCH FIRST 9223372036854775808 ROWS ONLY + +# Unexpected operation +statement error +SELECT a FROM test ORDER BY a FETCH FIRST 1 + 1 ROWS ONLY + statement error -SELECT a FROM test LIMIT a +SELECT a FROM test ORDER BY a FETCH FIRST -1 ROWS ONLY # LIMIT with non-scalar operation should also fail statement error