HDDS-14661. Handle split table writes for Multipart Upload

[spacemonkd]

What changes were proposed in this pull request?

HDDS-14661. Handle split table writes for Multipart Upload

Please describe your PR in detail:

Implements the runtime support for schemaVersion 1 multipart uploads using the split multipartPartsTable.

This patch covers the MPU lifecycle after an upload is already marked as schemaVersion 1:

• CommitPart writes part metadata into multipartPartsTable instead of growing MultipartInfoTable.
• Complete MPU reads parts from multipartPartsTable, assembles the final key, and deletes all consumed part rows.
• Abort MPU scans split-table parts, releases quota, moves part blocks to the deleted table, and removes part rows.
• Expired MPU cleanup follows the same split-table cleanup path for abandoned uploads.
• ListParts reads ordered part metadata from multipartPartsTable.
• Legacy schemaVersion 0 behavior remains unchanged.

This patch intentionally does not decide when new uploads become schemaVersion 1.
PR #10062 owns the upgrade/finalization and initiate-MPU schema selection. Once that branch starts creating schemaVersion 1 MPUs, this patch provides the runtime read/write/cleanup behavior needed to handle them end to end.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-14661

How was this patch tested?

Patch was tested using unit tests

[spacemonkd]

@errose28 @ivandika3 @kerneltime could you please take a look at this PR?
I plan on merging this PR first to put the wiring in place and then simultaneously merge the upgrade PR as they are co-dependent.

Having both of these ready to merge so we can do it in quick succession would be great!

[ivandika3]

Thanks @spacemonkd for the patch. Overall looks good. Left some comments.

Let's get this in asap, this feature has been outstanding quite a while.

[ivandika3]

Nit: I'd suggest to move this to OMMultipartUploadUtils so we only need a single util class for MPU.

[spacemonkd]

Addressed in the latest commit

[ivandika3]

Usually, in OmMetadataReader listKeys and getMultipartUploadKeys, we will iterate the cache first before the DB. Any reason here is reversed? Not saying it's wrong, just want to standardize.

[spacemonkd]

Sorry, it was a mistake on my part. I have reverted the order, ideally it should be reading from cache and if missed then go refer to DB.

[ivandika3]

Nit: A lot of unnecessary line wraps here which AI agent usually does. We can try to fix unnecessary wraps introduced in this patch as much as possible (within reason).

[spacemonkd]

Addressed in the latest commit

[ivandika3]

Nit: Let's make this inline instead of introducing another method.

[spacemonkd]

Addressed in the latest commit.

[ivandika3]

Do we have this logic in the schema version 0? OM might allow uploading an empty part (although S3G might fail it early).

[spacemonkd]

Yes, I reverted the extra check, however we still keep the ETag check as that is required.

[ivandika3]

Are we going to call the ETag validation here?

[spacemonkd]

I just realized that the partKeyInfoMap normalizes to add the split part information in the same object.
I simplified this in the latest commit. Can you please take a look now?

[ivandika3]

Just a thought, but if it's possible we can use deleteRange in the future to reduce the tombstones. However, deleteRange need to be done carefully since it can invalidate valid DB entries.

[spacemonkd]

Yes this is planned as part of the future work for MPU optimizations. It has been advised in the design document as well that we can use DeleteRange in future.

[ivandika3]

We might want to add a precondition here that if mutlipartPartKey and omMultipartPartInfo need to either be both null or both not null.

[spacemonkd]

Addressed in the latest commit

[spacemonkd]

Thanks for the extensive reviews @ivandika3, I caught a few redundant code blocks as well.
I addressed the comments, could you take another look?

[ivandika3]

@spacemonkd Thanks for the update, please kindly fix the tests first.

[rakeshadr]

Thanks @spacemonkd. Added a few comments, pls check it.

[rakeshadr]

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());
}

[rakeshadr]

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);
  }
}

[spacemonkd]

@ivandika3 @rakeshadr could you take another look? CI is green now.