feat(bigtable): add columnTransforms for generic cell value decoding

Add a columnTransforms parameter that maps column_family:column pairs
to value transformers. Supported types:
- BIG_ENDIAN_UINT64_TIMESTAMP_MS: 8-byte big-endian uint64 as timestamp

Example: columnTransforms=event_data:timestamp:BIG_ENDIAN_UINT64_TIMESTAMP_MS

This complements the proto decode feature — proto decode handles protobuf
columns while columnTransforms handles other binary encodings. Transform
priority: proto decode first, then columnTransforms.

Rename PROTO_DECODED_VALUE to TRANSFORMED_VALUE to reflect the generalized
transform pipeline.

Author: MattiasMTS

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/options/BigtableChangeStreamToBigQueryOptions.java

@@ -215,4 +215,19 @@ public interface BigtableChangeStreamToBigQueryOptions
   Boolean getPreserveProtoFieldNames();
 
   void setPreserveProtoFieldNames(Boolean value);
+
+  @TemplateParameter.Text(
+      order = 16,
+      optional = true,
+      description = "Column value transforms",
+      helpText =
+          "A comma-separated list of column value transformations in the format "
+              + "column_family:column:TRANSFORM_TYPE. Matched cell values are transformed "
+              + "before writing to BigQuery. Supported types: BIG_ENDIAN_UINT64_TIMESTAMP_MS "
+              + "(8-byte big-endian unsigned 64-bit integer as Unix epoch milliseconds, "
+              + "converted to a timestamp string).")
+  @Default.String("")
+  String getColumnTransforms();
+
+  void setColumnTransforms(String value);
 }

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstobigquery/BigtableChangeStreamsToBigQuery.java

@@ -37,6 +37,7 @@
 import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.ModType;
 import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.BigQueryUtils;
 import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.ProtoDecoder;
+import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.ValueTransformerRegistry;
 import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
 import com.google.cloud.teleport.v2.utils.BigtableSource;
 import java.util.ArrayList;
@@ -210,6 +211,13 @@ public static PipelineResult run(BigtableChangeStreamToBigQueryOptions options)
           options.getFullProtoMessageName());
     }
 
+    // Column value transforms (optional)
+    ValueTransformerRegistry transformerRegistry =
+        ValueTransformerRegistry.parse(options.getColumnTransforms());
+    if (transformerRegistry != null) {
+      LOG.info("Column transforms enabled: {}", options.getColumnTransforms());
+    }
+
     Pipeline pipeline = Pipeline.create(options);
     DeadLetterQueueManager dlqManager = buildDlqManager(options);
 
@@ -241,7 +249,9 @@ public static PipelineResult run(BigtableChangeStreamToBigQueryOptions options)
     PCollection<TableRow> changeStreamMutationToTableRow =
         dataChangeRecord.apply(
             "ChangeStreamMutation To TableRow",
-            ParDo.of(new ChangeStreamMutationToTableRowFn(sourceInfo, bigQuery, protoDecoder)));
+            ParDo.of(
+                new ChangeStreamMutationToTableRowFn(
+                    sourceInfo, bigQuery, protoDecoder, transformerRegistry)));
 
     Write<TableRow> bigQueryWrite =
         BigQueryIO.<TableRow>write()
@@ -353,12 +363,17 @@ static class ChangeStreamMutationToTableRowFn extends DoFn<ChangeStreamMutation,
     private final BigtableSource sourceInfo;
     private final BigQueryUtils bigQuery;
     private final ProtoDecoder protoDecoder;
+    private final ValueTransformerRegistry transformerRegistry;
 
     ChangeStreamMutationToTableRowFn(
-        BigtableSource source, BigQueryUtils bigQuery, ProtoDecoder protoDecoder) {
+        BigtableSource source,
+        BigQueryUtils bigQuery,
+        ProtoDecoder protoDecoder,
+        ValueTransformerRegistry transformerRegistry) {
       this.sourceInfo = source;
       this.bigQuery = bigQuery;
       this.protoDecoder = protoDecoder;
+      this.transformerRegistry = transformerRegistry;
     }
 
     @ProcessElement
