Light Cone Transpiler Pass (#12814)

* First commit with body of the class

* Optimized some features, added warnings and unittests

* commchecker fixed

* Fixed Pauli Indexing (proper unittests are still missing)

* Apply suggestions from code review

Co-authored-by: Julien Gacon <gaconju@gmail.com>

* d

* added get_final_ops

* Modified LightCone inputs and added tests

* Added tests for errors, modified to

* merging

* fixed commutation_checker version

* Sparse Observable tests for LightCone pass

* Fixed typehints

* Added release notes and max_num qubits for LightCone Pass

* Commented test issue #13828

* Commented decorator

* Fixed linting

* bug in reno

* Fixed non-existing cnot gate in releasenotes

* Apply suggestions from code review

Applied suggestions

Co-authored-by: Julien Gacon <gaconju@gmail.com>

* Changed registers and __init__ in unittest

* Changed warning and simplified if logic

* Fixed linting

* Use SessionCommutatorChecker + typos

---------

Co-authored-by: Samuele Piccinelli <Samuele.Piccinelli@ibm.com>
Co-authored-by: Julien Gacon <gaconju@gmail.com>

qiskit/transpiler/passes/optimization/light_cone.py

@@ -0,0 +1,135 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2017, 2019.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Cancel the redundant (self-adjoint) gates through commutation relations."""
+from __future__ import annotations
+import warnings
+from qiskit.circuit import Gate, Qubit
+from qiskit.circuit.commutation_library import SessionCommutationChecker as scc
+from qiskit.circuit.library import PauliGate, ZGate
+from qiskit.dagcircuit import DAGCircuit
+from qiskit.transpiler.basepasses import TransformationPass
+from qiskit.transpiler.passes.utils.remove_final_measurements import calc_final_ops
+
+translation_table = str.maketrans({"+": "X", "-": "X", "l": "Y", "r": "Y", "0": "Z", "1": "Z"})
+
+
+class LightCone(TransformationPass):
+    """Remove the gates that do not affect the outcome of a measurement on a circuit.
+
+    Pass for computing the light-cone of an observable or measurement. The Pass can handle
+    either an observable one would like to measure or a measurement on a set of qubits.
+    """
+
+    def __init__(self, bit_terms: str | None = None, indices: list[int] | None = None) -> None:
+        """
+        Args:
+            bit_terms: If ``None`` the light-cone will be computed for the set of measurements
+                in the circuit. If a string is specified, the light-cone will correspond to the
+                reduced circuit with the same expectation value for the observable.
+            indices: list of non-trivial indices corresponding to the observable in ``bit_terms``.
+        """
+        super().__init__()
+        valid_characters = {"X", "Y", "Z", "+", "-", "l", "r", "0", "1"}
+        self.bit_terms = None
+        if bit_terms is not None:
+            if not indices:
+                raise ValueError("`indices` must be non-empty when providing `bit_terms`.")
+            if not set(bit_terms).issubset(valid_characters):
+                raise ValueError(
+                    f"`bit_terms` should contain only characters in {valid_characters}."
+                )
+            if len(bit_terms) != len(indices):
+                raise ValueError("`bit_terms` must be the same length as `indices`.")
+            self.bit_terms = bit_terms.translate(translation_table)
+        self.indices = indices
+
+    @staticmethod
+    def _find_measurement_qubits(dag: DAGCircuit) -> set[Qubit]:
+        final_nodes = calc_final_ops(dag, {"measure"})
+        qubits_measured = set()
+        for node in final_nodes:
+            qubits_measured |= set(node.qargs)
+        return qubits_measured
+
+    def _get_initial_lightcone(
+        self, dag: DAGCircuit
+    ) -> tuple[set[Qubit], list[tuple[Gate, list[Qubit]]]]:
+        """Returns the initial light-cone.
+        If observable is `None`, the light-cone is the set of measured qubits.
+        If a `bit_terms` is provided, the qubits corresponding to the
+        non-trivial Paulis define the light-cone.
+        """
+        lightcone_qubits = self._find_measurement_qubits(dag)
+        if self.bit_terms is None:
+            lightcone_operations = [(ZGate(), [qubit_index]) for qubit_index in lightcone_qubits]
+        else:
+            # Having both measurements and an observable is not allowed
+            if len(dag.qubits) < max(self.indices) + 1:
+                raise ValueError("`indices` contains values outside the qubit range.")
+            if lightcone_qubits:
+                raise ValueError(
+                    "The circuit contains measurements and an observable has been given: "
+                    "remove the observable or the measurements."
+                )
+            lightcone_qubits = [dag.qubits[i] for i in self.indices]
+            # `lightcone_operations` is a list of tuples, each containing (operation, list_of_qubits)
+            lightcone_operations = [(PauliGate(self.bit_terms), lightcone_qubits)]
+
+        return set(lightcone_qubits), lightcone_operations
+
+    def run(self, dag: DAGCircuit) -> DAGCircuit:
+        """Run the LightCone pass on `dag`.
+
+        Args:
+            dag: The DAG to reduce.
+
+        Returns:
+            The DAG reduced to the light-cone of the observable.
+        """
+
+        # Get the initial light-cone and operations
+        lightcone_qubits, lightcone_operations = self._get_initial_lightcone(dag)
+
+        #  Initialize a new, empty DAG
+        new_dag = dag.copy_empty_like()
+
+        # Iterate over the nodes in reverse topological order
+        for node in reversed(list(dag.topological_op_nodes())):
+            # Check if the node belongs to the light-cone
+            if lightcone_qubits.intersection(node.qargs):
+                # Check commutation with all previous operations
+                commutes_bool = True
+                for op in lightcone_operations:
+                    max_num_qubits = max(len(op[1]), len(node.qargs))
+                    if max_num_qubits > 10:
+                        warnings.warn(
+                            "LightCone pass is checking commutation of"
+                            f"operators of size {max_num_qubits}."
+                            "This operation can be slow.",
+                            category=RuntimeWarning,
+                        )
+                    commute_bool = scc.commute(
+                        op[0], op[1], [], node.op, node.qargs, [], max_num_qubits=max_num_qubits
+                    )
+                    if not commute_bool:
+                        # If the current node does not commute, update the light-cone
+                        lightcone_qubits.update(node.qargs)
+                        lightcone_operations.append((node.op, node.qargs))
+                        commutes_bool = False
+                        break
+
+                # If the node is in the light-cone and commutes with previous `ops`,
+                # add it to the new DAG at the front
+                if not commutes_bool:
+                    new_dag.apply_operation_front(node.op, node.qargs, node.cargs)
+        return new_dag

qiskit/transpiler/passes/utils/remove_final_measurements.py

@@ -13,7 +13,45 @@
 """Remove final measurements and barriers at the end of a circuit."""
 
 from qiskit.transpiler.basepasses import TransformationPass
-from qiskit.dagcircuit import DAGOpNode
+from qiskit.dagcircuit import DAGCircuit, DAGOpNode
+
+
+def calc_final_ops(dag: DAGCircuit, final_op_names: set[str]) -> list[DAGOpNode]:
+    """Find the final operations of a circuit of a given type.
+    Args:
+        dag: the DAG circuit
+        final_op_names: names of the operations to find at the end of the circuit.
+
+    Returns:
+    List of nodes corresponding the the relevant operations at the end of the circuit.
+    """
+    final_ops = []
+
+    to_visit = [next(dag.predecessors(dag.output_map[qubit])) for qubit in dag.qubits]
+    barrier_encounters_remaining = {}
+
+    while to_visit:
+        node = to_visit.pop()
+        if not isinstance(node, DAGOpNode):
+            continue
+
+        if node.op.name == "barrier":
+            # Barrier is final if all children are final, so we track
+            # how many times we still need to encounter each barrier
+            # via a child node.
+            if node not in barrier_encounters_remaining:
+                barrier_encounters_remaining[node] = sum(1 for _ in dag.quantum_successors(node))
+            if barrier_encounters_remaining[node] - 1 > 0:
+                # We've encountered the barrier, but not (yet) via all children.
+                # Record the encounter, and bail!
+                barrier_encounters_remaining[node] -= 1
+                continue
+        if node.name in final_op_names:
+            # Current node is either a measure, or a barrier with all final op children.
+            final_ops.append(node)
+            to_visit.extend(dag.quantum_predecessors(node))
+
+    return final_ops
 
 
 class RemoveFinalMeasurements(TransformationPass):
@@ -32,37 +70,6 @@ class RemoveFinalMeasurements(TransformationPass):
     in a classical register that will remain.
     """
 
-    def _calc_final_ops(self, dag):
-        final_op_types = {"measure", "barrier"}
-        final_ops = []
-
-        to_visit = [next(dag.predecessors(dag.output_map[qubit])) for qubit in dag.qubits]
-        barrier_encounters_remaining = {}
-
-        while to_visit:
-            node = to_visit.pop()
-            if not isinstance(node, DAGOpNode):
-                continue
-            if node.op.name == "barrier":
-                # Barrier is final if all children are final, so we track
-                # how many times we still need to encounter each barrier
-                # via a child node.
-                if node not in barrier_encounters_remaining:
-                    barrier_encounters_remaining[node] = sum(
-                        1 for _ in dag.quantum_successors(node)
-                    )
-                if barrier_encounters_remaining[node] - 1 > 0:
-                    # We've encountered the barrier, but not (yet) via all children.
-                    # Record the encounter, and bail!
-                    barrier_encounters_remaining[node] -= 1
-                    continue
-            if node.name in final_op_types:
-                # Current node is either a measure, or a barrier with all final op children.
-                final_ops.append(node)
-                to_visit.extend(dag.quantum_predecessors(node))
-
-        return final_ops
-
     def run(self, dag):
         """Run the RemoveFinalMeasurements pass on `dag`.
 
@@ -72,7 +79,7 @@ def run(self, dag):
         Returns:
             DAGCircuit: the optimized DAG.
         """
-        final_ops = self._calc_final_ops(dag)
+        final_ops = calc_final_ops(dag, {"measure", "barrier"})
         if not final_ops:
             return dag
 

releasenotes/notes/add-light-cone-pass-6c56085734512e98.yaml

@@ -0,0 +1,43 @@
+---
+features_transpiler:
+  - |
+    Added a new transpiler pass, :class:`.LightCone` that is used
+    to get the lightcone of a circuit when measuring a subset of
+    qubits or a specific Pauli string.
+    For example if you had a circuit like:
+
+        .. plot::
+
+          from qiskit.transpiler.passes.optimization.light_cone import LightCone
+          from qiskit.transpiler.passmanager import PassManager
+          from qiskit.circuit import QuantumCircuit
+
+          qc = QuantumCircuit(3,1)
+          qc.h(range(3))
+          qc.cx(0,1)
+          qc.cx(2,1)
+          qc.h(range(3))
+          qc.measure(0,0)
+
+          qc.draw("mpl")
+
+    running the pass would eliminate the gates that do not affect the
+    outcome.
+
+        .. plot::
+           :include-source:
+
+            from qiskit.transpiler.passes.optimization.light_cone import LightCone
+            from qiskit.transpiler.passmanager import PassManager
+            from qiskit.circuit import QuantumCircuit
+
+            qc = QuantumCircuit(3,1)
+            qc.h(range(3))
+            qc.cx(0,1)
+            qc.cx(2,1)
+            qc.h(range(3))
+            qc.measure(0,0)
+
+            pm = PassManager([LightCone()])
+            new_circuit = pm.run(qc)
+            new_circuit.draw("mpl")