This is an automated email from the ASF dual-hosted git repository.
jinchengchenghh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 381c806ec2 [VL] Extend cuDF functions in Gluten with
row_constructor_with_null (#13104)
381c806ec2 is described below
commit 381c806ec2d65ed3262b67407f2e604902f11e49
Author: Reema <[email protected]>
AuthorDate: Wed Oct 28 15:15:31 2026 +0300
[VL] Extend cuDF functions in Gluten with row_constructor_with_null (#13104)
---
.../gluten/execution/CudfAggregateSuite.scala | 73 ++++++++++
cpp/velox/CMakeLists.txt | 1 +
cpp/velox/compute/VeloxBackend.cc | 2 +
cpp/velox/cudf/CudfFunctions.cc | 101 ++++++++++++++
cpp/velox/cudf/CudfFunctions.h | 25 ++++
cpp/velox/tests/CMakeLists.txt | 1 +
cpp/velox/tests/CudfFunctionsTest.cc | 151 +++++++++++++++++++++
7 files changed, 354 insertions(+)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/CudfAggregateSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/CudfAggregateSuite.scala
new file mode 100644
index 0000000000..fe937d3b20
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/CudfAggregateSuite.scala
@@ -0,0 +1,73 @@
+/*
+ * 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.
+ */
+package org.apache.gluten.execution
+
+import org.apache.gluten.tags.CudfTest
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.execution.adaptive.ColumnarAQEShuffleReadExec
+
+/** GPU regression coverage for Gluten-specific aggregate intermediate rows. */
+@CudfTest
+class CudfAggregateSuite extends VeloxWholeStageTransformerSuite {
+
+ override protected val resourcePath: String = "/tpch-data-parquet"
+ override protected val fileFormat: String = "parquet"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.gluten.sql.columnar.cudf", "true")
+ .set("spark.gluten.sql.columnar.backend.velox.cudf.allowCpuFallback",
"false")
+ .set("spark.sql.shuffle.partitions", "1")
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", "4g")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ createTPCHNotNullTables()
+ }
+
+ test("cuDF supports row_constructor_with_null for average states") {
+ // Three nullable AVG states mirror the row-construction shape in TPC-H q1
+ // without introducing the separate decimal or legacy-cast paths.
+ val query =
+ """
+ |SELECT
+ | l_returnflag,
+ | avg(CASE WHEN l_returnflag = 'R' THEN CAST(1 AS DOUBLE) END) AS
avg_r,
+ | avg(CASE WHEN l_returnflag = 'A' THEN CAST(2 AS DOUBLE) END) AS
avg_a,
+ | avg(CASE WHEN l_returnflag = 'N' THEN CAST(3 AS DOUBLE) END) AS
avg_n
+ |FROM lineitem
+ |GROUP BY l_returnflag
+ |""".stripMargin
+
+ runQueryAndCompare(query) {
+ df =>
+ val plan = df.queryExecution.executedPlan
+ val aggregates = collect(plan) { case aggregate:
HashAggregateExecTransformer => aggregate }
+ assert(aggregates.nonEmpty, s"expected an offloaded hash aggregate,
got:\n$plan")
+
+ val readers =
+ getExecutedPlan(df).collect { case reader:
ColumnarAQEShuffleReadExec => reader }
+ assert(
+ readers.exists(_.executionMode == MockGPUStageMode),
+ s"expected a cuDF-tagged aggregate stage, got:\n$plan")
+ }
+ }
+}
diff --git a/cpp/velox/CMakeLists.txt b/cpp/velox/CMakeLists.txt
index fd5ff0a9d4..5f416eadcc 100644
--- a/cpp/velox/CMakeLists.txt
+++ b/cpp/velox/CMakeLists.txt
@@ -229,6 +229,7 @@ if(ENABLE_GPU)
APPEND
VELOX_SRCS
cudf/CudfPlanValidator.cc
+ cudf/CudfFunctions.cc
cudf/GpuLock.cc
operators/plannodes/CudfVectorStream.cc
shuffle/VeloxGpuAsyncShuffleReader.cc
diff --git a/cpp/velox/compute/VeloxBackend.cc
b/cpp/velox/compute/VeloxBackend.cc
index 7633216249..e9db41783c 100644
--- a/cpp/velox/compute/VeloxBackend.cc
+++ b/cpp/velox/compute/VeloxBackend.cc
@@ -35,6 +35,7 @@
#include "utils/qat/QatCodec.h"
#endif
#ifdef GLUTEN_ENABLE_GPU
+#include "cudf/CudfFunctions.h"
#include "cudf/GpuLock.h"
#include "operators/plannodes/CudfVectorStream.h"
#include "velox/experimental/cudf/CudfConfig.h"
@@ -227,6 +228,7 @@ void VeloxBackend::init(
velox::exec::Operator::registerOperator(std::make_unique<CudfVectorStreamOperatorTranslator>());
velox::cudf_velox::registerSparkFunctions("");
velox::cudf_velox::registerSparkAggregateFunctions("");
+ registerCudfFunctions();
} else {
LOG(WARNING) << "No Cuda device found. Skip Cudf initialization.";
}
diff --git a/cpp/velox/cudf/CudfFunctions.cc b/cpp/velox/cudf/CudfFunctions.cc
new file mode 100644
index 0000000000..b0e8d838c4
--- /dev/null
+++ b/cpp/velox/cudf/CudfFunctions.cc
@@ -0,0 +1,101 @@
+/*
+ * 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 "cudf/CudfFunctions.h"
+
+#include "operators/functions/RowConstructorWithNull.h"
+#include "velox/common/base/Exceptions.h"
+#include "velox/experimental/cudf/expression/ExpressionEvaluator.h"
+
+#include <cudf/column/column_factories.hpp>
+#include <cudf/null_mask.hpp>
+#include <cudf/table/table_view.hpp>
+
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <utility>
+#include <variant>
+#include <vector>
+
+namespace gluten {
+namespace {
+
+using facebook::velox::core::TypedExprPtr;
+using facebook::velox::cudf_velox::ColumnOrView;
+using facebook::velox::cudf_velox::CudfFunction;
+
+class RowConstructorWithNullFunction final : public CudfFunction {
+ public:
+ explicit RowConstructorWithNullFunction(std::shared_ptr<CudfFunction>
rowConstructor)
+ : rowConstructor_(std::move(rowConstructor)) {
+ VELOX_CHECK_NOT_NULL(rowConstructor_);
+ }
+
+ ColumnOrView eval(std::vector<ColumnOrView>& inputColumns, cuda::stream_ref
stream, rmm::device_async_resource_ref mr)
+ const override {
+ auto result = rowConstructor_->eval(inputColumns, stream, mr);
+ auto* ownedResult = std::get_if<std::unique_ptr<cudf::column>>(&result);
+ VELOX_CHECK_NOT_NULL(ownedResult, "row_constructor must return an owning
cuDF column");
+
+ auto structColumn = std::move(*ownedResult);
+ VELOX_CHECK_NOT_NULL(structColumn);
+ VELOX_CHECK(structColumn->type().id() == cudf::type_id::STRUCT,
"row_constructor must return a cuDF STRUCT column");
+
+ const auto numRows = structColumn->size();
+ auto contents = structColumn->release();
+ std::vector<cudf::column_view> childViews;
+ childViews.reserve(contents.children.size());
+ for (const auto& child : contents.children) {
+ childViews.push_back(child->view());
+ }
+
+ auto [nullMask, nullCount] =
cudf::bitmask_and(cudf::table_view{childViews}, stream, mr);
+ // Rebuild through the struct factory instead of setting the parent mask in
+ // place. The factory superimposes parent nulls on every child, which is
+ // required when aggregate finalization reads a child column directly.
+ return cudf::make_structs_column(numRows, std::move(contents.children),
nullCount, std::move(nullMask), stream, mr);
+ }
+
+ private:
+ std::shared_ptr<CudfFunction> rowConstructor_;
+};
+
+bool canEvaluateRowConstructorWithNull(const TypedExprPtr& expr) {
+ return expr->type()->isRow() && !expr->inputs().empty() &&
+ std::any_of(
+ expr->inputs().begin(), expr->inputs().end(), [](const auto&
input) { return !input->isConstantKind(); });
+}
+
+} // namespace
+
+void registerCudfFunctions() {
+ facebook::velox::cudf_velox::registerCudfFunction(
+ RowConstructorWithNullCallToSpecialForm::kRowConstructorWithNull,
+ [](const std::string&,
+ const TypedExprPtr& expr,
+ facebook::velox::memory::MemoryPool* pool) ->
std::shared_ptr<CudfFunction> {
+ auto rowConstructor =
facebook::velox::cudf_velox::createCudfFunction("row_constructor", expr, pool);
+ VELOX_CHECK_NOT_NULL(rowConstructor, "cuDF row_constructor must be
registered first");
+ return
std::make_shared<RowConstructorWithNullFunction>(std::move(rowConstructor));
+ },
+ {},
+ false,
+ canEvaluateRowConstructorWithNull);
+}
+
+} // namespace gluten
diff --git a/cpp/velox/cudf/CudfFunctions.h b/cpp/velox/cudf/CudfFunctions.h
new file mode 100644
index 0000000000..dda05ae9fd
--- /dev/null
+++ b/cpp/velox/cudf/CudfFunctions.h
@@ -0,0 +1,25 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+namespace gluten {
+
+/// Registers Gluten-specific cuDF expression implementations.
+void registerCudfFunctions();
+
+} // namespace gluten
diff --git a/cpp/velox/tests/CMakeLists.txt b/cpp/velox/tests/CMakeLists.txt
index a5d7329830..82af2bacbc 100644
--- a/cpp/velox/tests/CMakeLists.txt
+++ b/cpp/velox/tests/CMakeLists.txt
@@ -106,6 +106,7 @@ if(ENABLE_GPU)
add_velox_gpu_test(velox_gpu_shuffle_writer_test SOURCES
VeloxGpuShuffleWriterTest.cc)
add_velox_gpu_test(velox_gpu_lock_test SOURCES GpuLockTest.cc)
+ add_velox_gpu_test(velox_gpu_cudf_functions_test SOURCES
CudfFunctionsTest.cc)
add_velox_gpu_test(velox_cudf_value_stream_test SOURCES
CudfValueStreamTest.cc)
endif()
diff --git a/cpp/velox/tests/CudfFunctionsTest.cc
b/cpp/velox/tests/CudfFunctionsTest.cc
new file mode 100644
index 0000000000..df62906c3b
--- /dev/null
+++ b/cpp/velox/tests/CudfFunctionsTest.cc
@@ -0,0 +1,151 @@
+/*
+ * 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 "cudf/CudfFunctions.h"
+#include "operators/functions/RowConstructorWithNull.h"
+#include "velox/experimental/cudf/exec/GpuResources.h"
+#include "velox/experimental/cudf/exec/ToCudf.h"
+#include "velox/experimental/cudf/exec/VeloxCudfInterop.h"
+#include "velox/experimental/cudf/expression/ExpressionEvaluator.h"
+#include "velox/vector/tests/utils/VectorTestBase.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+namespace gluten {
+namespace {
+
+using namespace facebook::velox;
+
+class CudfFunctionsTest : public ::testing::Test, public test::VectorTestBase {
+ protected:
+ static void SetUpTestCase() {
+
memory::MemoryManager::testingSetInstance(memory::MemoryManager::Options{});
+ cudf_velox::registerCudf();
+ registerCudfFunctions();
+ }
+
+ static void TearDownTestCase() {
+ cudf_velox::unregisterFunctions();
+ cudf_velox::unregisterCudf();
+ }
+
+ RowVectorPtr evaluateOnGpu(const core::TypedExprPtr& expr, const
RowVectorPtr& input) {
+ auto stream = cudf_velox::cudfGlobalStreamPool().get_stream();
+ auto mr = cudf_velox::get_output_mr();
+ auto cudfTable = cudf_velox::with_arrow::toCudfTable(input, pool_.get(),
stream, mr);
+ core::QueryConfig queryConfig{{}};
+ auto evaluator = cudf_velox::createCudfExpression(expr, input->rowType(),
pool_.get(), queryConfig);
+
+ auto ownedColumns = cudfTable->release();
+ std::vector<cudf::column_view> inputViews;
+ inputViews.reserve(ownedColumns.size());
+ for (const auto& column : ownedColumns) {
+ inputViews.push_back(column->view());
+ }
+
+ auto output = evaluator->eval(inputViews, stream, mr);
+ cudf::table_view outputTable{{cudf_velox::asView(output)}};
+ auto outputType = ROW({"result"}, {expr->type()});
+ auto result = cudf_velox::with_arrow::toVeloxColumn(
+ outputTable, pool_.get(), outputType, "", stream,
cudf::get_current_device_resource_ref());
+ stream.sync();
+ result->setType(outputType);
+ return result;
+ }
+
+ core::TypedExprPtr makeRowConstructorWithNull(const RowTypePtr& resultType) {
+ return std::make_shared<core::CallTypedExpr>(
+ resultType,
+ std::vector<core::TypedExprPtr>{
+ std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "c0"),
+ std::make_shared<core::FieldAccessTypedExpr>(VARCHAR(), "c1")},
+ RowConstructorWithNullCallToSpecialForm::kRowConstructorWithNull);
+ }
+};
+
+TEST_F(CudfFunctionsTest, rowConstructorWithNull) {
+ auto integers = makeNullableFlatVector<int32_t>({1, std::nullopt, 3,
std::nullopt});
+ auto strings = makeNullableFlatVector<std::string>({"a", "b", std::nullopt,
std::nullopt});
+ auto input = makeRowVector({"c0", "c1"}, {integers, strings});
+ auto resultType = ROW({"left", "right"}, {INTEGER(), VARCHAR()});
+ auto rowConstructor = makeRowConstructorWithNull(resultType);
+
+ ASSERT_TRUE(cudf_velox::canExprRunOnGpu(rowConstructor, nullptr,
pool_.get()));
+ auto output = evaluateOnGpu(rowConstructor, input);
+ auto result = output->childAt(0)->as<RowVector>();
+ ASSERT_NE(result, nullptr);
+ EXPECT_FALSE(result->isNullAt(0));
+ EXPECT_TRUE(result->isNullAt(1));
+ EXPECT_TRUE(result->isNullAt(2));
+ EXPECT_TRUE(result->isNullAt(3));
+
+ // cuDF structs require parent nulls to be superimposed on their children.
+ EXPECT_TRUE(result->childAt(0)->isNullAt(2));
+ EXPECT_TRUE(result->childAt(1)->isNullAt(1));
+
+ auto firstField = std::make_shared<core::DereferenceTypedExpr>(INTEGER(),
rowConstructor, 0);
+ auto dereferenceOutput = evaluateOnGpu(firstField, input);
+ auto actual = dereferenceOutput->childAt(0);
+ auto expected = makeNullableFlatVector<int32_t>({1, std::nullopt,
std::nullopt, std::nullopt});
+ test::assertEqualVectors(expected, actual);
+}
+
+TEST_F(CudfFunctionsTest, rowConstructorWithNullLiteral) {
+ auto integers = makeNullableFlatVector<int32_t>({1, std::nullopt, 3});
+ auto input = makeRowVector({"c0"}, {integers});
+ auto resultType = ROW({"left", "right"}, {INTEGER(), VARCHAR()});
+ auto rowConstructor = std::make_shared<core::CallTypedExpr>(
+ resultType,
+ std::vector<core::TypedExprPtr>{
+ std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "c0"),
+ std::make_shared<core::ConstantTypedExpr>(VARCHAR(),
variant::null(TypeKind::VARCHAR))},
+ RowConstructorWithNullCallToSpecialForm::kRowConstructorWithNull);
+
+ ASSERT_TRUE(cudf_velox::canExprRunOnGpu(rowConstructor, nullptr,
pool_.get()));
+ auto output = evaluateOnGpu(rowConstructor, input);
+ auto result = output->childAt(0)->as<RowVector>();
+ ASSERT_NE(result, nullptr);
+ for (vector_size_t row = 0; row < result->size(); ++row) {
+ EXPECT_TRUE(result->isNullAt(row));
+ EXPECT_TRUE(result->childAt(0)->isNullAt(row));
+ EXPECT_TRUE(result->childAt(1)->isNullAt(row));
+ }
+}
+
+TEST_F(CudfFunctionsTest, rejectsAllLiteralRow) {
+ auto resultType = ROW({"left", "right"}, {INTEGER(), VARCHAR()});
+ auto rowConstructor = std::make_shared<core::CallTypedExpr>(
+ resultType,
+ std::vector<core::TypedExprPtr>{
+ std::make_shared<core::ConstantTypedExpr>(INTEGER(),
variant(int32_t{1})),
+ std::make_shared<core::ConstantTypedExpr>(VARCHAR(), variant("x"))},
+ RowConstructorWithNullCallToSpecialForm::kRowConstructorWithNull);
+
+ EXPECT_FALSE(cudf_velox::canExprRunOnGpu(rowConstructor, nullptr,
pool_.get()));
+ EXPECT_EQ(
+ cudf_velox::createCudfFunction(
+ RowConstructorWithNullCallToSpecialForm::kRowConstructorWithNull,
rowConstructor, pool_.get()),
+ nullptr);
+}
+
+} // namespace
+} // namespace gluten
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
