Extract common class for WithTimestamps.AddTimestampsDoFn and Create.TimestampedValues.ConvertTimestamps

[kennknowles]

It seems like these APIs are predominantly duplicative of each other and, that it's hard to find one of them if you knew about the other.

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/WithTimestamps.java

Line 134 in 348d335

public void processElement(ProcessContext c) {

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java

Line 560 in 348d335

c.outputWithTimestamp(c.element().getValue(), c.element().getTimestamp());

What would make the most sense to me is if TimestampedValues were implemented in terms of both Values and WithTimestamps. I'm still learning about Beam though -- would this approach cause some kind of performance problem?

Imported from Jira BEAM-1811. Original Jira may contain additional context.
Reported by: wtanaka.

[Gaurav598]

I investigated the current implementation of this issue and wanted to leave a few findings here that may be useful to whoever picks it up next.

The duplication between WithTimestamps.AddTimestampsDoFn and Create.TimestampedValues.ConvertTimestamps is still present: both ultimately use outputWithTimestamp(...).

However, a direct Create.Values + WithTimestamps implementation does not appear to be a straightforward replacement. The timestamp is part of each TimestampedValue<T> and is not necessarily derivable from T, so duplicate/equal values can have different timestamps.

I also found that Create.TimestampedValues was previously related to Create.Values through inheritance, but that was intentionally removed because it caused runner matching issues. Reintroducing that relationship would therefore be risky.

The current TimestampedValues implementation also deliberately:

• infers the coder from the underlying T values,
• uses TimestampedValueCoder for the intermediate collection,
• unwraps the timestamped values through ConvertTimestamps,
• and explicitly restores the final Coder<T>.

Because of this, a conservative internal abstraction for the common timestamp-output behavior seems safer than restructuring TimestampedValues around Values + WithTimestamps. This would allow the existing execution path, coder behavior, and transform structure to remain unchanged.

The relevant tests also include renderer and Dataflow runner expectations around the current Create.TimestampedValues/ConvertTimestamps structure, so those are worth considering before changing the pipeline graph.
Hopefully these findings save some investigation time for the next contributor.