@@ -371,8 +386,8 @@ public void process(@Element ChangeStreamMutation input, OutputReceiver<TableRow
         switch (modType) {
           case SET_CELL:
             mod =
-                protoDecoder != null
-                    ? new Mod(sourceInfo, input, (SetCell) entry, protoDecoder)
+                (protoDecoder != null || transformerRegistry != null)
+                    ? new Mod(sourceInfo, input, (SetCell) entry, protoDecoder, transformerRegistry)
                     : new Mod(sourceInfo, input, (SetCell) entry);
             break;
           case DELETE_CELLS:

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstobigquery/model/Mod.java

@@ -24,6 +24,7 @@
 import com.google.cloud.bigtable.data.v2.models.Range.BoundType;
 import com.google.cloud.bigtable.data.v2.models.SetCell;
 import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.ProtoDecoder;
+import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.ValueTransformerRegistry;
 import com.google.cloud.teleport.v2.utils.BigtableSource;
 import com.google.common.collect.Maps;
 import com.google.protobuf.ByteString;
@@ -50,7 +51,7 @@ public final class Mod implements Serializable {
 
   private static final String PATTERN_FORMAT = "yyyy-MM-dd HH:mm:ss.SSSSSS";
 
-  public static final String PROTO_DECODED_VALUE = "PROTO_DECODED_VALUE";
+  public static final String TRANSFORMED_VALUE = "TRANSFORMED_VALUE";
 
   private static final ThreadLocal<ObjectMapper> OBJECT_MAPPER =
       ThreadLocal.withInitial(ObjectMapper::new);
@@ -87,21 +88,38 @@ public Mod(
       ChangeStreamMutation mutation,
       SetCell setCell,
       ProtoDecoder protoDecoder) {
+    this(source, mutation, setCell, protoDecoder, null);
+  }
+
+  public Mod(
+      BigtableSource source,
+      ChangeStreamMutation mutation,
+      SetCell setCell,
+      ProtoDecoder protoDecoder,
+      ValueTransformerRegistry transformerRegistry) {
     this(mutation.getCommitTimestamp(), ModType.SET_CELL);
 
     Map<String, Object> propertiesMap = Maps.newHashMap();
     setCommonProperties(propertiesMap, source, mutation);
     setSpecificProperties(propertiesMap, setCell);
 
-    // Attempt proto decode if the column matches
-    if (protoDecoder != null) {
-      String qualifierStr = setCell.getQualifier().toStringUtf8();
-      if (protoDecoder.matches(setCell.getFamilyName(), qualifierStr)) {
-        String decodedJson = protoDecoder.decode(setCell.getValue().toByteArray());
-        if (decodedJson != null) {
-          propertiesMap.put(PROTO_DECODED_VALUE, decodedJson);
-        }
-      }
+    String qualifierStr = setCell.getQualifier().toStringUtf8();
+    String transformed = null;
+
+    // Try proto decode first
+    if (protoDecoder != null && protoDecoder.matches(setCell.getFamilyName(), qualifierStr)) {
+      transformed = protoDecoder.decode(setCell.getValue().toByteArray());
+    }
+
+    // Then try column transforms
+    if (transformed == null && transformerRegistry != null) {
+      transformed =
+          transformerRegistry.transform(
+              setCell.getFamilyName(), qualifierStr, setCell.getValue().toByteArray());
+    }
+
+    if (transformed != null) {
+      propertiesMap.put(TRANSFORMED_VALUE, transformed);
     }
 
     this.changeJson = convertPropertiesToJson(propertiesMap);

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstobigquery/schemautils/BigEndianTimestampTransformer.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (C) 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils;
+
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Transforms an 8-byte big-endian unsigned 64-bit integer representing Unix epoch milliseconds into
+ * a BigQuery-compatible timestamp string.
+ */
+public class BigEndianTimestampTransformer implements ValueTransformer {
+
+  private static final long serialVersionUID = 1L;
+  private static final Logger LOG = LoggerFactory.getLogger(BigEndianTimestampTransformer.class);
+  private static final DateTimeFormatter FORMATTER =
+      DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS").withZone(ZoneOffset.UTC);
+
+  @Override
+  public String transform(byte[] bytes) {
+    if (bytes == null || bytes.length != 8) {
+      LOG.warn(
+          "Expected 8 bytes for big-endian uint64 timestamp, got {}",
+          bytes == null ? "null" : bytes.length);
+      return null;
+    }
+    try {
+      long millis = ByteBuffer.wrap(bytes).getLong();
+      return FORMATTER.format(Instant.ofEpochMilli(millis));
+    } catch (Exception e) {
+      LOG.warn("Failed to decode big-endian timestamp: {}", e.getMessage());
+      return null;
+    }
+  }
+}

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstobigquery/schemautils/BigQueryUtils.java

@@ -94,9 +94,9 @@ public class BigQueryUtils implements Serializable {
     FORMATTERS.put(
         ChangelogColumn.VALUE_STRING,
         (bq, chg) -> {
-          // Use proto-decoded JSON if available
-          if (chg.has(Mod.PROTO_DECODED_VALUE)) {
-            return chg.getString(Mod.PROTO_DECODED_VALUE);
+          // Use transformed value (proto decode or column transform) if available
+          if (chg.has(Mod.TRANSFORMED_VALUE)) {
+            return chg.getString(Mod.TRANSFORMED_VALUE);
           }
           if (!chg.has(ChangelogColumn.VALUE_BYTES.name())) {
             return null;

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstobigquery/schemautils/ValueTransformer.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright (C) 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils;
+
+import java.io.Serializable;
+
+/** Transforms a Bigtable cell value from raw bytes to a string representation for BigQuery. */
+public interface ValueTransformer extends Serializable {
+
+  /**
+   * Transforms raw cell value bytes to a string for BigQuery.
+   *
+   * @param bytes the raw cell value bytes
+   * @return the transformed string, or null if transformation fails
+   */
+  String transform(byte[] bytes);
+}

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstobigquery/schemautils/ValueTransformerRegistry.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright (C) 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Registry that maps column family:column qualifier pairs to {@link ValueTransformer} instances.
+ *
+ * <p>Parses a comma-separated configuration string with format: {@code
+ * column_family:column:TRANSFORM_TYPE}
+ *
+ * <p>Supported transform types:
+ *
+ * <ul>
+ *   <li>{@code BIG_ENDIAN_UINT64_TIMESTAMP_MS} - 8-byte big-endian uint64 Unix milliseconds
+ * </ul>
+ */
+public class ValueTransformerRegistry implements Serializable {
+
+  private static final long serialVersionUID = 1L;
+
+  private final Map<String, ValueTransformer> transformers;
+
+  private ValueTransformerRegistry(Map<String, ValueTransformer> transformers) {
+    this.transformers = transformers;
+  }
+
+  /**
+   * Parses a comma-separated transform configuration string.
+   *
+   * @param config format: "family:column:TYPE,family2:column2:TYPE2"
+   * @return a new registry, or null if config is null or empty
+   * @throws IllegalArgumentException if the config format is invalid or a transform type is unknown
+   */
+  public static ValueTransformerRegistry parse(String config) {
+    if (config == null || config.isEmpty()) {
+      return null;
+    }
+
+    Map<String, ValueTransformer> transformers = new HashMap<>();
+    for (String entry : config.split(",")) {
+      String trimmed = entry.trim();
+      if (trimmed.isEmpty()) {
+        continue;
+      }
+      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];
+      String key = family + ":" + column;
+
+      ValueTransformer transformer = createTransformer(type);
+      transformers.put(key, transformer);
+    }
+    return new ValueTransformerRegistry(transformers);
+  }
+
+  /**
+   * Transforms a cell value if a transformer is registered for the given column.
+   *
+   * @return the transformed string, or null if no transformer matches or transformation fails
+   */
+  public String transform(String family, String column, byte[] bytes) {
+    ValueTransformer transformer = transformers.get(family + ":" + column);
+    if (transformer == null) {
+      return null;
+    }
+    return transformer.transform(bytes);
+  }
+
+  /** Returns true if a transformer is registered for the given column. */
+  public boolean hasTransformer(String family, String column) {
+    return transformers.containsKey(family + ":" + column);
+  }
+
+  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");
+    }
+  }
+}