[jira] [Work logged] (BEAM-9650) Add consistent slowly changing side inputs support

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/BEAM-9650?focusedWorklogId=429599&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-429599 ]

ASF GitHub Bot logged work on BEAM-9650:
----------------------------------------

                Author: ASF GitHub Bot
            Created on: 01/May/20 18:28
            Start Date: 01/May/20 18:28
    Worklog Time Spent: 10m 
      Work Description: Ardagan commented on a change in pull request #11477:
URL: https://github.com/apache/beam/pull/11477#discussion_r418673674



##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PeriodicSequence.java
##########
@@ -21,33 +21,69 @@
 import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
 import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState;
 
-import java.util.List;
+import java.util.Objects;
 import javax.annotation.Nullable;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.schemas.JavaFieldSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
 import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
-import org.apache.beam.sdk.transforms.splittabledofn.Sizes;
 import org.apache.beam.sdk.transforms.splittabledofn.SplitResult;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
 import org.joda.time.Duration;
 import org.joda.time.Instant;
 
 /**
- * A {@link PTransform} which generates a sequence of timestamped elements at given interval in
- * runtime.
+ * A {@link PTransform} which generates a sequence of timestamped elements at given runtime
+ * interval.
  *
- * <p>Receives a PCollection<List<Long>> where each element triggers the generation of sequence and
- * has following elements: 0: first element timestamp 1: last element timestamp 2: interval
+ * <p>Transform will not output elements prior to target time. Transform can output elements at any
+ * time after target time.
  *
- * <p>All elements that have timestamp in the past will be output right away. Elements that have
- * timestamp in the future will be delayed.
- *
- * <p>Transform will not output elements prior to target timestamp. Transform can output elements at
- * any time after target timestamp.
+ * <p>Multiple elements can be output at given moment if their timestamp is earlier than current
+ * time.
  */
 @Experimental(Experimental.Kind.SPLITTABLE_DO_FN)
-public class PeriodicSequence extends PTransform<PCollection<List<Long>>, PCollection<Instant>> {
+public class PeriodicSequence
+    extends PTransform<PCollection<PeriodicSequence.SequenceDefinition>, PCollection<Instant>> {
+
+  @DefaultSchema(JavaFieldSchema.class)
+  public static class SequenceDefinition {
+    public Instant first;
+    public Instant last;
+    public Long durationMilliSec;
+
+    public SequenceDefinition() {}
+
+    public SequenceDefinition(Instant first, Instant last, Duration duration) {
+      this.first = first;
+      this.last = last;
+      this.durationMilliSec = duration.getMillis();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+
+      if (obj == null || obj.getClass() != this.getClass()) {
+        return false;
+      }
+
+      SequenceDefinition src = (SequenceDefinition) obj;
+      return src.first.equals(this.first)
+          && src.last.equals(this.last)
+          && src.durationMilliSec.equals(this.durationMilliSec);
+    }
+
+    @Override
+    public int hashCode() {
+      int result = Objects.hash(first, last, durationMilliSec);
+      return result;
+    }

Review comment:
       I tried to use it, but didn't manage to get AutoValue work. It failed to properly detect constructor for generated class. I tried to debug it, but it would be better to handle in separate PR.




----------------------------------------------------------------
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


Issue Time Tracking
-------------------

    Worklog Id:     (was: 429599)
    Time Spent: 9h 20m  (was: 9h 10m)

> Add consistent slowly changing side inputs support
> --------------------------------------------------
>
>                 Key: BEAM-9650
>                 URL: https://issues.apache.org/jira/browse/BEAM-9650
>             Project: Beam
>          Issue Type: Bug
>          Components: io-ideas
>            Reporter: Mikhail Gryzykhin
>            Assignee: Mikhail Gryzykhin
>            Priority: Major
>          Time Spent: 9h 20m
>  Remaining Estimate: 0h
>
> Add implementation for slowly changing dimentions based on [design doc](https://docs.google.com/document/d/1LDY_CtsOJ8Y_zNv1QtkP6AGFrtzkj1q5EW_gSChOIvg/edit]



--
This message was sent by Atlassian Jira
(v8.3.4#803005)