Side Input improvements (#38363)

* side-input improvements

* remove files

* add Override

* foo

* foo

* fix splittable dofn

* fix

* foo

* fix windows

* fix compilation

* foo

Author: reuvenlax

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

@@ -870,8 +870,17 @@ public InputT element(DoFn<InputT, OutputT> doFn) {
     }
 
     @Override
-    public Object sideInput(String tagId) {
-      throw new UnsupportedOperationException("SideInput parameters are not supported.");
+    public @Nullable Object sideInput(String tagId) {
+      PCollectionView<?> view =
+          checkStateNotNull(sideInputMapping.get(tagId), "Side input tag %s not found", tagId);
+      return sideInput(view);
+    }
+
+    @Override
+    public <T> T sideInput(PCollectionView<T> view) {
+      checkNotNull(view, "View passed to sideInput cannot be null");
+      return SimpleDoFnRunner.this.sideInput(
+          view, view.getWindowMappingFn().getSideInputWindow(window()));
     }
 
     @Override
@@ -1196,8 +1205,17 @@ public InputT element(DoFn<InputT, OutputT> doFn) {
     }
 
     @Override
-    public Object sideInput(String tagId) {
-      throw new UnsupportedOperationException("SideInput parameters are not supported.");
+    public @Nullable Object sideInput(String tagId) {
+      PCollectionView<?> view =
+          checkStateNotNull(sideInputMapping.get(tagId), "Side input tag %s not found", tagId);
+      return sideInput(view);
+    }
+
+    @Override
+    public <T> T sideInput(PCollectionView<T> view) {
+      checkNotNull(view, "View passed to sideInput cannot be null");
+      return SimpleDoFnRunner.this.sideInput(
+          view, view.getWindowMappingFn().getSideInputWindow(window()));
     }
 
     @Override

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleDoFnRunnerFactory.java

@@ -24,7 +24,6 @@
 import org.apache.beam.runners.core.SideInputReader;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.StreamingOptions;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
 import org.apache.beam.sdk.util.WindowedValueMultiReceiver;
@@ -68,17 +67,6 @@ public DoFnRunner<InputT, OutputT> createRunner(
             windowingStrategy,
             doFnSchemaInformation,
             sideInputMapping);
-    boolean hasStreamingSideInput =
-        options.as(StreamingOptions.class).isStreaming() && !sideInputReader.isEmpty();
-    if (hasStreamingSideInput) {
-      return new StreamingSideInputDoFnRunner<>(
-          fnRunner,
-          new StreamingSideInputFetcher<>(
-              sideInputViews,
-              inputCoder,
-              windowingStrategy,
-              (StreamingModeExecutionContext.StreamingModeStepContext) userStepContext));
-    }
     return fnRunner;
   }
 }