Add packed_table to CudfVector

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<cudf::table> 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;
   }
 

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<cudf::table> tbl,
+    cudf::table_view tableView,
     std::vector<column_index_t> const& groupByKeys,
     std::vector<std::unique_ptr<Aggregator>>& 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<cudf::groupby::aggregation_request> 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<cudf::table> tbl,
+    cudf::table_view tableView,
     rmm::cuda_stream_view stream) {
   std::vector<std::unique_ptr<cudf::column>> 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<cudf_velox::CudfVector>(
@@ -855,11 +859,11 @@ CudfVectorPtr CudfHashAggregation::doGlobalAggregation(
 }
 
 CudfVectorPtr CudfHashAggregation::getDistinctKeys(
-    std::unique_ptr<cudf::table> tbl,
+    cudf::table_view tableView,
     std::vector<column_index_t> 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);
   }
 }
 

velox/experimental/cudf/exec/CudfHashAggregation.h

@@ -104,15 +104,15 @@ class CudfHashAggregation : public exec::Operator, public NvtxHelper {
       std::vector<column_index_t>& groupingKeyOutputChannels) const;
 
   CudfVectorPtr doGroupByAggregation(
-      std::unique_ptr<cudf::table> tbl,
+      cudf::table_view tableView,
       std::vector<column_index_t> const& groupByKeys,
       std::vector<std::unique_ptr<Aggregator>>& aggregators,
       rmm::cuda_stream_view stream);
   CudfVectorPtr doGlobalAggregation(
-      std::unique_ptr<cudf::table> tbl,
+      cudf::table_view tableView,
       rmm::cuda_stream_view stream);
   CudfVectorPtr getDistinctKeys(
-      std::unique_ptr<cudf::table> tbl,
+      cudf::table_view tableView,
       std::vector<column_index_t> const& groupByKeys,
       rmm::cuda_stream_view stream);
 

velox/experimental/cudf/exec/CudfHashJoin.cpp

@@ -654,11 +654,10 @@ std::unique_ptr<cudf::table> CudfHashJoinProbe::filteredOutput(
 }
 
 std::vector<std::unique_ptr<cudf::table>> CudfHashJoinProbe::innerJoin(
-    std::unique_ptr<cudf::table> const& leftTable,
+    cudf::table_view leftTableView,
     rmm::cuda_stream_view stream) {
   std::vector<std::unique_ptr<cudf::table>> 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<std::unique_ptr<cudf::table>> CudfHashJoinProbe::innerJoin(
 }
 
 std::vector<std::unique_ptr<cudf::table>> CudfHashJoinProbe::leftJoin(
-    std::unique_ptr<cudf::table> const& leftTable,
+    cudf::table_view leftTableView,
     rmm::cuda_stream_view stream) {
   std::vector<std::unique_ptr<cudf::table>> 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<std::unique_ptr<cudf::table>> CudfHashJoinProbe::leftJoin(
 }
 
 std::vector<std::unique_ptr<cudf::table>> CudfHashJoinProbe::rightJoin(
-    std::unique_ptr<cudf::table> const& leftTable,
+    cudf::table_view leftTableView,
     rmm::cuda_stream_view stream) {
   std::vector<std::unique_ptr<cudf::table>> cudfOutputs;
 
-  auto leftTableView = leftTable->view();
   auto& rightTables = hashObject_.value().first;
   auto& hbs = hashObject_.value().second;
 
@@ -982,11 +979,10 @@ std::vector<std::unique_ptr<cudf::table>> CudfHashJoinProbe::rightJoin(
 }
 
 std::vector<std::unique_ptr<cudf::table>> CudfHashJoinProbe::leftSemiFilterJoin(
-    std::unique_ptr<cudf::table> const& leftTable,
+    cudf::table_view leftTableView,
     rmm::cuda_stream_view stream) {
   std::vector<std::unique_ptr<cudf::table>> cudfOutputs;
 
-  auto leftTableView = leftTable->view();
   auto& rightTables = hashObject_.value().first;
 
   for (auto i = 0; i < rightTables.size(); i++) {
@@ -1029,11 +1025,10 @@ std::vector<std::unique_ptr<cudf::table>> CudfHashJoinProbe::leftSemiFilterJoin(
 
 std::vector<std::unique_ptr<cudf::table>>
 CudfHashJoinProbe::rightSemiFilterJoin(
-    std::unique_ptr<cudf::table> const& leftTable,
+    cudf::table_view leftTableView,
     rmm::cuda_stream_view stream) {
   std::vector<std::unique_ptr<cudf::table>> cudfOutputs;
 
-  auto leftTableView = leftTable->view();
   auto& rightTables = hashObject_.value().first;
   auto rightTableView = rightTables[0]->view();
 
@@ -1077,7 +1072,7 @@ CudfHashJoinProbe::rightSemiFilterJoin(
 }
 
 std::vector<std::unique_ptr<cudf::table>> CudfHashJoinProbe::antiJoin(
-    std::unique_ptr<cudf::table>&& leftTable,
+    cudf::table_view leftTableViewParam,
     rmm::cuda_stream_view stream) {
   std::vector<std::unique_ptr<cudf::table>> cudfOutputs;
   auto& rightTables = hashObject_.value().first;
@@ -1088,21 +1083,28 @@ std::vector<std::unique_ptr<cudf::table>> 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<cudf::table> 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<rmm::device_uvector<cudf::size_type>> leftJoinIndices;
   if (joinNode_->filter()) {
     leftJoinIndices = cudf::mixed_left_anti_join(
@@ -1225,11 +1227,13 @@ RowVectorPtr CudfHashJoinProbe::getOutput() {
   auto cudfInput = std::dynamic_pointer_cast<CudfVector>(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<std::unique_ptr<cudf::table>> 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());

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<std::unique_ptr<cudf::table>> innerJoin(
-      std::unique_ptr<cudf::table> 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<std::unique_ptr<cudf::table>> leftJoin(
-      std::unique_ptr<cudf::table> 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<std::unique_ptr<cudf::table>> rightJoin(
-      std::unique_ptr<cudf::table> 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<std::unique_ptr<cudf::table>> leftSemiFilterJoin(
-      std::unique_ptr<cudf::table> 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<std::unique_ptr<cudf::table>> rightSemiFilterJoin(
-      std::unique_ptr<cudf::table> 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<std::unique_ptr<cudf::table>> antiJoin(
-      std::unique_ptr<cudf::table>&& leftTable,
+      cudf::table_view leftTableView,
       rmm::cuda_stream_view stream);
   /**
    * @brief Constructs join output table without applying filter conditions.