Verify values of parametrized rzz gates

qiskit_ibm_runtime/base_primitive.py

@@ -27,7 +27,7 @@
 from .options.utils import merge_options_v2
 from .runtime_job_v2 import RuntimeJobV2
 from .ibm_backend import IBMBackend
-from .utils import validate_isa_circuits, validate_no_dd_with_dynamic_circuits
+from .utils import validate_isa_circuits, validate_no_dd_with_dynamic_circuits, validate_rzz_pubs
 from .utils.default_session import get_cm_session
 from .utils.deprecation import issue_deprecation_msg
 from .utils.utils import is_simulator
@@ -171,6 +171,8 @@ def _run(self, pubs: Union[list[EstimatorPub], list[SamplerPub]]) -> RuntimeJobV
 
         validate_no_dd_with_dynamic_circuits([pub.circuit for pub in pubs], self.options)
         if self._backend:
+            if not is_simulator(self._backend):
+                validate_rzz_pubs(pubs)
             for pub in pubs:
                 if getattr(self._backend, "target", None) and not is_simulator(self._backend):
                     validate_isa_circuits([pub.circuit], self._backend.target)

qiskit_ibm_runtime/utils/__init__.py

@@ -33,6 +33,7 @@
     validate_no_dd_with_dynamic_circuits,
     validate_isa_circuits,
     validate_job_tags,
+    validate_rzz_pubs,
 )
 
 from .json import RuntimeEncoder, RuntimeDecoder, to_base64_string

qiskit_ibm_runtime/utils/utils.py

@@ -20,18 +20,22 @@
 import re
 from queue import Queue
 from threading import Condition
-from typing import List, Optional, Any, Dict, Union, Tuple
+from typing import List, Optional, Any, Dict, Union, Tuple, Set
 from urllib.parse import urlparse
+from itertools import chain
 import numpy as np
 
 import requests
 from ibm_cloud_sdk_core.authenticators import (  # pylint: disable=import-error
     IAMAuthenticator,
 )
 from ibm_platform_services import ResourceControllerV2  # pylint: disable=import-error
-from qiskit.circuit import QuantumCircuit, ControlFlowOp, ParameterExpression
+from qiskit.circuit import QuantumCircuit, ControlFlowOp, Parameter, ParameterExpression
 from qiskit.transpiler import Target
 from qiskit.providers.backend import BackendV1, BackendV2
+from qiskit.primitives.containers.estimator_pub import EstimatorPub
+from qiskit.primitives.containers.sampler_pub import SamplerPub
+
 from .deprecation import deprecate_function
 
 
@@ -116,6 +120,113 @@ def is_isa_circuit(circuit: QuantumCircuit, target: Target) -> str:
     return _is_isa_circuit_helper(circuit, target, qubit_map)
 
 
