thiagotnunes: @larkee thanks for the fix, I will test it out with a long running PDML to make sure everything we need is in place!
thiagotnunes: @larkee Quick update, the PDML long running task finished successfully in production testing. Thanks for fixing this!
google-cla[bot]: We found a Contributor License Agreement for you (the sender of this pull request), but were unable to find agreements for all the commit author(s) or Co-authors.  If you authored these, maybe you used a different email address in the git commits than was used to sign the CLA ([login here](https://cla.developers.google.com/) to double check)?  If these were authored by someone else, then they will need to sign a CLA as well, and confirm that they're okay with these being contributed to Google.
In order to pass this check, please resolve this problem and then comment `@googlebot I fixed it.`. If the bot doesn't comment, it means it doesn't think anything has changed.

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fgoogleapis%2Fpython-spanner%2Fpull%2F122) for more info**.

<!-- need_author_cla -->
diff --git a/google/cloud/spanner_v1/snapshot.py b/google/cloud/spanner_v1/snapshot.py
index 0b5ee1d..42e7154 100644
--- a/google/cloud/spanner_v1/snapshot.py
+++ b/google/cloud/spanner_v1/snapshot.py
@@ -20,6 +20,7 @@ from google.protobuf.struct_pb2 import Struct
 from google.cloud.spanner_v1.proto.transaction_pb2 import TransactionOptions
 from google.cloud.spanner_v1.proto.transaction_pb2 import TransactionSelector
 
+from google.api_core.exceptions import InternalServerError
 from google.api_core.exceptions import ServiceUnavailable
 import google.api_core.gapic_v1.method
 from google.cloud._helpers import _datetime_to_pb_timestamp
@@ -32,6 +33,11 @@ from google.cloud.spanner_v1.streamed import StreamedResultSet
 from google.cloud.spanner_v1.types import PartitionOptions
 from google.cloud.spanner_v1._opentelemetry_tracing import trace_call
 
+_STREAM_RESUMPTION_INTERNAL_ERROR_MESSAGES = (
+    "RST_STREAM",
+    "Received unexpected EOS on DATA frame from server",
+)
+
 
 def _restart_on_unavailable(restart, trace_name=None, session=None, attributes=None):
     """Restart iteration after :exc:`.ServiceUnavailable`.
@@ -55,6 +61,17 @@ def _restart_on_unavailable(restart, trace_name=None, session=None, attributes=N
             with trace_call(trace_name, session, attributes):
                 iterator = restart(resume_token=resume_token)
             continue
+        except InternalServerError as exc:
+            resumable_error = any(
+                resumable_message in exc.message
+                for resumable_message in _STREAM_RESUMPTION_INTERNAL_ERROR_MESSAGES
+            )
+            if not resumable_error:
+                raise
+            del item_buffer[:]
+            with trace_call(trace_name, session, attributes):
+                iterator = restart(resume_token=resume_token)
+            continue
 
         if len(item_buffer) == 0:
             break

spanner-client: Retry PDML on "Received unexpected EOS on DATA frame from server" 
This bug is related to the Spanner client library.

For long lived transactions (>= 30 minutes), in the case of large PDML changes, it is possible that the gRPC connection is terminated with an error "Received unexpected EOS on DATA frame from server".

In this case, we need to retry the transaction either with the received resume token obtained on reading the stream or from scratch. This will ensure that the PDML transaction continues to execute until it is successful or a hard timeout is reached.

We have already implemented such change in the Java client library, for more information see this PR: https://github.com/googleapis/java-spanner/pull/360.

In order to test the fix, we can use a large spanner database. Please speak to @thiagotnunes for more details.
diff --git a/tests/unit/test_snapshot.py b/tests/unit/test_snapshot.py
index 5c53ee6..8589a0c 100644
--- a/tests/unit/test_snapshot.py
+++ b/tests/unit/test_snapshot.py
@@ -86,12 +86,35 @@ class Test_restart_on_unavailable(OpenTelemetryBase):
         self.assertNoSpans()
 
     def test_iteration_w_raw_raising_unavailable_no_token(self):
+        from google.api_core.exceptions import ServiceUnavailable
+
+        ITEMS = (
+            self._make_item(0),
+            self._make_item(1, resume_token=RESUME_TOKEN),
+            self._make_item(2),
+        )
+        before = _MockIterator(fail_after=True, error=ServiceUnavailable("testing"))
+        after = _MockIterator(*ITEMS)
+        restart = mock.Mock(spec=[], side_effect=[before, after])
+        resumable = self._call_fut(restart)
+        self.assertEqual(list(resumable), list(ITEMS))
+        self.assertEqual(restart.mock_calls, [mock.call(), mock.call(resume_token=b"")])
+        self.assertNoSpans()
+
+    def test_iteration_w_raw_raising_retryable_internal_error_no_token(self):
+        from google.api_core.exceptions import InternalServerError
+
         ITEMS = (
             self._make_item(0),
             self._make_item(1, resume_token=RESUME_TOKEN),
             self._make_item(2),
         )
-        before = _MockIterator(fail_after=True)
+        before = _MockIterator(
+            fail_after=True,
+            error=InternalServerError(
+                "Received unexpected EOS on DATA frame from server"
+            ),
+        )
         after = _MockIterator(*ITEMS)
         restart = mock.Mock(spec=[], side_effect=[before, after])
         resumable = self._call_fut(restart)
@@ -99,11 +122,32 @@ class Test_restart_on_unavailable(OpenTelemetryBase):
         self.assertEqual(restart.mock_calls, [mock.call(), mock.call(resume_token=b"")])
         self.assertNoSpans()
 
+    def test_iteration_w_raw_raising_non_retryable_internal_error_no_token(self):
+        from google.api_core.exceptions import InternalServerError
+
+        ITEMS = (
+            self._make_item(0),
+            self._make_item(1, resume_token=RESUME_TOKEN),
+            self._make_item(2),
+        )
+        before = _MockIterator(fail_after=True, error=InternalServerError("testing"))
+        after = _MockIterator(*ITEMS)
+        restart = mock.Mock(spec=[], side_effect=[before, after])
+        resumable = self._call_fut(restart)
+        with self.assertRaises(InternalServerError):
+            list(resumable)
+        self.assertEqual(restart.mock_calls, [mock.call()])
+        self.assertNoSpans()
+
     def test_iteration_w_raw_raising_unavailable(self):
+        from google.api_core.exceptions import ServiceUnavailable
+
         FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
         SECOND = (self._make_item(2),)  # discarded after 503
         LAST = (self._make_item(3),)
-        before = _MockIterator(*(FIRST + SECOND), fail_after=True)
+        before = _MockIterator(
+            *(FIRST + SECOND), fail_after=True, error=ServiceUnavailable("testing")
+        )
         after = _MockIterator(*LAST)
         restart = mock.Mock(spec=[], side_effect=[before, after])
         resumable = self._call_fut(restart)
@@ -113,10 +157,53 @@ class Test_restart_on_unavailable(OpenTelemetryBase):
         )
         self.assertNoSpans()
 
+    def test_iteration_w_raw_raising_retryable_internal_error(self):
+        from google.api_core.exceptions import InternalServerError
+
+        FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
+        SECOND = (self._make_item(2),)  # discarded after 503
+        LAST = (self._make_item(3),)
+        before = _MockIterator(
+            *(FIRST + SECOND),
+            fail_after=True,
+            error=InternalServerError(
+                "Received unexpected EOS on DATA frame from server"
+            )
+        )
+        after = _MockIterator(*LAST)
+        restart = mock.Mock(spec=[], side_effect=[before, after])
+        resumable = self._call_fut(restart)
+        self.assertEqual(list(resumable), list(FIRST + LAST))
+        self.assertEqual(
+            restart.mock_calls, [mock.call(), mock.call(resume_token=RESUME_TOKEN)]
+        )
+        self.assertNoSpans()
+
+    def test_iteration_w_raw_raising_non_retryable_internal_error(self):
+        from google.api_core.exceptions import InternalServerError
+
+        FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
+        SECOND = (self._make_item(2),)  # discarded after 503
+        LAST = (self._make_item(3),)
+        before = _MockIterator(
+            *(FIRST + SECOND), fail_after=True, error=InternalServerError("testing")
+        )
+        after = _MockIterator(*LAST)
+        restart = mock.Mock(spec=[], side_effect=[before, after])
+        resumable = self._call_fut(restart)
+        with self.assertRaises(InternalServerError):
+            list(resumable)
+        self.assertEqual(restart.mock_calls, [mock.call()])
+        self.assertNoSpans()
+
     def test_iteration_w_raw_raising_unavailable_after_token(self):
+        from google.api_core.exceptions import ServiceUnavailable
+
         FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
         SECOND = (self._make_item(2), self._make_item(3))
-        before = _MockIterator(*FIRST, fail_after=True)
+        before = _MockIterator(
+            *FIRST, fail_after=True, error=ServiceUnavailable("testing")
+        )
         after = _MockIterator(*SECOND)
         restart = mock.Mock(spec=[], side_effect=[before, after])
         resumable = self._call_fut(restart)
@@ -126,6 +213,43 @@ class Test_restart_on_unavailable(OpenTelemetryBase):
         )
         self.assertNoSpans()
 
+    def test_iteration_w_raw_raising_retryable_internal_error_after_token(self):
+        from google.api_core.exceptions import InternalServerError
+
+        FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
+        SECOND = (self._make_item(2), self._make_item(3))
+        before = _MockIterator(
+            *FIRST,
+            fail_after=True,
+            error=InternalServerError(
+                "Received unexpected EOS on DATA frame from server"
+            )
+        )
+        after = _MockIterator(*SECOND)
+        restart = mock.Mock(spec=[], side_effect=[before, after])
+        resumable = self._call_fut(restart)
+        self.assertEqual(list(resumable), list(FIRST + SECOND))
+        self.assertEqual(
+            restart.mock_calls, [mock.call(), mock.call(resume_token=RESUME_TOKEN)]
+        )
+        self.assertNoSpans()
+
+    def test_iteration_w_raw_raising_non_retryable_internal_error_after_token(self):
+        from google.api_core.exceptions import InternalServerError
+
+        FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
+        SECOND = (self._make_item(2), self._make_item(3))
+        before = _MockIterator(
+            *FIRST, fail_after=True, error=InternalServerError("testing")
+        )
+        after = _MockIterator(*SECOND)
+        restart = mock.Mock(spec=[], side_effect=[before, after])
+        resumable = self._call_fut(restart)
+        with self.assertRaises(InternalServerError):
+            list(resumable)
+        self.assertEqual(restart.mock_calls, [mock.call()])
+        self.assertNoSpans()
+
     def test_iteration_w_span_creation(self):
         name = "TestSpan"
         extra_atts = {"test_att": 1}
@@ -136,11 +260,15 @@ class Test_restart_on_unavailable(OpenTelemetryBase):
         self.assertSpanAttributes(name, attributes=dict(BASE_ATTRIBUTES, test_att=1))
 
     def test_iteration_w_multiple_span_creation(self):
+        from google.api_core.exceptions import ServiceUnavailable
+
         if HAS_OPENTELEMETRY_INSTALLED:
             FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
             SECOND = (self._make_item(2),)  # discarded after 503
             LAST = (self._make_item(3),)
-            before = _MockIterator(*(FIRST + SECOND), fail_after=True)
+            before = _MockIterator(
+                *(FIRST + SECOND), fail_after=True, error=ServiceUnavailable("testing")
+            )
             after = _MockIterator(*LAST)
             restart = mock.Mock(spec=[], side_effect=[before, after])
             name = "TestSpan"
@@ -1153,18 +1281,17 @@ class _MockIterator(object):
     def __init__(self, *values, **kw):
         self._iter_values = iter(values)
         self._fail_after = kw.pop("fail_after", False)
+        self._error = kw.pop("error", Exception)
 
     def __iter__(self):
         return self
 
     def __next__(self):
-        from google.api_core.exceptions import ServiceUnavailable
-
         try:
             return next(self._iter_values)
         except StopIteration:
             if self._fail_after:
-                raise ServiceUnavailable("testing")
+                raise self._error
             raise
 
     next = __next__

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pytest-cov pytest-xdist pytest-mock pytest-asyncio mock
apt-get update apt-get install -y gcc
3.9
requirements/base.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
cachetools==4.2.4
certifi==2025.1.31
charset-normalizer==3.4.1
coverage==7.8.0
exceptiongroup==1.2.2
execnet==2.1.1
google-api-core==1.34.1
google-auth==1.35.0
google-cloud-core==1.7.3
-e git+https://github.com/googleapis/python-spanner.git@808837b5afb34ba7d745b83e53274b5709a9ef63#egg=google_cloud_spanner
googleapis-common-protos==1.69.2
grpc-google-iam-v1==0.12.7
grpcio==1.71.0
grpcio-gcp==0.2.2
grpcio-status==1.49.0rc1
idna==3.10
iniconfig==2.1.0
mock==5.2.0
packaging==24.2
pluggy==1.5.0
protobuf==3.20.3
pyasn1==0.6.1
pyasn1_modules==0.4.2
pytest==8.3.5
pytest-asyncio==0.26.0
pytest-cov==6.0.0
pytest-mock==3.14.0
pytest-xdist==3.6.1
requests==2.32.3
rsa==4.9
six==1.17.0
tomli==2.2.1
typing_extensions==4.13.0
urllib3==2.3.0

name: python-spanner
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - cachetools==4.2.4
      - certifi==2025.1.31
      - charset-normalizer==3.4.1
      - coverage==7.8.0
      - exceptiongroup==1.2.2
      - execnet==2.1.1
      - google-api-core==1.34.1
      - google-auth==1.35.0
      - google-cloud-core==1.7.3
      - googleapis-common-protos==1.69.2
      - grpc-google-iam-v1==0.12.7
      - grpcio==1.71.0
      - grpcio-gcp==0.2.2
      - grpcio-status==1.49.0rc1
      - idna==3.10
      - iniconfig==2.1.0
      - mock==5.2.0
      - packaging==24.2
      - pluggy==1.5.0
      - protobuf==3.20.3
      - pyasn1==0.6.1
      - pyasn1-modules==0.4.2
      - pytest==8.3.5
      - pytest-asyncio==0.26.0
      - pytest-cov==6.0.0
      - pytest-mock==3.14.0
      - pytest-xdist==3.6.1
      - requests==2.32.3
      - rsa==4.9
      - six==1.17.0
      - tomli==2.2.1
      - typing-extensions==4.13.0
      - urllib3==2.3.0
prefix: /opt/conda/envs/python-spanner

tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_retryable_internal_error tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_retryable_internal_error_after_token tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_retryable_internal_error_no_token
tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_empty_raw tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_multiple_span_creation tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_non_empty_raw tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_non_retryable_internal_error tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_non_retryable_internal_error_after_token tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_non_retryable_internal_error_no_token tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_unavailable tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_unavailable_after_token tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_raising_unavailable_no_token tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_raw_w_resume_tken tests/unit/test_snapshot.py::Test_restart_on_unavailable::test_iteration_w_span_creation tests/unit/test_snapshot.py::Test_SnapshotBase::test__make_txn_selector_virtual tests/unit/test_snapshot.py::Test_SnapshotBase::test_ctor tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_other_error tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_multi_use_w_first tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_multi_use_w_first_w_count_gt_0 tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_multi_use_wo_first tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_multi_use_wo_first_w_count_gt_0 tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_params_wo_param_types tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_query_options tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_retry tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_w_timeout tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_wo_multi_use tests/unit/test_snapshot.py::Test_SnapshotBase::test_execute_sql_wo_multi_use_w_read_request_count_gt_0 tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_query_ok_w_index_no_options tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_query_ok_w_max_partitions tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_query_ok_w_size tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_query_other_error tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_query_single_use_raises tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_query_w_params_wo_param_types tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_query_wo_transaction_raises tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_read_ok_w_index_no_options tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_read_ok_w_max_partitions tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_read_ok_w_size tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_read_other_error tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_read_single_use_raises tests/unit/test_snapshot.py::Test_SnapshotBase::test_partition_read_wo_existing_transaction_raises tests/unit/test_snapshot.py::Test_SnapshotBase::test_read_other_error tests/unit/test_snapshot.py::Test_SnapshotBase::test_read_w_multi_use_w_first_w_count_gt_0 tests/unit/test_snapshot.py::Test_SnapshotBase::test_read_w_multi_use_w_first_w_partition tests/unit/test_snapshot.py::Test_SnapshotBase::test_read_w_multi_use_wo_first tests/unit/test_snapshot.py::Test_SnapshotBase::test_read_w_multi_use_wo_first_w_count_gt_0 tests/unit/test_snapshot.py::Test_SnapshotBase::test_read_wo_multi_use tests/unit/test_snapshot.py::Test_SnapshotBase::test_read_wo_multi_use_w_read_request_count_gt_0 tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_strong tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_strong_w_multi_use tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_w_exact_staleness tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_w_exact_staleness_w_multi_use tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_w_max_staleness tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_w_min_read_timestamp tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_w_read_timestamp tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_w_read_timestamp_w_multi_use tests/unit/test_snapshot.py::TestSnapshot::test__make_txn_selector_w_transaction_id tests/unit/test_snapshot.py::TestSnapshot::test_begin_ok_exact_staleness tests/unit/test_snapshot.py::TestSnapshot::test_begin_ok_exact_strong tests/unit/test_snapshot.py::TestSnapshot::test_begin_w_existing_txn_id tests/unit/test_snapshot.py::TestSnapshot::test_begin_w_other_error tests/unit/test_snapshot.py::TestSnapshot::test_begin_w_read_request_count_gt_0 tests/unit/test_snapshot.py::TestSnapshot::test_begin_wo_multi_use tests/unit/test_snapshot.py::TestSnapshot::test_ctor_defaults tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_exact_staleness tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_max_staleness tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_min_read_timestamp tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_multi_use tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_multi_use_and_exact_staleness tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_multi_use_and_max_staleness tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_multi_use_and_min_read_timestamp tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_multi_use_and_read_timestamp tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_multiple_options tests/unit/test_snapshot.py::TestSnapshot::test_ctor_w_read_timestamp
google/cloud/spanner_v1/snapshot.py
google/cloud/spanner_v1/snapshot.py