Repository: Qiskit/qiskit-terra (python)

Issue:
QPY fails to deserialize some `ControlledGate`s
### Environment

- **Qiskit Terra version**: 0.20.1
- **Python version**: 3.8.12
- **Operating system**: Ubuntu 20.04.4


### What is happening?

QPY successfully serializes circuits containing instances of `ControlledGate`, but deserialization fails when `ControlledGate.__init__(...)` is called in `_read_instruction` with e.g.:
```
File "<..>/qiskit/qpy/binary_io/circuits.py", line 250, in _read_instruction
    gate = gate_class(*params)

TypeError: __init__() missing 3 required positional arguments: 'name', 'num_qubits', and 'params'
```

### How can we reproduce the issue?

```
from io import BytesIO
from qiskit import qpy, QuantumCircuit
from qiskit.circuit.library import DCXGate

controlled_gate = DCXGate().control(1)  # create instance of ControlledGate

circuit = QuantumCircuit(3)
circuit.append(controlled_gate, [0, 1, 2])

buf = BytesIO()
qpy.dump(circuit, buf)  # ok
qpy.load(BytesIO(buf.getbuffer()))  # TypeError: __init__() missing 3 required positional arguments: 'name',          'num_qubits', and 'params'
```

### What should happen?

`qpy.load()` should succeed

### Any suggestions?

_No response_

Pull request description:
Fix handling of ControlledGates in QPY
<!--
⚠️ If you do not respect this template, your pull request will be closed.
⚠️ Your pull request title should be short detailed and understandable for all.
⚠️ Also, please add a release note file using reno if the change needs to be
  documented in the release notes.
⚠️ If your pull request fixes an open issue, please link to the issue.

- [ ] I have added the tests to cover my changes.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the CONTRIBUTING document.
-->

### Summary

This commit fixes the handling of ControlledGates in QPY. Previously the
extra parameters needed to reconstruct a custom controlled gate were not
encoded into the QPY payload. Fixing this required a version bump to the
QPY format to modify the payload for a custom instruction entry. Once we
added to the format the extra data required for a controlled gate, the
number of control qubits, the control state, and the base gate object,
the deserializer has enough information to recreate the custom
ControlledGate objects. However, fixing this exposed another bug with
standard library multicontrolled gates where they often didn't contain
sufficient data in the payload to reconstruct either.

### Details and comments

Fixes #7999

TODO:

- [x]  Fix failing unittest
- [x] Add format docs for QPY v5
- [x] Add release note

Patch:
diff --git a/qiskit/qpy/__init__.py b/qiskit/qpy/__init__.py
index 74ef713a7..cd79d0291 100644
--- a/qiskit/qpy/__init__.py
+++ b/qiskit/qpy/__init__.py
@@ -100,6 +100,82 @@ There is a circuit payload for each circuit (where the total number is dictated
 by ``num_circuits`` in the file header). There is no padding between the
 circuits in the data.
 
+.. _qpy_version_5:
+
+Version 5
+=========
+
+Version 5 changes from :ref:`qpy_version_4` by changing two payloads the INSTRUCTION metadata
+payload and the CUSTOM_INSTRUCTION block. These now have new fields to better account for
+:class:`~.ControlledGate` objects in a circuit.
+
+INSTRUCTION
+-----------
+
+The INSTRUCTION block was modified to add two new fields ``num_ctrl_qubits`` and ``ctrl_state``
+which are used to model the :attr:`.ControlledGate.num_ctrl_qubits` and
+:attr:`.ControlledGate.ctrl_state` attributes. The new payload packed struct
+format is:
+
+.. code-block:: c
+
+    struct {
+        uint16_t name_size;
+        uint16_t label_size;
+        uint16_t num_parameters;
+        uint32_t num_qargs;
+        uint32_t num_cargs;
+        _Bool has_conditional;
+        uint16_t conditional_reg_name_size;
+        int64_t conditional_value;
+        uint32_t num_ctrl_qubits;
+        uint32_t ctrl_state;
+    }
+
+The rest of the instruction payload is the same. You can refer to
+:ref:`qpy_instructions` for the details of the full payload.
+
+CUSTOM_INSTRUCTION
+------------------
+
+The CUSTOM_INSTRUCTION block in QPY version 5 adds a new field
+``base_gate_size`` which is used to define the size of the
+:class:`qiskit.circuit.Instruction` object stored in the
+:attr:`.ControlledGate.base_gate` attribute for a custom
+:class:`~.ControlledGate` object. With this change the CUSTOM_INSTRUCTION
+metadata block becomes:
+
+.. code-block:: c
+
+    struct {
+        uint16_t name_size;
+        char type;
+        uint32_t num_qubits;
+        uint32_t num_clbits;
+        _Bool custom_definition;
+        uint64_t size;
+        uint32_t num_ctrl_qubits;
+        uint32_t ctrl_state;
+        uint64_t base_gate_size
+    }
+
+Immediately following the CUSTOM_INSTRUCTION struct is the utf8 encoded name
+of size ``name_size``.
+
+If ``custom_definition`` is ``True`` that means that the immediately following
+``size`` bytes contains a QPY circuit data which can be used for the custom
+definition of that gate. If ``custom_definition`` is ``False`` then the
+instruction can be considered opaque (ie no definition). The ``type`` field
+determines what type of object will get created with the custom definition.
+If it's ``'g'`` it will be a :class:`~qiskit.circuit.Gate` object, ``'i'``
+it will be a :class:`~qiskit.circuit.Instruction` object.
+
+Following this the next ``base_gate_size`` bytes contain the ``INSTRUCTION``
+payload for the :attr:`.ControlledGate.base_gate`.
+
+Additionally an addition value for ``type`` is added ``'c'`` which is used to
+indicate the custom instruction is a custom :class:`~.ControlledGate`.
+
 .. _qpy_version_4:
 
 Version 4
