feat: add created/started/ended properties to RowIterator. (#2260)

* feat: add created/started/ended attribute to RowIterator.

* fix annotation

* links update

* mypy fix

* Update google/cloud/bigquery/query.py

Co-authored-by: Tim Sweña (Swast) <[email protected]>

* Update google/cloud/bigquery/table.py

Co-authored-by: Tim Sweña (Swast) <[email protected]>

* Update google/cloud/bigquery/table.py

Co-authored-by: Tim Sweña (Swast) <[email protected]>

* Update google/cloud/bigquery/query.py

Co-authored-by: Tim Sweña (Swast) <[email protected]>

* Update google/cloud/bigquery/query.py

Co-authored-by: Tim Sweña (Swast) <[email protected]>

* Update google/cloud/bigquery/job/query.py

Co-authored-by: Tim Sweña (Swast) <[email protected]>

* fix unit test

---------

Co-authored-by: Tim Sweña (Swast) <[email protected]>
Co-authored-by: Lingqing Gan <[email protected]>

Author: 3 people

google/cloud/bigquery/_job_helpers.py

@@ -561,6 +561,9 @@ def do_query():
             query=query,
             total_bytes_processed=query_results.total_bytes_processed,
             slot_millis=query_results.slot_millis,
+            created=query_results.created,
+            started=query_results.started,
+            ended=query_results.ended,
         )
 
     if job_retry is not None:

google/cloud/bigquery/client.py

@@ -4145,6 +4145,9 @@ def _list_rows_from_query_results(
         query: Optional[str] = None,
         total_bytes_processed: Optional[int] = None,
         slot_millis: Optional[int] = None,
+        created: Optional[datetime.datetime] = None,
+        started: Optional[datetime.datetime] = None,
+        ended: Optional[datetime.datetime] = None,
     ) -> RowIterator:
         """List the rows of a completed query.
         See
@@ -4198,6 +4201,12 @@ def _list_rows_from_query_results(
                 total bytes processed from job statistics, if present.
             slot_millis (Optional[int]):
                 Number of slot ms the user is actually billed for.
+            created (Optional[datetime.datetime]):
+                Datetime at which the job was created.
+            started (Optional[datetime.datetime]):
+                Datetime at which the job was started.
+            ended (Optional[datetime.datetime]):
+                Datetime at which the job finished.
 
         Returns:
             google.cloud.bigquery.table.RowIterator:
@@ -4238,6 +4247,9 @@ def _list_rows_from_query_results(
             query=query,
             total_bytes_processed=total_bytes_processed,
             slot_millis=slot_millis,
+            created=created,
+            started=started,
+            ended=ended,
         )
         return row_iterator
 

google/cloud/bigquery/job/query.py

@@ -1767,6 +1767,9 @@ def is_job_done():
             query=self.query,
             total_bytes_processed=self.total_bytes_processed,
             slot_millis=self.slot_millis,
+            created=self.created,
+            started=self.started,
+            ended=self.ended,
             **list_rows_kwargs,
         )
         rows._preserve_order = _contains_order_by(self.query)

google/cloud/bigquery/query.py

@@ -1287,7 +1287,7 @@ def slot_millis(self):
         """Total number of slot ms the user is actually billed for.
 
         See:
-        https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query#body.QueryResponse.FIELDS.slot_millis
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query#body.QueryResponse.FIELDS.total_slot_ms
 
         Returns:
             Optional[int]: Count generated on the server (None until set by the server).
@@ -1310,6 +1310,56 @@ def num_dml_affected_rows(self):
         if num_dml_affected_rows is not None:
             return int(num_dml_affected_rows)
 
