feat: add preview support for incremental results (#2145)

* feat: add preview support for incremental results

Plumbs support to enable incremental results.

* fastpath allow

* add fastquery test

* lint

* lint

* blacken

Author: shollyman

google/cloud/bigquery/_job_helpers.py

@@ -564,6 +564,7 @@ def _supported_by_jobs_query(request_body: Dict[str, Any]) -> bool:
         "maximumBytesBilled",
         "requestId",
         "createSession",
+        "writeIncrementalResults",
     }
 
     unsupported_keys = request_keys - keys_allowlist

google/cloud/bigquery/job/query.py

@@ -674,6 +674,21 @@ def write_disposition(self):
     def write_disposition(self, value):
         self._set_sub_prop("writeDisposition", value)
 
+    @property
+    def write_incremental_results(self) -> Optional[bool]:
+        """This is only supported for a SELECT query using a temporary table.
+
+        If set, the query is allowed to write results incrementally to the temporary result
+        table. This may incur a performance penalty. This option cannot be used with Legacy SQL.
+
+        This feature is not generally available.
+        """
+        return self._get_sub_prop("writeIncrementalResults")
+
+    @write_incremental_results.setter
+    def write_incremental_results(self, value):
+        self._set_sub_prop("writeIncrementalResults", value)
+
     @property
     def table_definitions(self):
         """Dict[str, google.cloud.bigquery.external_config.ExternalConfig]:

tests/unit/job/test_query_config.py

@@ -167,6 +167,11 @@ def test_connection_properties(self):
         self.assertEqual(config.connection_properties[1].key, "time_zone")
         self.assertEqual(config.connection_properties[1].value, "America/Chicago")
 
+    def test_incremental_results(self):
+        config = self._get_target_class()()
+        config.write_incremental_results = True
+        self.assertEqual(config.write_incremental_results, True)
+
     def test_create_session(self):
         config = self._get_target_class()()
         self.assertIsNone(config.create_session)

tests/unit/test__job_helpers.py

@@ -194,6 +194,13 @@ def make_query_response(
             make_query_request({"maximumBytesBilled": "987654"}),
             id="job_config-with-maximum_bytes_billed",
         ),
+        pytest.param(
+            job_query.QueryJobConfig(
+                write_incremental_results=True,
+            ),
+            make_query_request({"writeIncrementalResults": True}),
+            id="job_config-with-incremental-results",
+        ),
     ),
 )
 def test__to_query_request(job_config, expected):
@@ -1141,6 +1148,11 @@ def test_make_job_id_w_job_id_overrides_prefix():
             False,
             id="priority=BATCH",
         ),
+        pytest.param(
+            job_query.QueryJobConfig(write_incremental_results=True),
+            True,
+            id="write_incremental_results",
+        ),
     ),
 )
 def test_supported_by_jobs_query_from_queryjobconfig(