tkalkirill commented on code in PR #13311:
URL: https://github.com/apache/ignite/pull/13311#discussion_r3711064137
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java:
##########
@@ -37,6 +37,9 @@
* Abstract node of execution tree.
*/
public abstract class AbstractNode<Row> implements Node<Row> {
+ /** Special value to highlight that all rows were received and we do not
expect more. */
+ static final int NOT_WAITING = -1;
Review Comment:
Why is this constant declared in `AbstractNode` if the class does not have a
corresponding waiting state? Should it be kept in the concrete nodes instead?
We should also handle the similar constants declared in other subclasses
consistently.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java:
##########
@@ -17,66 +17,73 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
-import java.util.function.Supplier;
import org.apache.calcite.rel.type.RelDataType;
import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteMath;
import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.Nullable;
/** Offset, fetch|limit support node. */
public class LimitNode<Row> extends AbstractNode<Row> implements
SingleNode<Row>, Downstream<Row> {
- /** Offset if its present, otherwise 0. */
- private final int offset;
+ /** Offset param. */
+ private final long offset;
- /** Fetch if its present, otherwise 0. */
- private final int fetch;
+ /** Fetch param. */
+ private final long fetch;
- /** Already processed (pushed to upstream) rows count. */
- private int rowsProcessed;
+ /** Summary rows to process. */
+ private final long rowsSummary;
- /** Fetch can be unset, in this case we need all rows. */
- private @Nullable Supplier<Integer> fetchNode;
+ /** Already processed (pushed to downstream) rows count. */
+ private long rowsProcessed;
/** Waiting results counter. */
private int waiting;
+ /** Upper requested rows. */
+ private int requested;
+
/**
* Constructor.
*
* @param ctx Execution context.
* @param rowType Row type.
+ * @param offset How many rows need to be skipped.
+ * @param fetch How many rows need to be processed.
*/
public LimitNode(
ExecutionContext<Row> ctx,
RelDataType rowType,
- Supplier<Integer> offsetNode,
- Supplier<Integer> fetchNode
+ long offset,
+ long fetch
Review Comment:
The documentation says nothing about what `-1` means.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java:
##########
@@ -17,66 +17,73 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
-import java.util.function.Supplier;
import org.apache.calcite.rel.type.RelDataType;
import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteMath;
import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.Nullable;
/** Offset, fetch|limit support node. */
public class LimitNode<Row> extends AbstractNode<Row> implements
SingleNode<Row>, Downstream<Row> {
- /** Offset if its present, otherwise 0. */
- private final int offset;
+ /** Offset param. */
+ private final long offset;
- /** Fetch if its present, otherwise 0. */
- private final int fetch;
+ /** Fetch param. */
+ private final long fetch;
- /** Already processed (pushed to upstream) rows count. */
- private int rowsProcessed;
+ /** Summary rows to process. */
+ private final long rowsSummary;
- /** Fetch can be unset, in this case we need all rows. */
- private @Nullable Supplier<Integer> fetchNode;
+ /** Already processed (pushed to downstream) rows count. */
+ private long rowsProcessed;
/** Waiting results counter. */
private int waiting;
+ /** Upper requested rows. */
+ private int requested;
+
/**
* Constructor.
*
* @param ctx Execution context.
* @param rowType Row type.
+ * @param offset How many rows need to be skipped.
+ * @param fetch How many rows need to be processed.
*/
public LimitNode(
ExecutionContext<Row> ctx,
RelDataType rowType,
- Supplier<Integer> offsetNode,
- Supplier<Integer> fetchNode
+ long offset,
+ long fetch
) {
super(ctx, rowType);
- offset = offsetNode == null ? 0 : offsetNode.get();
- fetch = fetchNode == null ? 0 : fetchNode.get();
- this.fetchNode = fetchNode;
+ this.offset = offset;
+ rowsSummary = fetch == -1 ? Long.MAX_VALUE :
IgniteMath.addExact(fetch, offset);
+ this.fetch = fetch == -1 ? 0 : fetch;
Review Comment:
The documentation says nothing about what `0` means.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java:
##########
@@ -1050,4 +1054,33 @@ private ScanStorageNode<Row> createStorageScan(
otherColMapping
);
}
+
+ /** */
+ private long validateAndGetFetchOffsetParams(RexNode node, String op) {
Review Comment:
Let’s have two separate methods for `fetch` and `offset`; validation can be
shared, but each will handle nulls in its own way. This will make the code
cleaner and easier to understand.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -241,54 +241,80 @@ 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);
+ validateFetchOffset(select.getFetch(), "fetch / limit");
+ validateFetchOffset(select.getOffset(), "offset");
}
/**
- * @param n Node to check limit.
+ * Invalidate fetch/offset params restrictions.
+ *
+ * @param n Node to check limit.
Review Comment:
Why just `LIMIT`? There’s `OFFSET`/`FETCH` and `LIMIT` here.
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitExecutionTest.java:
##########
@@ -78,10 +79,10 @@ private void checkLimitSort(int offset, int fetch) {
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, offset,
+ fetch == 0 ? -1 : fetch);
Review Comment:
Magic values.
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitExecutionTest.java:
##########
@@ -109,7 +110,7 @@ private void checkLimit(int offset, int fetch) {
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, offset,
fetch == 0 ? -1 : fetch);
Review Comment:
Magic values.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java:
##########
@@ -17,66 +17,73 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
-import java.util.function.Supplier;
import org.apache.calcite.rel.type.RelDataType;
import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteMath;
import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.Nullable;
/** Offset, fetch|limit support node. */
public class LimitNode<Row> extends AbstractNode<Row> implements
SingleNode<Row>, Downstream<Row> {
- /** Offset if its present, otherwise 0. */
- private final int offset;
+ /** Offset param. */
+ private final long offset;
- /** Fetch if its present, otherwise 0. */
- private final int fetch;
+ /** Fetch param. */
+ private final long fetch;
- /** Already processed (pushed to upstream) rows count. */
- private int rowsProcessed;
+ /** Summary rows to process. */
+ private final long rowsSummary;
- /** Fetch can be unset, in this case we need all rows. */
- private @Nullable Supplier<Integer> fetchNode;
+ /** Already processed (pushed to downstream) rows count. */
+ private long rowsProcessed;
/** Waiting results counter. */
private int waiting;
+ /** Upper requested rows. */
+ private int requested;
+
/**
* Constructor.
*
* @param ctx Execution context.
* @param rowType Row type.
+ * @param offset How many rows need to be skipped.
+ * @param fetch How many rows need to be processed.
*/
public LimitNode(
ExecutionContext<Row> ctx,
RelDataType rowType,
- Supplier<Integer> offsetNode,
- Supplier<Integer> fetchNode
+ long offset,
+ long fetch
) {
super(ctx, rowType);
- offset = offsetNode == null ? 0 : offsetNode.get();
- fetch = fetchNode == null ? 0 : fetchNode.get();
- this.fetchNode = fetchNode;
+ this.offset = offset;
+ rowsSummary = fetch == -1 ? Long.MAX_VALUE :
IgniteMath.addExact(fetch, offset);
Review Comment:
Check `offset + fetch` for overflow in the constructor and remove the
redundant `rowsSummary` field. After validation, use the immutable `offset` and
`fetch` values directly in subsequent calculations.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java:
##########
@@ -58,21 +57,21 @@ public class SortNode<Row> extends MemoryTrackingNode<Row>
implements SingleNode
public SortNode(
ExecutionContext<Row> ctx, RelDataType rowType,
Comparator<Row> comp,
- @Nullable Supplier<Integer> offset,
- @Nullable Supplier<Integer> fetch
+ long offset,
+ long fetch
Review Comment:
The documentation says nothing about what `-1` means.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java:
##########
@@ -58,21 +57,21 @@ public class SortNode<Row> extends MemoryTrackingNode<Row>
implements SingleNode
public SortNode(
ExecutionContext<Row> ctx, RelDataType rowType,
Comparator<Row> comp,
- @Nullable Supplier<Integer> offset,
- @Nullable Supplier<Integer> fetch
+ long offset,
+ long fetch
) {
super(ctx, rowType);
- assert fetch == null || fetch.get() >= 0;
- assert offset == null || offset.get() >= 0;
+ assert fetch == -1 || fetch > 0;
+ assert offset >= 0;
Review Comment:
You need to include at least the value in the error message.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java:
##########
@@ -58,21 +57,21 @@ public class SortNode<Row> extends MemoryTrackingNode<Row>
implements SingleNode
public SortNode(
ExecutionContext<Row> ctx, RelDataType rowType,
Comparator<Row> comp,
- @Nullable Supplier<Integer> offset,
- @Nullable Supplier<Integer> fetch
+ long offset,
+ long fetch
) {
super(ctx, rowType);
- assert fetch == null || fetch.get() >= 0;
- assert offset == null || offset.get() >= 0;
+ assert fetch == -1 || fetch > 0;
Review Comment:
You need to include at least the value in the error message.
##########
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();
Review Comment:
Missing, Double, Float, Long.
Missing floating-point value, for example, 1.5.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -241,54 +241,80 @@ 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);
+ validateFetchOffset(select.getFetch(), "fetch / limit");
+ validateFetchOffset(select.getOffset(), "offset");
}
/**
- * @param n Node to check limit.
+ * Invalidate fetch/offset params restrictions.
Review Comment:
Why "Invalidate"?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -241,54 +241,80 @@ 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);
+ validateFetchOffset(select.getFetch(), "fetch / limit");
+ validateFetchOffset(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 validateFetchOffset(@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<SqlValidatorException> 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) {
Review Comment:
What does this change have to do with it?
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java:
##########
@@ -104,15 +104,15 @@ public void testNestedLimitOffsetWithUnion() {
/** Tests correctness of fetch / offset params. */
@Test
public void testInvalidLimitOffset() {
Review Comment:
Missing, Double, Float, Long.
Missing floating-point value, for example, 1.5.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java:
##########
@@ -1050,4 +1054,33 @@ private ScanStorageNode<Row> createStorageScan(
otherColMapping
);
}
+
+ /** */
+ private long validateAndGetFetchOffsetParams(RexNode node, String op) {
+ Supplier<Object> scalar = expressionFactory.execute(node);
+ Object param = scalar.get();
+
+ if (!(param instanceof Number)) {
+ String actual = param == null ? "null" :
param.getClass().getSimpleName();
+ throw new
IgniteSQLException(IgniteResource.INSTANCE.incorrectDynamicParameterType("BIGINT",
actual).str(),
Review Comment:
Why use a string literal when you can use the type name `SqlTypeName.BIGINT`?
##########
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 {
Review Comment:
Missing, Double, Float, Long.
Missing floating-point value, for example, 1.5.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
