Clean up PValue and PCollection with clearer argument passing

Removes the deprecated "transform" argument to pvalue.PValue.

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

Author: charlesccychen

google/cloud/dataflow/io/iobase.py

@@ -731,7 +731,7 @@ def __init__(self, *args, **kwargs):
 
   def apply(self, pcoll):
     self._check_pcollection(pcoll)
-    return pvalue.PValue(pipeline=pcoll.pipeline, transform=self)
+    return pvalue.PValue(pcoll.pipeline)
 
 
 class Read(ptransform.PTransform):
@@ -753,7 +753,7 @@ def __init__(self, *args, **kwargs):
   def apply(self, pbegin):
     assert isinstance(pbegin, pvalue.PBegin)
     self.pipeline = pbegin.pipeline
-    return pvalue.PCollection(pipeline=self.pipeline, transform=self)
+    return pvalue.PCollection(self.pipeline)
 
   def get_windowing(self, unused_inputs):
     return core.Windowing(window.GlobalWindows())

google/cloud/dataflow/pvalue.py

@@ -30,38 +30,26 @@
 class PValue(object):
   """Base class for PCollection.
 
+  Dataflow users should not construct PValue objects directly in their
+  pipelines.
+
   A PValue has the following main characteristics:
     (1) Belongs to a pipeline. Added during object initialization.
     (2) Has a transform that can compute the value if executed.
     (3) Has a value which is meaningful if the transform was executed.
   """
 
-  def __init__(self, **kwargs):
+  def __init__(self, pipeline, tag=None, element_type=None):
     """Initializes a PValue with all arguments hidden behind keyword arguments.
 
     Args:
-      **kwargs: keyword arguments.
-
-    Raises:
-      ValueError: if the expected keyword arguments (pipeline, transform,
-        and optionally tag) are not present.
-
-    The method expects a pipeline and a transform keyword argument. However in
-    order to give a signal to users that they should not create these PValues
-    directly we obfuscate the arguments.
+      pipeline: Pipeline object for this PValue.
+      tag: Tag of this PValue.
+      element_type: The type of this PValue.
     """
-    if 'pipeline' not in kwargs or 'transform' not in kwargs:
-      raise ValueError(
-          'Missing required arguments (pipeline and transform): %s'
-          % kwargs.keys)
-    self.pipeline = kwargs.pop('pipeline')
-    # TODO(silviuc): Remove usage of the transform argument from all call sites.
-    # It is not used anymore and has been replaced with the producer attribute.
-    kwargs.pop('transform')
-    self.tag = kwargs.pop('tag', None)
-    self.element_type = kwargs.pop('element_type', None)
-    if kwargs:
-      raise ValueError('Unexpected keyword arguments: %s' % kwargs.keys())
+    self.pipeline = pipeline
+    self.tag = tag
+    self.element_type = element_type
     self.pipeline._add_pvalue(self)
     # The AppliedPTransform instance for the application of the PTransform
     # generating this PValue. The field gets initialized when a transform
@@ -103,11 +91,15 @@ def __or__(self, ptransform):
 
 
 class PCollection(PValue):
-  """A multiple values (potentially huge) container."""
+  """A multiple values (potentially huge) container.
+
+  Dataflow users should not construct PCollection objects directly in their
+  pipelines.
+  """
 
-  def __init__(self, **kwargs):
+  def __init__(self, pipeline, **kwargs):
     """Initializes a PCollection. Do not call directly."""
-    super(PCollection, self).__init__(**kwargs)
+    super(PCollection, self).__init__(pipeline, **kwargs)
 
   @property
   def windowing(self):
@@ -216,10 +208,7 @@ def __getitem__(self, tag):
       return self._pcolls[tag]
     if tag is not None:
       self._transform.side_output_tags.add(tag)
-    pcoll = PCollection(
-        pipeline=self._pipeline,
-        transform=self._transform,
-        tag=tag)
+    pcoll = PCollection(self._pipeline, tag=tag)
     # Transfer the producer from the DoOutputsTuple to the resulting
     # PCollection.
     pcoll.producer = self.producer

google/cloud/dataflow/pvalue_test.py

