Exclude partition from BigQuery table name

Please enter the commit message for your changes. Lines starting

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

@@ -2415,10 +2415,13 @@ def var_print(var_name):
             table_id = cmpt[1]
         else:
             raise ValueError(
-                f"{var_print(var_name)} Expect format of (<project.|<project:)<dataset>.<table>, "
+                f"{var_print(var_name)}Expect format of (<project.|<project:)<dataset>.<table>, "
                 f"got {table_input}"
             )
 
+        # Exclude partition from the table name
+        table_id = table_id.split("$")[0]
+
         if project_id is None:
             if var_name is not None:
                 self.log.info(
@@ -3330,6 +3333,9 @@ def var_print(var_name):
             f"{var_print(var_name)}Expect format of (<project.|<project:)<dataset>.<table>, got {table_input}"
         )
 
+    # Exclude partition from the table name
+    table_id = table_id.split("$")[0]
+
     if project_id is None:
         if var_name is not None:
             log.info(

tests/providers/google/cloud/hooks/test_bigquery.py

@@ -1030,12 +1030,69 @@ def test_query_results(self, _, selected_fields, result):
             == result
         )
 
+    def test_split_tablename_internal_need_default_project(self):
+        with pytest.raises(ValueError, match="INTERNAL: No default project is specified"):
+            self.hook.split_tablename("dataset.table", None)
+
+    @pytest.mark.parametrize("partition", ["$partition", ""])
+    @pytest.mark.parametrize(
+        "project_expected, dataset_expected, table_expected, table_input",
+        [
+            ("project", "dataset", "table", "dataset.table"),
+            ("alternative", "dataset", "table", "alternative:dataset.table"),
+            ("alternative", "dataset", "table", "alternative.dataset.table"),
+            ("alt1:alt", "dataset", "table", "alt1:alt.dataset.table"),
+            ("alt1:alt", "dataset", "table", "alt1:alt:dataset.table"),
+        ],
+    )
+    def test_split_tablename(
+        self, project_expected, dataset_expected, table_expected, table_input, partition
+    ):
+        default_project_id = "project"
+        project, dataset, table = self.hook.split_tablename(table_input + partition, default_project_id)
+        assert project_expected == project
+        assert dataset_expected == dataset
+        assert table_expected == table
+
+    @pytest.mark.parametrize(
+        "table_input, var_name, exception_message",
+        [
+            ("alt1:alt2:alt3:dataset.table", None, "Use either : or . to specify project got {}"),
+            (
+                "alt1.alt.dataset.table",
+                None,
+                r"Expect format of \(<project\.\|<project\:\)<dataset>\.<table>, got {}",
+            ),
+            (
+                "alt1:alt2:alt.dataset.table",
+                "var_x",
+                "Format exception for var_x: Use either : or . to specify project got {}",
+            ),
+            (
+                "alt1:alt2:alt:dataset.table",
+                "var_x",
+                "Format exception for var_x: Use either : or . to specify project got {}",
+            ),
+            (
+                "alt1.alt.dataset.table",
+                "var_x",
+                r"Format exception for var_x: Expect format of "
+                r"\(<project\.\|<project:\)<dataset>.<table>, got {}",
+            ),
+        ],
+    )
+    def test_split_tablename_invalid_syntax(self, table_input, var_name, exception_message):
+        default_project_id = "project"
+        with pytest.raises(ValueError, match=exception_message.format(table_input)):
+            self.hook.split_tablename(table_input, default_project_id, var_name)
+
 
 class TestBigQueryTableSplitter:
     def test_internal_need_default_project(self):
         with pytest.raises(ValueError, match="INTERNAL: No default project is specified"):
             split_tablename("dataset.table", None)
 
+    @pytest.mark.parametrize("partition", ["$partition", ""])
     @pytest.mark.parametrize(
         "project_expected, dataset_expected, table_expected, table_input",
         [
@@ -1046,9 +1103,11 @@ def test_internal_need_default_project(self):
             ("alt1:alt", "dataset", "table", "alt1:alt:dataset.table"),
         ],
     )
-    def test_split_tablename(self, project_expected, dataset_expected, table_expected, table_input):
+    def test_split_tablename(
+        self, project_expected, dataset_expected, table_expected, table_input, partition
+    ):
         default_project_id = "project"
-        project, dataset, table = split_tablename(table_input, default_project_id)
+        project, dataset, table = split_tablename(table_input + partition, default_project_id)
         assert project_expected == project
         assert dataset_expected == dataset
         assert table_expected == table