Validate side input parameters.

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java

@@ -53,6 +53,7 @@
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.MethodWithExtraParameters;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.OnTimerMethod;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.SchemaElementParameter;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.SideInputParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.WindowFn;
@@ -437,6 +438,29 @@ private static void validateStateApplicableForInput(DoFn<?, ?> fn, PCollection<?
  }
  }
 
+ private static void validateSideInputTypes(
+ Map<String, PCollectionView<?>> sideInputs, DoFn<?, ?> fn) {
+ DoFnSignature signature = DoFnSignatures.getSignature(fn.getClass());
+ DoFnSignature.ProcessElementMethod processElementMethod = signature.processElement();
+ for (SideInputParameter sideInput : processElementMethod.getSideInputParameters()) {
+ PCollectionView<?> view = sideInputs.get(sideInput.sideInputId());
+ checkArgument(
+ view != null,
+ "the ProcessElement method expects a side input identified with the tag %s, but no such side input was"
+ + " supplied. Use withSideInput(String, PCollectionView) to supply this side input.",
+ sideInput.sideInputId());
+ TypeDescriptor<?> viewType = view.getViewFn().getTypeDescriptor();
+
+ // Currently check that the types exactly match, even if the types are convertible.
+ checkArgument(
+ viewType.equals(sideInput.elementT()),
+ "Side Input with tag %s and type %s cannot be bound to ProcessElement parameter with type %s",
+ sideInput.sideInputId(),
+ viewType,
+ sideInput.elementT());
+ }
+ }
+
  private static FieldAccessDescriptor getFieldAccessDescriptorFromParameter(
  @Nullable String fieldAccessString,
  Schema inputSchema,
@@ -865,6 +889,8 @@ public PCollectionTuple expand(PCollection<? extends InputT> input) {
  validateStateApplicableForInput(fn, input);
  }
 
+ validateSideInputTypes(sideInputs, fn);
+
  // TODO: We should validate OutputReceiver<Row> only happens if the output PCollection
  // as schema. However coder/schema inference may not have happened yet at this point.
  // Need to figure out where to validate this.

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ViewFn.java

@@ -47,5 +47,6 @@ public abstract class ViewFn<PrimitiveViewT, ViewT> implements Serializable {
  /** A function to adapt a primitive view type to a desired view type. */
  public abstract ViewT apply(PrimitiveViewT primitiveViewT);
 
+ /** Return the {@link TypeDescriptor} describing the output of this fn. */
  public abstract TypeDescriptor<ViewT> getTypeDescriptor();
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java

@@ -40,6 +40,7 @@
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.OutputReceiverParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.RestrictionTrackerParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.SchemaElementParameter;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.SideInputParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.StateParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.TimerParameter;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.WindowParameter;
@@ -760,6 +761,14 @@ public List<SchemaElementParameter> getSchemaElementParameters() {
  .collect(Collectors.toList());
  }
 
+ @Nullable
+ public List<SideInputParameter> getSideInputParameters() {
+ return extraParameters().stream()
+ .filter(Predicates.instanceOf(SideInputParameter.class)::apply)
+ .map(SideInputParameter.class::cast)
+ .collect(Collectors.toList());
+ }
+
  /** The {@link OutputReceiverParameter} for a main output, or null if there is none. */
  @Nullable
  public OutputReceiverParameter getMainOutputReceiver() {

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

@@ -737,6 +737,76 @@ public void testParDoWithSideInputs() {
  pipeline.run();
  }
 
+ @Test
+ @Category({NeedsRunner.class, UsesSideInputs.class})
+ public void testSideInputAnnotationFailedValidationMissing() {
+ // SideInput tag id
+ final String sideInputTag1 = "tag1";
+
+ DoFn<Integer, List<Integer>> fn =
+ new DoFn<Integer, List<Integer>>() {
+ @ProcessElement
+ public void processElement(@SideInput(sideInputTag1) String tag1) {}
+ };
+
+ thrown.expect(IllegalArgumentException.class);
+ PCollection<List<Integer>> output =
+ pipeline.apply("Create main input", Create.of(2)).apply(ParDo.of(fn));
+ pipeline.run();
+ }
+
+ @Test
+ @Category({NeedsRunner.class, UsesSideInputs.class})
+ public void testSideInputAnnotationFailedValidationSingletonType() {
+
+ final PCollectionView<Integer> sideInput1 =
+ pipeline
+ .apply("CreateSideInput1", Create.of(2))
+ .apply("ViewSideInput1", View.asSingleton());
+
+ // SideInput tag id
+ final String sideInputTag1 = "tag1";
+
+ DoFn<Integer, List<Integer>> fn =
+ new DoFn<Integer, List<Integer>>() {
+ @ProcessElement
+ public void processElement(@SideInput(sideInputTag1) String tag1) {}
+ };
+
+ thrown.expect(IllegalArgumentException.class);
+ PCollection<List<Integer>> output =
+ pipeline
+ .apply("Create main input", Create.of(2))
+ .apply(ParDo.of(fn).withSideInput(sideInputTag1, sideInput1));
+ pipeline.run();
+ }
+
+ @Test
+ @Category({NeedsRunner.class, UsesSideInputs.class})
+ public void testSideInputAnnotationFailedValidationListType() {
+
+ final PCollectionView<List<Integer>> sideInput1 =
+ pipeline
+ .apply("CreateSideInput1", Create.of(2, 1, 0))
+ .apply("ViewSideInput1", View.asList());
+
+ // SideInput tag id
+ final String sideInputTag1 = "tag1";
+
+ DoFn<Integer, List<Integer>> fn =
+ new DoFn<Integer, List<Integer>>() {
+ @ProcessElement
+ public void processElement(@SideInput(sideInputTag1) List<String> tag1) {}
+ };
+
+ thrown.expect(IllegalArgumentException.class);
+ PCollection<List<Integer>> output =
+ pipeline
+ .apply("Create main input", Create.of(2))
+ .apply(ParDo.of(fn).withSideInput(sideInputTag1, sideInput1));
+ pipeline.run();
+ }
+
  @Test
  @Category({ValidatesRunner.class, UsesSideInputs.class})
  public void testSideInputAnnotation() {