Add properties for new job statistics

bigquery/google/cloud/bigquery/job.py

@@ -1390,6 +1390,20 @@ def referenced_tables(self):
 
         return tables
 
+    @property
+    def num_dml_affected_rows(self):
+        """Return total bytes billed from job statistics, if present.
+
+        See:
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#statistics.query.numDmlAffectedRows
+
+        :rtype: int or None
+        :returns: number of DML rows affectd by the job, or None if job is not
+                  yet complete.
+        """
+        query_stats = self._query_statistics()
+        return query_stats.get('numDmlAffectedRows')
+
     def query_results(self):
         """Construct a QueryResults instance, bound to this job.
 

bigquery/tests/unit/test_job.py

@@ -1766,6 +1766,21 @@ def test_referenced_tables(self):
         self.assertIsInstance(remote._dataset._client, _Client)
         self.assertEqual(remote._dataset._client.project, 'other-project-123')
 
+    def test_num_dml_affected_rows(self):
+        num_rows = 1234
+        client = _Client(self.PROJECT)
+        job = self._make_one(self.JOB_NAME, self.QUERY, client)
+        self.assertIsNone(job.num_dml_affected_rows)
+
+        statistics = job._properties['statistics'] = {}
+        self.assertIsNone(job.num_dml_affected_rows)
+
+        query_stats = statistics['query'] = {}
+        self.assertIsNone(job.num_dml_affected_rows)
+
+        query_stats['numDmlAffectedRows'] = num_rows
+        self.assertEqual(job.num_dml_affected_rows, num_rows)
+
     def test_query_results(self):
         from google.cloud.bigquery.query import QueryResults