code: def _BigQueryToElwc(
      pipeline: beam.Pipeline,
      exec_properties: Dict[str, Any],
      split_pattern: str,
  ) -> beam.pvalue.PCollection:
      """Read from BigQuery and transform to ExampleListWithContext.
  
      When a field has no value in BigQuery, a feature with no value will be
      generated in the tf.train.Features. This behavior is consistent with
      BigQueryExampleGen.
  
      Args:
        pipeline: beam pipeline.
        exec_properties: A dict of execution properties.
        split_pattern: Split.pattern in Input config, a BigQuery sql string.
  
      Returns:
        PCollection of ExampleListWithContext.
  
      Raises:
        RuntimeError: Context features must be included in the queried result.
      """
      project = utils.parse_gcp_project(
          exec_properties["_beam_pipeline_args"]
      )
      custom_config = example_gen_pb2.CustomConfig()
      json_format.Parse(exec_properties["custom_config"], custom_config)
      elwc_config = elwc_config_pb2.ElwcConfig()
      custom_config.custom_config.Unpack(elwc_config)
  
      client = bigquery.Client(project=project)
      # Dummy query to get the type information for each field.
      query_job = client.query(
          "SELECT * FROM ({}) LIMIT 0".format(split_pattern)
      )
      results = query_job.result()
      type_map = {}
      context_feature_fields = set(elwc_config.context_feature_fields)
      field_names = set()
      for field in results.schema:
          type_map[field.name] = field.field_type
          field_names.add(field.name)
      # Check whether the query contains necessary context fields.
      if not field_names.issuperset(context_feature_fields):
          raise RuntimeError(
              "Context feature fields are missing from the query."
          )
  
      return (
          pipeline
          | "ReadFromBigQuery"
          >> utils.ReadFromBigQuery(query=split_pattern)
          | "RowToContextFeatureAndExample"
          >> beam.ParDo(
              _RowToContextFeatureAndExample(
                  type_map, context_feature_fields
              )
          )
          | "CombineByContext"
          >> beam.CombinePerKey(beam.combiners.ToListCombineFn())
          | "ConvertContextAndExamplesToElwc"
          >> beam.Map(_ConvertContextAndExamplesToElwc)
      )
  
file_path: tfx/tfx/extensions/google_cloud_big_query/experimental/elwc_example_gen/component/executor.py
imports: from typing import Any, Dict, Iterable, List, Set, Tuple
  import apache_beam
  from google.cloud import bigquery
  import tensorflow
  from tfx.components.example_gen import base_example_gen_executor
  from tfx.extensions.google_cloud_big_query import utils
  from tfx.extensions.google_cloud_big_query.experimental.elwc_example_gen.proto import elwc_config_pb2
  from tfx.proto import example_gen_pb2
  from google.protobuf import json_format
  from tensorflow_serving.apis import input_pb2
label_desc: Use image node_id 2 for calling a global function with example usage: _BigQueryToElwc(pipeline, exec_properties, split_pattern) and returns: unknown