Repository navigation
Trigger BQ CDC IT. Fix impl - #40618
Conversation
81992ec to
6096ef9
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40618 +/- ##
==========================================
- Coverage 80.04% 79.93% -0.12%
==========================================
Files 490 493 +3
Lines 82916 83441 +525
==========================================
+ Hits 66373 66695 +322
- Misses 16543 16746 +203
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
6096ef9 to
2001dad
Compare
2001dad to
cea6e65
Compare
|
R: @jrmccluskey |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
jrmccluskey
left a comment
There was a problem hiding this comment.
A few questions on the review; just out of curiosity, how did the zoneinfo change surface the lack of test execution/breakage here?
| # [start_time, now - buffer) in a single poll. | ||
| config = _PollConfig(start_time=start_time) | ||
|
|
||
| poll = beam.ParDo( |
There was a problem hiding this comment.
Is there a reason for the refactor of defining the ParDos prior to the pipeline definition both here and elsewhere?
There was a problem hiding this comment.
No reason, moved back inline.
| start_ts: Timestamp, | ||
| end_ts: Timestamp, | ||
| watermark_estimator: _PollWatermarkEstimator | ||
| ) -> Iterable[TimestampedValue[_QueryRange]]: |
There was a problem hiding this comment.
Were the typehints applied here incorrect?
There was a problem hiding this comment.
I think there's an issue with typehinting where if process is annotated to yield TimestampedValue[T] related to #20344
E.g.
from typing import Iterable
import apache_beam as beam
from apache_beam.transforms.window import TimestampedValue
class AddTimestamp(beam.DoFn):
def process(self, x: int) -> Iterable[TimestampedValue[int]]:
yield TimestampedValue(x, x)
with beam.Pipeline() as p:
ints = p | beam.Create([1, 2, 3]) | beam.ParDo(AddTimestamp())
print('element_type:', ints.element_type)
_ = ints | beam.Map(lambda x: x + 1).with_input_types(int)
results in
element_type: apache_beam.transforms.window.TimestampedValue[int]
TypeCheckError: The transform 'Map(<lambda ...>)' requires PCollections of type '<class 'int'>'
but was applied to a PCollection of type 'apache_beam.transforms.window.TimestampedValue[int]'
https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/08daa781654625f26eeacb10c5bb7adadddffc21/python/src/main/python/bigquery-anomaly-detection/src/bqmonitor/cdc.py is a fork of the bigquery_change_history. So the pyarrow change that broke cdc.py should have also turned these tests red if they were actually running in post commits. |
jrmccluskey
left a comment
There was a problem hiding this comment.
LGTM, just have to resolve merge issues in CHANGES.md
#40613 showed that this IO tests were not executing due to missing pytest mark.it_postcommit. Also Fixed failing tests when they actually execute.