[
https://issues.apache.org/jira/browse/BEAM-7961?focusedWorklogId=377123&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-377123
]
ASF GitHub Bot logged work on BEAM-7961:
----------------------------------------
Author: ASF GitHub Bot
Created on: 25/Jan/20 01:14
Start Date: 25/Jan/20 01:14
Worklog Time Spent: 10m
Work Description: ihji commented on pull request #10051: [BEAM-7961] Add
tests for all runner native transforms for XLang
URL: https://github.com/apache/beam/pull/10051#discussion_r370902160
##########
File path:
sdks/java/testing/expansion-service/src/test/java/org/apache/beam/sdk/expansion/TestExpansionService.java
##########
@@ -55,39 +88,154 @@
@Override
public Map<String, ExpansionService.TransformProvider> knownTransforms() {
Schema schema = new Schema.Parser().parse(rawSchema);
- return ImmutableMap.of(
- TEST_COUNT_URN, spec -> Count.perElement(),
+ ImmutableMap.Builder<String, ExpansionService.TransformProvider> builder
=
+ ImmutableMap.builder();
+ builder.put(
+ TEST_PREFIX_URN,
+ spec ->
+ MapElements.into(TypeDescriptors.strings())
+ .via((String x) -> spec.getPayload().toStringUtf8() + x));
+ builder.put(TEST_MULTI_URN, spec -> new TestMultiPTransform());
+ builder.put(TEST_GBK_URN, spec -> GroupByKey.<Long, String>create());
+ builder.put(TEST_CGBK_URN, new TestCoGroupByKeyTransformProvider());
+ builder.put(TEST_COMGL_URN, spec -> Sum.longsGlobally());
+ builder.put(TEST_COMPK_URN, spec -> Sum.<String>longsPerKey());
+ builder.put(TEST_FLATTEN_URN, new TestFlattenTransformProvider());
+ builder.put(
+ TEST_PARTITION_URN,
+ spec -> Partition.of(2, (Long elem, int numP) -> elem % 2 == 0 ? 0 :
1));
+ builder.put(
+ TEST_PARQUET_WRITE_URN,
+ spec ->
+ new PTransform<PCollection<GenericRecord>,
PCollection<String>>() {
+ @Override
+ public PCollection<String> expand(PCollection<GenericRecord>
input) {
+ return input
+ .apply(
+ FileIO.<GenericRecord>write()
+ .via(ParquetIO.sink(schema))
+ .to(spec.getPayload().toStringUtf8()))
+ .getPerDestinationOutputFilenames()
+ .apply(Values.create());
+ }
+ });
+
+ builder.put(TEST_COUNT_URN, spec -> Count.perElement());
+ builder.put(
TEST_FILTER_URN,
- spec ->
- Filter.lessThanEq(
- // TODO(BEAM-6587): Use strings directly rather than
longs.
- (long) spec.getPayload().toStringUtf8().charAt(0)),
+ spec ->
+ Filter.lessThanEq(
+ // TODO(BEAM-6587): Use strings directly rather than longs.
+ (long) spec.getPayload().toStringUtf8().charAt(0)));
+ builder.put(
TEST_PARQUET_READ_URN,
- spec ->
- new PTransform<PBegin, PCollection<GenericRecord>>() {
- @Override
- public PCollection<GenericRecord> expand(PBegin input) {
- return input
-
.apply(FileIO.match().filepattern(spec.getPayload().toStringUtf8()))
- .apply(FileIO.readMatches())
- .apply(ParquetIO.readFiles(schema))
- .setCoder(AvroCoder.of(schema));
- }
- },
- TEST_PARQUET_WRITE_URN,
- spec ->
- new PTransform<PCollection<GenericRecord>,
PCollection<String>>() {
- @Override
- public PCollection<String>
expand(PCollection<GenericRecord> input) {
- return input
- .apply(
- FileIO.<GenericRecord>write()
- .via(ParquetIO.sink(schema))
- .to(spec.getPayload().toStringUtf8()))
- .getPerDestinationOutputFilenames()
- .apply(Values.create());
- }
- });
+ spec ->
+ new PTransform<PBegin, PCollection<GenericRecord>>() {
+ @Override
+ public PCollection<GenericRecord> expand(PBegin input) {
+ return input
+
.apply(FileIO.match().filepattern(spec.getPayload().toStringUtf8()))
+ .apply(FileIO.readMatches())
+ .apply(ParquetIO.readFiles(schema))
+ .setCoder(AvroCoder.of(schema));
+ }
+ });
+ return builder.build();
+ }
+
+ public static class TestFlattenTransformProvider
+ implements ExpansionService.TransformProvider<PCollectionList<Long>,
PCollection<Long>> {
Review comment:
Update to use `ExternalTransformRegistrar` whenever possible.
----------------------------------------------------------------
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: 377123)
Time Spent: 17h 10m (was: 17h)
> Add tests for all runner native transforms and some widely used composite
> transforms to cross-language validates runner test suite
> ----------------------------------------------------------------------------------------------------------------------------------
>
> Key: BEAM-7961
> URL: https://issues.apache.org/jira/browse/BEAM-7961
> Project: Beam
> Issue Type: Improvement
> Components: testing
> Reporter: Heejong Lee
> Assignee: Heejong Lee
> Priority: Major
> Time Spent: 17h 10m
> Remaining Estimate: 0h
>
> Add tests for all runner native transforms and some widely used composite
> transforms to cross-language validates runner test suite
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
