You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/10/06 17:06:45 UTC
[GitHub] [beam] tysonjh opened a new pull request #12915: [BEAM-7386] Introduce temporal inner join.
tysonjh opened a new pull request #12915:
URL: https://github.com/apache/beam/pull/12915
Similar to other inner joins except it includes a temporal predicate,
allowing users to join unbounded PCollection<KV>s in the GlobalWindow.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) 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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | ](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-704405996
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh closed pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh closed pull request #12915:
URL: https://github.com/apache/beam/pull/12915
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce EventTimeBoundedEquijoin.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-893821773
Obsolete.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r493905016
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,276 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+ private transient boolean evictionTimerInitialized;
+
+ @Setup
+ public void setup() {
+ evictionTimerInitialized = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ /**
+ * Finds an element in {@code search} state that satisfies {@code func} within the non-inclusive
+ * time range (timestamp - temporalBound, timestamp + temporalBound).
+ *
+ * @param timestamp Instant that scopes the match within temporalBound (i.e. midpoint).
+ * @param search State to search.
+ * @param func Join predicate.
+ * @param <T> Type of the state collection and return value.
+ * @return Matching element or null if none is found.
+ */
+ @Nullable
+ private <T> TimestampedValue<T> findMatch(
+ Instant timestamp,
+ OrderedListState<T> search,
+ Function<TimestampedValue<T>, Boolean> func) {
+ Iterable<TimestampedValue<T>> searchIterable =
+ search.readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound));
+ if (searchIterable != null) {
+ for (TimestampedValue<T> current : searchIterable) {
+ if (new Duration(current.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && func.apply(current)) {
+ return current;
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Removes an element from state. This method is the semantic equivalent of an
+ * OrderedListState#remove which does not exist.
+ */
+ private <T> void remove(OrderedListState<T> state, TimestampedValue<T> element) {
+ Instant upperBound = element.getTimestamp().plus(1);
Review comment:
A few lines down the returned collection is iterated, filtering out the one for removal, adding the rest back to state.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce EventTimeBoundedEquijoin.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r524920768
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
+ rightState.clearRange(new Instant(0L), ts);
Review comment:
Yes, you're right. I want to test the eviction but I don't know how. When I try to use two TestStreams with various watermark updates it doesn't seem to trigger. Is there a way I can test this behavior?
My current thought is to subclass `EventTimeEquijoinFn` for the test and emit the state based on some kind of sentinel input value. I'd have to change the class visibility for this though which is not great.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r497696164
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,276 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+ private transient boolean evictionTimerInitialized;
+
+ @Setup
+ public void setup() {
+ evictionTimerInitialized = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ /**
+ * Finds an element in {@code search} state that satisfies {@code func} within the non-inclusive
+ * time range (timestamp - temporalBound, timestamp + temporalBound).
+ *
+ * @param timestamp Instant that scopes the match within temporalBound (i.e. midpoint).
+ * @param search State to search.
+ * @param func Join predicate.
+ * @param <T> Type of the state collection and return value.
+ * @return Matching element or null if none is found.
+ */
+ @Nullable
+ private <T> TimestampedValue<T> findMatch(
+ Instant timestamp,
+ OrderedListState<T> search,
+ Function<TimestampedValue<T>, Boolean> func) {
+ Iterable<TimestampedValue<T>> searchIterable =
+ search.readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound));
+ if (searchIterable != null) {
+ for (TimestampedValue<T> current : searchIterable) {
+ if (new Duration(current.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && func.apply(current)) {
+ return current;
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Removes an element from state. This method is the semantic equivalent of an
+ * OrderedListState#remove which does not exist.
+ */
+ private <T> void remove(OrderedListState<T> state, TimestampedValue<T> element) {
+ Instant upperBound = element.getTimestamp().plus(1);
Review comment:
Done. Closing this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r516142798
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
Review comment:
I think we should explore another name for this class. IMO TemporalJoin is best used for a join that is actually based on temporal properties - e.g. FOR SYSTEM_TIME AS OF T in SQL. This seems like it's just a regular inner join with a join timeout, plus use of state to allow faster emission of joined elements.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
Review comment:
Use @link tags to link directly to functions
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
Review comment:
Consider making this an AutoValue class.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
Review comment:
Add name parameter to all applies
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
Review comment:
Don't mark the OrderedListStates as AlwaysFetched, as this will cause us to fetch both lists in entirety on every processElement
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
Review comment:
This will be cleared if a process restarts or of the in-memory object is ever recycled. Is this necessary? If so I would make it a ValueState so it persists, otherwise I would remove it.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
Review comment:
Can you explain the use case of compareFn?
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
Review comment:
timestamps can be negative - use BoundedWindow.TIMESTAMP_MIN_VALUE instead.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
+ rightState.clearRange(new Instant(0L), ts);
Review comment:
Don't you want to ts.minus(temporalBound) to be the upper bound of the eviction?
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
Review comment:
Using KV's just to encode a pair (i.e. V1 isn't.a "key" and V2 isn't a "value") reads slightly odd. There's nothing particularly wrong with it, but it just reads a bit oddly.
You could also look at using org.apache.beam.sdk.transforms.join.RawUnionValue for this, like CoGroupByKey does.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
Review comment:
Did Beam not properly deduce a coder here on its own?
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
Review comment:
why should users prefer innerJoin?
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
Review comment:
Instead of using ProcessContext, just add the following parameters to processElement:
@Element KV<...> e,
OutputReceiver<KV<...>> output
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
Review comment:
Thinking about this....
1. At least Dataflow will not effectively cache this read today, especially since the read range keeps changing (though maybe we should fix this)
2. Executing the full join on every element can result in worst case O(n^2) behavior. If temporalBound isn't too bad it might not be that bad.
Wondering if instead of doing this on every element we should instead have timers that fire at some configurable frequency (maybe default to once a second), and output the joined elements in the timer. you could do this easily using TimerMap - just round the timestamp to the next minute boundary, and make that the timer key in the map.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
Review comment:
Seems more obvious to me to write
if (r.getTimestamp().isAfter(timestamp.minus(temporalBound))
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
+ rightState.clearRange(new Instant(0L), ts);
+ }
+ }
+
+ /**
+ * Inner joins two PCollection<KV>s that satisfy a temporal predicate.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows elements
+ * to be expired when they are irrelevant according to the event-time watermark. This helps reduce
+ * the search space, storage, and memory requirements.
+ *
+ * @param <K> Join key type.
+ * @param <V1> Left element type in the left collection.
+ * @param <V2> Right element type in the right collection.
+ * @param name Name of the PTransform.
+ * @param leftCollection Left collection of the join.
+ * @param rightCollection Right collection of the join.
+ * @param temporalBound Time domain range used in the join predicate (non-inclusive).
+ * @param compareFn Function used when comparing elements in the join predicate.
+ */
+ public static <K, V1, V2> PCollection<KV<K, KV<V1, V2>>> temporalInnerJoin(
Review comment:
I assume outer joins are for a later PR?
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
Review comment:
Roughly the same code as the above block, could be refactored into a function.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
Review comment:
We will want to add a watermark hold for this as well. This can only be done in a timer, using withOutputTimestamp
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r496891714
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,276 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+ private transient boolean evictionTimerInitialized;
+
+ @Setup
+ public void setup() {
+ evictionTimerInitialized = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ /**
+ * Finds an element in {@code search} state that satisfies {@code func} within the non-inclusive
+ * time range (timestamp - temporalBound, timestamp + temporalBound).
+ *
+ * @param timestamp Instant that scopes the match within temporalBound (i.e. midpoint).
+ * @param search State to search.
+ * @param func Join predicate.
+ * @param <T> Type of the state collection and return value.
+ * @return Matching element or null if none is found.
+ */
+ @Nullable
+ private <T> TimestampedValue<T> findMatch(
+ Instant timestamp,
+ OrderedListState<T> search,
+ Function<TimestampedValue<T>, Boolean> func) {
+ Iterable<TimestampedValue<T>> searchIterable =
+ search.readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound));
+ if (searchIterable != null) {
+ for (TimestampedValue<T> current : searchIterable) {
+ if (new Duration(current.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && func.apply(current)) {
+ return current;
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Removes an element from state. This method is the semantic equivalent of an
+ * OrderedListState#remove which does not exist.
+ */
+ private <T> void remove(OrderedListState<T> state, TimestampedValue<T> element) {
+ Instant upperBound = element.getTimestamp().plus(1);
Review comment:
I'm going to change the semantics of the join, so that it doesn't produce a 1:1 match, which means having a remove is no longer necessary. It will take me a few days to make the change so no need to review for now. I'll update when the PR is ready.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-720290599
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-697981031
> Out of curiosity, why are you adding this here instead of the schema join library (which SQL uses)?
I wasn't aware of the other join library. I saw the join extension library implementations, plus the previously closed PR in BEAM-7386, and thought that this new one should be placed near and made the assumption that SQL would reuse the implementation. Looking at it now though it seems like the SQL joins don't use the join extension library.
Should I keep this one around or refactor into the SQL schema join library?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-718179094
Sorry for the delay.
AFAIK both this and the schema library are limited today to equijoins. The schema API is designed so that we can extend it later with non equijoins, however doing arbitrary join conditions in a distributed manner can be a hard problem.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-721414819
> We could call this one "timestamp-bounded equijoin" or some such.
Ya this is a tough one to name, more input is welcome. I floated the following: EventTimeLimitedDurationInnerJoin, EventTimeScopedDurationInnerJoin
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-720671151
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-721328234
> > I am a bit confused about the usage of compareFn here. State is per key, so I believe that your DoFn will only join items that have the same key - the compareFn will never even get to compare items with different keys. Is the idea to allow the user to generate a subset of an equijoin?
>
> Yes, it will be a subset of an equijoin. Sorry for the confusion.
> > I am a bit confused about the usage of compareFn here. State is per key, so I believe that your DoFn will only join items that have the same key - the compareFn will never even get to compare items with different keys. Is the idea to allow the user to generate a subset of an equijoin?
>
> Yes, it will be a subset of an equijoin. Sorry for the confusion.
Now that i'm thinking about this further, the compareFn may be unnecessarily complicating the API for this join. I imagined it would be helpful for a user who wants to add logic before emitting a matched result, like a filter, but it would be more idiomatic for the user to apply a filter transform to the join result instead.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r516915079
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
Review comment:
I want to include a note for users to ensure they are aware of the other, more general, `innerJoin` and use it unless they specifically require joining two streams in the global window.
I haven't measured this, but I think there will be overhead with this join in managing state according to the event time, plus the implementation uses a flatten + stateful pardo, which probably has less throughput than a standard cGBK. If this isn't the case please let me know and I can reword.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r516908263
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
Review comment:
OK. I'm open to ideas. Since event time is important to this join and it has a scoped duration (without using the word window): EventTimeLimitedDurationInnerJoin, EventTimeScopedDurationInnerJoin
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-697906615
Out of curiosity, why are you adding this here instead of the schema join library (which SQL uses)?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-721332587
+1 IMO it would be useful if it somehow happened prior to the CGBK, but
since it's after a subsequent ParDo or Filter will likely be fused anyway.
On Tue, Nov 3, 2020 at 11:23 AM Tyson Hamilton <no...@github.com>
wrote:
> I am a bit confused about the usage of compareFn here. State is per key,
> so I believe that your DoFn will only join items that have the same key -
> the compareFn will never even get to compare items with different keys. Is
> the idea to allow the user to generate a subset of an equijoin?
>
> Yes, it will be a subset of an equijoin. Sorry for the confusion.
>
> I am a bit confused about the usage of compareFn here. State is per key,
> so I believe that your DoFn will only join items that have the same key -
> the compareFn will never even get to compare items with different keys. Is
> the idea to allow the user to generate a subset of an equijoin?
>
> Yes, it will be a subset of an equijoin. Sorry for the confusion.
>
> Now that i'm thinking about this further, the compareFn may be
> unnecessarily complicating the API for this join. I imagined it would be
> helpful for a user who wants to add logic before emitting a matched result,
> like a filter, but it would be more idiomatic for the user to apply a
> filter transform to the join result instead.
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/beam/pull/12915#issuecomment-721328234>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AFAYJVLK76ST2YWOMQGMURLSOBKEJANCNFSM4RXFYIIQ>
> .
>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r516908263
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
Review comment:
OK. I'm open to ideas. Since event time is important to this join and it has a scoped duration (without using the word window): EventTimeLimitedDurationInnerJoin
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r516953662
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
+ rightState.clearRange(new Instant(0L), ts);
+ }
+ }
+
+ /**
+ * Inner joins two PCollection<KV>s that satisfy a temporal predicate.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows elements
+ * to be expired when they are irrelevant according to the event-time watermark. This helps reduce
+ * the search space, storage, and memory requirements.
+ *
+ * @param <K> Join key type.
+ * @param <V1> Left element type in the left collection.
+ * @param <V2> Right element type in the right collection.
+ * @param name Name of the PTransform.
+ * @param leftCollection Left collection of the join.
+ * @param rightCollection Right collection of the join.
+ * @param temporalBound Time domain range used in the join predicate (non-inclusive).
+ * @param compareFn Function used when comparing elements in the join predicate.
+ */
+ public static <K, V1, V2> PCollection<KV<K, KV<V1, V2>>> temporalInnerJoin(
Review comment:
Yes, I won't be doing it immediately however.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r517542010
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
Review comment:
cGBK also uses state under the covers and potentially needs to reiterate over state, so if anything this class is more efficient there. Likely what will makes the cGBK solution more efficient is processing all data once at EOW, instead of fetching state on every input element. However I would hesitate to make claims without actual bencvhmarks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-697541038
R: @reuvenlax
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r493832669
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,276 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+ private transient boolean evictionTimerInitialized;
+
+ @Setup
+ public void setup() {
+ evictionTimerInitialized = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ /**
+ * Finds an element in {@code search} state that satisfies {@code func} within the non-inclusive
+ * time range (timestamp - temporalBound, timestamp + temporalBound).
+ *
+ * @param timestamp Instant that scopes the match within temporalBound (i.e. midpoint).
+ * @param search State to search.
+ * @param func Join predicate.
+ * @param <T> Type of the state collection and return value.
+ * @return Matching element or null if none is found.
+ */
+ @Nullable
+ private <T> TimestampedValue<T> findMatch(
+ Instant timestamp,
+ OrderedListState<T> search,
+ Function<TimestampedValue<T>, Boolean> func) {
+ Iterable<TimestampedValue<T>> searchIterable =
+ search.readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound));
+ if (searchIterable != null) {
+ for (TimestampedValue<T> current : searchIterable) {
+ if (new Duration(current.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && func.apply(current)) {
+ return current;
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Removes an element from state. This method is the semantic equivalent of an
+ * OrderedListState#remove which does not exist.
+ */
+ private <T> void remove(OrderedListState<T> state, TimestampedValue<T> element) {
+ Instant upperBound = element.getTimestamp().plus(1);
Review comment:
If you have multiple elements with the same timestamp, this will remove all of those elements. Is that desired?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-720597130
> I am a bit confused about the usage of compareFn here. State is per key, so I believe that your DoFn will only join items that have the same key - the compareFn will never even get to compare items with different keys. Is the idea to allow the user to generate a subset of an equijoin?
Yes, it will be a subset of an equijoin. Sorry for the confusion.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r524898446
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
Review comment:
Done. Went with `EventTimeBoundedEquijoin`.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
Review comment:
Removed.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
Review comment:
I think I need it, so I made it into a state, but let me know if you know how to get rid of it please. I want to be able to differentiate between these states: never initialized, initialized but needs updating, doesn't need updating and track the last time the timer triggered.
What I'm doing here is:
`lastEviction == null` when the timer has never been initialized, use the first record received to set the timer
`lastEviction != null` when the timer has been initialized, rely on `evictionTimerSet`
`evictionTimerSet == true` don't update it, allow the trigger to reset the variable to `false`
`evictionTimerSet == false` set the timer on next record to `lastEvictionTime + evictionFrequency`.
I wouldn't need to do this if I could set timers in the `@OnTimer` method or if I could set some kind of looping trigger.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
Review comment:
Done.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
Review comment:
This wouldn't work. For example,
temporalBound = 3
r.getTimestamp() = 5
timestamp = 4
5.isAfter(4-3) == false
but this example should produce a join result since difference is within the temporalBound (i.e. 1.isShorterThan(3)).
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
Review comment:
Removed from other discussions.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
Review comment:
AutoValue doesn't support transient fields so the pcollection causes a problem.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
Review comment:
Done.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
Review comment:
I was following along with the other join methods in this class that use KVs. Using a `RawUnionValue` means I would need to create a results class (like cGBK) or un-tag the results afterwards just to produce a `KV<K, KV<V1, V2>>` like the other joins.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
Review comment:
Correct, it was unable to infer one.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
Review comment:
Done.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
Review comment:
Done.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
+ rightState.clearRange(new Instant(0L), ts);
Review comment:
Yes, you're right. I want to test the eviction but I don't know how. When I try to use two TestStreams with various watermark updates it doesn't seem to trigger. Is there a way I can test this behavior?
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
Review comment:
Could you elaborate a bit please? I don't understand why we want the hold, or what it accomplishes. The docs are a bit tricky to follow regarding this.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
Review comment:
Will the timer family reduce the worst case? The O(n^2) comes from searching through the state on each input, won't that still be required for finding the 'joined elements' to output in the timer?
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
Review comment:
The refactored function ends up looking pretty messy because of the types and the ordering required for creating them. Creating a `KV` causes problems, it requires passing a lambda to construct the KV with positional arguments (e.g. left vs. right).
It ends up looking pretty nasty, not worth the savings in lines IMO:
```java
void findAndOutputMatches(Instant timestamp, T searchValue, OrderedListState<U> searchState,
BiFunction<U, T, KV<K, KV<V1, V2>>> kvCreator, OutputReceiver<KV<K, KV<V1, V2>>> outputReceiver)
```
I pulled out the gnarly if condition to a function though, that cleans it up a bit.
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
Review comment:
Didn't know that, thanks. Done.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-721897966
FYI the actual class name can be a bit longer, as long there is a good builder method. e.g. You could do something like:
Join.boundedInnerJoin(pc1, pc2);
This would be easier to deal with if this contrib Join library used PTransforms instead of functions.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on a change in pull request #12915: [BEAM-7386] Introduce EventTimeBoundedEquijoin.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r534316752
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
+ rightState.clearRange(new Instant(0L), ts);
Review comment:
OK, bug fixed, testing added. Subclassing `EventTimeEquijoinFn` didn't work because of the DoFn reflection asserts that timer trigger callbacks must be defined int he same class as the class definition. It seems overly restrictive.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #12915: [BEAM-7386] Introduce EventTimeBoundedEquijoin.
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #12915:
URL: https://github.com/apache/beam/pull/12915#discussion_r534453439
##########
File path: sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
##########
@@ -350,6 +368,255 @@ public void processElement(ProcessContext c) {
return leftCollection.apply(name, InnerJoin.with(rightCollection));
}
+ /**
+ * PTransform representing a temporal inner join of PCollection<KV>s.
+ *
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of the values for the right collection.
+ */
+ public static class TemporalInnerJoin<K, V1, V2>
+ extends PTransform<PCollection<KV<K, V1>>, PCollection<KV<K, KV<V1, V2>>>> {
+ private final transient PCollection<KV<K, V2>> rightCollection;
+ private final Duration temporalBound;
+ private final SimpleFunction<KV<V1, V2>, Boolean> comparatorFn;
+
+ private TemporalInnerJoin(
+ final PCollection<KV<K, V2>> rightCollection,
+ final Duration temporalBound,
+ final SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.temporalBound = temporalBound;
+ this.rightCollection = rightCollection;
+ this.comparatorFn = compareFn;
+ }
+
+ /**
+ * Returns a TemporalInnerJoin PTransform that joins two PCollection<KV>s.
+ *
+ * <p>Similar to {@code innerJoin} but also supports unbounded PCollections in the GlobalWindow.
+ * Join results will be produced eagerly as new elements are received, regardless of windowing,
+ * however users should prefer {@code innerJoin} in most cases for better throughput.
+ *
+ * <p>The non-inclusive {@code temporalBound}, used as part of the join predicate, allows
+ * elements to be expired when they are irrelevant according to the event-time watermark. This
+ * helps reduce the search space, storage, and memory requirements.
+ *
+ * @param rightCollection Right side collection of the join.
+ * @param temporalBound Duration used in the join predicate (non-inclusive).
+ * @param compareFn Join predicate used for matching elements.
+ * @param <K> Type of the key for both collections.
+ * @param <V1> Type of the values for the left collection.
+ * @param <V2> Type of values for the right collection.
+ */
+ public static <K, V1, V2> TemporalInnerJoin<K, V1, V2> with(
+ PCollection<KV<K, V2>> rightCollection,
+ Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ return new TemporalInnerJoin<>(rightCollection, temporalBound, compareFn);
+ }
+
+ @Override
+ public PCollection<KV<K, KV<V1, V2>>> expand(PCollection<KV<K, V1>> leftCollection) {
+ // left right
+ // tag-left tag-right (create union type)
+ // \ /
+ // flatten
+ // join
+
+ Coder<K> keyCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getKeyCoder();
+ Coder<V1> leftValueCoder = ((KvCoder<K, V1>) leftCollection.getCoder()).getValueCoder();
+ Coder<V2> rightValueCoder = ((KvCoder<K, V2>) rightCollection.getCoder()).getValueCoder();
+
+ PCollection<KV<K, KV<V1, V2>>> leftUnion =
+ leftCollection
+ .apply("LeftUnionTag", MapElements.via(new LeftUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ PCollection<KV<K, KV<V1, V2>>> rightUnion =
+ rightCollection
+ .apply("RightUnionTag", MapElements.via(new RightUnionTagFn<K, V1, V2>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(
+ NullableCoder.of(leftValueCoder), NullableCoder.of(rightValueCoder))));
+
+ return PCollectionList.of(leftUnion)
+ .and(rightUnion)
+ .apply(Flatten.pCollections())
+ .apply(
+ "TemporalInnerJoinFn",
+ ParDo.of(
+ new TemporalInnerJoinFn<>(
+ leftValueCoder, rightValueCoder, temporalBound, comparatorFn)));
+ }
+ }
+
+ private static class LeftUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V1>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V1> element) {
+ return KV.of(element.getKey(), KV.of(element.getValue(), null));
+ }
+ }
+
+ private static class RightUnionTagFn<K, V1, V2>
+ extends SimpleFunction<KV<K, V2>, KV<K, KV<V1, V2>>> {
+ @Override
+ public KV<K, KV<V1, V2>> apply(KV<K, V2> element) {
+ return KV.of(element.getKey(), KV.of(null, element.getValue()));
+ }
+ }
+
+ private static class TemporalInnerJoinFn<K, V1, V2>
+ extends DoFn<KV<K, KV<V1, V2>>, KV<K, KV<V1, V2>>> {
+
+ @StateId("left")
+ private final StateSpec<OrderedListState<V1>> leftStateSpec;
+
+ @StateId("right")
+ private final StateSpec<OrderedListState<V2>> rightStateSpec;
+
+ // Null only when uninitialized. After first element is received this will always be non-null.
+ @StateId("lastEviction")
+ private final StateSpec<ValueState<Instant>> lastEvictionStateSpec;
+
+ @TimerId("eviction")
+ private final TimerSpec evictionSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private final Duration temporalBound;
+ private final Duration evictionFrequency;
+ private final SimpleFunction<KV<V1, V2>, Boolean> compareFn;
+
+ // Tracks the state of the eviction timer. Value is true when the timer has been set and
+ // execution is waiting for the event time watermark to fire the timer according to the
+ // evictionFrequency. False after the timer has been fired, so processElement can set the timer
+ // using the previous firing event time.
+ private transient boolean evictionTimerSet;
+
+ @Setup
+ public void setup() {
+ evictionTimerSet = false;
+ }
+
+ protected TemporalInnerJoinFn(
+ final Coder<V1> leftCoder,
+ final Coder<V2> rightCoder,
+ final Duration temporalBound,
+ SimpleFunction<KV<V1, V2>, Boolean> compareFn) {
+ this.leftStateSpec = StateSpecs.orderedList(leftCoder);
+ this.rightStateSpec = StateSpecs.orderedList(rightCoder);
+ this.lastEvictionStateSpec = StateSpecs.value(InstantCoder.of());
+ this.temporalBound = temporalBound;
+ this.compareFn = compareFn;
+ this.evictionFrequency =
+ temporalBound.getMillis() <= 4 ? Duration.millis(1) : temporalBound.dividedBy(4);
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c,
+ @AlwaysFetched @StateId("left") OrderedListState<V1> leftState,
+ @AlwaysFetched @StateId("right") OrderedListState<V2> rightState,
+ @AlwaysFetched @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant timestamp,
+ @TimerId("eviction") Timer evictionTimer) {
+ Instant lastEviction = lastEvictionState.read();
+ if (lastEviction == null) {
+ // Initialize timer for the first time relatively since event time watermark is unknown.
+ evictionTimerSet = true;
+ evictionTimer.offset(evictionFrequency).setRelative();
+ } else if (!evictionTimerSet) {
+ // Set timer using persisted event watermark from last timer firing event time.
+ checkNotNull(lastEviction);
+ evictionTimerSet = true;
+ evictionTimer.set(lastEviction.plus(evictionFrequency));
+ }
+
+ KV<K, KV<V1, V2>> e = c.element();
+ K key = e.getKey();
+ V1 left = e.getValue().getKey();
+ V2 right = e.getValue().getValue();
+ if (left != null) {
+ leftState.add(TimestampedValue.of(left, timestamp));
+ rightState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ r -> {
+ KV<V1, V2> matchCandidate = KV.of(left, r.getValue());
+ if (new Duration(r.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ } else {
+ rightState.add(TimestampedValue.of(right, timestamp));
+ leftState
+ .readRange(timestamp.minus(temporalBound), timestamp.plus(temporalBound))
+ .forEach(
+ l -> {
+ KV<V1, V2> matchCandidate = KV.of(l.getValue(), right);
+ if (new Duration(l.getTimestamp(), timestamp).abs().isShorterThan(temporalBound)
+ && compareFn.apply(matchCandidate)) {
+ c.output(KV.of(key, matchCandidate));
+ }
+ });
+ }
+ }
+
+ @OnTimer("eviction")
+ public void onEviction(
+ @StateId("left") OrderedListState<V1> leftState,
+ @StateId("right") OrderedListState<V2> rightState,
+ @StateId("lastEviction") ValueState<Instant> lastEvictionState,
+ @Timestamp Instant ts) {
+ evictionTimerSet = false;
+ lastEvictionState.write(ts);
+ leftState.clearRange(new Instant(0L), ts);
+ rightState.clearRange(new Instant(0L), ts);
Review comment:
I think this is basically trying to avoid complexities brought in by subclasses. We might lift such restrictions. You may know that personally I generally dislike implementation inheritance and particularly access to private members of superclasses, so it might just reflect my bias.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #12915: [BEAM-7386] Introduce temporal inner join.
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #12915:
URL: https://github.com/apache/beam/pull/12915#issuecomment-697542150
/cc @kennknowles
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh closed pull request #12915: [BEAM-7386] Introduce EventTimeBoundedEquijoin.
Posted by GitBox <gi...@apache.org>.
tysonjh closed pull request #12915:
URL: https://github.com/apache/beam/pull/12915
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org