

def run_ParDo(self, transform_node):
    transform = transform_node.transform
    input_tag = transform_node.inputs[0].tag
    input_step = self._cache.get_pvalue(transform_node.inputs[0])
    si_dict = {
        
    }
    si_tags_and_types = []
    for side_pval in transform_node.side_inputs:
        assert isinstance(side_pval, PCollectionView)
        side_input_step = self._cache.get_pvalue(side_pval)
        si_label = side_input_step.step_name
        si_dict[si_label] = {
            '@type': 'OutputReference',
            PropertyNames.STEP_NAME: side_tag,
            PropertyNames.OUTPUT_NAME: PropertyNames.OUT,
        }
        si_tags_and_types.append((si_label, side_pval.__class__, side_pval._view_options()))
    step = self._add_step(TransformNames.DO, transform_node.full_label, transform_node, transform_node.transform.side_output_tags)
    fn_data = (transform.fn, transform.args, transform.kwargs, si_tags_and_types, transform_node.inputs[0].windowing)
    step.add_property(PropertyNames.SERIALIZED_FN, pickler.dumps(fn_data))
    step.add_property(PropertyNames.PARALLEL_INPUT, {
        '@type': 'OutputReference',
        PropertyNames.STEP_NAME: input_step.proto.name,
        PropertyNames.OUTPUT_NAME: input_step.get_output(input_tag),
    })
    step.add_property(PropertyNames.NON_PARALLEL_INPUTS, si_dict)
    outputs = []
    step.encoding = self._get_typehint_based_encoding(self._get_transform_type_hint(transform_node))
    outputs.append({
        PropertyNames.USER_NAME: ('%s.%s' % (transform_node.full_label, PropertyNames.OUT)),
        PropertyNames.ENCODING: step.encoding,
        PropertyNames.OUTPUT_NAME: PropertyNames.OUT,
    })
    for side_tag in transform.side_output_tags:
        outputs.append({
            PropertyNames.USER_NAME: ('%s.%s' % (transform_node.full_label, side_tag)),
            PropertyNames.ENCODING: step.encoding,
            PropertyNames.OUTPUT_NAME: ('%s_%s' % (PropertyNames.OUT, side_tag)),
        })
    step.add_property(PropertyNames.OUTPUT_INFO, outputs)

dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/runners/dataflow_runner.py DataflowPipelineRunner.run_ParDo/VarMisuse@107/180 `si_label`->`side_tag`