[
https://issues.apache.org/jira/browse/BEAM-7961?focusedWorklogId=368376&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-368376
]
ASF GitHub Bot logged work on BEAM-7961:
----------------------------------------
Author: ASF GitHub Bot
Created on: 08/Jan/20 19:48
Start Date: 08/Jan/20 19:48
Worklog Time Spent: 10m
Work Description: chamikaramj 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_r364398245
##########
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:
Instead of implementing TransformProviders can we implement
TransformRegistrar and TransformBuilder combination which is our public API for
transform authors ? I think using low level API here will make this confusing
for users. I think we should ditch using TestExpansionService altogether and
try to implement integration tests using the real ExpansionService.
----------------------------------------------------------------
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: 368376)
Time Spent: 15h 10m (was: 15h)
> 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: 15h 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)
