[GLUTEN-8921][GLUTEN-8922][CH] Fix checkDecimalOverflowSparkOrNull and lead function (#8929)

Author: lwz9103

backends-clickhouse/src/test/scala/org/apache/gluten/execution/compatibility/GlutenClickhouseFunctionSuite.scala

@@ -442,4 +442,38 @@ class GlutenClickhouseFunctionSuite extends GlutenClickHouseTPCHAbstractSuite {
     }
   }
 
+  test("GLUTEN-8921: Type mismatch at checkDecimalOverflowSparkOrNull") {
+    compareResultsAgainstVanillaSpark(
+      """
+        |select l_shipdate, avg(l_quantity), count(0) over() COU,
+        |SUM(-1.1) over() SU, AVG(-2) over() AV,
+        |max(-1.1) over() MA, min(-3) over() MI
+        |from lineitem
+        |where l_shipdate <= date'1998-09-02'
+        |group by l_shipdate
+        |order by l_shipdate
+      """.stripMargin,
+      true,
+      { _ => }
+    )
+  }
+
+  test("GLUTEN-8922: Incorrect result in lead function with constant col") {
+    compareResultsAgainstVanillaSpark(
+      """
+        |select l_shipdate,
+        |FIRST_VALUE(-2) over() FI,
+        |LAST_VALUE(-2) over() LA,
+        |lag(-2) over(order by l_shipdate) lag0,
+        |lead(-2) over(order by l_shipdate) lead0
+        |from lineitem
+        |where l_shipdate <= date'1998-09-02'
+        |group by l_shipdate
+        |order by l_shipdate
+      """.stripMargin,
+      true,
+      { _ => }
+    )
+  }
+
 }

cpp-ch/local-engine/Functions/SparkFunctionCheckDecimalOverflow.cpp

@@ -100,14 +100,20 @@ class FunctionCheckDecimalOverflow : public IFunction
         UInt32 precision = extractArgument(arguments[1]);
         UInt32 scale = extractArgument(arguments[2]);
         auto return_type = createDecimal<DataTypeDecimal>(precision, scale);
-        if constexpr (exception_mode == CheckExceptionMode::Null)
-        {
-            if (!arguments[0].type->isNullable())
-                return std::make_shared<DataTypeNullable>(return_type);
-        }
+        if (isReturnTypeNullable(arguments[0]))
+            return std::make_shared<DataTypeNullable>(return_type);
         return return_type;
     }
 
+    bool isReturnTypeNullable(const ColumnWithTypeAndName & arg) const
+    {
+        if constexpr (exception_mode == CheckExceptionMode::Null)
+            return true;
+        if (arg.type->isNullable())
+            return true;
+        return false;
+    }
+
     ColumnPtr executeImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr &, size_t input_rows_count) const override
     {
         UInt32 to_precision = extractArgument(arguments[1]);
@@ -133,7 +139,10 @@ class FunctionCheckDecimalOverflow : public IFunction
                     auto from_scale = getDecimalScale(*src_col.type);
                     if (from_precision == to_precision && from_scale == to_scale)
                     {
-                        dst_col = src_col.column;
+                        if (isReturnTypeNullable(arguments[0]))
+                            dst_col = makeNullable(src_col.column);
+                        else
+                            dst_col = src_col.column;
                         return true;
                     }
                 }

cpp-ch/local-engine/Parser/aggregate_function_parser/LeadLagParser.cpp

@@ -33,15 +33,15 @@ LeadParser::parseFunctionArguments(const CommonFunctionInfo & func_info, DB::Act
     /// The 3rd arg is default value
     /// when it is set to null, the 1st arg must be nullable
     const auto & arg2 = func_info.arguments[2].value();
-    const auto * arg0_col = actions_dag.getInputs()[arg0.selection().direct_reference().struct_field().field()];
+    const auto * arg0_col = parseExpression(actions_dag, arg0);
     auto arg0_col_name = arg0_col->result_name;
     auto arg0_col_type= arg0_col->result_type;
     const DB::ActionsDAG::Node * node = nullptr;
     if (arg2.has_literal() && arg2.literal().has_null() && !arg0_col_type->isNullable())
     {
         node = ActionsDAGUtil::convertNodeType(
             actions_dag,
-            &actions_dag.findInOutputs(arg0_col_name),
+            arg0_col,
             DB::makeNullable(arg0_col_type),
             arg0_col_name);
         actions_dag.addOrReplaceInOutputs(*node);
@@ -76,15 +76,15 @@ LagParser::parseFunctionArguments(const CommonFunctionInfo & func_info, DB::Acti
     /// The 3rd arg is default value
     /// when it is set to null, the 1st arg must be nullable
     const auto & arg2 = func_info.arguments[2].value();
-    const auto * arg0_col = actions_dag.getInputs()[arg0.selection().direct_reference().struct_field().field()];
+    const auto * arg0_col = parseExpression(actions_dag, arg0);
     auto arg0_col_name = arg0_col->result_name;
     auto arg0_col_type = arg0_col->result_type;
     const DB::ActionsDAG::Node * node = nullptr;
     if (arg2.has_literal() && arg2.literal().has_null() && !arg0_col->result_type->isNullable())
     {
         node = ActionsDAGUtil::convertNodeType(
             actions_dag,
-            &actions_dag.findInOutputs(arg0_col_name),
+            arg0_col,
             makeNullable(arg0_col_type),
             arg0_col_name);
         actions_dag.addOrReplaceInOutputs(*node);