[CALCITE-7624] Support BigDecimal for FETCH and OFFSET in Enumerable

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java

@@ -72,6 +72,7 @@
 import java.lang.reflect.Modifier;
 import java.lang.reflect.Type;
 import java.math.BigDecimal;
+import java.math.BigInteger;
 import java.math.RoundingMode;
 import java.sql.Date;
 import java.sql.Time;
@@ -112,6 +113,36 @@ private EnumUtils() {}
   public static final List<String> LEFT_RIGHT =
       ImmutableList.of("left", "right");
 
+  /** Converts a FETCH or OFFSET value to the representation
+   * supported by the enumerable runtime. */
+  public static BigDecimal numberToBigDecimalLimit(Object value, String kind) {
+    return numberToBigDecimalLimit(value, kind, FetchOffsetRoundingPolicy.NONE);
+  }
+
+  /** Converts a FETCH or OFFSET value to the representation
+   * supported by the enumerable runtime. */
+  public static BigDecimal numberToBigDecimalLimit(Object value, String kind,
+      FetchOffsetRoundingPolicy roundingPolicy) {
+    if (!(value instanceof Number)) {
+      throw new IllegalArgumentException(kind + " must be a number");
+    }
+    final Number number = (Number) value;
+    final BigDecimal decimal;
+    if (number instanceof BigDecimal) {
+      decimal = (BigDecimal) number;
+    } else if (number instanceof BigInteger) {
+      decimal = new BigDecimal((BigInteger) number);
+    } else if (number instanceof Float || number instanceof Double) {
+      decimal = BigDecimal.valueOf(number.doubleValue());
+    } else {
+      decimal = BigDecimal.valueOf(number.longValue());
+    }
+    if (decimal.signum() < 0) {
+      throw new IllegalArgumentException(kind + " must not be negative");
+    }
+    return roundingPolicy.round(decimal);
+  }
+
   /** Declares a method that overrides another method. */
   public static MethodDeclaration overridingMethodDecl(Method method,
       Iterable<ParameterExpression> parameters,

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableLimit.java

@@ -101,36 +101,50 @@ public static EnumerableLimit create(final RelNode input, @Nullable RexNode offs
             result.format);
 
     Expression v = builder.append("child", result.block);
+    Expression roundingPolicyExp = getRoundingPolicy(implementor);
     if (offset != null) {
       v =
           builder.append("offset",
-              Expressions.call(v, BuiltInMethod.SKIP.method,
-                  getExpression(offset)));
+              Expressions.call(BuiltInMethod.SKIP_BIG_DECIMAL.method, v,
+                  getExpression(offset, "OFFSET", roundingPolicyExp)));
     }
     if (fetch != null) {
       v =
           builder.append("fetch",
-              Expressions.call(v, BuiltInMethod.TAKE.method,
-                  getExpression(fetch)));
+              Expressions.call(BuiltInMethod.TAKE_BIG_DECIMAL.method, v,
+                  getExpression(fetch, "FETCH", roundingPolicyExp)));
     }
 
     builder.add(Expressions.return_(null, v));
     return implementor.result(physType, builder.toBlock());
   }
 
-  static Expression getExpression(RexNode rexNode) {
+  static Expression getExpression(RexNode rexNode, String kind,
+      Expression roundingPolicy) {
+    final Expression value;
     if (rexNode instanceof RexDynamicParam) {
       final RexDynamicParam param = (RexDynamicParam) rexNode;
-      return Expressions.convert_(
+      value =
           Expressions.call(DataContext.ROOT,
               BuiltInMethod.DATA_CONTEXT_GET.method,
-              Expressions.constant("?" + param.getIndex())),
-          Integer.class);
+              Expressions.constant("?" + param.getIndex()));
     } else {
-      // TODO: Enumerable runtime only supports INT types for FETCH and OFFSET, not BIGINT types.
-      //  Currently, using BIGINT types for execution will result in an error message.
-      //  This issue needs to be fixed. For more information, see CALCITE-7156.
-      return Expressions.constant(RexLiteral.intValue(rexNode));
+      value = Expressions.constant(RexLiteral.bigDecimalValue(rexNode));
     }
+    return Expressions.call(
+        BuiltInMethod.NUMBER_TO_BIG_DECIMAL_LIMIT.method,
+        value,
+        Expressions.constant(kind),
+        roundingPolicy);
+  }
+
+  static Expression getRoundingPolicy(EnumerableRelImplementor implementor) {
+    final Object roundingPolicy =
+        implementor.map.get(EnumerableRelImplementor.FETCH_OFFSET_ROUNDING_POLICY);
+    if (roundingPolicy == null) {
+      return Expressions.field(null, FetchOffsetRoundingPolicy.class, "NONE");
+    }
+    return implementor.stash((FetchOffsetRoundingPolicy) roundingPolicy,
+        FetchOffsetRoundingPolicy.class);
   }
 }

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableLimitSort.java

