You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "afterincomparableyum (via GitHub)" <gi...@apache.org> on 2025/12/28 22:14:20 UTC
[PR] [WIP][CELEBORN-2221][CIP-14] Add support for write compression in CppClient [celeborn]
afterincomparableyum opened a new pull request, #3575:
URL: https://github.com/apache/celeborn/pull/3575
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
### Why are the changes needed?
### Does this PR resolve a correctness bug?
<!-- Yes/No. (Note: If yes, committer will add `correctness` label to current pull request). -->
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
Unit Tests, as well as compiling code
WIP
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3851507279
@SteNicholas I took a look at the comments of Copilot and made changes as needed.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3833196230
@afterincomparableyum, could you take a look at the review comment of Copilot?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2750571321
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -154,23 +160,37 @@ int ShuffleClientImpl::pushData(
auto pushState = getPushState(mapKey);
const int nextBatchId = pushState->nextBatchId();
- // TODO: compression in writing is not supported.
+ // Compression support: compress data if compression is enabled
+ const uint8_t* dataToWrite = data + offset;
+ size_t lengthToWrite = length;
+ std::unique_ptr<uint8_t[]> compressedBuffer;
+
+ if (shuffleCompressionEnabled_ && compressor_) {
+ // Allocate buffer for compressed data
+ const size_t compressedCapacity = compressor_->getDstCapacity(length);
+ compressedBuffer = std::make_unique<uint8_t[]>(compressedCapacity);
+
+ // Compress the data
+ lengthToWrite = compressor_->compress(
+ dataToWrite, 0, length, compressedBuffer.get(), 0);
+ dataToWrite = compressedBuffer.get();
+ }
auto writeBuffer =
- memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + length);
+ memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + lengthToWrite);
// TODO: the java side uses Platform to write the data. We simply assume
// littleEndian here.
writeBuffer->writeLE<int>(mapId);
writeBuffer->writeLE<int>(attemptId);
writeBuffer->writeLE<int>(nextBatchId);
- writeBuffer->writeLE<int>(length);
- writeBuffer->writeFromBuffer(data, offset, length);
+ writeBuffer->writeLE<int>(lengthToWrite);
+ writeBuffer->writeFromBuffer(dataToWrite, 0, lengthToWrite);
Review Comment:
this is a good suggestion I have applied
##########
cpp/celeborn/client/tests/CompressorFactoryTest.cpp:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+#include <gtest/gtest.h>
+
+#include "celeborn/client/compress/Compressor.h"
+#include "celeborn/conf/CelebornConf.h"
+
+using namespace celeborn;
+using namespace celeborn::client;
+using namespace celeborn::conf;
+using namespace celeborn::protocol;
+
+TEST(CompressorFactoryTest, CreateLz4CompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's an LZ4 compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
Review Comment:
done
##########
cpp/celeborn/client/tests/CompressorFactoryTest.cpp:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+#include <gtest/gtest.h>
+
+#include "celeborn/client/compress/Compressor.h"
+#include "celeborn/conf/CelebornConf.h"
+
+using namespace celeborn;
+using namespace celeborn::client;
+using namespace celeborn::conf;
+using namespace celeborn::protocol;
+
+TEST(CompressorFactoryTest, CreateLz4CompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's an LZ4 compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CreateZstdCompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel, "3");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's a ZSTD compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CompressionCodecNoneDisablesCompression) {
+ CelebornConf conf;
+ // Verify default is NONE
+ EXPECT_EQ(conf.shuffleCompressionCodec(), CompressionCodec::NONE);
+}
+
+TEST(CompressorFactoryTest, ZstdCompressionLevelFromConf) {
+ // Test that configuration correctly reads ZSTD compression levels
+ for (int level = -5; level <= 10; level++) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel,
+ std::to_string(level));
+
+ // Verify the compression level is set correctly
+ EXPECT_EQ(conf.shuffleCompressionZstdCompressLevel(), level);
+
+ // Verify the compressor is created correctly
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+ }
+}
+
+TEST(CompressorFactoryTest, CompressWithOffsetLz4) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ const std::string prefix = "SKIP_THIS_PREFIX";
+ const std::string testData =
+ "Celeborn compression offset test with structured data: "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4";
+ std::string fullData = prefix + testData;
+
+ const auto maxLength = compressor->getDstCapacity(testData.size());
+ std::vector<uint8_t> compressedData(maxLength);
+
+ // Compress with offset (simulating pushData usage pattern)
+ const size_t compressedSize = compressor->compress(
+ reinterpret_cast<const uint8_t*>(fullData.data()),
+ prefix.size(),
+ testData.size(),
+ compressedData.data(),
+ 0);
+
+ // Verify compression succeeded with offset
+ EXPECT_GT(compressedSize, 0);
+ EXPECT_LE(compressedSize, maxLength);
+}
Review Comment:
done
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2810937431
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -57,7 +58,16 @@ ShuffleClientImpl::ShuffleClientImpl(
: appUniqueId_(appUniqueId),
conf_(conf),
clientFactory_(clientEndpoint.clientFactory()),
- pushDataRetryPool_(clientEndpoint.pushDataRetryPool()) {
+ pushDataRetryPool_(clientEndpoint.pushDataRetryPool()),
+ shuffleCompressionEnabled_(
+ conf->shuffleCompressionCodec() != protocol::CompressionCodec::NONE),
+ compressorFactory_(
+ shuffleCompressionEnabled_
+ ? std::function<std::unique_ptr<compress::Compressor>()>(
+ [conf]() {
+ return compress::Compressor::createCompressor(*conf);
+ })
+ : std::function<std::unique_ptr<compress::Compressor>()>()) {
Review Comment:
The constructor initialization list order does not match the member declaration order in the header file. In ShuffleClient.h, members are declared in this order: appUniqueId_, shuffleCompressionEnabled_, conf_, clientFactory_, pushDataRetryPool_. However, in the constructor initialization list, they are initialized as: appUniqueId_, conf_, clientFactory_, pushDataRetryPool_, shuffleCompressionEnabled_. Members are always initialized in declaration order regardless of the initialization list order, which can cause issues when members depend on each other. This violates C++ best practices and may trigger -Wreorder warnings. Reorder the initialization list to match the declaration order in the header.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2767453251
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -57,7 +58,16 @@ ShuffleClientImpl::ShuffleClientImpl(
: appUniqueId_(appUniqueId),
conf_(conf),
clientFactory_(clientEndpoint.clientFactory()),
- pushDataRetryPool_(clientEndpoint.pushDataRetryPool()) {
+ pushDataRetryPool_(clientEndpoint.pushDataRetryPool()),
+ shuffleCompressionEnabled_(
+ conf->shuffleCompressionCodec() != protocol::CompressionCodec::NONE),
+ compressorFactory_(
+ shuffleCompressionEnabled_
+ ? std::function<std::unique_ptr<compress::Compressor>()>(
+ [conf]() {
+ return compress::Compressor::createCompressor(*conf);
+ })
+ : std::function<std::unique_ptr<compress::Compressor>()>()) {
Review Comment:
this is again wrong, the init list order in the constructor matches the declaration order in the header file.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3800309088
@SteNicholas I have fixed the clang format issues.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2767458255
##########
cpp/celeborn/client/ShuffleClient.h:
##########
@@ -266,6 +268,9 @@ class ShuffleClientImpl
mapperEndSets_;
utils::ConcurrentHashSet<int> stageEndShuffleSet_;
+ // Factory for creating compressor instances on demand to avoid sharing a
+ // single non-thread-safe compressor across concurrent operations.
+ std::function<std::unique_ptr<compress::Compressor>()> compressorFactory_;
Review Comment:
added
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3764463277
@HolyLow @SteNicholas @FMX @RexXiong Could you please help review this PR? Appreciate your help in improving this as needed!
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2750571474
##########
cpp/celeborn/client/tests/CompressorFactoryTest.cpp:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+#include <gtest/gtest.h>
+
+#include "celeborn/client/compress/Compressor.h"
+#include "celeborn/conf/CelebornConf.h"
+
+using namespace celeborn;
+using namespace celeborn::client;
+using namespace celeborn::conf;
+using namespace celeborn::protocol;
+
+TEST(CompressorFactoryTest, CreateLz4CompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's an LZ4 compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CreateZstdCompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel, "3");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's a ZSTD compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CompressionCodecNoneDisablesCompression) {
+ CelebornConf conf;
+ // Verify default is NONE
+ EXPECT_EQ(conf.shuffleCompressionCodec(), CompressionCodec::NONE);
+}
+
+TEST(CompressorFactoryTest, ZstdCompressionLevelFromConf) {
+ // Test that configuration correctly reads ZSTD compression levels
+ for (int level = -5; level <= 10; level++) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel,
+ std::to_string(level));
+
+ // Verify the compression level is set correctly
+ EXPECT_EQ(conf.shuffleCompressionZstdCompressLevel(), level);
+
+ // Verify the compressor is created correctly
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+ }
+}
+
+TEST(CompressorFactoryTest, CompressWithOffsetLz4) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ const std::string prefix = "SKIP_THIS_PREFIX";
+ const std::string testData =
+ "Celeborn compression offset test with structured data: "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4";
+ std::string fullData = prefix + testData;
+
+ const auto maxLength = compressor->getDstCapacity(testData.size());
+ std::vector<uint8_t> compressedData(maxLength);
+
+ // Compress with offset (simulating pushData usage pattern)
+ const size_t compressedSize = compressor->compress(
+ reinterpret_cast<const uint8_t*>(fullData.data()),
+ prefix.size(),
+ testData.size(),
+ compressedData.data(),
+ 0);
+
+ // Verify compression succeeded with offset
+ EXPECT_GT(compressedSize, 0);
+ EXPECT_LE(compressedSize, maxLength);
+}
+
+TEST(CompressorFactoryTest, CompressWithOffsetZstd) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel, "3");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ const std::string prefix = "SKIP_THIS_PREFIX";
+ const std::string testData =
+ "Celeborn compression offset test with structured data: "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4";
+ std::string fullData = prefix + testData;
+
+ const auto maxLength = compressor->getDstCapacity(testData.size());
+ std::vector<uint8_t> compressedData(maxLength);
+
+ // Compress with offset (simulating pushData usage pattern)
+ const size_t compressedSize = compressor->compress(
+ reinterpret_cast<const uint8_t*>(fullData.data()),
+ prefix.size(),
+ testData.size(),
+ compressedData.data(),
+ 0);
+
+ // Verify compression succeeded with offset
+ EXPECT_GT(compressedSize, 0);
+ EXPECT_LE(compressedSize, maxLength);
+}
Review Comment:
done
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2750572424
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -154,23 +160,37 @@ int ShuffleClientImpl::pushData(
auto pushState = getPushState(mapKey);
const int nextBatchId = pushState->nextBatchId();
- // TODO: compression in writing is not supported.
+ // Compression support: compress data if compression is enabled
+ const uint8_t* dataToWrite = data + offset;
+ size_t lengthToWrite = length;
+ std::unique_ptr<uint8_t[]> compressedBuffer;
+
+ if (shuffleCompressionEnabled_ && compressor_) {
+ // Allocate buffer for compressed data
+ const size_t compressedCapacity = compressor_->getDstCapacity(length);
+ compressedBuffer = std::make_unique<uint8_t[]>(compressedCapacity);
+
+ // Compress the data
+ lengthToWrite = compressor_->compress(
+ dataToWrite, 0, length, compressedBuffer.get(), 0);
+ dataToWrite = compressedBuffer.get();
+ }
auto writeBuffer =
- memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + length);
+ memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + lengthToWrite);
// TODO: the java side uses Platform to write the data. We simply assume
// littleEndian here.
writeBuffer->writeLE<int>(mapId);
writeBuffer->writeLE<int>(attemptId);
writeBuffer->writeLE<int>(nextBatchId);
- writeBuffer->writeLE<int>(length);
- writeBuffer->writeFromBuffer(data, offset, length);
+ writeBuffer->writeLE<int>(lengthToWrite);
+ writeBuffer->writeFromBuffer(dataToWrite, 0, lengthToWrite);
Review Comment:
Constructor Initialization List Order:
appUniqueId_
conf_
clientFactory_
pushDataRetryPool_
shuffleCompressionEnabled_
compressorFactory_
The initialization order matches the declaration order perfectly.
The compression members (shuffleCompressionEnabled_ and compressorFactory_) are declared after conf_ in the header (lines 271 and 274) in the latest commit I added, and they're initialized after conf_ in the constructor. This is now correct.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3798199601
@afterincomparableyum, please fix the failure of CI which is caused by format.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2731602949
##########
cpp/celeborn/client/ShuffleClient.h:
##########
@@ -249,6 +250,8 @@ class ShuffleClientImpl
static constexpr size_t kBatchHeaderSize = 4 * 4;
const std::string appUniqueId_;
+ const bool shuffleCompressionEnabled_;
+ std::unique_ptr<compress::Compressor> compressor_;
Review Comment:
@afterincomparableyum, PTAL.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2706673880
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -154,23 +160,37 @@ int ShuffleClientImpl::pushData(
auto pushState = getPushState(mapKey);
const int nextBatchId = pushState->nextBatchId();
- // TODO: compression in writing is not supported.
+ // Compression support: compress data if compression is enabled
+ const uint8_t* dataToWrite = data + offset;
+ size_t lengthToWrite = length;
+ std::unique_ptr<uint8_t[]> compressedBuffer;
+
+ if (shuffleCompressionEnabled_ && compressor_) {
+ // Allocate buffer for compressed data
+ const size_t compressedCapacity = compressor_->getDstCapacity(length);
+ compressedBuffer = std::make_unique<uint8_t[]>(compressedCapacity);
+
+ // Compress the data
+ lengthToWrite =
+ compressor_->compress(data, offset, length, compressedBuffer.get(), 0);
+ dataToWrite = compressedBuffer.get();
Review Comment:
Should `offset` set to zero?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2752353901
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -57,7 +58,16 @@ ShuffleClientImpl::ShuffleClientImpl(
: appUniqueId_(appUniqueId),
conf_(conf),
clientFactory_(clientEndpoint.clientFactory()),
- pushDataRetryPool_(clientEndpoint.pushDataRetryPool()) {
+ pushDataRetryPool_(clientEndpoint.pushDataRetryPool()),
+ shuffleCompressionEnabled_(
+ conf->shuffleCompressionCodec() != protocol::CompressionCodec::NONE),
+ compressorFactory_(
+ shuffleCompressionEnabled_
+ ? std::function<std::unique_ptr<compress::Compressor>()>(
+ [conf]() {
+ return compress::Compressor::createCompressor(*conf);
+ })
+ : std::function<std::unique_ptr<compress::Compressor>()>()) {
Review Comment:
The constructor initializer list order does not match the member declaration order in ShuffleClientImpl (e.g., conf_ is listed before shuffleCompressionEnabled_ even though shuffleCompressionEnabled_ is declared first). This can trigger -Wreorder (often treated as an error) and is confusing. Reorder the initializer list to match the declaration order in ShuffleClient.h.
##########
cpp/celeborn/client/ShuffleClient.h:
##########
@@ -266,6 +268,9 @@ class ShuffleClientImpl
mapperEndSets_;
utils::ConcurrentHashSet<int> stageEndShuffleSet_;
+ // Factory for creating compressor instances on demand to avoid sharing a
+ // single non-thread-safe compressor across concurrent operations.
+ std::function<std::unique_ptr<compress::Compressor>()> compressorFactory_;
Review Comment:
This header now introduces a `std::function` member (`compressorFactory_`) but does not include `<functional>`. Relying on transitive includes is brittle and can break builds when include graphs change; add an explicit `<functional>` include in this header.
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -154,23 +164,68 @@ int ShuffleClientImpl::pushData(
auto pushState = getPushState(mapKey);
const int nextBatchId = pushState->nextBatchId();
- // TODO: compression in writing is not supported.
+ // Validate input size fits in 32-bit int since it is required by compressor
+ // API and wire protocol
+ CELEBORN_CHECK(
+ length <= static_cast<size_t>(std::numeric_limits<int>::max()),
+ fmt::format(
+ "Data length {} exceeds maximum supported size {}",
+ length,
+ std::numeric_limits<int>::max()));
+
+ // Compression support: compress data if compression is enabled
+ const uint8_t* dataToWrite = data + offset;
+ int lengthToWrite = static_cast<int>(length);
+ std::unique_ptr<uint8_t[]> compressedBuffer;
+
+ if (shuffleCompressionEnabled_ && compressorFactory_) {
+ // Create a new compressor instance for thread-safety
+ auto compressor = compressorFactory_();
+ // Allocate buffer for compressed data
+ const size_t compressedCapacity =
+ compressor->getDstCapacity(static_cast<int>(length));
+ compressedBuffer = std::make_unique<uint8_t[]>(compressedCapacity);
+
+ // Compress the data
+ const size_t compressedSize = compressor->compress(
+ dataToWrite, 0, static_cast<int>(length), compressedBuffer.get(), 0);
+
+ CELEBORN_CHECK(
+ compressedSize <= static_cast<size_t>(std::numeric_limits<int>::max()),
+ fmt::format(
+ "Compressed size {} exceeds maximum supported size {}",
+ compressedSize,
+ std::numeric_limits<int>::max()));
+
+ lengthToWrite = static_cast<int>(compressedSize);
+ dataToWrite = compressedBuffer.get();
+ }
- auto writeBuffer =
- memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + length);
+ // Validate final buffer size fits in size_t and int
+ CELEBORN_CHECK(
+ static_cast<size_t>(lengthToWrite) <=
+ std::numeric_limits<size_t>::max() - kBatchHeaderSize,
+ fmt::format(
+ "Buffer size {} + header {} would overflow",
+ lengthToWrite,
+ kBatchHeaderSize));
+
+ auto writeBuffer = memory::ByteBuffer::createWriteOnly(
+ kBatchHeaderSize + static_cast<size_t>(lengthToWrite));
// TODO: the java side uses Platform to write the data. We simply assume
// littleEndian here.
writeBuffer->writeLE<int>(mapId);
writeBuffer->writeLE<int>(attemptId);
writeBuffer->writeLE<int>(nextBatchId);
- writeBuffer->writeLE<int>(length);
- writeBuffer->writeFromBuffer(data, offset, length);
+ writeBuffer->writeLE<int>(lengthToWrite);
+ writeBuffer->writeFromBuffer(
+ dataToWrite, 0, static_cast<size_t>(lengthToWrite));
auto hostAndPushPort = partitionLocation->hostAndPushPort();
// Check limit.
limitMaxInFlight(mapKey, *pushState, hostAndPushPort);
// Add inFlight requests.
Review Comment:
`batchBytesSize` is computed as `lengthToWrite + kBatchHeaderSize` and stored in an `int`. Since `lengthToWrite` can be `INT_MAX` (it is only checked against `INT_MAX`), this addition can overflow and produce a negative/incorrect in-flight byte count. Add a guard that `lengthToWrite <= INT_MAX - kBatchHeaderSize` (or change the accounting type to a wider integer) before computing/storing `batchBytesSize`.
```suggestion
// Add inFlight requests.
CELEBORN_CHECK(
lengthToWrite <= std::numeric_limits<int>::max() - kBatchHeaderSize,
fmt::format(
"Batch bytes size {} + header {} would overflow int",
lengthToWrite,
kBatchHeaderSize));
```
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2725988906
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -154,23 +160,37 @@ int ShuffleClientImpl::pushData(
auto pushState = getPushState(mapKey);
const int nextBatchId = pushState->nextBatchId();
- // TODO: compression in writing is not supported.
+ // Compression support: compress data if compression is enabled
+ const uint8_t* dataToWrite = data + offset;
+ size_t lengthToWrite = length;
+ std::unique_ptr<uint8_t[]> compressedBuffer;
+
+ if (shuffleCompressionEnabled_ && compressor_) {
+ // Allocate buffer for compressed data
+ const size_t compressedCapacity = compressor_->getDstCapacity(length);
+ compressedBuffer = std::make_unique<uint8_t[]>(compressedCapacity);
+
+ // Compress the data
+ lengthToWrite =
+ compressor_->compress(data, offset, length, compressedBuffer.get(), 0);
+ dataToWrite = compressedBuffer.get();
Review Comment:
@SteNicholas it can be, and if we set `offset` to 0, we will have to change `data` to `dataToWrite` since `dataToWrite = data + offset`.
I will make the change and push it.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3904863438
@RexXiong @FMX @HolyLow could you please take another look at this PR when you are available. I appreciate your review
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2750572779
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -57,7 +57,13 @@ ShuffleClientImpl::ShuffleClientImpl(
: appUniqueId_(appUniqueId),
conf_(conf),
clientFactory_(clientEndpoint.clientFactory()),
- pushDataRetryPool_(clientEndpoint.pushDataRetryPool()) {
+ pushDataRetryPool_(clientEndpoint.pushDataRetryPool()),
+ shuffleCompressionEnabled_(
+ conf->shuffleCompressionCodec() != protocol::CompressionCodec::NONE),
+ compressor_(
+ shuffleCompressionEnabled_
+ ? compress::Compressor::createCompressor(*conf)
+ : nullptr) {
Review Comment:
Constructor Initialization List Order:
appUniqueId_
conf_
clientFactory_
pushDataRetryPool_
shuffleCompressionEnabled_
compressorFactory_
The initialization order matches the declaration order.
The compression members (shuffleCompressionEnabled_ and compressorFactory_) are declared after conf_ in the header in latest commit, and they're initialized after conf_ in the constructor. This is correct.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2750570268
##########
cpp/celeborn/client/ShuffleClient.h:
##########
@@ -249,6 +250,8 @@ class ShuffleClientImpl
static constexpr size_t kBatchHeaderSize = 4 * 4;
const std::string appUniqueId_;
+ const bool shuffleCompressionEnabled_;
+ std::unique_ptr<compress::Compressor> compressor_;
Review Comment:
I've applied the suggestion recommended here @SteNicholas
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2767454531
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -154,23 +164,68 @@ int ShuffleClientImpl::pushData(
auto pushState = getPushState(mapKey);
const int nextBatchId = pushState->nextBatchId();
- // TODO: compression in writing is not supported.
+ // Validate input size fits in 32-bit int since it is required by compressor
+ // API and wire protocol
+ CELEBORN_CHECK(
+ length <= static_cast<size_t>(std::numeric_limits<int>::max()),
+ fmt::format(
+ "Data length {} exceeds maximum supported size {}",
+ length,
+ std::numeric_limits<int>::max()));
+
+ // Compression support: compress data if compression is enabled
+ const uint8_t* dataToWrite = data + offset;
+ int lengthToWrite = static_cast<int>(length);
+ std::unique_ptr<uint8_t[]> compressedBuffer;
+
+ if (shuffleCompressionEnabled_ && compressorFactory_) {
+ // Create a new compressor instance for thread-safety
+ auto compressor = compressorFactory_();
+ // Allocate buffer for compressed data
+ const size_t compressedCapacity =
+ compressor->getDstCapacity(static_cast<int>(length));
+ compressedBuffer = std::make_unique<uint8_t[]>(compressedCapacity);
+
+ // Compress the data
+ const size_t compressedSize = compressor->compress(
+ dataToWrite, 0, static_cast<int>(length), compressedBuffer.get(), 0);
+
+ CELEBORN_CHECK(
+ compressedSize <= static_cast<size_t>(std::numeric_limits<int>::max()),
+ fmt::format(
+ "Compressed size {} exceeds maximum supported size {}",
+ compressedSize,
+ std::numeric_limits<int>::max()));
+
+ lengthToWrite = static_cast<int>(compressedSize);
+ dataToWrite = compressedBuffer.get();
+ }
- auto writeBuffer =
- memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + length);
+ // Validate final buffer size fits in size_t and int
+ CELEBORN_CHECK(
+ static_cast<size_t>(lengthToWrite) <=
+ std::numeric_limits<size_t>::max() - kBatchHeaderSize,
+ fmt::format(
+ "Buffer size {} + header {} would overflow",
+ lengthToWrite,
+ kBatchHeaderSize));
+
+ auto writeBuffer = memory::ByteBuffer::createWriteOnly(
+ kBatchHeaderSize + static_cast<size_t>(lengthToWrite));
// TODO: the java side uses Platform to write the data. We simply assume
// littleEndian here.
writeBuffer->writeLE<int>(mapId);
writeBuffer->writeLE<int>(attemptId);
writeBuffer->writeLE<int>(nextBatchId);
- writeBuffer->writeLE<int>(length);
- writeBuffer->writeFromBuffer(data, offset, length);
+ writeBuffer->writeLE<int>(lengthToWrite);
+ writeBuffer->writeFromBuffer(
+ dataToWrite, 0, static_cast<size_t>(lengthToWrite));
auto hostAndPushPort = partitionLocation->hostAndPushPort();
// Check limit.
limitMaxInFlight(mapKey, *pushState, hostAndPushPort);
// Add inFlight requests.
Review Comment:
sure, I can add this check as well
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3575: [CELEBORN-2221][CIP-14] Support writing with compression in C++ client
URL: https://github.com/apache/celeborn/pull/3575
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3906781850
Merged to main(v0.7.0).
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][CELEBORN-2221][CIP-14] Add support for write compression in CppClient [celeborn]
Posted by "afterincomparableyum (via GitHub)" <gi...@apache.org>.
afterincomparableyum commented on PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#issuecomment-3695092633
@HolyLow this PR is a WIP, I will rebase it off of main after https://github.com/apache/celeborn/pull/3568 gets merged
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2221][CIP-14] Support writing with compression in C++ client [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3575:
URL: https://github.com/apache/celeborn/pull/3575#discussion_r2731560232
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -154,23 +160,37 @@ int ShuffleClientImpl::pushData(
auto pushState = getPushState(mapKey);
const int nextBatchId = pushState->nextBatchId();
- // TODO: compression in writing is not supported.
+ // Compression support: compress data if compression is enabled
+ const uint8_t* dataToWrite = data + offset;
+ size_t lengthToWrite = length;
+ std::unique_ptr<uint8_t[]> compressedBuffer;
+
+ if (shuffleCompressionEnabled_ && compressor_) {
+ // Allocate buffer for compressed data
+ const size_t compressedCapacity = compressor_->getDstCapacity(length);
+ compressedBuffer = std::make_unique<uint8_t[]>(compressedCapacity);
+
+ // Compress the data
+ lengthToWrite = compressor_->compress(
+ dataToWrite, 0, length, compressedBuffer.get(), 0);
+ dataToWrite = compressedBuffer.get();
+ }
auto writeBuffer =
- memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + length);
+ memory::ByteBuffer::createWriteOnly(kBatchHeaderSize + lengthToWrite);
// TODO: the java side uses Platform to write the data. We simply assume
// littleEndian here.
writeBuffer->writeLE<int>(mapId);
writeBuffer->writeLE<int>(attemptId);
writeBuffer->writeLE<int>(nextBatchId);
- writeBuffer->writeLE<int>(length);
- writeBuffer->writeFromBuffer(data, offset, length);
+ writeBuffer->writeLE<int>(lengthToWrite);
+ writeBuffer->writeFromBuffer(dataToWrite, 0, lengthToWrite);
Review Comment:
`length`/`offset` are `size_t`, but the compressor interface and on-wire header use 32-bit ints (`compress(..., int srcLength, ...)` and `writeLE<int>(lengthToWrite)`). Right now there are multiple implicit narrowings (`size_t` -> `int`) which can overflow/truncate for large buffers (and `lengthToWrite` can also be slightly larger than `length` due to compression framing). Add an explicit bounds check and cast (or change the local types to `int`/`int32_t` after validation) before calling `getDstCapacity/compress` and before writing the length/batchBytesSize.
##########
cpp/celeborn/client/tests/CompressorFactoryTest.cpp:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+#include <gtest/gtest.h>
+
+#include "celeborn/client/compress/Compressor.h"
+#include "celeborn/conf/CelebornConf.h"
+
+using namespace celeborn;
+using namespace celeborn::client;
+using namespace celeborn::conf;
+using namespace celeborn::protocol;
+
+TEST(CompressorFactoryTest, CreateLz4CompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's an LZ4 compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
Review Comment:
This test doesn’t actually verify that the factory produced an LZ4 compressor: `getDstCapacity()` will be >0 for any codec. To make this test meaningful, assert on codec-specific behavior (e.g., compress a small payload and check the codec magic/header, or dynamic_cast to `Lz4Compressor` if RTTI is available).
##########
cpp/celeborn/client/tests/CompressorFactoryTest.cpp:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+#include <gtest/gtest.h>
+
+#include "celeborn/client/compress/Compressor.h"
+#include "celeborn/conf/CelebornConf.h"
+
+using namespace celeborn;
+using namespace celeborn::client;
+using namespace celeborn::conf;
+using namespace celeborn::protocol;
+
+TEST(CompressorFactoryTest, CreateLz4CompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's an LZ4 compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CreateZstdCompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel, "3");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's a ZSTD compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CompressionCodecNoneDisablesCompression) {
+ CelebornConf conf;
+ // Verify default is NONE
+ EXPECT_EQ(conf.shuffleCompressionCodec(), CompressionCodec::NONE);
+}
+
+TEST(CompressorFactoryTest, ZstdCompressionLevelFromConf) {
+ // Test that configuration correctly reads ZSTD compression levels
+ for (int level = -5; level <= 10; level++) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel,
+ std::to_string(level));
+
+ // Verify the compression level is set correctly
+ EXPECT_EQ(conf.shuffleCompressionZstdCompressLevel(), level);
+
+ // Verify the compressor is created correctly
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+ }
+}
+
+TEST(CompressorFactoryTest, CompressWithOffsetLz4) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ const std::string prefix = "SKIP_THIS_PREFIX";
+ const std::string testData =
+ "Celeborn compression offset test with structured data: "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4";
+ std::string fullData = prefix + testData;
+
+ const auto maxLength = compressor->getDstCapacity(testData.size());
+ std::vector<uint8_t> compressedData(maxLength);
+
+ // Compress with offset (simulating pushData usage pattern)
+ const size_t compressedSize = compressor->compress(
+ reinterpret_cast<const uint8_t*>(fullData.data()),
+ prefix.size(),
+ testData.size(),
+ compressedData.data(),
+ 0);
+
+ // Verify compression succeeded with offset
+ EXPECT_GT(compressedSize, 0);
+ EXPECT_LE(compressedSize, maxLength);
+}
+
+TEST(CompressorFactoryTest, CompressWithOffsetZstd) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel, "3");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ const std::string prefix = "SKIP_THIS_PREFIX";
+ const std::string testData =
+ "Celeborn compression offset test with structured data: "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4";
+ std::string fullData = prefix + testData;
+
+ const auto maxLength = compressor->getDstCapacity(testData.size());
+ std::vector<uint8_t> compressedData(maxLength);
+
+ // Compress with offset (simulating pushData usage pattern)
+ const size_t compressedSize = compressor->compress(
+ reinterpret_cast<const uint8_t*>(fullData.data()),
+ prefix.size(),
+ testData.size(),
+ compressedData.data(),
+ 0);
+
+ // Verify compression succeeded with offset
+ EXPECT_GT(compressedSize, 0);
+ EXPECT_LE(compressedSize, maxLength);
+}
Review Comment:
Same issue as the LZ4 offset test: this only asserts `compressedSize > 0`, which won’t catch bugs where `srcOffset` is ignored. Decompress and compare the result to `testData` to ensure the offset is respected.
##########
cpp/celeborn/client/tests/CompressorFactoryTest.cpp:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+#include <gtest/gtest.h>
+
+#include "celeborn/client/compress/Compressor.h"
+#include "celeborn/conf/CelebornConf.h"
+
+using namespace celeborn;
+using namespace celeborn::client;
+using namespace celeborn::conf;
+using namespace celeborn::protocol;
+
+TEST(CompressorFactoryTest, CreateLz4CompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's an LZ4 compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CreateZstdCompressorFromConf) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel, "3");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ // Verify it's a ZSTD compressor
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+}
+
+TEST(CompressorFactoryTest, CompressionCodecNoneDisablesCompression) {
+ CelebornConf conf;
+ // Verify default is NONE
+ EXPECT_EQ(conf.shuffleCompressionCodec(), CompressionCodec::NONE);
+}
+
+TEST(CompressorFactoryTest, ZstdCompressionLevelFromConf) {
+ // Test that configuration correctly reads ZSTD compression levels
+ for (int level = -5; level <= 10; level++) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "ZSTD");
+ conf.registerProperty(
+ CelebornConf::kShuffleCompressionZstdCompressLevel,
+ std::to_string(level));
+
+ // Verify the compression level is set correctly
+ EXPECT_EQ(conf.shuffleCompressionZstdCompressLevel(), level);
+
+ // Verify the compressor is created correctly
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+ EXPECT_GT(compressor->getDstCapacity(100), 0);
+ }
+}
+
+TEST(CompressorFactoryTest, CompressWithOffsetLz4) {
+ CelebornConf conf;
+ conf.registerProperty(CelebornConf::kShuffleCompressionCodec, "LZ4");
+
+ auto compressor = compress::Compressor::createCompressor(conf);
+ ASSERT_NE(compressor, nullptr);
+
+ const std::string prefix = "SKIP_THIS_PREFIX";
+ const std::string testData =
+ "Celeborn compression offset test with structured data: "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4 "
+ "partition_0:shuffle_1:map_2:attempt_0:batch_3:data_block_4";
+ std::string fullData = prefix + testData;
+
+ const auto maxLength = compressor->getDstCapacity(testData.size());
+ std::vector<uint8_t> compressedData(maxLength);
+
+ // Compress with offset (simulating pushData usage pattern)
+ const size_t compressedSize = compressor->compress(
+ reinterpret_cast<const uint8_t*>(fullData.data()),
+ prefix.size(),
+ testData.size(),
+ compressedData.data(),
+ 0);
+
+ // Verify compression succeeded with offset
+ EXPECT_GT(compressedSize, 0);
+ EXPECT_LE(compressedSize, maxLength);
+}
Review Comment:
The “offset” coverage here is weak: the test only checks that compression returns a non-zero size, which would still pass even if `srcOffset` were ignored and the wrong bytes were compressed. To validate offset handling, decompress the output and assert the decompressed content equals `testData` (and does not include `prefix`).
##########
cpp/celeborn/client/ShuffleClient.h:
##########
@@ -249,6 +250,8 @@ class ShuffleClientImpl
static constexpr size_t kBatchHeaderSize = 4 * 4;
const std::string appUniqueId_;
+ const bool shuffleCompressionEnabled_;
+ std::unique_ptr<compress::Compressor> compressor_;
Review Comment:
`compressor_` is stored as a single shared instance on `ShuffleClientImpl`, but `pushData()` appears designed to be callable concurrently (e.g., `PushState` uses atomics/concurrent maps). `Lz4Compressor` has mutable internal state (`XXH32_state_t*`) and is not thread-safe, so concurrent `compressor_->compress()` calls can corrupt output/checksums. Consider making the compressor thread-safe (internal locking), using a per-call/per-thread compressor instance, or switching LZ4 checksum computation to a stateless API.
```suggestion
// Factory for creating compressor instances on demand to avoid sharing a
// single non-thread-safe compressor across concurrent operations.
std::function<std::unique_ptr<compress::Compressor>()> compressorFactory_;
```
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -57,7 +57,13 @@ ShuffleClientImpl::ShuffleClientImpl(
: appUniqueId_(appUniqueId),
conf_(conf),
clientFactory_(clientEndpoint.clientFactory()),
- pushDataRetryPool_(clientEndpoint.pushDataRetryPool()) {
+ pushDataRetryPool_(clientEndpoint.pushDataRetryPool()),
+ shuffleCompressionEnabled_(
+ conf->shuffleCompressionCodec() != protocol::CompressionCodec::NONE),
+ compressor_(
+ shuffleCompressionEnabled_
+ ? compress::Compressor::createCompressor(*conf)
+ : nullptr) {
Review Comment:
Constructor initializer list order doesn’t match member declaration order in `ShuffleClientImpl` (new `shuffleCompressionEnabled_`/`compressor_` are declared before `conf_`, but `conf_` appears earlier in the init list). Because C++ initializes members in declaration order, this can trigger `-Wreorder` warnings (and potentially build failures under `-Werror`). Reorder the initializer list to match the header’s member order (or reorder the member declarations).
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org