[CALCITE-7349] Upgrade the types of FETCH and OFFSET in SORT to BigDecimal

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

@@ -92,8 +92,8 @@ public EnumerableMergeUnionRule(Config config) {
         inputFetch = sort.fetch;
       } else if (sort.fetch instanceof RexLiteral && sort.offset instanceof RexLiteral) {
         inputFetch =
-            call.builder().literal(RexLiteral.longValue(sort.fetch)
-                + RexLiteral.longValue(sort.offset));
+            call.builder().literal(RexLiteral.bigDecimalValue(sort.fetch)
+                .add(RexLiteral.bigDecimalValue(sort.offset)));
       }
     }
 

core/src/main/java/org/apache/calcite/rel/metadata/RelMdUtil.java

@@ -1048,9 +1048,14 @@ private static boolean alreadySmaller(RelMetadataQuery mq, RelNode input,
       // Cannot be determined
       return false;
     }
-    final long offsetVal = offset == null ? 0 : RexLiteral.longValue(offset);
-    final long limit = RexLiteral.longValue(fetch);
-    return (double) offsetVal + (double) limit >= rowCount;
+    final BigDecimal offsetVal = offset == null
+        ? BigDecimal.ZERO
+        : RexLiteral.bigDecimalValue(offset);
+    final BigDecimal limit = RexLiteral.bigDecimalValue(fetch);
+    if (!Double.isFinite(rowCount)) {
+      return false;
+    }
+    return offsetVal.add(limit).compareTo(BigDecimal.valueOf(rowCount)) >= 0;
   }
 
   /**

core/src/main/java/org/apache/calcite/rel/rules/PruneEmptyRules.java

@@ -47,6 +47,7 @@
 
 import org.immutables.value.Value;
 
+import java.math.BigDecimal;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Predicate;
@@ -500,9 +501,8 @@ public interface SortFetchZeroRuleConfig extends PruneEmptyRule.Config {
           Sort sort = call.rel(0);
           return sort.fetch != null
               && !(sort.fetch instanceof RexDynamicParam)
-              && RexLiteral.longValue(sort.fetch) == 0;
+              && RexLiteral.bigDecimalValue(sort.fetch).equals(BigDecimal.ZERO);
         }
-
       };
     }
   }

core/src/main/java/org/apache/calcite/rel/rules/SortJoinTransposeRule.java

@@ -202,10 +202,12 @@ public SortJoinTransposeRule(Class<? extends Sort> sortClass,
     if (sort.fetch == null) {
       return null;
     }
-    final long outerFetch = RexLiteral.longValue(sort.fetch);
-    final long outerOffset = sort.offset != null ? RexLiteral.longValue(sort.offset) : 0;
-    final long totalFetch = outerOffset + outerFetch;
-    return rexBuilder.makeExactLiteral(BigDecimal.valueOf(totalFetch));
+    final BigDecimal outerFetch = RexLiteral.bigDecimalValue(sort.fetch);
+    final BigDecimal outerOffset = sort.offset != null
+        ? RexLiteral.bigDecimalValue(sort.offset)
+        : BigDecimal.ZERO;
+    final BigDecimal totalFetch = outerOffset.add(outerFetch);
+    return rexBuilder.makeExactLiteral(totalFetch);
   }
 
   /** Rule configuration. */

core/src/main/java/org/apache/calcite/rel/rules/SortRemoveRedundantRule.java

@@ -125,18 +125,17 @@ protected SortRemoveRedundantRule(final SortRemoveRedundantRule.Config config) {
     // then we could remove the redundant sort.
     if (inputMaxRowCount != null
         && Double.isFinite(inputMaxRowCount)
-        && new BigDecimal(inputMaxRowCount).compareTo(rowCountThreshold.get()) <= 0) {
+        && BigDecimal.valueOf(inputMaxRowCount).compareTo(rowCountThreshold.get()) <= 0) {
       call.transformTo(sort.getInput());
     }
   }
 
   private static Optional<BigDecimal> getRowCountThreshold(Sort sort) {
     if (RelOptUtil.isLimit(sort)) {
       assert sort.fetch != null;
-      final BigDecimal fetch = ((RexLiteral) sort.fetch).getValueAs(BigDecimal.class);
+      final BigDecimal fetch = RexLiteral.bigDecimalValue(sort.fetch);
 
       // We don't need to deal with fetch is 0.
-      assert fetch != null;
       if (fetch.equals(BigDecimal.ZERO)) {
         return Optional.empty();
       }

core/src/main/java/org/apache/calcite/rex/RexLiteral.java

@@ -1290,6 +1290,12 @@ public static long longValue(RexNode node) {
     return number.longValue();
   }
 
+  /** Returns the value of a literal, cast, or unary minus, as a BigDecimal;
+   * never null. */
+  public static BigDecimal bigDecimalValue(RexNode node) {
+    return (BigDecimal) numberValue(node);
+  }
+
   public static @Nullable String stringValue(RexNode node) {
     final Comparable value = findValue(node);
     return (value == null) ? null : ((NlsString) value).getValue();

core/src/main/java/org/apache/calcite/tools/RelBuilder.java

@@ -117,6 +117,7 @@
 import org.apache.calcite.util.ImmutableNullableList;
 import org.apache.calcite.util.Litmus;
 import org.apache.calcite.util.NlsString;
+import org.apache.calcite.util.NumberUtil;
 import org.apache.calcite.util.Optionality;
 import org.apache.calcite.util.Pair;
 import org.apache.calcite.util.Util;
@@ -494,8 +495,7 @@ public RexLiteral literal(@Nullable Object value) {
       return rexBuilder.makeApproxLiteral(
           ((Number) value).doubleValue(), getTypeFactory().createSqlType(SqlTypeName.DOUBLE));
     } else if (value instanceof Number) {
-      return rexBuilder.makeExactLiteral(
-          BigDecimal.valueOf(((Number) value).longValue()));
+      return rexBuilder.makeExactLiteral(NumberUtil.toBigDecimal((Number) value));
     } else if (value instanceof String) {
       return rexBuilder.makeLiteral((String) value);
     } else if (value instanceof Enum) {