diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
index 44dc80f01ec0..d59ac5b6d4f1 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
@@ -667,7 +667,7 @@ public final class OMConfigKeys {
   public static final String OZONE_OM_COMPACTION_SERVICE_COLUMNFAMILIES
       = "ozone.om.compaction.service.columnfamilies";
   public static final String OZONE_OM_COMPACTION_SERVICE_COLUMNFAMILIES_DEFAULT =
-      "keyTable,fileTable,directoryTable,deletedTable,deletedDirectoryTable,multipartInfoTable";
+      "keyTable,fileTable,directoryTable,deletedTable,deletedDirectoryTable,multipartInfoTable,multipartPartsTable";
 
   /**
    * Configuration to enable/disable non-snapshot diff table compaction when snapshots are evicted from cache.
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java
index 226871882d4b..88036d3e66a6 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java
@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.om.helpers;
 
+import com.google.common.collect.ImmutableList;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
@@ -31,6 +32,7 @@
 import org.apache.hadoop.hdds.utils.db.CopyObject;
 import org.apache.hadoop.hdds.utils.db.DelegatedCodec;
 import org.apache.hadoop.hdds.utils.db.Proto2Codec;
+import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.MultipartKeyInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PartKeyInfo;
 
@@ -46,6 +48,14 @@ public final class OmMultipartKeyInfo extends WithObjectID implements CopyObject
       OmMultipartKeyInfo.class);
 
   private final String uploadID;
+  private final String volumeName;
+  private final String bucketName;
+  private final String keyName;
+  private final String ownerName;
+  /**
+   * ACL information inherited during MPU initiation.
+   */
+  private final ImmutableList<OzoneAcl> acls;
   private final long creationTime;
   private final ReplicationConfig replicationConfig;
   private PartKeyInfoMap partKeyInfoMap;
@@ -74,6 +84,11 @@ public final class OmMultipartKeyInfo extends WithObjectID implements CopyObject
    */
   private final long parentID;
 
+  // This stores the schema version of the multipart key.
+  // 0 - Legacy Schema -> Uses the same table to store the multipart part info
+  // 1 - New Schema -> Uses a separate table to store the multipart part info
+  private final byte schemaVersion;
+
   public static Codec<OmMultipartKeyInfo> getCodec() {
     return CODEC;
   }
@@ -124,11 +139,11 @@ static PartKeyInfoMap put(PartKeyInfo partKeyInfo,
       return new PartKeyInfoMap(list);
     }
 
-    PartKeyInfoMap(List<PartKeyInfo> sorted) {
+    public PartKeyInfoMap(List<PartKeyInfo> sorted) {
       this.sorted = Collections.unmodifiableList(sorted);
     }
 
-    PartKeyInfoMap(SortedMap<Integer, PartKeyInfo> sorted) {
+    public PartKeyInfoMap(SortedMap<Integer, PartKeyInfo> sorted) {
       this(new ArrayList<>(sorted.values()));
     }
 
@@ -166,16 +181,27 @@ public PartKeyInfo lastEntry() {
   private OmMultipartKeyInfo(Builder b) {
     super(b);
     this.uploadID = b.uploadID;
+    this.volumeName = b.volumeName;
+    this.bucketName = b.bucketName;
+    this.keyName = b.keyName;
+    this.ownerName = b.ownerName;
+    this.acls = b.acls.build();
     this.creationTime = b.creationTime;
     this.replicationConfig = b.replicationConfig;
     this.partKeyInfoMap = new PartKeyInfoMap(b.partKeyInfoList);
     this.parentID = b.parentID;
+    this.schemaVersion = b.schemaVersion;
   }
 
   /** Copy constructor. */
   private OmMultipartKeyInfo(OmMultipartKeyInfo b) {
     super(b);
     this.uploadID = b.uploadID;
+    this.volumeName = b.volumeName;
+    this.bucketName = b.bucketName;
+    this.keyName = b.keyName;
+    this.ownerName = b.ownerName;
+    this.acls = b.acls;
     this.creationTime = b.creationTime;
     this.replicationConfig = b.replicationConfig;
     // PartKeyInfoMap is an immutable data structure. Whenever a PartKeyInfo
@@ -183,6 +209,7 @@ private OmMultipartKeyInfo(OmMultipartKeyInfo b) {
     // so here we can directly pass in partKeyInfoMap
     this.partKeyInfoMap = b.partKeyInfoMap;
     this.parentID = b.parentID;
+    this.schemaVersion = b.schemaVersion;
   }
 
   /**
@@ -206,11 +233,35 @@ public long getCreationTime() {
     return creationTime;
   }
 
+  public String getVolumeName() {
+    return volumeName;
+  }
+
+  public String getBucketName() {
+    return bucketName;
+  }
+
+  public String getKeyName() {
+    return keyName;
+  }
+
+  public String getOwnerName() {
+    return ownerName;
+  }
+
+  public List<OzoneAcl> getAcls() {
+    return acls;
+  }
+
   public PartKeyInfoMap getPartKeyInfoMap() {
     return partKeyInfoMap;
   }
 
   public void addPartKeyInfo(PartKeyInfo partKeyInfo) {
+    if (schemaVersion == 1) {
+      throw new IllegalStateException(
+          "PartKeyInfoMap is not supported for schemaVersion 1");
+    }
     this.partKeyInfoMap = PartKeyInfoMap.put(partKeyInfo, partKeyInfoMap);
   }
 
@@ -222,6 +273,10 @@ public ReplicationConfig getReplicationConfig() {
     return replicationConfig;
   }
 
+  public byte getSchemaVersion() {
+    return schemaVersion;
+  }
+
   public Builder toBuilder() {
     return new Builder(this);
   }
@@ -231,25 +286,42 @@ public Builder toBuilder() {
    */
   public static class Builder extends WithObjectID.Builder<OmMultipartKeyInfo> {
     private String uploadID;
+    private String volumeName;
+    private String bucketName;
+    private String keyName;
+    private String ownerName;
     private long creationTime;
     private ReplicationConfig replicationConfig;
+    private final AclListBuilder acls;
     private final TreeMap<Integer, PartKeyInfo> partKeyInfoList;
     private long parentID;
+    private byte schemaVersion;
 
     public Builder() {
+      this.acls = AclListBuilder.empty();
       this.partKeyInfoList = new TreeMap<>();
     }
 
     public Builder(OmMultipartKeyInfo multipartKeyInfo) {
       super(multipartKeyInfo);
       this.uploadID = multipartKeyInfo.uploadID;
+      this.volumeName = multipartKeyInfo.volumeName;
+      this.bucketName = multipartKeyInfo.bucketName;
+      this.keyName = multipartKeyInfo.keyName;
+      this.ownerName = multipartKeyInfo.ownerName;
       this.creationTime = multipartKeyInfo.creationTime;
       this.replicationConfig = multipartKeyInfo.replicationConfig;
+      this.acls = AclListBuilder.of(multipartKeyInfo.acls);
       this.partKeyInfoList = new TreeMap<>();
-      for (PartKeyInfo partKeyInfo : multipartKeyInfo.partKeyInfoMap) {
-        this.partKeyInfoList.put(partKeyInfo.getPartNumber(), partKeyInfo);
+
+      if (multipartKeyInfo.getSchemaVersion() == 0) {
+        for (PartKeyInfo partKeyInfo : multipartKeyInfo.partKeyInfoMap) {
+          this.partKeyInfoList.put(partKeyInfo.getPartNumber(), partKeyInfo);
+        }
       }
+
       this.parentID = multipartKeyInfo.parentID;
+      this.schemaVersion = multipartKeyInfo.schemaVersion;
     }
 
     public Builder setUploadID(String uploadId) {
@@ -257,6 +329,26 @@ public Builder setUploadID(String uploadId) {
       return this;
     }
 
+    public Builder setVolumeName(String volName) {
+      this.volumeName = volName;
+      return this;
+    }
+
+    public Builder setBucketName(String buckName) {
+      this.bucketName = buckName;
+      return this;
+    }
+
+    public Builder setKeyName(String keyObjName) {
+      this.keyName = keyObjName;
+      return this;
+    }
+
+    public Builder setOwnerName(String owner) {
+      this.ownerName = owner;
+      return this;
+    }
+
     public Builder setCreationTime(long crTime) {
       this.creationTime = crTime;
       return this;
@@ -281,6 +373,24 @@ public Builder addPartKeyInfoList(int partNum, PartKeyInfo partKeyInfo) {
       return this;
     }
 
+    public Builder setAcls(List<OzoneAcl> listOfAcls) {
+      if (listOfAcls != null) {
+        this.acls.set(listOfAcls);
+      }
+      return this;
+    }
+
+    public AclListBuilder acls() {
+      return acls;
+    }
+
+    public Builder addAcl(OzoneAcl ozoneAcl) {
+      if (ozoneAcl != null) {
+        this.acls.add(ozoneAcl);
+      }
+      return this;
+    }
+
     @Override
     public Builder setObjectID(long obId) {
       super.setObjectID(obId);
@@ -298,6 +408,11 @@ public Builder setParentID(long parentObjId) {
       return this;
     }
 
+    public Builder setSchemaVersion(byte schemaVersion) {
+      this.schemaVersion = schemaVersion;
+      return this;
+    }
+
     @Override
     protected OmMultipartKeyInfo buildObject() {
       return new OmMultipartKeyInfo(this);
@@ -312,8 +427,10 @@ protected OmMultipartKeyInfo buildObject() {
   public static Builder builderFromProto(
       MultipartKeyInfo multipartKeyInfo) {
     final SortedMap<Integer, PartKeyInfo> list = new TreeMap<>();
-    multipartKeyInfo.getPartKeyInfoListList().forEach(partKeyInfo ->
-        list.put(partKeyInfo.getPartNumber(), partKeyInfo));
+    if (!multipartKeyInfo.hasSchemaVersion() || multipartKeyInfo.getSchemaVersion() == 0) {
+      multipartKeyInfo.getPartKeyInfoListList().forEach(partKeyInfo ->
+          list.put(partKeyInfo.getPartNumber(), partKeyInfo));
+    }
 
     final ReplicationConfig replicationConfig = ReplicationConfig.fromProto(
         multipartKeyInfo.getType(),
@@ -323,12 +440,22 @@ public static Builder builderFromProto(
 
     return new Builder()
         .setUploadID(multipartKeyInfo.getUploadID())
+        .setVolumeName(multipartKeyInfo.hasVolumeName() ?
+            multipartKeyInfo.getVolumeName() : null)
+        .setBucketName(multipartKeyInfo.hasBucketName() ?
+            multipartKeyInfo.getBucketName() : null)
+        .setKeyName(multipartKeyInfo.hasKeyName() ?
+            multipartKeyInfo.getKeyName() : null)
+        .setOwnerName(multipartKeyInfo.hasOwnerName() ?
+            multipartKeyInfo.getOwnerName() : null)
         .setCreationTime(multipartKeyInfo.getCreationTime())
         .setReplicationConfig(replicationConfig)
+        .setAcls(OzoneAclUtil.fromProtobuf(multipartKeyInfo.getAclsList()))
         .setPartKeyInfoList(list)
         .setObjectID(multipartKeyInfo.getObjectID())
         .setUpdateID(multipartKeyInfo.getUpdateID())
-        .setParentID(multipartKeyInfo.getParentID());
+        .setParentID(multipartKeyInfo.getParentID())
+        .setSchemaVersion((byte) multipartKeyInfo.getSchemaVersion());
   }
 
   /**
@@ -346,13 +473,31 @@ public static OmMultipartKeyInfo getFromProto(
    * @return MultipartKeyInfo
    */
   public MultipartKeyInfo getProto() {
+    if (schemaVersion == 1 && partKeyInfoMap != null && partKeyInfoMap.size() > 0) {
+      throw new IllegalStateException(
+          "PartKeyInfoMap must be empty for schemaVersion 1");
+    }
+
     MultipartKeyInfo.Builder builder = MultipartKeyInfo.newBuilder()
         .setUploadID(uploadID)
         .setCreationTime(creationTime)
         .setType(replicationConfig.getReplicationType())
         .setObjectID(getObjectID())
         .setUpdateID(getUpdateID())
-        .setParentID(parentID);
+        .setParentID(parentID)
+        .setSchemaVersion(schemaVersion);
+    if (volumeName != null) {
+      builder.setVolumeName(volumeName);
+    }
+    if (bucketName != null) {
+      builder.setBucketName(bucketName);
+    }
+    if (keyName != null) {
+      builder.setKeyName(keyName);
+    }
+    if (ownerName != null) {
+      builder.setOwnerName(ownerName);
+    }
 
     if (replicationConfig instanceof ECReplicationConfig) {
       ECReplicationConfig ecConf = (ECReplicationConfig) replicationConfig;
@@ -361,7 +506,10 @@ public MultipartKeyInfo getProto() {
       builder.setFactor(ReplicationConfig.getLegacyFactor(replicationConfig));
     }
 
-    builder.addAllPartKeyInfoList(partKeyInfoMap);
+    builder.addAllAcls(OzoneAclUtil.toProtobuf(acls));
+    if (schemaVersion == 0) {
+      builder.addAllPartKeyInfoList(partKeyInfoMap);
+    }
     return builder.build();
   }
 
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartPartInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartPartInfo.java
new file mode 100644
index 000000000000..f8bf57de1498
--- /dev/null
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartPartInfo.java
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.helpers;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.FileChecksum;
+import org.apache.hadoop.fs.FileEncryptionInfo;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.hdds.utils.db.DelegatedCodec;
+import org.apache.hadoop.hdds.utils.db.Proto2Codec;
+import org.apache.hadoop.ozone.ClientVersion;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocationList;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.MultipartPartInfo;
+import org.apache.hadoop.ozone.protocolPB.OMPBHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class represents a part of a multipart upload key.
+ */
+public final class OmMultipartPartInfo {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OmMultipartPartInfo.class);
+  private static final Codec<OmMultipartPartInfo> CODEC = new DelegatedCodec<>(
+      Proto2Codec.get(MultipartPartInfo.getDefaultInstance()),
+      OmMultipartPartInfo::getFromProto,
+      OmMultipartPartInfo::getProto,
+      OmMultipartPartInfo.class);
+
+  private final String partName;
+  private final int partNumber;
+  private final long dataSize;
+  private final long modificationTime;
+  private final long objectID;
+  private final long updateID;
+  private final List<OmKeyLocationInfoGroup> keyLocationInfos;
+  private final String eTag;
+  private final FileEncryptionInfo encInfo;
+  private final FileChecksum fileChecksum;
+
+  public static Codec<OmMultipartPartInfo> getCodec() {
+    return CODEC;
+  }
+
+  private OmMultipartPartInfo(Builder b) {
+    if (StringUtils.isBlank(b.partName)) {
+      throw new IllegalArgumentException("partName is required");
+    }
+    if (b.partNumber <= 0) {
+      throw new IllegalArgumentException("partNumber is required and > 0");
+    }
+    if (StringUtils.isBlank(b.eTag)) {
+      throw new IllegalArgumentException("eTag is required");
+    }
+    if (b.keyLocationInfos == null || b.keyLocationInfos.isEmpty()) {
+      throw new IllegalArgumentException("keyLocationList is required");
+    }
+    this.partName = b.partName;
+    this.partNumber = b.partNumber;
+    this.dataSize = b.dataSize;
+    this.modificationTime = b.modificationTime;
+    this.objectID = b.objectID;
+    this.updateID = b.updateID;
+    this.keyLocationInfos = Collections.unmodifiableList(b.keyLocationInfos);
+    this.eTag = b.eTag;
+    this.encInfo = b.encInfo;
+    this.fileChecksum = b.fileChecksum;
+  }
+
+  /**
+   * Builder of OmMultipartPartInfo.
+   */
+  public static class Builder {
+    private String partName;
+    private int partNumber;
+    private long dataSize;
+    private long modificationTime;
+    private long objectID;
+    private long updateID;
+    private List<OmKeyLocationInfoGroup> keyLocationInfos;
+    private String eTag;
+    private FileEncryptionInfo encInfo;
+    private FileChecksum fileChecksum;
+
+    protected Builder() {
+      this.keyLocationInfos = new ArrayList<>();
+    }
+
+    public Builder(OmMultipartPartInfo obj) {
+      this.partName = obj.partName;
+      this.partNumber = obj.partNumber;
+      this.dataSize = obj.dataSize;
+      this.modificationTime = obj.modificationTime;
+      this.objectID = obj.objectID;
+      this.updateID = obj.updateID;
+      this.keyLocationInfos = new ArrayList<>(obj.keyLocationInfos);
+      this.eTag = obj.eTag;
+      this.encInfo = obj.encInfo;
+      this.fileChecksum = obj.fileChecksum;
+    }
+
+    public Builder setPartName(String partName) {
+      this.partName = partName;
+      return this;
+    }
+
+    public Builder setPartNumber(int partNumber) {
+      this.partNumber = partNumber;
+      return this;
+    }
+
+    public Builder setDataSize(long dataSize) {
+      this.dataSize = dataSize;
+      return this;
+    }
+
+    public Builder setModificationTime(long modificationTime) {
+      this.modificationTime = modificationTime;
+      return this;
+    }
+
+    public Builder setObjectID(long objectID) {
+      this.objectID = objectID;
+      return this;
+    }
+
+    public Builder setUpdateID(long updateID) {
+      this.updateID = updateID;
+      return this;
+    }
+
+    public Builder setETag(String eTagValue) {
+      if (StringUtils.isBlank(eTagValue)) {
+        throw new IllegalArgumentException("eTag is required");
+      }
+      this.eTag = eTagValue;
+      return this;
+    }
+
+    public Builder setKeyLocationInfos(
+        List<OmKeyLocationInfoGroup> keyLocationInfos) {
+      this.keyLocationInfos = new ArrayList<>(keyLocationInfos);
+      return this;
+    }
+
+    public Builder setEncInfo(FileEncryptionInfo encInfo) {
+      this.encInfo = encInfo;
+      return this;
+    }
+
+    public Builder setFileChecksum(FileChecksum fileChecksum) {
+      this.fileChecksum = fileChecksum;
+      return this;
+    }
+
+    public OmMultipartPartInfo build() {
+      return new OmMultipartPartInfo(this);
+    }
+  }
+
+  public static OmMultipartPartInfo getFromProto(
+      MultipartPartInfo multipartPartInfo) {
+    validateRequiredProtoFields(multipartPartInfo);
+    Builder builder = new Builder()
+        .setPartName(multipartPartInfo.getPartName())
+        .setPartNumber(multipartPartInfo.getPartNumber())
+        .setDataSize(multipartPartInfo.getDataSize())
+        .setModificationTime(multipartPartInfo.getModificationTime())
+        .setETag(multipartPartInfo.getETag())
+        .setKeyLocationInfos(getKeyLocationInfosFromProto(multipartPartInfo))
+        .setEncInfo(null);
+
+    if (!multipartPartInfo.hasObjectID()) {
+      LOG.warn("MultipartPartInfo missing objectID for part {}",
+          multipartPartInfo.getPartNumber());
+    }
+    builder.setObjectID(multipartPartInfo.getObjectID());
+
+    if (!multipartPartInfo.hasUpdateID()) {
+      LOG.warn("MultipartPartInfo missing updateID for part {}",
+          multipartPartInfo.getPartNumber());
+    }
+    builder.setUpdateID(multipartPartInfo.getUpdateID());
+
+    if (multipartPartInfo.hasFileEncryptionInfo()) {
+      builder.setEncInfo(
+          OMPBHelper.convert(multipartPartInfo.getFileEncryptionInfo()));
+    }
+
+    if (multipartPartInfo.hasFileChecksum()) {
+      builder.setFileChecksum(
+          OMPBHelper.convert(multipartPartInfo.getFileChecksum()));
+    }
+
+    return builder.build();
+  }
+
+  public MultipartPartInfo getProto() {
+    if (StringUtils.isBlank(partName)) {
+      throw new IllegalArgumentException("partName is required");
+    }
+    if (partNumber <= 0) {
+      throw new IllegalArgumentException("partNumber is required and > 0");
+    }
+    if (dataSize < 0) {
+      throw new IllegalArgumentException("dataSize is required");
+    }
+    if (modificationTime <= 0) {
+      throw new IllegalArgumentException("modificationTime is required");
+    }
+    if (keyLocationInfos == null || keyLocationInfos.isEmpty()) {
+      throw new IllegalArgumentException("keyLocationList is required");
+    }
+    MultipartPartInfo.Builder builder = MultipartPartInfo.newBuilder()
+        .setPartName(partName)
+        .setPartNumber(partNumber)
+        .setKeyLocationList(getKeyLocationInfosAsProto())
+        .setDataSize(dataSize)
+        .setModificationTime(modificationTime)
+        .setObjectID(objectID)
+        .setUpdateID(updateID)
+        .setETag(Objects.requireNonNull(eTag, "eTag is required"));
+
+    if (encInfo != null) {
+      builder.setFileEncryptionInfo(OMPBHelper.convert(encInfo));
+    }
+    if (fileChecksum != null) {
+      builder.setFileChecksum(OMPBHelper.convert(fileChecksum));
+    }
+    return builder.build();
+  }
+
+  public String getPartName() {
+    return partName;
+  }
+
+  public int getPartNumber() {
+    return partNumber;
+  }
+
+  public long getDataSize() {
+    return dataSize;
+  }
+
+  public long getModificationTime() {
+    return modificationTime;
+  }
+
+  public long getObjectID() {
+    return objectID;
+  }
+
+  public long getUpdateID() {
+    return updateID;
+  }
+
+  public List<OmKeyLocationInfoGroup> getKeyLocationInfos() {
+    return keyLocationInfos;
+  }
+
+  public String getETag() {
+    return eTag;
+  }
+
+  public FileEncryptionInfo getEncInfo() {
+    return encInfo;
+  }
+
+  public FileChecksum getFileChecksum() {
+    return fileChecksum;
+  }
+
+  public static OmMultipartPartInfo from(
+      String partName, int partNumber, OmKeyInfo omKeyInfo) {
+    if (omKeyInfo.getObjectID() == 0L) {
+      LOG.warn("Multipart part {} has unset objectID", partNumber);
+    }
+    if (omKeyInfo.getUpdateID() == 0L) {
+      LOG.warn("Multipart part {} has unset updateID", partNumber);
+    }
+    Builder builder = new Builder()
+        .setPartName(partName)
+        .setPartNumber(partNumber)
+        .setDataSize(omKeyInfo.getDataSize())
+        .setModificationTime(omKeyInfo.getModificationTime())
+        .setObjectID(omKeyInfo.getObjectID())
+        .setUpdateID(omKeyInfo.getUpdateID())
+        .setKeyLocationInfos(omKeyInfo.getKeyLocationVersions())
+        .setEncInfo(omKeyInfo.getFileEncryptionInfo())
+        .setFileChecksum(omKeyInfo.getFileChecksum())
+        .setETag(omKeyInfo.getMetadata().get(OzoneConsts.ETAG));
+    return builder.build();
+  }
+
+  private KeyLocationList getKeyLocationInfosAsProto() {
+    if (keyLocationInfos == null || keyLocationInfos.isEmpty()) {
+      throw new IllegalArgumentException("keyLocationList is required");
+    }
+    return keyLocationInfos.get(0).getProtobuf(true, ClientVersion.CURRENT_VERSION);
+  }
+
+  private static List<OmKeyLocationInfoGroup> getKeyLocationInfosFromProto(
+      MultipartPartInfo multipartPartInfo) {
+    return Collections.singletonList(
+        OmKeyLocationInfoGroup.getFromProtobuf(
+            multipartPartInfo.getKeyLocationList()));
+  }
+
+  private static void validateRequiredProtoFields(MultipartPartInfo partInfo) {
+    if (!partInfo.hasPartName() || StringUtils.isBlank(partInfo.getPartName())) {
+      throw new IllegalArgumentException("MultipartPartInfo missing partName");
+    }
+    if (!partInfo.hasPartNumber()) {
+      throw new IllegalArgumentException("MultipartPartInfo missing partNumber");
+    }
+    if (!partInfo.hasETag() || StringUtils.isBlank(partInfo.getETag())) {
+      throw new IllegalArgumentException("MultipartPartInfo missing eTag");
+    }
+    if (!partInfo.hasKeyLocationList()) {
+      throw new IllegalArgumentException("MultipartPartInfo missing keyLocationList");
+    }
+    if (!partInfo.hasDataSize()) {
+      throw new IllegalArgumentException("MultipartPartInfo missing dataSize");
+    }
+    if (!partInfo.hasModificationTime()) {
+      throw new IllegalArgumentException("MultipartPartInfo missing modificationTime");
+    }
+  }
+}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartPartKey.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartPartKey.java
new file mode 100644
index 000000000000..92b71e471908
--- /dev/null
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartPartKey.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.helpers;
+
+import jakarta.annotation.Nonnull;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.hdds.utils.db.CodecBuffer;
+
+/**
+ * Typed key for multipart parts table.
+ * Key encoding:
+ * <pre>
+ *   uploadId(utf8) + '/' + partNumber(int32, big-endian)
+ * </pre>
+ * Prefix encoding for iteration:
+ * <pre>
+ *   uploadId(utf8) + '/'
+ * </pre>
+ */
+public final class OmMultipartPartKey {
+  private static final byte SEPARATOR = (byte) '/';
+  private static final Codec<OmMultipartPartKey> CODEC =
+      new OmMultipartPartKeyCodec();
+
+  private final String uploadId;
+  private final Integer partNumber;
+
+  private OmMultipartPartKey(String uploadId, Integer partNumber) {
+    this.uploadId = Objects.requireNonNull(uploadId, "uploadId is null");
+    this.partNumber = partNumber;
+  }
+
+  public static OmMultipartPartKey of(String uploadId, Integer partNumber) {
+    Objects.requireNonNull(partNumber, "partNumber is null");
+    return new OmMultipartPartKey(uploadId, partNumber);
+  }
+
+  public static OmMultipartPartKey prefix(String uploadId) {
+    return new OmMultipartPartKey(uploadId, null);
+  }
+
+  public static Codec<OmMultipartPartKey> getCodec() {
+    return CODEC;
+  }
+
+  public String getUploadId() {
+    return uploadId;
+  }
+
+  public Integer getPartNumber() {
+    return partNumber;
+  }
+
+  public boolean hasPartNumber() {
+    return partNumber != null;
+  }
+
+  @Override
+  public String toString() {
+    return hasPartNumber() ? uploadId + "/" + partNumber : uploadId;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof OmMultipartPartKey)) {
+      return false;
+    }
+    OmMultipartPartKey that = (OmMultipartPartKey) o;
+    return Objects.equals(uploadId, that.uploadId)
+        && Objects.equals(partNumber, that.partNumber);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(uploadId, partNumber);
+  }
+
+  private static final class OmMultipartPartKeyCodec
+      implements Codec<OmMultipartPartKey> {
+
+    @Override
+    public Class<OmMultipartPartKey> getTypeClass() {
+      return OmMultipartPartKey.class;
+    }
+
+    @Override
+    public boolean supportCodecBuffer() {
+      return true;
+    }
+
+    @Override
+    public CodecBuffer toCodecBuffer(
+        @Nonnull OmMultipartPartKey key, CodecBuffer.Allocator allocator) {
+      byte[] uploadBytes = key.uploadId.getBytes(StandardCharsets.UTF_8);
+      int size = uploadBytes.length + 1
+          + (key.hasPartNumber() ? Integer.BYTES : 0);
+      CodecBuffer buffer = allocator.apply(size);
+      buffer.put(ByteBuffer.wrap(uploadBytes)).put(SEPARATOR);
+      if (key.hasPartNumber()) {
+        buffer.putInt(key.partNumber);
+      }
+      return buffer;
+    }
+
+    @Override
+    public OmMultipartPartKey fromCodecBuffer(@Nonnull CodecBuffer buffer)
+        throws IllegalArgumentException {
+      return fromByteBuffer(buffer.asReadOnlyByteBuffer());
+    }
+
+    /**
+     * Encodes the OmMultipartPartKey object into a byte array for storage in the key/value store.
+     * Key format:
+     *   prefix key: uploadId + '/'
+     *   full key:   uploadId + '/' + int32(partNumber)
+     * @param key The original java object. Should not be null.
+     * @return Byte array representation of the object for storage in the key/value store.
+     */
+    @Override
+    public byte[] toPersistedFormat(OmMultipartPartKey key) {
+      byte[] uploadBytes = key.uploadId.getBytes(StandardCharsets.UTF_8);
+      int size = uploadBytes.length + 1
+          + (key.hasPartNumber() ? Integer.BYTES : 0);
+      ByteBuffer buffer = ByteBuffer.allocate(size);
+      buffer.put(uploadBytes);
+      buffer.put(SEPARATOR);
+      if (key.hasPartNumber()) {
+        buffer.putInt(key.partNumber);
+      }
+      return buffer.array();
+    }
+
+    /**
+     * Decodes the raw byte array from the key/value store into an OmMultipartPartKey object.
+     * @param rawData Byte array from the key/value store. Should not be null.
+     * @return OmMultipartPartKey object represented by the raw byte array.
+     * @throws IllegalArgumentException if the rawData format is invalid
+     */
+    @Override
+    public OmMultipartPartKey fromPersistedFormat(byte[] rawData) throws IllegalArgumentException {
+      return fromByteBuffer(ByteBuffer.wrap(rawData));
+    }
+
+    private OmMultipartPartKey fromByteBuffer(ByteBuffer rawData)
+        throws IllegalArgumentException {
+      final ByteBuffer input = rawData.asReadOnlyBuffer();
+      final int start = input.position();
+      final int length = input.remaining();
+      if (length == 0) {
+        throw new IllegalArgumentException(
+            "Invalid multipart part key: empty key");
+      }
+
+      //   prefix key: uploadId + '/'
+      //   full key:   uploadId + '/' + int32(partNumber)
+      int suffixLength = getSuffixLength(input, start, length);
+
+      int separatorIndex = start + length - suffixLength - 1;
+      if (separatorIndex < start) {
+        throw new IllegalArgumentException(
+            "Invalid multipart part key: invalid separator position");
+      }
+      final ByteBuffer uploadIdBuffer = input.duplicate();
+      uploadIdBuffer.limit(separatorIndex);
+      uploadIdBuffer.position(start);
+      String uploadId = StandardCharsets.UTF_8.decode(uploadIdBuffer).toString();
+      if (suffixLength == 0) {
+        return prefix(uploadId);
+      }
+      if (start + length - (separatorIndex + 1) != Integer.BYTES) {
+        throw new IllegalArgumentException(
+            "Invalid multipart part key: unexpected part suffix length");
+      }
+      int part = input.getInt(separatorIndex + 1);
+      return of(uploadId, part);
+    }
+
+    @Override
+    public OmMultipartPartKey copyObject(OmMultipartPartKey object) {
+      return object;
+    }
+  }
+
+  /**
+   * Determines the length of the suffix (part number) in the raw key data.
+   * Check full-key first: if byte at len - 5 (size of int is 4) is /, decode as full key (used to identify full row)
+   * Else, if byte at len - 1 is /, decode as prefix. (this is used for prefix scan for iterating)
+   * Else invalid.
+   * @param rawData the raw byte buffer representing the key
+   * @param start the position where key bytes start
+   * @param length the number of bytes in the key
+   * @return the length of the suffix (0 for prefix keys, Integer.BYTES for full keys)
+   * @throws IllegalArgumentException if the key format is invalid (missing separator or unexpected suffix length)
+   */
+  private static int getSuffixLength(ByteBuffer rawData, int start, int length)
+      throws IllegalArgumentException {
+    int suffixLength = -1;
+    // Check full-key layout first. Otherwise, part numbers whose low byte is
+    // '/' (for example 47 -> 0x0000002f) are mis-classified as prefix keys.
+    if (length > Integer.BYTES
+        && rawData.get(start + length - Integer.BYTES - 1) == SEPARATOR) {
+      suffixLength = Integer.BYTES;
+    } else if (rawData.get(start + length - 1) == SEPARATOR) {
+      suffixLength = 0;
+    }
+    if (suffixLength < 0) {
+      throw new IllegalArgumentException(
+          "Invalid multipart part key: missing separator");
+    }
+    return suffixLength;
+  }
+}
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java
index e03bf4454b8a..3a5658dd5bde 100644
--- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java
@@ -20,7 +20,10 @@
 import static java.util.stream.Collectors.toList;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import java.util.Collections;
+import java.util.TreeMap;
 import java.util.UUID;
 import java.util.stream.Stream;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -28,9 +31,11 @@
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PartKeyInfo;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.util.Time;
 import org.junit.jupiter.api.Test;
 
@@ -85,6 +90,8 @@ private void protoConversion(ReplicationConfig replicationConfig) {
     // THEN
     assertEquals(subject, fromProto);
     assertEquals(replicationConfig, fromProto.getReplicationConfig());
+    assertEquals(subject.getOwnerName(), fromProto.getOwnerName());
+    assertEquals(subject.getAcls(), fromProto.getAcls());
   }
 
   private static Stream<ReplicationConfig> replicationConfigs() {
@@ -110,9 +117,43 @@ public void distinctListOfParts() {
     assertEquals(1, subject.getPartKeyInfoMap().size());
   }
 
+  @Test
+  public void addPartKeyInfoRejectsSchemaVersionOne() {
+    OmMultipartKeyInfo subject = createSubject()
+        .setSchemaVersion((byte) 1)
+        .build();
+
+    assertThrows(IllegalStateException.class,
+        () -> subject.addPartKeyInfo(createPart(createKeyInfo()).build()));
+  }
+
+  @Test
+  public void getProtoRejectsLegacyPartListForSchemaVersionOne() {
+    PartKeyInfo part = createPart(createKeyInfo()).build();
+    TreeMap<Integer, PartKeyInfo> legacyMap = new TreeMap<>();
+    legacyMap.put(part.getPartNumber(), part);
+
+    OmMultipartKeyInfo subject = new OmMultipartKeyInfo.Builder()
+        .setUploadID(UUID.randomUUID().toString())
+        .setCreationTime(Time.now())
+        .setSchemaVersion((byte) 1)
+        .setReplicationConfig(StandaloneReplicationConfig.getInstance(
+            HddsProtos.ReplicationFactor.ONE))
+        .setPartKeyInfoList(legacyMap)
+        .build();
+
+    assertThrows(IllegalStateException.class, subject::getProto);
+  }
+
   private static OmMultipartKeyInfo.Builder createSubject() {
     return new OmMultipartKeyInfo.Builder()
         .setUploadID(UUID.randomUUID().toString())
+        .setOwnerName("mpu-owner")
+        .setAcls(Collections.singletonList(OzoneAcl.of(
+            IAccessAuthorizer.ACLIdentityType.USER,
+            "mpu-owner",
+            OzoneAcl.AclScope.ACCESS,
+            IAccessAuthorizer.ACLType.WRITE)))
         .setCreationTime(Time.now());
   }
 
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartPartInfo.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartPartInfo.java
new file mode 100644
index 000000000000..860ec4d80ba0
--- /dev/null
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartPartInfo.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.helpers;
+
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Collections;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocationList;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.MultipartPartInfo;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link OmMultipartPartInfo}.
+ */
+public class TestOmMultipartPartInfo {
+
+  @Test
+  public void testProtoRoundTrip() {
+    OmMultipartPartInfo partInfo = OmMultipartPartInfo.from(
+        "part-name", 1, createOmKeyInfoWithEtag("etag-1"));
+
+    MultipartPartInfo proto = partInfo.getProto();
+    OmMultipartPartInfo decoded = OmMultipartPartInfo.getFromProto(proto);
+
+    assertEquals(partInfo.getPartName(), decoded.getPartName());
+    assertEquals(partInfo.getPartNumber(), decoded.getPartNumber());
+    assertEquals(partInfo.getETag(), decoded.getETag());
+    assertEquals(partInfo.getDataSize(), decoded.getDataSize());
+    assertEquals(partInfo.getModificationTime(), decoded.getModificationTime());
+    assertEquals(partInfo.getObjectID(), decoded.getObjectID());
+    assertEquals(partInfo.getUpdateID(), decoded.getUpdateID());
+    assertEquals(partInfo.getKeyLocationInfos().size(),
+        decoded.getKeyLocationInfos().size());
+  }
+
+  @Test
+  public void testGetFromProtoRejectsMissingRequiredFields() {
+    MultipartPartInfo base = createValidProto();
+
+    assertThrows(IllegalArgumentException.class,
+        () -> OmMultipartPartInfo.getFromProto(base.toBuilder().clearPartName().build()));
+    assertThrows(IllegalArgumentException.class,
+        () -> OmMultipartPartInfo.getFromProto(base.toBuilder().clearPartNumber().build()));
+    assertThrows(IllegalArgumentException.class,
+        () -> OmMultipartPartInfo.getFromProto(base.toBuilder().clearETag().build()));
+    assertThrows(IllegalArgumentException.class,
+        () -> OmMultipartPartInfo.getFromProto(base.toBuilder().clearKeyLocationList().build()));
+    assertThrows(IllegalArgumentException.class,
+        () -> OmMultipartPartInfo.getFromProto(base.toBuilder().clearDataSize().build()));
+    assertThrows(IllegalArgumentException.class,
+        () -> OmMultipartPartInfo.getFromProto(base.toBuilder().clearModificationTime().build()));
+  }
+
+  @Test
+  public void testGetFromProtoAllowsMissingObjectAndUpdateId() {
+    MultipartPartInfo proto = createValidProto().toBuilder()
+        .clearObjectID()
+        .clearUpdateID()
+        .build();
+
+    OmMultipartPartInfo decoded = OmMultipartPartInfo.getFromProto(proto);
+    assertEquals(0L, decoded.getObjectID());
+    assertEquals(0L, decoded.getUpdateID());
+  }
+
+  @Test
+  public void testFromOmKeyInfoRejectsMissingETag() {
+    OmKeyInfo keyInfo = createOmKeyInfoWithoutEtag();
+    assertThrows(IllegalArgumentException.class,
+        () -> OmMultipartPartInfo.from("part-name", 1, keyInfo));
+  }
+
+  private static MultipartPartInfo createValidProto() {
+    return MultipartPartInfo.newBuilder()
+        .setPartName("part-1")
+        .setPartNumber(1)
+        .setETag("etag-1")
+        .setDataSize(100)
+        .setModificationTime(10)
+        .setObjectID(11)
+        .setUpdateID(12)
+        .setKeyLocationList(KeyLocationList.newBuilder().setVersion(0).build())
+        .build();
+  }
+
+  private static OmKeyInfo createOmKeyInfoWithEtag(String eTag) {
+    return baseOmKeyInfoBuilder()
+        .addMetadata(OzoneConsts.ETAG, eTag)
+        .build();
+  }
+
+  private static OmKeyInfo createOmKeyInfoWithoutEtag() {
+    return baseOmKeyInfoBuilder().build();
+  }
+
+  private static OmKeyInfo.Builder baseOmKeyInfoBuilder() {
+    OmKeyLocationInfo location = new OmKeyLocationInfo.Builder()
+        .setBlockID(new BlockID(1, 1))
+        .setPipeline(Pipeline.newBuilder()
+            .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+            .setId(PipelineID.randomId())
+            .setNodes(Collections.emptyList())
+            .setState(Pipeline.PipelineState.OPEN)
+            .build())
+        .build();
+    OmKeyLocationInfoGroup group = new OmKeyLocationInfoGroup(0,
+        Collections.singletonList(location));
+
+    return new OmKeyInfo.Builder()
+        .setVolumeName("vol")
+        .setBucketName("bucket")
+        .setKeyName("key")
+        .setCreationTime(1)
+        .setModificationTime(2)
+        .setDataSize(100)
+        .setObjectID(11)
+        .setUpdateID(12)
+        .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+        .setOmKeyLocationInfos(Collections.singletonList(group));
+  }
+}
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartPartKey.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartPartKey.java
new file mode 100644
index 000000000000..309f39a9aa47
--- /dev/null
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartPartKey.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.helpers;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.stream.IntStream;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.hdds.utils.db.CodecBuffer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Unit tests for {@link OmMultipartPartKey}.
+ */
+public class TestOmMultipartPartKey {
+
+  private final Codec<OmMultipartPartKey> codec = OmMultipartPartKey.getCodec();
+
+  /**
+   * Generate a stream of part numbers with low byte separator as 2F,
+   * in the supported MPU range [1, 10000], values with low byte 0x2f.
+   * AND operation with 0xFF gives the low byte of the part number.
+   * @return IntStream of part numbers with low byte separator as 2F.
+   */
+  private static IntStream partNumbersWithLowByteSeparator() {
+    return IntStream.rangeClosed(1, 10000)
+        .filter(partNumber -> (partNumber & 0xFF) == 0x2F);
+  }
+
+  private static int compareUnsignedBytes(byte[] left, byte[] right) {
+    int length = Math.min(left.length, right.length);
+    for (int i = 0; i < length; i++) {
+      int a = left[i] & 0xFF;
+      int b = right[i] & 0xFF;
+      if (a != b) {
+        return Integer.compare(a, b);
+      }
+    }
+    return Integer.compare(left.length, right.length);
+  }
+
+  @Test
+  public void testRoundTripFullKey() throws Exception {
+    OmMultipartPartKey key = OmMultipartPartKey.of("upload-123", 9);
+
+    byte[] raw = codec.toPersistedFormat(key);
+    OmMultipartPartKey decoded = codec.fromPersistedFormat(raw);
+
+    assertEquals(key, decoded);
+    assertTrue(decoded.hasPartNumber());
+    assertEquals("upload-123/9", decoded.toString());
+  }
+
+  @Test
+  public void testRoundTripPrefixKey() throws Exception {
+    OmMultipartPartKey key = OmMultipartPartKey.prefix("upload-abc");
+
+    byte[] raw = codec.toPersistedFormat(key);
+    OmMultipartPartKey decoded = codec.fromPersistedFormat(raw);
+
+    assertEquals(key, decoded);
+    assertFalse(decoded.hasPartNumber());
+    assertEquals("upload-abc", decoded.toString());
+  }
+
+  @Test
+  public void testCodecBufferRoundTripFullKey() throws Exception {
+    OmMultipartPartKey key = OmMultipartPartKey.of("upload-buffer", 1024);
+
+    assertTrue(codec.supportCodecBuffer());
+    try (CodecBuffer buffer = codec.toHeapCodecBuffer(key)) {
+      OmMultipartPartKey decoded = codec.fromCodecBuffer(buffer);
+      assertEquals(key, decoded);
+      assertTrue(decoded.hasPartNumber());
+      assertEquals(1024, decoded.getPartNumber().intValue());
+    }
+  }
+
+  @Test
+  public void testCodecBufferRoundTripPrefixKey() throws Exception {
+    OmMultipartPartKey key = OmMultipartPartKey.prefix("upload-prefix-buffer");
+
+    assertTrue(codec.supportCodecBuffer());
+    try (CodecBuffer buffer = codec.toHeapCodecBuffer(key)) {
+      OmMultipartPartKey decoded = codec.fromCodecBuffer(buffer);
+      assertEquals(key, decoded);
+      assertFalse(decoded.hasPartNumber());
+    }
+  }
+
+  // Test to validate that the full key is decoded correctly
+  // when the part number has a low byte same as the separator byte i.e. 2F.
+  @ParameterizedTest
+  @MethodSource("partNumbersWithLowByteSeparator")
+  public void testDecodeFullKeyWhenPartLowByteIsSeparator(int partNumber)
+      throws Exception {
+    OmMultipartPartKey key = OmMultipartPartKey.of("upload-sep", partNumber);
+
+    byte[] raw = codec.toPersistedFormat(key);
+    OmMultipartPartKey decoded = codec.fromPersistedFormat(raw);
+
+    assertTrue(decoded.hasPartNumber());
+    assertEquals(partNumber, decoded.getPartNumber().intValue());
+    assertEquals("upload-sep", decoded.getUploadId());
+  }
+
+  @Test
+  public void testDecodeRejectsInvalidKeyWithoutSeparator() {
+    assertThrows(IllegalArgumentException.class,
+        () -> codec.fromPersistedFormat("invalid".getBytes(UTF_8)));
+  }
+
+  @Test
+  public void testDecodeRejectsEmptyKey() {
+    assertThrows(IllegalArgumentException.class,
+        () -> codec.fromPersistedFormat(new byte[0]));
+  }
+
+  @Test
+  public void testCodecBufferDecodeRejectsInvalidKeyWithoutSeparator() {
+    try (CodecBuffer buffer = CodecBuffer.wrap("invalid".getBytes(UTF_8))) {
+      assertThrows(IllegalArgumentException.class,
+          () -> codec.fromCodecBuffer(buffer));
+    }
+  }
+
+  @Test
+  public void testCodecBufferDecodeRejectsEmptyKey() {
+    try (CodecBuffer buffer = CodecBuffer.wrap(new byte[0])) {
+      assertThrows(IllegalArgumentException.class,
+          () -> codec.fromCodecBuffer(buffer));
+    }
+  }
+
+  @Test
+  public void testDecodeRejectsMalformedKeyWithMiddleSeparatorOnly() {
+    byte[] malformed = "up/xx".getBytes(UTF_8);
+    assertThrows(IllegalArgumentException.class,
+        () -> codec.fromPersistedFormat(malformed));
+  }
+
+  @Test
+  public void testFactoryRejectsNullPartNumber() {
+    assertThrows(NullPointerException.class,
+        () -> OmMultipartPartKey.of("upload-id", null));
+  }
+
+  @Test
+  public void testFactoryRejectsNullUploadId() {
+    assertThrows(NullPointerException.class,
+        () -> OmMultipartPartKey.of(null, 1));
+    assertThrows(NullPointerException.class,
+        () -> OmMultipartPartKey.prefix(null));
+  }
+
+  @Test
+  public void testDecodeRejectsNullRawData() {
+    assertThrows(NullPointerException.class,
+        () -> codec.fromPersistedFormat(null));
+  }
+
+  @Test
+  public void testSortOrderIsNumericalNotLexicographic() throws Exception {
+    // "9" sorts after "10" lexicographically, but 9 < 10 numerically.
+    // Big-endian int32 encoding must produce the numerical order.
+    byte[] raw9  = codec.toPersistedFormat(OmMultipartPartKey.of("uid", 9));
+    byte[] raw10 = codec.toPersistedFormat(OmMultipartPartKey.of("uid", 10));
+
+    assertTrue(compareUnsignedBytes(raw9, raw10) < 0,
+        "part 9 must sort before part 10 in byte order (big-endian encoding)");
+  }
+
+  @Test
+  public void testUploadIdContainingSlashRoundTrips() throws Exception {
+    // uploadId itself contains '/' — the codec separator character.
+    // A naive "find first slash" parser would split at the wrong position.
+    OmMultipartPartKey key = OmMultipartPartKey.of("upload/with/slashes", 5);
+    byte[] raw = codec.toPersistedFormat(key);
+    OmMultipartPartKey decoded = codec.fromPersistedFormat(raw);
+    assertEquals("upload/with/slashes", decoded.getUploadId());
+    assertEquals(5, decoded.getPartNumber().intValue());
+  }
+}
diff --git a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java
index 461324d00221..be66ffc195b5 100644
--- a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java
+++ b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java
@@ -48,6 +48,8 @@
 import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartKey;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUpload;
 import org.apache.hadoop.ozone.om.helpers.OmPrefixInfo;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
@@ -471,6 +473,13 @@ String getMultipartKeyFSO(String volume, String bucket, String key, String
    */
   Table<String, OmMultipartKeyInfo> getMultipartInfoTable();
 
+  /**
+   * Gets the multipart part info table which holds
+   * the information about the parts of a Multipart Upload Key.
+   * @return Table
+   */
+  Table<OmMultipartPartKey, OmMultipartPartInfo> getMultipartPartsTable();
+
   @Override
   Table<String, TransactionInfo> getTransactionInfoTable();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
index 9dc01e54f997..98c693584a67 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
@@ -108,6 +108,8 @@
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartKey;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUpload;
 import org.apache.hadoop.ozone.om.helpers.OmPrefixInfo;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
@@ -160,6 +162,7 @@ public class OmMetadataManagerImpl implements OMMetadataManager,
 
   private Table<String, OmKeyInfo> openKeyTable;
   private Table<String, OmMultipartKeyInfo> multipartInfoTable;
+  private Table<OmMultipartPartKey, OmMultipartPartInfo> multipartPartsTable;
   private Table<String, RepeatedOmKeyInfo> deletedTable;
 
   private Table<String, OmDirectoryInfo> dirTable;
@@ -386,6 +389,11 @@ public Table<String, OmMultipartKeyInfo> getMultipartInfoTable() {
     return multipartInfoTable;
   }
 
+  @Override
+  public Table<OmMultipartPartKey, OmMultipartPartInfo> getMultipartPartsTable() {
+    return multipartPartsTable;
+  }
+
   /**
    * Start metadata manager.
    */
@@ -464,6 +472,7 @@ protected void initializeOmTables(CacheType cacheType,
 
     openKeyTable = initializer.get(OMDBDefinition.OPEN_KEY_TABLE_DEF);
     multipartInfoTable = initializer.get(OMDBDefinition.MULTIPART_INFO_TABLE_DEF);
+    multipartPartsTable = initializer.get(OMDBDefinition.MULTIPART_PARTS_TABLE_DEF);
     deletedTable = initializer.get(OMDBDefinition.DELETED_TABLE_DEF);
 
     dirTable = initializer.get(OMDBDefinition.DIRECTORY_TABLE_DEF);
@@ -1521,6 +1530,7 @@ public List<ExpiredMultipartUploadsBucket> getExpiredMultipartUploads(
               .addMultipartUploads(builder.setName(dbMultipartInfoKey)
                   .build());
           numParts += omMultipartKeyInfo.getPartKeyInfoMap().size();
+          // TODO: Add the expired part handling from the new table when the complete flow is done
         }
 
       }
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/codec/OMDBDefinition.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/codec/OMDBDefinition.java
index 9894e8f5d6bf..323f11926af9 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/codec/OMDBDefinition.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/codec/OMDBDefinition.java
@@ -35,6 +35,8 @@
 import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartKey;
 import org.apache.hadoop.ozone.om.helpers.OmPrefixInfo;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
@@ -80,14 +82,15 @@
  * <pre>
  * {@code
  * Object Store (OBS) Tables:
- * |-----------------------------------------------------------------------|
- * |        Column Family |                           Mapping              |
- * |-----------------------------------------------------------------------|
- * |             keyTable | /volume/bucket/key          :- KeyInfo         |
- * |         deletedTable | /volume/bucket/key          :- RepeatedKeyInfo |
- * |         openKeyTable | /volume/bucket/key/id       :- KeyInfo         |
- * |   multipartInfoTable | /volume/bucket/key/uploadId :- parts           |
- * |-----------------------------------------------------------------------|
+ * |----------------------------------------------------------------------------------|
+ * |        Column Family |                           Mapping                         |
+ * |----------------------------------------------------------------------------------|
+ * |             keyTable | /volume/bucket/key                     :- KeyInfo         |
+ * |         deletedTable | /volume/bucket/key                     :- RepeatedKeyInfo |
+ * |         openKeyTable | /volume/bucket/key/id                  :- KeyInfo         |
+ * |   multipartInfoTable | /volume/bucket/key/uploadId            :- parts           |
+ * |  multipartPartsTable | uploadId/partNumber                    :- PartKeyInfo     |
+ * |----------------------------------------------------------------------------------|
  * }
  * </pre>
  * Note that "volume", "bucket" and "key" in OBS tables are names.
@@ -228,6 +231,13 @@ public final class OMDBDefinition extends DBDefinition.WithMap {
           StringCodec.get(),
           OmMultipartKeyInfo.getCodec());
 
+  public static final String MULTIPART_PARTS_TABLE = "multipartPartsTable";
+  /** multipartPartsTable: uploadId/partNumber :- PartKeyInfo. */
+  public static final DBColumnFamilyDefinition<OmMultipartPartKey, OmMultipartPartInfo> MULTIPART_PARTS_TABLE_DEF
+      = new DBColumnFamilyDefinition<>(MULTIPART_PARTS_TABLE,
+      OmMultipartPartKey.getCodec(),
+      OmMultipartPartInfo.getCodec());
+
   //---------------------------------------------------------------------------
   // File System Optimized (FSO) Tables:
   public static final String FILE_TABLE = "fileTable";
@@ -327,6 +337,7 @@ public final class OMDBDefinition extends DBDefinition.WithMap {
           KEY_TABLE_DEF,
           META_TABLE_DEF,
           MULTIPART_INFO_TABLE_DEF,
+          MULTIPART_PARTS_TABLE_DEF,
           OPEN_FILE_TABLE_DEF,
           OPEN_KEY_TABLE_DEF,
           PREFIX_TABLE_DEF,
@@ -369,4 +380,3 @@ public static List<String> getAllColumnFamilies() {
     return columnFamilies;
   }
 }
-
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
index d905fb23461c..fc2a9ca78b01 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
@@ -34,6 +34,7 @@
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.KEY_TABLE;
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.META_TABLE;
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.MULTIPART_INFO_TABLE;
+import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.MULTIPART_PARTS_TABLE;
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_FILE_TABLE;
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_KEY_TABLE;
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.PREFIX_TABLE;
@@ -122,6 +123,7 @@ public class TestOmMetadataManager {
       DELETED_TABLE,
       OPEN_KEY_TABLE,
       MULTIPART_INFO_TABLE,
+      MULTIPART_PARTS_TABLE,
       S3_SECRET_TABLE,
       DELEGATION_TOKEN_TABLE,
       PREFIX_TABLE,