[fix][core] Resolve the compatibility issue between double and decimal types in the lead/tag function

Author: guojn1

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

@@ -1927,14 +1927,17 @@ protected LeadLagImplementor(boolean isLead) {
       result.exitBlock();
       BlockStatement thenBranch = thenBlock.toBlock();
 
-      //Expression defaultValue = rexArgs.size() == 3
-      //    ? currentRowTranslator.translate(rexArgs.get(2), res.type)
-      //    : getDefaultValue(res.type);
-
       result.currentBlock().add(Expressions.declare(0, res, null));
-      result.currentBlock().add(
-          Expressions.ifThenElse(rowInRange, thenBranch,
-              Expressions.statement(Expressions.assign(res, defaultValue))));
+      if (("java.lang.Double".equalsIgnoreCase(res.getType().getTypeName()) || "double".equalsIgnoreCase(res.getType().getTypeName()))
+          && "java.math.BigDecimal".equalsIgnoreCase(defaultValue.getType().getTypeName())) {
+        result.currentBlock().add(
+                Expressions.ifThenElse(rowInRange, thenBranch,
+                        Expressions.statement(Expressions.assign(res, Expressions.call(defaultValue, BuiltInMethod.BIG_DECIMAL_DOUBLEVALUE.method)))));
+      } else {
+        result.currentBlock().add(
+                Expressions.ifThenElse(rowInRange, thenBranch,
+                        Expressions.statement(Expressions.assign(res, defaultValue))));
+      }
       return res;
     }
   }

core/src/main/java/org/apache/calcite/sql/fun/SqlLeadLagAggFunction.java

@@ -16,31 +16,45 @@
  */
 package org.apache.calcite.sql.fun;
 
+import com.google.common.collect.ImmutableList;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.SqlAggFunction;
 import org.apache.calcite.sql.SqlFunctionCategory;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperatorBinding;
 import org.apache.calcite.sql.type.OperandTypes;
 import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SameOperandTypeChecker;
 import org.apache.calcite.sql.type.SqlReturnTypeInference;
 import org.apache.calcite.sql.type.SqlSingleOperandTypeChecker;
+import org.apache.calcite.sql.type.SqlTypeFamily;
 import org.apache.calcite.sql.type.SqlTypeTransform;
 import org.apache.calcite.sql.type.SqlTypeTransforms;
 import org.apache.calcite.util.Optionality;
 
 import com.google.common.base.Preconditions;
 
+import java.util.List;
+
 /**
  * <code>LEAD</code> and <code>LAG</code> aggregate functions
  * return the value of given expression evaluated at given offset.
  */
 public class SqlLeadLagAggFunction extends SqlAggFunction {
   private static final SqlSingleOperandTypeChecker OPERAND_TYPES =
-      OperandTypes.ANY
-          .or(OperandTypes.ANY_NUMERIC)
-          .or(OperandTypes.ANY_NUMERIC_ANY
-              .and(OperandTypes.same(3, 0, 2)));
+          OperandTypes.or(
+                  OperandTypes.ANY,
+                  OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.NUMERIC),
+                  OperandTypes.and(
+                          OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.NUMERIC,
+                                  SqlTypeFamily.ANY),
+                          // Arguments 1 and 3 must have same type
+                          new SameOperandTypeChecker(3) {
+                            @Override protected List<Integer>
+                            getOperandList(int operandCount) {
+                              return ImmutableList.of(0, 2);
+                            }
+                          }));
 
   private static final SqlReturnTypeInference RETURN_TYPE =
       ReturnTypes.ARG0.andThen(SqlLeadLagAggFunction::transformType);

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

@@ -678,6 +678,7 @@ public enum BuiltInMethod {
       long.class),
   BIG_DECIMAL_ADD(BigDecimal.class, "add", BigDecimal.class),
   BIG_DECIMAL_NEGATE(BigDecimal.class, "negate"),
+  BIG_DECIMAL_DOUBLEVALUE(BigDecimal.class, "doubleValue"),
   COMPARE_TO(Comparable.class, "compareTo", Object.class);
 
   @SuppressWarnings("ImmutableEnumChecker")