+def _is_rzz_pub_helper(circuit: QuantumCircuit) -> Union[str, Set[Parameter]]:
+    """
+    For rzz gates:
+    - Verify that numeric angles are in the range [0, pi/2]
+    - Collect parameterized angles
+
+    Returns one of the following:
+    - A string, containing an error message, if a numeric angle is outside of the range [0, pi/2]
+    - A list of names of all the parameters that participate in an rzz gate
+
+    Note: we check for parametrized rzz gates inside control flow operation, although fractional
+    gates are actually impossible in combination with dynamic circuits. This is in order to remain
+    correct if this restriction is removed at some point.
+    """
+    angle_params = set()
+
+    for instruction in circuit.data:
+        operation = instruction.operation
+
+        # rzz gate is calibrated only for the range [0, pi/2].
+        # We allow an angle value of a bit more than pi/2, to compensate floating point rounding
+        # errors (beyond pi/2 does not trigger an error down the stack, only may become less
+        # accurate).
+        if operation.name == "rzz":
+            angle = instruction.operation.params[0]
+            if isinstance(angle, Parameter):
+                angle_params.add(angle.name)
+            elif not isinstance(angle, ParameterExpression) and (
+                angle < 0.0 or angle > 1.001 * np.pi / 2
+            ):
+                return (
+                    "The instruction rzz is supported only for angles in the "
+                    f"range [0, pi/2], but an angle of {angle} has been provided."
+                )
+
+        if isinstance(operation, ControlFlowOp):
+            for sub_circ in operation.blocks:
+                body_result = _is_rzz_pub_helper(sub_circ)
+                if isinstance(body_result, str):
+                    return body_result
+                angle_params.update(body_result)
+
+    return angle_params
+
+
+def is_rzz_pub(pub: Union[EstimatorPub, SamplerPub]) -> str:
+    """Verify that all rzz angles are in the range [0, pi/2].
+
+    Args:
+        pub: A pub to be checked
+
+    Returns:
+        An empty string if all angles are valid, otherwise an error message.
+    """
+    helper_result = _is_rzz_pub_helper(pub.circuit)
+
+    if isinstance(helper_result, str):
+        return helper_result
+
+    if len(helper_result) == 0:
+        return ""
+
+    # helper_result is a set of parameter names
+    rzz_params = list(helper_result)
+
+    param_values = pub.parameter_values
+    # param_values is of the form:
+    # BindingsArray(<shape=(2, 2, 3), num_parameters=4, parameters=['a', 'b', 'c', 'd']>)
+    # param_values.data is a dictionary, whose keys are tuples of parameter names.
+    # For examples, the keys can be: dict_keys([('a', 'b'), ('c',), ('d',)])
+
+    pub_params = list(chain(*[list(param_names) for param_names in param_values.data.keys()]))
+    # pub_params is the list of parameter names in the pub, for example: ['a', 'b', 'c', 'd']
+
+    col_indices = np.where(np.isin(pub_params, rzz_params))[0]
+    # col_indices is the indices of columns in the parameter value array that have to be checked
+
+    arr = param_values.as_array()
+
+    # almost-flatten the parameter:
+    # 'arr' will be a 2-dimensional array, where each line represents assignment of values to
+    # the circuit parameter. For example
+    # [[ 1.  2. 25. 45.]
+    #  [ 3.  4. 26. 46.]]
+    # The first line is an assignment of 1 to the first parameter, 2 to the second parameter,
+    # 25 to the third parameter, and 45 to the fourth parameter
+    arr = arr.reshape(-1, arr.shape[-1])
+
+    # project only to the parameters that have to be checked
+    arr = arr[:, col_indices]
+
+    # We allow an angle value of a bit more than pi/2, to compensate floating point rounding
+    # errors (beyond pi/2 does not trigger an error down the stack, only may become less
+    # accurate).
+    bad = np.where((arr < 0.0) | (arr > 1.001 * np.pi / 2))
+
+    # `bad` is a tuple of two arrays, which can be empty, like this:
+    # (array([], dtype=int64), array([], dtype=int64))
+    if len(bad[0]) > 0:
+        return (
+            f"Assignment of value {arr[bad[0][0], bad[1][0]]} to Parameter "
+            f"'{pub_params[col_indices[bad[1][0]]]}' is an invalid angle for the rzz gate"
+        )
+
+    return ""
+
+
 def are_circuits_dynamic(circuits: List[QuantumCircuit], qasm_default: bool = True) -> bool:
     """Checks if the input circuits are dynamic."""
     for circuit in circuits:

qiskit_ibm_runtime/utils/validations.py

@@ -11,15 +11,15 @@
 # that they have been altered from the originals.
 
 """Utilities for data validation."""
-from typing import List, Sequence, Optional, Any
+from typing import List, Sequence, Optional, Any, Union
 import warnings
 import keyword
 
 from qiskit import QuantumCircuit
 from qiskit.transpiler import Target
 from qiskit.primitives.containers.sampler_pub import SamplerPub
 from qiskit.primitives.containers.estimator_pub import EstimatorPub
-from qiskit_ibm_runtime.utils.utils import is_isa_circuit, are_circuits_dynamic
+from qiskit_ibm_runtime.utils.utils import is_isa_circuit, are_circuits_dynamic, is_rzz_pub
 from qiskit_ibm_runtime.exceptions import IBMInputValueError
 
 
@@ -98,6 +98,18 @@ def validate_isa_circuits(circuits: Sequence[QuantumCircuit], target: Target) ->
             )
 
 
