Skip to content

[BEAM-3161] Cannot output with timestamp XXXX #4101

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mingmxu wants to merge 1 commit into from
Closed

[BEAM-3161] Cannot output with timestamp XXXX #4101

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...ons/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamAggregationRel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,8 @@ public PCollection<BeamRecord> buildBeamPipeline(PCollectionTuple inputPCollecti
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)))
Copy link
Contributor

@xumingming xumingming Nov 9, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Two comments here:

  • Will other DoFns we defined in BeamSQL have the similar issue? (The classes in org.apache.beam.sdk.extensions.sql.impl.transform package)
  • In the future, the allowed timestamp skew might be better passed in as a config to the BeamSql api?(I am ok to keep it hardcoded in this PR.)

Copy link
Author

@mingmxu mingmxu Nov 9, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. This is the only place where we assign EventTimestamp explicitly;
  2. We may not need this parameter, now the optional aligned time in tumble, hop and session is used as max_deplay_time;

.setCoder(upstream.getCoder());
}

Expand Down