IGNITE-28971 SQL Calcite: Introduce value resolver for FETCH, OFFSET, and LIMIT

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java

@@ -18,7 +18,6 @@
 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;
@@ -126,6 +125,7 @@
 import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
 import org.apache.ignite.internal.processors.query.calcite.schema.IgniteIndex;
 import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlPaginationPolicy;
 import org.apache.ignite.internal.processors.query.calcite.trait.Destination;
 import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
 import org.apache.ignite.internal.processors.query.calcite.trait.TraitUtils;
@@ -1094,7 +1094,8 @@ private long validateAndGetFetchOffsetParams(RexNode node, String op) {
             if (paramAsDecimal.signum() < 0)
                 throw new IllegalArgumentException("Negative value for " + op);
 
-            return IgniteMath.convertToLongExact(paramAsDecimal, RoundingMode.DOWN);
+            IgniteSqlPaginationPolicy pagPlc = ctx.unwrap(IgniteSqlPaginationPolicy.class);
+            return IgniteSqlPaginationPolicy.convertToLongExact(paramAsDecimal, pagPlc);
         }
         catch (RuntimeException ex) {
             throw new IgniteSQLException(IgniteResource.INSTANCE.illegalFetchLimit(op).str(),

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/AbstractQueryContext.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.query.calcite.prepare;
 
 import org.apache.calcite.plan.Context;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Abstract query context.
@@ -32,7 +33,7 @@ public AbstractQueryContext(Context parentCtx) {
     }
 
     /** {@inheritDoc} */
-    @Override public <C> C unwrap(Class<C> aCls) {
+    @Override public <C> @Nullable C unwrap(Class<C> aCls) {
         if (aCls == getClass())
             return aCls.cast(this);
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgnitePlanner.java

@@ -91,6 +91,7 @@
 import org.apache.ignite.internal.processors.query.IgniteSQLException;
 import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMetadata;
 import org.apache.ignite.internal.processors.query.calcite.metadata.RelMetadataQueryEx;
+import org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlPaginationPolicy;
 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.util.typedef.F;
@@ -365,7 +366,7 @@ private static boolean isAsCall(SqlNode node) {
         }
 
         CalciteCatalogReader catalogReader = this.catalogReader.withSchemaPath(schemaPath);
-        SqlValidator validator = new IgniteSqlValidator(operatorTbl, catalogReader, typeFactory, validatorCfg, ctx.parameters());
+        SqlValidator validator = createSqlValidator(catalogReader);
         SqlToRelConverter sqlToRelConverter = sqlToRelConverter(validator, catalogReader, sqlToRelConverterCfg);
         RelRoot root = sqlToRelConverter.convertQuery(sqlNode, true, false);
         root = root.withRel(sqlToRelConverter.decorrelate(sqlNode, root.rel));
@@ -435,7 +436,7 @@ public boolean isAggregate(SqlSelect select, @Nullable SqlNodeList orderList) {
     /** */
     private SqlValidator validator() {
         if (validator == null)
-            validator = new IgniteSqlValidator(operatorTbl, catalogReader, typeFactory, validatorCfg, ctx.parameters());
+            validator = createSqlValidator();
 
         return validator;
     }
@@ -807,4 +808,21 @@ protected VolcanoPlannerExt(RelOptCostFactory costFactory, Context externalCtx)
             super.checkCancel();
         }
     }
+
+    /** */
+    private SqlValidator createSqlValidator(CalciteCatalogReader catalogReader) {
+        return new IgniteSqlValidator(
+            operatorTbl,
+            catalogReader,
+            typeFactory,
+            validatorCfg,
+            ctx.parameters(),
+            ctx.unwrap(IgniteSqlPaginationPolicy.class)
+        );
+    }
+
+    /** */
+    private SqlValidator createSqlValidator() {
+        return createSqlValidator(catalogReader);
+    }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java

@@ -18,7 +18,6 @@
 package org.apache.ignite.internal.processors.query.calcite.prepare;
 
 import java.math.BigDecimal;
-import java.math.RoundingMode;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.EnumSet;
@@ -77,6 +76,7 @@
 import org.apache.ignite.internal.processors.query.calcite.schema.IgniteCacheTable;
 import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
 import org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlDecimalLiteral;
+import org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlPaginationPolicy;
 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;
@@ -115,25 +115,31 @@ public class IgniteSqlValidator extends SqlValidatorImpl {
     /** */
     private final RelDataType nullType;
 
+    /** */
+    private final @Nullable IgniteSqlPaginationPolicy pagPlc;
+
     /**
      * Creates a validator.
      *
-     * @param opTab         Operator table
-     * @param catalogReader Catalog reader
-     * @param typeFactory   Type factory
-     * @param cfg           Config
-     * @param parameters    Dynamic parameters
+     * @param opTab Operator table.
+     * @param catalogReader Catalog reader.
+     * @param typeFactory Type factory.
+     * @param cfg Config.
+     * @param parameters Dynamic parameters.
+     * @param pagPlc Pagination policy.
      */
     public IgniteSqlValidator(
         SqlOperatorTable opTab,
         CalciteCatalogReader catalogReader,
         IgniteTypeFactory typeFactory,
         SqlValidator.Config cfg,
-        @Nullable Object[] parameters
+        @Nullable Object[] parameters,
+        @Nullable IgniteSqlPaginationPolicy pagPlc
     ) {
         super(opTab, catalogReader, typeFactory, cfg);
 
         this.parameters = parameters;
+        this.pagPlc = pagPlc;
 
         nullType = typeFactory.createSqlType(SqlTypeName.NULL);
     }
@@ -385,7 +391,7 @@ private void checkLimitOffset(Number offsetFetchLimit, SqlNode n, String nodeNam
             if (val.signum() < 0)
                 throw new IllegalArgumentException("Negative value for " + nodeName);
 
-            IgniteMath.convertToLongExact(val, RoundingMode.DOWN);
+            IgniteSqlPaginationPolicy.convertToLongExact(val, pagPlc);
         }
         catch (RuntimeException e) {
             throw newValidationError(n, IgniteResource.INSTANCE.illegalFetchLimit(nodeName));

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlanningContext.java

@@ -161,7 +161,7 @@ public RelOptCluster cluster() {
     }
 
     /** {@inheritDoc} */
-    @Override public <C> C unwrap(Class<C> aCls) {
+    @Override public <C> @Nullable C unwrap(Class<C> aCls) {
         if (aCls == getClass())
             return aCls.cast(this);
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/sql/IgniteSqlPaginationPolicy.java

@@ -0,0 +1,40 @@
+/*
+ * 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.sql;
+
+import java.math.RoundingMode;
+import org.apache.calcite.plan.Context;
+import org.apache.calcite.tools.Frameworks;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteMath;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Defines a policy for processing values of SQL pagination clauses: LIMIT, FETCH, and OFFSET.
+ *
+ * <p>Custom instance can be supplied through {@link Frameworks.ConfigBuilder#context(Context)}.</p>
+ */
+@FunctionalInterface
+public interface IgniteSqlPaginationPolicy {
+    /** Returns the rounding mode for FETCH, LIMIT and OFFSET values. */
+    RoundingMode roundingMode();
+
+    /** Rounds the given value according to the specified policy and converts it to {@code long}. */
+    static long convertToLongExact(Number value, @Nullable IgniteSqlPaginationPolicy policy) {
+        RoundingMode roundingMode = policy == null ? IgniteMath.NUMERIC_ROUNDING_MODE : policy.roundingMode();
+        return IgniteMath.convertToLongExact(value, roundingMode);
+    }
+}

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DynamicParametersIntegrationTest.java

@@ -193,15 +193,15 @@ public void testDynamicParameters() {
     public void testFractionalLimitOffset() {
         createAndPopulateTable();
 
-        assertQuery("SELECT id FROM person ORDER BY id LIMIT ?").withParams(0.5).resultSize(0).check();
+        assertQuery("SELECT id FROM person ORDER BY id LIMIT ?").withParams(0.5).returns(0).check();
         assertQuery("SELECT id FROM person ORDER BY id LIMIT ?").withParams(1.4).returns(0).check();
-        assertQuery("SELECT id FROM person ORDER BY id LIMIT ?").withParams(1.6).returns(0).check();
+        assertQuery("SELECT id FROM person ORDER BY id LIMIT ?").withParams(1.6).returns(0).returns(1).check();
         assertThrowsSqlException("SELECT id FROM person ORDER BY id LIMIT ?", null, BigDecimal.valueOf(-1.5));
         assertThrowsSqlException("SELECT id FROM person ORDER BY id LIMIT ?", null, BigDecimal.valueOf(-0.5));
 
         assertQuery("SELECT id FROM person ORDER BY id FETCH FIRST ? ROWS ONLY")
             .withParams(BigDecimal.valueOf(0.5))
-            .resultSize(0)
+            .returns(0)
             .check();
         assertQuery("SELECT id FROM person ORDER BY id FETCH FIRST ? ROWS ONLY")
             .withParams(BigDecimal.valueOf(1.3))
@@ -210,13 +210,13 @@ public void testFractionalLimitOffset() {
         assertQuery("SELECT id FROM person ORDER BY id FETCH FIRST ? ROWS ONLY")
             .withParams(BigDecimal.valueOf(1.6))
             .returns(0)
+            .returns(1)
             .check();
         assertThrowsSqlException("SELECT id FROM person ORDER BY id FETCH FIRST ? ROWS ONLY", null, BigDecimal.valueOf(-1.5));
         assertThrowsSqlException("SELECT id FROM person ORDER BY id FETCH FIRST ? ROWS ONLY", null, BigDecimal.valueOf(-0.5));
 
         assertQuery("SELECT id FROM person ORDER BY id OFFSET ? ROWS")
             .withParams(BigDecimal.valueOf(0.5))
-            .returns(0)
             .returns(1)
             .returns(2)
             .returns(3)
@@ -230,7 +230,6 @@ public void testFractionalLimitOffset() {
             .check();
         assertQuery("SELECT id FROM person ORDER BY id OFFSET ? ROWS")
             .withParams(BigDecimal.valueOf(2.6))
-            .returns(2)
             .returns(3)
             .returns(4)
             .check();

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java

@@ -149,18 +149,18 @@ public void testInvalidLimitOffset() {
     public void testFractionalLimitOffset() throws Exception {
         fillCache(cacheRepl, 4);
 
-        assertQuery("SELECT id FROM TEST_REPL ORDER BY id LIMIT 0.5").check();
+        assertQuery("SELECT id FROM TEST_REPL ORDER BY id LIMIT 0.5").returns(0).check();
         assertQuery("SELECT id FROM TEST_REPL ORDER BY id LIMIT 1.2").returns(0).check();
-        assertQuery("SELECT id FROM TEST_REPL ORDER BY id LIMIT 1.5").returns(0).check();
+        assertQuery("SELECT id FROM TEST_REPL ORDER BY id LIMIT 1.5").returns(0).returns(1).check();
 
-        assertQuery("SELECT id FROM TEST_REPL ORDER BY id FETCH FIRST 0.5 ROWS ONLY").check();
+        assertQuery("SELECT id FROM TEST_REPL ORDER BY id FETCH FIRST 0.5 ROWS ONLY").returns(0).check();
         assertQuery("SELECT id FROM TEST_REPL ORDER BY id FETCH FIRST 1.3 ROWS ONLY").returns(0).check();
-        assertQuery("SELECT id FROM TEST_REPL ORDER BY id FETCH FIRST 1.6 ROWS ONLY").returns(0).check();
+        assertQuery("SELECT id FROM TEST_REPL ORDER BY id FETCH FIRST 1.6 ROWS ONLY").returns(0).returns(1).check();
 
         assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET 0.5 ROWS")
-            .returns(0).returns(1).returns(2).returns(3).check();
+            .returns(1).returns(2).returns(3).check();
         assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET 2.3 ROWS").returns(2).returns(3).check();
-        assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET 2.6 ROWS").returns(2).returns(3).check();
+        assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET 2.6 ROWS").returns(3).check();
     }
 
     /**
@@ -352,7 +352,7 @@ public void testFetchExpressionWithoutPushDown() throws Exception {
 
         assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET 1 ROWS "
                 + "FETCH FIRST (ABS(0.5)) ROWS ONLY")
-            .resultSize(0)
+            .returns(1)
             .check();
 
         assertQuery("SELECT id FROM TEST_REPL ORDER BY id OFFSET 1 ROWS "

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/OperatorsExtensionIntegrationTest.java

@@ -17,6 +17,7 @@
 package org.apache.ignite.internal.processors.query.calcite.integration;
 
 import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.sql.Timestamp;
 import java.util.HashSet;
 import java.util.List;
@@ -67,6 +68,7 @@
 import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteConvertletTable;
 import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteSqlNodeRewriter;
 import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteSqlValidator;
+import org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlPaginationPolicy;
 import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
 import org.apache.ignite.plugin.AbstractTestPluginProvider;
 import org.apache.ignite.plugin.PluginContext;
@@ -96,6 +98,7 @@ public class OperatorsExtensionIntegrationTest extends AbstractBasicIntegrationT
                                     .withSqlNodeRewriter(new SqlRewriter()))
                             .context(Contexts.chain(
                                 CalciteQueryProcessor.FRAMEWORK_CONFIG.getContext(),
+                                Contexts.of((IgniteSqlPaginationPolicy)() -> RoundingMode.DOWN),
                                 Contexts.of(new AccumulatorFactoryProviderImpl())))
                             .build();
 
@@ -216,6 +219,23 @@ public void testRowNumRewrite() {
             .check();
     }
 
+    /** */
+    @Test
+    public void testPaginationRoundingPolicy() {
+        assertQuery("SELECT x FROM (VALUES (0), (1), (2)) t(x) ORDER BY x LIMIT 1.9")
+            .returns(0)
+            .check();
+
+        assertQuery("SELECT x FROM (VALUES (0), (1), (2)) t(x) ORDER BY x FETCH FIRST 1.9 ROWS ONLY")
+            .returns(0)
+            .check();
+
+        assertQuery("SELECT x FROM (VALUES (0), (1), (2)) t(x) ORDER BY x OFFSET 1.9 ROWS")
+            .returns(1)
+            .returns(2)
+            .check();
+    }
+
     /** Rewrites LTRIM with 2 parameters. */
     public static SqlCall rewriteLtrim(SqlValidator validator, SqlCall call) {
         if (call.operandCount() != 2)

modules/calcite/src/test/sql/order/test_limit.test

@@ -28,12 +28,14 @@ query I
 SELECT a FROM test ORDER BY a LIMIT 1.5
 ----
 11
+12
 
 # decimal limit
 query I
 SELECT a FROM test ORDER BY a LIMIT 1.6
 ----
 11
+12
 
 # decimal limit
 query I
@@ -57,12 +59,14 @@ query I
 SELECT a FROM test ORDER BY a FETCH FIRST 1.5 ROWS ONLY
 ----
 11
+12
 
 # decimal limit
 query I
 SELECT a FROM test ORDER BY a FETCH FIRST 1.6 ROWS ONLY
 ----
 11
+12
 
 # decimal offset/limit
 query I