Support standalone Var throughout transpiler (#12322)

* Support standalone `Var` throughout transpiler

This adds the missing pieces to fully support standalone `Var` nodes
through every part of the transpiler (that I could detect).  It's quite
possible there's some problem in a more esoteric non-preset pass
somewhere, but I couldn't spot them.

For the most part there were very few changes needed to the actual
passes, and only one place in `QuantumCircuit` that had previously been
missed.  Most of the commit is updating passes to correctly pass
`inline_captures=True` when appropriate for dealing with
`DAGCircuit.compose`, and making sure that any place that needed to
build a raw `DAGCircuit` for a rebuild _without_ using
`DAGCircuit.copy_empty_like` made sure to correctly add in the
variables.

This commit adds specific tests for every pass that I touched, plus the
general integration tests that we have for the transpiler to make sure
that OQ3 and QPY serialisation work afterwards.

* Clarify comment

qiskit/circuit/quantumcircuit.py

@@ -621,9 +621,7 @@ def reverse_ops(self) -> "QuantumCircuit":
                 q_1: ┤ RX(1.57) ├─────
                      └──────────┘
         """
-        reverse_circ = QuantumCircuit(
-            self.qubits, self.clbits, *self.qregs, *self.cregs, name=self.name + "_reverse"
-        )
+        reverse_circ = self.copy_empty_like(self.name + "_reverse")
 
         for instruction in reversed(self.data):
             reverse_circ._append(instruction.replace(operation=instruction.operation.reverse_ops()))

qiskit/circuit/store.py

@@ -59,6 +59,9 @@ class Store(Instruction):
     :class:`~.circuit.Measure` is a primitive for quantum measurement), and is not safe for
     subclassing."""
 
+    # This is a compiler/backend intrinsic operation, separate to any quantum processing.
+    _directive = True
+
     def __init__(self, lvalue: expr.Expr, rvalue: expr.Expr):
         """
         Args:

qiskit/transpiler/passes/basis/basis_translator.py

@@ -148,12 +148,12 @@ def run(self, dag):
 
         # Names of instructions assumed to supported by any backend.
         if self._target is None:
-            basic_instrs = ["measure", "reset", "barrier", "snapshot", "delay"]
+            basic_instrs = ["measure", "reset", "barrier", "snapshot", "delay", "store"]
             target_basis = set(self._target_basis)
             source_basis = set(self._extract_basis(dag))
             qargs_local_source_basis = {}
         else:
-            basic_instrs = ["barrier", "snapshot"]
+            basic_instrs = ["barrier", "snapshot", "store"]
             target_basis = self._target.keys() - set(self._non_global_operations)
             source_basis, qargs_local_source_basis = self._extract_basis_target(dag, qarg_indices)
 

qiskit/transpiler/passes/basis/unroll_custom_definitions.py

@@ -61,7 +61,7 @@ def run(self, dag):
             return dag
 
         if self._target is None:
-            basic_insts = {"measure", "reset", "barrier", "snapshot", "delay"}
+            basic_insts = {"measure", "reset", "barrier", "snapshot", "delay", "store"}
             device_insts = basic_insts | set(self._basis_gates)
 
         for node in dag.op_nodes():

qiskit/transpiler/passes/layout/apply_layout.py

@@ -61,6 +61,12 @@ def run(self, dag):
 
         new_dag = DAGCircuit()
         new_dag.add_qreg(q)
+        for var in dag.iter_input_vars():
+            new_dag.add_input_var(var)
+        for var in dag.iter_captured_vars():
+            new_dag.add_captured_var(var)
+        for var in dag.iter_declared_vars():
+            new_dag.add_declared_var(var)
         new_dag.metadata = dag.metadata
         new_dag.add_clbits(dag.clbits)
         for creg in dag.cregs.values():

qiskit/transpiler/passes/layout/sabre_layout.py

@@ -308,6 +308,12 @@ def run(self, dag):
         mapped_dag.add_clbits(dag.clbits)
         for creg in dag.cregs.values():
             mapped_dag.add_creg(creg)
+        for var in dag.iter_input_vars():
+            mapped_dag.add_input_var(var)
+        for var in dag.iter_captured_vars():
+            mapped_dag.add_captured_var(var)
+        for var in dag.iter_declared_vars():
+            mapped_dag.add_declared_var(var)
         mapped_dag._global_phase = dag._global_phase
         self.property_set["original_qubit_indices"] = {
             bit: index for index, bit in enumerate(dag.qubits)

qiskit/transpiler/passes/optimization/optimize_annotated.py

@@ -77,7 +77,7 @@ def __init__(
         self._top_level_only = not recurse or (self._basis_gates is None and self._target is None)
 
         if not self._top_level_only and self._target is None:
-            basic_insts = {"measure", "reset", "barrier", "snapshot", "delay"}
+            basic_insts = {"measure", "reset", "barrier", "snapshot", "delay", "store"}
             self._device_insts = basic_insts | set(self._basis_gates)
 
     def run(self, dag: DAGCircuit):

qiskit/transpiler/passes/routing/stochastic_swap.py

@@ -33,7 +33,6 @@
     ForLoopOp,
     SwitchCaseOp,
     ControlFlowOp,
-    Instruction,
     CASE_DEFAULT,
 )
 from qiskit._accelerate import stochastic_swap as stochastic_swap_rs
@@ -266,11 +265,15 @@ def _layer_update(self, dag, layer, best_layout, best_depth, best_circuit):
         # Output any swaps
         if best_depth > 0:
             logger.debug("layer_update: there are swaps in this layer, depth %d", best_depth)
-            dag.compose(best_circuit, qubits={bit: bit for bit in best_circuit.qubits})
+            dag.compose(
+                best_circuit, qubits={bit: bit for bit in best_circuit.qubits}, inline_captures=True
+            )
         else:
             logger.debug("layer_update: there are no swaps in this layer")
         # Output this layer
-        dag.compose(layer["graph"], qubits=best_layout.reorder_bits(dag.qubits))
+        dag.compose(
+            layer["graph"], qubits=best_layout.reorder_bits(dag.qubits), inline_captures=True
+        )
 
     def _mapper(self, circuit_graph, coupling_graph, trials=20):
         """Map a DAGCircuit onto a CouplingMap using swap gates.
@@ -438,7 +441,7 @@ def _controlflow_layer_update(self, dagcircuit_output, layer_dag, current_layout
                 root_dag, self.coupling_map, layout, final_layout, seed=self._new_seed()
             )
             if swap_dag.size(recurse=False):
-                updated_dag_block.compose(swap_dag, qubits=swap_qubits)
+                updated_dag_block.compose(swap_dag, qubits=swap_qubits, inline_captures=True)
             idle_qubits &= set(updated_dag_block.idle_wires())
 
         # Now for each block, expand it to be full width over all active wires (all blocks of a
@@ -504,10 +507,18 @@ def _dag_from_block(block, node, root_dag):
         out.add_qreg(qreg)
     # For clbits, we need to take more care.  Nested control-flow might need registers to exist for
     # conditions on inner blocks.  `DAGCircuit.substitute_node_with_dag` handles this register
-    # mapping when required, so we use that with a dummy block.
+    # mapping when required, so we use that with a dummy block that pretends to act on all variables
+    # in the DAG.
     out.add_clbits(node.cargs)
+    for var in block.iter_input_vars():
+        out.add_input_var(var)
+    for var in block.iter_captured_vars():
+        out.add_captured_var(var)
+    for var in block.iter_declared_vars():
+        out.add_declared_var(var)
+
     dummy = out.apply_operation_back(
-        Instruction("dummy", len(node.qargs), len(node.cargs), []),
+        IfElseOp(expr.lift(True), block.copy_empty_like(vars_mode="captures")),
         node.qargs,
         node.cargs,
         check=False,

qiskit/transpiler/passes/synthesis/high_level_synthesis.py

@@ -363,7 +363,7 @@ def __init__(
 
         # include path for when target exists but target.num_qubits is None (BasicSimulator)
         if not self._top_level_only and (self._target is None or self._target.num_qubits is None):
-            basic_insts = {"measure", "reset", "barrier", "snapshot", "delay"}
+            basic_insts = {"measure", "reset", "barrier", "snapshot", "delay", "store"}
             self._device_insts = basic_insts | set(self._basis_gates)
 
     def run(self, dag: DAGCircuit) -> DAGCircuit:

qiskit/transpiler/passes/utils/gates_basis.py

@@ -32,7 +32,7 @@ def __init__(self, basis_gates=None, target=None):
         self._basis_gates = None
         if basis_gates is not None:
             self._basis_gates = set(basis_gates).union(
-                {"measure", "reset", "barrier", "snapshot", "delay"}
+                {"measure", "reset", "barrier", "snapshot", "delay", "store"}
             )
         self._target = target
 
@@ -46,8 +46,8 @@ def run(self, dag):
 
             def _visit_target(dag, wire_map):
                 for gate in dag.op_nodes():
-                    # Barrier is universal and supported by all backends
-                    if gate.name == "barrier":
+                    # Barrier and store are assumed universal and supported by all backends
+                    if gate.name in ("barrier", "store"):
                         continue
                     if not self._target.instruction_supported(
                         gate.name, tuple(wire_map[bit] for bit in gate.qargs)

test/python/circuit/test_circuit_operations.py

@@ -1002,6 +1002,31 @@ def test_reverse(self):
 
         self.assertEqual(qc.reverse_ops(), expected)
 
+    def test_reverse_with_standlone_vars(self):
+        """Test that instruction-reversing works in the presence of stand-alone variables."""
+        a = expr.Var.new("a", types.Bool())
+        b = expr.Var.new("b", types.Uint(8))
+        c = expr.Var.new("c", types.Uint(8))
+
+        qc = QuantumCircuit(2, inputs=[a])
+        qc.add_var(b, 12)
+        qc.h(0)
+        qc.cx(0, 1)
+        with qc.if_test(a):
+            # We don't really comment on what should happen within control-flow operations in this
+            # method - it's not really defined in a non-linear CFG.  This deliberately uses a body
+            # of length 1 (a single `Store`), so there's only one possibility.
+            qc.add_var(c, 12)
+
+        expected = qc.copy_empty_like()
+        with expected.if_test(a):
+            expected.add_var(c, 12)
+        expected.cx(0, 1)
+        expected.h(0)
+        expected.store(b, 12)
+
+        self.assertEqual(qc.reverse_ops(), expected)
+
     def test_repeat(self):
         """Test repeating the circuit works."""
         qr = QuantumRegister(2)

test/python/compiler/test_transpiler.py

@@ -42,13 +42,13 @@
     SwitchCaseOp,
     WhileLoopOp,
 )
+from qiskit.circuit.classical import expr, types
 from qiskit.circuit.annotated_operation import (
     AnnotatedOperation,
     InverseModifier,
     ControlModifier,
     PowerModifier,
 )
-from qiskit.circuit.classical import expr
 from qiskit.circuit.delay import Delay
 from qiskit.circuit.measure import Measure
 from qiskit.circuit.reset import Reset
@@ -2175,6 +2175,38 @@ def _control_flow_expr_circuit(self):
                 base.append(CustomCX(), [3, 4])
         return base
 
+    def _standalone_var_circuit(self):
+        a = expr.Var.new("a", types.Bool())
+        b = expr.Var.new("b", types.Uint(8))
+        c = expr.Var.new("c", types.Uint(8))
+
+        qc = QuantumCircuit(5, 5, inputs=[a])
+        qc.add_var(b, 12)
+        qc.h(0)
+        qc.cx(0, 1)
+        qc.measure([0, 1], [0, 1])
+        qc.store(a, expr.bit_xor(qc.clbits[0], qc.clbits[1]))
+        with qc.if_test(a) as else_:
+            qc.cx(2, 3)
+            qc.cx(3, 4)
+            qc.cx(4, 2)
+        with else_:
+            qc.add_var(c, 12)
+        with qc.while_loop(a):
+            with qc.while_loop(a):
+                qc.add_var(c, 12)
+                qc.cz(1, 0)
+                qc.cz(4, 1)
+                qc.store(a, False)
+        with qc.switch(expr.bit_and(b, 7)) as case:
+            with case(0):
+                qc.cz(0, 1)
+                qc.cx(1, 2)
+                qc.cy(2, 0)
+            with case(case.DEFAULT):
+                qc.store(b, expr.bit_and(b, 7))
+        return qc
+
     @data(0, 1, 2, 3)
     def test_qpy_roundtrip(self, optimization_level):
         """Test that the output of a transpiled circuit can be round-tripped through QPY."""
@@ -2300,6 +2332,46 @@ def test_qpy_roundtrip_control_flow_expr_backendv2(self, optimization_level):
         round_tripped = qpy.load(buffer)[0]
         self.assertEqual(round_tripped, transpiled)
 
+    @data(0, 1, 2, 3)
+    def test_qpy_roundtrip_standalone_var(self, optimization_level):
+        """Test that the output of a transpiled circuit with control flow including standalone `Var`
+        nodes can be round-tripped through QPY."""
+        backend = GenericBackendV2(num_qubits=7)
+        transpiled = transpile(
+            self._standalone_var_circuit(),
+            backend=backend,
+            basis_gates=backend.operation_names
+            + ["if_else", "for_loop", "while_loop", "switch_case"],
+            optimization_level=optimization_level,
+            seed_transpiler=2024_05_01,
+        )
+        buffer = io.BytesIO()
+        qpy.dump(transpiled, buffer)
+        buffer.seek(0)
+        round_tripped = qpy.load(buffer)[0]
+        self.assertEqual(round_tripped, transpiled)
+
+    @data(0, 1, 2, 3)
+    def test_qpy_roundtrip_standalone_var_target(self, optimization_level):
+        """Test that the output of a transpiled circuit with control flow including standalone `Var`
+        nodes can be round-tripped through QPY."""
+        backend = GenericBackendV2(num_qubits=11)
+        backend.target.add_instruction(IfElseOp, name="if_else")
+        backend.target.add_instruction(ForLoopOp, name="for_loop")
+        backend.target.add_instruction(WhileLoopOp, name="while_loop")
+        backend.target.add_instruction(SwitchCaseOp, name="switch_case")
+        transpiled = transpile(
+            self._standalone_var_circuit(),
+            backend=backend,
+            optimization_level=optimization_level,
+            seed_transpiler=2024_05_01,
+        )
+        buffer = io.BytesIO()
+        qpy.dump(transpiled, buffer)
+        buffer.seek(0)
+        round_tripped = qpy.load(buffer)[0]
+        self.assertEqual(round_tripped, transpiled)
+
     @data(0, 1, 2, 3)
     def test_qasm3_output(self, optimization_level):
         """Test that the output of a transpiled circuit can be dumped into OpenQASM 3."""
@@ -2350,6 +2422,21 @@ def test_qasm3_output_control_flow_expr(self, optimization_level):
             str,
         )
 
+    @data(0, 1, 2, 3)
+    def test_qasm3_output_standalone_var(self, optimization_level):
+        """Test that the output of a transpiled circuit with control flow and standalone `Var` nodes
+        can be dumped into OpenQASM 3."""
+        transpiled = transpile(
+            self._standalone_var_circuit(),
+            backend=GenericBackendV2(num_qubits=13, control_flow=True),
+            optimization_level=optimization_level,
+            seed_transpiler=2024_05_01,
+        )
+        # TODO: There's not a huge amount we can sensibly test for the output here until we can
+        # round-trip the OpenQASM 3 back into a Terra circuit.  Mostly we're concerned that the dump
+        # itself doesn't throw an error, though.
+        self.assertIsInstance(qasm3.dumps(transpiled), str)
+
     @data(0, 1, 2, 3)
     def test_transpile_target_no_measurement_error(self, opt_level):
         """Test that transpile with a target which contains ideal measurement works

test/python/transpiler/test_apply_layout.py

@@ -15,6 +15,7 @@
 import unittest
 
 from qiskit.circuit import QuantumRegister, QuantumCircuit, ClassicalRegister
+from qiskit.circuit.classical import expr, types
 from qiskit.converters import circuit_to_dag
 from qiskit.transpiler.layout import Layout
 from qiskit.transpiler.passes import ApplyLayout, SetLayout
@@ -167,6 +168,31 @@ def test_final_layout_is_updated(self):
             ),
         )
 
+    def test_works_with_var_nodes(self):
+        """Test that standalone var nodes work."""
+        a = expr.Var.new("a", types.Bool())
+        b = expr.Var.new("b", types.Uint(8))
+
+        qc = QuantumCircuit(2, 2, inputs=[a])
+        qc.add_var(b, 12)
+        qc.h(0)
+        qc.cx(0, 1)
+        qc.measure([0, 1], [0, 1])
+        qc.store(a, expr.bit_and(a, expr.bit_xor(qc.clbits[0], qc.clbits[1])))
+
+        expected = QuantumCircuit(QuantumRegister(2, "q"), *qc.cregs, inputs=[a])
+        expected.add_var(b, 12)
+        expected.h(1)
+        expected.cx(1, 0)
+        expected.measure([1, 0], [0, 1])
+        expected.store(a, expr.bit_and(a, expr.bit_xor(qc.clbits[0], qc.clbits[1])))
+
+        pass_ = ApplyLayout()
+        pass_.property_set["layout"] = Layout(dict(enumerate(reversed(qc.qubits))))
+        after = pass_(qc)
+
+        self.assertEqual(after, expected)
+
 
 if __name__ == "__main__":
     unittest.main()