rakeshadr commented on code in PR #10588:
URL: https://github.com/apache/ozone/pull/10588#discussion_r3544415459
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/util/OMMultipartUploadUtils.java:
##########
@@ -103,4 +117,91 @@ public static boolean isMultipartKeySet(OmKeyInfo
openKeyInfo) {
return openKeyInfo.getLatestVersionLocations() != null
&& openKeyInfo.getLatestVersionLocations().isMultipartKey();
}
+
+ /**
+ * Cache-aware scan for multipart parts table rows belonging to a given
upload.
+ */
+ public static SortedMap<Integer, OmMultipartPartInfo> scanParts(
+ OMMetadataManager omMetadataManager, String uploadId) throws IOException
{
+ SortedMap<Integer, OmMultipartPartInfo> parts = new TreeMap<>();
+
+ Iterator<Map.Entry<CacheKey<OmMultipartPartKey>,
+ CacheValue<OmMultipartPartInfo>>> cacheIterator =
+ omMetadataManager.getMultipartPartsTable().cacheIterator();
+ while (cacheIterator.hasNext()) {
+ Map.Entry<CacheKey<OmMultipartPartKey>, CacheValue<OmMultipartPartInfo>>
+ cacheEntry = cacheIterator.next();
+ OmMultipartPartKey key = cacheEntry.getKey().getCacheKey();
+ if (!uploadId.equals(key.getUploadId()) || !key.hasPartNumber()) {
+ continue;
+ }
+ OmMultipartPartInfo value = cacheEntry.getValue().getCacheValue();
+ if (value == null) {
+ parts.remove(key.getPartNumber());
Review Comment:
The cache pass does `parts.remove(partNumber)` when it sees a null value
(tombstone), but that's a no-op on an empty TreeMap. The subsequent DB pass
then sees the row still in RocksDB and re-inserts it via:
```
if (key.hasPartNumber() && !parts.containsKey(key.getPartNumber())) {
parts.put(key.getPartNumber(), kv.getValue());
}
```
This means any part that has a cache tombstone (committed to delete but not
yet flushed to DB) will reappear from the DB pass. Since scanParts() is called
by listParts, complete, abort, and expired MPU abort, this affects all four
operations.
Can you change remove by tracking tombstoned keys explicitly during the
cache pass and skip them in the DB pass.
```
Set<Integer> tombstoned = new HashSet<>();
// cache pass
if (value == null) {
parts.remove(key.getPartNumber()); // ← always no-op
tombstoned.add(key.getPartNumber()); // ← now DB pass can check this
} else {
parts.put(key.getPartNumber(), value);
}
// DB pass
if (key.hasPartNumber()
&& !parts.containsKey(key.getPartNumber())
&& !tombstoned.contains(key.getPartNumber())) {
parts.put(key.getPartNumber(), kv.getValue());
}
```
##########
hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java:
##########
@@ -41,6 +41,9 @@
* upload part information of the key.
*/
public final class OmMultipartKeyInfo extends WithObjectID implements
CopyObject<OmMultipartKeyInfo> {
+ public static final byte LEGACY_SCHEMA_VERSION = 0;
+ public static final byte SPLIT_PARTS_TABLE_SCHEMA_VERSION = 1;
Review Comment:
Zero request-path test coverage for schema version 1.
S3InitiateMultipartUploadRequest always creates schema version 0.
One idea is - tests need to directly inject a schema v1 OmMultipartKeyInfo +
seed multipartPartsTable. Then extend the existing test classes.
**Imp Note:** LLM generated tests. Please validate and then add it.
```
Step 1 — Helper method in TestS3MultipartUploadCommitPartRequest
A schema v1 MPU bypasses the initiate request and directly populates the
tables:
private OmMultipartKeyInfo createSchemaV1MpuEntry(
String volumeName, String bucketName, String keyName,
String uploadId, long trxnIdx) throws IOException {
OmMultipartKeyInfo keyInfo = new OmMultipartKeyInfo.Builder()
.setUploadID(uploadId)
.setCreationTime(Time.now())
.setReplicationConfig(RatisReplicationConfig.getInstance(ReplicationFactor.ONE))
.setObjectID(trxnIdx)
.setUpdateID(trxnIdx)
.setSchemaVersion(OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION)
.build();
OmKeyInfo omKeyInfo = OMRequestTestUtils.createOmKeyInfo(
volumeName, bucketName, keyName,
RatisReplicationConfig.getInstance(ReplicationFactor.ONE));
OMRequestTestUtils.addMultipartInfoToTable(
false, omKeyInfo, keyInfo, trxnIdx, omMetadataManager);
return keyInfo;
}
Schema v1 CommitPart validates ETag. The existing addKeyToOpenKeyTable
doesn't set it. Add a variant:
protected String addKeyToOpenKeyTableWithETag(
String volumeName, String bucketName, String keyName,
long clientID, List<OmKeyLocationInfo> locationList) throws Exception {
OmKeyInfo keyInfo = OMRequestTestUtils.createOmKeyInfo(
volumeName, bucketName, keyName,
RatisReplicationConfig.getInstance(ReplicationFactor.ONE),
clientID, locationList);
keyInfo.addMetadata(OzoneConsts.ETAG, UUID.randomUUID().toString());
String openKey = omMetadataManager.getOpenKey(
volumeName, bucketName, keyName, clientID);
omMetadataManager.getOpenKeyTable(getBucketLayout())
.addCacheEntry(new CacheKey<>(openKey), CacheValue.get(clientID,
keyInfo));
omMetadataManager.getOpenKeyTable(getBucketLayout()).put(openKey, keyInfo);
return openKey;
}
Step 2 — New test methods to add to existing test classes
TestS3MultipartUploadCommitPartRequest — add:
@Test
public void testV1CommitWritesToPartsTable() throws Exception {
String vol = UUID.randomUUID().toString(), bkt =
UUID.randomUUID().toString();
String key = getKeyName(), uploadId = UUID.randomUUID().toString();
OMRequestTestUtils.addVolumeAndBucketToDB(vol, bkt, omMetadataManager,
getBucketLayout());
createParentPath(vol, bkt);
createSchemaV1MpuEntry(vol, bkt, key, uploadId, 1L);
long clientID = Time.now();
List<KeyLocation> locs = getKeyLocation(2);
addKeyToOpenKeyTableWithETag(vol, bkt, key, clientID,
locs.stream().map(OmKeyLocationInfo::getFromProtobuf).collect(toList()));
OMClientResponse resp = getS3MultipartUploadCommitReq(
doPreExecuteCommitMPU(vol, bkt, key, clientID, uploadId, 1, locs))
.validateAndUpdateCache(ozoneManager, 2L);
assertEquals(Status.OK, resp.getOMResponse().getStatus());
// Part must be in multipartPartsTable
assertNotNull(omMetadataManager.getMultipartPartsTable()
.get(OmMultipartPartKey.of(uploadId, 1)));
// multipartInfoTable entry must have empty part list (schema v1 never
writes there)
assertEquals(0, omMetadataManager.getMultipartInfoTable()
.get(omMetadataManager.getMultipartKey(vol, bkt, key, uploadId))
.getPartKeyInfoMap().size());
}
@Test
public void testV1CommitOverwriteQueuesOldBlocksForDeletion() throws
Exception {
String vol = UUID.randomUUID().toString(), bkt =
UUID.randomUUID().toString();
String key = getKeyName(), uploadId = UUID.randomUUID().toString();
OMRequestTestUtils.addVolumeAndBucketToDB(vol, bkt, omMetadataManager,
getBucketLayout());
createParentPath(vol, bkt);
createSchemaV1MpuEntry(vol, bkt, key, uploadId, 1L);
// First commit of part 1
long cid1 = Time.now();
List<KeyLocation> locs1 = getKeyLocation(2);
addKeyToOpenKeyTableWithETag(vol, bkt, key, cid1,
locs1.stream().map(OmKeyLocationInfo::getFromProtobuf).collect(toList()));
getS3MultipartUploadCommitReq(
doPreExecuteCommitMPU(vol, bkt, key, cid1, uploadId, 1, locs1))
.validateAndUpdateCache(ozoneManager, 2L);
// Overwrite part 1
long cid2 = Time.now();
List<KeyLocation> locs2 = getKeyLocation(3);
addKeyToOpenKeyTableWithETag(vol, bkt, key, cid2,
locs2.stream().map(OmKeyLocationInfo::getFromProtobuf).collect(toList()));
OMClientResponse resp = getS3MultipartUploadCommitReq(
doPreExecuteCommitMPU(vol, bkt, key, cid2, uploadId, 1, locs2))
.validateAndUpdateCache(ozoneManager, 3L);
assertEquals(Status.OK, resp.getOMResponse().getStatus());
// Old blocks must be in the deleted table
Map<String, RepeatedOmKeyInfo> toDelete =
((S3MultipartUploadCommitPartResponse) resp).getKeyToDelete();
assertNotNull(toDelete);
assertFalse(toDelete.isEmpty());
}
TestS3MultipartUploadAbortRequest — add:
@Test
public void testV1AbortMovesPartsToDeletedTable() throws Exception {
String vol = UUID.randomUUID().toString(), bkt =
UUID.randomUUID().toString();
String key = getKeyName(), uploadId = UUID.randomUUID().toString();
OMRequestTestUtils.addVolumeAndBucketToDB(vol, bkt, omMetadataManager,
getBucketLayout());
createSchemaV1MpuEntry(vol, bkt, key, uploadId, 1L);
// Seed two parts directly into multipartPartsTable
seedPartsInTable(uploadId, 2);
OMClientResponse resp = getS3MultipartUploadAbortReq(
doPreExecuteAbortMPU(vol, bkt, key, uploadId))
.validateAndUpdateCache(ozoneManager, 3L);
assertEquals(Status.OK, resp.getOMResponse().getStatus());
// multipartPartsTable must be cleared for this uploadId
assertNull(omMetadataManager.getMultipartPartsTable()
.get(OmMultipartPartKey.of(uploadId, 1)));
assertNull(omMetadataManager.getMultipartPartsTable()
.get(OmMultipartPartKey.of(uploadId, 2)));
// multipartInfoTable entry must be gone
assertNull(omMetadataManager.getMultipartInfoTable()
.get(omMetadataManager.getMultipartKey(vol, bkt, key, uploadId)));
}
TestS3MultipartUploadCompleteRequest — add (same pattern — seed parts,
complete, verify multipartPartsTable cleared and key written).
Step 3 — seedPartsInTable helper
private void seedPartsInTable(String uploadId, int count) throws IOException
{
for (int i = 1; i <= count; i++) {
List<OmKeyLocationInfo> locs = getKeyLocation(1).stream()
.map(OmKeyLocationInfo::getFromProtobuf).collect(toList());
OmMultipartPartInfo partInfo = new OmMultipartPartInfo.Builder()
.setPartName(uploadId + "/" + i)
.setPartNumber(i)
.setDataSize(200L)
.setModificationTime(Time.now())
.setObjectID((long) i)
.setUpdateID((long) i)
.setETag(UUID.randomUUID().toString())
.setKeyLocationInfos(Collections.singletonList(
new OmKeyLocationInfoGroup(0, locs, false)))
.build();
OmMultipartPartKey partKey = OmMultipartPartKey.of(uploadId, i);
omMetadataManager.getMultipartPartsTable().put(partKey, partInfo);
}
}
```
--
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]
