Gabriel39 commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3800769434
##########
be/src/exec/sort/sorter.cpp:
##########
@@ -184,35 +199,77 @@ bool FullSorter::has_enough_capacity(Block* input_block,
Block* unsorted_block)
}
size_t FullSorter::get_reserve_mem_size(RuntimeState* state, bool eos) const {
- size_t size_to_reserve = 0;
+ return get_reserve_mem_size_components(state, eos).total();
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state,
+ bool eos)
const {
+ return get_reserve_mem_size_components(state, eos,
std::numeric_limits<size_t>::max());
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
sort_threshold_bytes) const {
+ const auto rows = _state->unsorted_block()->rows();
+ const auto bytes = _state->unsorted_block()->bytes();
+ const auto bytes_per_row = rows == 0 ? 0 : bytes / rows;
+ return get_reserve_mem_size_components(
+ state, eos, state->batch_size(),
+ saturating_multiply_size(bytes_per_row, state->batch_size()),
sort_threshold_bytes);
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
incoming_rows,
+ size_t
incoming_bytes) const {
+ return get_reserve_mem_size_components(state, eos, incoming_rows,
incoming_bytes,
+ std::numeric_limits<size_t>::max());
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
incoming_rows,
+ size_t
incoming_bytes,
+ size_t
sort_threshold_bytes) const {
+ SorterReserveMemory reserve;
const auto rows = _state->unsorted_block()->rows();
if (rows != 0) {
const auto bytes = _state->unsorted_block()->bytes();
const auto allocated_bytes =
_state->unsorted_block()->allocated_bytes();
- const auto bytes_per_row = bytes / rows;
- const auto estimated_size_of_next_block = bytes_per_row *
state->batch_size();
- auto new_block_bytes = estimated_size_of_next_block + bytes;
- auto new_rows = rows + state->batch_size();
+ auto new_block_bytes = saturating_add_size(bytes, incoming_bytes);
+ auto new_rows = saturating_add_size(rows, incoming_rows);
// If the new size is greater than 85% of allocalted bytes, it maybe
need to realloc.
- if ((new_block_bytes * 100 / allocated_bytes) >= 85) {
- size_to_reserve += (size_t)(allocated_bytes * 1.15);
+ const auto growth_threshold = static_cast<size_t>(
+ (static_cast<unsigned __int128>(allocated_bytes) * 85 + 99) /
100);
+ const size_t growth_trigger_bytes = growth_threshold > bytes ?
growth_threshold - bytes : 0;
+ if (incoming_rows > 0 && growth_trigger_bytes <= incoming_bytes) {
+ reserve.retained_growth = static_cast<size_t>(std::min<unsigned
__int128>(
+ (static_cast<unsigned __int128>(allocated_bytes) * 115 +
99) / 100,
+ std::numeric_limits<size_t>::max()));
+ reserve.retained_growth_trigger_bytes = growth_trigger_bytes;
}
- auto sort = new_rows > _buffered_block_size || new_block_bytes >
_buffered_block_bytes;
+ // Iceberg close forces every nonempty pending run to sort at EOS,
even when the generic
+ // append thresholds are not reached, so admission must cover that
final allocation too.
+ // The reservation must mirror every caller-side rollover that
immediately invokes do_sort().
+ auto sort = (eos && new_rows > 0) || new_rows > _buffered_block_size ||
Review Comment:
Fixed by reserving transient sort workspace for every non-empty append once
FullSorter reaches its retained-capacity threshold. This conservatively covers
per-column pre-append rollover when the post-projection column distribution is
unavailable. Added a below-threshold capacity-rollover unit test.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2235,14 +2224,40 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
rpc_status = Status::create<false>(res.status);
} catch (apache::thrift::TException& e) {
+ report_outcome_ambiguous = true;
rpc_status = Status::InternalError("ReportExecStatus() to {} failed:
{}",
PrintThriftNetworkAddress(req.coord_addr), e.what());
}
+ // Iceberg requires the explicit new-protocol ACK. Legacy Hive/Paimon
coordinators transfer
+ // ownership through RPC success, which remains valid during a rolling FE
upgrade.
+ const bool requires_external_file_ack =
params.__isset.iceberg_commit_datas;
Review Comment:
Fixed. An explicit ACK is now required whenever any ownership-bearing vector
is present: Hive, Iceberg, MaxCompute, or Paimon. Added coverage for every
vector and re-ran the Hive report lifecycle tests, including S3 MPU abort on
rejection and retention on an ambiguous result.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -298,6 +310,271 @@ void
checkVariantBackendCompatibilityForCurrentScan(Iterable<Backend> backends)
checkVariantBackendCompatibility(projectsVariant, backends);
}
+ private boolean requiresIcebergScanSemanticsV2() throws UserException {
+ if (isSystemTable) {
+ // position_deletes already has its stricter dedicated
mixed-version gate.
+ return false;
+ }
+ TableScan scan = createTableScan();
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ if (hasApplicableEqualityDeletes(scan)) {
+ return true;
+ }
+ Schema scanSchema = scan.schema();
+ Set<Integer> projectedFieldIds = projectedFieldIds(scanSchema);
+ Set<Integer> topLevelIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelIds.add(field.fieldId());
+ }
+ Map<Integer, NestedField> fieldsById =
TypeUtil.indexById(scanSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = fieldsById.get(fieldId);
+ if (field.initialDefault() != null
+ && (!topLevelIds.contains(field.fieldId()) ||
field.type().isNestedType())) {
+ return true;
+ }
+ }
+ if (hasProjectedNameAliasCollision(scanSchema, projectedFieldIds,
extractNameMapping())) {
+ return true;
+ }
+ return schemaHistoryRequiresMissingRequiredFieldRejection(
+ scanSchema, projectedFieldIds, reachableSchemas(icebergTable,
snapshot));
+ }
+
+ private boolean hasApplicableEqualityDeletes(TableScan scan) throws
UserException {
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ String equalityDeleteCount =
snapshot.summary().get("total-equality-deletes");
+ if (equalityDeleteCount != null) {
+ try {
+ // A positive snapshot total proves V2 semantics are required
without opening every
+ // delete manifest; only old summaries that omit the counter
need the fallback.
+ return Long.parseLong(equalityDeleteCount) > 0;
+ } catch (NumberFormatException ignored) {
+ // Fall through for non-standard summaries instead of
weakening compatibility.
+ }
+ }
+ // Inspect only delete manifests, not data tasks: equality-delete
semantics are snapshot-wide
+ // compatibility state even when the current predicate happens to
prune their partitions.
+ for (ManifestFile manifest :
snapshot.deleteManifests(icebergTable.io())) {
+ if (!manifest.hasAddedFiles() && !manifest.hasExistingFiles()) {
+ continue;
+ }
+ try (ManifestReader<DeleteFile> deletes =
ManifestFiles.readDeleteManifest(
+ manifest, icebergTable.io(), icebergTable.specs())) {
+ for (DeleteFile delete : deletes) {
+ if (delete.content() == FileContent.EQUALITY_DELETES) {
+ return true;
+ }
+ }
+ } catch (IOException e) {
+ throw new UserException(
+ "Failed to inspect Iceberg delete manifest " +
manifest.path(), e);
+ }
+ }
+ return false;
+ }
+
+ private static boolean hasSmoothUpgradeSource(Iterable<Backend> backends) {
+ for (Backend backend : backends) {
+ if (backend.isSmoothUpgradeSrc()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private Set<Integer> projectedFieldIds(Schema scanSchema) {
+ return projectedFieldIds(scanSchema, desc.getSlots());
+ }
+
+ @VisibleForTesting
+ static Set<Integer> projectedFieldIds(Schema scanSchema,
Iterable<SlotDescriptor> slots) {
+ Set<Integer> projected = new HashSet<>();
+ for (SlotDescriptor slot : slots) {
+ int fieldId = slot.getColumn().getUniqueId();
+ // Stable Iceberg IDs prevent a dropped-and-readded name from
selecting the wrong history.
+ NestedField field = fieldId >= 0 ? scanSchema.findField(fieldId)
+ :
scanSchema.caseInsensitiveFindField(slot.getColumn().getName());
+ if (field != null) {
+ collectProjectedFieldIds(field, slot.getType(), projected);
+ }
+ }
+ return projected;
+ }
+
+ private static void collectProjectedFieldIds(
+ NestedField field, org.apache.doris.catalog.Type projectedType,
+ Set<Integer> projected) {
+ projected.add(field.fieldId());
+ if (projectedType instanceof StructType &&
field.type().isStructType()) {
+ for (StructField projectedChild : ((StructType)
projectedType).getFields()) {
+ NestedField icebergChild =
field.type().asStructType().fields().stream()
+ .filter(child ->
child.name().equalsIgnoreCase(projectedChild.getName()))
+ .findFirst().orElse(null);
+ if (icebergChild != null) {
+ collectProjectedFieldIds(icebergChild,
projectedChild.getType(), projected);
+ }
+ }
+ } else if (projectedType instanceof ArrayType &&
field.type().isListType()) {
+ collectProjectedFieldIds(field.type().asListType().fields().get(0),
+ ((ArrayType) projectedType).getItemType(), projected);
+ } else if (projectedType instanceof MapType &&
field.type().isMapType()) {
+ collectProjectedFieldIds(field.type().asMapType().fields().get(0),
+ ((MapType) projectedType).getKeyType(), projected);
+ collectProjectedFieldIds(field.type().asMapType().fields().get(1),
+ ((MapType) projectedType).getValueType(), projected);
+ }
+ }
+
+ @VisibleForTesting
+ static Iterable<Schema> reachableSchemas(Table table, Snapshot
selectedSnapshot) {
+ Map<Integer, Schema> schemas = table.schemas();
+ List<Schema> reachable = new ArrayList<>();
+ Set<Long> visitedSnapshots = new HashSet<>();
+ Set<Integer> visitedSchemaIds = new HashSet<>();
+ Snapshot snapshot = selectedSnapshot;
+ while (snapshot != null &&
visitedSnapshots.add(snapshot.snapshotId())) {
+ Schema schema = schemas.get(snapshot.schemaId());
+ if (schema != null && visitedSchemaIds.add(snapshot.schemaId())) {
+ reachable.add(schema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null : table.snapshot(parentId);
Review Comment:
Fixed. If the selected snapshot names a parent that is no longer present,
the V2 gate now conservatively evaluates every surviving table schema instead
of treating the lineage as complete. Added expired-parent optional-to-required
coverage.
--
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]
