Skip to content

Commit

Permalink
Simpler repro of problem
Browse files Browse the repository at this point in the history
  • Loading branch information
damccorm committed Sep 26, 2024
1 parent 65e3825 commit 948eb33
Show file tree
Hide file tree
Showing 4 changed files with 18 additions and 4 deletions.
2 changes: 1 addition & 1 deletion .github/trigger_files/beam_PostCommit_Python.json
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to run.",
"modification": 2
"modification": 1
}

Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to run",
"modification": "1"
"comment": "Modify this file in a trivial way to cause this test suite to run"
}
1 change: 0 additions & 1 deletion sdks/python/apache_beam/io/gcp/bigquery_write_it_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -491,7 +491,6 @@ def test_big_query_write_insert_non_transient_api_call_error(self):
# pylint: disable=expression-not-assigned
errors = (
p | 'create' >> beam.Create(input_data)
| beam.WindowInto(beam.transforms.window.FixedWindows(10))
| 'write' >> beam.io.WriteToBigQuery(
table_id,
schema=table_schema,
Expand Down
16 changes: 16 additions & 0 deletions sdks/python/apache_beam/utils/windowed_value_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from parameterized import parameterized
from parameterized import parameterized_class

import apache_beam as beam
from apache_beam.utils import windowed_value
from apache_beam.utils.timestamp import Timestamp

Expand Down Expand Up @@ -75,6 +76,21 @@ def test_pickle(self):
wv = windowed_value.WindowedValue(1, 3, (), pane_info)
self.assertTrue(pickle.loads(pickle.dumps(wv)) == wv)

def test_encoding_global_window_in_interval_window(self):
input_data = ['123']

class ComputeWordLengthFn(beam.DoFn):
def process(self, element):
pass

def finish_bundle(self):
yield beam.transforms.window.GlobalWindows.windowed_value('test')

with beam.Pipeline() as p:
(p | 'create' >> beam.Create(input_data)
| beam.WindowInto(beam.transforms.window.FixedWindows(10))
| beam.ParDo(ComputeWordLengthFn()))


WINDOWED_BATCH_INSTANCES = [
windowed_value.HomogeneousWindowedBatch.of(
Expand Down

0 comments on commit 948eb33

Please sign in to comment.