github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3840874517
##########
be/src/exec/sink/writer/iceberg/viceberg_sort_writer.cpp:
##########
@@ -85,6 +87,59 @@ size_t
VIcebergSortWriter::get_reserve_mem_size(RuntimeState* state, bool eos) c
return _sorter == nullptr ? 0 : _sorter->get_reserve_mem_size(state, eos);
}
+SorterReserveMemory
VIcebergSortWriter::get_reserve_mem_size_components(RuntimeState* state,
+ bool
eos) const {
+ std::lock_guard<std::mutex> lock(_sorter_mutex);
+ if (_sorter == nullptr) {
+ return {};
+ }
+ const size_t target = _target_file_size_bytes >= 0
+ ?
static_cast<size_t>(_target_file_size_bytes)
+ : std::numeric_limits<size_t>::max();
+ auto reservation = _sorter->get_reserve_mem_size_components(state, eos,
target);
+ _include_merge_reservation(state, eos, &reservation);
+ return reservation;
+}
+
+SorterReserveMemory VIcebergSortWriter::get_reserve_mem_size_components(
+ RuntimeState* state, bool eos, size_t incoming_rows, size_t
incoming_bytes) const {
+ std::lock_guard<std::mutex> lock(_sorter_mutex);
+ if (_sorter == nullptr) {
+ return {};
+ }
+ const size_t target = _target_file_size_bytes >= 0
+ ?
static_cast<size_t>(_target_file_size_bytes)
+ : std::numeric_limits<size_t>::max();
+ auto reservation = _sorter->get_reserve_mem_size_components(state, eos,
incoming_rows,
+
incoming_bytes, target);
+ _include_merge_reservation(state, eos, &reservation);
+ return reservation;
+}
+
+void VIcebergSortWriter::_include_merge_reservation(RuntimeState* state, bool
eos,
+ SorterReserveMemory*
reservation) const {
+ if (!eos) {
Review Comment:
[P1] Reserve merge output for target rollovers
This returns before adding merge workspace for every non-EOS block, but an
append that crosses `_target_file_size_bytes` immediately calls
`_flush_to_file()`. After `do_sort()` and `prepare_for_read(false)`,
`_write_sorted_data()` materializes the same byte-bounded merge-output block
that the EOS branch reserves while the sorted runs remain retained. A tiny
block can cross the target for a sorter whose earlier rows set `_max_row_bytes`
above the operator floor, so neither the incoming-block allowance nor the sort
destination/permutation reservation covers that output and the async writer can
cross the hard limit. Please include `iceberg_merge_output_workspace(...)`
whenever the predicted append reaches the target, and test a non-EOS rollover
with retained runs, an over-floor wide row, and a tiny tail.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSTransaction.java:
##########
@@ -175,17 +181,53 @@ public List<THivePartitionUpdate>
mergePartitions(List<THivePartitionUpdate> hiv
private void
collectUncompletedMpuPendingUploads(List<THivePartitionUpdate> hivePUs) {
for (THivePartitionUpdate pu : hivePUs) {
- if (pu.getS3MpuPendingUploads() != null) {
- for (TS3MPUPendingUpload s3MPUPendingUpload :
pu.getS3MpuPendingUploads()) {
- uncompletedMpuPendingUploads.add(
- new
UncompletedMpuPendingUpload(s3MPUPendingUpload,
pu.getLocation().getWritePath()));
+ List<TS3MPUPendingUpload> uploads = pu.getS3MpuPendingUploads();
+ if (uploads == null) {
+ continue;
+ }
+ String writePath = pu.getLocation() == null ? null :
pu.getLocation().getWritePath();
+ if (Strings.isNullOrEmpty(writePath)) {
+ // One malformed record must not prevent valid sibling uploads
from being cleaned up.
+ LOG.warn("Skipping MPU cleanup record without a write path");
+ continue;
+ }
+ for (TS3MPUPendingUpload upload : uploads) {
+ if (!isCompleteObjectStoreUpload(upload)) {
+ LOG.warn("Skipping incomplete MPU cleanup record for write
path {}", writePath);
+ continue;
}
+ uncompletedMpuPendingUploads.add(new
UncompletedMpuPendingUpload(upload, writePath));
+ }
+ }
+ }
+
+ private static boolean isCompleteObjectStoreUpload(TS3MPUPendingUpload
upload) {
+ return upload != null && !Strings.isNullOrEmpty(upload.getUploadId())
+ && !Strings.isNullOrEmpty(upload.getBucket()) &&
!Strings.isNullOrEmpty(upload.getKey());
+ }
+
+ private void validateObjectStoreCommitRecords() {
+ if (fileType != TFileType.FILE_S3) {
+ return;
+ }
+ for (THivePartitionUpdate update : hivePartitionUpdates) {
+ int fileCount = update.getFileNames() == null ? 0 :
update.getFileNames().size();
+ List<TS3MPUPendingUpload> uploads =
update.getS3MpuPendingUploads();
+ int uploadCount = uploads == null ? 0 : uploads.size();
+ boolean completeRecords = uploads != null
Review Comment:
[P1] Reject partial deferred multipart reports
This predicate treats an upload as complete when only its ID/bucket/key are
present, but the producing path can return such a record with only a subset of
its parts. `S3FileWriter::_complete()` checks `_failed` before waiting; after
`_wait_until_finish()` it returns `OK` immediately for `_used_by_s3_committer`,
before the later `_failed` and expected-part-count checks. If an async part
fails during that wait, `_build_s3_mpu_pending_upload()` therefore reports the
surviving `completed_parts()` map and this validation accepts it. Azure then
commits exactly those listed block IDs (and S3 can complete the supplied
subset), publishing a truncated object while Hive records the full row
count/file size. Please run the post-wait failure/count validation before the
committer return and cover a missing middle/tail part; validating
nonempty/contiguous part maps here would also provide defense in depth.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -309,6 +322,302 @@ 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);
+ }
+ Map<String, String> summary = snapshot.summary();
+ String sourceSnapshotId = summary == null
+ ? null :
summary.get(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP);
+ if (sourceSnapshotId != null) {
+ Snapshot sourceSnapshot;
+ try {
+ sourceSnapshot =
table.snapshot(Long.parseLong(sourceSnapshotId));
+ } catch (NumberFormatException e) {
+ sourceSnapshot = null;
+ }
+ if (sourceSnapshot == null ||
!schemas.containsKey(sourceSnapshot.schemaId())) {
+ // A cherry-picked snapshot can contribute live files
outside the selected
+ // parent chain; unverifiable provenance must
conservatively gate all schemas.
+ for (Schema historicalSchema : schemas.values()) {
+ if (visitedSchemaIds.add(historicalSchema.schemaId()))
{
+ reachable.add(historicalSchema);
+ }
+ }
+ } else if (visitedSchemaIds.add(sourceSnapshot.schemaId())) {
+ reachable.add(schemas.get(sourceSnapshot.schemaId()));
+ }
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null : table.snapshot(parentId);
+ if (parentId != null && snapshot == null) {
+ // Expiration may remove the parent metadata while descendants
still inherit its
+ // files, so an incomplete lineage must gate against every
surviving schema.
+ for (Schema historicalSchema : schemas.values()) {
+ if (visitedSchemaIds.add(historicalSchema.schemaId())) {
+ reachable.add(historicalSchema);
+ }
+ }
+ }
+ }
+ // Only ancestors of the selected ref can have produced files visible
to this scan.
+ return reachable;
+ }
+
+ @VisibleForTesting
+ static void checkIcebergScanSemanticsV2Compatibility(
+ boolean requiresV2, Iterable<Backend> backends) throws
UserException {
+ if (!requiresV2) {
+ return;
+ }
+ for (Backend backend : backends) {
+ if (backend.isSmoothUpgradeSrc()) {
+ // A V1 BE accepts the Thrift field but does not enforce
nested defaults/requiredness.
+ throw new UserException("Current Iceberg scan semantics are
unavailable while backend "
+ + backend.getId() + " is a smooth upgrade source");
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static boolean schemaHistoryRequiresMissingRequiredFieldRejection(
+ Schema scanSchema, Iterable<Schema> historicalSchemas) {
+ return schemaHistoryRequiresMissingRequiredFieldRejection(
+ scanSchema,
TypeUtil.indexById(scanSchema.asStruct()).keySet(), historicalSchemas);
+ }
+
+ private static boolean schemaHistoryRequiresMissingRequiredFieldRejection(
+ Schema scanSchema, Set<Integer> projectedFieldIds,
Iterable<Schema> historicalSchemas) {
+ Map<Integer, NestedField> currentFields =
TypeUtil.indexById(scanSchema.asStruct());
+ Map<Integer, Integer> parentById =
TypeUtil.indexParents(scanSchema.asStruct());
+ Set<Integer> collectionWrapperIds = new HashSet<>();
+ collectCollectionWrapperFieldIds(scanSchema.asStruct(),
collectionWrapperIds);
+ for (Schema historicalSchema : historicalSchemas) {
+ Map<Integer, NestedField> historicalFields =
TypeUtil.indexById(historicalSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = currentFields.get(fieldId);
+ if (field == null ||
collectionWrapperIds.contains(field.fieldId())
Review Comment:
[P1] Enforce required collection entries
This skips list-element and map-value IDs from the requiredness-history
gate, but those wrapper fields can change from optional to required while
retaining the same ID, and reachable old files may still contain NULL entries.
FE now transports `is_optional=false`, yet `icebergTypeToDorisType()` keeps
collection children nullable; BE validates present children from `table_type`
(and ARRAY explicitly makes the element mapping nullable) and consults
`is_optional` only when a field is physically missing. The V2 reader therefore
returns the old NULL under the current required schema, and this skip also
permits a V1 smooth-upgrade BE. Please gate optional-to-required element/value
history and validate present entries from the transported optionality, with
historical-NULL coverage on both V1 and V2 paths.
--
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]
