[
https://issues.apache.org/jira/browse/BEAM-10093?focusedWorklogId=442063&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-442063
]
ASF GitHub Bot logged work on BEAM-10093:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 05/Jun/20 22:27
Start Date: 05/Jun/20 22:27
Worklog Time Spent: 10m
Work Description: apilloud commented on a change in pull request #11820:
URL: https://github.com/apache/beam/pull/11820#discussion_r436191652
##########
File path:
sdks/java/testing/nexmark/src/test/java/org/apache/beam/sdk/nexmark/queries/sql/SqlBoundedSideInputJoinTest.java
##########
@@ -48,166 +47,182 @@
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
-import org.junit.experimental.categories.Category;
+import org.junit.experimental.runners.Enclosed;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
/** Test the various NEXMark queries yield results coherent with their models.
*/
-@RunWith(JUnit4.class)
+@RunWith(Enclosed.class)
public class SqlBoundedSideInputJoinTest {
- @Rule public TestPipeline p = TestPipeline.create();
+ private abstract static class SqlBoundedSideInputJoinTestCases {
- @Before
- public void setupPipeline() {
- NexmarkUtils.setupPipeline(NexmarkUtils.CoderStrategy.HAND, p);
- }
+ protected abstract SqlBoundedSideInputJoin getQuery(NexmarkConfiguration
configuration);
+
+ @Rule public TestPipeline p = TestPipeline.create();
+
+ @Before
+ public void setupPipeline() {
+ NexmarkUtils.setupPipeline(NexmarkUtils.CoderStrategy.HAND, p);
+ }
- /** Test {@code query} matches {@code model}. */
- private <T extends KnownSize> void queryMatchesModel(
- String name,
- NexmarkConfiguration config,
- NexmarkQueryTransform<T> query,
- NexmarkQueryModel<T> model,
- boolean streamingMode)
- throws Exception {
-
- ResourceId sideInputResourceId =
- FileSystems.matchNewResource(
- String.format(
- "%s/JoinToFiles-%s", p.getOptions().getTempLocation(), new
Random().nextInt()),
- false);
- config.sideInputUrl = sideInputResourceId.toString();
-
- try {
+ /** Test {@code query} matches {@code model}. */
+ private <T extends KnownSize> void queryMatchesModel(
+ String name,
+ NexmarkConfiguration config,
+ NexmarkQueryTransform<T> query,
+ NexmarkQueryModel<T> model,
+ boolean streamingMode)
+ throws Exception {
+
+ ResourceId sideInputResourceId =
+ FileSystems.matchNewResource(
+ String.format(
+ "%s/JoinToFiles-%s", p.getOptions().getTempLocation(), new
Random().nextInt()),
+ false);
+ config.sideInputUrl = sideInputResourceId.toString();
+
+ try {
+ PCollection<KV<Long, String>> sideInput =
NexmarkUtils.prepareSideInput(p, config);
+ query.setSideInput(sideInput);
+
+ PCollection<Event> events =
+ p.apply(
+ name + ".Read",
+ streamingMode
+ ? NexmarkUtils.streamEventsSource(config)
+ : NexmarkUtils.batchEventsSource(config));
+
+ PCollection<TimestampedValue<T>> results =
+ (PCollection<TimestampedValue<T>>) events.apply(new
NexmarkQuery<>(config, query));
+ PAssert.that(results).satisfies(model.assertionFor());
+ PipelineResult result = p.run();
+ result.waitUntilFinish();
+ } finally {
+ NexmarkUtils.cleanUpSideInput(config);
+ }
+ }
+
+ /**
+ * A smoke test that the count of input bids and outputs are the same, to
help diagnose
+ * flakiness in more complex tests.
+ */
+ @Test
+ public void inputOutputSameEvents() throws Exception {
+ NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
+ config.sideInputType = NexmarkUtils.SideInputType.DIRECT;
+ config.numEventGenerators = 1;
+ config.numEvents = 5000;
+ config.sideInputRowCount = 10;
+ config.sideInputNumShards = 3;
PCollection<KV<Long, String>> sideInput =
NexmarkUtils.prepareSideInput(p, config);
- query.setSideInput(sideInput);
-
- PCollection<Event> events =
- p.apply(
- name + ".Read",
- streamingMode
- ? NexmarkUtils.streamEventsSource(config)
- : NexmarkUtils.batchEventsSource(config));
-
- PCollection<TimestampedValue<T>> results =
- (PCollection<TimestampedValue<T>>) events.apply(new
NexmarkQuery<>(config, query));
- PAssert.that(results).satisfies(model.assertionFor());
- PipelineResult result = p.run();
- result.waitUntilFinish();
- } finally {
- NexmarkUtils.cleanUpSideInput(config);
+
+ try {
+ PCollection<Event> input =
p.apply(NexmarkUtils.batchEventsSource(config));
+ PCollection<Bid> justBids = input.apply(NexmarkQueryUtil.JUST_BIDS);
+ PCollection<Long> bidCount = justBids.apply("Count Bids",
Count.globally());
+
+ NexmarkQueryTransform<Bid> query = getQuery(config);
+ query.setSideInput(sideInput);
+
+ PCollection<TimestampedValue<Bid>> output =
+ (PCollection<TimestampedValue<Bid>>) input.apply(new
NexmarkQuery(config, query));
+ PCollection<Long> outputCount = output.apply("Count outputs",
Count.globally());
+
+
PAssert.that(PCollectionList.of(bidCount).and(outputCount).apply(Flatten.pCollections()))
+ .satisfies(
+ counts -> {
+ assertThat(Iterables.size(counts), equalTo(2));
+ assertThat(Iterables.get(counts, 0), greaterThan(0L));
+ assertThat(Iterables.get(counts, 0),
equalTo(Iterables.get(counts, 1)));
+ return null;
+ });
+ p.run();
+ } finally {
+ NexmarkUtils.cleanUpSideInput(config);
+ }
}
- }
- /**
- * A smoke test that the count of input bids and outputs are the same, to
help diagnose flakiness
- * in more complex tests.
- */
- @Test
- @Category(NeedsRunner.class)
- public void inputOutputSameEvents() throws Exception {
- NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
- config.sideInputType = NexmarkUtils.SideInputType.DIRECT;
- config.numEventGenerators = 1;
- config.numEvents = 5000;
- config.sideInputRowCount = 10;
- config.sideInputNumShards = 3;
- PCollection<KV<Long, String>> sideInput = NexmarkUtils.prepareSideInput(p,
config);
-
- try {
- PCollection<Event> input =
p.apply(NexmarkUtils.batchEventsSource(config));
- PCollection<Bid> justBids = input.apply(NexmarkQueryUtil.JUST_BIDS);
- PCollection<Long> bidCount = justBids.apply("Count Bids",
Count.globally());
-
- NexmarkQueryTransform<Bid> query = new SqlBoundedSideInputJoin(config);
- query.setSideInput(sideInput);
-
- PCollection<TimestampedValue<Bid>> output =
- (PCollection<TimestampedValue<Bid>>) input.apply(new
NexmarkQuery(config, query));
- PCollection<Long> outputCount = output.apply("Count outputs",
Count.globally());
-
-
PAssert.that(PCollectionList.of(bidCount).and(outputCount).apply(Flatten.pCollections()))
- .satisfies(
- counts -> {
- assertThat(Iterables.size(counts), equalTo(2));
- assertThat(Iterables.get(counts, 0), greaterThan(0L));
- assertThat(Iterables.get(counts, 0),
equalTo(Iterables.get(counts, 1)));
- return null;
- });
- p.run();
- } finally {
- NexmarkUtils.cleanUpSideInput(config);
+ @Test
+ public void queryMatchesModelBatchDirect() throws Exception {
+ NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
+ config.sideInputType = NexmarkUtils.SideInputType.DIRECT;
+ config.numEventGenerators = 1;
+ config.numEvents = 5000;
+ config.sideInputRowCount = 10;
+ config.sideInputNumShards = 3;
+
+ queryMatchesModel(
+ "SqlBoundedSideInputJoinTestBatch",
+ config,
+ getQuery(config),
+ new BoundedSideInputJoinModel(config),
+ false);
}
- }
- @Test
- @Category(NeedsRunner.class)
- public void queryMatchesModelBatchDirect() throws Exception {
- NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
- config.sideInputType = NexmarkUtils.SideInputType.DIRECT;
- config.numEventGenerators = 1;
- config.numEvents = 5000;
- config.sideInputRowCount = 10;
- config.sideInputNumShards = 3;
-
- queryMatchesModel(
- "SqlBoundedSideInputJoinTestBatch",
- config,
- new SqlBoundedSideInputJoin(config),
- new BoundedSideInputJoinModel(config),
- false);
- }
+ @Test
+ public void queryMatchesModelStreamingDirect() throws Exception {
+ NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
+ config.sideInputType = NexmarkUtils.SideInputType.DIRECT;
+ config.numEventGenerators = 1;
+ config.numEvents = 5000;
+ config.sideInputRowCount = 10;
+ config.sideInputNumShards = 3;
+ queryMatchesModel(
+ "SqlBoundedSideInputJoinTestStreaming",
+ config,
+ getQuery(config),
+ new BoundedSideInputJoinModel(config),
+ true);
+ }
+
+ @Test
+ public void queryMatchesModelBatchCsv() throws Exception {
+ NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
+ config.sideInputType = NexmarkUtils.SideInputType.CSV;
+ config.numEventGenerators = 1;
+ config.numEvents = 5000;
+ config.sideInputRowCount = 10;
+ config.sideInputNumShards = 3;
+
+ queryMatchesModel(
+ "SqlBoundedSideInputJoinTestBatch",
+ config,
+ getQuery(config),
+ new BoundedSideInputJoinModel(config),
+ false);
+ }
- @Test
- @Category(NeedsRunner.class)
- public void queryMatchesModelStreamingDirect() throws Exception {
- NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
- config.sideInputType = NexmarkUtils.SideInputType.DIRECT;
- config.numEventGenerators = 1;
- config.numEvents = 5000;
- config.sideInputRowCount = 10;
- config.sideInputNumShards = 3;
- queryMatchesModel(
- "SqlBoundedSideInputJoinTestStreaming",
- config,
- new SqlBoundedSideInputJoin(config),
- new BoundedSideInputJoinModel(config),
- true);
+ @Test
+ public void queryMatchesModelStreamingCsv() throws Exception {
+ NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
+ config.sideInputType = NexmarkUtils.SideInputType.CSV;
+ config.numEventGenerators = 1;
+ config.numEvents = 5000;
+ config.sideInputRowCount = 10;
+ config.sideInputNumShards = 3;
+ queryMatchesModel(
+ "SqlBoundedSideInputJoinTestStreaming",
+ config,
+ getQuery(config),
+ new BoundedSideInputJoinModel(config),
+ true);
+ }
}
- @Test
- @Category(NeedsRunner.class)
- public void queryMatchesModelBatchCsv() throws Exception {
- NexmarkConfiguration config = NexmarkConfiguration.DEFAULT.copy();
- config.sideInputType = NexmarkUtils.SideInputType.CSV;
- config.numEventGenerators = 1;
- config.numEvents = 5000;
- config.sideInputRowCount = 10;
- config.sideInputNumShards = 3;
-
- queryMatchesModel(
- "SqlBoundedSideInputJoinTestBatch",
- config,
- new SqlBoundedSideInputJoin(config),
- new BoundedSideInputJoinModel(config),
- false);
+ @RunWith(JUnit4.class)
+ public static class SqlBoundedSideInputJoinTest_Calcite extends
SqlBoundedSideInputJoinTestCases {
Review comment:
Checkstyle says no on the underscore. It looks to be failing in a few
other places as well.
----------------------------------------------------------------
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:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 442063)
Time Spent: 2h 20m (was: 2h 10m)
> Add ZetaSQL Nexmark variant
> ---------------------------
>
> Key: BEAM-10093
> URL: https://issues.apache.org/jira/browse/BEAM-10093
> Project: Beam
> Issue Type: New Feature
> Components: testing-nexmark
> Reporter: Kenneth Knowles
> Assignee: Kenneth Knowles
> Priority: P2
> Time Spent: 2h 20m
> Remaining Estimate: 0h
>
> Most queries will be identical, but best to simply stay decoupled, so this is
> a copy/paste/modify job.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
