[ 
https://issues.apache.org/jira/browse/BEAM-7499?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16858916#comment-16858916
 ] 

Tanay Tummalapalli edited comment on BEAM-7499 at 6/7/19 7:06 PM:
------------------------------------------------------------------

Hi [~lcwik],
Thanks for fixing the test failure. I found the following:

A. The last *BeamAssertExceptiom* is because the actual TestWindowedValue in 
the assert contains PaneInfo along with the window in a tuple. 
This is not the case with the current code in master[1]. I tested the specific 
tests and they work.  I'm not able to reproduce this.
Were any changes made related to ParDo evaluation or PaneInfo, or any PR that 
made use of/modified Reify? 


B. The *ValueError* is caused by `_IdentityWindowFn` that is used in 
`ReshufflePerKey`. 
It's docstring states that it will raise an exception when used after DoFns 
that return TimestampedValue elements. This seems to be caused by a change to 
the test - `test_no_window_context_fails`. This test is expected to raise a 
ValueError as in the master branch[2]. I am yet to investigate why 
_IdentityWindowFn raises an Exception in this case.

I need two more pieces to solve this mystery:
*  Was this caused by a PR?
*  Are the two tracebacks(ValueError, BeamAssertException) related? They seem 
to be caused by two different tests.

[1] 
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/transforms/util.py#L758
[2] 
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/transforms/util_test.py#L267



was (Author: ttanay):
Hi [~lcwik],
Thanks for fixing the test failure. I found the following:

A. The last *Failed Assert* is because the actual TestWindowedValue in the 
assert contains PaneInfo along with the window in a tuple. 
This is not the case with the current code in master[1]. I tested the specific 
tests and they work.  I'm not able to reproduce this.
Were any changes made related to ParDo evaluation or PaneInfo, or any PR that 
made use of/modified Reify? 


B. The ValueError is caused by `_IdentityWindowFn` that is used in 
`ReshufflePerKey`. 
It's docstring states that it will raise an exception when used after DoFns 
that return TimestampedValue elements. This seems to be caused by a change to 
the test - `test_no_window_context_fails`. This test is expected to raise a 
ValueError as in the master branch[2]. I am yet to investigate why 
_IdentityWindowFn raises an Exception in this case.

I need two more pieces to solve this mystery:
*  Was this caused by a PR?
*  Are the two tracebacks(ValueError, BeamAssertException) related? They seem 
to be caused by two different tests.

[1] 
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/transforms/util.py#L758
[2] 
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/transforms/util_test.py#L267


