Repository navigation
Conversation
|
Reminder, please take a look at this pr: @Abacn |
|
Load balancing out some reviews. R: @derrickaw |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
There was a problem hiding this comment.
Is there a missing integration yaml test that uses these changes?
|
|
||
| static final Schema CHANGE_STREAM_MUTATION_SCHEMA = | ||
| Schema.builder() | ||
| .addByteArrayField("row_key") |
There was a problem hiding this comment.
lets use "row" to match bigtablereadschematransformprovider
There was a problem hiding this comment.
Changed to "key" as in the BigtableReadSchemaTransformProvider
| Schema.builder() | ||
| .addStringField("type") | ||
| .addStringField("family_name") | ||
| .addNullableByteArrayField("qualifier") |
There was a problem hiding this comment.
lets use "column_qualifier" to match bigtablereadschematransformprovider
|
|
||
| @Override | ||
| public String description() { | ||
| return "Reads change stream records from a Google Cloud Bigtable table."; |
There was a problem hiding this comment.
can you provide an example usage? thanks
| .addValue(new Instant(mutation.getCommitTime().toEpochMilli())) | ||
| .addValue(mutation.getTieBreaker()) | ||
| .addValue(mutation.getToken()) | ||
| .addValue(new Instant(mutation.getEstimatedLowWatermarkTime().toEpochMilli())) | ||
| .addValue(entries) | ||
| .build(); |
There was a problem hiding this comment.
I don't think we use milliseconds. In the regular bigtablereadschematransformprovider, it uses micro
There was a problem hiding this comment.
These fields are Beam DATETIMEs, so they are represented as Joda Instant and use millisecond precision. The regular Bigtable provider uses microseconds for cell timestamps because those are exposed as raw INT64 fields (timestamp_micros). I’d prefer to keep commit_timestamp and estimated_low_watermark as typed timestamps rather than numeric micros.
There was a problem hiding this comment.
Can we add a mutationToRow test?
There was a problem hiding this comment.
Added a mutationToRow test covering the top-level mutation fields and nested entries.
|
R: @ahmedabu98, can you take a quick peek at this to see if anything else is missing? Thanks |
ahmedabu98
left a comment
ahmedabu98
left a comment
There was a problem hiding this comment.
Looks good from a skim, just consider adding a flattened output option.
Also please add a YAML IT
| .addNullableRowField("value_input", VALUE_SCHEMA) | ||
| .build(); | ||
|
|
||
| static final Schema CHANGE_STREAM_MUTATION_SCHEMA = |
There was a problem hiding this comment.
This output schema looks complex, which will be noticed when using it with YAML.
In the regular BigtableReadSchemaTransformProvider, we added an option to use a flattened schema (one output per entry, instead of per mutation). Consider doing something similar here
|
cc @mutianf might be relevant for you |
I tried adding a Bigtable CDC YAML IT using a temporary change-stream-enabled table and a bounded end time, but the pipeline consistently times out in the YAML Direct/FnApi integration-test harness. Do you have a recommendation for how this integration test should be structured for a streaming CDC source? |
4 participants
Adds YAML support for reading Google Cloud Bigtable change streams.
This PR:
Tested with:
Successful run: https://github.com/apache/beam/actions/runs/35729484761
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.