[GLUTEN-4652][VL] Fix min_by/max_by result mismatch when RDD partition num > 1

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxAggregateFunctionsSuite.scala

@@ -194,18 +194,12 @@ abstract class VeloxAggregateFunctionsSuite extends VeloxWholeStageTransformerSu
   }
 
   test("min_by/max_by") {
-    withTempPath {
-      path =>
-        Seq((5: Integer, 6: Integer), (null: Integer, 11: Integer), (null: Integer, 5: Integer))
-          .toDF("a", "b")
-          .write
-          .parquet(path.getCanonicalPath)
-        spark.read
-          .parquet(path.getCanonicalPath)
-          .createOrReplaceTempView("test")
-        runQueryAndCompare("select min_by(a, b), max_by(a, b) from test") {
-          checkGlutenOperatorMatch[HashAggregateExecTransformer]
-        }
+    withSQLConf(("spark.sql.leafNodeDefaultParallelism", "2")) {
+      runQueryAndCompare(
+        "select min_by(a, b), max_by(a, b) from " +
+          "values (5, 6), (null, 11), (null, 5) test(a, b)") {
+        checkGlutenOperatorMatch[HashAggregateExecTransformer]
+      }
     }
   }
 

cpp/velox/operators/functions/RegistrationAllFunctions.cc

@@ -15,11 +15,10 @@
  * limitations under the License.
  */
 #include "operators/functions/RegistrationAllFunctions.h"
+
 #include "operators/functions/Arithmetic.h"
-#include "operators/functions/RowConstructorWithAllNull.h"
 #include "operators/functions/RowConstructorWithNull.h"
 #include "operators/functions/RowFunctionWithNull.h"
-
 #include "velox/expression/SpecialFormRegistry.h"
 #include "velox/expression/VectorFunction.h"
 #include "velox/functions/lib/RegistrationHelpers.h"
@@ -45,29 +44,32 @@ void registerFunctionOverwrite() {
   velox::registerFunction<RoundFunction, double, double, int32_t>({"round"});
   velox::registerFunction<RoundFunction, float, float, int32_t>({"round"});
 
+  auto kRowConstructorWithNull = RowConstructorWithNullCallToSpecialForm::kRowConstructorWithNull;
   velox::exec::registerVectorFunction(
-      "row_constructor_with_null",
+      kRowConstructorWithNull,
       std::vector<std::shared_ptr<velox::exec::FunctionSignature>>{},
       std::make_unique<RowFunctionWithNull</*allNull=*/false>>(),
       RowFunctionWithNull</*allNull=*/false>::metadata());
   velox::exec::registerFunctionCallToSpecialForm(
-      RowConstructorWithNullCallToSpecialForm::kRowConstructorWithNull,
-      std::make_unique<RowConstructorWithNullCallToSpecialForm>());
+      kRowConstructorWithNull, std::make_unique<RowConstructorWithNullCallToSpecialForm>(kRowConstructorWithNull));
+
+  auto kRowConstructorWithAllNull = RowConstructorWithNullCallToSpecialForm::kRowConstructorWithAllNull;
   velox::exec::registerVectorFunction(
-      "row_constructor_with_all_null",
+      kRowConstructorWithAllNull,
       std::vector<std::shared_ptr<velox::exec::FunctionSignature>>{},
       std::make_unique<RowFunctionWithNull</*allNull=*/true>>(),
       RowFunctionWithNull</*allNull=*/true>::metadata());
   velox::exec::registerFunctionCallToSpecialForm(
-      RowConstructorWithAllNullCallToSpecialForm::kRowConstructorWithAllNull,
-      std::make_unique<RowConstructorWithAllNullCallToSpecialForm>());
+      kRowConstructorWithAllNull,
+      std::make_unique<RowConstructorWithNullCallToSpecialForm>(kRowConstructorWithAllNull));
   velox::functions::sparksql::registerBitwiseFunctions("spark_");
 }
 } // namespace
 
 void registerAllFunctions() {
   // The registration order matters. Spark sql functions are registered after
-  // presto sql functions to overwrite the registration for same named functions.
+  // presto sql functions to overwrite the registration for same named
+  // functions.
   velox::functions::prestosql::registerAllScalarFunctions();
   velox::functions::sparksql::registerFunctions("");
   velox::aggregate::prestosql::registerAllAggregateFunctions(
@@ -76,7 +78,8 @@ void registerAllFunctions() {
       "", true /*registerCompanionFunctions*/, true /*overwrite*/);
   velox::window::prestosql::registerAllWindowFunctions();
   velox::functions::window::sparksql::registerWindowFunctions("");
-  // Using function overwrite to handle function names mismatch between Spark and Velox.
+  // Using function overwrite to handle function names mismatch between Spark
+  // and Velox.
   registerFunctionOverwrite();
 }
 

cpp/velox/operators/functions/RowConstructorWithNull.cc

@@ -32,11 +32,11 @@ facebook::velox::TypePtr RowConstructorWithNullCallToSpecialForm::resolveType(
 }
 
 facebook::velox::exec::ExprPtr RowConstructorWithNullCallToSpecialForm::constructSpecialForm(
-    const std::string& name,
     const facebook::velox::TypePtr& type,
     std::vector<facebook::velox::exec::ExprPtr>&& compiledChildren,
     bool trackCpuUsage,
     const facebook::velox::core::QueryConfig& config) {
+  auto name = this->rowFunctionName;
   auto [function, metadata] = facebook::velox::exec::vectorFunctionFactories().withRLock(
       [&config, &name](auto& functionMap) -> std::pair<
                                               std::shared_ptr<facebook::velox::exec::VectorFunction>,
@@ -52,12 +52,4 @@ facebook::velox::exec::ExprPtr RowConstructorWithNullCallToSpecialForm::construc
   return std::make_shared<facebook::velox::exec::Expr>(
       type, std::move(compiledChildren), function, metadata, name, trackCpuUsage);
 }
-
-facebook::velox::exec::ExprPtr RowConstructorWithNullCallToSpecialForm::constructSpecialForm(
-    const facebook::velox::TypePtr& type,
-    std::vector<facebook::velox::exec::ExprPtr>&& compiledChildren,
-    bool trackCpuUsage,
-    const facebook::velox::core::QueryConfig& config) {
-  return constructSpecialForm(kRowConstructorWithNull, type, std::move(compiledChildren), trackCpuUsage, config);
-}
 } // namespace gluten

cpp/velox/operators/functions/RowConstructorWithNull.h

@@ -23,6 +23,10 @@
 namespace gluten {
 class RowConstructorWithNullCallToSpecialForm : public facebook::velox::exec::FunctionCallToSpecialForm {
  public:
+  RowConstructorWithNullCallToSpecialForm(const std::string& rowFunctionName) {
+    this->rowFunctionName = rowFunctionName;
+  }
+
   facebook::velox::TypePtr resolveType(const std::vector<facebook::velox::TypePtr>& argTypes) override;
 
   facebook::velox::exec::ExprPtr constructSpecialForm(
@@ -32,6 +36,7 @@ class RowConstructorWithNullCallToSpecialForm : public facebook::velox::exec::Fu
       const facebook::velox::core::QueryConfig& config) override;
 
   static constexpr const char* kRowConstructorWithNull = "row_constructor_with_null";
+  static constexpr const char* kRowConstructorWithAllNull = "row_constructor_with_all_null";
 
  protected:
   facebook::velox::exec::ExprPtr constructSpecialForm(
@@ -40,5 +45,8 @@ class RowConstructorWithNullCallToSpecialForm : public facebook::velox::exec::Fu
       std::vector<facebook::velox::exec::ExprPtr>&& compiledChildren,
       bool trackCpuUsage,
       const facebook::velox::core::QueryConfig& config);
+
+ private:
+  std::string rowFunctionName;
 };
 } // namespace gluten