Repository: Qiskit/qiskit (python)

Issue:
`EquivalenceLibrary.set_entry` does not clear old entries in graph
### Environment

- **Qiskit version**: 1.0.1
- **Python version**: 3.10
- **Operating system**: macOS


### What is happening?

When using `EquivalenceLibrary.set_entry` to replace a set of entries with another, the `get_entry` method will update with the new entries, but the `graph` attribute will not have suitably changed the edges, meaning that the `BasisTranslator` will use the old rules.

### How can we reproduce the issue?

```python
import math
from qiskit.transpiler.passes import BasisTranslator
from qiskit.circuit.equivalence import EquivalenceLibrary
from qiskit.circuit import library, QuantumCircuit

equiv = EquivalenceLibrary()

cx_to_cz = QuantumCircuit(2)
cx_to_cz.ry(-math.pi / 4, 1)
cx_to_cz.cz(0, 1)
cx_to_cz.ry(math.pi / 2, 1)
equiv.add_equivalence(library.CXGate(), cx_to_cz)

cx_to_cz_worse = QuantumCircuit(2, global_phase=math.pi)
cx_to_cz_worse.ry(math.pi / 2, 1)
cx_to_cz_worse.rx(math.pi, 1)
cx_to_cz_worse.cz(0, 1)
cx_to_cz_worse.ry(math.pi / 2, 1)
cx_to_cz_worse.rx(math.pi, 1)
equiv.set_entry(library.CXGate(), [cx_to_cz_worse])

qc = QuantumCircuit(2)
qc.cx(0, 1)
BasisTranslator(equiv, ["cz", "rx", "ry", "rz"])(qc).draw()
```
```

q_0: ─────────────■────────────
     ┌──────────┐ │ ┌─────────┐
q_1: ┤ Ry(-π/4) ├─■─┤ Ry(π/2) ├
     └──────────┘   └─────────┘
```

### What should happen?

Since we replaced the "good" rule with a "bad" rule, the basis translator should only be able to find the bad rule and the output should be:
```
global phase: π

q_0: ─────────────────────■─────────────────────
     ┌─────────┐┌───────┐ │ ┌─────────┐┌───────┐
q_1: ┤ Ry(π/2) ├┤ Rx(π) ├─■─┤ Ry(π/2) ├┤ Rx(π) ├
     └─────────┘└───────┘   └─────────┘└───────┘
```

### Any suggestions?

_No response_

Pull request description:
Remove old graph structure during `EquivalenceLibrary.set_entry`
### Summary

The previous implementation of `EquivalenceLibrary.set_entry` changed the equivalence rules attached to particular nodes within the graph structure, but didn't correctly update the edges, so the graph was no longer representing the correct data, and queries within `BasisTranslator` would still use the old equivalence sets.

This correctly clears out the old rules and adds the new structure when `set_entry` is used.  The inner private variable `_rule_count` is replaced with `_rule_id`, since it would actually be misleading for its use to decrement it; this could cause clashes, and what's actually intended and used is that it's an id for rules.

<!--
⚠️ 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.
-->

### Details and comments

Fix #11958


Patch:
diff --git a/qiskit/circuit/commutation_checker.py b/qiskit/circuit/commutation_checker.py
index 7e025554d..a3283de37 100644
--- a/qiskit/circuit/commutation_checker.py
+++ b/qiskit/circuit/commutation_checker.py
@@ -16,11 +16,15 @@ from functools import lru_cache
 from typing import List, Union
 import numpy as np
 
+from qiskit import QiskitError
 from qiskit.circuit import Qubit
 from qiskit.circuit.operation import Operation
-from qiskit.circuit.controlflow import ControlFlowOp
+from qiskit.circuit.controlflow import CONTROL_FLOW_OP_NAMES
 from qiskit.quantum_info.operators import Operator
 
+_skipped_op_names = {"measure", "reset", "delay", "initialize"}
+_no_cache_op_names = {"annotated"}
+
 
 @lru_cache(maxsize=None)
 def _identity_op(num_qubits):
@@ -94,8 +98,11 @@ class CommutationChecker:
         )
         first_op, first_qargs, _ = first_op_tuple
         second_op, second_qargs, _ = second_op_tuple