@@ -455,6 +531,8 @@ Each custom instruction is defined with a CUSTOM_INSTRUCTION block defined as:
     struct {
         uint16_t name_size;
         char type;
+        uint32_t num_qubits;
+        uint32_t num_clbits;
         _Bool custom_definition;
         uint64_t size;
     }
@@ -470,6 +548,8 @@ determines what type of object will get created with the custom definition.
 If it's ``'g'`` it will be a :class:`~qiskit.circuit.Gate` object, ``'i'``
 it will be a :class:`~qiskit.circuit.Instruction` object.
 
+.. _qpy_instructions:
+
 INSTRUCTIONS
 ------------
 
diff --git a/qiskit/qpy/binary_io/circuits.py b/qiskit/qpy/binary_io/circuits.py
index e09515263..408942a7e 100644
--- a/qiskit/qpy/binary_io/circuits.py
+++ b/qiskit/qpy/binary_io/circuits.py
@@ -27,6 +27,7 @@ from qiskit import extensions
 from qiskit.circuit import library, controlflow
 from qiskit.circuit.classicalregister import ClassicalRegister, Clbit
 from qiskit.circuit.gate import Gate
+from qiskit.circuit.controlledgate import ControlledGate
 from qiskit.circuit.instruction import Instruction
 from qiskit.circuit.quantumcircuit import QuantumCircuit
 from qiskit.circuit.quantumregister import QuantumRegister, Qubit
@@ -150,12 +151,20 @@ def _read_instruction_parameter(file_obj, version, vectors):
 
 
 def _read_instruction(file_obj, circuit, registers, custom_instructions, version, vectors):
