def run_load(self,
                 destination_project_dataset_table,
                 source_uris,
                 schema_fields=None,
                 source_format='CSV',
                 create_disposition='CREATE_IF_NEEDED',
                 skip_leading_rows=0,
                 write_disposition='WRITE_EMPTY',
                 field_delimiter=',',
                 max_bad_records=0,
                 quote_character=None,
                 ignore_unknown_values=False,
                 allow_quoted_newlines=False,
                 allow_jagged_rows=False,
                 schema_update_options=(),
                 src_fmt_configs=None,
                 time_partitioning=None,
                 cluster_fields=None,
                 autodetect=False):
        """
        
        """

        # bigquery only allows certain source formats
        # we check to make sure the passed source format is valid
        # if it's not, we raise a ValueError
        # Refer to this link for more details:
        #   https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query.tableDefinitions.(key).sourceFormat

        if schema_fields is None and not autodetect:
            raise ValueError(
                'You must either pass a schema or autodetect=True.')

        if src_fmt_configs is None:
            src_fmt_configs = {}

        source_format = source_format.upper()
        allowed_formats = [
            "CSV", "NEWLINE_DELIMITED_JSON", "AVRO", "GOOGLE_SHEETS",
            "DATASTORE_BACKUP", "PARQUET"
        ]
        if source_format not in allowed_formats:
            raise ValueError("{0} is not a valid source format. "
                             "Please use one of the following types: {1}"
                             .format(source_format, allowed_formats))

        # bigquery also allows you to define how you want a table's schema to change
        # as a side effect of a load
        # for more details:
        # https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.load.schemaUpdateOptions
        allowed_schema_update_options = [
            'ALLOW_FIELD_ADDITION', "ALLOW_FIELD_RELAXATION"
        ]
        if not set(allowed_schema_update_options).issuperset(
                set(schema_update_options)):
            raise ValueError(
                "{0} contains invalid schema update options."
                "Please only use one or more of the following options: {1}"
                .format(schema_update_options, allowed_schema_update_options))

        destination_project, destination_dataset, destination_table = \
            _split_tablename(table_input=destination_project_dataset_table,
                             default_project_id=self.project_id,
                             var_name='destination_project_dataset_table')

        configuration = {
            'load': {
                'autodetect': autodetect,
                'createDisposition': create_disposition,
                'destinationTable': {
                    'projectId': destination_project,
                    'datasetId': destination_dataset,
                    'tableId': destination_table,
                },
                'sourceFormat': source_format,
                'sourceUris': source_uris,
                'writeDisposition': write_disposition,
                'ignoreUnknownValues': ignore_unknown_values
            }
        }

        time_partitioning = _cleanse_time_partitioning(
            destination_project_dataset_table,
            time_partitioning
        )
        if time_partitioning:
            configuration['load'].update({
                'timePartitioning': time_partitioning
            })

        if cluster_fields:
            configuration['load'].update({'clustering': {'fields': cluster_fields}})

        if schema_fields:
            configuration['load']['schema'] = {'fields': schema_fields}

        if schema_update_options:
            if write_disposition not in ["WRITE_APPEND", "WRITE_TRUNCATE"]:
                raise ValueError("schema_update_options is only "
                                 "allowed if write_disposition is "
                                 "'WRITE_APPEND' or 'WRITE_TRUNCATE'.")
            else:
                self.log.info(
                    "Adding experimental 'schemaUpdateOptions': %s",
                    schema_update_options
                )
                configuration['load'][
                    'schemaUpdateOptions'] = schema_update_options

        if max_bad_records:
            configuration['load']['maxBadRecords'] = max_bad_records

        # if following fields are not specified in src_fmt_configs,
        # honor the top-level params for backward-compatibility
        if 'skipLeadingRows' not in src_fmt_configs:
            src_fmt_configs['skipLeadingRows'] = skip_leading_rows
        if 'fieldDelimiter' not in src_fmt_configs:
            src_fmt_configs['fieldDelimiter'] = field_delimiter
        if 'ignoreUnknownValues' not in src_fmt_configs:
            src_fmt_configs['ignoreUnknownValues'] = ignore_unknown_values
        if quote_character is not None:
            src_fmt_configs['quote'] = quote_character
        if allow_quoted_newlines:
            src_fmt_configs['allowQuotedNewlines'] = allow_quoted_newlines

        src_fmt_to_configs_mapping = {
            'CSV': [
                'allowJaggedRows', 'allowQuotedNewlines', 'autodetect',
                'fieldDelimiter', 'skipLeadingRows', 'ignoreUnknownValues',
                'nullMarker', 'quote'
            ],
            'DATASTORE_BACKUP': ['projectionFields'],
            'NEWLINE_DELIMITED_JSON': ['autodetect', 'ignoreUnknownValues'],
            'PARQUET': ['autodetect', 'ignoreUnknownValues'],
            'AVRO': ['useAvroLogicalTypes'],
        }
        valid_configs = src_fmt_to_configs_mapping[source_format]
        src_fmt_configs = {
            k: v
            for k, v in src_fmt_configs.items() if k in valid_configs
        }
        configuration['load'].update(src_fmt_configs)

        if allow_jagged_rows:
            configuration['load']['allowJaggedRows'] = allow_jagged_rows

        return self.run_with_configuration(configuration)