[AIRFLOW-2772] Fix Bug in BigQuery hook for Partitioned Table (#3901)

Author: xnuinside

airflow/contrib/hooks/bigquery_hook.py

@@ -562,9 +562,7 @@ def run_query(self,
             The default value is INTERACTIVE.
         :type priority: str
         :param time_partitioning: configure optional time partitioning fields i.e.
-            partition by field, type and
-            expiration as per API specifications. Note that 'field' is not available in
-            conjunction with dataset.table$partition.
+            partition by field, type and expiration as per API specifications.
         :type time_partitioning: dict
         :param cluster_fields: Request that the result of this query be stored sorted
             by one or more columns. This is only available in combination with
@@ -926,9 +924,7 @@ def run_load(self,
         :param src_fmt_configs: configure optional fields specific to the source format
         :type src_fmt_configs: dict
         :param time_partitioning: configure optional time partitioning fields i.e.
-            partition by field, type and
-            expiration as per API specifications. Note that 'field' is not available in
-            conjunction with dataset.table$partition.
+            partition by field, type and  expiration as per API specifications.
         :type time_partitioning: dict
         :param cluster_fields: Request that the result of this load be stored sorted
             by one or more columns. This is only available in combination with
@@ -1662,12 +1658,7 @@ def _cleanse_time_partitioning(destination_dataset_table, time_partitioning_in):
 
     time_partitioning_out = {}
     if destination_dataset_table and '$' in destination_dataset_table:
-        if time_partitioning_in.get('field'):
-            raise ValueError(
-                "Cannot specify field partition and partition name"
-                "(dataset.table$partition) at the same time")
         time_partitioning_out['type'] = 'DAY'
-
     time_partitioning_out.update(time_partitioning_in)
     return time_partitioning_out
 

airflow/contrib/operators/bigquery_operator.py

@@ -96,9 +96,7 @@ class BigQueryOperator(BaseOperator):
         The default value is INTERACTIVE.
     :type priority: str
     :param time_partitioning: configure optional time partitioning fields i.e.
-        partition by field, type and
-        expiration as per API specifications. Note that 'field' is not available in
-        conjunction with dataset.table$partition.
+        partition by field, type and expiration as per API specifications.
     :type time_partitioning: dict
     :param cluster_fields: Request that the result of this query be stored sorted
         by one or more columns. This is only available in conjunction with

tests/contrib/hooks/test_bigquery_hook.py

@@ -447,13 +447,6 @@ def test_extra_time_partitioning_options(self):
         }
         self.assertEqual(tp_out, expect)
 
-    def test_cant_add_dollar_and_field_name(self):
-        with self.assertRaises(ValueError):
-            _cleanse_time_partitioning(
-                'test.teast$20170101',
-                {'type': 'DAY', 'field': 'test_field', 'expirationMs': 1000}
-            )
-
 
 class TestClusteringInRunJob(unittest.TestCase):