fix(spanner): add Options save/restore to PartialResultSetSource

[devbww]

Save the options in force during PartialResultSetSource construction,
and then restore them over PartialResultSetReader calls. This means
that NextRow(), Finish() and TryCancel() will have the right
options installed during any RPCs they execute.

Tighten up the test by installing a non-matching OptionsSpan before
any NextRow() calls are made, and by expecting no TryCancel()s in
cases that complete their enumeration.

This is the Spanner equivalent of #8256.

---

This change is 

[google-cloud-cpp-bot]

Google Cloud Build Logs
For commit: 1c104a1a6c68272d388db673cf628b3ce9914ffe

• Public Access Build Logs
• Cloud Build Console (requires auth)
• Kokoro Logs (requires auth)

ℹ️ NOTE: Kokoro logs are linked from "Details" below.

[codecov]

Codecov Report

Merging #8355 (3e475aa) into main (fe2d568) will increase coverage by 0.00%.
The diff coverage is 100.00%.

@@           Coverage Diff           @@
##             main    #8355   +/-   ##
=======================================
  Coverage   94.99%   94.99%           
=======================================
  Files        1355     1355           
  Lines      120749   120790   +41     
=======================================
+ Hits       114700   114748   +48     
+ Misses       6049     6042    -7     

Impacted Files	Coverage Δ
google/cloud/spanner/results.h	91.66% <ø> (ø)
...loud/spanner/internal/partial_result_set_source.cc	95.18% <100.00%> (+0.11%)	⬆️
...cloud/spanner/internal/partial_result_set_source.h	100.00% <100.00%> (ø)
...spanner/internal/partial_result_set_source_test.cc	100.00% <100.00%> (ø)
google/cloud/bigtable/async_read_stream_test.cc	97.32% <0.00%> (-0.67%)	⬇️
...le/cloud/internal/default_completion_queue_impl.cc	97.15% <0.00%> (-0.57%)	⬇️
...cloud/pubsub/internal/subscription_session_test.cc	97.75% <0.00%> (-0.50%)	⬇️
google/cloud/completion_queue_test.cc	96.95% <0.00%> (-0.20%)	⬇️
google/cloud/pubsub/samples/samples.cc	92.10% <0.00%> (+0.07%)	⬆️
...le/cloud/spanner/database_admin_connection_test.cc	99.70% <0.00%> (+0.29%)	⬆️
... and 4 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a5a5458...3e475aa. Read the comment docs.

[coryan]

This got me thinking, should we make analogous changes to the resumable streaming RPC:

google-cloud-cpp/google/cloud/internal/resumable_streaming_read_rpc.h

Lines 93 to 99 in 18082df

	absl::variant<Status, ResponseType> Read() override {
	auto response = impl_->Read();
	if (absl::holds_alternative<ResponseType>(response)) {
	updater_(absl::get<ResponseType>(response), request_);
	has_received_data_ = true;
	return response;
	}

[coryan]

optional:

auto make_read_mock = [&response](int i) {
  return [&response, i]() {
    EXPECT_EQ(internal::CurrentOptions().get<StringOption>(), "ErrorOnIncompleteRow");
    return response[i];
  }
};
EXPECT_CALL(*grpc_reader, Read)
  .WillOnce(make_read_mock(0))
  .WillOnce(make_read_mock(1))
....

[devbww]

Good idea. I did something similar that eliminates the duplication and makes the mocks easier to use and understand. PTAL.

[google-cloud-cpp-bot]

Google Cloud Build Logs
For commit: 3e475aab1df59db0360992887bafbd1a0c15de5c

• Public Access Build Logs
• Cloud Build Console (requires auth)
• Kokoro Logs (requires auth)

ℹ️ NOTE: Kokoro logs are linked from "Details" below.

[devbww]

This got me thinking, should we make analogous changes to the resumable streaming RPC

I'll look into it.

[devbww]

This got me thinking, should we make analogous changes to the resumable streaming RPC

I'll look into it.

I had thought that the StreamRange changes would handle this, but it looks like there is a hole when the StreamRange destroys its reader_, as that might cause a cancellation RPC without the right options if the reader is not yet finished. I'm investigating fixing that, and making the generated and Spanner cases look as similar as possible.

[devbww]

This got me thinking, should we make analogous changes to the resumable streaming RPC

I'll look into it.

I had thought that the StreamRange changes would handle this, but it looks like there is a hole when the StreamRange destroys its reader_, as that might cause a cancellation RPC without the right options if the reader is not yet finished. I'm investigating fixing that, and making the generated and Spanner cases look as similar as possible.

I don't think there was really a hole, as destroying the unfinished streaming reader only calls directly into gRPC to try to cancel the stream. That is, there is no cancellation RPC or other code that might look at our Options. But, we should probably always restore the operation-start options whenever the user can call back into the library, no matter what code we think that executes. So, I created #8403.

[coryan]

I don't think there was really a hole, as destroying the unfinished streaming reader only calls directly into gRPC to try to cancel the stream.

I was thinking about google::cloud::internal::ResumableStreamingReadRpc<>, I just created #8404, you can tell me there if the changes are not needed.