Batch QuantumCircuit with multiple parameter_binds (Qiskit#1840)

`Estimator` in Aer did not use parameter bindings appropriately. Backend is called multiple times 
to simulate a circuit with multiple parameter-sets. This commit is to call backend once for multiple
parameter-sets for a circuit.

* WIP
* refactor
* lint
* add reno

qiskit_aer/primitives/estimator.py

@@ -22,7 +22,7 @@
 from warnings import warn
 
 import numpy as np
-from qiskit.circuit import QuantumCircuit
+from qiskit.circuit import ParameterExpression, QuantumCircuit
 from qiskit.compiler import transpile
 from qiskit.opflow import PauliSumOp
 from qiskit.primitives import BaseEstimator, EstimatorResult
@@ -364,55 +364,80 @@ def _compute_with_approximation(
     ):
         # Key for cache
         key = (tuple(circuits), tuple(observables), self.approximation)
-        parameter_binds = []
         shots = run_options.pop("shots", None)
+
         # Create expectation value experiments.
         if key in self._cache:  # Use a cache
-            experiments, experiment_data = self._cache[key]
+            parameter_binds = defaultdict(dict)
             for i, j, value in zip(circuits, observables, parameter_values):
                 self._validate_parameter_length(value, i)
-                parameter_binds.append({k: [v] for k, v in zip(self._parameters[i], value)})
+                for k, v in zip(self._parameters[i], value):
+                    if k in parameter_binds[(i, j)]:
+                        parameter_binds[(i, j)][k].append(v)
+                    else:
+                        parameter_binds[(i, j)][k] = [v]
+            experiment_manager = self._cache[key]
+            experiment_manager.parameter_binds = list(parameter_binds.values())
         else:
             self._transpile_circuits(circuits)
-            experiments = []
-            experiment_data = []
+            experiment_manager = _ExperimentManager()
             for i, j, value in zip(circuits, observables, parameter_values):
-                self._validate_parameter_length(value, i)
-                circuit = (
-                    self._circuits[i].copy()
-                    if self._skip_transpilation
-                    else self._transpiled_circuits[i].copy()
-                )
-                observable = self._observables[j]
-                experiment_data.append(observable)
-                if shots is None:
-                    circuit.save_expectation_value(observable, self._layouts[i])
+                if (i, j) in experiment_manager.keys:
+                    self._validate_parameter_length(value, i)
+                    experiment_manager.append(
+                        key=(i, j),
+                        parameter_bind=dict(zip(self._parameters[i], value)),
+                    )
                 else:
-                    for term_ind, pauli in enumerate(observable.paulis):
-                        circuit.save_expectation_value(pauli, self._layouts[i], label=str(term_ind))
-                experiments.append(circuit)
-                parameter_binds.append({k: [v] for k, v in zip(self._parameters[i], value)})
-            self._cache[key] = (experiments, experiment_data)
-        parameter_binds = parameter_binds if any(parameter_binds) else None
+                    self._validate_parameter_length(value, i)
+                    circuit = (
+                        self._circuits[i].copy()
+                        if self._skip_transpilation
+                        else self._transpiled_circuits[i].copy()
+                    )
+
+                    observable = self._observables[j]
+                    if shots is None:
+                        circuit.save_expectation_value(observable, self._layouts[i])
+                    else:
+                        for term_ind, pauli in enumerate(observable.paulis):
+                            circuit.save_expectation_value(
+                                pauli, self._layouts[i], label=str(term_ind)
+                            )
+                    experiment_manager.append(
+                        key=(i, j),
+                        parameter_bind=dict(zip(self._parameters[i], value)),
+                        experiment_circuit=circuit,
+                    )
+
+            self._cache[key] = experiment_manager
         result = self._backend.run(
-            experiments, parameter_binds=parameter_binds, **run_options
+            experiment_manager.experiment_circuits,
+            parameter_binds=experiment_manager.parameter_binds,
+            **run_options,
         ).result()
 
         # Post processing (calculate expectation values)
         if shots is None:
-            expectation_values = [result.data(i)["expectation_value"] for i in range(len(circuits))]
+            expectation_values = [
+                result.data(i)["expectation_value"] for i in experiment_manager.experiment_indices
+            ]
             metadata = [
-                {"simulator_metadata": result.results[i].metadata} for i in range(len(experiments))
+                {"simulator_metadata": result.results[i].metadata}
+                for i in experiment_manager.experiment_indices
             ]
         else:
             expectation_values = []
             rng = np.random.default_rng(seed)
             metadata = []
-            for i in range(len(experiments)):
+            experiment_indices = experiment_manager.experiment_indices
+            for i in range(len(experiment_manager)):
                 combined_expval = 0.0
                 combined_var = 0.0
-                coeffs = np.real_if_close(experiment_data[i].coeffs)
-                for term_ind, expval in result.data(i).items():
+                result_index = experiment_indices[i]
+                observable_key = experiment_manager.get_observable_key(i)
+                coeffs = np.real_if_close(self._observables[observable_key].coeffs)
+                for term_ind, expval in result.data(result_index).items():
                     var = 1 - expval**2
                     coeff = coeffs[int(term_ind)]
                     combined_expval += expval * coeff
@@ -424,7 +449,7 @@ def _compute_with_approximation(
                     {
                         "variance": np.real_if_close(combined_var).item(),
                         "shots": shots,
-                        "simulator_metadata": result.results[i].metadata,
+                        "simulator_metadata": result.results[result_index].metadata,
                     }
                 )
 
@@ -575,3 +600,49 @@ def _paulis2basis(paulis: PauliList) -> Pauli:
             np.logical_or.reduce(paulis.x),  # pylint:disable=no-member
         )
     )
+
+
+class _ExperimentManager:
+    def __init__(self):
+        self.keys: list[tuple[int, int]] = []
+        self.experiment_circuits: list[QuantumCircuit] = []
+        self.parameter_binds: list[dict[ParameterExpression, list[float]]] = []
+        self._input_indices: list[list[int]] = []
+        self._num_experiment: int = 0
+
+    def __len__(self):
+        return self._num_experiment
+
+    @property
+    def experiment_indices(self):
+        """indices of experiments"""
+        return sum(self._input_indices, [])
+
+    def append(
+        self,
+        key: tuple[int, int],
+        parameter_bind: dict[ParameterExpression, float],
+        experiment_circuit: QuantumCircuit | None = None,
+    ):
+        """append experiments"""
+        if experiment_circuit is not None:
+            self.experiment_circuits.append(experiment_circuit)
+
+        if key in self.keys:
+            key_index = self.keys.index(key)
+            for k, vs in self.parameter_binds[key_index].items():
+                vs.append(parameter_bind[k])
+            self._input_indices[key_index].append(self._num_experiment)
+        else:
+            self.keys.append(key)
+            self.parameter_binds.append({k: [v] for k, v in parameter_bind.items()})
+            self._input_indices.append([self._num_experiment])
+
+        self._num_experiment += 1
+
+    def get_observable_key(self, index):
+        """return key of observables"""
+        for i, inputs in enumerate(self._input_indices):
+            if index in inputs:
+                return self.keys[i][1]
+        raise AerError("Unexpected behavior.")

releasenotes/notes/estimator-performance-da83a59b9fd69086.yaml

@@ -0,0 +1,5 @@
+---
+upgrade:
+  - |
+    Improved performance when the same circuits and multiple parameters are passed to
+    :class:`~.Estimator` with ``approximation=True``.