[VL] Fix empty GPU broadcast joins and CudfVector host read crashes in the cuDF backend

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala

@@ -16,6 +16,7 @@
  */
 package org.apache.gluten.execution
 
+import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.iterator.Iterators
 
 import org.apache.spark.{broadcast, SparkContext}
@@ -38,7 +39,14 @@ case class VeloxBroadcastBuildSideRDD(
       case unsafe: UnsafeColumnarBuildSideRelation =>
         unsafe.isOffload
     }
-    val output = if (isBNL || !offload) {
+    // With cuDF enabled the hash join runs as CudfHashJoin, which builds its own GPU
+    // hash table from the build-side value stream and cannot consume the prebuilt CPU
+    // OpaqueHashTable. Feeding Iterator.empty here silently produces empty join
+    // results, so stream the broadcast batches instead (the same path shuffle joins
+    // use on GPU). Skipping the CPU cache build also keeps hybrid mode correct:
+    // VeloxBroadcastBuildSideCache.get finds no table, so the HashJoinNode carries no
+    // reusable table and a CPU-fallback join builds from this stream as usual.
+    val output = if (isBNL || !offload || GlutenConfig.get.enableColumnarCudf) {
       val relation = broadcasted.value.asReadOnlyCopy()
       Iterators
         .wrap(relation.deserialized)

cpp/velox/compute/VeloxRuntime.h

@@ -17,18 +17,18 @@
 
 #pragma once
 
+#include <folly/Executor.h>
+#include "IcebergNestedField.pb.h"
 #include "WholeStageResultIterator.h"
 #include "compute/Runtime.h"
 #include "compute/VeloxConnectorIds.h"
 #include "iceberg/IcebergWriter.h"
-#include <folly/Executor.h>
 #include "memory/VeloxMemoryManager.h"
 #include "operators/serializer/VeloxColumnarBatchSerializer.h"
 #include "operators/serializer/VeloxColumnarToRowConverter.h"
 #include "operators/writer/VeloxParquetDataSource.h"
 #include "shuffle/ShuffleReader.h"
 #include "shuffle/ShuffleWriter.h"
-#include "IcebergNestedField.pb.h"
 
 namespace gluten {
 

cpp/velox/jni/JniHashTable.cc

@@ -26,6 +26,7 @@
 #include "operators/hashjoin/HashTableSerializer.h"
 #include "substrait/algebra.pb.h"
 #include "substrait/type.pb.h"
+#include "utils/CudfVectorUtils.h"
 #include "velox/core/PlanNode.h"
 #include "velox/type/Type.h"
 
@@ -150,7 +151,8 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild(
       abandonHashBuildDedupMinPct);
 
   for (auto i = 0; i < batches.size(); i++) {
-    auto rowVector = VeloxColumnarBatch::from(memoryPool.get(), batches[i])->getRowVector();
+    auto rowVector = materializeVeloxRowVector(
+        VeloxColumnarBatch::from(memoryPool.get(), batches[i])->getRowVector(), memoryPool.get());
     hashTableBuilder->addInput(rowVector);
     if (hashTableBuilder->noMoreInput()) {
       break;

cpp/velox/jni/VeloxJniWrapper.cc

@@ -41,6 +41,7 @@
 #include "operators/hashjoin/HashTableBuilder.h"
 #include "shuffle/rss/RssPartitionWriter.h"
 #include "substrait/SubstraitToVeloxPlanValidator.h"
+#include "utils/CudfVectorUtils.h"
 #include "utils/ObjectStore.h"
 #include "utils/VeloxBatchResizer.h"
 #include "velox/common/base/BloomFilter.h"
@@ -702,7 +703,7 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_columnarbatch_VeloxColumnarBatchJ
   auto veloxBatch = std::dynamic_pointer_cast<VeloxColumnarBatch>(batch);
   VELOX_CHECK_NOT_NULL(veloxBatch, "Expected VeloxColumnarBatch but got a different type.");
 
-  auto rowVector = veloxBatch->getRowVector();
+  auto rowVector = materializeVeloxRowVector(veloxBatch->getRowVector(), veloxBatch->getRowVector()->pool());
   auto prunedVector = rowVector->slice(offset, limit);
 
   auto prunedRowVector = std::dynamic_pointer_cast<facebook::velox::RowVector>(prunedVector);

cpp/velox/memory/VeloxColumnarBatch.cc

@@ -16,6 +16,7 @@
  */
 #include "VeloxColumnarBatch.h"
 #include "compute/VeloxRuntime.h"
+#include "utils/CudfVectorUtils.h"
 #include "utils/Timer.h"
 #include "utils/VeloxArrowUtils.h"
 #include "velox/row/UnsafeRowFast.h"
@@ -46,6 +47,7 @@ RowVectorPtr makeRowVector(
 } // namespace
 
 void VeloxColumnarBatch::ensureFlattened() {
+  rowVector_ = materializeVeloxRowVector(rowVector_, rowVector_->pool());
   if (flattened_) {
     return;
   }
@@ -118,7 +120,7 @@ std::shared_ptr<VeloxColumnarBatch> VeloxColumnarBatch::compose(
       throw GlutenException("Mismatched row counts among the input batches during composing columnar batches");
     }
     auto vb = std::dynamic_pointer_cast<VeloxColumnarBatch>(batch);
-    auto rv = vb->getRowVector();
+    auto rv = materializeVeloxRowVector(vb->getRowVector(), pool);
     GLUTEN_CHECK(rv->nulls() == nullptr, "Vectors to compose contain null bits");
   }
 
@@ -128,7 +130,7 @@ std::shared_ptr<VeloxColumnarBatch> VeloxColumnarBatch::compose(
   std::vector<VectorPtr> children;
   for (const auto& batch : batches) {
     auto vb = std::dynamic_pointer_cast<VeloxColumnarBatch>(batch);
-    auto rv = vb->getRowVector();
+    auto rv = materializeVeloxRowVector(vb->getRowVector(), pool);
     for (const std::string& name : rv->type()->asRow().names()) {
       childNames.push_back(name);
     }
@@ -147,21 +149,24 @@ std::shared_ptr<VeloxColumnarBatch> VeloxColumnarBatch::select(
   std::vector<VectorPtr> childVectors;
   childNames.reserve(columnIndices.size());
   childVectors.reserve(columnIndices.size());
-  auto type = facebook::velox::asRowType(rowVector_->type());
+  auto inputRowVector = materializeVeloxRowVector(rowVector_, pool);
+  auto type = facebook::velox::asRowType(inputRowVector->type());
 
   for (uint32_t i = 0; i < columnIndices.size(); i++) {
     auto index = columnIndices[i];
-    auto child = rowVector_->childAt(index);
+    auto child = inputRowVector->childAt(index);
     childNames.push_back(type->nameOf(index));
     childVectors.push_back(child);
   }
 
-  auto rowVector = makeRowVector(pool, numRows(), std::move(childNames), rowVector_->nulls(), std::move(childVectors));
+  auto rowVector =
+      makeRowVector(pool, numRows(), std::move(childNames), inputRowVector->nulls(), std::move(childVectors));
   return std::make_shared<VeloxColumnarBatch>(rowVector);
 }
 
 std::vector<char> VeloxColumnarBatch::toUnsafeRow(int32_t rowId) const {
-  auto fast = std::make_unique<facebook::velox::row::UnsafeRowFast>(rowVector_);
+  auto rowVector = materializeVeloxRowVector(rowVector_, rowVector_->pool());
+  auto fast = std::make_unique<facebook::velox::row::UnsafeRowFast>(rowVector);
   auto size = fast->rowSize(rowId);
   std::vector<char> bytes(size);
   std::memset(bytes.data(), 0, bytes.size());

cpp/velox/operators/hashjoin/HashTableBuilder.cc

@@ -23,6 +23,8 @@
 
 #include "velox/exec/OperatorUtils.h"
 
+#include "utils/CudfVectorUtils.h"
+
 namespace gluten {
 namespace {
 facebook::velox::RowTypePtr hashJoinTableType(
@@ -255,6 +257,11 @@ void HashTableBuilder::setupTable() {
 }
 
 void HashTableBuilder::addInput(facebook::velox::RowVectorPtr input) {
+  // The build side (including broadcast build) may arrive as a GPU-resident
+  // CudfVector, which has no host-side children. Materialize it to a host
+  // RowVector before reading key channels below; otherwise childAt() below
+  // dereferences a null child and crashes.
+  input = materializeVeloxRowVector(input, pool_);
   activeRows_.resize(input->size());
   activeRows_.setAll();
 

cpp/velox/operators/plannodes/CudfVectorStream.cc

@@ -17,6 +17,7 @@
 
 #include "CudfVectorStream.h"
 #include "memory/VeloxColumnarBatch.h"
+#include "utils/CudfVectorUtils.h"
 #include "velox/exec/Driver.h"
 #include "velox/exec/Operator.h"
 #include "velox/exec/Task.h"
@@ -41,7 +42,6 @@ class SuspendedSection {
  private:
   facebook::velox::exec::Driver* const driver_;
 };
-
 } // namespace
 
 namespace gluten {
@@ -87,9 +87,9 @@ std::shared_ptr<ColumnarBatch> CudfVectorStreamBase::nextInternal() {
 facebook::velox::RowVectorPtr CudfVectorStreamBase::next() {
   auto cb = nextInternal();
   const std::shared_ptr<VeloxColumnarBatch>& vb = VeloxColumnarBatch::from(pool_, cb);
-  auto vp = vb->getRowVector();
+  auto vp = materializeVeloxRowVector(vb->getRowVector(), pool_);
   VELOX_DCHECK(vp != nullptr);
   return std::make_shared<facebook::velox::RowVector>(
       vp->pool(), outputType_, facebook::velox::BufferPtr(0), vp->size(), vp->children());
 }
-} // namespace gluten
+} // namespace gluten

cpp/velox/operators/plannodes/RowVectorStream.cc

@@ -17,6 +17,7 @@
 
 #include "RowVectorStream.h"
 #include "memory/VeloxColumnarBatch.h"
+#include "utils/CudfVectorUtils.h"
 #include "velox/exec/Driver.h"
 #include "velox/exec/Operator.h"
 #include "velox/exec/OperatorUtils.h"
@@ -43,7 +44,6 @@ class SuspendedSection {
  private:
   facebook::velox::exec::Driver* const driver_;
 };
-
 } // namespace
 
 namespace gluten {
@@ -100,7 +100,7 @@ std::shared_ptr<ColumnarBatch> RowVectorStream::nextInternal() {
 facebook::velox::RowVectorPtr RowVectorStream::next() {
   auto cb = nextInternal();
   const std::shared_ptr<VeloxColumnarBatch>& vb = VeloxColumnarBatch::from(pool_, cb);
-  auto vp = vb->getRowVector();
+  auto vp = materializeVeloxRowVector(vb->getRowVector(), pool_);
   VELOX_DCHECK(vp != nullptr);
   return std::make_shared<facebook::velox::RowVector>(
       vp->pool(), outputType_, facebook::velox::BufferPtr(0), vp->size(), vp->children());
@@ -278,4 +278,4 @@ void ValueStreamDataSource::applyFilterOnColumn(
   rows.updateBounds();
 }
 
-} // namespace gluten
+} // namespace gluten

cpp/velox/operators/serializer/VeloxColumnarBatchSerializer.cc

@@ -27,6 +27,7 @@
 
 #include "memory/ArrowMemory.h"
 #include "memory/VeloxColumnarBatch.h"
+#include "utils/CudfVectorUtils.h"
 #include "velox/common/compression/Compression.h"
 #include "velox/common/memory/Memory.h"
 #include "velox/vector/FlatVector.h"
@@ -69,6 +70,10 @@ VeloxColumnarBatchSerializer::VeloxColumnarBatchSerializer(
 
 void VeloxColumnarBatchSerializer::append(const std::shared_ptr<ColumnarBatch>& batch) {
   auto rowVector = VeloxColumnarBatch::from(veloxPool_.get(), batch)->getRowVector();
+  // A GPU-resident CudfVector serializes as zero rows because its host-side
+  // children are empty. Materialize to host so broadcast/shuffle payloads carry
+  // the real data instead of an empty build side.
+  rowVector = materializeVeloxRowVector(rowVector, veloxPool_.get());
   if (serializer_ == nullptr) {
     // Using first batch's schema to create the Velox serializer. This logic was introduced in
     // https://github.com/apache/gluten/pull/1568. It's a bit suboptimal because the schemas
@@ -396,6 +401,9 @@ std::vector<uint8_t> VeloxColumnarBatchSerializer::framedSerializeWithStats(
   // Compute stats over the inbound rowVector BEFORE delegating to the append path (which may
   // consume / mutate iterator state on subsequent calls).
   auto rowVector = VeloxColumnarBatch::from(veloxPool_.get(), batch)->getRowVector();
+  // Materialize a GPU-resident CudfVector to host so stats are computed over
+  // real rows (the subsequent append() re-materializes independently).
+  rowVector = materializeVeloxRowVector(rowVector, veloxPool_.get());
   const uint32_t numRows = static_cast<uint32_t>(rowVector->size());
   std::vector<ColumnStats> perCol = computeStats(rowVector);
   const uint32_t numCols = static_cast<uint32_t>(perCol.size());

cpp/velox/operators/serializer/VeloxColumnarToRowConverter.cc

@@ -67,7 +67,12 @@ void VeloxColumnarToRowConverter::refreshStates(facebook::velox::RowVectorPtr ro
 
 void VeloxColumnarToRowConverter::convert(std::shared_ptr<ColumnarBatch> cb, int64_t startRow) {
   auto veloxBatch = VeloxColumnarBatch::from(veloxPool_.get(), cb);
-  refreshStates(veloxBatch->getRowVector(), startRow);
+  // Materialize a GPU-resident CudfVector to host before the UnsafeRow read below.
+  // getFlattenedRowVector() -> ensureFlattened() -> materializeVeloxRowVector(). Without
+  // this, a device-resident CudfVector's host children are empty/garbage, so columnar->row
+  // (broadcast build side + result return) produces empty/corrupt rows -> empty broadcast
+  // joins and VARCHAR-offset SIGSEGVs. No-op for plain host RowVectors.
+  refreshStates(veloxBatch->getFlattenedRowVector(), startRow);
 
   // Initialize the offsets_ , lengths_
   lengths_.clear();

cpp/velox/utils/CudfVectorUtils.h

@@ -0,0 +1,47 @@
+/*
+ * 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
+
+#include "velox/common/memory/MemoryPool.h"
+#include "velox/vector/ComplexVector.h"
+
+#ifdef GLUTEN_ENABLE_GPU
+#include "velox/experimental/cudf/exec/VeloxCudfInterop.h"
+#include "velox/experimental/cudf/vector/CudfVector.h"
+#endif
+
+namespace gluten {
+
+// Converts a GPU-resident CudfVector to a host RowVector. A CudfVector holds
+// its data in a device-side cudf::table and exposes no host-side children, so
+// any host code that reads childAt()/serializes it would see zero rows. When
+// the input is a plain host RowVector it is returned unchanged.
+inline facebook::velox::RowVectorPtr materializeVeloxRowVector(
+    const facebook::velox::RowVectorPtr& rowVector,
+    facebook::velox::memory::MemoryPool* memoryPool) {
+#ifdef GLUTEN_ENABLE_GPU
+  auto cudfVector = std::dynamic_pointer_cast<facebook::velox::cudf_velox::CudfVector>(rowVector);
+  if (cudfVector != nullptr) {
+    return facebook::velox::cudf_velox::with_arrow::toVeloxColumn(
+        cudfVector->getTableView(), memoryPool, "", cudfVector->stream(), cudf::get_current_device_resource_ref());
+  }
+#endif
+  return rowVector;
+}
+
+} // namespace gluten

cpp/velox/utils/VeloxBatchResizer.cc

@@ -16,6 +16,7 @@
  */
 
 #include "VeloxBatchResizer.h"
+#include "utils/CudfVectorUtils.h"
 
 namespace gluten {
 namespace {
@@ -69,7 +70,9 @@ bool supportsCopyRanges(const facebook::velox::RowVectorPtr& rowVector) {
 class SliceRowVector : public ColumnarBatchIterator {
  public:
   SliceRowVector(int32_t maxOutputBatchSize, facebook::velox::RowVectorPtr in)
-      : maxOutputBatchSize_(maxOutputBatchSize), in_(in) {}
+      : maxOutputBatchSize_(maxOutputBatchSize), in_(std::move(in)) {
+    in_ = materializeVeloxRowVector(in_, in_->pool());
+  }
 
   std::shared_ptr<ColumnarBatch> next() override {
     int32_t remainingLength = in_->size() - cursor_;
@@ -209,17 +212,13 @@ std::shared_ptr<ColumnarBatch> VeloxBatchResizer::next() {
   uint64_t numBytes = cb->numBytes();
   if (cb->numRows() < minOutputBatchSize_ && numBytes <= preferredBatchBytes_) {
     auto vb = VeloxColumnarBatch::from(pool_, cb);
-    auto rv = vb->getRowVector();
-    if (enableCopyRanges_) {
-      return collectAndCopy(std::move(rv), numBytes);
-    }
-
+    auto rv = materializeVeloxRowVector(vb->getRowVector(), pool_);
     auto buffer = facebook::velox::RowVector::createEmpty(rv->type(), pool_);
     appendToBuffer(buffer, rv);
 
     for (cb = in_->next(); cb != nullptr; cb = in_->next()) {
       vb = VeloxColumnarBatch::from(pool_, cb);
-      rv = vb->getRowVector();
+      rv = materializeVeloxRowVector(vb->getRowVector(), pool_);
       uint64_t addedBytes = cb->numBytes();
       if (buffer->size() + rv->size() > maxOutputBatchSize_ ||
           numBytes + addedBytes > static_cast<uint64_t>(preferredBatchBytes_)) {
@@ -243,7 +242,7 @@ std::shared_ptr<ColumnarBatch> VeloxBatchResizer::next() {
 
   if (cb->numRows() > maxOutputBatchSize_) {
     auto vb = VeloxColumnarBatch::from(pool_, cb);
-    auto rv = vb->getRowVector();
+    auto rv = materializeVeloxRowVector(vb->getRowVector(), pool_);
     GLUTEN_CHECK(next_ == nullptr, "Invalid state");
     next_ = std::make_unique<SliceRowVector>(maxOutputBatchSize_, rv);
     auto next = next_->next();