Copilot commented on code in PR #12370:
URL: https://github.com/apache/gluten/pull/12370#discussion_r3535486295
##########
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 used in `createDeserializer()` but never initialized
in the `VeloxShuffleReader` constructor, so the `switch` is undefined behavior
and can select the wrong deserializer at runtime. Initialize it from the
provided `options` (e.g., `options->shuffleWriterType`).
##########
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 not a member of `VeloxShuffleReader` (it was removed
when refactoring to `ShuffleReaderOptions`), so this reference will not
compile. Use `options_->readerBufferSize` here.
##########
cpp/velox/benchmarks/GenericBenchmark.cc:
##########
@@ -237,9 +237,9 @@ std::shared_ptr<VeloxShuffleWriter> createShuffleWriter(
}
std::shared_ptr<ShuffleReader> createShuffleReader(Runtime* runtime, const
std::shared_ptr<arrow::Schema>& schema) {
- auto readerOptions = ShuffleReaderOptions{};
- readerOptions.shuffleWriterType =
ShuffleWriter::stringToType(FLAGS_shuffle_writer),
- setCompressionTypeFromFlag(readerOptions.compressionType,
readerOptions.codecBackend);
+ auto readerOptions = std::make_shared<ShuffleReaderOptions>();
+ readerOptions->shuffleWriterType =
ShuffleWriter::stringToType(FLAGS_shuffle_writer),
+ setCompressionTypeFromFlag(readerOptions->compressionType,
readerOptions->codecBackend);
Review Comment:
This assignment ends with a comma, which turns it into a comma-operator
expression / invalid statement sequence and breaks compilation. It should be a
normal statement terminated with `;` before calling
`setCompressionTypeFromFlag`.
##########
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:
`deserializerBufferSize` is assigned twice with the same value; this is
redundant and looks like a copy/paste error.
##########
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 `VeloxShuffleReader`
member, so this reference will not compile. Use `options_->readerBufferSize`
when constructing the deserializer.
--
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]
