Add option max_block_width to CollectLinearFunctions and CollectClifford passes

qiskit/dagcircuit/collect_blocks.py

@@ -139,7 +139,9 @@ def _have_uncollected_nodes(self):
         """Returns whether there are uncollected (pending) nodes"""
         return len(self._pending_nodes) > 0
 
-    def collect_matching_block(self, filter_fn: Callable) -> list[DAGOpNode | DAGDepNode]:
+    def collect_matching_block(
+        self, filter_fn: Callable, max_block_width: int | None
+    ) -> list[DAGOpNode | DAGDepNode]:
         """Iteratively collects the largest block of input nodes (that is, nodes with
         ``_in_degree`` equal to 0) that match a given filtering function.
         Examples of this include collecting blocks of swap gates,
@@ -150,6 +152,7 @@ def collect_matching_block(self, filter_fn: Callable) -> list[DAGOpNode | DAGDep
         Returns the block of collected nodes.
         """
         current_block = []
+        current_block_qargs = set()
         unprocessed_pending_nodes = self._pending_nodes
         self._pending_nodes = []
 
@@ -158,19 +161,28 @@ def collect_matching_block(self, filter_fn: Callable) -> list[DAGOpNode | DAGDep
         # - any node that match filter_fn is added to the current_block,
         #   and some of its successors may be moved to unprocessed_pending_nodes.
         while unprocessed_pending_nodes:
-            new_pending_nodes = []
-            for node in unprocessed_pending_nodes:
-                if filter_fn(node):
-                    current_block.append(node)
-
-                    # update the _in_degree of node's successors
-                    for suc in self._direct_succs(node):
-                        self._in_degree[suc] -= 1
-                        if self._in_degree[suc] == 0:
-                            new_pending_nodes.append(suc)
-                else:
-                    self._pending_nodes.append(node)
-            unprocessed_pending_nodes = new_pending_nodes
+            node = unprocessed_pending_nodes.pop()
+
+            if max_block_width is not None:
+                # for efficiency, only update new_qargs when max_block_width is specified
+                new_qargs = current_block_qargs.copy()
+                new_qargs.update(node.qargs)
+                width_within_budget = len(new_qargs) <= max_block_width
+            else:
+                new_qargs = set()
+                width_within_budget = True
+
+            if filter_fn(node) and width_within_budget:
+                current_block.append(node)
+                current_block_qargs = new_qargs
+
+                # update the _in_degree of node's successors
+                for suc in self._direct_succs(node):
+                    self._in_degree[suc] -= 1
+                    if self._in_degree[suc] == 0:
+                        unprocessed_pending_nodes.append(suc)
+            else:
+                self._pending_nodes.append(node)
 
         return current_block
 
@@ -181,6 +193,7 @@ def collect_all_matching_blocks(
         min_block_size=2,
         split_layers=False,
         collect_from_back=False,
+        max_block_width=None,
     ):
         """Collects all blocks that match a given filtering function filter_fn.
         This iteratively finds the largest block that does not match filter_fn,
@@ -193,6 +206,8 @@ def collect_all_matching_blocks(
         qubit subsets. The option ``split_layers`` allows to split collected blocks
         into layers of non-overlapping instructions. The option ``min_block_size``
         specifies the minimum number of gates in the block for the block to be collected.
+        The option ``max_block_width`` specificies the maximum number of qubits over
+        which a block can be defined.
 
         By default, blocks are collected in the direction from the inputs towards the outputs
         of the circuit. The option ``collect_from_back`` allows to change this direction,
@@ -212,8 +227,8 @@ def not_filter_fn(node):
         # Iteratively collect non-matching and matching blocks.
         matching_blocks: list[list[DAGOpNode | DAGDepNode]] = []
         while self._have_uncollected_nodes():
-            self.collect_matching_block(not_filter_fn)
-            matching_block = self.collect_matching_block(filter_fn)
+            self.collect_matching_block(not_filter_fn, max_block_width=None)
+            matching_block = self.collect_matching_block(filter_fn, max_block_width=max_block_width)
             if matching_block:
                 matching_blocks.append(matching_block)
 

qiskit/transpiler/passes/optimization/collect_and_collapse.py

@@ -95,6 +95,7 @@ def collect_using_filter_function(
     min_block_size,
     split_layers=False,
     collect_from_back=False,
+    max_block_width=None,
 ):
     """Corresponds to an important block collection strategy that greedily collects
     maximal blocks of nodes matching a given ``filter_function``.
@@ -105,6 +106,7 @@ def collect_using_filter_function(
         min_block_size=min_block_size,
         split_layers=split_layers,
         collect_from_back=collect_from_back,
+        max_block_width=max_block_width,
     )
 
 

qiskit/transpiler/passes/optimization/collect_cliffords.py

@@ -39,6 +39,7 @@ def __init__(
         split_layers=False,
         collect_from_back=False,
         matrix_based=False,
+        max_block_width=None,
     ):
         """CollectCliffords initializer.
 
@@ -55,6 +56,9 @@ def __init__(
                 from the end of the circuit.
             matrix_based (bool): specifies whether to collect unitary gates
                which are Clifford gates only for certain parameters (based on their unitary matrix).
+            max_block_width (int | None): specifies the maximum width of the block
+                (that is, the number of qubits over which the block is defined)
+                for the block to be collected.
         """
 
         collect_function = partial(
@@ -64,6 +68,7 @@ def __init__(
             min_block_size=min_block_size,
             split_layers=split_layers,
             collect_from_back=collect_from_back,
+            max_block_width=max_block_width,
         )
         collapse_function = partial(collapse_to_operation, collapse_function=_collapse_to_clifford)
 

qiskit/transpiler/passes/optimization/collect_linear_functions.py

@@ -34,6 +34,7 @@ def __init__(
         min_block_size=2,
         split_layers=False,
         collect_from_back=False,
+        max_block_width=None,
     ):
         """CollectLinearFunctions initializer.
 
@@ -48,6 +49,9 @@ def __init__(
                 over disjoint qubit subsets.
             collect_from_back (bool): specifies if blocks should be collected started
                 from the end of the circuit.
+            max_block_width (int | None): specifies the maximum width of the block
+                (that is, the number of qubits over which the block is defined)
+                for the block to be collected.
         """
 
         collect_function = partial(
@@ -57,6 +61,7 @@ def __init__(
             min_block_size=min_block_size,
             split_layers=split_layers,
             collect_from_back=collect_from_back,
+            max_block_width=max_block_width,
         )
         collapse_function = partial(
             collapse_to_operation, collapse_function=_collapse_to_linear_function

releasenotes/notes/add-max-block-width-arg-e3677a2d26575a73.yaml

@@ -0,0 +1,27 @@
+---
+features_transpiler:
+  - |
+    Added a new argument ``max_block_width`` to the class :class:`.BlockCollector`
+    and to the transpiler passes :class:`.CollectLinearFunctions` and :class:`.CollectCliffords`.
+    This argument allows to restrict the maximum number of qubits over which a block of nodes is 
+    defined. 
+
+    For example::
+
+      from qiskit.circuit import QuantumCircuit
+      from qiskit.transpiler.passes import CollectLinearFunctions
+
+      qc = QuantumCircuit(5)
+      qc.h(0)
+      qc.cx(0, 1)
+      qc.cx(1, 2)
+      qc.cx(2, 3)
+      qc.cx(3, 4)
+
+      # Collects all CX-gates into a single block 
+      qc1 = CollectLinearFunctions()(qc)
+      qc1.draw(output='mpl')
+
+      # Collects CX-gates into two blocks of width 3 
+      qc2 = CollectLinearFunctions(max_block_width=3)(qc)
+      qc2.draw(output='mpl')

test/python/dagcircuit/test_collect_blocks.py

@@ -14,6 +14,7 @@
 
 
 import unittest
+import ddt
 
 from qiskit import QuantumRegister, ClassicalRegister
 from qiskit.converters import (
@@ -28,6 +29,7 @@
 from test import QiskitTestCase  # pylint: disable=wrong-import-order
 
 
+@ddt.ddt
 class TestCollectBlocks(QiskitTestCase):
     """Tests to verify correctness of collecting, splitting, and consolidating blocks
     from DAGCircuit and DAGDependency. Additional tests appear as a part of
@@ -878,6 +880,64 @@ def test_collect_blocks_backwards_dagdependency(self):
         self.assertEqual(len(blocks[0]), 1)
         self.assertEqual(len(blocks[1]), 7)
 
+    @ddt.data(circuit_to_dag, circuit_to_dagdependency)
+    def test_max_block_width_default(self, converter):
+        """Test that not explicitly specifying ``max_block_width`` works as expected."""
+
+        # original circuit
+        circuit = QuantumCircuit(6)
+        circuit.h(0)
+        circuit.cx(0, 1)
+        circuit.cx(1, 2)
+        circuit.cx(2, 3)
+        circuit.cx(3, 4)
+        circuit.cx(4, 5)
+
+        block_collector = BlockCollector(converter(circuit))
+
+        # When max_block_width is not specified, we should obtain 1 block
+        blocks = block_collector.collect_all_matching_blocks(
+            lambda node: True,
+            min_block_size=1,
+        )
+        self.assertEqual(len(blocks), 1)
+
+    @ddt.data(
+        (circuit_to_dag, None, 1),
+        (circuit_to_dag, 2, 5),
+        (circuit_to_dag, 3, 3),
+        (circuit_to_dag, 4, 2),
+        (circuit_to_dag, 6, 1),
+        (circuit_to_dag, 10, 1),
+        (circuit_to_dagdependency, None, 1),
+        (circuit_to_dagdependency, 2, 5),
+        (circuit_to_dagdependency, 3, 3),
+        (circuit_to_dagdependency, 4, 2),
+        (circuit_to_dagdependency, 6, 1),
+        (circuit_to_dagdependency, 10, 1),
+    )
+    @ddt.unpack
+    def test_max_block_width(self, converter, max_block_width, num_expected_blocks):
+        """Test that the option ``max_block_width`` for collecting blocks works correctly."""
+
+        # original circuit
+        circuit = QuantumCircuit(6)
+        circuit.h(0)
+        circuit.cx(0, 1)
+        circuit.cx(1, 2)
+        circuit.cx(2, 3)
+        circuit.cx(3, 4)
+        circuit.cx(4, 5)
+
+        block_collector = BlockCollector(converter(circuit))
+
+        blocks = block_collector.collect_all_matching_blocks(
+            lambda node: True,
+            min_block_size=1,
+            max_block_width=max_block_width,
+        )
+        self.assertEqual(len(blocks), num_expected_blocks)
+
     def test_split_layers_dagcircuit(self):
         """Test that splitting blocks of nodes into layers works correctly."""
 

test/python/transpiler/test_clifford_passes.py

@@ -400,6 +400,29 @@ def test_collect_cliffords_default(self):
         self.assertEqual(qct.size(), 1)
         self.assertIn("clifford", qct.count_ops().keys())
 
+    def test_collect_cliffords_max_block_width(self):
+        """Make sure that collecting Clifford gates and replacing them by Clifford
+        works correctly when the option ``max_block_width`` is specified."""
+
+        # original circuit (consisting of Clifford gates only)
+        qc = QuantumCircuit(3)
+        qc.h(0)
+        qc.s(1)
+        qc.cx(0, 1)
+        qc.sdg(0)
+        qc.x(1)
+        qc.swap(2, 1)
+        qc.h(1)
+        qc.swap(1, 2)
+
+        # We should end up with two Clifford objects
+        qct = PassManager(CollectCliffords(max_block_width=2)).run(qc)
+        self.assertEqual(qct.size(), 2)
+        self.assertEqual(qct[0].name, "clifford")
+        self.assertEqual(len(qct[0].qubits), 2)
+        self.assertEqual(qct[1].name, "clifford")
+        self.assertEqual(len(qct[1].qubits), 2)
+
     def test_collect_cliffords_multiple_blocks(self):
         """Make sure that when collecting Clifford gates, non-Clifford gates
         are not collected, and the pass correctly splits disconnected Clifford

test/python/transpiler/test_linear_functions_passes.py

@@ -532,6 +532,29 @@ def test_min_block_size(self):
         self.assertNotIn("linear_function", circuit4.count_ops().keys())
         self.assertEqual(circuit4.count_ops()["cx"], 6)
 
+    def test_max_block_width(self):
+        """Test that the option max_block_width for collecting linear functions works correctly."""
+        circuit = QuantumCircuit(6)
+        circuit.cx(0, 1)
+        circuit.cx(1, 2)
+        circuit.cx(2, 3)
+        circuit.cx(3, 4)
+        circuit.cx(4, 5)
+
+        # When max_block_width = 3, we should obtain 3 linear blocks
+        circuit1 = PassManager(CollectLinearFunctions(min_block_size=1, max_block_width=3)).run(
+            circuit
+        )
+        self.assertEqual(circuit1.count_ops()["linear_function"], 3)
+        self.assertNotIn("cx", circuit1.count_ops().keys())
+
+        # When max_block_width = 4, we should obtain 2 linear blocks
+        circuit1 = PassManager(CollectLinearFunctions(min_block_size=1, max_block_width=4)).run(
+            circuit
+        )
+        self.assertEqual(circuit1.count_ops()["linear_function"], 2)
+        self.assertNotIn("cx", circuit1.count_ops().keys())
+
     @combine(do_commutative_analysis=[False, True])
     def test_collect_from_back_correctness(self, do_commutative_analysis):
         """Test that collecting from the back of the circuit works correctly."""