From 6c3d5ce1835e6fe1287b60ac66138841454df5d8 Mon Sep 17 00:00:00 2001 From: Daniel Bauer Date: Fri, 5 Dec 2025 11:08:35 +0100 Subject: [PATCH 1/2] Extend CudfVector with packed table ownership. --- velox/experimental/cudf/vector/CudfVector.cpp | 41 +++++++++++++++++++ velox/experimental/cudf/vector/CudfVector.h | 37 ++++++++++++++--- 2 files changed, 73 insertions(+), 5 deletions(-) diff --git a/velox/experimental/cudf/vector/CudfVector.cpp b/velox/experimental/cudf/vector/CudfVector.cpp index 93e1c22bd67..8753838ce91 100644 --- a/velox/experimental/cudf/vector/CudfVector.cpp +++ b/velox/experimental/cudf/vector/CudfVector.cpp @@ -21,6 +21,7 @@ #include "velox/vector/TypeAliases.h" #include +#include #include namespace facebook::velox::cudf_velox { @@ -110,10 +111,50 @@ CudfVector::CudfVector( std::vector(), std::nullopt), table_{std::move(table)}, + packedTable_{nullptr}, stream_{stream} { auto [bytes, tableOut] = getTableSize(std::move(table_)); flatSize_ = bytes; table_ = std::move(tableOut); + tabView_ = table_->view(); +} + +CudfVector::CudfVector( + velox::memory::MemoryPool* pool, + TypePtr type, + vector_size_t size, + std::unique_ptr&& packedTable, + rmm::cuda_stream_view stream) + : RowVector( + pool, + std::move(type), + BufferPtr(nullptr), + size, + std::vector(), + std::nullopt), + table_{nullptr}, + packedTable_{std::move(packedTable)}, + tabView_{packedTable_->table}, + stream_{stream} { + // For packed table, flatSize is the size of the GPU data buffer + flatSize_ = packedTable_->data.gpu_data->size(); +} + +std::unique_ptr CudfVector::release() { + flatSize_ = 0; + if (table_) { + // Constructed from owned table - just move it out + return std::move(table_); + } + // Constructed from packed_table - materialize a table from the view. + // This copies the data since the view references the packed buffer. + auto mr = cudf::get_current_device_resource_ref(); + auto materializedTable = + std::make_unique(tabView_, stream_, mr); + stream_.synchronize(); + // Clear the packed table since we've materialized + packedTable_.reset(); + return materializedTable; } uint64_t CudfVector::estimateFlatSize() const { diff --git a/velox/experimental/cudf/vector/CudfVector.h b/velox/experimental/cudf/vector/CudfVector.h index bb27bd4a889..0a172ba07b1 100644 --- a/velox/experimental/cudf/vector/CudfVector.h +++ b/velox/experimental/cudf/vector/CudfVector.h @@ -19,6 +19,7 @@ #include "velox/vector/ComplexVector.h" #include "velox/vector/TypeAliases.h" +#include #include #include @@ -29,8 +30,13 @@ namespace facebook::velox::cudf_velox { // Vector class which holds GPU data from cuDF. +// Can be constructed either from an owned cudf::table or from packed_table. +// When constructed from packed_table, the data remains packed and tabView_ +// references the table view inside packed_table without copying the underlying +// GPU data. class CudfVector : public RowVector { public: + /// Constructs a CudfVector from an owned cudf::table. CudfVector( velox::memory::MemoryPool* pool, TypePtr type, @@ -38,23 +44,44 @@ class CudfVector : public RowVector { std::unique_ptr&& table, rmm::cuda_stream_view stream); + /// Constructs a CudfVector from packed_table. + /// The packed data is retained and tabView_ references the table view inside + /// packed_table. This avoids copying the underlying GPU data. + CudfVector( + velox::memory::MemoryPool* pool, + TypePtr type, + vector_size_t size, + std::unique_ptr&& packedTable, + rmm::cuda_stream_view stream); + rmm::cuda_stream_view stream() const { return stream_; } cudf::table_view getTableView() const { - return table_->view(); + return tabView_; } - std::unique_ptr&& release() { - flatSize_ = 0; - return std::move(table_); - } + /// Releases ownership of the underlying table. + /// If constructed from packed_table, materializes a table from the view + /// first (which copies the data). + std::unique_ptr release(); uint64_t estimateFlatSize() const override; private: + // Owned table - set when constructed from cudf::table, or materialized + // on release() when constructed from packed_table. std::unique_ptr table_; + + // Packed table - set when constructed from packed_table. + // Keeps the packed data alive while tabView_ references it. + std::unique_ptr packedTable_; + + // Table view - always valid, points to either table_->view() or + // packedTable_->table. + cudf::table_view tabView_; + rmm::cuda_stream_view stream_; uint64_t flatSize_; }; From 7774ca686148ba12bd39dc83fdb683f4fa5f630b Mon Sep 17 00:00:00 2001 From: Daniel Bauer Date: Thu, 18 Dec 2025 16:24:59 +0100 Subject: [PATCH 2/2] Avoid expensive CudfVector::release() by using getTableView() When CudfVector is constructed from a packed_table, calling release() materializes a new table from the view, which copies all data. This change refactors callers to use getTableView() instead, keeping the CudfVector alive during operations to avoid unnecessary copies. Changes: - CudfConversion: Use getTableView() in passthrough mode - CudfHashAggregation: Change doGroupByAggregation, doGlobalAggregation, and getDistinctKeys to accept table_view instead of unique_ptr - CudfHashJoin: Change all join functions (innerJoin, leftJoin, rightJoin, leftSemiFilterJoin, rightSemiFilterJoin, antiJoin) to accept table_view --- .../experimental/cudf/exec/CudfConversion.cpp | 11 ++-- .../cudf/exec/CudfHashAggregation.cpp | 38 ++++++++------ .../cudf/exec/CudfHashAggregation.h | 6 +-- velox/experimental/cudf/exec/CudfHashJoin.cpp | 52 ++++++++++--------- velox/experimental/cudf/exec/CudfHashJoin.h | 22 ++++---- velox/experimental/cudf/vector/CudfVector.cpp | 1 - 6 files changed, 71 insertions(+), 59 deletions(-) diff --git a/velox/experimental/cudf/exec/CudfConversion.cpp b/velox/experimental/cudf/exec/CudfConversion.cpp index 86f192e3d2b..805189e0e9a 100644 --- a/velox/experimental/cudf/exec/CudfConversion.cpp +++ b/velox/experimental/cudf/exec/CudfConversion.cpp @@ -213,19 +213,22 @@ RowVectorPtr CudfToVelox::getOutput() { // batch size if (isPassthroughMode() || (inputs_.size() == 1 && inputs_.front()->size() <= targetBatchSize)) { - std::unique_ptr tbl = inputs_.front()->release(); + // Move the CudfVector out to keep it alive while we use the view. + // This avoids expensive materialization when constructed from packed_table. + auto cudfVector = std::move(inputs_.front()); inputs_.pop_front(); - VELOX_CHECK_NOT_NULL(tbl); - if (tbl->num_rows() == 0) { + auto tableView = cudfVector->getTableView(); + if (tableView.num_rows() == 0) { finished_ = noMoreInput_ && inputs_.empty(); return nullptr; } RowVectorPtr output = - with_arrow::toVeloxColumn(tbl->view(), pool(), "", stream); + with_arrow::toVeloxColumn(tableView, pool(), "", stream); stream.synchronize(); finished_ = noMoreInput_ && inputs_.empty(); output->setType(outputType_); + // cudfVector goes out of scope here, freeing the GPU memory return output; } diff --git a/velox/experimental/cudf/exec/CudfHashAggregation.cpp b/velox/experimental/cudf/exec/CudfHashAggregation.cpp index ff8cbd20c0f..5bd7856f43f 100644 --- a/velox/experimental/cudf/exec/CudfHashAggregation.cpp +++ b/velox/experimental/cudf/exec/CudfHashAggregation.cpp @@ -683,8 +683,10 @@ void CudfHashAggregation::computeIntermediateGroupbyPartial(CudfVectorPtr tbl) { // intermediate groupby results. auto inputTableStream = tbl->stream(); + // Use getTableView() to avoid expensive materialization for packed_table. + // tbl stays alive during this function call, keeping the view valid. auto groupbyOnInput = doGroupByAggregation( - tbl->release(), + tbl->getTableView(), groupingKeyInputChannels_, aggregators_, inputTableStream); @@ -708,8 +710,9 @@ void CudfHashAggregation::computeIntermediateGroupbyPartial(CudfVectorPtr tbl) { cudf::concatenate(tablesToConcat, partialOutputStream); // Now we have to groupby again but this time with intermediate aggregators. + // Keep concatenatedTable alive while we use its view. auto compactedOutput = doGroupByAggregation( - std::move(concatenatedTable), + concatenatedTable->view(), groupingKeyOutputChannels_, intermediateAggregators_, partialOutputStream); @@ -746,16 +749,18 @@ void CudfHashAggregation::computeIntermediateDistinctPartial( cudf::concatenate(tablesToConcat, partialOutputStream); // Do a distinct on the concatenated results. + // Keep concatenatedTable alive while we use its view. auto distinctOutput = getDistinctKeys( - std::move(concatenatedTable), + concatenatedTable->view(), groupingKeyOutputChannels_, inputTableStream); partialOutput_ = distinctOutput; } else { // First time processing, just store the result of the input batch's - // distinct. + // distinct. Use getTableView() to avoid expensive materialization for + // packed_table. tbl stays alive during this function call. partialOutput_ = getDistinctKeys( - tbl->release(), groupingKeyInputChannels_, inputTableStream); + tbl->getTableView(), groupingKeyInputChannels_, inputTableStream); } } @@ -785,11 +790,11 @@ void CudfHashAggregation::addInput(RowVectorPtr input) { } CudfVectorPtr CudfHashAggregation::doGroupByAggregation( - std::unique_ptr tbl, + cudf::table_view tableView, std::vector const& groupByKeys, std::vector>& aggregators, rmm::cuda_stream_view stream) { - auto groupbyKeyView = tbl->select(groupByKeys.begin(), groupByKeys.end()); + auto groupbyKeyView = tableView.select(groupByKeys.begin(), groupByKeys.end()); size_t const numGroupingKeys = groupbyKeyView.num_columns(); @@ -802,7 +807,7 @@ CudfVectorPtr CudfHashAggregation::doGroupByAggregation( std::vector requests; for (auto& aggregator : aggregators) { - aggregator->addGroupbyRequest(tbl->view(), requests); + aggregator->addGroupbyRequest(tableView, requests); } auto [groupKeys, results] = groupByOwner.aggregate(requests, stream); @@ -836,14 +841,13 @@ CudfVectorPtr CudfHashAggregation::doGroupByAggregation( } CudfVectorPtr CudfHashAggregation::doGlobalAggregation( - std::unique_ptr tbl, + cudf::table_view tableView, rmm::cuda_stream_view stream) { std::vector> resultColumns; resultColumns.reserve(aggregators_.size()); for (auto i = 0; i < aggregators_.size(); i++) { resultColumns.push_back( - aggregators_[i]->doReduce( - tbl->view(), outputType_->childAt(i), stream)); + aggregators_[i]->doReduce(tableView, outputType_->childAt(i), stream)); } return std::make_shared( @@ -855,11 +859,11 @@ CudfVectorPtr CudfHashAggregation::doGlobalAggregation( } CudfVectorPtr CudfHashAggregation::getDistinctKeys( - std::unique_ptr tbl, + cudf::table_view tableView, std::vector const& groupByKeys, rmm::cuda_stream_view stream) { auto result = cudf::distinct( - tbl->view().select(groupByKeys.begin(), groupByKeys.end()), + tableView.select(groupByKeys.begin(), groupByKeys.end()), {groupingKeyOutputChannels_.begin(), groupingKeyOutputChannels_.end()}, cudf::duplicate_keep_option::KEEP_FIRST, cudf::null_equality::EQUAL, @@ -946,13 +950,15 @@ RowVectorPtr CudfHashAggregation::getOutput() { VELOX_CHECK_NOT_NULL(tbl); + // Use tbl->view() instead of moving the table. + // tbl stays alive until the end of this function, keeping the view valid. if (isDistinct_) { - return getDistinctKeys(std::move(tbl), groupingKeyInputChannels_, stream); + return getDistinctKeys(tbl->view(), groupingKeyInputChannels_, stream); } else if (isGlobal_) { - return doGlobalAggregation(std::move(tbl), stream); + return doGlobalAggregation(tbl->view(), stream); } else { return doGroupByAggregation( - std::move(tbl), groupingKeyInputChannels_, aggregators_, stream); + tbl->view(), groupingKeyInputChannels_, aggregators_, stream); } } diff --git a/velox/experimental/cudf/exec/CudfHashAggregation.h b/velox/experimental/cudf/exec/CudfHashAggregation.h index dfd1fc37874..b92bd63bdb4 100644 --- a/velox/experimental/cudf/exec/CudfHashAggregation.h +++ b/velox/experimental/cudf/exec/CudfHashAggregation.h @@ -104,15 +104,15 @@ class CudfHashAggregation : public exec::Operator, public NvtxHelper { std::vector& groupingKeyOutputChannels) const; CudfVectorPtr doGroupByAggregation( - std::unique_ptr tbl, + cudf::table_view tableView, std::vector const& groupByKeys, std::vector>& aggregators, rmm::cuda_stream_view stream); CudfVectorPtr doGlobalAggregation( - std::unique_ptr tbl, + cudf::table_view tableView, rmm::cuda_stream_view stream); CudfVectorPtr getDistinctKeys( - std::unique_ptr tbl, + cudf::table_view tableView, std::vector const& groupByKeys, rmm::cuda_stream_view stream); diff --git a/velox/experimental/cudf/exec/CudfHashJoin.cpp b/velox/experimental/cudf/exec/CudfHashJoin.cpp index e889b4ea4f9..134d822a7d5 100644 --- a/velox/experimental/cudf/exec/CudfHashJoin.cpp +++ b/velox/experimental/cudf/exec/CudfHashJoin.cpp @@ -654,11 +654,10 @@ std::unique_ptr CudfHashJoinProbe::filteredOutput( } std::vector> CudfHashJoinProbe::innerJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream) { std::vector> cudfOutputs; - auto leftTableView = leftTable->view(); auto& rightTables = hashObject_.value().first; auto& hbs = hashObject_.value().second; for (auto i = 0; i < rightTables.size(); i++) { @@ -719,11 +718,10 @@ std::vector> CudfHashJoinProbe::innerJoin( } std::vector> CudfHashJoinProbe::leftJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream) { std::vector> cudfOutputs; - auto leftTableView = leftTable->view(); auto& rightTables = hashObject_.value().first; auto& hbs = hashObject_.value().second; for (auto i = 0; i < rightTables.size(); i++) { @@ -866,11 +864,10 @@ std::vector> CudfHashJoinProbe::leftJoin( } std::vector> CudfHashJoinProbe::rightJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream) { std::vector> cudfOutputs; - auto leftTableView = leftTable->view(); auto& rightTables = hashObject_.value().first; auto& hbs = hashObject_.value().second; @@ -982,11 +979,10 @@ std::vector> CudfHashJoinProbe::rightJoin( } std::vector> CudfHashJoinProbe::leftSemiFilterJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream) { std::vector> cudfOutputs; - auto leftTableView = leftTable->view(); auto& rightTables = hashObject_.value().first; for (auto i = 0; i < rightTables.size(); i++) { @@ -1029,11 +1025,10 @@ std::vector> CudfHashJoinProbe::leftSemiFilterJoin( std::vector> CudfHashJoinProbe::rightSemiFilterJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream) { std::vector> cudfOutputs; - auto leftTableView = leftTable->view(); auto& rightTables = hashObject_.value().first; auto rightTableView = rightTables[0]->view(); @@ -1077,7 +1072,7 @@ CudfHashJoinProbe::rightSemiFilterJoin( } std::vector> CudfHashJoinProbe::antiJoin( - std::unique_ptr&& leftTable, + cudf::table_view leftTableViewParam, rmm::cuda_stream_view stream) { std::vector> cudfOutputs; auto& rightTables = hashObject_.value().first; @@ -1088,21 +1083,28 @@ std::vector> CudfHashJoinProbe::antiJoin( "Multiple right tables not yet supported for antiJoin"); auto rightTableView = rightTables[0]->view(); + + // For the special case where we need to drop nulls, we create a local table. + // Otherwise, we use the input view directly. + std::unique_ptr modifiedLeftTable; + cudf::table_view leftTableView = leftTableViewParam; + // Special case for null-aware anti join where // build table is not empty, no nulls, and probe table has nulls if (joinNode_->isNullAware() and !joinNode_->filter()) { auto const leftTableHasNulls = - cudf::has_nulls(leftTable->view().select(leftKeyIndices_)); + cudf::has_nulls(leftTableViewParam.select(leftKeyIndices_)); auto const rightTableHasNulls = cudf::has_nulls(rightTableView.select(rightKeyIndices_)); if (rightTables[0]->num_rows() > 0 and !rightTableHasNulls and leftTableHasNulls) { - // drop nulls on probe table - leftTable = cudf::drop_nulls(leftTable->view(), leftKeyIndices_, stream); + // drop nulls on probe table - creates a new table + modifiedLeftTable = + cudf::drop_nulls(leftTableViewParam, leftKeyIndices_, stream); + leftTableView = modifiedLeftTable->view(); } } - auto leftTableView = leftTable->view(); std::unique_ptr> leftJoinIndices; if (joinNode_->filter()) { leftJoinIndices = cudf::mixed_left_anti_join( @@ -1225,11 +1227,13 @@ RowVectorPtr CudfHashJoinProbe::getOutput() { auto cudfInput = std::dynamic_pointer_cast(input_); VELOX_CHECK_NOT_NULL(cudfInput); auto stream = cudfInput->stream(); - auto leftTable = cudfInput->release(); // probe table + // Use getTableView() to avoid expensive materialization for packed_table. + // cudfInput stays alive during this function, keeping the view valid. + auto leftTableView = cudfInput->getTableView(); if (CudfConfig::getInstance().debugEnabled) { - VLOG(1) << "Probe table number of columns: " << leftTable->num_columns() + VLOG(1) << "Probe table number of columns: " << leftTableView.num_columns() << std::endl; - VLOG(1) << "Probe table number of rows: " << leftTable->num_rows() + VLOG(1) << "Probe table number of rows: " << leftTableView.num_rows() << std::endl; } @@ -1252,22 +1256,22 @@ RowVectorPtr CudfHashJoinProbe::getOutput() { std::vector> cudfOutputs; switch (joinNode_->joinType()) { case core::JoinType::kInner: - cudfOutputs = innerJoin(leftTable, stream); + cudfOutputs = innerJoin(leftTableView, stream); break; case core::JoinType::kLeft: - cudfOutputs = leftJoin(leftTable, stream); + cudfOutputs = leftJoin(leftTableView, stream); break; case core::JoinType::kRight: - cudfOutputs = rightJoin(leftTable, stream); + cudfOutputs = rightJoin(leftTableView, stream); break; case core::JoinType::kLeftSemiFilter: - cudfOutputs = leftSemiFilterJoin(leftTable, stream); + cudfOutputs = leftSemiFilterJoin(leftTableView, stream); break; case core::JoinType::kRightSemiFilter: - cudfOutputs = rightSemiFilterJoin(leftTable, stream); + cudfOutputs = rightSemiFilterJoin(leftTableView, stream); break; case core::JoinType::kAnti: - cudfOutputs = antiJoin(std::move(leftTable), stream); + cudfOutputs = antiJoin(leftTableView, stream); break; default: VELOX_FAIL("Unsupported join type: ", joinNode_->joinType()); diff --git a/velox/experimental/cudf/exec/CudfHashJoin.h b/velox/experimental/cudf/exec/CudfHashJoin.h index bce949ac15d..6214bdd4343 100644 --- a/velox/experimental/cudf/exec/CudfHashJoin.h +++ b/velox/experimental/cudf/exec/CudfHashJoin.h @@ -219,54 +219,54 @@ class CudfHashJoinProbe : public exec::Operator, public NvtxHelper { * @return Vector of result tables (multiple if build data was batched) */ std::vector> innerJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream); /** * @brief Performs left join between probe table and all build tables. - * @param leftTable Probe-side table to join + * @param leftTableView Probe-side table view to join * @param stream CUDA stream for operations * @return Vector of result tables (multiple if build data was batched) */ std::vector> leftJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream); /** * @brief Performs right join between probe table and all build tables. - * @param leftTable Probe-side table to join + * @param leftTableView Probe-side table view to join * @param stream CUDA stream for operations * @return Vector of result tables (multiple if build data was batched) */ std::vector> rightJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream); /** * @brief Performs left semi filter join between probe table and all build * tables. - * @param leftTable Probe-side table to join + * @param leftTableView Probe-side table view to join * @param stream CUDA stream for operations * @return Vector of result tables (multiple if build data was batched) */ std::vector> leftSemiFilterJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream); /** * @brief Performs right semi filter join between probe table and all build * tables. - * @param leftTable Probe-side table to join + * @param leftTableView Probe-side table view to join * @param stream CUDA stream for operations * @return Vector of result tables (multiple if build data was batched) */ std::vector> rightSemiFilterJoin( - std::unique_ptr const& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream); /** * @brief Performs anti join between probe table and all build tables. - * @param leftTable Probe-side table to join (moved) + * @param leftTableView Probe-side table view to join * @param stream CUDA stream for operations * @return Vector of result tables (multiple if build data was batched) */ std::vector> antiJoin( - std::unique_ptr&& leftTable, + cudf::table_view leftTableView, rmm::cuda_stream_view stream); /** * @brief Constructs join output table without applying filter conditions. diff --git a/velox/experimental/cudf/vector/CudfVector.cpp b/velox/experimental/cudf/vector/CudfVector.cpp index 8753838ce91..5982b9816bf 100644 --- a/velox/experimental/cudf/vector/CudfVector.cpp +++ b/velox/experimental/cudf/vector/CudfVector.cpp @@ -21,7 +21,6 @@ #include "velox/vector/TypeAliases.h" #include -#include #include namespace facebook::velox::cudf_velox {