You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "ReemaAlzaid (via GitHub)" <gi...@apache.org> on 2026/06/16 10:52:17 UTC
[PR] [GLUTEN-12303][VL] Support async multipart upload for S3 writes [gluten]
ReemaAlzaid opened a new pull request, #12305:
URL: https://github.com/apache/gluten/pull/12305
## What changes were proposed in this pull request?
This PR adds Gluten-side support for multi-threaded asynchronous multipart upload for S3 compatible object storage in the Velox backend.
When enabled, Gluten uses an async S3 write path that uploads multipart upload parts through a shared upload thread pool. The number of in-flight part uploads per file and the upload thread pool size are configurable.
The existing synchronous S3 write path remains the default behavior.
This PR also adds a benchmark for comparing synchronous and asynchronous S3 multipart upload performance.
https://github.com/apache/gluten/issues/12303
## How was this patch tested?
Built the Velox backend with S3 enabled and ran the S3 async upload benchmark against MinIO.
Example benchmark command:
```bash
export AWS_ACCESS_KEY_ID=minioadmin
export AWS_SECRET_ACCESS_KEY=minioadmin
export GLUTEN_S3_BENCH_BUCKET=writedata
export GLUTEN_S3_BENCH_ENDPOINT=http://127.0.0.1:9000
export GLUTEN_S3_BENCH_REGION=us-east-1
export GLUTEN_S3_BENCH_PATH_STYLE_ACCESS=true
export GLUTEN_S3_BENCH_SSL_ENABLED=false
export GLUTEN_S3_BENCH_MAX_CONCURRENCY=8
export GLUTEN_S3_BENCH_UPLOAD_THREADS=8
export GLUTEN_S3_BENCH_MIN_PART_SIZE=32MB
./cpp/build/velox/benchmarks/s3_async_upload_benchmark \
--bm_min_iters=1 \
--bm_regex='(sync|async)_upload_(16|32|64|128|256|512|1024)M'
============================================================================
[...]/benchmarks/S3AsyncUploadBenchmark.cc relative time/iter iters/s
============================================================================
sync_upload_16M 80.58ms 12.41
async_upload_16M 128.28% 62.81ms 15.92
sync_upload_32M 117.05ms 8.54
async_upload_32M 93.949% 124.59ms 8.03
sync_upload_64M 204.63ms 4.89
async_upload_64M 132.00% 155.03ms 6.45
sync_upload_128M 360.72ms 2.77
async_upload_128M 135.32% 266.56ms 3.75
sync_upload_256M 709.47ms 1.41
async_upload_256M 203.33% 348.93ms 2.87
sync_upload_512M 1.50s 667.72m
async_upload_512M 237.77% 629.87ms 1.59
sync_upload_1024M 2.90s 345.41m
async_upload_1024M 313.82% 922.55ms 1.08
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12303][VL] Support async multipart upload for S3 writes [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12305:
URL: https://github.com/apache/gluten/pull/12305#discussion_r3508735122
##########
cpp/velox/filesystem/GlutenS3FileSystem.cc:
##########
@@ -29,6 +66,460 @@ namespace filesystems = facebook::velox::filesystems;
namespace {
+using namespace facebook::velox::filesystems;
+
+constexpr std::string_view kPartUploadAsync{"part-upload-async"};
+constexpr std::string_view kPartUploadAsyncLegacy{"uploadPartAsync"};
+constexpr std::string_view kMaxConcurrentUploadNum{"max-concurrent-upload-num"};
+constexpr std::string_view kUploadThreads{"upload-threads"};
+constexpr uint32_t kDefaultMaxConcurrentUploadNum{4};
+constexpr uint32_t kDefaultUploadThreads{16};
+constexpr const char* kApplicationOctetStream = "application/octet-stream";
+
+std::string baseS3ConfigKey(std::string_view suffix) {
+ std::string key(filesystems::S3Config::kS3Prefix);
+ key.append(suffix);
+ return key;
+}
+
+std::string bucketS3ConfigKey(std::string_view bucketName, std::string_view suffix) {
+ std::string key(filesystems::S3Config::kS3BucketPrefix);
+ key.append(bucketName);
+ key.append(".");
+ key.append(suffix);
+ return key;
+}
+
+std::optional<std::string> getS3ConfigValue(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix) {
+ if (auto value = config->get<std::string>(bucketS3ConfigKey(bucketName, suffix))) {
+ return value;
+ }
+ return config->get<std::string>(baseS3ConfigKey(suffix));
+}
+
+std::optional<std::string> getS3ConfigValue(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix,
+ std::string_view legacySuffix) {
+ auto value = getS3ConfigValue(config, bucketName, suffix);
+ if (value.has_value()) {
+ return value;
+ }
+ return getS3ConfigValue(config, bucketName, legacySuffix);
+}
+
+uint32_t getUInt32S3Config(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix,
+ uint32_t defaultValue) {
+ const auto value = getS3ConfigValue(config, bucketName, suffix);
+ return value.has_value() ? folly::to<uint32_t>(value.value()) : defaultValue;
+}
+
+// Supported values are "Always", "RequestDependent", "Never"(default).
+Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy inferPayloadSign(std::string sign) {
+ std::transform(sign.begin(), sign.end(), sign.begin(), [](unsigned char c) { return std::toupper(c); });
+ if (sign == "ALWAYS") {
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always;
+ }
+ if (sign == "REQUESTDEPENDENT") {
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::RequestDependent;
+ }
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never;
+}
+
+std::optional<std::shared_ptr<Aws::Client::RetryStrategy>> getRetryStrategy(
+ const std::shared_ptr<filesystems::S3Config>& s3Config) {
+ auto retryMode = s3Config->retryMode();
+ auto maxAttempts = s3Config->maxAttempts();
+ if (!retryMode.has_value()) {
+ return std::nullopt;
+ }
+
+ if (retryMode.value() == "standard") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::StandardRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::StandardRetryStrategy>();
+ }
+
+ if (retryMode.value() == "adaptive") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::AdaptiveRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::AdaptiveRetryStrategy>();
+ }
+
+ if (retryMode.value() == "legacy") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::DefaultRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::DefaultRetryStrategy>();
+ }
+
+ VELOX_USER_FAIL("Invalid retry mode for S3: {}", retryMode.value());
+ return std::nullopt;
+}
+
+std::shared_ptr<Aws::Auth::AWSCredentialsProvider> getCredentialsProvider(const filesystems::S3Config& s3Config) {
+ VELOX_USER_CHECK(
+ !s3Config.credentialsProvider().has_value(),
+ "Gluten async S3 multipart upload does not support custom AWS credentials providers yet.");
+
+ auto accessKey = s3Config.accessKey();
+ auto secretKey = s3Config.secretKey();
+ const auto iamRole = s3Config.iamRole();
+
+ int keyCount = accessKey.has_value() + secretKey.has_value();
+ VELOX_USER_CHECK(keyCount != 1, "Invalid configuration: both access key and secret key must be specified");
+
+ int configCount =
+ (accessKey.has_value() && secretKey.has_value()) + iamRole.has_value() + s3Config.useInstanceCredentials();
+ VELOX_USER_CHECK(
+ configCount <= 1,
+ "Invalid configuration: specify only one among 'access/secret keys', 'use instance credentials', 'IAM role'");
+
+ if (accessKey.has_value() && secretKey.has_value()) {
+ return std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(
+ filesystems::awsString(accessKey.value()), filesystems::awsString(secretKey.value()));
+ }
+
+ if (s3Config.useInstanceCredentials()) {
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+ }
+
+ if (iamRole.has_value()) {
+ return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(
+ filesystems::awsString(iamRole.value()), filesystems::awsString(s3Config.iamRoleSessionName()));
+ }
+
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+}
+
+std::shared_ptr<Aws::S3::S3Client> createWriteClient(const std::shared_ptr<filesystems::S3Config>& s3Config) {
+ Aws::Client::ClientConfigurationInitValues initValues;
+ initValues.shouldDisableIMDS = !s3Config->useIMDS();
+ Aws::S3::S3ClientConfiguration clientConfig(initValues);
+ clientConfig.checksumConfig.requestChecksumCalculation = Aws::Client::RequestChecksumCalculation::WHEN_REQUIRED;
+ clientConfig.checksumConfig.responseChecksumValidation = Aws::Client::ResponseChecksumValidation::WHEN_REQUIRED;
+
+ if (s3Config->endpoint().has_value()) {
+ clientConfig.endpointOverride = s3Config->endpoint().value();
+ }
+ if (s3Config->endpointRegion().has_value()) {
+ clientConfig.region = s3Config->endpointRegion().value();
+ }
+ if (s3Config->useProxyFromEnv()) {
+ auto proxyConfig =
+ filesystems::S3ProxyConfigurationBuilder(s3Config->endpoint().has_value() ? s3Config->endpoint().value() : "")
+ .useSsl(s3Config->useSSL())
+ .build();
+ if (proxyConfig.has_value()) {
+ clientConfig.proxyScheme = Aws::Http::SchemeMapper::FromString(proxyConfig.value().scheme().c_str());
+ clientConfig.proxyHost = filesystems::awsString(proxyConfig.value().host());
+ clientConfig.proxyPort = proxyConfig.value().port();
+ clientConfig.proxyUserName = filesystems::awsString(proxyConfig.value().username());
+ clientConfig.proxyPassword = filesystems::awsString(proxyConfig.value().password());
+ }
+ }
+
+ clientConfig.scheme = s3Config->useSSL() ? Aws::Http::Scheme::HTTPS : Aws::Http::Scheme::HTTP;
+
+ if (s3Config->connectTimeout().has_value()) {
+ clientConfig.connectTimeoutMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+ velox::config::toDuration(s3Config->connectTimeout().value()))
+ .count();
+ }
+ if (s3Config->socketTimeout().has_value()) {
+ clientConfig.requestTimeoutMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+ velox::config::toDuration(s3Config->socketTimeout().value()))
+ .count();
+ }
+ if (s3Config->maxConnections().has_value()) {
+ clientConfig.maxConnections = s3Config->maxConnections().value();
+ }
+
+ auto retryStrategy = getRetryStrategy(s3Config);
+ if (retryStrategy.has_value()) {
+ clientConfig.retryStrategy = retryStrategy.value();
+ }
+
+ clientConfig.useVirtualAddressing = s3Config->useVirtualAddressing();
+ clientConfig.payloadSigningPolicy = inferPayloadSign(s3Config->payloadSigningPolicy());
+
+ return std::make_shared<Aws::S3::S3Client>(
+ getCredentialsProvider(*s3Config), nullptr /* endpointProvider */, clientConfig);
+}
+
+class GlutenS3WriteFile : public velox::WriteFile {
+ public:
+ GlutenS3WriteFile(
+ std::string_view path,
+ Aws::S3::S3Client* client,
+ velox::memory::MemoryPool* pool,
+ size_t minPartSize,
+ uint32_t maxConcurrentUploadNum,
+ uint32_t uploadThreads)
+ : client_(client),
+ pool_(pool),
+ minPartSize_(minPartSize),
+ uploadThrottle_(std::make_unique<folly::ThrottledLifoSem>(maxConcurrentUploadNum)),
+ uploadThreadPool_(uploadThreadPool(uploadThreads)) {
+ VELOX_CHECK_NOT_NULL(client_);
+ VELOX_CHECK_NOT_NULL(pool_);
+ filesystems::getBucketAndKeyFromPath(path, bucket_, key_);
+ currentPart_ = std::make_unique<velox::dwio::common::DataBuffer<char>>(*pool_);
+ currentPart_->reserve(minPartSize_);
+ ensureObjectDoesNotExist();
+ createBucketIfMissing();
+ fileSize_ = 0;
+ }
+
+ ~GlutenS3WriteFile() override {
+ if (!uploadFutures_.empty()) {
+ try {
+ waitForAsyncUploads();
+ } catch (const std::exception& e) {
+ LOG(ERROR) << "Failed while waiting for S3 async uploads: " << e.what();
+ }
+ }
+ }
+
+ void append(std::string_view data) override {
+ VELOX_CHECK(!closed(), "File is closed");
+ if (data.size() + currentPart_->size() >= minPartSize_) {
+ if (uploadState_.partNumber == 0) {
+ createMultipartUploadRequest();
+ }
+ upload(data);
+ } else {
+ currentPart_->unsafeAppend(data.data(), data.size());
+ }
+ fileSize_ += data.size();
+ }
+
+ void flush() override {
+ VELOX_CHECK(!closed(), "File is closed");
+ VELOX_CHECK_LT(currentPart_->size(), minPartSize_);
+ }
+
+ void close() override {
+ if (closed()) {
+ return;
+ }
+ if (uploadState_.partNumber == 0) {
+ putObjectRequest();
+ currentPart_->clear();
+ return;
+ }
+
+ RECORD_METRIC_VALUE(filesystems::kMetricS3StartedUploads);
+ uploadPart({currentPart_->data(), currentPart_->size()}, true);
+ waitForAsyncUploads();
+ VELOX_CHECK_EQ(uploadState_.partNumber, uploadState_.completedParts.size());
+ completeMultipartUpload();
+ currentPart_->clear();
+ }
+
+ uint64_t size() const override {
+ return fileSize_;
+ }
+
+ private:
+ struct UploadState {
+ Aws::Vector<Aws::S3::Model::CompletedPart> completedParts;
+ int64_t partNumber = 0;
+ Aws::String id;
+ };
+
+ static std::shared_ptr<folly::CPUThreadPoolExecutor> uploadThreadPool(uint32_t uploadThreads) {
+ std::lock_guard<std::mutex> l(uploadThreadPoolMutex_);
+ if (!sharedUploadThreadPool_) {
+ sharedUploadThreadPool_ = std::make_shared<folly::CPUThreadPoolExecutor>(
+ uploadThreads, std::make_shared<folly::NamedThreadFactory>("s3-upload-thread"));
+ }
+ return sharedUploadThreadPool_;
+ }
+
+ bool closed() const {
+ return currentPart_->capacity() == 0;
+ }
+
+ void ensureObjectDoesNotExist() {
+ Aws::S3::Model::HeadObjectRequest request;
+ request.SetBucket(filesystems::awsString(bucket_));
+ request.SetKey(filesystems::awsString(key_));
+ RECORD_METRIC_VALUE(filesystems::kMetricS3MetadataCalls);
+ auto objectMetadata = client_->HeadObject(request);
+ if (!objectMetadata.IsSuccess()) {
+ RECORD_METRIC_VALUE(filesystems::kMetricS3GetMetadataErrors);
+ }
+ RECORD_METRIC_VALUE(filesystems::kMetricS3GetObjectRetries, objectMetadata.GetRetryCount());
+ VELOX_CHECK(!objectMetadata.IsSuccess(), "S3 object already exists: bucket={}, key={}", bucket_, key_);
+ }
+
+ void createBucketIfMissing() {
+ Aws::S3::Model::HeadBucketRequest request;
+ request.SetBucket(filesystems::awsString(bucket_));
+ auto bucketMetadata = client_->HeadBucket(request);
+ if (bucketMetadata.IsSuccess()) {
+ return;
+ }
+
+ Aws::S3::Model::CreateBucketRequest createRequest;
+ createRequest.SetBucket(filesystems::awsString(bucket_));
+ auto outcome = client_->CreateBucket(createRequest);
+ VELOX_CHECK_AWS_OUTCOME(outcome, "Failed to create S3 bucket", bucket_, "");
+ }
+
+ void createMultipartUploadRequest() {
+ Aws::S3::Model::CreateMultipartUploadRequest request;
+ request.SetBucket(filesystems::awsString(bucket_));
+ request.SetKey(filesystems::awsString(key_));
+ request.SetContentType(kApplicationOctetStream);
+ auto outcome = client_->CreateMultipartUpload(request);
+ VELOX_CHECK_AWS_OUTCOME(outcome, "Failed initiating multiple part upload", bucket_, key_);
+ uploadState_.id = outcome.GetResult().GetUploadId();
+ }
+
+ void putObjectRequest() {
+ Aws::S3::Model::PutObjectRequest request;
+ request.SetBucket(filesystems::awsString(bucket_));
+ request.SetKey(filesystems::awsString(key_));
+ request.SetContentType(kApplicationOctetStream);
+ request.SetContentLength(currentPart_->size());
+ request.SetBody(std::make_shared<filesystems::StringViewStream>(currentPart_->data(), currentPart_->size()));
+ RECORD_METRIC_VALUE(filesystems::kMetricS3StartedUploads);
+ auto outcome = client_->PutObject(request);
+ if (outcome.IsSuccess()) {
+ RECORD_METRIC_VALUE(filesystems::kMetricS3SuccessfulUploads);
+ } else {
+ RECORD_METRIC_VALUE(filesystems::kMetricS3FailedUploads);
+ }
+ VELOX_CHECK_AWS_OUTCOME(outcome, "Failed single object upload", bucket_, key_);
+ }
+
+ void completeMultipartUpload() {
+ Aws::S3::Model::CompletedMultipartUpload completedUpload;
+ completedUpload.SetParts(uploadState_.completedParts);
+ Aws::S3::Model::CompleteMultipartUploadRequest request;
+ request.SetBucket(filesystems::awsString(bucket_));
+ request.SetKey(filesystems::awsString(key_));
+ request.SetUploadId(uploadState_.id);
+ request.SetMultipartUpload(std::move(completedUpload));
+
+ auto outcome = client_->CompleteMultipartUpload(request);
+ if (outcome.IsSuccess()) {
+ RECORD_METRIC_VALUE(filesystems::kMetricS3SuccessfulUploads);
+ } else {
+ RECORD_METRIC_VALUE(filesystems::kMetricS3FailedUploads);
+ }
+ VELOX_CHECK_AWS_OUTCOME(outcome, "Failed to complete multiple part upload", bucket_, key_);
+ }
+
+ void upload(std::string_view data) {
+ auto dataPtr = data.data();
+ auto dataSize = data.size();
+ auto remainingBufferSize = currentPart_->capacity() - currentPart_->size();
+ currentPart_->unsafeAppend(dataPtr, remainingBufferSize);
+ uploadPart({currentPart_->data(), currentPart_->size()});
+ dataPtr += remainingBufferSize;
+ dataSize -= remainingBufferSize;
+ while (dataSize > minPartSize_) {
+ uploadPart({dataPtr, minPartSize_});
+ dataPtr += minPartSize_;
+ dataSize -= minPartSize_;
+ }
+ currentPart_->unsafeAppend(0, dataPtr, dataSize);
+ }
Review Comment:
In upload(), the buffer holding the current part is uploaded via uploadPart(...) but never reset before more data is appended. This can re-upload the same bytes and/or append past the reserved capacity, corrupting the uploaded object.
##########
cpp/velox/filesystem/GlutenS3FileSystem.cc:
##########
@@ -29,6 +66,460 @@ namespace filesystems = facebook::velox::filesystems;
namespace {
+using namespace facebook::velox::filesystems;
+
+constexpr std::string_view kPartUploadAsync{"part-upload-async"};
+constexpr std::string_view kPartUploadAsyncLegacy{"uploadPartAsync"};
+constexpr std::string_view kMaxConcurrentUploadNum{"max-concurrent-upload-num"};
+constexpr std::string_view kUploadThreads{"upload-threads"};
+constexpr uint32_t kDefaultMaxConcurrentUploadNum{4};
+constexpr uint32_t kDefaultUploadThreads{16};
+constexpr const char* kApplicationOctetStream = "application/octet-stream";
+
+std::string baseS3ConfigKey(std::string_view suffix) {
+ std::string key(filesystems::S3Config::kS3Prefix);
+ key.append(suffix);
+ return key;
+}
+
+std::string bucketS3ConfigKey(std::string_view bucketName, std::string_view suffix) {
+ std::string key(filesystems::S3Config::kS3BucketPrefix);
+ key.append(bucketName);
+ key.append(".");
+ key.append(suffix);
+ return key;
+}
+
+std::optional<std::string> getS3ConfigValue(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix) {
+ if (auto value = config->get<std::string>(bucketS3ConfigKey(bucketName, suffix))) {
+ return value;
+ }
+ return config->get<std::string>(baseS3ConfigKey(suffix));
+}
+
+std::optional<std::string> getS3ConfigValue(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix,
+ std::string_view legacySuffix) {
+ auto value = getS3ConfigValue(config, bucketName, suffix);
+ if (value.has_value()) {
+ return value;
+ }
+ return getS3ConfigValue(config, bucketName, legacySuffix);
+}
+
+uint32_t getUInt32S3Config(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix,
+ uint32_t defaultValue) {
+ const auto value = getS3ConfigValue(config, bucketName, suffix);
+ return value.has_value() ? folly::to<uint32_t>(value.value()) : defaultValue;
+}
+
+// Supported values are "Always", "RequestDependent", "Never"(default).
+Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy inferPayloadSign(std::string sign) {
+ std::transform(sign.begin(), sign.end(), sign.begin(), [](unsigned char c) { return std::toupper(c); });
+ if (sign == "ALWAYS") {
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always;
+ }
+ if (sign == "REQUESTDEPENDENT") {
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::RequestDependent;
+ }
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never;
+}
+
+std::optional<std::shared_ptr<Aws::Client::RetryStrategy>> getRetryStrategy(
+ const std::shared_ptr<filesystems::S3Config>& s3Config) {
+ auto retryMode = s3Config->retryMode();
+ auto maxAttempts = s3Config->maxAttempts();
+ if (!retryMode.has_value()) {
+ return std::nullopt;
+ }
+
+ if (retryMode.value() == "standard") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::StandardRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::StandardRetryStrategy>();
+ }
+
+ if (retryMode.value() == "adaptive") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::AdaptiveRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::AdaptiveRetryStrategy>();
+ }
+
+ if (retryMode.value() == "legacy") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::DefaultRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::DefaultRetryStrategy>();
+ }
+
+ VELOX_USER_FAIL("Invalid retry mode for S3: {}", retryMode.value());
+ return std::nullopt;
+}
+
+std::shared_ptr<Aws::Auth::AWSCredentialsProvider> getCredentialsProvider(const filesystems::S3Config& s3Config) {
+ VELOX_USER_CHECK(
+ !s3Config.credentialsProvider().has_value(),
+ "Gluten async S3 multipart upload does not support custom AWS credentials providers yet.");
+
+ auto accessKey = s3Config.accessKey();
+ auto secretKey = s3Config.secretKey();
+ const auto iamRole = s3Config.iamRole();
+
+ int keyCount = accessKey.has_value() + secretKey.has_value();
+ VELOX_USER_CHECK(keyCount != 1, "Invalid configuration: both access key and secret key must be specified");
+
+ int configCount =
+ (accessKey.has_value() && secretKey.has_value()) + iamRole.has_value() + s3Config.useInstanceCredentials();
+ VELOX_USER_CHECK(
+ configCount <= 1,
+ "Invalid configuration: specify only one among 'access/secret keys', 'use instance credentials', 'IAM role'");
+
+ if (accessKey.has_value() && secretKey.has_value()) {
+ return std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(
+ filesystems::awsString(accessKey.value()), filesystems::awsString(secretKey.value()));
+ }
+
+ if (s3Config.useInstanceCredentials()) {
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+ }
+
+ if (iamRole.has_value()) {
+ return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(
+ filesystems::awsString(iamRole.value()), filesystems::awsString(s3Config.iamRoleSessionName()));
+ }
+
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+}
+
+std::shared_ptr<Aws::S3::S3Client> createWriteClient(const std::shared_ptr<filesystems::S3Config>& s3Config) {
+ Aws::Client::ClientConfigurationInitValues initValues;
+ initValues.shouldDisableIMDS = !s3Config->useIMDS();
+ Aws::S3::S3ClientConfiguration clientConfig(initValues);
+ clientConfig.checksumConfig.requestChecksumCalculation = Aws::Client::RequestChecksumCalculation::WHEN_REQUIRED;
+ clientConfig.checksumConfig.responseChecksumValidation = Aws::Client::ResponseChecksumValidation::WHEN_REQUIRED;
+
+ if (s3Config->endpoint().has_value()) {
+ clientConfig.endpointOverride = s3Config->endpoint().value();
+ }
+ if (s3Config->endpointRegion().has_value()) {
+ clientConfig.region = s3Config->endpointRegion().value();
+ }
+ if (s3Config->useProxyFromEnv()) {
+ auto proxyConfig =
+ filesystems::S3ProxyConfigurationBuilder(s3Config->endpoint().has_value() ? s3Config->endpoint().value() : "")
+ .useSsl(s3Config->useSSL())
+ .build();
+ if (proxyConfig.has_value()) {
+ clientConfig.proxyScheme = Aws::Http::SchemeMapper::FromString(proxyConfig.value().scheme().c_str());
+ clientConfig.proxyHost = filesystems::awsString(proxyConfig.value().host());
+ clientConfig.proxyPort = proxyConfig.value().port();
+ clientConfig.proxyUserName = filesystems::awsString(proxyConfig.value().username());
+ clientConfig.proxyPassword = filesystems::awsString(proxyConfig.value().password());
+ }
+ }
+
+ clientConfig.scheme = s3Config->useSSL() ? Aws::Http::Scheme::HTTPS : Aws::Http::Scheme::HTTP;
+
+ if (s3Config->connectTimeout().has_value()) {
+ clientConfig.connectTimeoutMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+ velox::config::toDuration(s3Config->connectTimeout().value()))
+ .count();
+ }
+ if (s3Config->socketTimeout().has_value()) {
+ clientConfig.requestTimeoutMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+ velox::config::toDuration(s3Config->socketTimeout().value()))
+ .count();
+ }
+ if (s3Config->maxConnections().has_value()) {
+ clientConfig.maxConnections = s3Config->maxConnections().value();
+ }
+
+ auto retryStrategy = getRetryStrategy(s3Config);
+ if (retryStrategy.has_value()) {
+ clientConfig.retryStrategy = retryStrategy.value();
+ }
+
+ clientConfig.useVirtualAddressing = s3Config->useVirtualAddressing();
+ clientConfig.payloadSigningPolicy = inferPayloadSign(s3Config->payloadSigningPolicy());
+
+ return std::make_shared<Aws::S3::S3Client>(
+ getCredentialsProvider(*s3Config), nullptr /* endpointProvider */, clientConfig);
+}
+
+class GlutenS3WriteFile : public velox::WriteFile {
+ public:
+ GlutenS3WriteFile(
+ std::string_view path,
+ Aws::S3::S3Client* client,
+ velox::memory::MemoryPool* pool,
+ size_t minPartSize,
+ uint32_t maxConcurrentUploadNum,
+ uint32_t uploadThreads)
+ : client_(client),
+ pool_(pool),
+ minPartSize_(minPartSize),
+ uploadThrottle_(std::make_unique<folly::ThrottledLifoSem>(maxConcurrentUploadNum)),
+ uploadThreadPool_(uploadThreadPool(uploadThreads)) {
+ VELOX_CHECK_NOT_NULL(client_);
+ VELOX_CHECK_NOT_NULL(pool_);
+ filesystems::getBucketAndKeyFromPath(path, bucket_, key_);
+ currentPart_ = std::make_unique<velox::dwio::common::DataBuffer<char>>(*pool_);
+ currentPart_->reserve(minPartSize_);
+ ensureObjectDoesNotExist();
+ createBucketIfMissing();
+ fileSize_ = 0;
+ }
+
+ ~GlutenS3WriteFile() override {
+ if (!uploadFutures_.empty()) {
+ try {
+ waitForAsyncUploads();
+ } catch (const std::exception& e) {
+ LOG(ERROR) << "Failed while waiting for S3 async uploads: " << e.what();
+ }
+ }
+ }
+
+ void append(std::string_view data) override {
+ VELOX_CHECK(!closed(), "File is closed");
+ if (data.size() + currentPart_->size() >= minPartSize_) {
+ if (uploadState_.partNumber == 0) {
+ createMultipartUploadRequest();
+ }
+ upload(data);
+ } else {
+ currentPart_->unsafeAppend(data.data(), data.size());
+ }
+ fileSize_ += data.size();
+ }
+
+ void flush() override {
+ VELOX_CHECK(!closed(), "File is closed");
+ VELOX_CHECK_LT(currentPart_->size(), minPartSize_);
+ }
+
+ void close() override {
+ if (closed()) {
+ return;
+ }
+ if (uploadState_.partNumber == 0) {
+ putObjectRequest();
+ currentPart_->clear();
+ return;
+ }
+
+ RECORD_METRIC_VALUE(filesystems::kMetricS3StartedUploads);
+ uploadPart({currentPart_->data(), currentPart_->size()}, true);
+ waitForAsyncUploads();
+ VELOX_CHECK_EQ(uploadState_.partNumber, uploadState_.completedParts.size());
+ completeMultipartUpload();
Review Comment:
If any async part upload or CompleteMultipartUpload fails, the code doesn’t appear to abort the multipart upload. This can leave incomplete multipart uploads behind (storage leakage / cleanup burden), and the destructor path only logs and swallows exceptions rather than cleaning up.
##########
cpp/velox/filesystem/GlutenS3FileSystem.cc:
##########
@@ -29,6 +66,460 @@ namespace filesystems = facebook::velox::filesystems;
namespace {
+using namespace facebook::velox::filesystems;
+
+constexpr std::string_view kPartUploadAsync{"part-upload-async"};
+constexpr std::string_view kPartUploadAsyncLegacy{"uploadPartAsync"};
+constexpr std::string_view kMaxConcurrentUploadNum{"max-concurrent-upload-num"};
+constexpr std::string_view kUploadThreads{"upload-threads"};
+constexpr uint32_t kDefaultMaxConcurrentUploadNum{4};
+constexpr uint32_t kDefaultUploadThreads{16};
+constexpr const char* kApplicationOctetStream = "application/octet-stream";
+
+std::string baseS3ConfigKey(std::string_view suffix) {
+ std::string key(filesystems::S3Config::kS3Prefix);
+ key.append(suffix);
+ return key;
+}
+
+std::string bucketS3ConfigKey(std::string_view bucketName, std::string_view suffix) {
+ std::string key(filesystems::S3Config::kS3BucketPrefix);
+ key.append(bucketName);
+ key.append(".");
+ key.append(suffix);
+ return key;
+}
+
+std::optional<std::string> getS3ConfigValue(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix) {
+ if (auto value = config->get<std::string>(bucketS3ConfigKey(bucketName, suffix))) {
+ return value;
+ }
+ return config->get<std::string>(baseS3ConfigKey(suffix));
+}
+
+std::optional<std::string> getS3ConfigValue(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix,
+ std::string_view legacySuffix) {
+ auto value = getS3ConfigValue(config, bucketName, suffix);
+ if (value.has_value()) {
+ return value;
+ }
+ return getS3ConfigValue(config, bucketName, legacySuffix);
+}
+
+uint32_t getUInt32S3Config(
+ const std::shared_ptr<const velox::config::ConfigBase>& config,
+ std::string_view bucketName,
+ std::string_view suffix,
+ uint32_t defaultValue) {
+ const auto value = getS3ConfigValue(config, bucketName, suffix);
+ return value.has_value() ? folly::to<uint32_t>(value.value()) : defaultValue;
+}
+
+// Supported values are "Always", "RequestDependent", "Never"(default).
+Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy inferPayloadSign(std::string sign) {
+ std::transform(sign.begin(), sign.end(), sign.begin(), [](unsigned char c) { return std::toupper(c); });
+ if (sign == "ALWAYS") {
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always;
+ }
+ if (sign == "REQUESTDEPENDENT") {
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::RequestDependent;
+ }
+ return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never;
+}
+
+std::optional<std::shared_ptr<Aws::Client::RetryStrategy>> getRetryStrategy(
+ const std::shared_ptr<filesystems::S3Config>& s3Config) {
+ auto retryMode = s3Config->retryMode();
+ auto maxAttempts = s3Config->maxAttempts();
+ if (!retryMode.has_value()) {
+ return std::nullopt;
+ }
+
+ if (retryMode.value() == "standard") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::StandardRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::StandardRetryStrategy>();
+ }
+
+ if (retryMode.value() == "adaptive") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::AdaptiveRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::AdaptiveRetryStrategy>();
+ }
+
+ if (retryMode.value() == "legacy") {
+ if (maxAttempts.has_value()) {
+ VELOX_USER_CHECK_GE(
+ maxAttempts.value(),
+ 0,
+ "Invalid configuration: specified 'hive.s3.max-attempts' value {} is < 0.",
+ maxAttempts.value());
+ return std::make_shared<Aws::Client::DefaultRetryStrategy>(maxAttempts.value());
+ }
+ return std::make_shared<Aws::Client::DefaultRetryStrategy>();
+ }
+
+ VELOX_USER_FAIL("Invalid retry mode for S3: {}", retryMode.value());
+ return std::nullopt;
+}
+
+std::shared_ptr<Aws::Auth::AWSCredentialsProvider> getCredentialsProvider(const filesystems::S3Config& s3Config) {
+ VELOX_USER_CHECK(
+ !s3Config.credentialsProvider().has_value(),
+ "Gluten async S3 multipart upload does not support custom AWS credentials providers yet.");
+
+ auto accessKey = s3Config.accessKey();
+ auto secretKey = s3Config.secretKey();
+ const auto iamRole = s3Config.iamRole();
+
+ int keyCount = accessKey.has_value() + secretKey.has_value();
+ VELOX_USER_CHECK(keyCount != 1, "Invalid configuration: both access key and secret key must be specified");
+
+ int configCount =
+ (accessKey.has_value() && secretKey.has_value()) + iamRole.has_value() + s3Config.useInstanceCredentials();
+ VELOX_USER_CHECK(
+ configCount <= 1,
+ "Invalid configuration: specify only one among 'access/secret keys', 'use instance credentials', 'IAM role'");
+
+ if (accessKey.has_value() && secretKey.has_value()) {
+ return std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(
+ filesystems::awsString(accessKey.value()), filesystems::awsString(secretKey.value()));
+ }
+
+ if (s3Config.useInstanceCredentials()) {
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+ }
+
+ if (iamRole.has_value()) {
+ return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(
+ filesystems::awsString(iamRole.value()), filesystems::awsString(s3Config.iamRoleSessionName()));
+ }
+
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+}
+
+std::shared_ptr<Aws::S3::S3Client> createWriteClient(const std::shared_ptr<filesystems::S3Config>& s3Config) {
+ Aws::Client::ClientConfigurationInitValues initValues;
+ initValues.shouldDisableIMDS = !s3Config->useIMDS();
+ Aws::S3::S3ClientConfiguration clientConfig(initValues);
+ clientConfig.checksumConfig.requestChecksumCalculation = Aws::Client::RequestChecksumCalculation::WHEN_REQUIRED;
+ clientConfig.checksumConfig.responseChecksumValidation = Aws::Client::ResponseChecksumValidation::WHEN_REQUIRED;
+
+ if (s3Config->endpoint().has_value()) {
+ clientConfig.endpointOverride = s3Config->endpoint().value();
+ }
+ if (s3Config->endpointRegion().has_value()) {
+ clientConfig.region = s3Config->endpointRegion().value();
+ }
+ if (s3Config->useProxyFromEnv()) {
+ auto proxyConfig =
+ filesystems::S3ProxyConfigurationBuilder(s3Config->endpoint().has_value() ? s3Config->endpoint().value() : "")
+ .useSsl(s3Config->useSSL())
+ .build();
+ if (proxyConfig.has_value()) {
+ clientConfig.proxyScheme = Aws::Http::SchemeMapper::FromString(proxyConfig.value().scheme().c_str());
+ clientConfig.proxyHost = filesystems::awsString(proxyConfig.value().host());
+ clientConfig.proxyPort = proxyConfig.value().port();
+ clientConfig.proxyUserName = filesystems::awsString(proxyConfig.value().username());
+ clientConfig.proxyPassword = filesystems::awsString(proxyConfig.value().password());
+ }
+ }
+
+ clientConfig.scheme = s3Config->useSSL() ? Aws::Http::Scheme::HTTPS : Aws::Http::Scheme::HTTP;
+
+ if (s3Config->connectTimeout().has_value()) {
+ clientConfig.connectTimeoutMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+ velox::config::toDuration(s3Config->connectTimeout().value()))
+ .count();
+ }
+ if (s3Config->socketTimeout().has_value()) {
+ clientConfig.requestTimeoutMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+ velox::config::toDuration(s3Config->socketTimeout().value()))
+ .count();
+ }
+ if (s3Config->maxConnections().has_value()) {
+ clientConfig.maxConnections = s3Config->maxConnections().value();
+ }
+
+ auto retryStrategy = getRetryStrategy(s3Config);
+ if (retryStrategy.has_value()) {
+ clientConfig.retryStrategy = retryStrategy.value();
+ }
+
+ clientConfig.useVirtualAddressing = s3Config->useVirtualAddressing();
+ clientConfig.payloadSigningPolicy = inferPayloadSign(s3Config->payloadSigningPolicy());
+
+ return std::make_shared<Aws::S3::S3Client>(
+ getCredentialsProvider(*s3Config), nullptr /* endpointProvider */, clientConfig);
+}
+
+class GlutenS3WriteFile : public velox::WriteFile {
+ public:
+ GlutenS3WriteFile(
+ std::string_view path,
+ Aws::S3::S3Client* client,
+ velox::memory::MemoryPool* pool,
+ size_t minPartSize,
+ uint32_t maxConcurrentUploadNum,
+ uint32_t uploadThreads)
+ : client_(client),
+ pool_(pool),
+ minPartSize_(minPartSize),
+ uploadThrottle_(std::make_unique<folly::ThrottledLifoSem>(maxConcurrentUploadNum)),
+ uploadThreadPool_(uploadThreadPool(uploadThreads)) {
+ VELOX_CHECK_NOT_NULL(client_);
+ VELOX_CHECK_NOT_NULL(pool_);
+ filesystems::getBucketAndKeyFromPath(path, bucket_, key_);
+ currentPart_ = std::make_unique<velox::dwio::common::DataBuffer<char>>(*pool_);
+ currentPart_->reserve(minPartSize_);
+ ensureObjectDoesNotExist();
+ createBucketIfMissing();
+ fileSize_ = 0;
+ }
+
+ ~GlutenS3WriteFile() override {
+ if (!uploadFutures_.empty()) {
+ try {
+ waitForAsyncUploads();
+ } catch (const std::exception& e) {
+ LOG(ERROR) << "Failed while waiting for S3 async uploads: " << e.what();
+ }
+ }
+ }
+
+ void append(std::string_view data) override {
+ VELOX_CHECK(!closed(), "File is closed");
+ if (data.size() + currentPart_->size() >= minPartSize_) {
+ if (uploadState_.partNumber == 0) {
+ createMultipartUploadRequest();
+ }
+ upload(data);
+ } else {
+ currentPart_->unsafeAppend(data.data(), data.size());
+ }
+ fileSize_ += data.size();
+ }
+
+ void flush() override {
+ VELOX_CHECK(!closed(), "File is closed");
+ VELOX_CHECK_LT(currentPart_->size(), minPartSize_);
+ }
+
+ void close() override {
+ if (closed()) {
+ return;
+ }
+ if (uploadState_.partNumber == 0) {
+ putObjectRequest();
+ currentPart_->clear();
+ return;
+ }
+
+ RECORD_METRIC_VALUE(filesystems::kMetricS3StartedUploads);
+ uploadPart({currentPart_->data(), currentPart_->size()}, true);
+ waitForAsyncUploads();
+ VELOX_CHECK_EQ(uploadState_.partNumber, uploadState_.completedParts.size());
+ completeMultipartUpload();
+ currentPart_->clear();
+ }
+
+ uint64_t size() const override {
+ return fileSize_;
+ }
+
+ private:
+ struct UploadState {
+ Aws::Vector<Aws::S3::Model::CompletedPart> completedParts;
+ int64_t partNumber = 0;
+ Aws::String id;
+ };
+
+ static std::shared_ptr<folly::CPUThreadPoolExecutor> uploadThreadPool(uint32_t uploadThreads) {
+ std::lock_guard<std::mutex> l(uploadThreadPoolMutex_);
+ if (!sharedUploadThreadPool_) {
+ sharedUploadThreadPool_ = std::make_shared<folly::CPUThreadPoolExecutor>(
+ uploadThreads, std::make_shared<folly::NamedThreadFactory>("s3-upload-thread"));
Review Comment:
uploadThreads is read per bucket/file, but the implementation uses a single static shared thread pool that is created only once. If different buckets (or even the same app in different runs) specify different hive.s3.upload-threads values, only the first value will take effect and the rest will be silently ignored.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12303][VL] Support async multipart upload for S3 writes [gluten]
Posted by "FelixYBW (via GitHub)" <gi...@apache.org>.
FelixYBW commented on PR #12305:
URL: https://github.com/apache/gluten/pull/12305#issuecomment-4813982071
FYI, @weixiuli. Since the PR isn't imported into Velox. We extend Velox's S3 file system and move your enhancement into Gluten.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12303][VL] Support async multipart upload for S3 writes [gluten]
Posted by "zhouyuan (via GitHub)" <gi...@apache.org>.
zhouyuan commented on PR #12305:
URL: https://github.com/apache/gluten/pull/12305#issuecomment-4842053204
@ReemaAlzaid the configuration issue should be re-generated by https://github.com/apache/gluten/blob/main/dev/gen-all-config-docs.sh
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12303][VL] Support async multipart upload for S3 writes [gluten]
Posted by "FelixYBW (via GitHub)" <gi...@apache.org>.
FelixYBW merged PR #12305:
URL: https://github.com/apache/gluten/pull/12305
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org