Gabriel39 commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3698475844
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -216,35 +237,78 @@ ObjectStorageResponse
AzureObjStorageClient::put_object(const ObjectStoragePathO
ObjectStorageUploadResponse AzureObjStorageClient::upload_part(const
ObjectStoragePathOptions& opts,
std::string_view stream,
int part_num) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto client =
_client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ std::string block_id = azure_block_id(opts, part_num);
auto resp = do_azure_client_call(
[&]() {
Azure::Core::IO::MemoryBodyStream memory_body(
reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
// The blockId must be base64 encoded
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.StageBlock(base64_encode_part_num(part_num),
memory_body);
+ client.StageBlock(block_id, memory_body);
+ if (opts.deferred_completion) {
Review Comment:
Fixed by removing the legacy mirror entirely. BE now stages only
upload-scoped block IDs on the target blob, and FE commits the exact IDs
reported by that upload. A new BE with an old FE therefore fails closed instead
of sharing deterministic legacy IDs and silently mixing writers. The concurrent
same-key test now verifies that the losing completion fails and the first
committed value remains visible.
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -216,35 +237,78 @@ ObjectStorageResponse
AzureObjStorageClient::put_object(const ObjectStoragePathO
ObjectStorageUploadResponse AzureObjStorageClient::upload_part(const
ObjectStoragePathOptions& opts,
std::string_view stream,
int part_num) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto client =
_client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ std::string block_id = azure_block_id(opts, part_num);
auto resp = do_azure_client_call(
[&]() {
Azure::Core::IO::MemoryBodyStream memory_body(
reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
// The blockId must be base64 encoded
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.StageBlock(base64_encode_part_num(part_num),
memory_body);
+ client.StageBlock(block_id, memory_body);
+ if (opts.deferred_completion) {
+ // During rolling upgrades an old FE still commits
deterministic IDs on the final blob.
+ Azure::Core::IO::MemoryBodyStream legacy_body(
+ reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
+ _client->GetBlockBlobClient(opts.key).StageBlock(
Review Comment:
Fixed together with the compatibility isolation issue: the second StageBlock
transfer was removed. Each part is now transferred exactly once, so the
existing rate-limited upload_part accounting matches the bytes sent.
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -216,35 +237,78 @@ ObjectStorageResponse
AzureObjStorageClient::put_object(const ObjectStoragePathO
ObjectStorageUploadResponse AzureObjStorageClient::upload_part(const
ObjectStoragePathOptions& opts,
std::string_view stream,
int part_num) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto client =
_client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ std::string block_id = azure_block_id(opts, part_num);
auto resp = do_azure_client_call(
[&]() {
Azure::Core::IO::MemoryBodyStream memory_body(
reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
// The blockId must be base64 encoded
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.StageBlock(base64_encode_part_num(part_num),
memory_body);
+ client.StageBlock(block_id, memory_body);
+ if (opts.deferred_completion) {
+ // During rolling upgrades an old FE still commits
deterministic IDs on the final blob.
+ Azure::Core::IO::MemoryBodyStream legacy_body(
+ reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
+ _client->GetBlockBlobClient(opts.key).StageBlock(
+ legacy_azure_block_id(part_num), legacy_body);
+ }
},
opts, _tls_debug_context);
return ObjectStorageUploadResponse {
.resp = resp,
+ // Hive defers completion to FE, so the exact staged ID must cross
that boundary.
+ .etag = block_id,
};
}
ObjectStorageResponse AzureObjStorageClient::complete_multipart_upload(
const ObjectStoragePathOptions& opts,
const std::vector<ObjectCompleteMultiPart>& completed_parts) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto temp_client =
+ _client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ auto target_client = _client->GetBlockBlobClient(opts.key);
std::vector<std::string> string_block_ids;
std::ranges::transform(
completed_parts, std::back_inserter(string_block_ids),
- [](const ObjectCompleteMultiPart& i) { return
base64_encode_part_num(i.part_num); });
- return do_azure_client_call(
+ [&opts](const ObjectCompleteMultiPart& i) { return
azure_block_id(opts, i.part_num); });
+ auto response = do_azure_client_call(
[&]() {
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.CommitBlockList(string_block_ids);
+ // Per-upload temporary blobs keep Azure's blob-wide
staged-block namespace isolated.
+ temp_client.CommitBlockList(string_block_ids);
+ auto copy =
target_client.StartCopyFromUri(temp_client.GetUrl());
Review Comment:
Fixed by removing Copy Blob from both BE and FE completion. Parts are staged
as upload-scoped blocks on the final blob, and Put Block List is the single
atomic publication point. A failed or conflicting commit does not replace the
prior committed value; the same-key integration test verifies that the first
value remains after the second completion fails.
##########
be/src/runtime/runtime_state.cpp:
##########
@@ -52,12 +52,37 @@
#include "runtime/thread_context.h"
#include "storage/id_manager.h"
#include "storage/storage_engine.h"
+#include "util/thrift_util.h"
#include "util/timezone_utils.h"
#include "util/uid_util.h"
namespace doris {
using namespace ErrorCode;
+Status RuntimeState::add_iceberg_commit_datas(TIcebergCommitData
iceberg_commit_data) {
+ ThriftSerializer serializer(false, 256);
+ uint32_t serialized_size = 0;
+ uint8_t* buffer = nullptr;
+ RETURN_IF_ERROR(serializer.serialize(&iceberg_commit_data,
&serialized_size, &buffer));
+
+ constexpr size_t report_envelope_headroom = 1024 * 1024;
+ const size_t thrift_limit =
static_cast<size_t>(std::max(config::thrift_max_message_size, 0));
Review Comment:
Fixed by adding coordinator_thrift_max_message_size to TQueryOptions. FE
sends its receiver limit, BE budgets against the smaller FE/BE value, and an
old FE that omits the field retains the local fallback. The asymmetric
FE-lower-than-BE unit test and FE propagation test both pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
