Gabriel39 commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3703975920
##########
fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureObjStorage.java:
##########
@@ -230,7 +232,7 @@ public UploadPartResult uploadPart(String remotePath,
String uploadId, int partN
AzureUri uri = AzureUri.parse(remotePath);
BlockBlobClient blockBlobClient =
getClient().getBlobContainerClient(uri.container())
.getBlobClient(uri.key()).getBlockBlobClient();
- String blockId = toBlockId(partNum);
+ String blockId = multipartBlockId(uploadId, partNum);
Review Comment:
Fixed across 149613cede and af29c53bf8. Azure block IDs retain the legacy
four-byte decoded length, and the regression seeds the legacy residual format
before staging a new upload. Exact staged IDs are carried to the new
coordinator, so completion never guesses a legacy ID.
##########
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 across ccac4295e3 and af29c53bf8. The 32-bit namespace is no longer
treated as writer identity: every upload holds a target blob lease, all stage
and commit operations are lease-fenced, and an expired lease can be reacquired
with the same proposed ID only when no competing writer owns the target. The
collision pair remains in the regression to prove that the lease, not the hash,
supplies isolation.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +55,62 @@ 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();
+ per_partition_reservations.reserve(active_writers->size());
+ for (const auto& writer : *active_writers) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+ auto reservation =
sort_writer->get_reserve_mem_size_components(state, eos);
Review Comment:
Fixed in 6d95c1fd22 and covered again by af29c53bf8. Iceberg sorted writers
use state-dependent admission with at most one block outstanding; the producer
cannot reserve a later block until the dequeued block has updated sorter state.
The deterministic admission regression verifies that dequeue alone does not
reopen the producer.
--
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]
