title: Apache Beam : WriteToBigQuery does not work if preceded by a stateful Transform unless re-windowing is applied
body: <p>I have a simple pipeline that does not work as expected, and any documentation I find does not explain its behaviour. In short, WriteToBigQuery in streaming mode fails if it is preceded by a stateful Transform, like GroupIntoBatches, unless global windows are re-applied before streaming it into BQ. Does anyone have a meaningful explanation?</p>
  <p>This does work:</p>
  <pre><code>    result = (
          p
          | &quot;Read Data&quot;
              &gt;&gt; beam.io.ReadFromPubSub(
                  subscription=&quot;projects/myproject/subscriptions/mysubscription&quot;,
                  with_attributes=False,
              ).with_output_types(bytes)
          | &quot;Decompose Data&quot; &gt;&gt; beam.ParDo(DecomposeData())
          | &quot;Window into Fixed Intervals&quot;
                  &gt;&gt; beam.WindowInto(window.FixedWindows(self.window_size))
          | &quot;Transform Data&quot; &gt;&gt; beam.ParDo(TransformData())
          | &quot;Write to BigQuery Table&quot; &gt;&gt; beam.io.WriteToBigQuery(
              table=_choose_table,
              schema=_choose_schema,
              create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))
  </code></pre>
  <p>This fails:</p>
  <pre><code>    result = (
          p
          | &quot;Read Data&quot;
              &gt;&gt; beam.io.ReadFromPubSub(
                  subscription=&quot;projects/myproject/subscriptions/mysubscription&quot;,
                  with_attributes=False,
              ).with_output_types(bytes)
          | &quot;Decompose Data&quot; &gt;&gt; beam.ParDo(DecomposeData())
          | &quot;Window into Fixed Intervals&quot;
                  &gt;&gt; beam.WindowInto(window.FixedWindows(self.window_size))
          | &quot;Add Dummy Key&quot; &gt;&gt; beam.Map(lambda elem: (None, elem))
          | &quot;Group into Batches&quot;
              &gt;&gt; beam.GroupIntoBatches(
                  max_buffering_duration_secs=self.window_size,
                  batch_size=self.batch_size,
              )
          | &quot;Discard Dummy Key&quot; &gt;&gt; beam.MapTuple(lambda _, val: val)**
          | &quot;Transform Data&quot; &gt;&gt; beam.ParDo(TransformData())
          | &quot;Write to BigQuery Table&quot; &gt;&gt; beam.io.WriteToBigQuery(
              table=_choose_table,
              schema=_choose_schema,
              create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))
  </code></pre>
  <p>This works again:</p>
  <pre><code>    result = (
          p
          | &quot;Read Data&quot;
              &gt;&gt; beam.io.ReadFromPubSub(
                  subscription=&quot;projects/myproject/subscriptions/mysubscription&quot;,
                  with_attributes=False,
              ).with_output_types(bytes)
          | &quot;Decompose Data&quot; &gt;&gt; beam.ParDo(DecomposeData())
          | &quot;Window into Fixed Intervals&quot;
                  &gt;&gt; beam.WindowInto(window.FixedWindows(self.window_size))
          | &quot;Add Dummy Key&quot; &gt;&gt; beam.Map(lambda elem: (None, elem))
          | &quot;Group into Batches&quot;
              &gt;&gt; beam.GroupIntoBatches(
                  max_buffering_duration_secs=self.window_size,
                  batch_size=self.batch_size,
              )
          | &quot;Discard Dummy Key&quot; &gt;&gt; beam.MapTuple(lambda _, val: val)
          | &quot;Transform Data&quot; &gt;&gt; beam.ParDo(TransformData())
          | &quot;Re-window&quot; &gt;&gt; beam.WindowInto(window.GlobalWindows())
          | &quot;Write to BigQuery Table&quot; &gt;&gt; beam.io.WriteToBigQuery(
              table=_choose_table,
              schema=_choose_schema,
              create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))
  </code></pre>