-    instruction = formats.CIRCUIT_INSTRUCTION._make(
-        struct.unpack(
-            formats.CIRCUIT_INSTRUCTION_PACK,
-            file_obj.read(formats.CIRCUIT_INSTRUCTION_SIZE),
+    if version < 5:
+        instruction = formats.CIRCUIT_INSTRUCTION._make(
+            struct.unpack(
+                formats.CIRCUIT_INSTRUCTION_PACK,
+                file_obj.read(formats.CIRCUIT_INSTRUCTION_SIZE),
+            )
+        )
+    else:
+        instruction = formats.CIRCUIT_INSTRUCTION_V2._make(
+            struct.unpack(
+                formats.CIRCUIT_INSTRUCTION_V2_PACK,
+                file_obj.read(formats.CIRCUIT_INSTRUCTION_V2_SIZE),
+            )
         )
-    )
     gate_name = file_obj.read(instruction.name_size).decode(common.ENCODE)
     label = file_obj.read(instruction.label_size).decode(common.ENCODE)
     condition_register = file_obj.read(instruction.condition_register_size).decode(common.ENCODE)
@@ -179,30 +188,35 @@ def _read_instruction(file_obj, circuit, registers, custom_instructions, version
                 )
         else:
             condition_tuple = (registers["c"][condition_register], instruction.condition_value)
-    qubit_indices = dict(enumerate(circuit.qubits))
-    clbit_indices = dict(enumerate(circuit.clbits))
+    if circuit is not None:
+        qubit_indices = dict(enumerate(circuit.qubits))
+        clbit_indices = dict(enumerate(circuit.clbits))
+    else:
+        qubit_indices = {}
+        clbit_indices = {}
 
     # Load Arguments
-    for _qarg in range(instruction.num_qargs):
-        qarg = formats.CIRCUIT_INSTRUCTION_ARG._make(
-            struct.unpack(
-                formats.CIRCUIT_INSTRUCTION_ARG_PACK,
-                file_obj.read(formats.CIRCUIT_INSTRUCTION_ARG_SIZE),
+    if circuit is not None:
+        for _qarg in range(instruction.num_qargs):
+            qarg = formats.CIRCUIT_INSTRUCTION_ARG._make(
+                struct.unpack(
+                    formats.CIRCUIT_INSTRUCTION_ARG_PACK,
+                    file_obj.read(formats.CIRCUIT_INSTRUCTION_ARG_SIZE),
+                )
             )
-        )
-        if qarg.type.decode(common.ENCODE) == "c":
-            raise TypeError("Invalid input carg prior to all qargs")
-        qargs.append(qubit_indices[qarg.size])
-    for _carg in range(instruction.num_cargs):
-        carg = formats.CIRCUIT_INSTRUCTION_ARG._make(
-            struct.unpack(
-                formats.CIRCUIT_INSTRUCTION_ARG_PACK,
-                file_obj.read(formats.CIRCUIT_INSTRUCTION_ARG_SIZE),
+            if qarg.type.decode(common.ENCODE) == "c":
+                raise TypeError("Invalid input carg prior to all qargs")
+            qargs.append(qubit_indices[qarg.size])
+        for _carg in range(instruction.num_cargs):
+            carg = formats.CIRCUIT_INSTRUCTION_ARG._make(
+                struct.unpack(
+                    formats.CIRCUIT_INSTRUCTION_ARG_PACK,
+                    file_obj.read(formats.CIRCUIT_INSTRUCTION_ARG_SIZE),
+                )
             )
-        )
-        if carg.type.decode(common.ENCODE) == "q":
-            raise TypeError("Invalid input qarg after all qargs")
-        cargs.append(clbit_indices[carg.size])
+            if carg.type.decode(common.ENCODE) == "q":
+                raise TypeError("Invalid input qarg after all qargs")
+            cargs.append(clbit_indices[carg.size])
 
     # Load Parameters
     for _param in range(instruction.num_parameters):
@@ -210,20 +224,28 @@ def _read_instruction(file_obj, circuit, registers, custom_instructions, version
         params.append(param)
 
     # Load Gate object
-    if gate_name in ("Gate", "Instruction"):
-        inst_obj = _parse_custom_instruction(custom_instructions, gate_name, params)
+    if gate_name in {"Gate", "Instruction", "ControlledGate"}:
+        inst_obj = _parse_custom_instruction(
+            custom_instructions, gate_name, params, version, vectors, registers
+        )
         inst_obj.condition = condition_tuple
         if instruction.label_size > 0:
             inst_obj.label = label
+        if circuit is None:
+            return inst_obj
         circuit._append(inst_obj, qargs, cargs)
-        return
+        return None
     elif gate_name in custom_instructions:
-        inst_obj = _parse_custom_instruction(custom_instructions, gate_name, params)
+        inst_obj = _parse_custom_instruction(
+            custom_instructions, gate_name, params, version, vectors, registers
+        )
         inst_obj.condition = condition_tuple
         if instruction.label_size > 0:
             inst_obj.label = label
+        if circuit is None:
+            return inst_obj
         circuit._append(inst_obj, qargs, cargs)
-        return
+        return None
     elif hasattr(library, gate_name):
         gate_class = getattr(library, gate_name)
     elif hasattr(circuit_mod, gate_name):
@@ -239,6 +261,14 @@ def _read_instruction(file_obj, circuit, registers, custom_instructions, version
 
     if gate_name in {"IfElseOp", "WhileLoopOp"}:
         gate = gate_class(condition_tuple, *params)
+    elif version >= 5 and issubclass(gate_class, ControlledGate):
+        if gate_name in {"MCPhaseGate", "MCU1Gate"}:
+            gate = gate_class(*params, instruction.num_ctrl_qubits)
+        else:
+            gate = gate_class(*params)
+            gate.num_ctrl_qubits = instruction.num_ctrl_qubits
+            gate.ctrl_state = instruction.ctrl_state
+        gate.condition = condition_tuple
     else:
         if gate_name in {"Initialize", "UCRXGate", "UCRYGate", "UCRZGate"}:
             gate = gate_class(params)
@@ -251,14 +281,28 @@ def _read_instruction(file_obj, circuit, registers, custom_instructions, version
         gate.condition = condition_tuple
     if instruction.label_size > 0:
         gate.label = label
+    if circuit is None:
+        return gate
     if not isinstance(gate, Instruction):
         circuit.append(gate, qargs, cargs)
     else:
         circuit._append(gate, qargs, cargs)
+    return None
 
 
-def _parse_custom_instruction(custom_instructions, gate_name, params):
-    type_str, num_qubits, num_clbits, definition = custom_instructions[gate_name]
+def _parse_custom_instruction(custom_instructions, gate_name, params, version, vectors, registers):
+    if version >= 5:
+        (
+            type_str,
+            num_qubits,
+            num_clbits,
+            definition,
+            num_ctrl_qubits,
+            ctrl_state,
+            base_gate_raw,
+        ) = custom_instructions[gate_name]
+    else:
+        type_str, num_qubits, num_clbits, definition = custom_instructions[gate_name]
     type_key = common.CircuitInstructionTypeKey(type_str)
 
     if type_key == common.CircuitInstructionTypeKey.INSTRUCTION:
@@ -272,6 +316,22 @@ def _parse_custom_instruction(custom_instructions, gate_name, params):
         inst_obj.definition = definition
         return inst_obj
 
+    if version >= 5 and type_key == common.CircuitInstructionTypeKey.CONTROLLED_GATE:
+        with io.BytesIO(base_gate_raw) as base_gate_obj:
+            base_gate = _read_instruction(
+                base_gate_obj, None, registers, custom_instructions, version, vectors
+            )
+        inst_obj = ControlledGate(
+            gate_name,
+            num_qubits,
+            params,
+            num_ctrl_qubits=num_ctrl_qubits,
+            ctrl_state=ctrl_state,
+            base_gate=base_gate,
+        )
+        inst_obj.definition = definition
+        return inst_obj
+
     if type_key == common.CircuitInstructionTypeKey.PAULI_EVOL_GATE:
         return definition
 
@@ -327,12 +387,21 @@ def _read_custom_instructions(file_obj, version, vectors):
     )
     if custom_definition_header.size > 0:
         for _ in range(custom_definition_header.size):
-            data = formats.CUSTOM_CIRCUIT_INST_DEF._make(
-                struct.unpack(
-                    formats.CUSTOM_CIRCUIT_INST_DEF_PACK,
-                    file_obj.read(formats.CUSTOM_CIRCUIT_INST_DEF_SIZE),
+            if version < 5:
+                data = formats.CUSTOM_CIRCUIT_INST_DEF._make(
+                    struct.unpack(
+                        formats.CUSTOM_CIRCUIT_INST_DEF_PACK,
+                        file_obj.read(formats.CUSTOM_CIRCUIT_INST_DEF_SIZE),
+                    )
                 )
-            )
+            else:
+                data = formats.CUSTOM_CIRCUIT_INST_DEF_V2._make(
+                    struct.unpack(
+                        formats.CUSTOM_CIRCUIT_INST_DEF_V2_PACK,
+                        file_obj.read(formats.CUSTOM_CIRCUIT_INST_DEF_V2_SIZE),
+                    )
+                )
+
             name = file_obj.read(data.gate_name_size).decode(common.ENCODE)
             type_str = data.type
             definition_circuit = None
@@ -346,12 +415,20 @@ def _read_custom_instructions(file_obj, version, vectors):
                     definition_circuit = common.data_from_binary(
                         def_binary, _read_pauli_evolution_gate, version=version, vectors=vectors
                     )
-            custom_instructions[name] = (
-                type_str,
-                data.num_qubits,
-                data.num_clbits,
-                definition_circuit,
-            )
+            if version < 5:
+                data_payload = (type_str, data.num_qubits, data.num_clbits, definition_circuit)
+            else:
+                base_gate = file_obj.read(data.base_gate_size)
+                data_payload = (
+                    type_str,
+                    data.num_qubits,
+                    data.num_clbits,
+                    definition_circuit,
+                    data.num_ctrl_qubits,
+                    data.ctrl_state,
+                    base_gate,
+                )
+            custom_instructions[name] = data_payload
     return custom_instructions
 
 
@@ -381,6 +458,7 @@ def _write_instruction_parameter(file_obj, param):
 # pylint: disable=too-many-boolean-expressions
 def _write_instruction(file_obj, instruction_tuple, custom_instructions, index_map):
     gate_class_name = instruction_tuple[0].__class__.__name__
+    custom_instructions_list = []
     if (
         (
             not hasattr(library, gate_class_name)
@@ -391,15 +469,18 @@ def _write_instruction(file_obj, instruction_tuple, custom_instructions, index_m
         )
         or gate_class_name == "Gate"
         or gate_class_name == "Instruction"
+        or gate_class_name == "ControlledGate"
         or isinstance(instruction_tuple[0], library.BlueprintCircuit)
     ):
         if instruction_tuple[0].name not in custom_instructions:
             custom_instructions[instruction_tuple[0].name] = instruction_tuple[0]
+            custom_instructions_list.append(instruction_tuple[0].name)
         gate_class_name = instruction_tuple[0].name
 
     elif isinstance(instruction_tuple[0], library.PauliEvolutionGate):
         gate_class_name = r"###PauliEvolutionGate_" + str(uuid.uuid4())
         custom_instructions[gate_class_name] = instruction_tuple[0]
+        custom_instructions_list.append(gate_class_name)
 
     has_condition = False
     condition_register = b""
@@ -420,8 +501,11 @@ def _write_instruction(file_obj, instruction_tuple, custom_instructions, index_m
         label_raw = label.encode(common.ENCODE)
     else:
         label_raw = b""
+
+    num_ctrl_qubits = getattr(instruction_tuple[0], "num_ctrl_qubits", 0)
+    ctrl_state = getattr(instruction_tuple[0], "ctrl_state", 0)
     instruction_raw = struct.pack(
-        formats.CIRCUIT_INSTRUCTION_PACK,
+        formats.CIRCUIT_INSTRUCTION_V2_PACK,
         len(gate_class_name),
         len(label_raw),
         len(instruction_tuple[0].params),
@@ -430,6 +514,8 @@ def _write_instruction(file_obj, instruction_tuple, custom_instructions, index_m
         has_condition,
         len(condition_register),
         condition_value,
+        num_ctrl_qubits,
+        ctrl_state,
     )
     file_obj.write(instruction_raw)
     file_obj.write(gate_class_name)
@@ -449,6 +535,7 @@ def _write_instruction(file_obj, instruction_tuple, custom_instructions, index_m
     # Encode instruction params
     for param in instruction_tuple[0].params:
         _write_instruction_parameter(file_obj, param)
+    return custom_instructions_list
 
 
 def _write_pauli_evolution_gate(file_obj, evolution_gate):
@@ -491,13 +578,17 @@ def _write_pauli_evolution_gate(file_obj, evolution_gate):
     file_obj.write(synth_data)
 
 
-def _write_custom_instruction(file_obj, name, instruction):
+def _write_custom_instruction(file_obj, name, instruction, custom_instructions):
     type_key = common.CircuitInstructionTypeKey.assign(instruction)
     has_definition = False
     size = 0
     data = None
     num_qubits = instruction.num_qubits
     num_clbits = instruction.num_clbits
+    ctrl_state = 0
+    num_ctrl_qubits = 0
+    base_gate = None
+    new_custom_instruction = []
 
     if type_key == common.CircuitInstructionTypeKey.PAULI_EVOL_GATE:
         has_definition = True
@@ -507,20 +598,37 @@ def _write_custom_instruction(file_obj, name, instruction):
         has_definition = True
         data = common.data_to_binary(instruction.definition, write_circuit)
         size = len(data)
+    if type_key == common.CircuitInstructionTypeKey.CONTROLLED_GATE:
+        num_ctrl_qubits = instruction.num_ctrl_qubits
+        ctrl_state = instruction.ctrl_state
+        base_gate = instruction.base_gate
+    if base_gate is None:
+        base_gate_raw = b""
+    else:
+        with io.BytesIO() as base_gate_buffer:
+            new_custom_instruction = _write_instruction(
+                base_gate_buffer, (base_gate, [], []), custom_instructions, {}
+            )
+            base_gate_raw = base_gate_buffer.getvalue()
     name_raw = name.encode(common.ENCODE)
     custom_instruction_raw = struct.pack(
-        formats.CUSTOM_CIRCUIT_INST_DEF_PACK,
+        formats.CUSTOM_CIRCUIT_INST_DEF_V2_PACK,
         len(name_raw),
         type_key,
         num_qubits,
         num_clbits,
         has_definition,
         size,
+        num_ctrl_qubits,
+        ctrl_state,
+        len(base_gate_raw),
     )
     file_obj.write(custom_instruction_raw)
     file_obj.write(name_raw)
     if data:
         file_obj.write(data)
+    file_obj.write(base_gate_raw)
+    return new_custom_instruction
 
 
 def _write_registers(file_obj, in_circ_regs, full_bits):
@@ -612,13 +720,23 @@ def write_circuit(file_obj, circuit, metadata_serializer=None):
     index_map["c"] = {bit: index for index, bit in enumerate(circuit.clbits)}
     for instruction in circuit.data:
         _write_instruction(instruction_buffer, instruction, custom_instructions, index_map)
-    file_obj.write(struct.pack(formats.CUSTOM_CIRCUIT_DEF_HEADER_PACK, len(custom_instructions)))
 
-    for name, instruction in custom_instructions.items():
-        _write_custom_instruction(file_obj, name, instruction)
+    with io.BytesIO() as custom_instructions_buffer:
+        new_custom_instructions = list(custom_instructions.keys())
+        while new_custom_instructions:
+            instructions_to_serialize = new_custom_instructions.copy()
+            for name in instructions_to_serialize:
+                instruction = custom_instructions[name]
+                new_custom_instructions = _write_custom_instruction(
+                    custom_instructions_buffer, name, instruction, custom_instructions
+                )
+
+        file_obj.write(
+            struct.pack(formats.CUSTOM_CIRCUIT_DEF_HEADER_PACK, len(custom_instructions))
+        )
+        file_obj.write(custom_instructions_buffer.getvalue())
 
-    instruction_buffer.seek(0)
-    file_obj.write(instruction_buffer.read())
+    file_obj.write(instruction_buffer.getvalue())
     instruction_buffer.close()
 
 
diff --git a/qiskit/qpy/common.py b/qiskit/qpy/common.py
index 3658f3c84..c98803424 100644
--- a/qiskit/qpy/common.py
+++ b/qiskit/qpy/common.py
@@ -26,10 +26,10 @@ from qiskit.circuit.parameter import Parameter
 from qiskit.circuit.parameterexpression import ParameterExpression
 from qiskit.circuit.parametervector import ParameterVectorElement
 from qiskit.circuit.library import PauliEvolutionGate
-from qiskit.circuit import Gate, Instruction as CircuitInstruction, QuantumCircuit
+from qiskit.circuit import Gate, Instruction as CircuitInstruction, QuantumCircuit, ControlledGate
 from qiskit.qpy import formats, exceptions
 
-QPY_VERSION = 4
+QPY_VERSION = 5
 ENCODE = "utf8"
 
 
@@ -39,6 +39,7 @@ class CircuitInstructionTypeKey(bytes, Enum):
     INSTRUCTION = b"i"
     GATE = b"g"
     PAULI_EVOL_GATE = b"p"
+    CONTROLLED_GATE = b"c"
 
     @classmethod
     def assign(cls, obj):
@@ -55,6 +56,8 @@ class CircuitInstructionTypeKey(bytes, Enum):
         """
         if isinstance(obj, PauliEvolutionGate):
             return cls.PAULI_EVOL_GATE
+        if isinstance(obj, ControlledGate):
+            return cls.CONTROLLED_GATE
         if isinstance(obj, Gate):
             return cls.GATE
         if isinstance(obj, CircuitInstruction):
diff --git a/qiskit/qpy/formats.py b/qiskit/qpy/formats.py
index ba8f900a1..6e8f869ec 100644
--- a/qiskit/qpy/formats.py
+++ b/qiskit/qpy/formats.py
@@ -85,6 +85,26 @@ CIRCUIT_INSTRUCTION = namedtuple(
 CIRCUIT_INSTRUCTION_PACK = "!HHHII?Hq"
 CIRCUIT_INSTRUCTION_SIZE = struct.calcsize(CIRCUIT_INSTRUCTION_PACK)
 
+# CIRCUIT_INSTRUCTION_V2
+CIRCUIT_INSTRUCTION_V2 = namedtuple(
+    "CIRCUIT_INSTRUCTION",
+    [
+        "name_size",
+        "label_size",
+        "num_parameters",
+        "num_qargs",
+        "num_cargs",
+        "has_condition",
+        "condition_register_size",
+        "condition_value",
+        "num_ctrl_qubits",
+        "ctrl_state",
+    ],
+)
+CIRCUIT_INSTRUCTION_V2_PACK = "!HHHII?HqII"
+CIRCUIT_INSTRUCTION_V2_SIZE = struct.calcsize(CIRCUIT_INSTRUCTION_V2_PACK)
+
+
 # CIRCUIT_INSTRUCTION_ARG
 CIRCUIT_INSTRUCTION_ARG = namedtuple("CIRCUIT_INSTRUCTION_ARG", ["type", "size"])
 CIRCUIT_INSTRUCTION_ARG_PACK = "!1cI"
@@ -108,6 +128,25 @@ CUSTOM_CIRCUIT_DEF_HEADER = namedtuple("CUSTOM_CIRCUIT_DEF_HEADER", ["size"])
 CUSTOM_CIRCUIT_DEF_HEADER_PACK = "!Q"
 CUSTOM_CIRCUIT_DEF_HEADER_SIZE = struct.calcsize(CUSTOM_CIRCUIT_DEF_HEADER_PACK)
 
+# CUSTOM_CIRCUIT_INST_DEF_V2
+CUSTOM_CIRCUIT_INST_DEF_V2 = namedtuple(
+    "CUSTOM_CIRCUIT_INST_DEF",
+    [
+        "gate_name_size",
+        "type",
+        "num_qubits",
+        "num_clbits",
+        "custom_definition",
+        "size",
+        "num_ctrl_qubits",
+        "ctrl_state",
+        "base_gate_size",
+    ],
+)
+CUSTOM_CIRCUIT_INST_DEF_V2_PACK = "!H1cII?QIIQ"
+CUSTOM_CIRCUIT_INST_DEF_V2_SIZE = struct.calcsize(CUSTOM_CIRCUIT_INST_DEF_V2_PACK)
+
+
 # CUSTOM_CIRCUIT_INST_DEF
 CUSTOM_CIRCUIT_INST_DEF = namedtuple(
     "CUSTOM_CIRCUIT_INST_DEF",
diff --git a/releasenotes/notes/fix-qpy-controlled-gates-e653cbeee067f90b.yaml b/releasenotes/notes/fix-qpy-controlled-gates-e653cbeee067f90b.yaml
new file mode 100644
index 000000000..2ccf42474
--- /dev/null
+++ b/releasenotes/notes/fix-qpy-controlled-gates-e653cbeee067f90b.yaml
@@ -0,0 +1,28 @@
+---
+upgrade:
+  - |
+    The QPY version format version emitted by :func:`.qpy.dump` has been
+    increased to version 5. This new format version is incompatible with the
+    previous versions and will result in an error when trying to load it with
+    a deserializer that isn't able to handle QPY version 5. This change was
+    necessary to fix support for representing controlled gates properly and
+    representing non-default control states.
+fixes:
+  - |
+    Fixed support for QPY serialization (:func:`.qpy.dump`) and deserialization
+    (:func:`.qpy.load`) of a :class:`~.QuantumCircuit` object containing custom
+    :class:`~.ControlledGate` objects. Previously, an exception would be raised
+    by :func:`.qpy.load` when trying to reconstruct the custom
+    :class:`~.ControlledGate`.
+    Fixed `#7999 <https://github.com/Qiskit/qiskit-terra/issues/7999>`__
+  - |
+    Fixed support for QPY serialization (:func:`.qpy.dump`) and deserialization
+    (:func:`.qpy.load`) of a :class:`~.QuantumCircuit` object containing custom
+    :class:`~.MCPhaseGate` objects. Previously, an exception would be raised
+    by :func:`.qpy.load` when trying to reconstruct the :class:`~.MCPhaseGate`.
+  - |
+    Fixed support for QPY serialization (:func:`.qpy.dump`) and deserialization
+    (:func:`.qpy.load`) of a :class:`~.QuantumCircuit` object containing
+    controlled gates with an open control state. Previously, the open control
+    state would be lost by the serialization process and the reconstructed
+    circuit.


Test patch:
diff --git a/test/python/circuit/test_circuit_load_from_qpy.py b/test/python/circuit/test_circuit_load_from_qpy.py
index 241779447..de5fdcbf2 100644
--- a/test/python/circuit/test_circuit_load_from_qpy.py
+++ b/test/python/circuit/test_circuit_load_from_qpy.py
@@ -24,7 +24,7 @@ from qiskit.circuit.classicalregister import Clbit
 from qiskit.circuit.quantumregister import Qubit
 from qiskit.circuit.random import random_circuit
 from qiskit.circuit.gate import Gate
-from qiskit.circuit.library import XGate, QFT, QAOAAnsatz, PauliEvolutionGate
+from qiskit.circuit.library import XGate, QFT, QAOAAnsatz, PauliEvolutionGate, DCXGate, MCU1Gate
 from qiskit.circuit.instruction import Instruction
 from qiskit.circuit.parameter import Parameter
 from qiskit.circuit.parametervector import ParameterVector
@@ -954,3 +954,60 @@ class TestLoadFromQPY(QiskitTestCase):
         qpy_file.seek(0)
         new_circuit = load(qpy_file)[0]
         self.assertEqual(qc.decompose().decompose(), new_circuit.decompose().decompose())
+
+    def test_controlled_gate(self):
+        """Test a custom controlled gate."""
+        qc = QuantumCircuit(3)
+        controlled_gate = DCXGate().control(1)
+        qc.append(controlled_gate, [0, 1, 2])
+        qpy_file = io.BytesIO()
+        dump(qc, qpy_file)
+        qpy_file.seek(0)
+        new_circuit = load(qpy_file)[0]
+        self.assertEqual(qc, new_circuit)
+
+    def test_nested_controlled_gate(self):
+        """Test a custom nested controlled gate."""
+        custom_gate = Gate("black_box", 1, [])
+        custom_definition = QuantumCircuit(1)
+        custom_definition.h(0)
+        custom_definition.rz(1.5, 0)
+        custom_definition.sdg(0)
+        custom_gate.definition = custom_definition
+
+        qc = QuantumCircuit(3)
+        qc.append(custom_gate, [0])
+        controlled_gate = custom_gate.control(2)
+        qc.append(controlled_gate, [0, 1, 2])
+        qpy_file = io.BytesIO()
+        dump(qc, qpy_file)
+        qpy_file.seek(0)
+        new_circ = load(qpy_file)[0]
+        self.assertEqual(qc, new_circ)
+        self.assertEqual(qc.decompose(), new_circ.decompose())
+
+    def test_open_controlled_gate(self):
+        """Test an open control is preserved across serialization."""
+        qc = QuantumCircuit(2)
+        qc.cx(0, 1, ctrl_state=0)
+        with io.BytesIO() as fd:
+            dump(qc, fd)
+            fd.seek(0)
+            new_circ = load(fd)[0]
+        self.assertEqual(qc, new_circ)
+        self.assertEqual(qc.data[0][0].ctrl_state, new_circ.data[0][0].ctrl_state)
+
+    def test_standard_control_gates(self):
+        """Test standard library controlled gates."""
+        qc = QuantumCircuit(3)
+        mcu1_gate = MCU1Gate(np.pi, 2)
+        qc.append(mcu1_gate, [0, 2, 1])
+        qc.mcp(np.pi, [0, 2], 1)
+        qc.mct([0, 2], 1)
+        qc.mcx([0, 2], 1)
+        qc.measure_all()
+        qpy_file = io.BytesIO()
+        dump(qc, qpy_file)
+        qpy_file.seek(0)
+        new_circuit = load(qpy_file)[0]
+        self.assertEqual(qc, new_circuit)
diff --git a/test/qpy_compat/test_qpy.py b/test/qpy_compat/test_qpy.py
index eb69653e2..3956da417 100755
--- a/test/qpy_compat/test_qpy.py
+++ b/test/qpy_compat/test_qpy.py
@@ -27,7 +27,8 @@ from qiskit.circuit.parametervector import ParameterVector
 from qiskit.circuit.qpy_serialization import dump, load
 from qiskit.opflow import X, Y, Z, I
 from qiskit.quantum_info.random import random_unitary
-from qiskit.circuit.library import U1Gate, U2Gate, U3Gate, QFT
+from qiskit.circuit.library import U1Gate, U2Gate, U3Gate, QFT, DCXGate
+from qiskit.circuit.gate import Gate
 
 
 def generate_full_circuit():
@@ -340,6 +341,31 @@ def generate_control_flow_circuits():
     return circuits
 
 
+def generate_controlled_gates():
+    """Test QPY serialization with custom ControlledGates."""
+    circuits = []
+    qc = QuantumCircuit(3)
+    controlled_gate = DCXGate().control(1)
+    qc.append(controlled_gate, [0, 1, 2])
+    circuits.append(qc)
+    custom_gate = Gate("black_box", 1, [])
+    custom_definition = QuantumCircuit(1)
+    custom_definition.h(0)
+    custom_definition.rz(1.5, 0)
+    custom_definition.sdg(0)
+    custom_gate.definition = custom_definition
+    nested_qc = QuantumCircuit(3)
+    qc.append(custom_gate, [0])
+    controlled_gate = custom_gate.control(2)
+    nested_qc.append(controlled_gate, [0, 1, 2])
+    nested_qc.measure_all()
+    circuits.append(nested_qc)
+    qc_open = QuantumCircuit(2)
+    qc_open.cx(0, 1, ctrl_state=0)
+    circuits.append(qc_open)
+    return circuits
+
+
 def generate_circuits(version_str=None):
     """Generate reference circuits."""
     version_parts = None
@@ -372,6 +398,8 @@ def generate_circuits(version_str=None):
         ]
     if version_parts >= (0, 19, 2):
         output_circuits["control_flow.qpy"] = generate_control_flow_circuits()
+    if version_parts >= (0, 21, 0):
+        output_circuits["controlled_gates.qpy"] = generate_controlled_gates()
 
     return output_circuits