Register custom row constructor function for Gluten. (oap-project#404)

cpp/velox/CMakeLists.txt

@@ -156,6 +156,7 @@ set(VELOX_SRCS
     compute/VeloxPlanConverter.cc
     compute/ArrowTypeUtils.cc
     compute/VeloxToRowConverter.cc
+    compute/RowConstructor.cc
     compute/DwrfDatasource.cc
     compute/bridge.cc
     memory/velox_memory_pool.cc

cpp/velox/compute/RegistrationAllFunctions.cc

@@ -0,0 +1,39 @@
+/*
+ * 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.
+ */
+#include "RegistrationAllFunctions.h"
+#include "RowConstructor.cc"
+#include "velox/functions/sparksql/Register.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::exec;
+
+namespace velox::compute {
+
+void registerCustomFunctions() {
+  exec::registerVectorFunction(
+      "row_constructor",
+      std::vector<std::shared_ptr<exec::FunctionSignature>>{},
+      std::make_unique<RowConstructor>());
+}
+
+void registerAllFunctions() {
+  functions::prestosql::registerAllScalarFunctions();
+  functions::sparksql::registerFunctions("");
+  registerCustomFunctions();
+}
+
+} // namespace velox::compute

cpp/velox/compute/RegistrationAllFunctions.h

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+namespace velox::compute {
+
+void registerAllFunctions();
+
+} // namespace velox::compute

cpp/velox/compute/RowConstructor.cc

@@ -0,0 +1,69 @@
+/*
+ * 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.
+ */
+#include "velox/expression/VectorFunction.h"
+#include "velox/type/Type.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::exec;
+
+namespace velox::compute {
+
+namespace {
+class RowConstructor : public exec::VectorFunction {
+  void apply(
+      const SelectivityVector& rows,
+      std::vector<VectorPtr>& args,
+      const TypePtr& outputType,
+      exec::EvalCtx& context,
+      VectorPtr& result) const override {
+    auto argsCopy = args;
+
+    BufferPtr nulls = AlignedBuffer::allocate<char>(
+        bits::nbytes(rows.size()), context.pool(), 1);
+    auto* nullsPtr = nulls->asMutable<uint64_t>();
+    auto cntNull = 0;
+    rows.applyToSelected([&](vector_size_t i) {
+      bits::clearNull(nullsPtr, i);
+      if (!bits::isBitNull(nullsPtr, i)) {
+        for (size_t c = 0; c < argsCopy.size(); c++) {
+          auto arg = argsCopy[c].get();
+          if (arg->mayHaveNulls() && arg->isNullAt(i)) {
+            bits::setNull(nullsPtr, i, true);
+            cntNull++;
+            break;
+          }
+        }
+      }
+    });
+
+    RowVectorPtr localResult = std::make_shared<RowVector>(
+        context.pool(),
+        outputType,
+        nulls,
+        rows.size(),
+        std::move(argsCopy),
+        cntNull /*nullCount*/);
+    context.moveOrCopyResult(localResult, rows, result);
+  }
+
+  bool isDefaultNullBehavior() const override {
+    return false;
+  }
+};
+
+} // namespace
+} // namespace velox::compute

cpp/velox/compute/VeloxPlanConverter.cc

@@ -24,6 +24,7 @@
 #include <string>
 
 #include "ArrowTypeUtils.h"
+#include "RegistrationAllFunctions.cc"
 #include "arrow/c/Bridge.h"
 #include "arrow/c/bridge.h"
 #include "bridge.h"
@@ -33,7 +34,6 @@
 #include "velox/functions/prestosql/aggregates/AverageAggregate.h"
 #include "velox/functions/prestosql/aggregates/CountAggregate.h"
 #include "velox/functions/prestosql/aggregates/MinMaxAggregates.h"
-#include "velox/functions/sparksql/Register.h"
 
 using namespace facebook::velox;
 using namespace facebook::velox::exec;
@@ -95,9 +95,8 @@ void VeloxInitializer::Init() {
   parquet::registerParquetReaderFactory(ParquetReaderType::NATIVE);
   // parquet::registerParquetReaderFactory(ParquetReaderType::DUCKDB);
   dwrf::registerDwrfReaderFactory();
-  // Register Velox functions.
-  functions::sparksql::registerFunctions("");
-  functions::prestosql::registerAllScalarFunctions();
+  // Register Velox functions
+  registerAllFunctions();
   aggregate::registerSumAggregate<aggregate::SumAggregate>("sum");
   aggregate::registerAverageAggregate("avg");
   aggregate::registerCountAggregate("count");

cpp/velox/jni/jni_wrapper.cc

@@ -21,6 +21,7 @@
 
 #include <jni/jni_common.h>
 #include "compute/DwrfDatasource.h"
+#include "compute/RegistrationAllFunctions.h"
 #include "compute/VeloxPlanConverter.h"
 #include "jni/jni_errors.h"
 #include "memory/velox_memory_pool.h"
@@ -127,6 +128,7 @@ Java_io_glutenproject_vectorized_ExpressionEvaluatorJniWrapper_nativeDoValidate(
   std::unique_ptr<core::ExecCtx> execCtx_ =
       std::make_unique<core::ExecCtx>(pool.get(), queryCtx_.get());
 
+  velox::compute::registerAllFunctions();
   auto planValidator = std::make_shared<
       facebook::velox::substrait::SubstraitToVeloxPlanValidator>(
       pool.get(), execCtx_.get());