### Description

There might be an use case where the export_format can be based on some dynamic values. So, adding export_format will help developers in future

### Use case/motivation

_No response_

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

logical_date=context["logical_date"],
            configuration=configuration,
            force_rerun=self.force_rerun,
        )
        try:
            self.log.info("Executing: %s", configuration)
            job: ExtractJob = self._submit_job(hook=hook, job_id=job_id, configuration=configuration)
        except Conflict:
            
            job = hook.get_job(
                project_id=self.project_id,
                location=self.location,
                job_id=job_id,
            )
            if job.state in self.reattach_states:
                
                job.result(timeout=self.result_timeout, retry=self.result_retry)
                self._handle_job_error(job)
            else:
                
                raise AirflowException(
                    f"Job with id: {job_id} already exists and is in {job.state} state. If you "
                    f"want to force rerun it consider setting `force_rerun=True`."
                    f"Or, if you want to reattach in this scenario add {job.state} to `reattach_states`"
                )
        conf = job.to_api_repr()["configuration"]["extract"]["sourceTable"]
        dataset_id, project_id, table_id = conf["datasetId"], conf["projectId"], conf["tableId"]
        BigQueryTableLink.persist(
            context=context,
            task_instance=self,