Add AirflowDeprecationWarning for deprecated project_id in BigQueryGetDataOperator

providers/google/docs/operators/cloud/bigquery.rst

@@ -198,6 +198,10 @@ To fetch data from a BigQuery table you can use
 Alternatively you can fetch data for selected columns if you pass fields to
 ``selected_fields``.
 
+.. note::
+   The ``project_id`` parameter is **deprecated** and will be removed in a future
+   release. Please use ``table_project_id`` instead.
+
 The result of this operator can be retrieved in two different formats based on the value of the ``as_dict`` parameter:
 ``False`` (default) - A Python list of lists, where the number of elements in the nesting list will be equal to the number of rows fetched. Each element in the
 nesting will a nested list where elements would represent the column values for

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

@@ -59,6 +59,7 @@
     BigQueryValueCheckTrigger,
 )
 from airflow.providers.google.cloud.utils.bigquery import convert_job_id
+from airflow.providers.google.common.deprecated import deprecated
 from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID
 from airflow.utils.helpers import exactly_one
 
@@ -1088,17 +1089,21 @@ def generate_query(self, hook: BigQueryHook) -> str:
         )
         return query
 
-    def execute(self, context: Context):
-        if self.project_id:
-            self.log.warning(
-                "The project_id parameter is deprecated, and will be removed in a future release."
-                " Please use table_project_id instead.",
-            )
-            if not self.table_project_id:
-                self.table_project_id = self.project_id
-            else:
-                self.log.info("Ignoring 'project_id' parameter, as 'table_project_id' is found.")
+    """Deprecated method to assign project_id to table_project_id."""
 
+    @deprecated(
+        planned_removal_date="June 30, 2026",
+        use_instead="table_project_id",
+        category=AirflowProviderDeprecationWarning,
+    )
+    def _assign_project_id(self, project_id: str) -> str:
+        return project_id
+
+    def execute(self, context: Context):
+        if self.project_id != PROVIDE_PROJECT_ID and not self.table_project_id:
+            self.table_project_id = self._assign_project_id(self.project_id)
+        elif self.project_id != PROVIDE_PROJECT_ID and self.table_project_id:
+            self.log.info("Ignoring 'project_id' parameter, as 'table_project_id' is found.")
         if not exactly_one(self.job_id, self.table_id):
             raise AirflowException(
                 "'job_id' and 'table_id' parameters are mutually exclusive, "