[GitHub] [beam] Ardagan commented on a change in pull request #11477: [BEAM-9650] Add PeriodicSequence generator.

[GitBox <gi...@apache.org>]

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