[BEAM-2914] Add TimestampPrefixingWindowCoder as beam:coder:custom_window:v1. #15137

y1chi · 2021-07-07T22:16:35Z

Add a new common standard coder beam:coder:custom_window:v1. The coder encodes arbitrary bounded window by prefixing the max timestamp to its encoded form using the window's registered coder. e.g.:
max_timestamp
encoded_window_bytes

The idea is that, any arbitrary window extending bounded window must have max timestamp defined and runner only needs to know the max timestamp when processing trigger timers. The encoded bytes using the original custom window's registered coder can be used transparently on runner side as the identity of the window.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Samza	Twister2
Go	---	---	---
Java
Python	---	---	---
XLang		---	---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

codecov · 2021-07-07T22:42:36Z

Codecov Report

Merging #15137 (65a7ce8) into master (e616db2) will increase coverage by 0.06%.
The diff coverage is 85.54%.

@@            Coverage Diff             @@
##           master   #15137      +/-   ##
==========================================
+ Coverage   83.77%   83.83%   +0.06%     
==========================================
  Files         441      441              
  Lines       59800    59692     -108     
==========================================
- Hits        50095    50043      -52     
+ Misses       9705     9649      -56

Impacted Files	Coverage Δ
sdks/python/apache_beam/dataframe/__init__.py	`100.00% <ø> (ø)`
...dks/python/apache_beam/examples/cookbook/coders.py	`62.16% <0.00%> (ø)`
...ython/apache_beam/examples/kafkataxi/kafka_taxi.py	`0.00% <0.00%> (ø)`
...on/apache_beam/runners/portability/spark_runner.py	`67.34% <33.33%> (ø)`
...m/runners/portability/spark_uber_jar_job_server.py	`84.32% <50.00%> (ø)`
...n/apache_beam/ml/gcp/recommendations_ai_test_it.py	`69.76% <69.76%> (ø)`
sdks/python/apache_beam/coders/coders.py	`87.80% <80.00%> (-0.23%)`	⬇️
...ks/python/apache_beam/ml/gcp/recommendations_ai.py	`87.26% <87.26%> (ø)`
sdks/python/apache_beam/dataframe/frames.py	`94.87% <97.95%> (+0.21%)`	⬆️
sdks/python/apache_beam/coders/coder_impl.py	`95.28% <100.00%> (+0.08%)`	⬆️
... and 40 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 2c6f42c...65a7ce8. Read the comment docs.

y1chi · 2021-07-13T19:22:20Z

Run Java PreCommit

y1chi · 2021-07-13T19:22:26Z

Run GoPortable PreCommit

model/pipeline/src/main/proto/beam_runner_api.proto

...nstruction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java

sdks/java/core/src/main/java/org/apache/beam/sdk/coders/TimestampPrefixingWindowCoder.java

sdks/java/core/src/test/java/org/apache/beam/sdk/coders/TimestampPrefixingWindowCoderTest.java

sdks/java/core/src/main/java/org/apache/beam/sdk/coders/TimestampPrefixingWindowCoder.java

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

sdks/java/core/src/test/java/org/apache/beam/sdk/coders/TimestampPrefixingWindowCoderTest.java

lukecwik · 2021-07-16T15:42:17Z

Run Java PreCommit

y1chi · 2021-07-16T16:47:17Z

Run Java PreCommit

robertwb · 2021-07-19T21:12:45Z

model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml

+#  }
+#  System.out.println(example);
+coder:
+  urn: "beam:coder:custom_window:v1"


Should this be timestamp_prefixed (similar to length_prefixed)?

I think custom_window makes it a bit more clear that this is a coder for window type. timestamp_prefixed also works as long as sdk and runner agree with the urn I guess.

robertwb · 2021-07-19T21:14:10Z

model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml

+#  System.out.println(example);
+coder:
+  urn: "beam:coder:custom_window:v1"
+  components: [{urn: "beam:coder:interval_window:v1"}]


This shouldn't be limited to interval_windows, any windowed_coder should do.

yep, using interval_window coder here makes it easier to verify in standard coder test.

robertwb · 2021-07-19T21:15:25Z

model/pipeline/src/main/proto/beam_runner_api.proto

+    //   maxTimestamp - A big endian 8 byte integer representing millis-since-epoch.
+    //     The encoded representation is shifted so that the byte representation
+    //     of negative values are lexicographically ordered before the byte
+    //     representation of positive values. This is typically done by


"typically done" makes it sound like any transformation preserving lexicographic ordering is acceptable. There is no choice here.

