Repository: Qiskit/qiskit-ibm-runtime
Pull request #546 (2022-09-26 18:53:37)

Update for flexible session with data caching
<!--
⚠️ The 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.
-->

### Summary
Calculate circuit id by serializing and hashing the circuit key. Pass `circuits_map` and `circuit_ids` instead of `circuit_indices`.


### Details and comments
Fixes # 



CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/Qiskit/qiskit-ibm-runtime?pullRequest=546) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/Qiskit/qiskit-ibm-runtime?pullRequest=546) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/Qiskit/qiskit-ibm-runtime?pullRequest=546) it.</sub>
coveralls: ## Pull Request Test Coverage Report for [Build 3130269279](https://coveralls.io/builds/52768983)

* **35** of **41**   **(85.37%)**  changed or added relevant lines in **3** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.3%**) to **64.515%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [qiskit_ibm_runtime/estimator.py](https://coveralls.io/builds/52768983/source?filename=qiskit_ibm_runtime%2Festimator.py#L148) | 16 | 19 | 84.21%
| [qiskit_ibm_runtime/sampler.py](https://coveralls.io/builds/52768983/source?filename=qiskit_ibm_runtime%2Fsampler.py#L121) | 16 | 19 | 84.21%
<!-- | **Total:** | **35** | **41** | **85.37%** | -->


|  Totals | [![Coverage Status](https://coveralls.io/builds/52768983/badge)](https://coveralls.io/builds/52768983) |
| :-- | --: |
| Change from base [Build 3100068626](https://coveralls.io/builds/52659656): |  0.3% |
| Covered Lines: | 3278 |
| Relevant Lines: | 5081 |

---
##### 💛  - [Coveralls](https://coveralls.io)

coveralls: ## Pull Request Test Coverage Report for [Build 3130269279](https://coveralls.io/builds/52768983)

* **-6** of **41**   **(85.37%)**  changed or added relevant lines in **3** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.3%**) to **64.515%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [qiskit_ibm_runtime/estimator.py](https://coveralls.io/builds/52768983/source?filename=qiskit_ibm_runtime%2Festimator.py#L148) | 16 | 19 | 84.21%
| [qiskit_ibm_runtime/sampler.py](https://coveralls.io/builds/52768983/source?filename=qiskit_ibm_runtime%2Fsampler.py#L121) | 16 | 19 | 84.21%
<!-- | **Total:** | **35** | **41** | **85.37%** | -->


|  Totals | [![Coverage Status](https://coveralls.io/builds/52768983/badge)](https://coveralls.io/builds/52768983) |
| :-- | --: |
| Change from base [Build 3100068626](https://coveralls.io/builds/52659656): |  0.3% |
| Covered Lines: | 3278 |
| Relevant Lines: | 5081 |

---
##### 💛  - [Coveralls](https://coveralls.io)

jyu00: @ChitsimranGill Can you add some test cases? Both unit and integration. 

Also caching doesn't work with simulators, so you'll need to check that (which may be a bit tricky for cloud channel where backend can be omitted). 
ChitsimranGill: > @ChitsimranGill Can you add some test cases? Both unit and integration.
> 
> Also caching doesn't work with simulators, so you'll need to check that (which may be a bit tricky for cloud channel where backend can be omitted).

I think the existing integration tests are good enough, I will write unit tests for this tomorrow. 
jyu00: > I think the existing integration tests are good enough, I will write unit tests for this tomorrow.

Does any of the existing integration tests actually test data caching? 
ChitsimranGill: We added unit tests for data caching in sampler and estimator. We are not sure how to assert this in integration tests. Hopefully these unit tests will sufficient.
jyu00: > We are not sure how to assert this in integration tests. Hopefully these unit tests will sufficient.

You can do something similar to what you did in unit test, e.g. 

```
with Session(...):
    sampler.run([circ1, circ2], ...)
    sampler.run([circ1], ...)
    sampler.run([circ3], ...)
    sampler.run([circ1, circ4], ...)
```
where all the circuits all have different number of qubits, so you can tell from the result whether the correct circuit was used. Also different transpilation options - i.e. if you send in `circ1 + skip_transpilation=False` and `circ1+skip_transpilation=True`, it should not be reused. `skip_transpilation` is the easiest to test with because `skip_transpilation=True` will fail the job. 


Also don't forget this comment:

> Also caching doesn't work with simulators, so you'll need to check that (which may be a bit tricky for cloud channel where backend can be omitted).
ChitsimranGill: > 

@jyu00 we wrote integration tests for sampler circuit caching.

@jyu00 @ikkoham @t-imamichi could you provide four sample circuits and observables for estimator with 2,3,4, and 5 qubits, which provide consistent results which we can assert. 

@jyu00 we tried `skip_transpilation` example that you suggested, but it did not fail on simulator. Right now we can only test on simulators from qiskit-ibm-runtime. Do you have an example that might actually fail on simulator without transpilation. 
jyu00: > @jyu00 we wrote integration tests for sampler circuit caching.

Excellent! 
 
> @jyu00 @ikkoham @t-imamichi could you provide four sample circuits and observables for estimator with 2,3,4, and 5 qubits, which provide consistent results which we can assert.

You can use different gates instead of different qubit counts. For example, 

```
qc1 = QuantumCircuit(2)
qc1.x(range(2))
qc1.measure_all()
qc2 = QuantumCircuit(2)
qc2.measure_all()
qc3 = QuantumCircuit(2)
qc3.h(range(2))
qc3.measure_all()
```
And observable `H = SparsePauliOp.from_list([("IZ", 1)])`  will give you values of -1 for qc1, 1 for qc2, and somewhere in between for qc3. 


> @jyu00 we tried `skip_transpilation` example that you suggested, but it did not fail on simulator. Right now we can only test on simulators from qiskit-ibm-runtime. Do you have an example that might actually fail on simulator without transpilation.

Right, I forgot the simulator supports all gates. Try specifying `initial_layout=list(range(x))`, where x is smaller than the number of qubits you have. It should give you an error saying the length of the list needs to match that of the qubit count.

ChitsimranGill: > > @jyu00 we wrote integration tests for sampler circuit caching.
> 
> Excellent!
> 
> > @jyu00 @ikkoham @t-imamichi could you provide four sample circuits and observables for estimator with 2,3,4, and 5 qubits, which provide consistent results which we can assert.
> 
> You can use different gates instead of different qubit counts. For example,
> 
> ```
> qc1 = QuantumCircuit(2)
> qc1.x(range(2))
> qc1.measure_all()
> qc2 = QuantumCircuit(2)
> qc2.measure_all()
> qc3 = QuantumCircuit(2)
> qc3.h(range(2))
> qc3.measure_all()
> ```
> 
> And observable `H = SparsePauliOp.from_list([("IZ", 1)])` will give you values of -1 for qc1, 1 for qc2, and somewhere in between for qc3.
> 
> > @jyu00 we tried `skip_transpilation` example that you suggested, but it did not fail on simulator. Right now we can only test on simulators from qiskit-ibm-runtime. Do you have an example that might actually fail on simulator without transpilation.
> 
> Right, I forgot the simulator supports all gates. Try specifying `initial_layout=list(range(x))`, where x is smaller than the number of qubits you have. It should give you an error saying the length of the list needs to match that of the qubit count.

@jyu00 added integration tests to test circuit caching with transpilation options for both sampler and estimator.

Patch:
diff --git a/qiskit_ibm_runtime/estimator.py b/qiskit_ibm_runtime/estimator.py
index 15c9d240..7923100f 100644
--- a/qiskit_ibm_runtime/estimator.py
+++ b/qiskit_ibm_runtime/estimator.py
@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 import copy
+import json
 from typing import Iterable, Optional, Dict, Sequence, Any, Union, Callable
 
 from qiskit.circuit import QuantumCircuit, Parameter
@@ -25,8 +26,11 @@ from qiskit.quantum_info.operators.base_operator import BaseOperator
 
 # TODO import BaseEstimator and EstimatorResult from terra once released
 from .qiskit.primitives import BaseEstimator, EstimatorResult
+from .qiskit.primitives.utils import _circuit_key
 from .qiskit_runtime_service import QiskitRuntimeService
 from .utils.estimator_result_decoder import EstimatorResultDecoder
+from .utils.json import RuntimeEncoder
+from .utils.utils import _hash
 from .runtime_job import RuntimeJob
 from .utils.deprecation import (
     deprecate_arguments,
@@ -189,6 +193,17 @@ class Estimator(BaseEstimator):
             backend = session or backend
             self._session = get_default_session(service, backend)
 
+        self._first_run = True
+        self._circuits_map = {}
+        if self.circuits:
+            for circuit in self.circuits:
+                circuit_id = _hash(
+                    json.dumps(_circuit_key(circuit), cls=RuntimeEncoder)
+                )
+                if circuit_id not in self._session._circuits_map:
+                    self._circuits_map[circuit_id] = circuit
+                    self._session._circuits_map[circuit_id] = circuit
+
     def run(  # pylint: disable=arguments-differ
         self,
         circuits: QuantumCircuit | Sequence[QuantumCircuit],
@@ -262,11 +277,25 @@ class Estimator(BaseEstimator):
             **kwargs: Individual options to overwrite the default primitive options.
 
         Returns:
-            Submitted job.
+            Submitted job
         """
+        circuits_map = {}
+        circuit_ids = []
+        for circuit in circuits:
+            circuit_id = _hash(json.dumps(_circuit_key(circuit), cls=RuntimeEncoder))
+            circuit_ids.append(circuit_id)
+            if circuit_id in self._session._circuits_map:
+                continue
+            self._session._circuits_map[circuit_id] = circuit
+            circuits_map[circuit_id] = circuit
+
+        if self._first_run:
+            self._first_run = False
+            circuits_map.update(self._circuits_map)
+
         inputs = {
-            "circuits": circuits,
-            "circuit_indices": list(range(len(circuits))),
+            "circuits": circuits_map,
+            "circuit_ids": circuit_ids,
             "observables": observables,
             "observable_indices": list(range(len(observables))),
             "parameters": [circ.parameters for circ in circuits],
diff --git a/qiskit_ibm_runtime/qiskit/primitives/utils.py b/qiskit_ibm_runtime/qiskit/primitives/utils.py
index dfad5b9a..8c939bcb 100644
--- a/qiskit_ibm_runtime/qiskit/primitives/utils.py
+++ b/qiskit_ibm_runtime/qiskit/primitives/utils.py
@@ -17,6 +17,8 @@ from __future__ import annotations
 
 from collections.abc import Iterator
 from typing import TypeVar
+from qiskit.circuit import QuantumCircuit
+from qiskit.circuit.bit import Bit
 
 
 T = TypeVar("T")  # pylint: disable=invalid-name
@@ -25,3 +27,49 @@ T = TypeVar("T")  # pylint: disable=invalid-name
 def _finditer(obj: T, objects: list[T]) -> Iterator[int]:
     """Return an iterator yielding the indices matching obj."""
     return map(lambda x: x[0], filter(lambda x: x[1] == obj, enumerate(objects)))
+
+
+def _bits_key(bits: tuple[Bit, ...], circuit: QuantumCircuit) -> tuple:
+    return tuple(
+        (
+            circuit.find_bit(bit).index,
+            tuple(
+                (reg[0].size, reg[0].name, reg[1])
+                for reg in circuit.find_bit(bit).registers
+            ),
+        )
+        for bit in bits
+    )
+
+
+def _circuit_key(circuit: QuantumCircuit, functional: bool = True) -> tuple:
+    """Private key function for QuantumCircuit.
+    This is the workaround until :meth:`QuantumCircuit.__hash__` will be introduced.
+    If key collision is found, please add elements to avoid it.
+    Args:
+        circuit: Input quantum circuit.
+        functional: If True, the returned key only includes functional data (i.e. execution related).
+    Returns:
+        Composite key for circuit.
+    """
+    functional_key: tuple = (
+        circuit.num_qubits,
+        circuit.num_clbits,
+        circuit.num_parameters,
+        tuple(  # circuit.data
+            (
+                _bits_key(data.qubits, circuit),  # qubits
+                _bits_key(data.clbits, circuit),  # clbits
+                data.operation.name,  # operation.name
+                tuple(data.operation.params),  # operation.params
+            )
+            for data in circuit.data
+        ),
+        None if circuit._op_start_times is None else tuple(circuit._op_start_times),
+    )
+    if functional:
+        return functional_key
+    return (
+        circuit.name,
+        *functional_key,
+    )
diff --git a/qiskit_ibm_runtime/sampler.py b/qiskit_ibm_runtime/sampler.py
index fdd3ec67..9425d80a 100644
--- a/qiskit_ibm_runtime/sampler.py
+++ b/qiskit_ibm_runtime/sampler.py
@@ -15,14 +15,18 @@
 from __future__ import annotations
 from typing import Dict, Iterable, Optional, Sequence, Any, Union, Callable
 import copy
+import json
 
 from qiskit.circuit import QuantumCircuit, Parameter
 
 # TODO import BaseSampler and SamplerResult from terra once released
 from .qiskit.primitives import BaseSampler, SamplerResult
+from .qiskit.primitives.utils import _circuit_key
 from .qiskit_runtime_service import QiskitRuntimeService
 from .options import Options
 from .utils.sampler_result_decoder import SamplerResultDecoder
+from .utils.json import RuntimeEncoder
+from .utils.utils import _hash
 from .runtime_job import RuntimeJob
 from .ibm_backend import IBMBackend
 from .session import get_default_session
@@ -157,6 +161,17 @@ class Sampler(BaseSampler):
             backend = session or backend
             self._session = get_default_session(service, backend)
 
+        self._first_run = True
+        self._circuits_map = {}
+        if self.circuits:
+            for circuit in self.circuits:
+                circuit_id = _hash(
+                    json.dumps(_circuit_key(circuit), cls=RuntimeEncoder)
+                )
+                if circuit_id not in self._session._circuits_map:
+                    self._circuits_map[circuit_id] = circuit
+                    self._session._circuits_map[circuit_id] = circuit
+
     def run(  # pylint: disable=arguments-differ
         self,
         circuits: QuantumCircuit | Sequence[QuantumCircuit],
@@ -219,10 +234,24 @@ class Sampler(BaseSampler):
         Returns:
             Submitted job.
         """
+        circuits_map = {}
+        circuit_ids = []
+        for circuit in circuits:
+            circuit_id = _hash(json.dumps(_circuit_key(circuit), cls=RuntimeEncoder))
+            circuit_ids.append(circuit_id)
+            if circuit_id in self._session._circuits_map:
+                continue
+            self._session._circuits_map[circuit_id] = circuit
+            circuits_map[circuit_id] = circuit
+
+        if self._first_run:
+            self._first_run = False
+            circuits_map.update(self._circuits_map)
+
         inputs = {
-            "circuits": circuits,
+            "circuits": circuits_map,
             "parameters": [circ.parameters for circ in circuits],
-            "circuit_indices": list(range(len(circuits))),
+            "circuit_ids": circuit_ids,
             "parameter_values": parameter_values,
         }
         combined = self.options._merge_options(kwargs)
diff --git a/qiskit_ibm_runtime/session.py b/qiskit_ibm_runtime/session.py
index 436ffab4..10f00a32 100644
--- a/qiskit_ibm_runtime/session.py
+++ b/qiskit_ibm_runtime/session.py
@@ -16,6 +16,8 @@ from typing import Dict, Optional, Type, Union, Callable
 from types import TracebackType
 from functools import wraps
 
+from qiskit.circuit import QuantumCircuit
+
 from qiskit_ibm_runtime import QiskitRuntimeService
 from .runtime_job import RuntimeJob
 from .runtime_program import ParameterNamespace
@@ -96,6 +98,8 @@ class Session:
         self._session_id: Optional[str] = None
         self._active = True
 
+        self._circuits_map: Dict[str, QuantumCircuit] = {}
+
         self._max_time = (
             max_time
             if max_time is None or isinstance(max_time, int)
diff --git a/qiskit_ibm_runtime/utils/utils.py b/qiskit_ibm_runtime/utils/utils.py
index 8a7dcc50..71aa3bb4 100644
--- a/qiskit_ibm_runtime/utils/utils.py
+++ b/qiskit_ibm_runtime/utils/utils.py
@@ -16,6 +16,7 @@ import keyword
 import logging
 import os
 import re
+import hashlib
 from queue import Queue
 from threading import Condition
 from typing import List, Optional, Any, Dict, Union, Tuple, Type
@@ -261,6 +262,13 @@ def _filter_value(
                 _filter_value(value, filter_keys)
 
 
+def _hash(hash_str: str) -> str:
+    """Hashes and returns a digest.
+    blake2s is supposedly faster than SHAs.
+    """
+    return hashlib.blake2s(hash_str.encode()).hexdigest()
+
+
 class RefreshQueue(Queue):
     """A queue that replaces the oldest item with the new item being added when full.
 


Test patch:
diff --git a/test/integration/test_estimator.py b/test/integration/test_estimator.py
index 8fad0dca..f04636d9 100644
--- a/test/integration/test_estimator.py
+++ b/test/integration/test_estimator.py
@@ -12,11 +12,13 @@
 
 """Integration tests for Estimator primitive."""
 
+from qiskit.circuit import QuantumCircuit
 from qiskit.circuit.library import RealAmplitudes
 from qiskit.quantum_info import SparsePauliOp
 from qiskit.test.reference_circuits import ReferenceCircuits
 
 from qiskit_ibm_runtime import Estimator, EstimatorResult, BaseEstimator, Session
+from qiskit_ibm_runtime.exceptions import RuntimeJobFailureError
 
 from ..decorators import run_integration_test
 from ..ibm_test_case import IBMIntegrationTestCase
@@ -104,6 +106,92 @@ class TestIntegrationEstimator(IBMIntegrationTestCase):
             self.assertEqual(len(result5.values), len(circuits5))
             self.assertEqual(len(result5.metadata), len(circuits5))
 
+    @run_integration_test
+    def test_estimator_session_circuit_caching(self, service):
+        """Verify if estimator primitive circuit caching works"""
+
+        backend = "ibmq_qasm_simulator"
+
+        qc1 = QuantumCircuit(2)
+        qc1.x(range(2))
+        qc1.measure_all()
+        qc2 = QuantumCircuit(2)
+        qc2.measure_all()
+        qc3 = QuantumCircuit(2)
+        qc3.h(range(2))
+        qc3.measure_all()
+
+        # pylint: disable=invalid-name
+        H = SparsePauliOp.from_list([("IZ", 1)])
+
+        with Session(service, backend) as session:
+            estimator = Estimator(session=session)
+
+            job = estimator.run(
+                circuits=[qc1, qc2],
+                observables=[H, H],
+            )
+            result = job.result()
+            self.assertEqual(len(result.values), 2)
+            self.assertEqual(len(result.metadata), 2)
+            self.assertEqual(result.values[0], -1)
+            self.assertEqual(result.values[1], 1)
+
+            job = estimator.run(
+                circuits=[qc3],
+                observables=[H],
+            )
+            result = job.result()
+            self.assertEqual(len(result.values), 1)
+            self.assertEqual(len(result.metadata), 1)
+            self.assertNotEqual(result.values[0], -1)
+            self.assertNotEqual(result.values[0], 1)
+
+            job = estimator.run(
+                circuits=[qc1, qc3],
+                observables=[H, H],
+            )
+            result = job.result()
+            self.assertEqual(len(result.values), 2)
+            self.assertEqual(len(result.metadata), 2)
+            self.assertEqual(result.values[0], -1)
+            self.assertNotEqual(result.values[1], -1)
+            self.assertNotEqual(result.values[1], 1)
+
+    @run_integration_test
+    def test_estimator_circuit_caching_with_transpilation_options(self, service):
+        """Verify if circuit caching works in estimator primitive
+        by passing correct and incorrect transpilation options."""
+
+        qc1 = QuantumCircuit(2)
+        qc1.x(range(2))
+        qc1.measure_all()
+
+        # pylint: disable=invalid-name
+        H = SparsePauliOp.from_list([("IZ", 1)])
+
+        with Session(service, self.backend) as session:
+            estimator = Estimator(session=session)
+            # pass correct initial_layout
+            transpilation = {"initial_layout": [0, 1]}
+            job = estimator.run(
+                circuits=[qc1], observables=[H], transpilation=transpilation
+            )
+            result = job.result()
+            self.assertEqual(len(result.values), 1)
+            self.assertEqual(len(result.metadata), 1)
+            self.assertEqual(result.values[0], -1)
+
+            # pass incorrect initial_layout
+            # since a new transpilation option is passed it should not use the
+            # cached transpiled circuit from the first run above
+            transpilation = {"initial_layout": [0]}
+            job = estimator.run(
+                circuits=[qc1], observables=[H], transpilation=transpilation
+            )
+            with self.assertRaises(RuntimeJobFailureError):
+                job.result()
+
     @run_integration_test
     def test_estimator_primitive(self, service):
         """Test to verify that estimator as a primitive still works."""
diff --git a/test/integration/test_sampler.py b/test/integration/test_sampler.py
index 248d97aa..c3e10ebb 100644
--- a/test/integration/test_sampler.py
+++ b/test/integration/test_sampler.py
@@ -46,8 +46,8 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
             self.assertIsInstance(result, SamplerResult)
             self.assertEqual(len(result.quasi_dists), 1)
             self.assertEqual(len(result.metadata), 1)
-            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.05)
-            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.05)
+            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.1)
+            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.1)
             self.assertTrue(session.session_id)
 
     @run_integration_test
@@ -65,8 +65,8 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
             self.assertEqual(len(result1.quasi_dists), len(circuits1))
             self.assertEqual(len(result1.metadata), len(circuits1))
             for i in range(len(circuits1)):
-                self.assertAlmostEqual(result1.quasi_dists[i][3], 0.5, delta=0.05)
-                self.assertAlmostEqual(result1.quasi_dists[i][0], 0.5, delta=0.05)
+                self.assertAlmostEqual(result1.quasi_dists[i][3], 0.5, delta=0.1)
+                self.assertAlmostEqual(result1.quasi_dists[i][0], 0.5, delta=0.1)
 
             circuits2 = [circuits[0], circuits[2]]
             result2 = sampler.run(circuits=circuits2).result()
@@ -74,8 +74,8 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
             self.assertEqual(len(result2.quasi_dists), len(circuits2))
             self.assertEqual(len(result2.metadata), len(circuits2))
             for i in range(len(circuits2)):
-                self.assertAlmostEqual(result2.quasi_dists[i][3], 0.5, delta=0.05)
-                self.assertAlmostEqual(result2.quasi_dists[i][0], 0.5, delta=0.05)
+                self.assertAlmostEqual(result2.quasi_dists[i][3], 0.5, delta=0.1)
+                self.assertAlmostEqual(result2.quasi_dists[i][0], 0.5, delta=0.1)
 
             circuits3 = [circuits[1], circuits[2]]
             result3 = sampler.run(circuits=circuits3).result()
@@ -83,8 +83,84 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
             self.assertEqual(len(result3.quasi_dists), len(circuits3))
             self.assertEqual(len(result3.metadata), len(circuits3))
             for i in range(len(circuits3)):
-                self.assertAlmostEqual(result3.quasi_dists[i][3], 0.5, delta=0.05)
-                self.assertAlmostEqual(result3.quasi_dists[i][0], 0.5, delta=0.05)
+                self.assertAlmostEqual(result3.quasi_dists[i][3], 0.5, delta=0.1)
+                self.assertAlmostEqual(result3.quasi_dists[i][0], 0.5, delta=0.1)
+
+    @run_integration_test
+    def test_sampler_non_parameterized_circuit_caching(self, service):
+        """Verify if circuit caching works in sampler primitive
+        and returns expected results for non-parameterized circuits."""
+
+        qc1 = QuantumCircuit(2)
+        qc1.x(range(2))
+        qc1.measure_all()
+        qc2 = QuantumCircuit(3)
+        qc2.x(range(3))
+        qc2.measure_all()
+        qc3 = QuantumCircuit(4)
+        qc3.x(range(4))
+        qc3.measure_all()
+        qc4 = QuantumCircuit(5)
+        qc4.x(range(5))
+        qc4.measure_all()
+
+        with Session(service, self.backend) as session:
+            sampler = Sampler(session=session)
+            self.assertIsInstance(sampler, BaseSampler)
+            job = sampler.run(circuits=[qc1, qc2])
+            result = job.result()
+            self.assertEqual(len(result.quasi_dists), 2)
+            self.assertEqual(len(result.metadata), 2)
+            self.assertEqual(result.quasi_dists[0][3], 1)
+            self.assertEqual(result.quasi_dists[1][7], 1)
+
+            job = sampler.run(circuits=[qc1])
+            result = job.result()
+            self.assertEqual(len(result.quasi_dists), 1)
+            self.assertEqual(len(result.metadata), 1)
+            self.assertEqual(result.quasi_dists[0][3], 1)
+
+            job = sampler.run(circuits=[qc3])
+            result = job.result()
+            self.assertEqual(len(result.quasi_dists), 1)
+            self.assertEqual(len(result.metadata), 1)
+            self.assertEqual(result.quasi_dists[0][15], 1)
+
+            job = sampler.run(circuits=[qc1, qc4])
+            result = job.result()
+            self.assertEqual(len(result.quasi_dists), 2)
+            self.assertEqual(len(result.metadata), 2)
+            self.assertEqual(result.quasi_dists[0][3], 1)
+            self.assertEqual(result.quasi_dists[1][31], 1)
+
+    @run_integration_test
+    def test_sampler_non_parameterized_circuit_caching_with_transpilation_options(
+        self, service
+    ):
+        """Verify if circuit caching works in sampler primitive
+        by passing correct and incorrect transpilation options."""
+
+        qc1 = QuantumCircuit(2)
+        qc1.x(range(2))
+        qc1.measure_all()
+
+        with Session(service, self.backend) as session:
+            sampler = Sampler(session=session)
+            # pass correct initial_layout
+            transpilation = {"initial_layout": [0, 1]}
+            job = sampler.run(circuits=[qc1], transpilation=transpilation)
+            result = job.result()
+            self.assertEqual(len(result.quasi_dists), 1)
+            self.assertEqual(len(result.metadata), 1)
+            self.assertEqual(result.quasi_dists[0][3], 1)
+
+            # pass incorrect initial_layout
+            # since a new transpilation option is passed it should not use the
+            # cached transpiled circuit from the first run above
+            transpilation = {"initial_layout": [0]}
+            job = sampler.run(circuits=[qc1], transpilation=transpilation)
+            with self.assertRaises(RuntimeJobFailureError):
+                job.result()
 
     @run_integration_test
     def test_sampler_primitive_parameterized_circuits(self, service):
@@ -139,10 +215,10 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
             result = sampler.run(self.bell, shots=shots).result()
             self.assertEqual(result.quasi_dists[0].shots, shots)
             self.assertAlmostEqual(
-                result.quasi_dists[0]._stddev_upper_bound, sqrt(1 / shots), delta=0.05
+                result.quasi_dists[0]._stddev_upper_bound, sqrt(1 / shots), delta=0.1
             )
-            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.05)
-            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.05)
+            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.1)
+            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.1)
 
     @run_integration_test
     def test_sampler_primitive_as_session(self, service):
diff --git a/test/integration/test_session.py b/test/integration/test_session.py
index 469a6ade..b7b4219e 100644
--- a/test/integration/test_session.py
+++ b/test/integration/test_session.py
@@ -56,8 +56,8 @@ class TestIntegrationSession(IBMIntegrationTestCase):
             self.assertEqual(len(result.quasi_dists), 1)
             self.assertEqual(len(result.metadata), 1)
             self.assertEqual(result.metadata[0]["shots"], 200)
-            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.05)
-            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.05)
+            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.1)
+            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.1)
 
             result = estimator.run(
                 circuits=[psi1], observables=[H1], parameter_values=[theta1], shots=300
@@ -72,5 +72,5 @@ class TestIntegrationSession(IBMIntegrationTestCase):
             self.assertEqual(len(result.quasi_dists), 1)
             self.assertEqual(len(result.metadata), 1)
             self.assertEqual(result.metadata[0]["shots"], 400)
-            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.05)
-            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.05)
+            self.assertAlmostEqual(result.quasi_dists[0][3], 0.5, delta=0.1)
+            self.assertAlmostEqual(result.quasi_dists[0][0], 0.5, delta=0.1)
diff --git a/test/unit/test_estimator.py b/test/unit/test_estimator.py
new file mode 100644
index 00000000..a277fe09
--- /dev/null
+++ b/test/unit/test_estimator.py
@@ -0,0 +1,83 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# 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.
+
+"""Tests for estimator class."""
+
+import json
+from unittest.mock import patch, ANY
+
+from qiskit.circuit.library import RealAmplitudes
+from qiskit.quantum_info import SparsePauliOp
+
+from qiskit_ibm_runtime.utils.json import RuntimeEncoder
+from qiskit_ibm_runtime.utils.utils import _hash
+from qiskit_ibm_runtime.qiskit.primitives.utils import _circuit_key
+
+from qiskit_ibm_runtime import Estimator, Session
+from ..ibm_test_case import IBMTestCase
+
+from .mock.fake_runtime_service import FakeRuntimeService
+
+
+class TestEstimator(IBMTestCase):
+    """Class for testing the Estimator class."""
+
+    def test_estimator_circuit_caching(self):
+        """Test circuit caching in Estimator class"""
+        psi1 = RealAmplitudes(num_qubits=2, reps=2)
+        psi2 = RealAmplitudes(num_qubits=2, reps=3)
+        psi3 = RealAmplitudes(num_qubits=2, reps=2)
+        psi4 = RealAmplitudes(num_qubits=2, reps=3)
+        psi1_id = _hash(json.dumps(_circuit_key(psi1), cls=RuntimeEncoder))
+        psi2_id = _hash(json.dumps(_circuit_key(psi2), cls=RuntimeEncoder))
+        psi3_id = _hash(json.dumps(_circuit_key(psi3), cls=RuntimeEncoder))
+        psi4_id = _hash(json.dumps(_circuit_key(psi4), cls=RuntimeEncoder))
+
+        # pylint: disable=invalid-name
+        H1 = SparsePauliOp.from_list([("II", 1), ("IZ", 2), ("XI", 3)])
+        H2 = SparsePauliOp.from_list([("IZ", 1)])
+
+        with Session(
+            service=FakeRuntimeService(channel="ibm_quantum", token="abc"),
+            backend="ibmq_qasm_simulator",
+        ) as session:
+            estimator = Estimator(session=session)
+
+            # calculate [ <psi1(theta1)|H1|psi1(theta1)> ]
+            with patch.object(estimator._session, "run") as mock_run:
+                estimator.run([psi1, psi2], [H1, H2], [[ANY] * 6, [ANY] * 8])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {psi1_id: psi1, psi2_id: psi2})
+                self.assertEqual(inputs["circuit_ids"], [psi1_id, psi2_id])
+
+            # calculate [ <psi2(theta2)|H2|psi2(theta2)> ]
+            with patch.object(estimator._session, "run") as mock_run:
+                estimator.run([psi2], [H1], [[ANY] * 8])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {})
+                self.assertEqual(inputs["circuit_ids"], [psi2_id])
+
+            with patch.object(estimator._session, "run") as mock_run:
+                estimator.run([psi3], [H1], [[ANY] * 6])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {psi3_id: psi3})
+                self.assertEqual(inputs["circuit_ids"], [psi3_id])
+
+            with patch.object(estimator._session, "run") as mock_run:
+                estimator.run([psi4, psi1], [H2, H1], [[ANY] * 8, [ANY] * 6])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {psi4_id: psi4})
+                self.assertEqual(inputs["circuit_ids"], [psi4_id, psi1_id])
diff --git a/test/unit/test_ibm_primitives.py b/test/unit/test_ibm_primitives.py
index 05899476..28acb659 100644
--- a/test/unit/test_ibm_primitives.py
+++ b/test/unit/test_ibm_primitives.py
@@ -14,21 +14,36 @@
 
 import sys
 import copy
+import json
 from unittest.mock import MagicMock, patch, ANY
 import warnings
 from dataclasses import asdict
+from typing import Dict
 
+from qiskit.circuit import QuantumCircuit
+from qiskit.circuit.library import RealAmplitudes
 from qiskit.test.reference_circuits import ReferenceCircuits
 from qiskit.quantum_info import SparsePauliOp
-from qiskit_ibm_runtime import Sampler, Estimator, Options, Session
+
+from qiskit_ibm_runtime import Sampler, Estimator, Options, Session, RuntimeEncoder
 from qiskit_ibm_runtime.ibm_backend import IBMBackend
 import qiskit_ibm_runtime.session as session_pkg
+from qiskit_ibm_runtime.utils.utils import _hash
+from qiskit_ibm_runtime.qiskit.primitives.utils import _circuit_key
 from ..ibm_test_case import IBMTestCase
 from ..utils import dict_paritally_equal
 
+from .mock.fake_runtime_service import FakeRuntimeService
+
+
+class MockSession(Session):
+    """Mock for session class"""
+
+    _circuits_map: Dict[str, QuantumCircuit] = {}
+
 
 class TestPrimitives(IBMTestCase):
-    """Class for testing the Sampler class."""
+    """Class for testing the Sampler and Estimator classes."""
 
     @classmethod
     def setUpClass(cls):
@@ -45,7 +60,7 @@ class TestPrimitives(IBMTestCase):
         primitives = [Sampler, Estimator]
         for cls in primitives:
             with self.subTest(primitive=cls):
-                inst = cls(session=MagicMock(spec=Session), skip_transpilation=True)
+                inst = cls(session=MagicMock(spec=MockSession), skip_transpilation=True)
                 self.assertTrue(inst.options.transpilation.skip_transpilation)
 
     def test_skip_transpilation_overwrite(self):
@@ -56,7 +71,7 @@ class TestPrimitives(IBMTestCase):
         for cls in primitives:
             with self.subTest(primitive=cls):
                 inst = cls(
-                    session=MagicMock(spec=Session),
+                    session=MagicMock(spec=MockSession),
                     options=options,
                     skip_transpilation=True,
                 )
@@ -77,7 +92,7 @@ class TestPrimitives(IBMTestCase):
         for cls in primitives:
             for options in options_vars:
                 with self.subTest(primitive=cls, options=options):
-                    inst = cls(session=MagicMock(spec=Session), options=options)
+                    inst = cls(session=MagicMock(spec=MockSession), options=options)
                     expected = asdict(Options())
                     self._update_dict(expected, copy.deepcopy(options))
                     self.assertDictEqual(expected, asdict(inst.options))
@@ -109,7 +124,7 @@ class TestPrimitives(IBMTestCase):
 
     def test_runtime_options(self):
         """Test RuntimeOptions specified as primitive options."""
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         primitives = [Sampler, Estimator]
         env_vars = [
             {"log_level": "DEBUG"},
@@ -138,7 +153,7 @@ class TestPrimitives(IBMTestCase):
         for cls in primitives:
             with self.subTest(primitive=cls):
                 options.transpilation.skip_transpilation = True
-                inst = cls(session=MagicMock(spec=Session), options=options)
+                inst = cls(session=MagicMock(spec=MockSession), options=options)
                 options.transpilation.skip_transpilation = False
                 self.assertTrue(inst.options.transpilation.skip_transpilation)
 
@@ -225,7 +240,7 @@ class TestPrimitives(IBMTestCase):
 
     def test_run_default_options(self):
         """Test run using default options."""
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         options_vars = [
             (Options(resilience_level=9), {"resilience_settings": {"level": 9}}),
             (
@@ -256,7 +271,7 @@ class TestPrimitives(IBMTestCase):
 
     def test_run_updated_default_options(self):
         """Test run using updated default options."""
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         primitives = [Sampler, Estimator]
         for cls in primitives:
             with self.subTest(primitive=cls):
@@ -283,7 +298,7 @@ class TestPrimitives(IBMTestCase):
 
     def test_run_overwrite_options(self):
         """Test run using overwritten options."""
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         options_vars = [
             ({"resilience_level": 9}, {"resilience_settings": {"level": 9}}),
             ({"shots": 200}, {"run_options": {"shots": 200}}),
@@ -317,7 +332,7 @@ class TestPrimitives(IBMTestCase):
 
     def test_kwarg_options(self):
         """Test specifying arbitrary options."""
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         primitives = [Sampler, Estimator]
         for cls in primitives:
             with self.subTest(primitive=cls):
@@ -333,7 +348,7 @@ class TestPrimitives(IBMTestCase):
 
     def test_run_kwarg_options(self):
         """Test specifying arbitrary options in run."""
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         primitives = [Sampler, Estimator]
         for cls in primitives:
             with self.subTest(primitive=cls):
@@ -348,7 +363,7 @@ class TestPrimitives(IBMTestCase):
 
     def test_run_multiple_different_options(self):
         """Test multiple runs with different options."""
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         primitives = [Sampler, Estimator]
         for cls in primitives:
             with self.subTest(primitive=cls):
@@ -366,13 +381,68 @@ class TestPrimitives(IBMTestCase):
         """Test multiple runs within a session."""
         num_runs = 5
         primitives = [Sampler, Estimator]
-        session = MagicMock(spec=Session)
+        session = MagicMock(spec=MockSession)
         for idx in range(num_runs):
             cls = primitives[idx % 2]
             inst = cls(session=session)
             inst.run(self.qx, observables=self.obs)
         self.assertEqual(session.run.call_count, num_runs)
 
+    def test_primitives_circuit_caching(self):
+        """Test circuit caching in Estimator and Sampler classes"""
+        psi1 = RealAmplitudes(num_qubits=2, reps=2)
+        psi1.measure_all()
+        psi2 = RealAmplitudes(num_qubits=2, reps=3)
+        psi2.measure_all()
+        psi3 = RealAmplitudes(num_qubits=2, reps=2)
+        psi3.measure_all()
+        psi4 = RealAmplitudes(num_qubits=2, reps=3)
+        psi4.measure_all()
+        psi1_id = _hash(json.dumps(_circuit_key(psi1), cls=RuntimeEncoder))
+        psi2_id = _hash(json.dumps(_circuit_key(psi2), cls=RuntimeEncoder))
+        psi3_id = _hash(json.dumps(_circuit_key(psi3), cls=RuntimeEncoder))
+        psi4_id = _hash(json.dumps(_circuit_key(psi4), cls=RuntimeEncoder))
+
+        # pylint: disable=invalid-name
+        H1 = SparsePauliOp.from_list([("II", 1), ("IZ", 2), ("XI", 3)])
+        H2 = SparsePauliOp.from_list([("IZ", 1)])
+
+        with Session(
+            service=FakeRuntimeService(channel="ibm_quantum", token="abc"),
+            backend="ibmq_qasm_simulator",
+        ) as session:
+            estimator = Estimator(session=session)
+
+            # calculate [ <psi1(theta1)|H1|psi1(theta1)> ]
+            with patch.object(estimator._session, "run") as mock_run:
+                estimator.run([psi1, psi2], [H1, H2], [[ANY] * 6, [ANY] * 8])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {psi1_id: psi1, psi2_id: psi2})
+                self.assertEqual(inputs["circuit_ids"], [psi1_id, psi2_id])
+
+            sampler = Sampler(session=session)
+            with patch.object(sampler._session, "run") as mock_run:
+                sampler.run([psi1, psi2], [[ANY] * 6, [ANY] * 8])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {})
+                self.assertEqual(inputs["circuit_ids"], [psi1_id, psi2_id])
+
+            with patch.object(estimator._session, "run") as mock_run:
+                estimator.run([psi3], [H1], [[ANY] * 6])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {psi3_id: psi3})
+                self.assertEqual(inputs["circuit_ids"], [psi3_id])
+
+            with patch.object(sampler._session, "run") as mock_run:
+                sampler.run([psi4, psi1], [[ANY] * 8, [ANY] * 6])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {psi4_id: psi4})
+                self.assertEqual(inputs["circuit_ids"], [psi4_id, psi1_id])
+
     def _update_dict(self, dict1, dict2):
         for key, val in dict1.items():
             if isinstance(val, dict):
diff --git a/test/unit/test_sampler.py b/test/unit/test_sampler.py
new file mode 100644
index 00000000..e9a757e3
--- /dev/null
+++ b/test/unit/test_sampler.py
@@ -0,0 +1,80 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# 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.
+
+"""Tests for sampler class."""
+
+import json
+from unittest.mock import patch, ANY
+
+from qiskit.circuit.library import RealAmplitudes
+
+from qiskit_ibm_runtime.utils.json import RuntimeEncoder
+from qiskit_ibm_runtime.utils.utils import _hash
+from qiskit_ibm_runtime.qiskit.primitives.utils import _circuit_key
+
+from qiskit_ibm_runtime import Sampler, Session
+from ..ibm_test_case import IBMTestCase
+
+from .mock.fake_runtime_service import FakeRuntimeService
+
+
+class TestSampler(IBMTestCase):
+    """Class for testing the Sampler class."""
+
+    def test_sampler_circuit_caching(self):
+        """Test circuit caching in Sampler class"""
+
+        pqc = RealAmplitudes(num_qubits=2, reps=2)
+        pqc.measure_all()
+        pqc2 = RealAmplitudes(num_qubits=2, reps=3)
+        pqc2.measure_all()
+        pqc3 = RealAmplitudes(num_qubits=2, reps=2)
+        pqc3.measure_all()
+        pqc4 = RealAmplitudes(num_qubits=2, reps=3)
+        pqc4.measure_all()
+        pqc_id = _hash(json.dumps(_circuit_key(pqc), cls=RuntimeEncoder))
+        pqc2_id = _hash(json.dumps(_circuit_key(pqc2), cls=RuntimeEncoder))
+        pqc3_id = _hash(json.dumps(_circuit_key(pqc3), cls=RuntimeEncoder))
+        pqc4_id = _hash(json.dumps(_circuit_key(pqc4), cls=RuntimeEncoder))
+
+        with Session(
+            service=FakeRuntimeService(channel="ibm_quantum", token="abc"),
+            backend="ibmq_qasm_simulator",
+        ) as session:
+            sampler = Sampler(session=session)
+            with patch.object(sampler._session, "run") as mock_run:
+                sampler.run([pqc, pqc2], [[ANY] * 6, [ANY] * 8])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {pqc_id: pqc, pqc2_id: pqc2})
+                self.assertEqual(inputs["circuit_ids"], [pqc_id, pqc2_id])
+
+            with patch.object(sampler._session, "run") as mock_run:
+                sampler.run([pqc2], [[ANY] * 8])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {})
+                self.assertEqual(inputs["circuit_ids"], [pqc2_id])
+
+            with patch.object(sampler._session, "run") as mock_run:
+                sampler.run([pqc3], [[ANY] * 6])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {pqc3_id: pqc3})
+                self.assertEqual(inputs["circuit_ids"], [pqc3_id])
+
+            with patch.object(sampler._session, "run") as mock_run:
+                sampler.run([pqc4, pqc], [[ANY] * 8, [ANY] * 6])
+                _, kwargs = mock_run.call_args
+                inputs = kwargs["inputs"]
+                self.assertDictEqual(inputs["circuits"], {pqc4_id: pqc4})
+                self.assertEqual(inputs["circuit_ids"], [pqc4_id, pqc_id])