Always write to BQ from global window

.github/trigger_files/beam_PostCommit_Python.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
-  "modification": 2
+  "modification": 3
 }
 

.github/trigger_files/beam_PostCommit_Python_ValidatesContainer_Dataflow.json

@@ -1,3 +1,4 @@
 {
-    "comment": "Modify this file in a trivial way to cause this test suite to run"
+    "comment": "Modify this file in a trivial way to cause this test suite to run",
+    "modification": 1
 }

sdks/python/apache_beam/io/gcp/bigquery.py

@@ -1864,6 +1864,9 @@ def _restore_table_ref(sharded_table_ref_elems_kv):
     return (
         tagged_data
         | 'FromHashableTableRef' >> beam.Map(_restore_table_ref)
+        # Use global window for writes since we're outputting back into the
+        # global window.
+        | 'Window into Global Window' >> beam.WindowInto(GlobalWindows())
         | 'StreamInsertRows' >> ParDo(
             bigquery_write_fn, *self.schema_side_inputs).with_outputs(
                 BigQueryWriteFn.FAILED_ROWS,

sdks/python/apache_beam/io/gcp/bigquery_write_it_test.py

@@ -470,7 +470,7 @@ def test_big_query_write_insert_non_transient_api_call_error(self):
     input_data = [{
         'number': 1,
         'str': 'some_string',
-    }]
+    }]*500
 
     table_schema = {
         "fields": [{
@@ -483,7 +483,7 @@ def test_big_query_write_insert_non_transient_api_call_error(self):
     bq_result_errors = [({
         'number': 1,
         'str': 'some_string',
-    }, "Not Found")]
+    }, "Not Found")]*500
 
     args = self.test_pipeline.get_full_options_as_args()