+def validate_rzz_pubs(pubs: Union[List[EstimatorPub], List[SamplerPub]]) -> None:
+    """Validate that rzz angles are always in the range [0, pi/2]
+
+    Args:
+        pubs: A list of pubs.
+    """
+    for pub in pubs:
+        message = is_rzz_pub(pub)
+        if message:
+            raise IBMInputValueError(message)
+
+
 def validate_no_dd_with_dynamic_circuits(circuits: List[QuantumCircuit], options: Any) -> None:
     """Validate that if dynamical decoupling options are enabled,
     no circuit in the pubs is dynamic

test/unit/test_sampler.py

@@ -285,7 +285,7 @@ def test_isa_inside_condition_block_body_in_separate_circuit(self, backend):
                 SamplerV2(backend).run(pubs=[(circ)])
 
     @data(-1, 1, 2)
-    def test_rzz_angle_validation(self, angle):
+    def test_rzz_fixed_angle_validation(self, angle):
         """Test exception when rzz gate is used with an angle outside the range [0, pi/2]"""
         backend = FakeFractionalBackend()
 
@@ -295,12 +295,13 @@ def test_rzz_angle_validation(self, angle):
         if angle == 1:
             SamplerV2(backend).run(pubs=[(circ)])
         else:
-            with self.assertRaises(IBMInputValueError):
+            with self.assertRaisesRegex(IBMInputValueError, f"{angle}"):
                 SamplerV2(backend).run(pubs=[(circ)])
 
-    def test_rzz_validates_only_for_fixed_angles(self):
-        """Verify that the rzz validation occurs only when the angle is a number, and not a
-        parameter"""
+    @data(-1, 1, 2)
+    def test_rzz_parametrized_angle_validation(self, angle):
+        """Test exception when rzz gate is used with a parameter which is assigned a value outside
+        the range [0, pi/2]"""
         backend = FakeFractionalBackend()
         param = Parameter("p")
 
@@ -310,8 +311,82 @@ def test_rzz_validates_only_for_fixed_angles(self):
             # Should run without an error
             SamplerV2(backend).run(pubs=[(circ, [1])])
 
-        with self.subTest("parameter expression"):
-            circ = QuantumCircuit(2)
-            circ.rzz(2 * param, 0, 1)
-            # Should run without an error
-            SamplerV2(backend).run(pubs=[(circ, [0.5])])
+        if angle == 1:
+            SamplerV2(backend).run(pubs=[(circ, [angle])])
+        else:
+            with self.assertRaisesRegex(IBMInputValueError, f"{angle}.*Parameter 'p'"):
+                SamplerV2(backend).run(pubs=[(circ, [angle])])
+
+    @data(("a", -1), ("b", 2), ("d", 3), (-1, 1), (1, 2), None)
+    def test_rzz_complex(self, flawed_params):
+        """Testing rzz validation in the currently non-existing case of dynamic instructions"""
+        # pylint: disable=not-context-manager
+
+        # FakeFractionalBackend has both fractional and dynamic instructions
+        backend = FakeFractionalBackend()
+
+        aparam = Parameter("a")
+        bparam = Parameter("b")
+        cparam = Parameter("c")
+        dparam = Parameter("d")
+
+        angle1 = 1
+        angle2 = 1
+        if flawed_params is not None and not isinstance(flawed_params[0], str):
+            angle1 = flawed_params[0]
+            angle2 = flawed_params[1]
+
+        circ = QuantumCircuit(2, 1)
+        circ.rzz(bparam, 0, 1)
+        circ.rzz(angle1, 0, 1)
+        circ.measure(0, 0)
+        with circ.if_test((0, 1)):
+            circ.rzz(aparam, 0, 1)
+            circ.rzz(angle2, 0, 1)
+        circ.rx(cparam, 0)
+        circ.rzz(dparam, 0, 1)
+        circ.rzz(1, 0, 1)
+        circ.rzz(aparam, 0, 1)
+
+        val_ab = np.ones([2, 2, 3, 2])
+        val_c = (-1) * np.ones([2, 2, 3])
+        val_d = np.ones([2, 2, 3])
+
+        if flawed_params is not None and isinstance(flawed_params[0], str):
+            if flawed_params[0] == "a":
+                val_ab[0, 1, 1, 0] = flawed_params[1]
+                val_ab[1, 0, 2, 1] = flawed_params[1]
+            if flawed_params[0] == "b":
+                val_ab[1, 0, 2, 1] = flawed_params[1]
+                val_d[1, 1, 1] = flawed_params[1]
+            if flawed_params[0] == "d":
+                val_d[1, 1, 1] = flawed_params[1]
+                val_ab[1, 1, 2, 1] = flawed_params[1]
+
+        pub = (circ, {("a", "b"): val_ab, "c": val_c, "d": val_d})
+
+        if flawed_params is None:
+            SamplerV2(backend).run(pubs=[pub])
+        else:
+            if isinstance(flawed_params[0], str):
+                with self.assertRaisesRegex(
+                    IBMInputValueError, f"{flawed_params[1]}.*Parameter '{flawed_params[0]}'"
+                ):
+                    SamplerV2(backend).run(pubs=[pub])
+            else:
+                with self.assertRaisesRegex(
+                    IBMInputValueError, f"{flawed_params[0] * flawed_params[1]}"
+                ):
+                    SamplerV2(backend).run(pubs=[pub])
+
+    def test_rzz_validation_skips_param_exp(self):
+        """Verify that the rzz validation occurs only when the angle is a number or a parameter,
+        but not a parameter expression"""
+        backend = FakeFractionalBackend()
+        param = Parameter("p")
+
+        circ = QuantumCircuit(2)
+        circ.rzz(2 * param, 0, 1)
+
+        # Should run without an error
+        SamplerV2(backend).run(pubs=[(circ, [1])])