Copilot commented on code in PR #12370:
URL: https://github.com/apache/gluten/pull/12370#discussion_r3535571224
##########
cpp/velox/shuffle/VeloxShuffleReader.cc:
##########
@@ -909,88 +949,85 @@ size_t
VeloxRssSortShuffleReaderDeserializer::VeloxInputStream::remainingSize()
return std::numeric_limits<unsigned long>::max();
}
-VeloxShuffleReaderDeserializerFactory::VeloxShuffleReaderDeserializerFactory(
+VeloxShuffleReader::VeloxShuffleReader(
const std::shared_ptr<arrow::Schema>& schema,
- const std::shared_ptr<arrow::util::Codec>& codec,
- facebook::velox::common::CompressionKind veloxCompressionType,
- const RowTypePtr& rowType,
- int32_t batchSize,
- int64_t readerBufferSize,
- int64_t deserializerBufferSize,
VeloxMemoryManager* memoryManager,
- ShuffleWriterType shuffleWriterType,
- bool enableHashShuffleReaderStreamMerge)
- : schema_(schema),
- codec_(codec),
- veloxCompressionType_(veloxCompressionType),
- rowType_(rowType),
- batchSize_(batchSize),
- readerBufferSize_(readerBufferSize),
- deserializerBufferSize_(deserializerBufferSize),
- memoryManager_(memoryManager),
- shuffleWriterType_(shuffleWriterType),
- enableHashShuffleReaderStreamMerge_(enableHashShuffleReaderStreamMerge) {
+ const std::shared_ptr<ShuffleReaderOptions>& options)
+ : schema_(schema), memoryManager_(memoryManager), options_(options) {
Review Comment:
`shuffleWriterType_` is never initialized from `options`, but
`createDeserializer()` switches on it. This is undefined behavior and can
select the wrong deserializer at runtime. Initialize it from
`options->shuffleWriterType` (or switch on `options_->shuffleWriterType`
directly).
##########
cpp/velox/shuffle/VeloxShuffleReader.cc:
##########
@@ -909,88 +949,85 @@ size_t
VeloxRssSortShuffleReaderDeserializer::VeloxInputStream::remainingSize()
return std::numeric_limits<unsigned long>::max();
}
-VeloxShuffleReaderDeserializerFactory::VeloxShuffleReaderDeserializerFactory(
+VeloxShuffleReader::VeloxShuffleReader(
const std::shared_ptr<arrow::Schema>& schema,
- const std::shared_ptr<arrow::util::Codec>& codec,
- facebook::velox::common::CompressionKind veloxCompressionType,
- const RowTypePtr& rowType,
- int32_t batchSize,
- int64_t readerBufferSize,
- int64_t deserializerBufferSize,
VeloxMemoryManager* memoryManager,
- ShuffleWriterType shuffleWriterType,
- bool enableHashShuffleReaderStreamMerge)
- : schema_(schema),
- codec_(codec),
- veloxCompressionType_(veloxCompressionType),
- rowType_(rowType),
- batchSize_(batchSize),
- readerBufferSize_(readerBufferSize),
- deserializerBufferSize_(deserializerBufferSize),
- memoryManager_(memoryManager),
- shuffleWriterType_(shuffleWriterType),
- enableHashShuffleReaderStreamMerge_(enableHashShuffleReaderStreamMerge) {
+ const std::shared_ptr<ShuffleReaderOptions>& options)
+ : schema_(schema), memoryManager_(memoryManager), options_(options) {
+ codec_ = gluten::createCompressionCodec(options->compressionType,
options->codecBackend);
+ veloxCompressionType_ =
arrowCompressionTypeToVelox(options->compressionType);
+ rowType_ = facebook::velox::asRowType(gluten::fromArrowSchema(schema));
initFromSchema();
}
-std::unique_ptr<ColumnarBatchIterator>
VeloxShuffleReaderDeserializerFactory::createDeserializer(
- const std::shared_ptr<StreamReader>& streamReader) {
+void VeloxShuffleReader::createDeserializer(const
std::shared_ptr<StreamReader>& streamReader) {
switch (shuffleWriterType_) {
- case ShuffleWriterType::kGpuHashShuffle:
+ case ShuffleWriterType::kGpuHashShuffle: {
#ifdef GLUTEN_ENABLE_GPU
VELOX_CHECK(!hasComplexType_);
- return std::make_unique<VeloxGpuHashShuffleReaderDeserializer>(
- streamReader,
- schema_,
- codec_,
- rowType_,
- readerBufferSize_,
- memoryManager_,
- deserializeTime_,
- decompressTime_);
+ if (options_->enableGpuAsyncReader) {
+ deserializer_ =
std::make_unique<VeloxGpuAsyncHashShuffleReaderDeserializer>(
+ streamReader,
+ schema_,
+ codec_,
+ rowType_,
+ readerBufferSize_,
Review Comment:
`readerBufferSize_` is referenced here, but it is not declared as a member
of `VeloxShuffleReader` (and the other cases already use
`options_->readerBufferSize`). This will not compile / will break the GPU
reader construction. Use `options_->readerBufferSize` instead.
##########
cpp/velox/compute/VeloxBackend.cc:
##########
@@ -303,6 +303,18 @@ facebook::velox::cache::AsyncDataCache*
VeloxBackend::getAsyncDataCache() const
return asyncDataCache_.get();
}
+ReaderThreadPool* VeloxBackend::getReaderThreadPool() {
+ static std::once_flag readerThreadPoolInit;
+ std::call_once(readerThreadPoolInit, [this] {
+ const auto configuredThreads =
+ backendConf_->get<int32_t>(kShuffleReaderThreads,
static_cast<int32_t>(std::thread::hardware_concurrency()));
+ // std::thread::hardware_concurrency() can return 0;
+ const auto numThreads = configuredThreads > 0 ? configuredThreads : 1;
+ readerThreadPool_ = std::make_unique<ReaderThreadPool>(numThreads);
Review Comment:
The default thread count here uses `std::thread::hardware_concurrency()`,
but the Spark-side conf
`spark.gluten.sql.columnar.backend.velox.gpuShuffleReader.threadPoolSize` has a
default of 1 (see `VeloxConfig.scala`). If the key is absent from
`backendConf_`, this will unexpectedly create many native threads, changing
behavior vs the documented/Scala defaults.
##########
backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarBatchSerializer.scala:
##########
@@ -253,7 +254,6 @@ private class CelebornColumnarBatchSerializerInstance(
if (wrappedOut != null) {
wrappedOut.close()
}
- streamReader.close()
if (cb != null) {
cb.close()
Review Comment:
`ShuffleStreamReader` no longer closes the previous JNI stream
automatically, and this Celeborn deserialization path doesn’t call the new
`ShuffleReaderJniWrapper.stop()` on close/release. If the consumer stops early
(e.g. LIMIT), the native reader can keep holding an in-flight stream/buffer
until task cleanup. Call `stop` in `close0()` to promptly release resources.
##########
cpp/velox/tests/VeloxShuffleWriterTest.cc:
##########
@@ -318,24 +322,17 @@ class VeloxShuffleWriterTest : public
::testing::TestWithParam<ShuffleTestParams
const RowTypePtr& rowType,
std::vector<facebook::velox::RowVectorPtr>& vectors,
std::shared_ptr<arrow::io::InputStream> in) {
- const auto veloxCompressionType =
arrowCompressionTypeToVelox(compressionType);
const auto schema = toArrowSchema(rowType,
getDefaultMemoryManager()->getLeafMemoryPool().get());
-
- auto codec = createCompressionCodec(compressionType, CodecBackend::NONE);
-
+ const auto options = std::make_shared<ShuffleReaderOptions>();
+ options->compressionType = compressionType;
// Set batchSize to a large value to make all batches are merged by reader.
- auto deserializerFactory =
std::make_unique<gluten::VeloxShuffleReaderDeserializerFactory>(
- schema,
- std::move(codec),
- veloxCompressionType,
- rowType,
- kDefaultBatchSize,
- kDefaultReadBufferSize,
- GetParam().deserializerBufferSize,
- getDefaultMemoryManager(),
- GetParam().shuffleWriterType);
-
- const auto reader =
std::make_shared<VeloxShuffleReader>(std::move(deserializerFactory));
+ options->batchSize = kDefaultBatchSize;
+ options->readerBufferSize = kDefaultReadBufferSize;
+ options->deserializerBufferSize = GetParam().deserializerBufferSize;
+ options->deserializerBufferSize = GetParam().deserializerBufferSize;
Review Comment:
`options->deserializerBufferSize` is assigned twice consecutively, which is
redundant and makes it easy to miss a missing assignment for another field.
##########
cpp/velox/shuffle/VeloxShuffleReader.cc:
##########
@@ -909,88 +949,85 @@ size_t
VeloxRssSortShuffleReaderDeserializer::VeloxInputStream::remainingSize()
return std::numeric_limits<unsigned long>::max();
}
-VeloxShuffleReaderDeserializerFactory::VeloxShuffleReaderDeserializerFactory(
+VeloxShuffleReader::VeloxShuffleReader(
const std::shared_ptr<arrow::Schema>& schema,
- const std::shared_ptr<arrow::util::Codec>& codec,
- facebook::velox::common::CompressionKind veloxCompressionType,
- const RowTypePtr& rowType,
- int32_t batchSize,
- int64_t readerBufferSize,
- int64_t deserializerBufferSize,
VeloxMemoryManager* memoryManager,
- ShuffleWriterType shuffleWriterType,
- bool enableHashShuffleReaderStreamMerge)
- : schema_(schema),
- codec_(codec),
- veloxCompressionType_(veloxCompressionType),
- rowType_(rowType),
- batchSize_(batchSize),
- readerBufferSize_(readerBufferSize),
- deserializerBufferSize_(deserializerBufferSize),
- memoryManager_(memoryManager),
- shuffleWriterType_(shuffleWriterType),
- enableHashShuffleReaderStreamMerge_(enableHashShuffleReaderStreamMerge) {
+ const std::shared_ptr<ShuffleReaderOptions>& options)
+ : schema_(schema), memoryManager_(memoryManager), options_(options) {
+ codec_ = gluten::createCompressionCodec(options->compressionType,
options->codecBackend);
+ veloxCompressionType_ =
arrowCompressionTypeToVelox(options->compressionType);
+ rowType_ = facebook::velox::asRowType(gluten::fromArrowSchema(schema));
initFromSchema();
}
-std::unique_ptr<ColumnarBatchIterator>
VeloxShuffleReaderDeserializerFactory::createDeserializer(
- const std::shared_ptr<StreamReader>& streamReader) {
+void VeloxShuffleReader::createDeserializer(const
std::shared_ptr<StreamReader>& streamReader) {
switch (shuffleWriterType_) {
- case ShuffleWriterType::kGpuHashShuffle:
+ case ShuffleWriterType::kGpuHashShuffle: {
#ifdef GLUTEN_ENABLE_GPU
VELOX_CHECK(!hasComplexType_);
- return std::make_unique<VeloxGpuHashShuffleReaderDeserializer>(
- streamReader,
- schema_,
- codec_,
- rowType_,
- readerBufferSize_,
- memoryManager_,
- deserializeTime_,
- decompressTime_);
+ if (options_->enableGpuAsyncReader) {
+ deserializer_ =
std::make_unique<VeloxGpuAsyncHashShuffleReaderDeserializer>(
+ streamReader,
+ schema_,
+ codec_,
+ rowType_,
+ readerBufferSize_,
+ memoryManager_,
+ deserializeTime_,
+ decompressTime_);
+ } else {
+ deserializer_ =
std::make_unique<VeloxGpuHashShuffleReaderDeserializer>(
+ streamReader,
+ schema_,
+ codec_,
+ rowType_,
+ readerBufferSize_,
Review Comment:
Same issue as above: `readerBufferSize_` is not a member and should be
`options_->readerBufferSize` to match the new options-based construction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
