Make ConsolidateBlocks transpiler pass target aware (#7778)

* Make ConsolidateBlocks transpiler pass target aware

This commit updates the consolidate blocks pass target aware so that at
its instantiation a Target object can be specified to define the target
device for compilation. When specified this target will be used for
all device aware operations in the pass. When a target is specified the
consolidate blocks pass will only consolidate blocks with instructions
outside the target (both operation and qubits).

Part of #7113

* Fix typo

* Add release note

* Add tests for target instruction_supported() method

* Add Target tests for consolidate block

* Apply suggestions from code review

Co-authored-by: Jake Lishman <[email protected]>

* Check number of arguments in ideal operation case

This commit adds a check in the ideal operation case that the number of
qargs is equal to the number of qubits the operation object supports.
This way if you call instruction_supported() with a qargs parameter that
has the wrong number of arguments even an ideally implemented gate will
return false. For example, target.instruction_supported(cx, (0, 1, 2))
will return False since a CXGate object only supports 2 qubits.

* Cast qargs to tuple to prevent issues with qarg list lookups

* Deduplicate qubit index map creation

Co-authored-by: Jake Lishman <[email protected]>

qiskit/transpiler/passes/optimization/consolidate_blocks.py

@@ -37,16 +37,18 @@ class ConsolidateBlocks(TransformationPass):
         collected by a previous pass, such as `Collect2qBlocks`.
     """
 
-    def __init__(self, kak_basis_gate=None, force_consolidate=False, basis_gates=None):
+    def __init__(self, kak_basis_gate=None, force_consolidate=False, basis_gates=None, target=None):
         """ConsolidateBlocks initializer.
 
         Args:
             kak_basis_gate (Gate): Basis gate for KAK decomposition.
             force_consolidate (bool): Force block consolidation
             basis_gates (List(str)): Basis gates from which to choose a KAK gate.
+            target (Target): The target object for the compilation target backend
         """
         super().__init__()
         self.basis_gates = None
+        self.target = target
         if basis_gates is not None:
             self.basis_gates = set(basis_gates)
         self.force_consolidate = force_consolidate
@@ -73,7 +75,9 @@ def run(self, dag):
         basis_gate_name = self.decomposer.gate.name
         all_block_gates = set()
         for block in blocks:
-            if len(block) == 1 and (self.basis_gates and block[0].name not in self.basis_gates):
+            if len(block) == 1 and self._check_not_in_basis(
+                block[0].name, block[0].qargs, global_index_map
+            ):
                 all_block_gates.add(block[0])
                 dag.substitute_node(block[0], UnitaryGate(block[0].op.to_matrix()))
             else:
@@ -95,7 +99,7 @@ def run(self, dag):
                 for nd in block:
                     if nd.op.name == basis_gate_name:
                         basis_count += 1
-                    if self.basis_gates and nd.op.name not in self.basis_gates:
+                    if self._check_not_in_basis(nd.op.name, nd.qargs, global_index_map):
                         outside_basis = True
                     qc.append(nd.op, [q[block_index_map[i]] for i in nd.qargs])
                 unitary = UnitaryGate(Operator(qc))
@@ -106,15 +110,18 @@ def run(self, dag):
                     or unitary.num_qubits > 2
                     or self.decomposer.num_basis_gates(unitary) < basis_count
                     or len(block) > max_2q_depth
-                    or (self.basis_gates is not None and outside_basis)
+                    or ((self.basis_gates is not None) and outside_basis)
+                    or ((self.target is not None) and outside_basis)
                 ):
                     dag.replace_block_with_op(block, unitary, block_index_map, cycle_check=False)
         # If 1q runs are collected before consolidate those too
         runs = self.property_set["run_list"] or []
         for run in runs:
             if run[0] in all_block_gates:
                 continue
-            if len(run) == 1 and self.basis_gates and run[0].name not in self.basis_gates:
+            if len(run) == 1 and not self._check_not_in_basis(
+                run[0].name, run[0].qargs, global_index_map
+            ):
                 dag.substitute_node(run[0], UnitaryGate(run[0].op.to_matrix()))
             else:
                 qubit = run[0].qargs[0]
@@ -130,6 +137,14 @@ def run(self, dag):
                 dag.replace_block_with_op(run, unitary, {qubit: 0}, cycle_check=False)
         return dag
 
+    def _check_not_in_basis(self, gate_name, qargs, global_index_map):
+        if self.target is not None:
+            return not self.target.instruction_supported(
+                gate_name, tuple(global_index_map[qubit] for qubit in qargs)
+            )
+        else:
+            return self.basis_gates and gate_name not in self.basis_gates
+
     def _block_qargs_to_indices(self, block_qargs, global_index_map):
         """Map each qubit in block_qargs to its wire position among the block's wires.
         Args:

qiskit/transpiler/preset_passmanagers/level0.py

@@ -189,7 +189,7 @@ def _swap_condition(property_set):
             Unroll3qOrMore(),
             Collect2qBlocks(),
             Collect1qRuns(),
-            ConsolidateBlocks(basis_gates=basis_gates),
+            ConsolidateBlocks(basis_gates=basis_gates, target=target),
             UnitarySynthesis(
                 basis_gates,
                 approximation_degree=approximation_degree,

qiskit/transpiler/preset_passmanagers/level1.py

@@ -247,7 +247,7 @@ def _swap_condition(property_set):
             ),
             Unroll3qOrMore(),
             Collect2qBlocks(),
-            ConsolidateBlocks(basis_gates=basis_gates),
+            ConsolidateBlocks(basis_gates=basis_gates, target=target),
             UnitarySynthesis(
                 basis_gates,
                 approximation_degree=approximation_degree,

qiskit/transpiler/preset_passmanagers/level2.py

@@ -232,7 +232,7 @@ def _swap_condition(property_set):
             ),
             Unroll3qOrMore(),
             Collect2qBlocks(),
-            ConsolidateBlocks(basis_gates=basis_gates),
+            ConsolidateBlocks(basis_gates=basis_gates, target=target),
             UnitarySynthesis(
                 basis_gates,
                 approximation_degree=approximation_degree,

qiskit/transpiler/preset_passmanagers/level3.py

@@ -230,7 +230,7 @@ def _swap_condition(property_set):
             ),
             Unroll3qOrMore(),
             Collect2qBlocks(),
-            ConsolidateBlocks(basis_gates=basis_gates),
+            ConsolidateBlocks(basis_gates=basis_gates, target=target),
             UnitarySynthesis(
                 basis_gates,
                 approximation_degree=approximation_degree,
@@ -264,7 +264,7 @@ def _opt_control(property_set):
 
     _opt = [
         Collect2qBlocks(),
-        ConsolidateBlocks(basis_gates=basis_gates),
+        ConsolidateBlocks(basis_gates=basis_gates, target=target),
         UnitarySynthesis(
             basis_gates,
             approximation_degree=approximation_degree,

qiskit/transpiler/target.py

@@ -484,6 +484,28 @@ def operations_for_qargs(self, qargs):
             raise KeyError(f"{qargs} not in target.")
         return [self._gate_name_map[x] for x in self._qarg_gate_map[qargs]]
 
+    def instruction_supported(self, operation_name, qargs):
+        """Return whether the instruction (operation + qubits) is supported by the target
+
+        Args:
+            operation_name (str): The name of the operation for the instruction
+            qargs (tuple): The tuple of qubit indices for the instruction
+
+        Returns:
+            bool: Returns ``True`` if the instruction is supported and ``False`` if it isn't.
+
+        """
+        # Case a list if passed in by mistake
+        qargs = tuple(qargs)
+        if operation_name in self._gate_map:
+            if qargs in self._gate_map[operation_name]:
+                return True
+            if self._gate_map[operation_name] is None or None in self._gate_map[operation_name]:
+                return self._gate_name_map[operation_name].num_qubits == len(qargs) and all(
+                    x < self.num_qubits for x in qargs
+                )
+        return False
+
     @property
     def operation_names(self):
         """Get the operation names in the target."""

releasenotes/notes/consolidate-blocks-target-aware-6482e65d6ee2d18c.yaml

@@ -0,0 +1,16 @@
+---
+
+features:
+  - |
+    The :class:`~qiskit.transpiler.ConsolidateBlocks` pass has a new keyword
+    argument on its constructor, ``target``. This argument is used to specify a
+    :class:`~qiskit.transpiler.Target` object representing the compilation
+    target for the pass. If it is specified it supersedes the ``basis_gates``
+    kwarg. If a target is specified the pass will respect the gates and qubits
+    when deciding which gates to consolidate into a unitary.
+  - |
+    The :class:`~qiskit.transpiler.Target` class has a new method,
+    :meth:`~qiskit.transpiler.Target.instruction_supported` which is used
+    to query the target to see if an instruction (the combination of an
+    operation and the qubit(s) it is executed on) is supported on the backend
+    modelled by the :class:`~qiskit.transpiler.Target`.

test/python/transpiler/test_consolidate_blocks.py

@@ -18,14 +18,15 @@
 import numpy as np
 
 from qiskit.circuit import QuantumCircuit, QuantumRegister
-from qiskit.circuit.library import U2Gate
+from qiskit.circuit.library import U2Gate, SwapGate, CXGate
 from qiskit.extensions import UnitaryGate
 from qiskit.converters import circuit_to_dag
 from qiskit.transpiler.passes import ConsolidateBlocks
 from qiskit.quantum_info.operators import Operator
 from qiskit.quantum_info.operators.measures import process_fidelity
 from qiskit.test import QiskitTestCase
 from qiskit.transpiler import PassManager
+from qiskit.transpiler import Target
 from qiskit.transpiler.passes import Collect1qRuns
 from qiskit.transpiler.passes import Collect2qBlocks
 
@@ -359,6 +360,53 @@ def test_single_gate_block_outside_basis(self):
         expected.unitary(np.array([[1, 0, 0, 0], [0, 0, 1, 0], [0, 1, 0, 0], [0, 0, 0, 1]]), [0, 1])
         self.assertEqual(expected, pass_manager.run(qc))
 
+    def test_single_gate_block_outside_basis_with_target(self):
+        """Test a gate outside basis defined in target gets converted."""
+        qc = QuantumCircuit(2)
+        target = Target(num_qubits=2)
+        # Add ideal basis gates to all qubits
+        target.add_instruction(CXGate())
+        qc.swap(0, 1)
+        consolidate_block_pass = ConsolidateBlocks(target=target)
+        pass_manager = PassManager()
+        pass_manager.append(Collect2qBlocks())
+        pass_manager.append(consolidate_block_pass)
+        expected = QuantumCircuit(2)
+        expected.unitary(np.array([[1, 0, 0, 0], [0, 0, 1, 0], [0, 1, 0, 0], [0, 0, 0, 1]]), [0, 1])
+        self.assertEqual(expected, pass_manager.run(qc))
+
+    def test_single_gate_block_outside_local_basis_with_target(self):
+        """Test that a gate in basis but outside valid qubits is treated as outside basis with target."""
+        qc = QuantumCircuit(2)
+        target = Target(num_qubits=2)
+        # Add ideal cx to (1, 0) only
+        target.add_instruction(CXGate(), {(1, 0): None})
+        qc.cx(0, 1)
+        consolidate_block_pass = ConsolidateBlocks(target=target)
+        pass_manager = PassManager()
+        pass_manager.append(Collect2qBlocks())
+        pass_manager.append(consolidate_block_pass)
+        expected = QuantumCircuit(2)
+        expected.unitary(np.array([[1, 0, 0, 0], [0, 0, 0, 1], [0, 0, 1, 0], [0, 1, 0, 0]]), [0, 1])
+        self.assertEqual(expected, pass_manager.run(qc))
+
+    def test_single_gate_block_outside_target_with_matching_basis_gates(self):
+        """Ensure the target is the source of truth with basis_gates also set."""
+        qc = QuantumCircuit(2)
+        target = Target(num_qubits=2)
+        # Add ideal cx to (1, 0) only
+        target.add_instruction(SwapGate())
+        qc.swap(0, 1)
+        consolidate_block_pass = ConsolidateBlocks(
+            basis_gates=["id", "cx", "rz", "sx", "x"], target=target
+        )
+        pass_manager = PassManager()
+        pass_manager.append(Collect2qBlocks())
+        pass_manager.append(consolidate_block_pass)
+        expected = QuantumCircuit(2)
+        expected.swap(0, 1)
+        self.assertEqual(expected, pass_manager.run(qc))
+
 
 if __name__ == "__main__":
     unittest.main()

test/python/transpiler/test_target.py

@@ -904,6 +904,15 @@ def test_get_non_global_operation_name_ideal_backend_strict_direction(self):
             self.fake_backend_target.get_non_global_operation_names(True), ["cx", "ecr"]
         )
 
+    def test_instruction_supported(self):
+        self.assertTrue(self.aqt_target.instruction_supported("r", (0,)))
+        self.assertFalse(self.aqt_target.instruction_supported("cx", (0, 1)))
+        self.assertTrue(self.ideal_sim_target.instruction_supported("cx", (0, 1)))
+        self.assertFalse(self.ideal_sim_target.instruction_supported("cx", (0, 524)))
+        self.assertTrue(self.fake_backend_target.instruction_supported("cx", (0, 1)))
+        self.assertFalse(self.fake_backend_target.instruction_supported("cx", (1, 0)))
+        self.assertFalse(self.ideal_sim_target.instruction_supported("cx", (0, 1, 2)))
+
 
 class TestPulseTarget(QiskitTestCase):
     def setUp(self):