Repository navigation
feat(bigtable): add protobuf decoding to Bigtable Change Streams to BigQuery - #3572
MattiasMTS wants to merge 8 commits into
Conversation
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request enhances the Bigtable Change Streams to BigQuery template by enabling the decoding of protobuf-encoded data directly within the pipeline. By providing a schema descriptor file, users can now automatically transform binary Bigtable cell values into readable JSON format in BigQuery. The implementation is designed to be backwards compatible and includes robust error handling to ensure data integrity when decoding fails or when specific columns do not require transformation. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
|
Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA). View this failed invocation of the CLA check for more information. For the most up to date status, view the checks section at the bottom of the pull request. |
0bb6d5e to
024da4e
Compare
|
/gemini review |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces support for decoding protobuf-encoded cell values and applying custom transformations to Bigtable change stream data before it is written to BigQuery. Key additions include a ProtoDecoder for schema-based JSON conversion and a ValueTransformerRegistry for handling specific data types like big-endian uint64 timestamps. The feedback identifies a potential issue in ValueTransformerRegistry.java where parsing configuration strings with split(":") may fail if column qualifiers contain colons, and suggests a more robust parsing approach using first and last index lookups.
| String[] parts = trimmed.split(":"); | ||
| if (parts.length != 3) { | ||
| throw new IllegalArgumentException( | ||
| "Invalid columnTransforms entry '" | ||
| + trimmed | ||
| + "'. Expected format: column_family:column:TRANSFORM_TYPE"); | ||
| } | ||
| String family = parts[0]; | ||
| String column = parts[1]; | ||
| String type = parts[2]; |
There was a problem hiding this comment.
Using split(":") to parse the configuration string is fragile because Bigtable column qualifiers can contain colons. Since the column_family name is restricted and cannot contain colons, and the TRANSFORM_TYPE is the last segment, it is safer to find the first and last colons to extract the components. This ensures that column qualifiers containing colons are handled correctly.
| String[] parts = trimmed.split(":"); | |
| if (parts.length != 3) { | |
| throw new IllegalArgumentException( | |
| "Invalid columnTransforms entry '" | |
| + trimmed | |
| + "'. Expected format: column_family:column:TRANSFORM_TYPE"); | |
| } | |
| String family = parts[0]; | |
| String column = parts[1]; | |
| String type = parts[2]; | |
| int firstColon = trimmed.indexOf(':'); | |
| int lastColon = trimmed.lastIndexOf(':'); | |
| if (firstColon == -1 || firstColon == lastColon) { | |
| throw new IllegalArgumentException( | |
| "Invalid columnTransforms entry '" | |
| + trimmed | |
| + "'. Expected format: column_family:column:TRANSFORM_TYPE"); | |
| } | |
| String family = trimmed.substring(0, firstColon); | |
| String column = trimmed.substring(firstColon + 1, lastColon); | |
| String type = trimmed.substring(lastColon + 1); |
|
Please also add new test to BigtableChangeStreamsToBigQueryIT |
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (55.04%) is below the target coverage (80.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## main #3572 +/- ##
============================================
+ Coverage 52.12% 53.16% +1.03%
- Complexity 5644 6089 +445
============================================
Files 1040 1078 +38
Lines 63118 65222 +2104
Branches 6922 7223 +301
============================================
+ Hits 32903 34674 +1771
- Misses 27981 28220 +239
- Partials 2234 2328 +94
🚀 New features to boost your workflow:
|
stankiewicz
left a comment
stankiewicz
left a comment
There was a problem hiding this comment.
great contribution! Few comments added
| @Default.String("") | ||
| String getFullProtoMessageName(); | ||
|
|
||
| void setFullProtoMessageName(String value); |
There was a problem hiding this comment.
thanks for being consistent with src/main/java/com/google/cloud/teleport/v2/templates/PubsubProtoToBigQuery.java
| + "(8-byte big-endian unsigned 64-bit integer as Unix epoch milliseconds, " | ||
| + "converted to a timestamp string).") | ||
| @Default.String("") | ||
| String getColumnTransforms(); |
There was a problem hiding this comment.
There is writeNumericTimestamps that converts int64 to timestamp for limited amount of columns. In your change, you are not only converting to some format (String in this case), but you also change endianness. Why not to timestamp type? why not to little endian uint? This looks like UDF to me with very limited scope that would require releasing a template whenever you want to add new transform.
Would JS UDF be good alternative?
| return null; | ||
| } | ||
| try { | ||
| long millis = ByteBuffer.wrap(bytes).getLong(); |
There was a problem hiding this comment.
add .order(ByteOrder.BIG_ENDIAN) to make it explicit and make code self documenting.
| /** | ||
| * Parses a comma-separated transform configuration string. | ||
| * | ||
| * @param config format: "family:column:TYPE,family2:column2:TYPE2" |
There was a problem hiding this comment.
TYPE is confusing, TRANSFORM_TYPE is used in other parts of javadoc.
| } | ||
|
|
||
| Map<String, Map<String, ValueTransformer>> transformersByFamily = new HashMap<>(); | ||
| for (String entry : config.split(",")) { |
There was a problem hiding this comment.
column name can be any value, can contain comma. Maybe state somewhere about such limitation.
| private static ValueTransformer createTransformer(String type) { | ||
| switch (type) { | ||
| case "BIG_ENDIAN_UINT64_TIMESTAMP_MS": | ||
| return new BigEndianTimestampTransformer(); | ||
| default: | ||
| throw new IllegalArgumentException( | ||
| "Unknown transform type '" | ||
| + type | ||
| + "'. Supported types: BIG_ENDIAN_UINT64_TIMESTAMP_MS"); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
while few transformers look small, it's cleaner and more efficient to reuse them. tranformers are stateless and thread safe (maybe worth writing this somewhere), you can implement memoization strategy:
private static final Map<String, ValueTransformer> TRANSFORMER_CACHE = new HashMap<>();
private static ValueTransformer createTransformer(String type) {
return TRANSFORMER_CACHE.computeIfAbsent(type, t -> {
switch (t) {
case "BIG_ENDIAN_UINT64_TIMESTAMP_MS":
return new BigEndianTimestampTransformer();
default:
throw new IllegalArgumentException("Unknown transform type '" + t + "'");
}
});
}
| boolean hasProtoColumn = !StringUtils.isBlank(options.getProtoColumn()); | ||
|
|
||
| if (hasProtoSchema || hasProtoMessage || hasProtoColumnFamily || hasProtoColumn) { | ||
| if (!hasProtoSchema || !hasProtoMessage || !hasProtoColumnFamily || !hasProtoColumn) { |
There was a problem hiding this comment.
difficult to read.. maybe:
boolean anyProtoSet = hasProtoSchema || hasProtoMessage || hasProtoColumnFamily || hasProtoColumn;
boolean allProtoSet = hasProtoSchema && hasProtoMessage && hasProtoColumnFamily && hasProtoColumn;
if (anyProtoSet && !allProtoSet) { throw new IllegalArgumentException( [..]
| "When using protobuf decoding, all of protoSchemaPath, fullProtoMessageName, " | ||
| + "protoColumnFamily, and protoColumn must be specified."); | ||
| } | ||
| protoDecoder = |
There was a problem hiding this comment.
ProtoDecoder protoDecoder = null;
if (allProtoSet) { protoDecoder = new ProtoDecoder( [..]
| @TemplateParameter.Text( | ||
| order = 14, | ||
| optional = true, | ||
| description = "Column qualifier containing proto values", |
There was a problem hiding this comment.
Grouping together multiple data into single proto column is a best practice, but it's not always the case, what if you have second proto in other column family or another column.
|
PubSub Proto to BQ has single proto so it's ok that config is only for single proto message. In BT there is a limit of 10 schema bundles per table, which suggest that there may be multiple proto types used. Eventually you may end up creating another transform_type: |
|
Let me know when it's ready. |
yes absolutely I'll let you know, boss! It's not ready yet. Sorry, been swamped in other stuff 😿 I'll try to focus getting this in shortly with high quality. |
e22bd73 to
c2626e7
Compare
|
@stankiewicz should be ready for review now. It grew a bit - let me know if you want me to split this into stacked PRs instead |
| static final String REASON = "reason"; | ||
| static final String REASON_VALUE = "decoded_value_exceeds_max_bytes"; | ||
|
|
||
| private static final ThreadLocal<Gson> GSON = ThreadLocal.withInitial(Gson::new); |
There was a problem hiding this comment.
Gson instances are natively thread-safe and intended to be shared concurrently across threads. Using ThreadLocal consumes unnecessary memory and initialization time per thread without any benefit.
why not static constant?
|
|
||
| if (maxBytes > 0 && rawBytes > maxBytes) { | ||
| return TransformResult.oversized(rawBytes, rawBytes); | ||
| } |
There was a problem hiding this comment.
just to note down on this cheap pre-check. if a protobuf payload contains a large number of unknown fields, those fields take up space in the rawBytes wire format but are subsequently dropped during JsonFormat.printer() rendering. Thus, a 10MB raw protobuf message might result in a 1MB JSON string. This pre-check would falsely reject the message if maxBytes was set to 8MB.
It is acceptable to leave it as-is, but the tradeoff should be explicitly understood.
|
/gemini review |
|
fix spotless :) |
There was a problem hiding this comment.
Code Review
This pull request introduces column value transformations, including BigEndian timestamps and Protobuf decoding, to the Bigtable Change Streams to BigQuery template. It also adds a mechanism to route oversized decoded values to a severe Dead Letter Queue to prevent pipeline stalls. The review feedback points out a serialization issue in ProtoDecodeTransformer that could cause a NullPointerException and suggests an optimization in Mod.java to avoid unnecessary memory allocations by checking raw value sizes before processing.
| private final String fullMessageName; | ||
| private final boolean preserveFieldNames; | ||
|
|
||
| private volatile boolean initialized = false; |
There was a problem hiding this comment.
The initialized flag should be marked as transient. Currently, because it is not transient, if the transformer instance is serialized and then deserialized (which can happen in a distributed Beam pipeline), initialized will remain true while the transient fields descriptor and printer will be reset to null. This would cause ensureInitialized() to skip re-initialization on the worker, leading to a NullPointerException when the transformer is used.
| private volatile boolean initialized = false; | |
| private transient volatile boolean initialized = false; |
| byte[] valueBytes = setCell.getValue().toByteArray(); | ||
| result = transformer.transformBounded(valueBytes, maxDecodedValueBytes); |
There was a problem hiding this comment.
To improve efficiency and avoid unnecessary memory allocation, consider checking if the raw cell value size already exceeds maxDecodedValueBytes before calling toByteArray(). Since maxDecodedValueBytes is the limit for the decoded JSON string, any raw value larger than this limit will certainly exceed it after transformation (as JSON is typically more verbose than the protobuf wire format).
if (maxDecodedValueBytes > 0 && setCell.getValue().size() > maxDecodedValueBytes) {
result = TransformResult.oversized(setCell.getValue().size(), setCell.getValue().size());
} else {
byte[] valueBytes = setCell.getValue().toByteArray();
result = transformer.transformBounded(valueBytes, maxDecodedValueBytes);
}
|
|
||
| /** Tests for {@link ValueTransformerRegistry#transformBounded}. */ | ||
| @RunWith(JUnit4.class) | ||
| public class ValueTransformerRegistryTest { |
There was a problem hiding this comment.
missing some negative tests
| if (type.startsWith("PROTO_DECODE(") && type.endsWith(")")) { | ||
| String messageName = type.substring("PROTO_DECODE(".length(), type.length() - 1); | ||
| if (messageName.isEmpty()) { |
There was a problem hiding this comment.
maybe regex?
Pattern.compile("PROTO_DECODE\\s*\\(\\s*(?<messageName>[^\\s\\)]+)\\s*\\)");
// Inside createTransformer method
Matcher matcher = PROTO_PATTERN.matcher(type.trim());
if (matcher.matches()) {
String messageName = matcher.group("messageName");```
| + "BIG_ENDIAN_TIMESTAMP (interprets 8-byte big-endian values as Unix epoch millis), " | ||
| + "PROTO_DECODE(package.MessageName) (decodes protobuf-encoded values to JSON; " | ||
| + "requires protoSchemaPath). For complex transformations, consider using a " | ||
| + "JavaScript UDF. Note that column qualifiers containing commas are not supported " |
There was a problem hiding this comment.
there is no Javascript UDF in this template.
| if (transformerRegistry == null) { | ||
| result = TransformResult.noTransformer(); | ||
| } else { | ||
| String qualifierStr = setCell.getQualifier().toStringUtf8(); | ||
| ValueTransformer transformer = | ||
| transformerRegistry.getTransformer(setCell.getFamilyName(), qualifierStr); | ||
| if (transformer == null) { | ||
| result = TransformResult.noTransformer(); | ||
| } else { | ||
| byte[] valueBytes = setCell.getValue().toByteArray(); | ||
| result = transformer.transformBounded(valueBytes, maxDecodedValueBytes); | ||
| } | ||
| } | ||
|
|
||
| if (result.status() == TransformResult.Status.OVERSIZED) { | ||
| return new SetCellBuildResult(null, result); | ||
| } |
There was a problem hiding this comment.
Determine TransformResult early with simplified logic
TransformResult result = TransformResult.noTransformer();
if (transformerRegistry != null) {
String qualifierStr = setCell.getQualifier().toStringUtf8();
ValueTransformer transformer =
transformerRegistry.getTransformer(setCell.getFamilyName(), qualifierStr);
if (transformer != null) {
byte[] valueBytes = setCell.getValue().toByteArray();
result = transformer.transformBounded(valueBytes, maxDecodedValueBytes);
}
}```
| // Exact for SUCCESS/OVERSIZED, 0 otherwise. | ||
| private final long decodedBytes; | ||
|
|
||
| private TransformResult(Status status, @Nullable String value, long rawBytes, long decodedBytes) { |
There was a problem hiding this comment.
naming - rawBytes -> rawBytesCount ? Length?
| TransformResult result = buildResult.transformResult(); | ||
| if (result.status() == TransformResult.Status.SUCCESS) { | ||
| decodedValueBytesDistribution.update(result.decodedBytes()); | ||
| } else if (result.status() == TransformResult.Status.OVERSIZED) { |
There was a problem hiding this comment.
what about. DECODE_ERROR
There was a problem hiding this comment.
The parser currently relies on string splitting. You need tests to ensure it doesn't crash or behave unpredictably when given:
- Missing Parts: e.g., family:qualifier (missing the transform type).
- Extra Colons: e.g., family:qualifier:type:extra.
- Empty Entries: e.g., family:qualifier:TYPE,,next:col:TYPE (double commas).
- Invalid Types: e.g., family:qualifier:NOT_A_REAL_TRANSFORM
There was a problem hiding this comment.
PROTO_DECODE missing tests - this is a complex transformation, it is the most likely to be misconfigured - make sure that if fails correctly for :
- Missing Message Name: PROTO_DECODE() or PROTO_DECODE.
- Invalid Message Name: PROTO_DECODE(nonexistent.Message).
- Schema Issues: Using PROTO_DECODE but providing a null or empty protoSchemaPath.
- Corrupt Descriptor: Testing what happens if the .pb file at the protoSchemaPath is malformed or not a valid FileDescriptorSet.
| @@ -33,6 +33,10 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat | |||
| * **bigQueryChangelogTablePartitionExpirationMs**: Sets the changelog table partition expiration time, in milliseconds. When set to `true`, partitions older than the specified number of milliseconds are deleted. By default, no expiration is set. | |||
| * **bigQueryChangelogTableFieldsToIgnore**: A comma-separated list of the changelog columns that, when specified, aren't created and populated. Use one of the following supported values: `is_gc`, `source_instance`, `source_cluster`, `source_table`, `tiebreaker`, or `big_query_commit_timestamp`. By default, all columns are populated. | |||
| * **dlqDirectory**: The directory to use for the dead-letter queue. Records that fail to be processed are stored in this directory. The default is a directory under the Dataflow job's temp location. In most cases, you can use the default path. | |||
| * **columnTransforms**: A comma-separated list of column value transforms. Each entry has the format column_family:column_qualifier:TRANSFORM_TYPE. Supported TRANSFORM_TYPE values: BIG_ENDIAN_TIMESTAMP (interprets 8-byte big-endian values as Unix epoch millis), PROTO_DECODE(package.MessageName) (decodes protobuf-encoded values to JSON; requires protoSchemaPath). For complex transformations, consider using a JavaScript UDF. Note that column qualifiers containing commas are not supported since comma is used as the entry delimiter. Defaults to empty. | |||
There was a problem hiding this comment.
JavaScript UDF is not available for this template
|
Hi @MattiasMTS, friendly ping. Thanks |
Adds protobuf decoding support to the Bigtable Change Streams to BigQuery
template. Cell values matching a registered transform are decoded into
JSON before being written to BigQuery.
- New columnTransforms option: comma-separated list of
family:qualifier:TRANSFORM_TYPE entries. Supported types:
- PROTO_DECODE(fully.qualified.MessageName): decodes proto bytes to
JSON via JsonFormat.printer, with optional preserveProtoFieldNames.
- BIG_ENDIAN_TIMESTAMP: decodes an 8-byte big-endian uint64 cell
value to an ISO-8601 timestamp string.
- protoSchemaPath: GCS path to a serialized FileDescriptorSet loaded by
ProtoDecodeTransformer on worker startup.
- Decoded output is capped at WINDMILL_ELEMENT_LIMIT_BYTES - 1 MiB
(~82.8 MiB) via Utf8BoundedAppendable, which counts exact UTF-8 bytes
and aborts mid-decode when the bound is crossed. Cells whose decoded
form would exceed the bound are routed to the existing severe DLQ as
OversizedValue records (serialized by OversizedValueDlqSanitizer) via
a new TupleTag side output, preventing Dataflow Streaming Engine
per-element commit failures on Bigtable cells approaching the 100 MB
cell limit.
- BigQuery-level size rejections (Storage Write API: 10 MB per row;
legacy batch load: 100 MB per row) continue to flow through the
existing writeResult.getFailedStorageApiInserts() path, sanitized by
BigQueryDeadLetterQueueSanitizer as today.
Unit tests for the new transformers, bounded-decode plumbing, and DLQ sanitizer; integration test covering the full pipeline with a proto schema and the oversized-value DLQ route.
- ProtoDecodeTransformer: mark `initialized` as transient so the initialized flag is reset after Beam serializes the DoFn for workers, preventing NPE when transient `descriptor`/`printer` would otherwise short-circuit through a stale `initialized=true`. - OversizedValueDlqSanitizer: replace the per-thread Gson with a single static instance; Gson is documented thread-safe. - Mod.buildSetCell: pre-check `setCell.getValue().size()` against the decoded-value budget before calling `toByteArray()`, avoiding a copy of multi-MB values that would be rejected anyway. The defensive pre-check inside ProtoDecodeTransformer.transformBounded is preserved as defense-in-depth and remains unit-tested. - BigtableChangeStreamsToBigQuery: emit a counter and warning log when a SET_CELL transform reports DECODE_ERROR so silent decode failures are observable; existing fallback to raw bytes is unchanged.
- Delete BigEndianTimestampTransformer and its dispatch case in ValueTransformerRegistry; the only supported transform is now PROTO_DECODE. - Switch PROTO_DECODE parsing to a compiled regex with a named capture group, and reject malformed PROTO_DECODE input (missing closing paren, blank message name, etc.) with a clear message. - Rename TransformResult.rawBytes -> rawBytesLength to make it obvious it is a count, not the original byte[]; update all callers including BigtableChangeStreamsToBigQuery and tests. - Flatten Mod.buildSetCell branching by initializing result to TransformResult.noTransformer() and only descending when a transformer is registered. - Drop the JS UDF reference and BIG_ENDIAN_TIMESTAMP entry from the columnTransforms helpText and the generated README parameter list. - Document on the ValueTransformer interface that implementations must be stateless and thread-safe. - Drop ValueTransformerRegistryTest BigEndian-specific tests and rewrite the non-proto OVERSIZED case to use a small inline ValueTransformer; drop the BigEndian column-transform IT.
- ValueTransformerRegistryTest: add parse-time negative tests for missing transform type, double-comma empty entries (skipped gracefully), unknown transform types, malformed PROTO_DECODE syntax (no parens, empty/blank message name) and missing protoSchemaPath. - ProtoDecodeTransformerTest: add negatives for nonexistent message name, malformed FileDescriptorSet bytes, and null/empty schema path. Validate non-null/non-empty constructor inputs in ProtoDecodeTransformer so the schema-path negative test fails fast at construction with a clear message instead of NPE during lazy init.
Transformer lookup, decode-error logs and oversized DLQ records now decode the qualifier and row key with bigtableChangeStreamCharset instead of always UTF-8, matching how the rest of the template renders them.
…heck - Add a registry test asserting that a trailing segment after the transform type is rejected as an unknown TRANSFORM_TYPE. - Rename the double-comma test to reflect that empty entries are skipped, not rejected. - Document that the raw-size pre-check counts unknown fields that the JSON printer drops, so such payloads can be rejected even when their JSON would fit.
c2626e7 to
dbcea39
Compare
Summary
Adds optional protobuf decoding to the Bigtable Change Streams to BigQuery template. Configured cells are decoded from protobuf wire format to JSON and written as a STRING to the BigQuery
valuecolumn.The Pub/Sub Proto to BigQuery template already supports this. Bigtable is a common store for protobuf data (see Query protobuf data), and a table can hold several message types, so this template needs per-column configuration.
New parameters (all optional, backwards compatible)
columnTransformscolumn_family:column_qualifier:TRANSFORM_TYPEentries. Supported:PROTO_DECODE(package.MessageName). Each column can use a different message type. Qualifiers may contain:, but not,.protoSchemaPathFileDescriptorSet(protoc --descriptor_set_out --include_imports). Required when anyPROTO_DECODEentry is used.preserveProtoFieldNamesmaxDecodedValueBytesExample:
columnTransforms=cf:profile:PROTO_DECODE(pkg.Profile),cf:event:PROTO_DECODE(pkg.Event)Behavior
PROTO_DECODE(...)and a missingprotoSchemaPathfail fast.maxDecodedValueBytes) go to the severe DLQ with row key, column, sizes and source info, instead of stalling the pipeline on a commit that is too large. Values whose raw protobuf size already exceeds the bound are rejected before decoding (unknown fields count toward raw size but are dropped from JSON).Appendablethat aborts mid-serialization once the budget is exceeded, so huge messages are never fully materialized.bigtableChangeStreamCharset, consistent with the rest of the template.PROTO_DECODE(...)types reuse one instance.Descriptor/JsonFormat.Printerare lazily initialized on workers.Reuses
SchemaUtils.getProtoDomain()fromv2/commonand theDynamicMessage+JsonFormat.Printerpattern fromPubsubProtoToBigQuery.Test plan
ValueTransformerRegistryTest,ProtoDecodeTransformerTest,Utf8BoundedAppendableTest,OversizedValueDlqSanitizerTest,BigQueryUtilTest(48 tests, including negative cases for parser and decoder misconfiguration)BigtableChangeStreamsToBigQueryIT: new cases for multi-columnPROTO_DECODEand oversized values routed to the DLQ (needs GCP resources, not run locally)mvn spotless:checkandcheckstyle:check