@@ -18,7 +18,6 @@
 
 from google.cloud.dataflow.pipeline import Pipeline
 from google.cloud.dataflow.pvalue import PValue
-from google.cloud.dataflow.transforms import PTransform
 
 
 class FakePipeline(Pipeline):
@@ -33,15 +32,9 @@ class PValueTest(unittest.TestCase):
 
   def test_pvalue_expected_arguments(self):
     pipeline = Pipeline('DirectPipelineRunner')
-    transform = PTransform()
-    value = PValue(pipeline=pipeline, transform=transform)
+    value = PValue(pipeline)
     self.assertEqual(pipeline, value.pipeline)
 
-  def test_pvalue_missing_arguments(self):
-    self.assertRaises(ValueError, PValue,
-                      pipeline=Pipeline('DirectPipelineRunner'))
-    self.assertRaises(ValueError, PValue, transform=PTransform())
-
 
 if __name__ == '__main__':
   unittest.main()

google/cloud/dataflow/runners/dataflow_runner.py

@@ -322,7 +322,7 @@ def apply_GroupByKey(self, transform, pcoll):
     coders.registry.verify_deterministic(
         coder.key_coder(), 'GroupByKey operation "%s"' % transform.label)
 
-    return pvalue.PCollection(pipeline=pcoll.pipeline, transform=transform)
+    return pvalue.PCollection(pcoll.pipeline)
 
   def run_GroupByKey(self, transform_node):
     input_tag = transform_node.inputs[0].tag
@@ -413,7 +413,7 @@ def run_ParDo(self, transform_node):
     step.add_property(PropertyNames.OUTPUT_INFO, outputs)
 
   def apply_CombineValues(self, transform, pcoll):
-    return pvalue.PCollection(pipeline=pcoll.pipeline, transform=transform)
+    return pvalue.PCollection(pcoll.pipeline)
 
   def run_CombineValues(self, transform_node):
     transform = transform_node.transform

google/cloud/dataflow/transforms/core.py

@@ -537,7 +537,7 @@ def apply(self, pcoll):
     self.side_output_tags = set()
     # TODO(robertwb): Change all uses of the dofn attribute to use fn instead.
     self.dofn = self.fn
-    return pvalue.PCollection(pipeline=pcoll.pipeline, transform=self)
+    return pvalue.PCollection(pcoll.pipeline)
 
   def with_outputs(self, *tags, **main_kw):
     """Returns a tagged tuple allowing access to the outputs of a ParDo.
@@ -1032,7 +1032,7 @@ def infer_output_type(self, input_type):
 
   def apply(self, pcoll):
     self._check_pcollection(pcoll)
-    return pvalue.PCollection(pipeline=pcoll.pipeline, transform=self)
+    return pvalue.PCollection(pcoll.pipeline)
 
 
 class Partition(PTransformWithSideInputs):
@@ -1207,7 +1207,7 @@ def _extract_input_pvalues(self, pvalueish):
   def apply(self, pcolls):
     for pcoll in pcolls:
       self._check_pcollection(pcoll)
-    return pvalue.PCollection(pipeline=self.pipeline, transform=self)
+    return pvalue.PCollection(self.pipeline)
 
 
 class Create(PTransform):
@@ -1240,7 +1240,7 @@ def infer_output_type(self, unused_input_type):
   def apply(self, pbegin):
     assert isinstance(pbegin, pvalue.PBegin)
     self.pipeline = pbegin.pipeline
-    return pvalue.PCollection(pipeline=self.pipeline, transform=self)
+    return pvalue.PCollection(self.pipeline)
 
   def get_windowing(self, unused_inputs):
     return Windowing(window.GlobalWindows())

google/cloud/dataflow/transforms/ptransform.py

@@ -410,7 +410,7 @@ def _extract_input_pvalues(self, pvalueish):
     from google.cloud.dataflow import pipeline
     # pylint: enable=g-import-not-at-top
     if isinstance(pvalueish, pipeline.Pipeline):
-      pvalueish = pvalue.PBegin(pipeline=pvalueish, transform=None)
+      pvalueish = pvalue.PBegin(pvalueish)
 
     return pvalueish, (pvalueish,)