

def run_Read(self, transform_node):
    transform = transform_node.transform
    step = self._add_step(TransformNames.READ, transform_node.full_label, transform_node)
    if (transform.source.format == 'text'):
        step.add_property(PropertyNames.FILE_PATTERN, transform.source.path)
    elif (transform.source.format == 'bigquery'):
        if (transform.source.table_reference is not None):
            step.add_property(PropertyNames.BIGQUERY_DATASET, transform.source.table_reference.datasetId)
            step.add_property(PropertyNames.BIGQUERY_TABLE, transform.source.table_reference.tableId)
            if (transform.source.table_reference.projectId is not None):
                step.add_property(PropertyNames.BIGQUERY_PROJECT, transform.source.table_reference.projectId)
        elif (transform.source.query is not None):
            standard_options.add_property(PropertyNames.BIGQUERY_QUERY, transform.source.query)
        else:
            raise ValueError('BigQuery source %r must specify either a table or a query', transform.source)
    elif (transform.source.format == 'pubsub'):
        standard_options = transform_node.inputs[0].pipeline.options.view_as(StandardOptions)
        if (not standard_options.streaming):
            raise ValueError('PubSubSource is currently available for use only in streaming pipelines.')
        step.add_property(PropertyNames.PUBSUB_TOPIC, transform.source.topic)
        if transform.source.subscription:
            step.add_property(PropertyNames.PUBSUB_SUBSCRIPTION, transform.source.topic)
        if transform.source.id_label:
            step.add_property(PropertyNames.PUBSUB_ID_LABEL, transform.source.id_label)
    elif (transform.source.format == 'custom'):
        raise NotImplementedError
    else:
        raise ValueError(('Source %r has unexpected format %s.' % (transform.source, transform.source.format)))
    step.add_property(PropertyNames.FORMAT, transform.source.format)
    step.encoding = self._get_cloud_encoding(transform.source.coder)
    step.add_property(PropertyNames.OUTPUT_INFO, [{
        PropertyNames.USER_NAME: ('%s.%s' % (transform_node.full_label, PropertyNames.OUT)),
        PropertyNames.ENCODING: step.encoding,
        PropertyNames.OUTPUT_NAME: PropertyNames.OUT,
    }])

dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/runners/dataflow_runner.py DataflowPipelineRunner.run_Read/VarMisuse@174/227 `step`->`standard_options`