+    @property
+    def created(self):
+        """Creation time of this query.
+
+        See:
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query#body.QueryResponse.FIELDS.creation_time
+
+        Returns:
+            Optional[datetime.datetime]:
+                the creation time (None until set from the server).
+        """
+        millis = self._properties.get("creationTime")
+        if millis is not None:
+            return _helpers._datetime_from_microseconds(int(millis) * 1000.0)
+
+    @property
+    def started(self):
+        """Start time of this query.
+
+        This field will be present when the query transitions from the
+        PENDING state to either RUNNING or DONE.
+
+        See:
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query#body.QueryResponse.FIELDS.start_time
+
+        Returns:
+            Optional[datetime.datetime]:
+                the start time (None until set from the server).
+        """
+        millis = self._properties.get("startTime")
+        if millis is not None:
+            return _helpers._datetime_from_microseconds(int(millis) * 1000.0)
+
+    @property
+    def ended(self):
+        """End time of this query.
+
+        This field will be present whenever a query is in the DONE state.
+
+        See:
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query#body.QueryResponse.FIELDS.end_time
+
+        Returns:
+            Optional[datetime.datetime]:
+                the end time (None until set from the server).
+        """
+        millis = self._properties.get("endTime")
+        if millis is not None:
+            return _helpers._datetime_from_microseconds(int(millis) * 1000.0)
+
     @property
     def rows(self):
         """Query results.

google/cloud/bigquery/table.py

@@ -1788,7 +1788,15 @@ class RowIterator(HTTPIterator):
         query (Optional[str]):
             The query text used.
         total_bytes_processed (Optional[int]):
-            total bytes processed from job statistics, if present.
+            If representing query results, the total bytes processed by the associated query.
+        slot_millis (Optional[int]):
+            If representing query results, the number of slot ms billed for the associated query.
+        created (Optional[datetime.datetime]):
+            If representing query results, the creation time of the associated query.
+        started (Optional[datetime.datetime]):
+            If representing query results, the start time of the associated query.
+        ended (Optional[datetime.datetime]):
+            If representing query results, the end time of the associated query.
     """
 
     def __init__(
@@ -1813,6 +1821,9 @@ def __init__(
         query: Optional[str] = None,
         total_bytes_processed: Optional[int] = None,
         slot_millis: Optional[int] = None,
+        created: Optional[datetime.datetime] = None,
+        started: Optional[datetime.datetime] = None,
+        ended: Optional[datetime.datetime] = None,
     ):
         super(RowIterator, self).__init__(
             client,
@@ -1843,6 +1854,9 @@ def __init__(
         self._query = query
         self._total_bytes_processed = total_bytes_processed
         self._slot_millis = slot_millis
+        self._job_created = created
+        self._job_started = started
+        self._job_ended = ended
 
     @property
     def _billing_project(self) -> Optional[str]:
@@ -1905,6 +1919,21 @@ def slot_millis(self) -> Optional[int]:
         """Number of slot ms the user is actually billed for."""
         return self._slot_millis
 
+    @property
+    def created(self) -> Optional[datetime.datetime]:
+        """If representing query results, the creation time of the associated query."""
+        return self._job_created
+
+    @property
+    def started(self) -> Optional[datetime.datetime]:
+        """If representing query results, the start time of the associated query."""
+        return self._job_started
+
+    @property
+    def ended(self) -> Optional[datetime.datetime]:
+        """If representing query results, the end time of the associated query."""
+        return self._job_ended
+
     def _is_almost_completely_cached(self):
         """Check if all results are completely cached.
 

tests/unit/job/test_query.py

@@ -889,6 +889,9 @@ def test_result_reloads_job_state_until_done(self):
         job_resource_done = self._make_resource(started=True, ended=True, location="EU")
         job_resource_done["statistics"]["query"]["totalBytesProcessed"] = str(1234)
         job_resource_done["statistics"]["query"]["totalSlotMs"] = str(5678)
+        job_resource_done["statistics"]["creationTime"] = str(11)
+        job_resource_done["statistics"]["startTime"] = str(22)
+        job_resource_done["statistics"]["endTime"] = str(33)
         job_resource_done["configuration"]["query"]["destinationTable"] = {
             "projectId": "dest-project",
             "datasetId": "dest_dataset",
@@ -971,6 +974,9 @@ def test_result_reloads_job_state_until_done(self):
         self.assertEqual(result.query, job.query)
         self.assertEqual(result.total_bytes_processed, 1234)
         self.assertEqual(result.slot_millis, 5678)
+        self.assertEqual(result.created.timestamp() * 1000, 11)
+        self.assertEqual(result.started.timestamp() * 1000, 22)
+        self.assertEqual(result.ended.timestamp() * 1000, 33)
 
         query_results_path = f"/projects/{self.PROJECT}/queries/{self.JOB_ID}"
         query_results_call = mock.call(

tests/unit/test_client.py

@@ -5719,6 +5719,9 @@ def test_query_and_wait_defaults(self):
             "queryId": "job_abcDEF_",
             "totalBytesProcessed": 1234,
             "totalSlotMs": 5678,
+            "creationTime": "1437767599006",
+            "startTime": "1437767600007",
+            "endTime": "1437767601008",
         }
         creds = _make_credentials()
         http = object()
@@ -5737,6 +5740,9 @@ def test_query_and_wait_defaults(self):
         self.assertEqual(rows.query, query)
         self.assertEqual(rows.total_bytes_processed, 1234)
         self.assertEqual(rows.slot_millis, 5678)
+        self.assertEqual(rows.created.timestamp() * 1000, 1437767599006)
+        self.assertEqual(rows.started.timestamp() * 1000, 1437767600007)
+        self.assertEqual(rows.ended.timestamp() * 1000, 1437767601008)
 
         # Verify the request we send is to jobs.query.
         conn.api_request.assert_called_once()

tests/unit/test_query.py

@@ -2016,6 +2016,54 @@ def test_slot_millis_present_string(self):
         query = self._make_one(resource)
         self.assertEqual(query.slot_millis, 123456)
 
+    def test_created_missing(self):
+        query = self._make_one(self._make_resource())
+        self.assertIsNone(query.created)
+
+    def test_created_present_integer(self):
+        resource = self._make_resource()
+        resource["creationTime"] = 1437767599006
+        query = self._make_one(resource)
+        self.assertEqual(query.created.timestamp() * 1000, 1437767599006)
+
+    def test_created_present_string(self):
+        resource = self._make_resource()
+        resource["creationTime"] = "1437767599006"
+        query = self._make_one(resource)
+        self.assertEqual(query.created.timestamp() * 1000, 1437767599006)
+
+    def test_started_missing(self):
+        query = self._make_one(self._make_resource())
+        self.assertIsNone(query.started)
+
+    def test_started_present_integer(self):
+        resource = self._make_resource()
+        resource["startTime"] = 1437767599006
+        query = self._make_one(resource)
+        self.assertEqual(query.started.timestamp() * 1000, 1437767599006)
+
+    def test_started_present_string(self):
+        resource = self._make_resource()
+        resource["startTime"] = "1437767599006"
+        query = self._make_one(resource)
+        self.assertEqual(query.started.timestamp() * 1000, 1437767599006)
+
+    def test_ended_missing(self):
+        query = self._make_one(self._make_resource())
+        self.assertIsNone(query.ended)
+
+    def test_ended_present_integer(self):
+        resource = self._make_resource()
+        resource["endTime"] = 1437767599006
+        query = self._make_one(resource)
+        self.assertEqual(query.ended.timestamp() * 1000, 1437767599006)
+
+    def test_ended_present_string(self):
+        resource = self._make_resource()
+        resource["endTime"] = "1437767599006"
+        query = self._make_one(resource)
+        self.assertEqual(query.ended.timestamp() * 1000, 1437767599006)
+
     def test_num_dml_affected_rows_missing(self):
         query = self._make_one(self._make_resource())
         self.assertIsNone(query.num_dml_affected_rows)