stankiewicz opened a new pull request, #40480:
URL: https://github.com/apache/beam/pull/40480
### Summary
Adds `withWriteResults()` to `SpannerIO.Write` and
`SpannerIO.WriteGrouped` (modeled after `BigtableIO.Write#withWriteResults()`)
to optionally enable `Options.commitStats()` on Cloud Spanner mutation writes
and return a
`PCollection<WriteResults>` containing commit metadata from
`com.google.cloud.spanner.CommitResponse`.
### Changes
* **`WriteResults`**:
* Added `WriteResults` annotated with `@DefaultCoder(AvroCoder.class)`
(using `TimestampEncoding` for `Timestamp` fields) exposing:
* `getMutationCount()` (`long`): populated from
`CommitResponse.getCommitStats().getMutationCount()`.
* `getCommitTimestamp()` (`@Nullable Timestamp`): populated from
`CommitResponse.getCommitTimestamp()`.
* `getSnapshotTimestamp()` (`@Nullable Timestamp`): populated from
`CommitResponse.getSnapshotTimestamp()`.
* **`SpannerIO.Write#withWriteResults()` &
`SpannerIO.WriteGrouped#withWriteResults()`**:
* Added `WriteWithResults` (`PTransform<PCollection<Mutation>,
PCollection<WriteResults>>`) and `WriteGroupedWithResults`
(`PTransform<PCollection<MutationGroup>, PCollection<WriteResults>>`).
* Enabled `Options.commitStats()` in
`WriteToSpannerFn#getTransactionOptions()` only when `withWriteResults()` is
used, preserving full backwards compatibility and zero overhead for existing
`SpannerIO.write()` users.
* Exposed `@Nullable PCollection<WriteResults> getWriteResults()` on
`SpannerWriteResult`.
* **Bundle Minimum Timestamp in `GatherSortCreateBatchesFn`**:
* Updated `GatherSortCreateBatchesFn` and
`OutputReceiverForFinishBundle` to track the minimum element timestamp in the
bundle (`minBundleTimestamp`) and assign it to batches flushed at
`@FinishBundle` instead of `Instant.
now()`.
* **Tests**:
* Added unit tests in `SpannerIOWriteTest` covering `WriteResults`
default `AvroCoder`, minimum bundle timestamp in `GatherSortCreateBatchesFn`,
`Write#withWriteResults()`, `WriteGrouped#withWriteResults()`, and timestamp
preservation when batching is disabled.
* Added integration test `testWriteWithResults` in `SpannerWriteIT`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
