feat: add preview support for incremental results

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):