Gabriel39 commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3709304892
##########
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:
Fixed in 861f4b884c. Both Java and C++ now only renew the original lease and
propagate every renewal failure; the reacquisition branches and helper were
removed entirely. The Java regression executes the concrete collision
interleaving (B stages the colliding block and releases its lease, then stale A
receives a renewal conflict) and verifies that A neither reacquires nor
commits. The full 26-test Azure extension suite passes.
##########
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 in 861f4b884c. The report result now carries an explicit external-file
acceptance bit, and new BE writers require that capability before creating
Iceberg files. FE sets acceptance only after the complete commit vector is
serialized and owned; missing coordinators, missing handlers, incomplete final
reports, and handler exceptions return a real rejection without marking the
task done. Accepted report identities remain retryable after coordinator
removal, while BE distinguishes ACKNOWLEDGED, REJECTED, and AMBIGUOUS outcomes
so a consumed request with a lost response retains its files. The 29 focused FE
tests and 14 focused BE ASAN tests pass.
##########
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:
Fixed in 861f4b884c. PipelineTask now passes the current Block into sink
admission. Cold Iceberg admission derives a saturating upper bound from the
block allocated bytes, two possible selected/retained copies, row-index
storage, and writer workspace instead of using the fixed operator floor as the
incoming-block estimate. The regression constructs a real 40 MiB variable-width
first block, verifies that it exceeds the 32 MiB floor, and confirms the larger
reservation. The focused BE ASAN suite passes.
--
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]