Perhaps we could reference other coders here (e.g. the encoding used for TimestmpCoder, or in WindowedValueCoder?

I suggested copying it verbatim as that is the text in those segments already.

robertwb · 2021-07-19T21:16:03Z

model/pipeline/src/main/proto/beam_runner_api.proto

+
+    // Encodes an arbitrary user defined window and its max timestamp (inclusive).
+    // The encoding format is:
+    //   maxTimestamp window


How is the window encoded? Should we length prefix it as well? Or do we expect TimestampPrefixed(LenghtPrefixed(CustomWindowCoder))?

I think whether it should be length prefixed can be determined by runner with coder overwrite, since we are not limiting the wrapped window coder type, known coder can be also used(though it is unlikely in real use case, but it is handy for unit tests) which don't need length prefix.

I would rather have the length prefix to be a component as Yichi describes since this will allow other length prefixing schemes then the current one we have.

robertwb · 2021-07-19T21:16:32Z

sdks/python/apache_beam/coders/coders.py

+  def is_deterministic(self) -> bool:
+    return self._window_coder.is_deterministic()
+
+  def as_cloud_object(self, coders_context=None):


Is this needed?

robertwb · 2021-07-19T21:16:48Z

...loud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/CloudObjectKinds.java

@@ -21,6 +21,7 @@
 class CloudObjectKinds {
  static final String KIND_GLOBAL_WINDOW = "kind:global_window";
  static final String KIND_INTERVAL_WINDOW = "kind:interval_window";
+  static final String KIND_CUSTOM_WINDOW = "kind:custom_window";


Why is this needed?

For translating it as a known coder on runner side.

It isn't needed since we shouldn't need to pass these in anymore to Dataflow since the beam proto -> DFE conversion should happen internally but I didn't ask to remove it since it was already done.

robertwb · 2021-07-19T21:18:56Z

sdks/java/core/src/test/java/org/apache/beam/sdk/coders/TimestampPrefixingWindowCoderTest.java

+  private static class CustomWindowCoder extends CustomCoder<CustomWindow> {
+
+    private static final Coder<IntervalWindow> INTERVAL_WINDOW_CODER = IntervalWindow.getCoder();
+    private static final int REGISTER_BYTE_SIZE = 1234;


I'm not following what is meant by the REGISTER_BYTE_SIZE constant. BYTE_SIZE_TO_REGISTER?

y1chi force-pushed the custom_window branch 9 times, most recently from a416bd0 to b7e1e7a Compare July 12, 2021 23:39

[WIP] Add TimestampPrefixingWindowCoder to java sdk.

40b0cbf

y1chi force-pushed the custom_window branch from b7e1e7a to 40b0cbf Compare July 13, 2021 18:07

y1chi changed the title ~~[WIP] Add TimestampPrefixingWindowCoder to java sdk.~~ [BEAM-2914] Add TimestampPrefixingWindowCoder as beam:coder:custom_window:v1. Jul 13, 2021

y1chi requested review from robertwb and lukecwik July 13, 2021 19:23

lukecwik reviewed Jul 14, 2021

View reviewed changes

y1chi force-pushed the custom_window branch from e9dea38 to 928766d Compare July 15, 2021 18:17

Address comments

2d095d3

y1chi force-pushed the custom_window branch from 928766d to 2d095d3 Compare July 15, 2021 18:49

lukecwik reviewed Jul 15, 2021

View reviewed changes

Tests update

65a7ce8

y1chi force-pushed the custom_window branch from 2ae0a1d to 65a7ce8 Compare July 16, 2021 01:38

lukecwik approved these changes Jul 16, 2021

View reviewed changes

y1chi merged commit 7d98ad2 into apache:master Jul 16, 2021

robertwb reviewed Jul 19, 2021

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-2914] Add TimestampPrefixingWindowCoder as beam:coder:custom_window:v1. #15137

[BEAM-2914] Add TimestampPrefixingWindowCoder as beam:coder:custom_window:v1. #15137

y1chi commented Jul 7, 2021 •

edited

Loading

codecov bot commented Jul 7, 2021 •

edited

Loading

y1chi commented Jul 13, 2021

y1chi commented Jul 13, 2021

lukecwik commented Jul 16, 2021

y1chi commented Jul 16, 2021

robertwb Jul 19, 2021

y1chi Jul 19, 2021

robertwb Jul 19, 2021

y1chi Jul 19, 2021

robertwb Jul 19, 2021

lukecwik Jul 19, 2021

robertwb Jul 19, 2021

y1chi Jul 19, 2021

lukecwik Jul 19, 2021

robertwb Jul 19, 2021

robertwb Jul 19, 2021

y1chi Jul 19, 2021

lukecwik Jul 19, 2021

robertwb Jul 19, 2021

[BEAM-2914] Add TimestampPrefixingWindowCoder as beam:coder:custom_window:v1. #15137

[BEAM-2914] Add TimestampPrefixingWindowCoder as beam:coder:custom_window:v1. #15137

Conversation

y1chi commented Jul 7, 2021 • edited Loading

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

codecov bot commented Jul 7, 2021 • edited Loading

Codecov Report

y1chi commented Jul 13, 2021

y1chi commented Jul 13, 2021

lukecwik commented Jul 16, 2021

y1chi commented Jul 16, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

y1chi commented Jul 7, 2021 •

edited

Loading

`ValidatesRunner` compliance status (on master branch)

codecov bot commented Jul 7, 2021 •

edited

Loading