-        first_params = first_op.params
-        second_params = second_op.params
+
+        skip_cache = first_op.name in _no_cache_op_names or second_op.name in _no_cache_op_names
+
+        if skip_cache:
+            return _commute_matmul(first_op, first_qargs, second_op, second_qargs)
 
         commutation_lookup = self.check_commutation_entries(
             first_op, first_qargs, second_op, second_qargs
@@ -113,6 +120,8 @@ class CommutationChecker:
         if self._current_cache_entries >= self._cache_max_entries:
             self.clear_cached_commutations()
 
+        first_params = getattr(first_op, "params", [])
+        second_params = getattr(second_op, "params", [])
         if len(first_params) > 0 or len(second_params) > 0:
             self._cached_commutations.setdefault((first_op.name, second_op.name), {}).setdefault(
                 _get_relative_placement(first_qargs, second_qargs), {}
@@ -184,7 +193,11 @@ class CommutationChecker:
 
 
 def _hashable_parameters(params):
-    """Convert the parameters of a gate into a hashable format for lookup in a dictionary."""
+    """Convert the parameters of a gate into a hashable format for lookup in a dictionary.
+
+    This aims to be fast in common cases, and is not intended to work outside of the lifetime of a
+    single commutation pass; it does not handle mutable state correctly if the state is actually
+    changed."""
     try:
         hash(params)
         return params
@@ -201,7 +214,53 @@ def _hashable_parameters(params):
     return ("fallback", str(params))
 
 
-_skipped_op_names = {"measure", "reset", "delay"}
+def is_commutation_supported(op):
+    """
+    Filter operations whose commutation is not supported due to bugs in transpiler passes invoking
+    commutation analysis.
+    Args:
+        op (Operation): operation to be checked for commutation relation
+    Return:
+        True if determining the commutation of op is currently supported
+    """
+    # Bug in CommutativeCancellation, e.g. see gh-8553
+    if getattr(op, "condition", False):
+        return False
+
+    # Commutation of ControlFlow gates also not supported yet. This may be pending a control flow graph.
+    if op.name in CONTROL_FLOW_OP_NAMES:
+        return False
+
+    return True
+
+
+def is_commutation_skipped(op, qargs, max_num_qubits):
+    """
+    Filter operations whose commutation will not be determined.
+    Args:
+        op (Operation): operation to be checked for commutation relation
+        qargs (List): operation qubits
+        max_num_qubits (int): the maximum number of qubits to consider, the check may be skipped if
+                the number of qubits for either operation exceeds this amount.
+    Return:
+        True if determining the commutation of op is currently not supported
+    """
+    if (
+        len(qargs) > max_num_qubits
+        or getattr(op, "_directive", False)
+        or op.name in _skipped_op_names
+    ):
+        return True
+
+    if getattr(op, "is_parameterized", False) and op.is_parameterized():
+        return True
+
+    # we can proceed if op has defined: to_operator, to_matrix and __array__, or if its definition can be
+    # recursively resolved by operations that have a matrix. We check this by constructing an Operator.
+    if (hasattr(op, "to_matrix") and hasattr(op, "__array__")) or hasattr(op, "to_operator"):
+        return False
+
+    return False
 
 
 def _commutation_precheck(
@@ -213,43 +272,14 @@ def _commutation_precheck(
     cargs2: List,
     max_num_qubits,
 ):
-    # pylint: disable=too-many-return-statements
-
-    # We don't support commutation of conditional gates for now due to bugs in
-    # CommutativeCancellation.  See gh-8553.
-    if getattr(op1, "condition", None) is not None or getattr(op2, "condition", None) is not None:
+    if not is_commutation_supported(op1) or not is_commutation_supported(op2):
         return False
 
-    # Commutation of ControlFlow gates also not supported yet. This may be
-    # pending a control flow graph.
-    if isinstance(op1, ControlFlowOp) or isinstance(op2, ControlFlowOp):
-        return False
-
-    # These lines are adapted from dag_dependency and say that two gates over
-    # different quantum and classical bits necessarily commute. This is more
-    # permissive that the check from commutation_analysis, as for example it
-    # allows to commute X(1) and Measure(0, 0).
-    # Presumably this check was not present in commutation_analysis as
-    # it was only called on pairs of connected nodes from DagCircuit.
-    intersection_q = set(qargs1).intersection(set(qargs2))
-    intersection_c = set(cargs1).intersection(set(cargs2))
-    if not (intersection_q or intersection_c):
+    if set(qargs1).isdisjoint(qargs2) and set(cargs1).isdisjoint(cargs2):
         return True
 
-    # Skip the check if the number of qubits for either operation is too large
-    if len(qargs1) > max_num_qubits or len(qargs2) > max_num_qubits:
-        return False
-
-    # These lines are adapted from commutation_analysis, which is more restrictive than the
-    # check from dag_dependency when considering nodes with "_directive".  It would be nice to
-    # think which optimizations from dag_dependency can indeed be used.
-    if op1.name in _skipped_op_names or op2.name in _skipped_op_names:
-        return False
-
-    if getattr(op1, "_directive", False) or getattr(op2, "_directive", False):
-        return False
-    if (getattr(op1, "is_parameterized", False) and op1.is_parameterized()) or (
-        getattr(op2, "is_parameterized", False) and op2.is_parameterized()
+    if is_commutation_skipped(op1, qargs1, max_num_qubits) or is_commutation_skipped(
+        op2, qargs2, max_num_qubits
     ):
         return False
 
@@ -264,13 +294,11 @@ def _get_relative_placement(first_qargs: List[Qubit], second_qargs: List[Qubit])
         second_qargs (DAGOpNode): second gate
 
     Return:
-        A tuple that describes the relative qubit placement. The relative placement is defined by the
-        gate qubit arrangements as q2^{-1}[q1[i]] where q1[i] is the ith qubit of the first gate and
-        q2^{-1}[q] returns the qubit index of qubit q in the second gate (possibly 'None'). E.g.
+        A tuple that describes the relative qubit placement: E.g.
         _get_relative_placement(CX(0, 1), CX(1, 2)) would return (None, 0) as there is no overlap on
         the first qubit of the first gate but there is an overlap on the second qubit of the first gate,
-        i.e. qubit 0 of the second gate. Likewise, _get_relative_placement(CX(1, 2), CX(0, 1)) would
-        return (1, None)
+        i.e. qubit 0 of the second gate. Likewise,
+        _get_relative_placement(CX(1, 2), CX(0, 1)) would return (1, None)
     """
     qubits_g2 = {q_g1: i_g1 for i_g1, q_g1 in enumerate(second_qargs)}
     return tuple(qubits_g2.get(q_g0, None) for q_g0 in first_qargs)
@@ -355,8 +383,10 @@ def _query_commutation(
         # if we have another dict in commutation_after_placement, commutation depends on params
         if isinstance(commutation_after_placement, dict):
             # Param commutation entry exists and must be a dict
+            first_params = getattr(first_op, "params", [])
+            second_params = getattr(second_op, "params", [])
             return commutation_after_placement.get(
-                (_hashable_parameters(first_op.params), _hashable_parameters(second_op.params)),
+                (_hashable_parameters(first_params), _hashable_parameters(second_params)),
                 None,
             )
         else:
@@ -379,12 +409,17 @@ def _commute_matmul(
     first_qarg = tuple(qarg[q] for q in first_qargs)
     second_qarg = tuple(qarg[q] for q in second_qargs)
 
-    operator_1 = Operator(
-        first_ops, input_dims=(2,) * len(first_qarg), output_dims=(2,) * len(first_qarg)
-    )
-    operator_2 = Operator(
-        second_op, input_dims=(2,) * len(second_qarg), output_dims=(2,) * len(second_qarg)
-    )
+    # try to generate an Operator out of op, if this succeeds we can determine commutativity, otherwise
+    # return false
+    try:
+        operator_1 = Operator(
+            first_ops, input_dims=(2,) * len(first_qarg), output_dims=(2,) * len(first_qarg)
+        )
+        operator_2 = Operator(
+            second_op, input_dims=(2,) * len(second_qarg), output_dims=(2,) * len(second_qarg)
+        )
+    except QiskitError:
+        return False
 
     if first_qarg == second_qarg:
         # Use full composition if possible to get the fastest matmul paths.
diff --git a/qiskit/circuit/equivalence.py b/qiskit/circuit/equivalence.py
index 08525905f..45921c3f2 100644
--- a/qiskit/circuit/equivalence.py
+++ b/qiskit/circuit/equivalence.py
@@ -45,11 +45,12 @@ class EquivalenceLibrary:
         if base is None:
             self._graph = rx.PyDiGraph()
             self._key_to_node_index = {}
-            self._rule_count = 0
+            # Some unique identifier for rules.
+            self._rule_id = 0
         else:
             self._graph = base._graph.copy()
             self._key_to_node_index = copy.deepcopy(base._key_to_node_index)
-            self._rule_count = base._rule_count
+            self._rule_id = base._rule_id
 
     @property
     def graph(self) -> rx.PyDiGraph:
@@ -104,12 +105,12 @@ class EquivalenceLibrary:
             (
                 self._set_default_node(source),
                 target,
-                EdgeData(index=self._rule_count, num_gates=len(sources), rule=equiv, source=source),
+                EdgeData(index=self._rule_id, num_gates=len(sources), rule=equiv, source=source),
             )
             for source in sources
         ]
         self._graph.add_edges_from(edges)
-        self._rule_count += 1
+        self._rule_id += 1
 
     def has_entry(self, gate):
         """Check if a library contains any decompositions for gate.
@@ -142,10 +143,15 @@ class EquivalenceLibrary:
             _raise_if_shape_mismatch(gate, equiv)
             _raise_if_param_mismatch(gate.params, equiv.parameters)
 
-        key = Key(name=gate.name, num_qubits=gate.num_qubits)
-        equivs = [Equivalence(params=gate.params.copy(), circuit=equiv.copy()) for equiv in entry]
-
-        self._graph[self._set_default_node(key)] = NodeData(key=key, equivs=equivs)
+        node_index = self._set_default_node(Key(name=gate.name, num_qubits=gate.num_qubits))
+        # Remove previous equivalences of this node, leaving in place any later equivalences that
+        # were added that use `gate`.
+        self._graph[node_index].equivs.clear()
+        for parent, child, _ in self._graph.in_edges(node_index):
+            # `child` should always be ourselves, but there might be parallel edges.
+            self._graph.remove_edge(parent, child)
+        for equivalence in entry:
+            self.add_equivalence(gate, equivalence)
 
     def get_entry(self, gate):
         """Gets the set of QuantumCircuits circuits from the library which
diff --git a/releasenotes/notes/abstract-commutation-analysis-3518129e91a33599.yaml b/releasenotes/notes/abstract-commutation-analysis-3518129e91a33599.yaml
new file mode 100644
index 000000000..eee58d430
--- /dev/null
+++ b/releasenotes/notes/abstract-commutation-analysis-3518129e91a33599.yaml
@@ -0,0 +1,12 @@
+---
+features:
+  - |
+    Extended the commutation analysis performed by :class:`.CommutationChecker` to only operate on
+    hardware circuits to also work with abstract circuits, i.e. each operation in
+    the input quantum circuit is now checked for its matrix representation before proceeding to the
+    analysis. In addition, the operation is now checked for its ability to be cached in the session
+    commutation library. For example, this now enables computing whether :class:`.AnnotatedOperation`
+    commute. This enables transpiler passes that rely on :class:`.CommutationChecker` internally,
+    such as :class:`.CommutativeCancellation`, during earlier stages of a default transpilation pipeline
+    (prior to basis translation).
+
diff --git a/releasenotes/notes/fix-equivalence-setentry-5a30b0790666fcf2.yaml b/releasenotes/notes/fix-equivalence-setentry-5a30b0790666fcf2.yaml
new file mode 100644
index 000000000..9df1eb36c
--- /dev/null
+++ b/releasenotes/notes/fix-equivalence-setentry-5a30b0790666fcf2.yaml
@@ -0,0 +1,7 @@
+---
+fixes:
+  - |
+    Calling :meth:`.EquivalenceLibrary.set_entry` will now correctly update the internal graph
+    object of the library.  Previously, the metadata would be updated, but the graph structure would
+    be unaltered, meaning that users like :class:`.BasisTranslator` would still use the old rules.
+    Fixed `#11958 <https://github.com/Qiskit/qiskit/issues/11958>`__.


Test patch:
diff --git a/test/python/circuit/test_commutation_checker.py b/test/python/circuit/test_commutation_checker.py
index 4a3e6e3b0..d9b70b56e 100644
--- a/test/python/circuit/test_commutation_checker.py
+++ b/test/python/circuit/test_commutation_checker.py
@@ -17,7 +17,14 @@ import unittest
 import numpy as np
 
 from qiskit import ClassicalRegister
-from qiskit.circuit import QuantumRegister, Parameter, Qubit
+from qiskit.circuit import (
+    QuantumRegister,
+    Parameter,
+    Qubit,
+    AnnotatedOperation,
+    InverseModifier,
+    ControlModifier,
+)
 from qiskit.circuit.commutation_library import SessionCommutationChecker as scc
 
 from qiskit.circuit.library import (
@@ -31,6 +38,7 @@ from qiskit.circuit.library import (
     Barrier,
     Reset,
     LinearFunction,
+    SGate,
 )
 from test import QiskitTestCase  # pylint: disable=wrong-import-order
 
@@ -384,6 +392,30 @@ class TestCommutationChecker(QiskitTestCase):
         res = scc.commute(lf3, [0, 1, 2], [], lf4, [0, 1, 2], [])
         self.assertTrue(res)
 
+    def test_equal_annotated_operations_commute(self):
+        """Check commutativity involving the same annotated operation."""
+        op1 = AnnotatedOperation(SGate(), [InverseModifier(), ControlModifier(1)])
+        op2 = AnnotatedOperation(SGate(), [InverseModifier(), ControlModifier(1)])
+        # the same, so true
+        self.assertTrue(scc.commute(op1, [0, 1], [], op2, [0, 1], []))
+
+    def test_annotated_operations_commute_with_unannotated(self):
+        """Check commutativity involving annotated operations and unannotated operations."""
+        op1 = AnnotatedOperation(SGate(), [InverseModifier(), ControlModifier(1)])
+        op2 = AnnotatedOperation(ZGate(), [InverseModifier()])
+        op3 = ZGate()
+        # all true
+        self.assertTrue(scc.commute(op1, [0, 1], [], op2, [1], []))
+        self.assertTrue(scc.commute(op1, [0, 1], [], op3, [1], []))
+        self.assertTrue(scc.commute(op2, [1], [], op3, [1], []))
+
+    def test_annotated_operations_no_commute(self):
+        """Check non-commutativity involving annotated operations."""
+        op1 = AnnotatedOperation(XGate(), [InverseModifier(), ControlModifier(1)])
+        op2 = AnnotatedOperation(XGate(), [InverseModifier()])
+        # false
+        self.assertFalse(scc.commute(op1, [0, 1], [], op2, [0], []))
+
     def test_c7x_gate(self):
         """Test wide gate works correctly."""
         qargs = [Qubit() for _ in [None] * 8]
diff --git a/test/python/circuit/test_equivalence.py b/test/python/circuit/test_equivalence.py
index bc2f3c957..286941313 100644
--- a/test/python/circuit/test_equivalence.py
+++ b/test/python/circuit/test_equivalence.py
@@ -103,24 +103,87 @@ class TestEquivalenceLibraryWithoutBase(QiskitTestCase):
         self.assertEqual(entry[1], second_equiv)
 
     def test_set_entry(self):
-        """Verify setting an entry overrides any previously added."""
+        """Verify setting an entry overrides any previously added, without affecting entries that
+        depended on the set entry."""
         eq_lib = EquivalenceLibrary()
 
-        gate = OneQubitZeroParamGate()
-        first_equiv = QuantumCircuit(1)
-        first_equiv.h(0)
+        gates = {key: Gate(key, 1, []) for key in "abcd"}
+        target = Gate("target", 1, [])
+
+        old = QuantumCircuit(1)
+        old.append(gates["a"], [0])
+        old.append(gates["b"], [0])
+        eq_lib.add_equivalence(target, old)
+
+        outbound = QuantumCircuit(1)
+        outbound.append(target, [0])
+        eq_lib.add_equivalence(gates["c"], outbound)
+
+        self.assertEqual(eq_lib.get_entry(target), [old])
+        self.assertEqual(eq_lib.get_entry(gates["c"]), [outbound])
+        # Assert the underlying graph structure is correct as well.
+        gate_indices = {eq_lib.graph[node].key.name: node for node in eq_lib.graph.node_indices()}
+        self.assertTrue(eq_lib.graph.has_edge(gate_indices["a"], gate_indices["target"]))
+        self.assertTrue(eq_lib.graph.has_edge(gate_indices["b"], gate_indices["target"]))
+        self.assertTrue(eq_lib.graph.has_edge(gate_indices["target"], gate_indices["c"]))
+
+        new = QuantumCircuit(1)
+        new.append(gates["d"], [0])
+        eq_lib.set_entry(target, [new])
+
+        self.assertEqual(eq_lib.get_entry(target), [new])
+        self.assertEqual(eq_lib.get_entry(gates["c"]), [outbound])
+        # Assert the underlying graph structure is correct as well.
+        gate_indices = {eq_lib.graph[node].key.name: node for node in eq_lib.graph.node_indices()}
+        self.assertFalse(eq_lib.graph.has_edge(gate_indices["a"], gate_indices["target"]))
+        self.assertFalse(eq_lib.graph.has_edge(gate_indices["b"], gate_indices["target"]))
+        self.assertTrue(eq_lib.graph.has_edge(gate_indices["d"], gate_indices["target"]))
+        self.assertTrue(eq_lib.graph.has_edge(gate_indices["target"], gate_indices["c"]))
+
+    def test_set_entry_parallel_edges(self):
+        """Test that `set_entry` works correctly in the case of parallel wires."""
+        eq_lib = EquivalenceLibrary()
+        gates = {key: Gate(key, 1, []) for key in "abcd"}
+        target = Gate("target", 1, [])
 
-        eq_lib.add_equivalence(gate, first_equiv)
+        old_1 = QuantumCircuit(1, name="a")
+        old_1.append(gates["a"], [0])
+        old_1.append(gates["b"], [0])
+        eq_lib.add_equivalence(target, old_1)
 
-        second_equiv = QuantumCircuit(1)
-        second_equiv.append(U2Gate(0, np.pi), [0])
+        old_2 = QuantumCircuit(1, name="b")
+        old_2.append(gates["b"], [0])
+        old_2.append(gates["a"], [0])
+        eq_lib.add_equivalence(target, old_2)
 
-        eq_lib.set_entry(gate, [second_equiv])
+        # This extra rule is so that 'a' still has edges, so we can do an exact isomorphism test.
+        # There's not particular requirement for `set_entry` to remove orphan nodes, so we'll just
+        # craft a test that doesn't care either way.
+        a_to_b = QuantumCircuit(1)
+        a_to_b.append(gates["b"], [0])
+        eq_lib.add_equivalence(gates["a"], a_to_b)
 
-        entry = eq_lib.get_entry(gate)
+        self.assertEqual(sorted(eq_lib.get_entry(target), key=lambda qc: qc.name), [old_1, old_2])
 
-        self.assertEqual(len(entry), 1)
-        self.assertEqual(entry[0], second_equiv)
+        new = QuantumCircuit(1, name="c")
+        # No more use of 'a', but re-use 'b' and introduce 'c'.
+        new.append(gates["b"], [0])
+        new.append(gates["c"], [0])
+        eq_lib.set_entry(target, [new])
+
+        self.assertEqual(eq_lib.get_entry(target), [new])
+
+        expected = EquivalenceLibrary()
+        expected.add_equivalence(gates["a"], a_to_b)
+        expected.add_equivalence(target, new)
+
+        def node_fn(left, right):
+            return left == right
+
+        def edge_fn(left, right):
+            return left.rule == right.rule
+
+        self.assertTrue(rx.is_isomorphic(eq_lib.graph, expected.graph, node_fn, edge_fn))
 
     def test_raise_if_gate_entry_shape_mismatch(self):
         """Verify we raise if adding a circuit and gate with different shapes."""