diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala index 2d4b1570565..074eae17dfc 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala +++ b/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) diff --git a/cpp/velox/compute/VeloxRuntime.h b/cpp/velox/compute/VeloxRuntime.h index 3f53653e5ba..52ca25f76be 100644 --- a/cpp/velox/compute/VeloxRuntime.h +++ b/cpp/velox/compute/VeloxRuntime.h @@ -17,18 +17,18 @@ #pragma once +#include +#include "IcebergNestedField.pb.h" #include "WholeStageResultIterator.h" #include "compute/Runtime.h" #include "compute/VeloxConnectorIds.h" #include "iceberg/IcebergWriter.h" -#include #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 { diff --git a/cpp/velox/jni/JniHashTable.cc b/cpp/velox/jni/JniHashTable.cc index d7259879c59..0b76bd419e2 100644 --- a/cpp/velox/jni/JniHashTable.cc +++ b/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 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; diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc index 36798f5cded..d855db8dc71 100644 --- a/cpp/velox/jni/VeloxJniWrapper.cc +++ b/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(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(prunedVector); diff --git a/cpp/velox/memory/VeloxColumnarBatch.cc b/cpp/velox/memory/VeloxColumnarBatch.cc index 038bd9f798f..f6ce16cee14 100644 --- a/cpp/velox/memory/VeloxColumnarBatch.cc +++ b/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::compose( throw GlutenException("Mismatched row counts among the input batches during composing columnar batches"); } auto vb = std::dynamic_pointer_cast(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::compose( std::vector children; for (const auto& batch : batches) { auto vb = std::dynamic_pointer_cast(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::select( std::vector 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(rowVector); } std::vector VeloxColumnarBatch::toUnsafeRow(int32_t rowId) const { - auto fast = std::make_unique(rowVector_); + auto rowVector = materializeVeloxRowVector(rowVector_, rowVector_->pool()); + auto fast = std::make_unique(rowVector); auto size = fast->rowSize(rowId); std::vector bytes(size); std::memset(bytes.data(), 0, bytes.size()); diff --git a/cpp/velox/operators/hashjoin/HashTableBuilder.cc b/cpp/velox/operators/hashjoin/HashTableBuilder.cc index 4c3129fccf8..28d6c02b809 100644 --- a/cpp/velox/operators/hashjoin/HashTableBuilder.cc +++ b/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(); diff --git a/cpp/velox/operators/plannodes/CudfVectorStream.cc b/cpp/velox/operators/plannodes/CudfVectorStream.cc index 2cec9b67b23..840fde61719 100644 --- a/cpp/velox/operators/plannodes/CudfVectorStream.cc +++ b/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 CudfVectorStreamBase::nextInternal() { facebook::velox::RowVectorPtr CudfVectorStreamBase::next() { auto cb = nextInternal(); const std::shared_ptr& vb = VeloxColumnarBatch::from(pool_, cb); - auto vp = vb->getRowVector(); + auto vp = materializeVeloxRowVector(vb->getRowVector(), pool_); VELOX_DCHECK(vp != nullptr); return std::make_shared( vp->pool(), outputType_, facebook::velox::BufferPtr(0), vp->size(), vp->children()); } -} // namespace gluten \ No newline at end of file +} // namespace gluten diff --git a/cpp/velox/operators/plannodes/RowVectorStream.cc b/cpp/velox/operators/plannodes/RowVectorStream.cc index 4baa1f84712..e8757134b9a 100644 --- a/cpp/velox/operators/plannodes/RowVectorStream.cc +++ b/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 RowVectorStream::nextInternal() { facebook::velox::RowVectorPtr RowVectorStream::next() { auto cb = nextInternal(); const std::shared_ptr& vb = VeloxColumnarBatch::from(pool_, cb); - auto vp = vb->getRowVector(); + auto vp = materializeVeloxRowVector(vb->getRowVector(), pool_); VELOX_DCHECK(vp != nullptr); return std::make_shared( vp->pool(), outputType_, facebook::velox::BufferPtr(0), vp->size(), vp->children()); @@ -278,4 +278,4 @@ void ValueStreamDataSource::applyFilterOnColumn( rows.updateBounds(); } -} // namespace gluten \ No newline at end of file +} // namespace gluten diff --git a/cpp/velox/operators/serializer/VeloxColumnarBatchSerializer.cc b/cpp/velox/operators/serializer/VeloxColumnarBatchSerializer.cc index cc1cd6bb4ef..fec876bf565 100644 --- a/cpp/velox/operators/serializer/VeloxColumnarBatchSerializer.cc +++ b/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& 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 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(rowVector->size()); std::vector perCol = computeStats(rowVector); const uint32_t numCols = static_cast(perCol.size()); diff --git a/cpp/velox/operators/serializer/VeloxColumnarToRowConverter.cc b/cpp/velox/operators/serializer/VeloxColumnarToRowConverter.cc index cbf25ff7ba7..705a55de817 100644 --- a/cpp/velox/operators/serializer/VeloxColumnarToRowConverter.cc +++ b/cpp/velox/operators/serializer/VeloxColumnarToRowConverter.cc @@ -67,7 +67,12 @@ void VeloxColumnarToRowConverter::refreshStates(facebook::velox::RowVectorPtr ro void VeloxColumnarToRowConverter::convert(std::shared_ptr 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(); diff --git a/cpp/velox/utils/CudfVectorUtils.h b/cpp/velox/utils/CudfVectorUtils.h new file mode 100644 index 00000000000..a652f52915c --- /dev/null +++ b/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(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 diff --git a/cpp/velox/utils/VeloxBatchResizer.cc b/cpp/velox/utils/VeloxBatchResizer.cc index c67b1c75111..84ccc5ca77a 100644 --- a/cpp/velox/utils/VeloxBatchResizer.cc +++ b/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 next() override { int32_t remainingLength = in_->size() - cursor_; @@ -209,17 +212,13 @@ std::shared_ptr 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(preferredBatchBytes_)) { @@ -243,7 +242,7 @@ std::shared_ptr 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(maxOutputBatchSize_, rv); auto next = next_->next();