So i have been dealing with this issue for a while now without any light…
I have a DAG that queries data from BigQuery, and depending on the results some Dynamic Task Mappings are created to insert an entry in another BigQuery table using BigQueryInsertJobOperator…

For Example:

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryGetDataOperator, BigQueryInsertJobOperator
from airflow.utils.dates import days_ago
from airflow.decorators import task
from airflow import XComArg

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
    'retries': 1,
}

dag = DAG(
    dag_id='bigquery_data_transfer_mapped_correct',
    default_args=default_args,
    schedule_interval="@daily",
    catchup=False,
    tags=['example'],
)

  @task
  def get_data(sql):
      bq_hook = BigQueryHook(...)
  
      self.log.info('Fetching Data from:')
      self.log.info('Query: %s', sql)
  
      bq_client = bq_hook.get_client()
      query_job = bq_client.query(sql)
      client_results = query_job.result()  # Waits for the query to finish
  
      results = list(dict(result) for  result in client_results)
  
      self.log.info(f"Retrieved {len(results)} rows from BigQuery")
      self.log.info('Response: %s', results)
  
      return results

  query_data = get_data("SELECT * FROM some_table WHERE some_conditions;")
  
  insert = BigQueryInsertJobOperator.partial(
      task_id=f"insert_data",
      configuration={
          'query': {
              'query': "INSERT INTO `project.dataset.table` (field1, field2) VALUES ('{{ params.field1 }}', '{{ params.field2 }}')",
              'useLegacySql': False,
          }
      }
  ).expand(
      params=XComArg(query_data)
  )
  
  query_data >> insert

Please note that this code is just a basic example that i just wrote and in my usecase, i actually have a task_group that expands and takes in a parameter to be sent to the params in one of the BigQueryInsertJobOperator task.

After running my DAG i get an error saying:

Broken DAG: [/opt/airflow/dags/src/dag.py] Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 407, in apply_defaults
    default_args, merged_params = get_merged_defaults(
                                  ^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 167, in get_merged_defaults
    raise TypeError("params must be a mapping")
TypeError: params must be a mapping

The airflow version is:

Version: [v2.8.1](https://pypi.python.org/pypi/apache-airflow/2.8.1)
Git Version: .release:c0ffa9c5d96625c68ded9562632674ed366b5eb3