[ https://issues.apache.org/jira/browse/BEAM-3749?focusedWorklogId=79440&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-79440 ]
ASF GitHub Bot logged work on BEAM-3749: ---------------------------------------- Author: ASF GitHub Bot Created on: 12/Mar/18 13:43 Start Date: 12/Mar/18 13:43 Worklog Time Spent: 10m Work Description: kennknowles closed pull request #4826: [BEAM-3749] Add test for a trigger with windowed SQL query URL: https://github.com/apache/beam/pull/4826 This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java index cdea0f8ce1f..ed19668d3b8 100644 --- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java +++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationTest.java @@ -27,6 +27,8 @@ import java.util.Iterator; import java.util.List; import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestStream; +import org.apache.beam.sdk.testing.UsesTestStream; import org.apache.beam.sdk.transforms.Create; import org.apache.beam.sdk.transforms.SerializableFunction; import org.apache.beam.sdk.transforms.windowing.AfterPane; @@ -45,6 +47,7 @@ import org.joda.time.Duration; import org.junit.Before; import org.junit.Test; +import org.junit.experimental.categories.Category; /** * Tests for GROUP-BY/aggregation, with global_window/fix_time_window/sliding_window/session_window @@ -328,6 +331,65 @@ private void runTumbleWindow(PCollection<Row> input) throws Exception { pipeline.run().waitUntilFinish(); } + /** + * Tests that a trigger set up prior to a SQL statement still is effective + * within the SQL statement. + */ + @Test + @Category(UsesTestStream.class) + public void testTriggeredTumble() throws Exception { + RowType inputRowType = + RowSqlType.builder().withIntegerField("f_int").withTimestampField("f_timestamp").build(); + + PCollection<Row> input = + pipeline.apply( + TestStream.create(inputRowType.getRowCoder()) + .addElements( + Row.withRowType(inputRowType) + .addValues(1, FORMAT.parse("2017-01-01 01:01:01")) + .build(), + Row.withRowType(inputRowType) + .addValues(2, FORMAT.parse("2017-01-01 01:01:01")) + .build()) + .addElements( + Row.withRowType(inputRowType) + .addValues(3, FORMAT.parse("2017-01-01 01:01:01")) + .build()) + .addElements( + Row.withRowType(inputRowType) + .addValues(4, FORMAT.parse("2017-01-01 01:01:01")) + .build()) + .advanceWatermarkToInfinity()); + + String sql = + "SELECT SUM(f_int) AS f_int_sum FROM PCOLLECTION" + + " GROUP BY TUMBLE(f_timestamp, INTERVAL '1' HOUR)"; + + RowType outputRowType = RowSqlType.builder().withIntegerField("fn_int_sum").build(); + + PCollection<Row> result = + input + .apply( + "Triggering", + Window.<Row>configure() + .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1))) + .withAllowedLateness(Duration.ZERO) + .withOnTimeBehavior(Window.OnTimeBehavior.FIRE_IF_NON_EMPTY) + .accumulatingFiredPanes()) + .apply("Windowed Query", BeamSql.query(sql)); + + PAssert.that(result) + .containsInAnyOrder( + TestUtils.RowsBuilder.of(outputRowType) + .addRows(3) // first bundle 1+2 + .addRows(6) // next bundle 1+2+3 + .addRows(10) // next bundle 1+2+3+4) + .getRows()); + + pipeline.run().waitUntilFinish(); + + } + /** * GROUP-BY with HOP window(aka sliding_window) with bounded PCollection. */ ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking ------------------- Worklog Id: (was: 79440) Time Spent: 3h 20m (was: 3h 10m) > support customized trigger/accumulationMode in BeamSql > ------------------------------------------------------ > > Key: BEAM-3749 > URL: https://issues.apache.org/jira/browse/BEAM-3749 > Project: Beam > Issue Type: Improvement > Components: dsl-sql > Reporter: Xu Mingmin > Assignee: Xu Mingmin > Priority: Major > Fix For: 2.5.0 > > Time Spent: 3h 20m > Remaining Estimate: 0h > > Currently BeamSql use {{DefaultTrigger}} for aggregation operations. > By adding two options {{withTrigger(Trigger)}} and > {{withAccumulationMode(AccumulationMode)}}, developers can specify their own > aggregation strategies with BeamSql. > [~xumingming] [~kedin] [~kenn] for any comments. -- This message was sent by Atlassian JIRA (v7.6.3#76005)