Set bigquery use_legacy_sql param in job config correctly (#28522)

Author: pankajastro

airflow/providers/google/cloud/operators/bigquery.py

@@ -205,7 +205,7 @@ def _submit_job(
         job_id: str,
     ) -> BigQueryJob:
         """Submit a new job and get the job id for polling the status using Trigger."""
-        configuration = {"query": {"query": self.sql}}
+        configuration = {"query": {"query": self.sql, "useLegacySql": self.use_legacy_sql}}
 
         return hook.insert_job(
             configuration=configuration,
@@ -319,11 +319,9 @@ def _submit_job(
         configuration = {
             "query": {
                 "query": self.sql,
-                "useLegacySql": False,
-            }
+                "useLegacySql": self.use_legacy_sql,
+            },
         }
-        if self.use_legacy_sql:
-            configuration["query"]["useLegacySql"] = self.use_legacy_sql
 
         return hook.insert_job(
             configuration=configuration,
@@ -453,7 +451,7 @@ def _submit_job(
         job_id: str,
     ) -> BigQueryJob:
         """Submit a new job and get the job id for polling the status using Triggerer."""
-        configuration = {"query": {"query": sql}}
+        configuration = {"query": {"query": sql, "useLegacySql": self.use_legacy_sql}}
         return hook.insert_job(
             configuration=configuration,
             project_id=hook.project_id,
@@ -577,7 +575,7 @@ def _submit_job(
         job_id: str,
     ) -> BigQueryJob:
         """Submit a new job and get the job id for polling the status using Trigger."""
-        configuration = {"query": {"query": self.sql}}
+        configuration = {"query": {"query": self.sql, "useLegacySql": self.use_legacy_sql}}
 
         return hook.insert_job(
             configuration=configuration,
@@ -689,7 +687,7 @@ def _submit_job(
         job_id: str,
     ) -> BigQueryJob:
         """Submit a new job and get the job id for polling the status using Trigger."""
-        configuration = {"query": {"query": self.sql}}
+        configuration = {"query": {"query": self.sql, "useLegacySql": self.use_legacy_sql}}
 
         return hook.insert_job(
             configuration=configuration,