@@ -30,7 +30,10 @@
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
+import java.math.BigDecimal;
+
 import static org.apache.calcite.adapter.enumerable.EnumerableLimit.getExpression;
+import static org.apache.calcite.adapter.enumerable.EnumerableLimit.getRoundingPolicy;
 
 /**
  * Implementation of {@link org.apache.calcite.rel.core.Sort} in
@@ -95,19 +98,20 @@ public static EnumerableLimitSort create(
     final PhysType inputPhysType = result.physType;
     final Pair<Expression, Expression> pair =
         inputPhysType.generateCollationKey(this.collation.getFieldCollations());
+    final Expression roundingPolicyExp = getRoundingPolicy(implementor);
 
     final Expression fetchVal;
     if (this.fetch == null) {
-      fetchVal = Expressions.constant(Integer.MAX_VALUE);
+      fetchVal = Expressions.constant(BigDecimal.valueOf(Integer.MAX_VALUE));
     } else {
-      fetchVal = getExpression(this.fetch);
+      fetchVal = getExpression(this.fetch, "FETCH", roundingPolicyExp);
     }
 
     final Expression offsetVal;
     if (this.offset == null) {
-      offsetVal = Expressions.constant(0);
+      offsetVal = Expressions.constant(BigDecimal.ZERO);
     } else {
-      offsetVal = getExpression(this.offset);
+      offsetVal = getExpression(this.offset, "OFFSET", roundingPolicyExp);
     }
 
     builder.add(

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRelImplementor.java

@@ -78,6 +78,9 @@
  * operators of {@link EnumerableConvention} calling convention.
  */
 public class EnumerableRelImplementor extends JavaRelImplementor {
+  public static final String FETCH_OFFSET_ROUNDING_POLICY =
+      "_fetchOffsetRoundingPolicy";
+
   public final Map<String, Object> map;
   private final Map<String, RexToLixTranslator.InputGetter> corrVars =
       new HashMap<>();

core/src/main/java/org/apache/calcite/adapter/enumerable/FetchOffsetRoundingPolicy.java

@@ -0,0 +1,29 @@
+/*
+ * 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. */
+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. */
+  BigDecimal round(BigDecimal value);
+}

core/src/main/java/org/apache/calcite/prepare/CalcitePrepareImpl.java

@@ -21,7 +21,9 @@
 import org.apache.calcite.adapter.enumerable.EnumerableConvention;
 import org.apache.calcite.adapter.enumerable.EnumerableInterpretable;
 import org.apache.calcite.adapter.enumerable.EnumerableRel;
+import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor;
 import org.apache.calcite.adapter.enumerable.EnumerableRules;
+import org.apache.calcite.adapter.enumerable.FetchOffsetRoundingPolicy;
 import org.apache.calcite.adapter.enumerable.RexToLixTranslator;
 import org.apache.calcite.adapter.java.JavaTypeFactory;
 import org.apache.calcite.avatica.AvaticaParameter;
