github-actions[bot] commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3705920258
##########
fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureObjStorage.java:
##########
@@ -524,4 +495,15 @@ private static String toBlockId(int partNum) {
byte[] bytes =
ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(partNum).array();
return Base64.getEncoder().encodeToString(bytes);
}
+
+ static String multipartBlockId(String uploadId, int partNum) {
+ int uploadNamespace = 0x811C9DC5;
+ for (byte value : uploadId.getBytes(StandardCharsets.UTF_8)) {
+ uploadNamespace = (uploadNamespace ^ (value & 0xFF)) * 0x01000193;
+ }
+ int namespacedPart = uploadNamespace + partNum;
Review Comment:
[P1] The lease fix is still incomplete because the reacquire branch is
entered after this upload's generation fence has already been lost. Azure
permits an expired lease to renew with its old ID when the blob has not been
modified or leased since expiry. If that renewal returns 409/412, session B may
already have acquired, staged or committed this collision pair, and released;
A's unconditional `acquireLease(60)` then succeeds and its `Latest` block list
can adopt B's staged or committed bytes. The C++ path has the same fallback.
This also means an ETag-only reacquire is insufficient when B staged
colliding uncommitted blocks and released without committing. Please fail
closed on any old-ID renewal failure in both implementations, and add
A-expire/B-stage-or-commit-and-release/A-resume tests using the checked-in
collision pair. The relevant service contracts are [Lease
Blob](https://learn.microsoft.com/en-us/rest/api/storageservices/lease-blob)
and [Put Block
List](https://learn.microsoft.com/en-us/rest/api/storageservices/put-block-list).
##########
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:
[P1] The exact sizing fix is sound, but an OK/failed RPC is not yet an
ownership-safe coordinator acknowledgement. The ambiguity goes both ways:
1. FE can mark the pipeline done, feed `iceberg_commit_datas`, and unblock
the external commit, then lose the response. If BE cannot reopen/retry,
`finalize_iceberg_report_cleanup(false)` deletes files that FE can still commit
into metadata.
2. `QeProcessorImpl.reportExecStatus` returns OK when the coordinator is
absent and catches `updateFragmentExecStatus` exceptions while retaining OK. BE
then clears its callbacks even if the vectors were never accepted; because
`done` is set before `CommitDataSerializer.feed`, a retry after a partial
exception is ignored and cannot repair the handoff.
Please make acceptance explicit and idempotent only after all external
commit vectors are durably owned, return a real rejection for missing/failed
handlers, and keep ambiguous transport outcomes recoverable or queryable
instead of eagerly deleting. Tests should cover both
consumed-request/lost-response and false-OK-before-feed paths.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +55,60 @@ size_t
SpillIcebergTableSinkLocalState::get_reserve_mem_size(RuntimeState* state
if (!_writer) {
return 0;
}
- auto current_writer = _writer->current_writer();
- auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
- if (!sort_writer) {
- return 0;
+ std::vector<IcebergSorterReserveMemory> per_partition_reservations;
+ auto active_writers = _writer->active_writers();
Review Comment:
[P1] The cold-writer allowance is not actually tied to the incoming block.
The production call passes `minimum_operator_memory_required_bytes()` to the
helper's `incoming_block_bytes` parameter, but that setting is a fixed
execution floor (about 31.25 MiB by default), not an upper bound on the block
already held by `PipelineTask`. A legal larger variable-width block is admitted
with only that floor while the active-writer snapshot is empty; partition
selection and `FullSorter::append_block` can then hold selected and retained
column copies before the first writer is published.
The helper test injects an artificial byte count directly, so it cannot
catch this production wiring mismatch. Please feed an upper bound derived from
the current block into admission (including cold partition/sorter workspace)
and add a production-path test whose first block exceeds
`minimum_operator_memory_required_bytes()`.
--
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]
