diff --git a/docs/index.md b/docs/index.md
index 2b097bf8b..1d90d4ee7 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -48,6 +48,10 @@ To launch reverse replication, refer details [here](./reverse-replication/Revers
 
 To find out how to monitor your migration, refer [here](./monitoring/MonitoringUserGuide.md).
 
+### Custom transformation
+
+To find out how to configure custom transformations, refer [here](./transformations/CustomTransformation.md)
+
 ## Supported Sources and Targets
 
 - **Schema Migrations**: SMT supports schema migrations for MySQL, PostgreSQL, SQLServer and Oracle.
diff --git a/docs/reverse-replication/ReverseReplicationUserGuide.md b/docs/reverse-replication/ReverseReplicationUserGuide.md
index dbf0a3a3a..57ca1cbb3 100644
--- a/docs/reverse-replication/ReverseReplicationUserGuide.md
+++ b/docs/reverse-replication/ReverseReplicationUserGuide.md
@@ -113,6 +113,9 @@ The Dataflow job that writes to source database exposes the following per shard
 | metadata_file_create_lag_retry_\<logical shard name\> | Count of file lookup retries done when the job that writes to GCS is lagging |
 | mySQL_retry_\<logical shard name\> | Number of retries done when MySQL is not reachable|
 | shard_failed_\<logical shard name\> | Published when there is a failure while processing the shard |
+| custom_transformation_exception | Number of exception encountered in the custom transformation jar |
+| filtered_events_\<logical shard name\> | Number of events filtered via custom transformation per shard |
+| apply_custom_transformation_impl_latency_ms | Time taken for the execution of custom transformation logic. |
 
 These can be used to track the pipeline progress.
 However, there is a limit of 100 on the total number of metrics per project. So if this limit is exhausted, the Dataflow job will give a message like so:
@@ -378,10 +381,13 @@ In order to make it easier for users to customize the shard routing logic, the [
 Steps to perfrom customization:
 1. Write custom shard id fetcher logic [CustomShardIdFetcher.java](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-custom-shard/src/main/java/com/custom/CustomShardIdFetcher.java). Details of the ShardIdRequest class can be found [here](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-migrations-sdk/src/main/java/com/google/cloud/teleport/v2/spanner/utils/ShardIdRequest.java).
 2. Build the [JAR](https://github.com/GoogleCloudPlatform/DataflowTemplates/tree/main/v2/spanner-custom-shard) and upload the jar to GCS
-3. Invoke the reverse replication flow by passing the [custom jar path and custom class path](RunnigReverseReplication.md#custom-jar).
+3. Invoke the reverse replication flow by passing the [custom jar path and custom class path](RunnigReverseReplication.md#custom-shard-identification).
 4. If any custom parameters are needed in the custom shard identification logic, they can be passed via the *readerShardingCustomParameters* input to the runner. These parameters will be passed to the *init* method of the custom class. The *init* method is invoked once per worker setup.
 
-
+### Custom transformations
+For cases where a user wants to handle a custom transformation logic, they need to specify the following parameters in the [GCS to Sourcedb](https://github.com/GoogleCloudPlatform/DataflowTemplates/tree/main/v2/gcs-to-sourcedb) template - a GCS path that points to a custom jar, fully classified custom class name of the class containing custom transformation logic and custom parameters which might be used by the jar to invoke custom logic to perform transformation.
+ 
+For details on how to implement and use custom transformations, please refer to the [Custom Transformation](../transformations/CustomTransformation.md) section.
 
 ## Cost
 
diff --git a/docs/reverse-replication/RunnigReverseReplication.md b/docs/reverse-replication/RunnigReverseReplication.md
index 4174b6637..2dedb936b 100644
--- a/docs/reverse-replication/RunnigReverseReplication.md
+++ b/docs/reverse-replication/RunnigReverseReplication.md
@@ -69,6 +69,10 @@ The script takes in multiple arguments to orchestrate the pipeline. They are:
 - `vpcHostProjectId`: project ID hosting the subnetwork. If unspecified, the 'projectId' parameter value will be used for subnetwork.
 - `vpcNetwork`: name of the VPC network to be used for the dataflow jobs
 - `vpcSubnetwork`: name of the VPC subnetwork to be used for the dataflow jobs. Subnet should exist in the same region as the 'dataflowRegion' parameter.
+- `writeFilteredEventsToGcs`: Whether to write filtered events to GCS. Default is false.
+- `writerTransformationCustomJarPath`: The GCS path to custom jar for custom transformation logic.
+- `writerTransformationCustomClassName`: The fully qualified custom class name for custom transformation logic.
+- `writerTransformationCustomParameters`: Any custom parameters to be supplied to custom transformation class.
 
 
 ## Pre-requisites
@@ -139,9 +143,9 @@ Launched writer job:  id:"<>" project_id:"<>" name:"<>" current_state_time:{} cr
 
 ```
 
-### Custom Jar
+### Custom shard identification
 
-In order to specify custom shard identification function, custom jar and class names need to give. The command to do that is below:
+In order to specify custom shard identification function, custom jar and class names need to be given. The command to do that is below:
 
 ```
 go run reverse-replication-runner.go -projectId=<project-id> -dataflowRegion=<region> -instanceId=<spanner-instance> -dbName=<spanner-database> -sourceShardsFilePath=gs://bucket-name/shards.json -sessionFilePath=gs://bucket-name/session.json -gcsPath=gs://bucket-name/<directory> -readerShardingCustomJarPath=gs://bucket-name/custom.jar -readerShardingCustomClassName=com.custom.classname -readerShardingCustomParameters='a=b,c=d'
@@ -154,6 +158,26 @@ gcloud dataflow flex-template run smt-reverse-replication-reader-2024-01-05t10-3
 ```
 
 
+### Custom transformation
+
+In order to specify custom transformation, custom jar and class names need to be given. The command to do that is below:
+
+```
+go run reverse-replication-runner.go -projectId=<project-id> -dataflowRegion=<region> -instanceId=<spanner-instance> -dbName=<spanner-database> -sourceShardsFilePath=gs://bucket-name/shards.json -sessionFilePath=gs://bucket-name/session.json -gcsPath=gs://bucket-name/<directory> -writerTransformationCustomJarPath=gs://bucket-name/custom.jar -writerTransformationCustomClassName=com.custom.classname -writerTransformationCustomParameters='a=b,c=d'
+``` 
+
+If a user wants to write the records filtered via custom transformation to GCS, they can use the command below:
+```
+go run reverse-replication-runner.go -projectId=<project-id> -dataflowRegion=<region> -instanceId=<spanner-instance> -dbName=<spanner-database> -sourceShardsFilePath=gs://bucket-name/shards.json -sessionFilePath=gs://bucket-name/session.json -gcsPath=gs://bucket-name/<directory> -writerTransformationCustomJarPath=gs://bucket-name/custom.jar -writerTransformationCustomClassName=com.custom.classname -writerTransformationCustomParameters='a=b,c=d' -writeFilteredEventsToGcs
+```
+
+The sample reader job gcloud command for the same
+
+```
+gcloud dataflow flex-template run smt-reverse-replication-writer-2024-04-17t08-00-37z-8dac-d95f --project=<project> --region=<region> --template-file-gcs-location=<template location> --parameters sessionFilePath=<session path>,sourceDbTimezoneOffset=+00:00,spannerProjectId=span-cloud-testing,runMode=regular,sourceShardsFilePath=<shard file path>,metadataTableSuffix=,GCSInputDirectoryPath=<gcs path>,metadataInstance=<spanner instance>,metadataDatabase=<spanner metadata database>,runIdentifier=2024-04-17t08-00-37z,transformationJarPath=<jar path>,transformationClassName=<custom class name>,transformationCustomParameters=a=b,c=d,writeFilteredEventsToGcs=true --num-workers=5 --worker-machine-type=n2-standard-4 --additional-experiments=use_runner_v2
+```
+
+
 ### Network and Subnetwork specification
 
 If the dataflow workers need to be run on a different network and subnetwork with custom network tags, sample command is below.
diff --git a/docs/transformations/CustomTransformation.md b/docs/transformations/CustomTransformation.md
new file mode 100644
index 000000000..a83b8451f
--- /dev/null
+++ b/docs/transformations/CustomTransformation.md
@@ -0,0 +1,212 @@
+---
+layout: default
+title: Custom Transformations
+nav_order: 15
+permalink: /custom-transformation
+---
+
+# Custom transformation
+{: .no_toc }
+Users can follow this guide to apply custom transformations at the row level to all columns in their schema, including primary key columns. This guide can be used to specify custom transformation logic for existing columns, to populate new columns in Spanner that didn't exist in the source, or to filter a row during migration. For ease of use, we have implemented the [CustomTransformationFetcher](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-custom-shard/src/main/java/com/custom/CustomTransformationFetcher.java) class in `v2/spanner-custom-shard`, which can be updated according to the user's logic.
+
+<details open markdown="block">
+  <summary>
+    Table of contents
+  </summary>
+  {: .text-delta }
+1. TOC
+{:toc}
+</details>
+
+## Custom transformation workflow
+
+![](https://services.google.com/fh/files/misc/transformation-workflow.png)
+
+Points to note regarding the workflow:
+- The workflow will execute once per event and will also be triggered during retries.
+- Not all column values need to be returned; only the returned columns will be updated. The rest will be migrated as they are from the source.
+- Users can add new columns in the response that are not present in the source but exist in Spanner. These column values will be written to Spanner, and the data types of the returned values must be compatible with the Spanner schema.
+
+## Methods in CustomTransformationFetcher
+
+- **init()** - This is an initialization method that will be called once during the pipeline setup. It is used to initialize the custom jar with custom parameters.  
+- **toSpannerRow()** - This method applies custom transformations to the incoming source record and is expected to return a subset of spanner row.
+- **toSourceRow()** - This method applies custom transformations to the incoming spanner record and is expected to return a subset of source row.
+
+## Parameter details
+
+### Forward migration
+
+#### Request
+[MigrationTransformationRequest](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-migrations-sdk/src/main/java/com/google/cloud/teleport/v2/spanner/utils/MigrationTransformationRequest.java) contains the following information - 
+- tableName - Name of the source table to which the event belongs to.
+- shardId - Logical shard id of the record.
+- eventType - The event type can either be INSERT, UPDATE-INSERT, UPDATE, UPDATE_DELETE or DELETE. Please refer to the [datastream documentation](https://cloud.google.com/datastream/docs/events-and-streams) for more details.
+- requestRow - It is a map of type `Map<java.lang.String, java.lang.Object>` where key is the source column name and value is source column value.
+
+The following table outlines the Java object type that will be sent in the request to the `toSpannerRow` for each **MySQL** datatype:
+
+| MYSQL datatype | Java object type    |
+|----------------|---------------------|
+| BIGINT         | Long                |
+| BINARY         | String (hex encoded)|
+| BIT            | Long                |
+| BLOB           | String (hex encoded)|
+| BOOLEAN        | Long                |
+| CHAR           | String              |
+| DATE           | String (Format: yyyy-MM-dd) |
+| DATETIME       | String (UTC format) |
+| DECIMAL        | String              |
+| DOUBLE         | Double              |
+| ENUM           | String              |
+| FLOAT          | Double              |
+| INT            | Long                |
+| JSON           | String              |
+| LONGBLOB       | String (hex encoded)|
+| LONGTEXT       | String              |
+| MEDIUMBLOB     | String (hex encoded)|
+| MEDIUMINT      | Long                |
+| MEDIUMTEXT     | String              |
+| SET            | String              |
+| SMALLINT       | Long                |
+| TEXT           | String              |
+| TIME           | Long ([time-micros](https://avro.apache.org/docs/current/specification/_print/#time-microsecond-precision) e.g., 45296000000 for 12:34:56) |
+| TIMESTAMP      | String (UTC format) |
+| TINYBLOB       | String (hex encoded)|
+| TINYINT        | Long                |
+| TINYTEXT       | String              |
+| VARBINARY      | String (hex encoded)|
+| VARCHAR        | String              |
+
+{: .highlight }
+Datastream does not support [spatial data type](https://dev.mysql.com/doc/refman/8.0/en/spatial-type-overview.html) columns. Values in the request for these columns will be NULL.
+
+#### Response
+[MigrationTransformationResponse](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-migrations-sdk/src/main/java/com/google/cloud/teleport/v2/spanner/utils/MigrationTransformationResponse.java) contains the following information - 
+- responseRow - It is a map of type `Map<java.lang.String, java.lang.Object>` where key is the spanner column name and value is spanner column value.
+- isEventFiltered - If set to true, event will be skipped and not written to spanner.
+
+{: .highlight }
+Values in the response row must be compatible with their corresponding Spanner column types.
+
+The following table outlines the recommended Java object types for each Spanner data type(GSQL dialect and PostgreSQL dialect) to ensure successful data insertion:
+
+| Spanner datatype (GSQL dialect) | Spanner datatype (PostgreSQL dialect) | Java object type                                |
+|---------------------------------|---------------------------------------|-------------------------------------------------|
+| INT64                           | bigint                                | Long                                            |
+| FLOAT32                         | real                                  | Double                                          |
+| FLOAT64                         | double precision                      | Double                                          |
+| NUMERIC                         | numeric                               | String                                          |
+| BOOL                            | boolean                               | Boolean                                         |
+| STRING                          | text/character varying                | String                                          |
+| BYTES                           | bytea                                 | Hex Encoded string                              |
+| DATE                            | date                                  | String (Format: yyyy-MM-dd)                     |
+| TIMESTAMP                       | timestamp with time zone              | String (in UTC format, e.g., 2023-05-23T12:34:56Z) |
+| JSON                            | jsonb                                 | String                                          |
+
+{: .highlight }
+Please refer to the sample implementation of **toSpannerRow** for most MySQL datatype columns [here](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-custom-shard/src/main/java/com/custom/CustomTransformationWithShardForIT.java#L44).
+
+### Reverse replication
+
+#### Request
+[MigrationTransformationRequest](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-migrations-sdk/src/main/java/com/google/cloud/teleport/v2/spanner/utils/MigrationTransformationRequest.java) contains the following information - 
+- tableName - Name of the spanner table to which the event belongs to.
+- shardId - Logical shard id of the record.
+- eventType - The event type can either be INSERT, UPDATE or DELETE
+- requestRow - It is a map of type `Map<java.lang.String, java.lang.Object>` where key is the spanner column name and value is spanner column value.
+
+The following table outlines the Java object type that will be sent in the request to the `toSourceRow` for each **Spanner** datatype:
+
+| Spanner datatype | Java object type                    |
+|-----------------|--------------------------------------|
+| INT64           | String                               |
+| FLOAT32         | BigDecimal                           |
+| FLOAT64         | BigDecimal                           |
+| NUMERIC         | String                               |
+| BOOL            | String (e.g., "false")               |
+| STRING          | String                               |
+| BYTES           | String (Base64 encoded)              |
+| DATE            | String (Format: yyyy-MM-dd)          |
+| TIMESTAMP       | String (UTC format)                  |
+| JSON            | String                               |                                                   
+
+#### Response
+[MigrationTransformationResponse](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-migrations-sdk/src/main/java/com/google/cloud/teleport/v2/spanner/utils/MigrationTransformationResponse.java) contains the following information - 
+- responseRow - It is a map of type `Map<java.lang.String, java.lang.Object>` where key is the source column name and value is source column value.
+- isEventFiltered - If set to true, event will be skipped and not written to source.
+
+{: .highlight }
+Values in the response row should be exactly in the format compatible with source schema, which means users also need to enclose string values in **single quotes** as they would normally do in an INSERT statement.
+
+The following table outlines format of the response string for each **MySQL** datatype to ensure successful data insertion:
+
+| Response format                                                                    | MYSQL datatype                                                                   |
+|-------------------------------------------------------------------------------------|----------------------------------------------------------------------------------|
+| String                                                                              | TINYINT, INT, BIGINT, YEAR, SMALLINT                                            |
+| String (Format: Time value **enclosed in single quotes**, e.g., '14:30:00')         | TIME                                                                             |
+| String                                                                              | FLOAT, DOUBLE, DECIMAL                                                           |
+| String                                                                              | BOOLEAN                                                                          |
+| String (**enclosed in single quotes**, e.g., 'Transformed text')                    | TEXT, ENUM, JSON, CHAR, LONGTEXT, MEDIUMTEXT, SET, TINYTEXT, VARCHAR             |
+| Binary String                                                                       | BLOB, BINARY, BIT, LONGBLOB, MEDIUMBLOB, TINYBLOB, VARBINARY                     |
+| String (Format: yyyy-MM-dd **enclosed in single quotes**, e.g., '1995-01-13')       | DATE                                                                             |
+| String (UTC format,e.g. 2024-01-01T12:34:56Z)                                       | DATETIME, TIMESTAMP                                                              |
+| String (**enclosed in single quotes**)                                              | [Spatial Datatypes](https://dev.mysql.com/doc/refman/8.0/en/spatial-type-overview.html) |
+
+{: .highlight }
+Please refer to the sample implementation of **toSourceRow** for most MySQL datatype columns [here](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-custom-shard/src/main/java/com/custom/CustomTransformationWithShardForIT.java#L145).
+
+## Steps to implement custom transformation
+1. Checkout the dataflow code from [github](https://github.com/GoogleCloudPlatform/DataflowTemplates)
+    ```
+    git clone https://github.com/GoogleCloudPlatform/DataflowTemplates.git
+    ```
+2. Update the logic in [CustomTransformationFetcher](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-custom-shard/src/main/java/com/custom/CustomTransformationFetcher.java) class in `v2/spanner-custom-shard`. In case of only forward migration only updating the toSpannerRow is sufficient and for toSourceRow just use the below sample:
+    ```
+    @Override
+    public MigrationTransformationResponse toSourceRow(MigrationTransformationRequest request throws InvalidTransformationException {
+    return new MigrationTransformationResponse(null, false);
+    }
+    ```
+
+    Similarly, in case of only reverse replication updating the toSourceRow is sufficient and for toSpannerRow just use the below sample:
+    ```
+    @Override
+    public MigrationTransformationResponse toSpannerRow(MigrationTransformationRequest request throws InvalidTransformationException {
+    return new MigrationTransformationResponse(null, false);
+    }
+    ```
+3. If any custom parameters are needed in the custom transformation logic, they can be passed to the *init* method of the custom class. The *init* method is invoked once per worker setup.
+3. Please test the modified code by writing unit and cross functional tests in [CustomTransformationFetcherTest.java](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/spanner-custom-shard/src/test/java/com/custom/CustomTransformationFetcherTest.java) 
+4. Build the `spanner-custom-shard` module by running the below commands:
+    ```
+    cd v2/spanner-custom-shard
+    mvn install
+    ```
+5. Upload the built JAR located in v2/spanner-custom-shard/target with the name `spanner-custom-shard-1.0-SNAPSHOT.jar` to a GCS bucket
+
+## Error handling
+
+### Forward migration
+- If the value of a column, received as a response from the custom JAR, cannot be successfully parsed into the corresponding Spanner column datatype, pipeline will deem the record as failed and label it as a SEVERE error in the DLQ.
+- If the custom JAR sends a NULL response for a NOT NULL column, the record will encounter a failure during insertion and will be labelled as a SEVERE error in the DLQ.
+- If the custom JAR returns an exception while processing a record then pipeline will deem the record as failed, label it as a SEVERE error in the DLQ and will increment the `Custom Transformation Exceptions` metric.
+<!---
+[TODO]: Once the fix to ignore extra columns in response during forward migration is released, update the documentation.
+-->
+- If the custom JAR returns an extra column in response which is not present in the spanner schema then the record will be labelled as SEVERE error and moved to DLQ.
+
+### Reverse replication
+- If the value of a column, received as a response from the custom JAR, cannot be successfully inserted into the source database then the reverse replication pipeline will stop processing records for the particular shard.
+- If the custom JAR returns a NULL value for a NOT NULL column, the record will encounter a failure during insertion and will stop processing records for the particular shard.
+- If the custom JAR returns an exception while processing a record then pipeline will deem the record as failed, will stop processing records for the particular shard and will increment the `custom_transformation_exception` metric.
+- If the custom JAR returns an extra column in response which is not present in the spanner schema then the extra column will be ignored.
+
+
+## Best practices
+- Avoid time-consuming operations in the custom JAR, as they can slow down the pipeline since it is executed at a per-record level.
+- Ensure idempotency and account for retries.
+- Be cautious with logging in the **toSourceRow/toSpannerRow** methods, as logging per row can cause throttling.
+- Throw an **InvalidTransformationException** if an error occurs while processing a particular event in the custom JAR.
+
+
diff --git a/docs/troubleshoot/minimal.md b/docs/troubleshoot/minimal.md
index 48ce80831..31d4f3c78 100644
--- a/docs/troubleshoot/minimal.md
+++ b/docs/troubleshoot/minimal.md
@@ -50,29 +50,38 @@ Migration progress can be tracked by monitoring the Dataflow job and following c
 
 ### Metrics for regular run
 
-| Metric Name                           | Description                                                                                                                      |
-|---------------------------------------|----------------------------------------------------------------------------------------------------------------------------------|
-| Successful events                     | Total number of events successfully processed and applied to Spanner database                                                    |
-| Retryable errors                      | The count of events that were errored out but will be retried                                                                    |
-| Total permanent errors                | The number of events that are errored out with non-retriable errors in addition to the number of errors after exhausting retries |
-| Conversion errors                     | Number of events that could not be converted to Spanner.This is a permanent error category.                                      |
-| Skipped events                        | The events that are skipped from migration since the table was dropped from migration                                            |
-| Other permanent errors                | The remaining permanent errors.                                                                                                  |
-| Total events processed                | The number of events that were tried for forward migration, including retries and permanent errors.                              |
-| elementsReconsumedFromDeadLetterQueue | The total number of events consumed from DLQ for retry                                                                           |
+| Metric Name                                   | Description                                                                                                                      |
+|-----------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------|
+| Successful events                             | Total number of events successfully processed and applied to Spanner database                                                    |
+| Retryable errors                              | The count of events that were errored out but will be retried                                                                    |
+| Total permanent errors                        | The number of events that are errored out with non-retriable errors in addition to the number of errors after exhausting retries |
+| Conversion errors                             | Number of events that could not be converted to Spanner. This is a permanent error category.                                      |
+| Skipped events                                | The events that are skipped from migration since the table was dropped from migration                                            |
+| Other permanent errors                        | The remaining permanent errors.                                                                                                  |
+| Transformed events                            | The number of events that were successfully transformed, including retries and permanent errors.                                 |
+| Filtered events                               | The number of events that were were filtered as a part of custom transformation.                                                 |
+| Custom Transformation Exceptions              | The number of events that were errored out due to some exception in custom transformation jar.                                   |
+| Total events processed                        | The number of events that were tried for forward migration, including retries and permanent errors.                              |
+| apply_custom_transformation_impl_latency_ms   | Latency of applying custom transformation to the event.                                                                          |
+| elementsReconsumedFromDeadLetterQueue         | The total number of events consumed from DLQ for retry.                                                                          |
 
 
 ### Metrics for retryDLQ run
 
-| Metric Name                           | Description                                                                                         |
-|---------------------------------------|-----------------------------------------------------------------------------------------------------|
-| Successful events                     | Total number of events successfully processed and applied to Spanner database                       |
-| elementsReconsumedFromDeadLetterQueue | The total number of events consumed from DLQ for retry                                              |
-| Elements requeued for retry           | The total number of events that were re queued for retry                                            |
-| Conversion errors                     | Number of events that could not be converted to Spanner.This is a permanent error category.         |
-| Skipped events                        | The events that are skipped from migration since the table was dropped from migration               |
-| Other permanent errors                | The remaining permanent errors.                                                                     |
-| Total events processed                | The number of events that were tried for forward migration, including retries and permanent errors. |
+| Metric Name                                   | Description                                                                                         |
+|-----------------------------------------------|-----------------------------------------------------------------------------------------------------|
+| Successful events                             | Total number of events successfully processed and applied to Spanner database                       |
+| elementsReconsumedFromDeadLetterQueue         | The total number of events consumed from DLQ for retry                                              |
+| Elements requeued for retry                   | The total number of events that were re queued for retry                                            |
+| Conversion errors                             | Number of events that could not be converted to Spanner.This is a permanent error category.         |
+| Skipped events                                | The events that are skipped from migration since the table was dropped from migration               |
+| Other permanent errors                        | The remaining permanent errors.                                                                     |
+| Total events processed                        | The number of events that were tried for forward migration, including retries and permanent errors. |
+| Transformed events                            | The number of events that were successfully transformed, including retries and permanent errors.    |
+| Filtered events                               | The number of events that were were filtered as a part of custom transformation.                    |
+| Custom Transformation Exceptions              | The number of events that were errored out due to some exception in custom transformation jar.      |
+| Total events processed                        | The number of events that were tried for forward migration, including retries and permanent errors. |
+| apply_custom_transformation_impl_latency_ms   | Latency of applying custom transformation to the event.                                             |
 
 It can happen that in retryDLQ mode, there are still permanent errors. To identify that all the retryable errors have been processed and only permanent errors remain for reprocessing - one can look at the ‘Successful events' count - it would remain constant after every retry iteration. Each retry iteration, the ‘elementsReconsumedFromDeadLetterQueue' would increment.
 
diff --git a/docs/ui/prepare-migration/dataflow-tuning.md b/docs/ui/prepare-migration/dataflow-tuning.md
index 29c46338d..23ed6e820 100644
--- a/docs/ui/prepare-migration/dataflow-tuning.md
+++ b/docs/ui/prepare-migration/dataflow-tuning.md
@@ -23,13 +23,14 @@ Some use cases when the user would want to tweak the jobs are:
 - Dataflow and Spanner should run in separate projects for cost tracking.
 - Use a custom service account to launch the Dataflow job.
 - Apply labels for better cost tracking for the jobs.
+- Apply custom transformations to the records.
 
 
 {: .highlight }
 
 To tune dataflow, first specify the target database in the 'Configure Spanner Database' step. This enables the configure button for the remaining steps.
 
-![](https://services.google.com/fh/gumdrop/preview/misc/dataflow-form.png)
+![](https://services.google.com/fh/files/misc/dataflow-tuning.png)
 
 <details open markdown="block">
   <summary>
@@ -44,6 +45,35 @@ To tune dataflow, first specify the target database in the 'Configure Spanner Da
 
 SMT exposes the most frequently changed dataflow configurations to the user. Please reach out to us if you have a use-case that is not satisfied by the provided configurations.
 
+### Custom transformations
+For cases where a user wants to handle a custom transformation logic, they need to specify the following parameters in the [dataflow](https://github.com/GoogleCloudPlatform/DataflowTemplates/tree/main/v2/datastream-to-spanner) template from the SMT UI - a GCS path that points to a custom jar, fully classified custom class name of the class containing custom transformation logic and custom parameters which might be used by the jar to invoke custom logic to perform transformation.
+
+The records processed through custom transformation will be written to the `filteredEvents/` directory in the destination connection profile's GCS bucket. 
+
+#### Custom JAR GCS Path
+Specify the GCS path of the jar containing custom transformation logic. For custom transformation, specify both custom jar GCS path and fully classified class name. If no custom jar and class name are provided, only the default transformations will be used.
+
+Present under the Custom Transformations section of the form.
+
+{: .highlight }
+
+Please make sure that the custom JAR code is idempotent to manage transaction retries effectively.
+
+Refer to [Custom transformations](../../transformations/CustomTransformation.md) for more details.
+
+#### Custom Class Name
+Specify the fully classified class name of the class containing custom transformation logic (e.g.: `com.custom.CustomTransformationFetcher.java`). For custom transformation, specify both custom jar GCS path and fully classified class name. If no custom jar and class name are provided, only the default transformations will be used.
+
+Present under the Custom Transformations section of the form.
+
+{: .highlight }
+Specify both the custom class name and custom jar GCS path, or specify neither.
+
+#### Custom Parameter
+Specify the custom parameters to be passed to the custom transformation logic implementation.
+
+Present under the Custom Transformations section of the form.
+
 ### VPC Host ProjectId
 Specify the project id of the VPC that you want to use. This is required in order to use private connectivity. By default, this is assumed to be the same as Spanner project. Ensure this is specified if also specifying a network and subnetwork.
 
diff --git a/docs/ui/schema-conv/spanner-draft.md b/docs/ui/schema-conv/spanner-draft.md
index f47f1ef3e..43335b464 100644
--- a/docs/ui/schema-conv/spanner-draft.md
+++ b/docs/ui/schema-conv/spanner-draft.md
@@ -30,6 +30,16 @@ Column tab provides information on the columns that are a part of the selected t
 
 ![](https://services.google.com/fh/files/misc/column-info-edit.png)
 
+#### Add Column
+
+In addition to editing the existing columns in the Spanner draft mapped from the source database, users can also add new columns to the selected table.
+
+![](https://services.google.com/fh/files/misc/add_column.png)
+![](https://services.google.com/fh/files/misc/add_column_form.png)
+![](https://services.google.com/fh/files/misc/new_column.png)
+
+To specify custom transformation logic in the dataflow pipeline to populate these columns, please refer to the [Custom Transformation](../../transformations/CustomTransformation.md) section.
+
 ### Primary Key
 
 Users can view and edit the primary key of a table from the primary key tab. They can remove/add a column from the primary key or change the order of columns in the primary key. Once these changes are made, the session file is updated and they can also be verified from the [SQL tab](#sql).  
diff --git a/reverse_replication/reverse-replication-runner.go b/reverse_replication/reverse-replication-runner.go
index d8ec40112..1de1ddfb0 100644
--- a/reverse_replication/reverse-replication-runner.go
+++ b/reverse_replication/reverse-replication-runner.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"strconv"
 	"strings"
 	"time"
 
@@ -20,44 +21,48 @@ import (
 )
 
 var (
-	projectId                      string
-	dataflowRegion                 string
-	jobNamePrefix                  string
-	changeStreamName               string
-	instanceId                     string
-	dbName                         string
-	metadataInstance               string
-	metadataDatabase               string
-	startTimestamp                 string
-	sourceShardsFilePath           string
-	sessionFilePath                string
-	machineType                    string
-	vpcNetwork                     string
-	vpcSubnetwork                  string
-	vpcHostProjectId               string
-	serviceAccountEmail            string
-	readerWorkers                  int
-	writerWorkers                  int
-	windowDuration                 string
-	gcsPath                        string
-	filtrationMode                 string
-	metadataTableSuffix            string
-	sourceDbTimezoneOffset         string
-	writerRunMode                  string
-	readerRunMode                  string
-	readerShardingCustomJarPath    string
-	readerShardingCustomClassName  string
-	readerShardingCustomParameters string
-	readerSkipDirectoryName        string
-	spannerReaderTemplateLocation  string
-	sourceWriterTemplateLocation   string
-	jobsToLaunch                   string
-	skipChangeStreamCreation       bool
-	skipMetadataDatabaseCreation   bool
-	networkTags                    string
-	runIdentifier                  string
-	readerMaxWorkers               int
-	spannerProjectId               string
+	projectId                            string
+	dataflowRegion                       string
+	jobNamePrefix                        string
+	changeStreamName                     string
+	instanceId                           string
+	dbName                               string
+	metadataInstance                     string
+	metadataDatabase                     string
+	startTimestamp                       string
+	sourceShardsFilePath                 string
+	sessionFilePath                      string
+	machineType                          string
+	vpcNetwork                           string
+	vpcSubnetwork                        string
+	vpcHostProjectId                     string
+	serviceAccountEmail                  string
+	readerWorkers                        int
+	writerWorkers                        int
+	windowDuration                       string
+	gcsPath                              string
+	filtrationMode                       string
+	metadataTableSuffix                  string
+	sourceDbTimezoneOffset               string
+	writerRunMode                        string
+	readerRunMode                        string
+	readerShardingCustomJarPath          string
+	readerShardingCustomClassName        string
+	readerShardingCustomParameters       string
+	readerSkipDirectoryName              string
+	spannerReaderTemplateLocation        string
+	sourceWriterTemplateLocation         string
+	jobsToLaunch                         string
+	skipChangeStreamCreation             bool
+	skipMetadataDatabaseCreation         bool
+	networkTags                          string
+	runIdentifier                        string
+	readerMaxWorkers                     int
+	spannerProjectId                     string
+	writeFilteredEventsToGcs             bool
+	writerTransformationCustomJarPath    string
+	writerTransformationCustomClassName  string
+	writerTransformationCustomParameters string
 )
 
 const (
@@ -103,7 +108,10 @@ func setupGlobalFlags() {
 	flag.StringVar(&readerShardingCustomParameters, "readerShardingCustomParameters", "", "Any custom parameters to be supplied to custom sharding class.")
 	flag.IntVar(&readerMaxWorkers, "readerMaxWorkers", 20, "Number of max workers for reader job.")
 	flag.StringVar(&spannerProjectId, "spannerProjectId", "", "The project id where Cloud Spanner resides, for use case when Cloud Spanner is in a different project than where Dataflow would run.")
-
+	flag.BoolVar(&writeFilteredEventsToGcs, "writeFilteredEventsToGcs", false, "Whether to write filtered events to GCS. Default is false.")
+	flag.StringVar(&writerTransformationCustomJarPath, "writerTransformationCustomJarPath", "", "The GCS path to custom jar for custom transformation logic.")
+	flag.StringVar(&writerTransformationCustomClassName, "writerTransformationCustomClassName", "", "The fully qualified custom class name for custom transformation logic.")
+	flag.StringVar(&writerTransformationCustomParameters, "writerTransformationCustomParameters", "", "Any custom parameters to be supplied to custom transformation class.")
 }
 
 func prechecks() error {
@@ -354,21 +362,30 @@ func main() {
 			additionalExpr = []string{"use_network_tags=" + networkTags, "use_network_tags_for_flex_templates=" + networkTags}
 		}
 
+		writerParams := map[string]string{
+			"sourceShardsFilePath":   sourceShardsFilePath,
+			"sessionFilePath":        sessionFilePath,
+			"sourceDbTimezoneOffset": sourceDbTimezoneOffset,
+			"metadataTableSuffix":    metadataTableSuffix,
+			"GCSInputDirectoryPath":  gcsPath,
+			"spannerProjectId":       spannerProjectId,
+			"metadataInstance":       metadataInstance,
+			"metadataDatabase":       metadataDatabase,
+			"runMode":                writerRunMode,
+			"runIdentifier":          runId,
+		}
+
+		if writerTransformationCustomJarPath != "" {
+			writerParams["transformationJarPath"] = writerTransformationCustomJarPath
+			writerParams["transformationClassName"] = writerTransformationCustomClassName
+			writerParams["transformationCustomParameters"] = writerTransformationCustomParameters
+			writerParams["writeFilteredEventsToGcs"] = strconv.FormatBool(writeFilteredEventsToGcs)
+		}
+
 		launchParameters := &dataflowpb.LaunchFlexTemplateParameter{
-			JobName:  fmt.Sprintf("%s-writer-%s-%s", jobNamePrefix, runId, utils.GenerateHashStr()),
-			Template: &dataflowpb.LaunchFlexTemplateParameter_ContainerSpecGcsPath{ContainerSpecGcsPath: sourceWriterTemplateLocation},
-			Parameters: map[string]string{
-				"sourceShardsFilePath":   sourceShardsFilePath,
-				"sessionFilePath":        sessionFilePath,
-				"sourceDbTimezoneOffset": sourceDbTimezoneOffset,
-				"metadataTableSuffix":    metadataTableSuffix,
-				"GCSInputDirectoryPath":  gcsPath,
-				"spannerProjectId":       spannerProjectId,
-				"metadataInstance":       metadataInstance,
-				"metadataDatabase":       metadataDatabase,
-				"runMode":                writerRunMode,
-				"runIdentifier":          runId,
-			},
+			JobName:    fmt.Sprintf("%s-writer-%s-%s", jobNamePrefix, runId, utils.GenerateHashStr()),
+			Template:   &dataflowpb.LaunchFlexTemplateParameter_ContainerSpecGcsPath{ContainerSpecGcsPath: sourceWriterTemplateLocation},
+			Parameters: writerParams,
 			Environment: &dataflowpb.FlexTemplateRuntimeEnvironment{
 				NumWorkers:            int32(writerWorkers),
 				AdditionalExperiments: additionalExpr,
diff --git a/streaming/streaming.go b/streaming/streaming.go
index f3f28a7c7..e96780bc3 100644
--- a/streaming/streaming.go
+++ b/streaming/streaming.go
@@ -771,6 +771,7 @@ func LaunchDataflowJob(ctx context.Context, migrationProjectId string, targetPro
 		launchParameters.Parameters["transformationJarPath"] = dataflowCfg.CustomJarPath
 		launchParameters.Parameters["transformationClassName"] = dataflowCfg.CustomClassName
 		launchParameters.Parameters["transformationCustomParameters"] = dataflowCfg.CustomParameter
+		launchParameters.Parameters["filteredEventsDirectory"] = utils.ConcatDirectoryPath(inputFilePattern, "filteredEvents")
 	} else if (dataflowCfg.CustomClassName != "" && dataflowCfg.CustomJarPath == "") || (dataflowCfg.CustomClassName == "" && dataflowCfg.CustomJarPath != "") {
 		return internal.DataflowOutput{}, fmt.Errorf("specify both the custom class name and custom jar GCS path, or specify neither")
 	}