> ReifyTest.test_window fails in DirectRunner due to 'assign_context.window 
> should not be None.'
> ----------------------------------------------------------------------------------------------
>
>                 Key: BEAM-7499
>                 URL: https://issues.apache.org/jira/browse/BEAM-7499
>             Project: Beam
>          Issue Type: Improvement
>          Components: sdk-py-core
>            Reporter: Luke Cwik
>            Assignee: Pablo Estrada
>            Priority: Minor
>          Time Spent: 50m
>  Remaining Estimate: 0h
>
>  
> [PR 8717|https://github.com/apache/beam/pull/8717] added 
> ReifyWindow.test_window which fails on the DirectRunner.
> {code:java}
> ERROR:root:Exception at bundle 
> <apache_beam.runners.direct.bundle_factory._Bundle object at 0x7f087ab31248>, 
> due to an exception.
>  Traceback (most recent call last):
>  File "apache_beam/runners/direct/executor.py", line 343, in call
>  finish_state)
>  File "apache_beam/runners/direct/executor.py", line 380, in attempt_call
>  evaluator.process_element(value)
>  File "apache_beam/runners/direct/transform_evaluator.py", line 636, in 
> process_element
>  self.runner.process(element)
>  File "apache_beam/runners/common.py", line 780, in 
> apache_beam.runners.common.DoFnRunner.process
>  def process(self, windowed_value):
>  File "apache_beam/runners/common.py", line 784, in 
> apache_beam.runners.common.DoFnRunner.process
>  self._reraise_augmented(exn)
>  File "apache_beam/runners/common.py", line 851, in 
> apache_beam.runners.common.DoFnRunner._reraise_augmented
>  raise_with_traceback(new_exn)
>  File "apache_beam/runners/common.py", line 782, in 
> apache_beam.runners.common.DoFnRunner.process
>  return self.do_fn_invoker.invoke_process(windowed_value)
>  File "apache_beam/runners/common.py", line 453, in 
> apache_beam.runners.common.SimpleInvoker.invoke_process
>  output_processor.process_outputs(
>  File "apache_beam/runners/common.py", line 915, in 
> apache_beam.runners.common._OutputProcessor.process_outputs
>  self.window_fn.assign(assign_context))
>  File "apache_beam/transforms/util.py", line 557, in assign
>  'assign_context.window should not be None. '
> ValueError: assign_context.window should not be None. This might be due to a 
> DoFn returning a TimestampedValue. [while running 'add_timestamps2']
> Traceback (most recent call last):
>  File "apache_beam/transforms/util_test.py", line 501, in test_window
>  assert_that(reified_pc, equal_to(expected), reify_windows=True)
>  File "apache_beam/pipeline.py", line 426, in __exit__
>  self.run().wait_until_finish()
>  File "apache_beam/testing/test_pipeline.py", line 109, in run
>  state = result.wait_until_finish()
>  File "apache_beam/runners/direct/direct_runner.py", line 430, in 
> wait_until_finish
>  self._executor.await_completion()
>  File "apache_beam/runners/direct/executor.py", line 400, in await_completion
>  self._executor.await_completion()
>  File "apache_beam/runners/direct/executor.py", line 446, in await_completion
>  raise_(t, v, tb)
>  File "apache_beam/runners/direct/executor.py", line 343, in call
>  finish_state)
>  File "apache_beam/runners/direct/executor.py", line 380, in attempt_call
>  evaluator.process_element(value)
>  File "apache_beam/runners/direct/transform_evaluator.py", line 636, in 
> process_element
>  self.runner.process(element)
>  File "apache_beam/runners/common.py", line 780, in 
> apache_beam.runners.common.DoFnRunner.process
>  def process(self, windowed_value):
>  File "apache_beam/runners/common.py", line 784, in 
> apache_beam.runners.common.DoFnRunner.process
>  self._reraise_augmented(exn)
>  File "apache_beam/runners/common.py", line 851, in 
> apache_beam.runners.common.DoFnRunner._reraise_augmented
>  raise_with_traceback(new_exn)
>  File "apache_beam/runners/common.py", line 782, in 
> apache_beam.runners.common.DoFnRunner.process
>  return self.do_fn_invoker.invoke_process(windowed_value)
>  File "apache_beam/runners/common.py", line 454, in 
> apache_beam.runners.common.SimpleInvoker.invoke_process
>  windowed_value, self.process_method(windowed_value.value))
>  File "apache_beam/transforms/core.py", line 1292, in <lambda>
>  wrapper = lambda x: [fn(x)]
>  File "apache_beam/testing/util.py", line 129, in _equal
>  'Failed assert: %r == %r' % (sorted_expected, sorted_actual))
> BeamAssertException: Failed assert: [TestWindowedValue(value=('a', 100, 
> GlobalWindow), timestamp=100, windows=[GlobalWindow]), 
> TestWindowedValue(value=('b', 200, GlobalWindow), timestamp=200, 
> windows=[GlobalWindow]), TestWindowedValue(value=('c', 300, GlobalWindow), 
> timestamp=300, windows=[GlobalWindow])] == [TestWindowedValue(value=(('a', 
> 100.0, (GlobalWindow,), PaneInfo(first: True, last: True, timing: 3, index: 
> 0, nonspeculative_index: 0)), Timestamp(-9223372036854.775000), 
> GlobalWindow), timestamp=Timestamp(-9223372036854.775000), 
> windows=[GlobalWindow]), TestWindowedValue(value=(('c', 300.0, 
> (GlobalWindow,), PaneInfo(first: True, last: True, timing: 3, index: 0, 
> nonspeculative_index: 0)), Timestamp(-9223372036854.775000), GlobalWindow), 
> timestamp=Timestamp(-9223372036854.775000), windows=[GlobalWindow]), 
> TestWindowedValue(value=(('b', 200.0, (GlobalWindow,), PaneInfo(first: True, 
> last: True, timing: 3, index: 0, nonspeculative_index: 0)), 
> Timestamp(-9223372036854.775000), GlobalWindow), 
> timestamp=Timestamp(-9223372036854.775000), windows=[GlobalWindow])] [while 
> running 'assert_that/Match']
> {code}
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Reply via email to