def METHOD_NAME(context, source, source_name):
    destination = context.op_config.get("destination")
    load_job_config = _preprocess_config(context.op_config.get("load_job_config", {}))
    cfg = LoadJobConfig(**load_job_config) if load_job_config else None
    context.log.info(
        "executing BQ load with config: %s for source %s"
        % (cfg.to_api_repr() if cfg else "(no config provided)", source)
    )
    if source_name == BigQueryLoadSource.DataFrame:
        context.resources.bigquery.load_table_from_dataframe(
            source, destination, job_config=cfg
        ).result()
    # Load from file. See: https://cloud.google.com/bigquery/docs/loading-data-local
    elif source_name == BigQueryLoadSource.File:
        with open(source, "rb") as file_obj:
            context.resources.bigquery.load_table_from_file(
                file_obj, destination, job_config=cfg
            ).result()
    # Load from GCS. See: https://cloud.google.com/bigquery/docs/loading-data-cloud-storage
    elif source_name == BigQueryLoadSource.GCS:
        context.resources.bigquery.load_table_from_uri(source, destination, job_config=cfg).result()