Repository: beam Updated Branches: refs/heads/master 867d81684 -> 4451d5566
change `withAllowedTimestampSkew` Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/4a5741f2 Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/4a5741f2 Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/4a5741f2 Branch: refs/heads/master Commit: 4a5741f2f1aabac098e41cfe453b52876a1bcdb0 Parents: 867d816 Author: mingmxu <ming...@ebay.com> Authored: Wed Nov 8 13:44:18 2017 -0800 Committer: James Xu <xumingmi...@gmail.com> Committed: Thu Nov 9 13:42:04 2017 +0800 ---------------------------------------------------------------------- .../beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/4a5741f2/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java index e49e79c..6ed30b6 100644 --- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java +++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java @@ -80,7 +80,8 @@ public class BeamAggregationRel extends Aggregate implements BeamRelNode { BeamSqlRelUtils.getBeamRelInput(input).buildBeamPipeline(inputPCollections, sqlEnv); if (windowFieldIdx != -1) { upstream = upstream.apply(stageName + "assignEventTimestamp", WithTimestamps - .of(new BeamAggregationTransforms.WindowTimestampFn(windowFieldIdx))) + .of(new BeamAggregationTransforms.WindowTimestampFn(windowFieldIdx)) + .withAllowedTimestampSkew(new Duration(Long.MAX_VALUE))) .setCoder(upstream.getCoder()); }