@@ -1196,6 +1198,13 @@ protected SqlValidator createSqlValidator(CatalogReader catalogReader,
           CatalogReader.THREAD_LOCAL.set(catalogReader);
           final SqlConformance conformance = context.config().conformance();
           internalParameters.put("_conformance", conformance);
+          final FetchOffsetRoundingPolicy fetchOffsetRoundingPolicy =
+              planner.getContext().unwrap(FetchOffsetRoundingPolicy.class);
+          if (fetchOffsetRoundingPolicy != null) {
+            internalParameters.put(
+                EnumerableRelImplementor.FETCH_OFFSET_ROUNDING_POLICY,
+                fetchOffsetRoundingPolicy);
+          }
           bindable =
               EnumerableInterpretable.toBindable(internalParameters,
                   context.spark(), enumerable,

core/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java

@@ -1743,11 +1743,11 @@ SqlValidatorNamespace getNamespaceOrThrow(SqlIdentifier id,
   private void handleOffsetFetch(@Nullable SqlNode offset, @Nullable SqlNode fetch) {
     if (offset instanceof SqlDynamicParam) {
       setValidatedNodeType(offset,
-          typeFactory.createSqlType(SqlTypeName.INTEGER));
+          typeFactory.createSqlType(SqlTypeName.DECIMAL));
     }
     if (fetch instanceof SqlDynamicParam) {
       setValidatedNodeType(fetch,
-          typeFactory.createSqlType(SqlTypeName.INTEGER));
+          typeFactory.createSqlType(SqlTypeName.DECIMAL));
     }
   }
 

core/src/main/java/org/apache/calcite/util/BuiltInMethod.java

@@ -21,6 +21,7 @@
 import org.apache.calcite.adapter.enumerable.BasicAggregateLambdaFactory;
 import org.apache.calcite.adapter.enumerable.BasicLazyAccumulator;
 import org.apache.calcite.adapter.enumerable.EnumUtils;
+import org.apache.calcite.adapter.enumerable.FetchOffsetRoundingPolicy;
 import org.apache.calcite.adapter.enumerable.LazyAggregateLambdaFactory;
 import org.apache.calcite.adapter.enumerable.MatchUtils;
 import org.apache.calcite.adapter.enumerable.SourceSorter;
@@ -296,7 +297,7 @@ public enum BuiltInMethod {
   ORDER_BY(ExtendedEnumerable.class, "orderBy", Function1.class,
       Comparator.class),
   ORDER_BY_WITH_FETCH_AND_OFFSET(EnumerableDefaults.class, "orderBy", Enumerable.class,
-      Function1.class, Comparator.class, int.class, int.class),
+      Function1.class, Comparator.class, BigDecimal.class, BigDecimal.class),
   UNION(ExtendedEnumerable.class, "union", Enumerable.class),
   CONCAT(ExtendedEnumerable.class, "concat", Enumerable.class),
   REPEAT_UNION(EnumerableDefaults.class, "repeatUnion", Enumerable.class,
@@ -308,7 +309,11 @@ public enum BuiltInMethod {
   INTERSECT(ExtendedEnumerable.class, "intersect", Enumerable.class, boolean.class),
   EXCEPT(ExtendedEnumerable.class, "except", Enumerable.class, boolean.class),
   SKIP(ExtendedEnumerable.class, "skip", int.class),
+  SKIP_BIG_DECIMAL(EnumerableDefaults.class, "skip", Enumerable.class,
+      BigDecimal.class),
   TAKE(ExtendedEnumerable.class, "take", int.class),
+  TAKE_BIG_DECIMAL(EnumerableDefaults.class, "take", Enumerable.class,
+      BigDecimal.class),
   SINGLETON_ENUMERABLE(Linq4j.class, "singletonEnumerable", Object.class),
   EMPTY_ENUMERABLE(Linq4j.class, "emptyEnumerable"),
   NULLS_COMPARATOR(Functions.class, "nullsComparator", boolean.class,
@@ -390,6 +395,8 @@ public enum BuiltInMethod {
       Calendar.class),
   TIME_ZONE_GET_OFFSET(TimeZone.class, "getOffset", long.class),
   LONG_VALUE(Number.class, "longValue"),
+  NUMBER_TO_BIG_DECIMAL_LIMIT(EnumUtils.class, "numberToBigDecimalLimit",
+      Object.class, String.class, FetchOffsetRoundingPolicy.class),
   STRING_TO_UPPER(String.class, "toUpperCase"),
   COMPARATOR_COMPARE(Comparator.class, "compare", Object.class, Object.class),
   COLLECTIONS_REVERSE_ORDER(Collections.class, "reverseOrder"),

core/src/test/java/org/apache/calcite/adapter/enumerable/CodeGeneratorTest.java

@@ -16,7 +16,9 @@
  */
 package org.apache.calcite.adapter.enumerable;
 
+import org.apache.calcite.DataContexts;
 import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.linq4j.Enumerable;
 import org.apache.calcite.linq4j.tree.ClassDeclaration;
 import org.apache.calcite.linq4j.tree.Expressions;
 import org.apache.calcite.plan.ConventionTraitDef;
@@ -26,19 +28,32 @@
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelRoot;
 import org.apache.calcite.rel.rules.CoreRules;
+import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
 import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.runtime.Bindable;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.sql.test.SqlTestFactory;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.validate.SqlValidator;
 import org.apache.calcite.sql2rel.SqlToRelConverter;
 
+import com.google.common.collect.ImmutableList;
+
 import org.junit.jupiter.api.Test;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 
 /**
@@ -97,4 +112,46 @@ public class CodeGeneratorTest {
     assertFalse(javaCode.contains("case_when_value3"));
     assertFalse(javaCode.contains("case_when_value4"));
   }
+
+  @Test public void testFetchOffsetRoundingPolicy() {
+    final JavaTypeFactoryImpl typeFactory =
+        new JavaTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+    final RexBuilder rexBuilder = new RexBuilder(typeFactory);
+    final VolcanoPlanner planner = new VolcanoPlanner();
+    planner.addRelTraitDef(ConventionTraitDef.INSTANCE);
+    final RelOptCluster cluster = RelOptCluster.create(planner, rexBuilder);
+    final RelDataType rowType =
+        typeFactory.builder().add("i", SqlTypeName.INTEGER).build();
+    final RelDataType integerType = rowType.getFieldList().get(0).getType();
+    final ImmutableList<ImmutableList<RexLiteral>> tuples =
+        ImmutableList.of(
+            ImmutableList.of(
+                rexBuilder.makeExactLiteral(BigDecimal.ONE,
+                integerType)),
+            ImmutableList.of(
+                rexBuilder.makeExactLiteral(BigDecimal.valueOf(2),
+                integerType)),
+            ImmutableList.of(
+                rexBuilder.makeExactLiteral(BigDecimal.valueOf(3),
+                integerType)));
+    final EnumerableValues values =
+        EnumerableValues.create(cluster, rowType, tuples);
+    final RelDataType decimalType =
+        typeFactory.createSqlType(SqlTypeName.DECIMAL, 2, 1);
+    final EnumerableLimit limit =
+        EnumerableLimit.create(values, null,
+            rexBuilder.makeExactLiteral(new BigDecimal("1.5"), decimalType));
+
+    final Map<String, Object> parameters = new HashMap<>();
+    parameters.put(EnumerableRelImplementor.FETCH_OFFSET_ROUNDING_POLICY,
+        (FetchOffsetRoundingPolicy) value ->
+            value.setScale(0, RoundingMode.DOWN));
+    final Bindable bindable =
+        EnumerableInterpretable.toBindable(parameters, null, limit,
+            EnumerableRel.Prefer.ARRAY);
+
+    final Enumerable<?> enumerable = bindable.bind(DataContexts.of(parameters));
+    final List<?> result = enumerable.toList();
+    assertThat(result, hasSize(1));
+  }
 }

core/src/test/java/org/apache/calcite/adapter/enumerable/EnumUtilsTest.java

@@ -29,6 +29,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.util.Arrays;
 
 import static org.hamcrest.CoreMatchers.is;
@@ -171,6 +172,17 @@ public final class EnumUtilsTest {
     assertThat(Expressions.toString(e2), is("(String) (Object) null"));
   }
 
+  @Test void testNumberToBigDecimalLimitPreservesFractionalNumber() {
+    final FetchOffsetRoundingPolicy ceiling =
+        value -> value.setScale(0, RoundingMode.CEILING);
+    assertThat(EnumUtils.numberToBigDecimalLimit(1.5D, "FETCH"),
+        is(new BigDecimal("1.5")));
+    assertThat(EnumUtils.numberToBigDecimalLimit(1.5F, "OFFSET"),
+        is(new BigDecimal("1.5")));
+    assertThat(EnumUtils.numberToBigDecimalLimit(1.5D, "FETCH", ceiling),
+        is(BigDecimal.valueOf(2)));
+  }
+
   @Test void testMethodCallExpression() {
     // test for Object.class method parameter type
     final ConstantExpression arg0 = Expressions.constant(1, int.class);