Remove sdk pipeline options from the DoFn context

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=118403483

Author: aaltay

google/cloud/dataflow/runners/common.py

@@ -171,14 +171,9 @@ def existing_windows(self):
 
 class DoFnState(object):
   """Keeps track of state that DoFns want, currently, user counters.
-
-  Attributes:
-    pipeline_options: a PipelineOptions object associated with this DoFn.
-    step_name: name of the step as a string.
   """
 
-  def __init__(self, pipeline_options):
-    self.pipeline_options = pipeline_options
+  def __init__(self):
     self.step_name = ''
     self._user_counters = {}
 

google/cloud/dataflow/runners/direct_runner.py

@@ -77,10 +77,8 @@ def func_wrapper(self, pvalue, *args, **kwargs):
   @skip_if_cached
   def run_ParDo(self, transform_node):
     transform = transform_node.transform
-    options = transform_node.inputs[0].pipeline.options
     # TODO(gildea): what is the appropriate object to attach the state to?
-    context = DoFnProcessContext(label=transform.label,
-                                 state=DoFnState(options))
+    context = DoFnProcessContext(label=transform.label, state=DoFnState())
 
     # Construct the list of values from side-input PCollections that we'll
     # substitute into the arguments for DoFn methods.
@@ -107,6 +105,7 @@ def get_side_input_value(si):
 
     # TODO(robertwb): Do this type checking inside DoFnRunner to get it on
     # remote workers as well?
+    options = transform_node.inputs[0].pipeline.options
     if options is not None and options.view_as(TypeOptions).runtime_type_check:
       transform.dofn = TypeCheckWrapperDoFn(
           transform.dofn, transform.get_type_hints())

google/cloud/dataflow/transforms/core.py

@@ -61,7 +61,6 @@ class DoFnProcessContext(object):
       (in process method only; always None in start_bundle and finish_bundle)
     windows: windows of the element
       (in process method only; always None in start_bundle and finish_bundle)
-    pipeline_options: PipelineOptions object used for creating the pipeline.
     state: a DoFnState object, which holds the runner's internal state
       for this element.  For example, aggregator state is here.
       Not used by the pipeline code.
@@ -107,10 +106,6 @@ def aggregate_to(self, aggregator, input_value):
     """
     self.state.counter_for(aggregator).update(input_value)
 
-  @property
-  def pipeline_options(self):
-    return self.state.pipeline_options
-
 
 class DoFn(WithTypeHints):
   """A function object used by a transform with custom processing.

google/cloud/dataflow/worker/batchworker.py

@@ -405,8 +405,7 @@ def do_work(self, work_item):
         self.dynamic_split_result_to_report = None
 
         self.set_current_work_item_and_executor(work_item,
-                                                executor.MapTaskExecutor(
-                                                    self.pipeline_options))
+                                                executor.MapTaskExecutor())
       self.report_progress = True
       self.current_executor.execute(work_item.map_task)
     except Exception:  # pylint: disable=broad-except

google/cloud/dataflow/worker/executor.py

@@ -383,9 +383,9 @@ def process(self, o):
 class DoOperation(Operation):
   """A Do operation that will execute a custom DoFn for each input element."""
 
-  def __init__(self, spec, pipeline_options):
+  def __init__(self, spec):
     super(DoOperation, self).__init__(spec)
-    self.state = common.DoFnState(pipeline_options)
+    self.state = common.DoFnState()
 
   def _read_side_inputs(self, tags_and_types):
     """Generator reading side inputs in the order prescribed by tags_and_types.
@@ -728,8 +728,7 @@ class MapTaskExecutor(object):
   multiple_read_instruction_error_msg = (
       'Found more than one \'read instruction\' in a single \'map task\'')
 
-  def __init__(self, pipeline_options=None):
-    self.pipeline_options = pipeline_options
+  def __init__(self):
     self._ops = []
     self._read_operation = None
 
@@ -776,7 +775,7 @@ def execute(self, map_task, test_shuffle_source=None, test_shuffle_sink=None):
       elif isinstance(spec, maptask.WorkerPartialGroupByKey):
         op = create_pgbk_op(spec)
       elif isinstance(spec, maptask.WorkerDoFn):
-        op = DoOperation(spec, self.pipeline_options)
+        op = DoOperation(spec)
       elif isinstance(spec, maptask.WorkerGroupingShuffleRead):
         op = GroupedShuffleReadOperation(
             spec, shuffle_source=test_shuffle_source)