Repository: xataio/pgstream
Pull request #329 (2025-05-09 12:26:44)

[Transformers] Add instrumentation
This PR adds instrumentation to the transformation layer. It records the latency metrics for now, as well as track a span trace. This can help identify long running transformations. Main changes:
- Adding context to the `Transform` method (which is required for the metrics/traces).
- Adding a `Type` method to the transformer interface to be able to enrich the metric/traces labels with the type of transformer being used.
- Update transformer builder package to support instrumentation.

Commits have been split for ease of reviewing.

Relates to #323 


Patch:
diff --git a/pkg/stream/stream.go b/pkg/stream/stream.go
index 393ea60..a89dfdc 100644
--- a/pkg/stream/stream.go
+++ b/pkg/stream/stream.go
@@ -10,6 +10,7 @@ import (
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/otel"
 	"github.com/xataio/pgstream/pkg/schemalog"
+	"github.com/xataio/pgstream/pkg/transformers/builder"
 	"github.com/xataio/pgstream/pkg/wal/checkpointer"
 	"github.com/xataio/pgstream/pkg/wal/processor"
 	"github.com/xataio/pgstream/pkg/wal/processor/filter"
@@ -155,6 +156,12 @@ func addProcessorModifiers(ctx context.Context, config *Config, logger loglib.Lo
 	var err error
 	if config.Processor.Transformer != nil {
 		logger.Info("adding transformation layer to processor...")
+		builderOpts := []builder.Option{}
+		if instrumentation.IsEnabled() {
+			builderOpts = append(builderOpts, builder.WithInstrumentation(instrumentation))
+		}
+		transformerBuilder := builder.NewTransformerBuilder(builderOpts...)
+
 		opts := []transformer.Option{transformer.WithLogger(logger)}
 		// if a source pg url is provided, use it to validate the transformer
 		pgURL := ""
@@ -165,14 +172,14 @@ func addProcessorModifiers(ctx context.Context, config *Config, logger loglib.Lo
 			pgURL = config.Listener.Snapshot.Generator.URL
 		}
 		if pgURL != "" {
-			pgParser, err := transformer.NewPostgresTransformerParser(ctx, pgURL)
+			pgParser, err := transformer.NewPostgresTransformerParser(ctx, pgURL, transformerBuilder)
 			if err != nil {
 				return nil, nil, fmt.Errorf("creating transformer validator: %w", err)
 			}
 			closerAgg.addCloserFn(pgParser.Close)
 			opts = append(opts, transformer.WithParser(pgParser.ParseAndValidate))
 		}
-		processor, err = transformer.New(ctx, config.Processor.Transformer, processor, opts...)
+		processor, err = transformer.New(ctx, config.Processor.Transformer, processor, transformerBuilder, opts...)
 		if err != nil {
 			logger.Error(err, "creating transformer layer")
 			return nil, nil, err
diff --git a/pkg/stream/stream_status_checker.go b/pkg/stream/stream_status_checker.go
index 72081e8..478116b 100644
--- a/pkg/stream/stream_status_checker.go
+++ b/pkg/stream/stream_status_checker.go
@@ -11,6 +11,7 @@ import (
 
 	pglib "github.com/xataio/pgstream/internal/postgres"
 	pgmigrations "github.com/xataio/pgstream/migrations/postgres"
+	"github.com/xataio/pgstream/pkg/transformers/builder"
 	"github.com/xataio/pgstream/pkg/wal/processor/transformer"
 
 	"github.com/jackc/pgx/v5"
@@ -51,7 +52,7 @@ func NewStatusChecker() *StatusChecker {
 		configParser:    pgx.ParseConfig,
 		migratorBuilder: func(pgURL string) (migrator, error) { return newPGMigrator(pgURL) },
 		ruleValidatorBuilder: func(ctx context.Context, pgURL string) (ruleValidator, error) {
-			validator, err := transformer.NewPostgresTransformerParser(ctx, pgURL)
+			validator, err := transformer.NewPostgresTransformerParser(ctx, pgURL, builder.NewTransformerBuilder())
 			if err != nil {
 				return nil, err
 			}
diff --git a/pkg/transformers/builder/transformer_builder.go b/pkg/transformers/builder/transformer_builder.go
index 45589c5..51213bb 100644
--- a/pkg/transformers/builder/transformer_builder.go
+++ b/pkg/transformers/builder/transformer_builder.go
@@ -5,12 +5,40 @@ package builder
 import (
 	"fmt"
 
+	"github.com/xataio/pgstream/pkg/otel"
 	"github.com/xataio/pgstream/pkg/transformers"
 	"github.com/xataio/pgstream/pkg/transformers/greenmask"
+	"github.com/xataio/pgstream/pkg/transformers/instrumentation"
 	"github.com/xataio/pgstream/pkg/transformers/neosync"
 )
 
-func New(cfg *transformers.Config) (transformers.Transformer, error) {
+type TransformerBuilder struct {
+	instrumentation *otel.Instrumentation
+}
+
+type Option func(b *TransformerBuilder)
+
+func NewTransformerBuilder(opts ...Option) *TransformerBuilder {
+	b := &TransformerBuilder{}
+	for _, opt := range opts {
+		opt(b)
+	}
+	return b
+}
+
+func WithInstrumentation(i *otel.Instrumentation) Option {
+	return func(b *TransformerBuilder) {
+		b.instrumentation = i
+	}
+}
+
+func (b *TransformerBuilder) New(cfg *transformers.Config) (t transformers.Transformer, err error) {
+	defer func() {
+		if b.instrumentation != nil {
+			t, err = instrumentation.NewTransformer(t, b.instrumentation)
+		}
+	}()
+
 	switch cfg.Name {
 	case transformers.GreenmaskString:
 		return greenmask.NewStringTransformer(cfg.Parameters)
diff --git a/pkg/transformers/greenmask/greenmask_boolean_transformer.go b/pkg/transformers/greenmask/greenmask_boolean_transformer.go
index f79d70d..a053fec 100644
--- a/pkg/transformers/greenmask/greenmask_boolean_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_boolean_transformer.go
@@ -3,6 +3,8 @@
 package greenmask
 
 import (
+	"context"
+
 	greenmasktransformers "github.com/eminano/greenmask/pkg/generators/transformers"
 	"github.com/xataio/pgstream/pkg/transformers"
 )
@@ -26,7 +28,7 @@ func NewBooleanTransformer(params transformers.Parameters) (*BooleanTransformer,
 	}, nil
 }
 
-func (bt *BooleanTransformer) Transform(value transformers.Value) (any, error) {
+func (bt *BooleanTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case bool:
@@ -54,3 +56,7 @@ func (bt *BooleanTransformer) CompatibleTypes() []transformers.SupportedDataType
 		transformers.ByteArrayDataType,
 	}
 }
+
+func (bt *BooleanTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskBoolean
+}
diff --git a/pkg/transformers/greenmask/greenmask_choice_transformer.go b/pkg/transformers/greenmask/greenmask_choice_transformer.go
index 49cc87c..42d13b7 100644
--- a/pkg/transformers/greenmask/greenmask_choice_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_choice_transformer.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"errors"
 	"fmt"
 
@@ -50,7 +51,7 @@ func NewChoiceTransformer(params transformers.Parameters) (*ChoiceTransformer, e
 	}, nil
 }
 
-func (t *ChoiceTransformer) Transform(value transformers.Value) (any, error) {
+func (t *ChoiceTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case []byte:
@@ -76,3 +77,7 @@ func (t *ChoiceTransformer) CompatibleTypes() []transformers.SupportedDataType {
 		transformers.StringDataType,
 	}
 }
+
+func (t *ChoiceTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskChoice
+}
diff --git a/pkg/transformers/greenmask/greenmask_date_transformer.go b/pkg/transformers/greenmask/greenmask_date_transformer.go
index a4cd504..64144da 100644
--- a/pkg/transformers/greenmask/greenmask_date_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_date_transformer.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -67,7 +68,7 @@ func NewDateTransformer(params transformers.Parameters) (*DateTransformer, error
 	}, nil
 }
 
-func (t *DateTransformer) Transform(value transformers.Value) (any, error) {
+func (t *DateTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case time.Time:
@@ -95,3 +96,7 @@ func (t *DateTransformer) CompatibleTypes() []transformers.SupportedDataType {
 		transformers.StringDataType,
 	}
 }
+
+func (t *DateTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskDate
+}
diff --git a/pkg/transformers/greenmask/greenmask_firstname_transformer.go b/pkg/transformers/greenmask/greenmask_firstname_transformer.go
index ccec5d7..c20b8bf 100644
--- a/pkg/transformers/greenmask/greenmask_firstname_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_firstname_transformer.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"fmt"
 
 	greenmasktransformers "github.com/eminano/greenmask/pkg/generators/transformers"
@@ -44,7 +45,7 @@ func NewFirstNameTransformer(params, dynamicParams transformers.Parameters) (*Fi
 	}, nil
 }
 
-func (fnt *FirstNameTransformer) Transform(value transformers.Value) (any, error) {
+func (fnt *FirstNameTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case string:
@@ -91,3 +92,7 @@ func (fnt *FirstNameTransformer) CompatibleTypes() []transformers.SupportedDataT
 		transformers.ByteArrayDataType,
 	}
 }
+
+func (fnt *FirstNameTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskFirstName
+}
diff --git a/pkg/transformers/greenmask/greenmask_float_transformer.go b/pkg/transformers/greenmask/greenmask_float_transformer.go
index beb5212..9dd05b2 100644
--- a/pkg/transformers/greenmask/greenmask_float_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_float_transformer.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"encoding/binary"
 	"fmt"
 	"math"
@@ -55,7 +56,7 @@ func NewFloatTransformer(params transformers.Parameters) (*FloatTransformer, err
 	}, nil
 }
 
-func (ft *FloatTransformer) Transform(value transformers.Value) (any, error) {
+func (ft *FloatTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case float32:
@@ -82,6 +83,10 @@ func (ft *FloatTransformer) CompatibleTypes() []transformers.SupportedDataType {
 	}
 }
 
+func (ft *FloatTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskFloat
+}
+
 func getBytesForFloat(f float64) []byte {
 	var buf [8]byte
 	binary.BigEndian.PutUint64(buf[:], math.Float64bits(f))
diff --git a/pkg/transformers/greenmask/greenmask_integer_transformer.go b/pkg/transformers/greenmask/greenmask_integer_transformer.go
index 768681c..c00213c 100644
--- a/pkg/transformers/greenmask/greenmask_integer_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_integer_transformer.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"math"
@@ -75,7 +76,7 @@ func NewIntegerTransformer(params transformers.Parameters) (*IntegerTransformer,
 // Supported input types are int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, and byte.
 // If the input value is a byte slice, it is passed through the transformer without modification.
 // If the input value is of an unsupported type, an error is returned.
-func (t *IntegerTransformer) Transform(value transformers.Value) (any, error) {
+func (t *IntegerTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case int:
@@ -132,6 +133,10 @@ func (t *IntegerTransformer) CompatibleTypes() []transformers.SupportedDataType
 	}
 }
 
+func (t *IntegerTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskInteger
+}
+
 func minMaxValueForSize(size int) (int, int, error) {
 	switch size {
 	case 2:
diff --git a/pkg/transformers/greenmask/greenmask_string_transformer.go b/pkg/transformers/greenmask/greenmask_string_transformer.go
index 18ec134..27b7215 100644
--- a/pkg/transformers/greenmask/greenmask_string_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_string_transformer.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"fmt"
 
 	greenmasktransformers "github.com/eminano/greenmask/pkg/generators/transformers"
@@ -51,7 +52,7 @@ func NewStringTransformer(params transformers.Parameters) (*StringTransformer, e
 	}, nil
 }
 
-func (st *StringTransformer) Transform(value transformers.Value) (any, error) {
+func (st *StringTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case string:
@@ -72,3 +73,7 @@ func (st *StringTransformer) CompatibleTypes() []transformers.SupportedDataType
 		transformers.ByteArrayDataType,
 	}
 }
+
+func (st *StringTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskString
+}
diff --git a/pkg/transformers/greenmask/greenmask_timestamp_transformer.go b/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
index c4ce639..59fd3c7 100644
--- a/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"time"
@@ -73,7 +74,7 @@ func NewUTCTimestampTransformer(params transformers.Parameters) (*UTCTimestampTr
 	}, nil
 }
 
-func (t *UTCTimestampTransformer) Transform(value transformers.Value) (any, error) {
+func (t *UTCTimestampTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case time.Time:
@@ -95,3 +96,7 @@ func (t *UTCTimestampTransformer) CompatibleTypes() []transformers.SupportedData
 		transformers.StringDataType,
 	}
 }
+
+func (t *UTCTimestampTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskUTCTimestamp
+}
diff --git a/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer.go b/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer.go
index e85205d..fb62248 100644
--- a/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer.go
@@ -15,7 +15,11 @@ var errMinMaxValueNotSpecified = errors.New("min_value and max_value must be spe
 
 var unixTimestampTransformerParams = []string{"min_value", "max_value", "generator"}
 
-func NewUnixTimestampTransformer(params transformers.Parameters) (*IntegerTransformer, error) {
+type UnixTimestampTransformer struct {
+	*IntegerTransformer
+}
+
+func NewUnixTimestampTransformer(params transformers.Parameters) (*UnixTimestampTransformer, error) {
 	if err := transformers.ValidateParameters(params, unixTimestampTransformerParams); err != nil {
 		return nil, err
 	}
@@ -58,7 +62,13 @@ func NewUnixTimestampTransformer(params transformers.Parameters) (*IntegerTransf
 		return nil, err
 	}
 
-	return &IntegerTransformer{
-		transformer: t,
+	return &UnixTimestampTransformer{
+		IntegerTransformer: &IntegerTransformer{
+			transformer: t,
+		},
 	}, nil
 }
+
+func (t *UnixTimestampTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskUnixTimestamp
+}
diff --git a/pkg/transformers/greenmask/greenmask_uuid_transformer.go b/pkg/transformers/greenmask/greenmask_uuid_transformer.go
index 86b430e..d0239fe 100644
--- a/pkg/transformers/greenmask/greenmask_uuid_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_uuid_transformer.go
@@ -3,6 +3,8 @@
 package greenmask
 
 import (
+	"context"
+
 	greenmasktransformers "github.com/eminano/greenmask/pkg/generators/transformers"
 	"github.com/google/uuid"
 	"github.com/xataio/pgstream/pkg/transformers"
@@ -27,7 +29,7 @@ func NewUUIDTransformer(params transformers.Parameters) (*UUIDTransformer, error
 	}, nil
 }
 
-func (ut *UUIDTransformer) Transform(value transformers.Value) (any, error) {
+func (ut *UUIDTransformer) Transform(_ context.Context, value transformers.Value) (any, error) {
 	var toTransform []byte
 	switch val := value.TransformValue.(type) {
 	case string:
@@ -56,3 +58,7 @@ func (ut *UUIDTransformer) CompatibleTypes() []transformers.SupportedDataType {
 		transformers.UInt8ArrayOf16DataType,
 	}
 }
+
+func (ut *UUIDTransformer) Type() transformers.TransformerType {
+	return transformers.GreenmaskUUID
+}
diff --git a/pkg/transformers/instrumentation/instrumented_transformer.go b/pkg/transformers/instrumentation/instrumented_transformer.go
new file mode 100644
index 0000000..773168f
--- /dev/null
+++ b/pkg/transformers/instrumentation/instrumented_transformer.go
@@ -0,0 +1,92 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package instrumentation
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/xataio/pgstream/pkg/otel"
+	"github.com/xataio/pgstream/pkg/transformers"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type Transformer struct {
+	inner   transformers.Transformer
+	tracer  trace.Tracer
+	meter   metric.Meter
+	metrics *metrics
+}
+
+type metrics struct {
+	transformLatency metric.Int64Histogram
+}
+
+const typeAttributeKey = "transformer_type"
+
+func NewTransformer(t transformers.Transformer, instrumentation *otel.Instrumentation) (transformers.Transformer, error) {
+	if instrumentation == nil {
+		return t, nil
+	}
+
+	transformer := &Transformer{
+		inner:   t,
+		tracer:  instrumentation.Tracer,
+		meter:   instrumentation.Meter,
+		metrics: &metrics{},
+	}
+
+	if err := transformer.initMetrics(); err != nil {
+		return nil, fmt.Errorf("initialising transformer metrics: %w", err)
+	}
+
+	return transformer, nil
+}
+
+func (i *Transformer) Transform(ctx context.Context, v transformers.Value) (res any, err error) {
+	ctx, span := otel.StartSpan(ctx, i.tracer, "transformer.Transform", trace.WithAttributes(i.typeAttribute()))
+	defer otel.CloseSpan(span, err)
+
+	if i.meter != nil {
+		startTime := time.Now()
+		defer func() {
+			i.metrics.transformLatency.Record(ctx, int64(time.Since(startTime).Milliseconds()), metric.WithAttributes(i.typeAttribute()))
+		}()
+	}
+	return i.inner.Transform(ctx, v)
+}
+
+func (i *Transformer) CompatibleTypes() []transformers.SupportedDataType {
+	return i.inner.CompatibleTypes()
+}
+
+func (i *Transformer) Type() transformers.TransformerType {
+	return i.inner.Type()
+}
+
+func (i *Transformer) initMetrics() error {
+	if i.meter == nil {
+		return nil
+	}
+
+	var err error
+	i.metrics.transformLatency, err = i.meter.Int64Histogram("pgstream.transformer.latency",
+		metric.WithUnit("ms"),
+		metric.WithDescription("Distribution of the time taken to transform the value"))
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (i *Transformer) typeAttribute() attribute.KeyValue {
+	return attribute.KeyValue{
+		Key:   typeAttributeKey,
+		Value: attribute.StringValue(string(i.inner.Type())),
+	}
+}
diff --git a/pkg/transformers/literal_string_transformer.go b/pkg/transformers/literal_string_transformer.go
index f944fa0..bb2aa1f 100644
--- a/pkg/transformers/literal_string_transformer.go
+++ b/pkg/transformers/literal_string_transformer.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"errors"
 	"fmt"
 )
@@ -34,7 +35,7 @@ func NewLiteralStringTransformer(params Parameters) (*LiteralStringTransformer,
 	}, nil
 }
 
-func (lst *LiteralStringTransformer) Transform(value Value) (any, error) {
+func (lst *LiteralStringTransformer) Transform(_ context.Context, value Value) (any, error) {
 	return lst.literal, nil
 }
 
@@ -43,3 +44,7 @@ func (lst *LiteralStringTransformer) CompatibleTypes() []SupportedDataType {
 		AllDataTypes,
 	}
 }
+
+func (lst *LiteralStringTransformer) Type() TransformerType {
+	return LiteralString
+}
diff --git a/pkg/transformers/masking_transformer.go b/pkg/transformers/masking_transformer.go
index d53f037..47d5486 100644
--- a/pkg/transformers/masking_transformer.go
+++ b/pkg/transformers/masking_transformer.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strconv"
@@ -94,7 +95,7 @@ func NewMaskingTransformer(params Parameters) (*MaskingTransformer, error) {
 }
 
 // Transform applies the masking function to the input value and returns the masked value.
-func (t *MaskingTransformer) Transform(value Value) (any, error) {
+func (t *MaskingTransformer) Transform(_ context.Context, value Value) (any, error) {
 	switch val := value.TransformValue.(type) {
 	case string:
 		return t.maskingFunction(val), nil
@@ -113,6 +114,10 @@ func (t *MaskingTransformer) CompatibleTypes() []SupportedDataType {
 	}
 }
 
+func (t *MaskingTransformer) Type() TransformerType {
+	return Masking
+}
+
 func getCustomMaskingFn(params Parameters) (maskingFunction, error) {
 	maskBegin, maskBeginFound, err := FindParameter[string](params, "mask_begin")
 	if err != nil {
diff --git a/pkg/transformers/mocks/mock_builder.go b/pkg/transformers/mocks/mock_builder.go
new file mode 100644
index 0000000..cde2a13
--- /dev/null
+++ b/pkg/transformers/mocks/mock_builder.go
@@ -0,0 +1,15 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package mocks
+
+import (
+	"github.com/xataio/pgstream/pkg/transformers"
+)
+
+type TransformerBuilder struct {
+	NewFn func(*transformers.Config) (transformers.Transformer, error)
+}
+
+func (m *TransformerBuilder) New(cfg *transformers.Config) (transformers.Transformer, error) {
+	return m.NewFn(cfg)
+}
diff --git a/pkg/transformers/mocks/mock_transformer.go b/pkg/transformers/mocks/mock_transformer.go
index b778eed..057b09b 100644
--- a/pkg/transformers/mocks/mock_transformer.go
+++ b/pkg/transformers/mocks/mock_transformer.go
@@ -2,17 +2,25 @@
 
 package mocks
 
-import "github.com/xataio/pgstream/pkg/transformers"
+import (
+	"context"
+
+	"github.com/xataio/pgstream/pkg/transformers"
+)
 
 type Transformer struct {
 	TransformFn       func(transformers.Value) (any, error)
 	CompatibleTypesFn func() []transformers.SupportedDataType
 }
 
-func (m *Transformer) Transform(val transformers.Value) (any, error) {
+func (m *Transformer) Transform(_ context.Context, val transformers.Value) (any, error) {
 	return m.TransformFn(val)
 }
 
 func (m *Transformer) CompatibleTypes() []transformers.SupportedDataType {
 	return m.CompatibleTypesFn()
 }
+
+func (m *Transformer) Type() transformers.TransformerType {
+	return transformers.TransformerType("mock")
+}
diff --git a/pkg/transformers/neosync/neosync_email_transformer.go b/pkg/transformers/neosync/neosync_email_transformer.go
index 6a34816..c30448e 100644
--- a/pkg/transformers/neosync/neosync_email_transformer.go
+++ b/pkg/transformers/neosync/neosync_email_transformer.go
@@ -103,3 +103,7 @@ func (t *EmailTransformer) CompatibleTypes() []transformers.SupportedDataType {
 		transformers.StringDataType,
 	}
 }
+
+func (t *EmailTransformer) Type() transformers.TransformerType {
+	return transformers.NeosyncEmail
+}
diff --git a/pkg/transformers/neosync/neosync_firstname_transformer.go b/pkg/transformers/neosync/neosync_firstname_transformer.go
index fe34e26..0ecc871 100644
--- a/pkg/transformers/neosync/neosync_firstname_transformer.go
+++ b/pkg/transformers/neosync/neosync_firstname_transformer.go
@@ -50,3 +50,7 @@ func (t *FirstNameTransformer) CompatibleTypes() []transformers.SupportedDataTyp
 		transformers.StringDataType,
 	}
 }
+
+func (t *FirstNameTransformer) Type() transformers.TransformerType {
+	return transformers.NeosyncFirstName
+}
diff --git a/pkg/transformers/neosync/neosync_lastname_transformer.go b/pkg/transformers/neosync/neosync_lastname_transformer.go
index ddce4ed..e654d07 100644
--- a/pkg/transformers/neosync/neosync_lastname_transformer.go
+++ b/pkg/transformers/neosync/neosync_lastname_transformer.go
@@ -50,3 +50,7 @@ func (t *LastNameTransformer) CompatibleTypes() []transformers.SupportedDataType
 		transformers.StringDataType,
 	}
 }
+
+func (t *LastNameTransformer) Type() transformers.TransformerType {
+	return transformers.NeosyncLastName
+}
diff --git a/pkg/transformers/neosync/neosync_string_transformer.go b/pkg/transformers/neosync/neosync_string_transformer.go
index 705ab39..6c26269 100644
--- a/pkg/transformers/neosync/neosync_string_transformer.go
+++ b/pkg/transformers/neosync/neosync_string_transformer.go
@@ -55,3 +55,7 @@ func (t *StringTransformer) CompatibleTypes() []transformers.SupportedDataType {
 		transformers.StringDataType,
 	}
 }
+
+func (t *StringTransformer) Type() transformers.TransformerType {
+	return transformers.NeosyncString
+}
diff --git a/pkg/transformers/neosync/neosync_transformer.go b/pkg/transformers/neosync/neosync_transformer.go
index b17c8be..0982e1a 100644
--- a/pkg/transformers/neosync/neosync_transformer.go
+++ b/pkg/transformers/neosync/neosync_transformer.go
@@ -3,6 +3,8 @@
 package neosync
 
 import (
+	"context"
+
 	"github.com/xataio/pgstream/pkg/transformers"
 )
 
@@ -25,7 +27,7 @@ func New[T any](t neosyncTransformer, opts any) *transformer[T] {
 	}
 }
 
-func (t *transformer[T]) Transform(value transformers.Value) (any, error) {
+func (t *transformer[T]) Transform(_ context.Context, value transformers.Value) (any, error) {
 	retPtr, err := t.neosyncTransformer.Transform(value.TransformValue, t.opts)
 	if err != nil {
 		return nil, err
diff --git a/pkg/transformers/phone_number_transformer.go b/pkg/transformers/phone_number_transformer.go
index 7aa7846..f33cb16 100644
--- a/pkg/transformers/phone_number_transformer.go
+++ b/pkg/transformers/phone_number_transformer.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/xataio/pgstream/pkg/transformers/generators"
@@ -72,7 +73,7 @@ func NewPhoneNumberTransformer(params Parameters) (*PhoneNumberTransformer, erro
 	}, nil
 }
 
-func (t *PhoneNumberTransformer) Transform(value Value) (any, error) {
+func (t *PhoneNumberTransformer) Transform(_ context.Context, value Value) (any, error) {
 	switch v := value.TransformValue.(type) {
 	case string:
 		return t.transform([]byte(v))
@@ -121,3 +122,7 @@ func (t *PhoneNumberTransformer) CompatibleTypes() []SupportedDataType {
 		ByteArrayDataType,
 	}
 }
+
+func (t *PhoneNumberTransformer) Type() TransformerType {
+	return PhoneNumber
+}
diff --git a/pkg/transformers/string_transformer.go b/pkg/transformers/string_transformer.go
index 6b2e42f..cda35e3 100644
--- a/pkg/transformers/string_transformer.go
+++ b/pkg/transformers/string_transformer.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"fmt"
 
 	"golang.org/x/exp/rand"
@@ -23,7 +24,7 @@ func NewStringTransformer(params Parameters) (*StringTransformer, error) {
 	return &StringTransformer{}, nil
 }
 
-func (st *StringTransformer) Transform(v Value) (any, error) {
+func (st *StringTransformer) Transform(_ context.Context, v Value) (any, error) {
 	switch str := v.TransformValue.(type) {
 	case string:
 		return st.transform(str), nil
@@ -50,3 +51,7 @@ func (st *StringTransformer) CompatibleTypes() []SupportedDataType {
 		ByteArrayDataType,
 	}
 }
+
+func (st *StringTransformer) Type() TransformerType {
+	return String
+}
diff --git a/pkg/transformers/transformer.go b/pkg/transformers/transformer.go
index 777fb15..abac35f 100644
--- a/pkg/transformers/transformer.go
+++ b/pkg/transformers/transformer.go
@@ -3,13 +3,15 @@
 package transformers
 
 import (
+	"context"
 	"errors"
 	"fmt"
 )
 
 type Transformer interface {
-	Transform(Value) (any, error)
+	Transform(context.Context, Value) (any, error)
 	CompatibleTypes() []SupportedDataType
+	Type() TransformerType
 }
 
 type Value struct {
diff --git a/pkg/wal/processor/transformer/wal_transformer_validator.go b/pkg/wal/processor/transformer/wal_postgres_transformer_parser.go
similarity index 95%
rename from pkg/wal/processor/transformer/wal_transformer_validator.go
rename to pkg/wal/processor/transformer/wal_postgres_transformer_parser.go
index d0b4861..f8f5dfc 100644
--- a/pkg/wal/processor/transformer/wal_transformer_validator.go
+++ b/pkg/wal/processor/transformer/wal_postgres_transformer_parser.go
@@ -10,23 +10,24 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 	pglib "github.com/xataio/pgstream/internal/postgres"
 	"github.com/xataio/pgstream/pkg/transformers"
-	"github.com/xataio/pgstream/pkg/transformers/builder"
 	"golang.org/x/exp/slices"
 )
 
 type PostgresTransformerParser struct {
-	conn pglib.Querier
+	conn    pglib.Querier
+	builder transformerBuilder
 }
 
 const fieldDescriptionsQuery = "SELECT * FROM %s LIMIT 0"
 
-func NewPostgresTransformerParser(ctx context.Context, pgURL string) (*PostgresTransformerParser, error) {
+func NewPostgresTransformerParser(ctx context.Context, pgURL string, builder transformerBuilder) (*PostgresTransformerParser, error) {
 	pool, err := pglib.NewConnPool(ctx, pgURL)
 	if err != nil {
 		return nil, err
 	}
 	return &PostgresTransformerParser{
-		conn: pool,
+		conn:    pool,
+		builder: builder,
 	}, nil
 }
 
@@ -70,7 +71,7 @@ func (v *PostgresTransformerParser) ParseAndValidate(rules []TableRules) (map[st
 			}
 
 			// build the transformer
-			transformer, err := builder.New(cfg)
+			transformer, err := v.builder.New(cfg)
 			if err != nil {
 				return nil, err
 			}
diff --git a/pkg/wal/processor/transformer/wal_transformer.go b/pkg/wal/processor/transformer/wal_transformer.go
index 8140bcf..c2ddc3f 100644
--- a/pkg/wal/processor/transformer/wal_transformer.go
+++ b/pkg/wal/processor/transformer/wal_transformer.go
@@ -9,7 +9,6 @@ import (
 	pglib "github.com/xataio/pgstream/internal/postgres"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/transformers"
-	"github.com/xataio/pgstream/pkg/transformers/builder"
 	"github.com/xataio/pgstream/pkg/wal"
 	"github.com/xataio/pgstream/pkg/wal/processor"
 )
@@ -27,6 +26,10 @@ type ParseFn func(rules []TableRules) (map[string]ColumnTransformers, error)
 
 type ColumnTransformers map[string]transformers.Transformer
 
+type transformerBuilder interface {
+	New(*transformers.Config) (transformers.Transformer, error)
+}
+
 type Config struct {
 	TransformerRules []TableRules
 }
@@ -39,11 +42,11 @@ var errValidatorRequiredForStrictMode = errors.New("strict validation mode requi
 
 // New will return a transformer processor wrapper that will transform incoming
 // wal event column values as configured by the transformation rules.
-func New(ctx context.Context, cfg *Config, processor processor.Processor, opts ...Option) (*Transformer, error) {
+func New(ctx context.Context, cfg *Config, processor processor.Processor, builder transformerBuilder, opts ...Option) (*Transformer, error) {
 	t := &Transformer{
 		logger:    loglib.NewNoopLogger(),
 		processor: processor,
-		parser:    transformerMapFromRules,
+		parser:    newTransformerParser(builder).parse,
 	}
 
 	for _, opt := range opts {
@@ -74,7 +77,7 @@ func WithParser(parser ParseFn) Option {
 }
 
 func (t *Transformer) ProcessWALEvent(ctx context.Context, event *wal.Event) error {
-	err := t.applyTransformations(event)
+	err := t.applyTransformations(ctx, event)
 	if err != nil {
 		return err
 	}
@@ -90,7 +93,7 @@ func (t *Transformer) Close() error {
 	return t.processor.Close()
 }
 
-func (t *Transformer) applyTransformations(event *wal.Event) error {
+func (t *Transformer) applyTransformations(ctx context.Context, event *wal.Event) error {
 	if event.Data == nil || len(t.transformerMap) == 0 {
 		return nil
 	}
@@ -112,7 +115,7 @@ func (t *Transformer) applyTransformations(event *wal.Event) error {
 			continue
 		}
 
-		newValue, err := columnTransformer.Transform(t.getTransformValue(&col, event.Data.Columns))
+		newValue, err := columnTransformer.Transform(ctx, t.getTransformValue(&col, event.Data.Columns))
 		if err != nil {
 			t.logger.Error(err, "transforming column", loglib.Fields{
 				"severity":    "DATALOSS",
@@ -143,34 +146,3 @@ func (t *Transformer) getTransformValue(column *wal.Column, columns []wal.Column
 func schemaTableKey(schema, table string) string {
 	return pglib.QuoteQualifiedIdentifier(schema, table)
 }
-
-func transformerMapFromRules(rules []TableRules) (map[string]ColumnTransformers, error) {
-	var err error
-	transformerMap := map[string]ColumnTransformers{}
-	for _, table := range rules {
-		if table.ValidationMode == validationModeStrict {
-			return nil, errValidatorRequiredForStrictMode
-		}
-		schemaTableTransformers := make(map[string]transformers.Transformer)
-		transformerMap[schemaTableKey(table.Schema, table.Table)] = schemaTableTransformers
-		for colName, transformerRules := range table.ColumnRules {
-			cfg := transformerRulesToConfig(transformerRules)
-			if cfg.Name == "" || cfg.Name == "noop" {
-				// noop transformer, skip
-				continue
-			}
-			if schemaTableTransformers[colName], err = builder.New(cfg); err != nil {
-				return nil, err
-			}
-		}
-	}
-	return transformerMap, nil
-}
-
-func transformerRulesToConfig(rules TransformerRules) *transformers.Config {
-	return &transformers.Config{
-		Name:              transformers.TransformerType(rules.Name),
-		Parameters:        rules.Parameters,
-		DynamicParameters: rules.DynamicParameters,
-	}
-}
diff --git a/pkg/wal/processor/transformer/wal_transformer_parser.go b/pkg/wal/processor/transformer/wal_transformer_parser.go
new file mode 100644
index 0000000..308047f
--- /dev/null
+++ b/pkg/wal/processor/transformer/wal_transformer_parser.go
@@ -0,0 +1,48 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformer
+
+import (
+	"github.com/xataio/pgstream/pkg/transformers"
+)
+
+type transformerParser struct {
+	builder transformerBuilder
+}
+
+func newTransformerParser(b transformerBuilder) *transformerParser {
+	return &transformerParser{
+		builder: b,
+	}
+}
+
+func (p *transformerParser) parse(rules []TableRules) (map[string]ColumnTransformers, error) {
+	var err error
+	transformerMap := map[string]ColumnTransformers{}
+	for _, table := range rules {
+		if table.ValidationMode == validationModeStrict {
+			return nil, errValidatorRequiredForStrictMode
+		}
+		schemaTableTransformers := make(map[string]transformers.Transformer)
+		transformerMap[schemaTableKey(table.Schema, table.Table)] = schemaTableTransformers
+		for colName, transformerRules := range table.ColumnRules {
+			cfg := transformerRulesToConfig(transformerRules)
+			if cfg.Name == "" || cfg.Name == "noop" {
+				// noop transformer, skip
+				continue
+			}
+			if schemaTableTransformers[colName], err = p.builder.New(cfg); err != nil {
+				return nil, err
+			}
+		}
+	}
+	return transformerMap, nil
+}
+
+func transformerRulesToConfig(rules TransformerRules) *transformers.Config {
+	return &transformers.Config{
+		Name:              transformers.TransformerType(rules.Name),
+		Parameters:        rules.Parameters,
+		DynamicParameters: rules.DynamicParameters,
+	}
+}


Test patch:
diff --git a/pkg/transformers/builder/transformer_builder_test.go b/pkg/transformers/builder/transformer_builder_test.go
index ffd5d39..5a091f1 100644
--- a/pkg/transformers/builder/transformer_builder_test.go
+++ b/pkg/transformers/builder/transformer_builder_test.go
@@ -9,7 +9,7 @@ import (
 	"github.com/xataio/pgstream/pkg/transformers"
 )
 
-func TestNew(t *testing.T) {
+func TestTransformerBuilder_New(t *testing.T) {
 	tests := []struct {
 		name    string
 		config  *transformers.Config
@@ -44,7 +44,8 @@ func TestNew(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
-			_, err := New(tt.config)
+			tb := NewTransformerBuilder()
+			_, err := tb.New(tt.config)
 			require.ErrorIs(t, err, tt.wantErr)
 		})
 	}
diff --git a/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go b/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go
index cbd1e64..c814d8e 100644
--- a/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -96,7 +97,7 @@ func Test_BooleanTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.input})
 			require.ErrorIs(t, err, tc.wantErr)
 			if err != nil {
 				return
@@ -107,7 +108,7 @@ func Test_BooleanTransformer_Transform(t *testing.T) {
 
 			// if deterministic, the same input should always produce the same output
 			if mustGetGeneratorType(t, tc.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/greenmask/greenmask_choice_transformer_test.go b/pkg/transformers/greenmask/greenmask_choice_transformer_test.go
index 38eb55b..f177e32 100644
--- a/pkg/transformers/greenmask/greenmask_choice_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_choice_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"testing"
 
 	"github.com/eminano/greenmask/pkg/toolkit"
@@ -106,7 +107,7 @@ func TestChoiceTransformer_Transform(t *testing.T) {
 			transformer, err := NewChoiceTransformer(tt.params)
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
-			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 			require.Equal(t, tt.wantErr, err)
 			if err != nil {
 				return
@@ -120,7 +121,7 @@ func TestChoiceTransformer_Transform(t *testing.T) {
 
 			// if deterministic, check if we get the same result again
 			if mustGetGeneratorType(t, tt.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/greenmask/greenmask_date_transformer_test.go b/pkg/transformers/greenmask/greenmask_date_transformer_test.go
index 5387a22..8e95c1e 100644
--- a/pkg/transformers/greenmask/greenmask_date_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_date_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -123,7 +124,7 @@ func TestDateTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 			require.ErrorIs(t, err, tt.wantErr)
 			if err != nil {
 				return
@@ -143,7 +144,7 @@ func TestDateTransformer_Transform(t *testing.T) {
 			require.True(t, result.Before(maxDate) || result.Equal(maxDate))
 
 			if mustGetGeneratorType(t, tt.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/greenmask/greenmask_firstname_transformer_test.go b/pkg/transformers/greenmask/greenmask_firstname_transformer_test.go
index 79330c7..0d472b2 100644
--- a/pkg/transformers/greenmask/greenmask_firstname_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_firstname_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"errors"
 	"testing"
 
@@ -205,7 +206,7 @@ func TestFirstNameTransformer_Transform(t *testing.T) {
 			transformer, err := NewFirstNameTransformer(tc.params, tc.dynamicParams)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(transformers.NewValue(tc.value, tc.dynamicValues))
+			got, err := transformer.Transform(context.Background(), transformers.NewValue(tc.value, tc.dynamicValues))
 			require.Equal(t, err, tc.wantErr)
 			if err != nil {
 				return
diff --git a/pkg/transformers/greenmask/greenmask_float_transformer_test.go b/pkg/transformers/greenmask/greenmask_float_transformer_test.go
index f62d904..855fa57 100644
--- a/pkg/transformers/greenmask/greenmask_float_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_float_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"testing"
 
 	greenmasktransformers "github.com/eminano/greenmask/pkg/generators/transformers"
@@ -184,7 +185,7 @@ func TestFloatTransformer_Transform(t *testing.T) {
 			transformer, err := NewFloatTransformer(tc.params)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tc.value})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			if err != nil {
 				return
@@ -207,7 +208,7 @@ func TestFloatTransformer_Transform(t *testing.T) {
 
 			// if deterministic, check if we get the same result again
 			if mustGetGeneratorType(t, tc.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tc.value})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.value})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/greenmask/greenmask_integer_transformer_test.go b/pkg/transformers/greenmask/greenmask_integer_transformer_test.go
index f26bc6b..64a5bd3 100644
--- a/pkg/transformers/greenmask/greenmask_integer_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_integer_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"math"
 	"testing"
 
@@ -255,7 +256,7 @@ func TestIntegerTransformer_Transform(t *testing.T) {
 			transformer, err := NewIntegerTransformer(tt.params)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 			require.ErrorIs(t, err, tt.wantErr)
 			if err != nil {
 				return
@@ -280,7 +281,7 @@ func TestIntegerTransformer_Transform(t *testing.T) {
 
 			// if deterministic, check if we get the same result again
 			if mustGetGeneratorType(t, tt.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/greenmask/greenmask_string_transformer_test.go b/pkg/transformers/greenmask/greenmask_string_transformer_test.go
index a14373e..bd5003c 100644
--- a/pkg/transformers/greenmask/greenmask_string_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_string_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -161,7 +162,7 @@ func TestStringTransformer_Transform(t *testing.T) {
 			transformer, err := NewStringTransformer(tc.params)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tc.value})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			if err != nil {
 				return
diff --git a/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go b/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
index 76e4d82..c3b3af1 100644
--- a/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -181,7 +182,7 @@ func TestUTCTimestampTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 			require.ErrorIs(t, err, tt.wantErr)
 			if err != nil {
 				return
@@ -234,7 +235,7 @@ func TestUTCTimestampTransformer_Transform(t *testing.T) {
 
 			// if deterministic, check that the same input always produces the same output
 			if mustGetGeneratorType(t, tt.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer_test.go b/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer_test.go
index 8e54ca3..961ec2c 100644
--- a/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"strconv"
 	"testing"
 
@@ -116,7 +117,7 @@ func TestUnixTimestampTransformer_Transform(t *testing.T) {
 			t.Parallel()
 			transformer, err := NewUnixTimestampTransformer(tt.params)
 			require.NoError(t, err)
-			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 			require.NoError(t, err)
 			require.NotNil(t, got)
 
@@ -135,7 +136,7 @@ func TestUnixTimestampTransformer_Transform(t *testing.T) {
 
 			// if deterministic, check if we get the same result again
 			if mustGetGeneratorType(t, tt.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tt.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/greenmask/greenmask_uuid_transformer_test.go b/pkg/transformers/greenmask/greenmask_uuid_transformer_test.go
index 7302401..eca64bf 100644
--- a/pkg/transformers/greenmask/greenmask_uuid_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_uuid_transformer_test.go
@@ -3,6 +3,7 @@
 package greenmask
 
 import (
+	"context"
 	"errors"
 	"testing"
 
@@ -109,7 +110,7 @@ func Test_UUIDTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.input})
 			if !errors.Is(err, tc.wantErr) {
 				require.Error(t, err, tc.wantErr.Error())
 			}
@@ -120,7 +121,7 @@ func Test_UUIDTransformer_Transform(t *testing.T) {
 
 			// if deterministic, the same input should always produce the same output
 			if mustGetGeneratorType(t, tc.params) == deterministic {
-				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
+				gotAgain, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/literal_string_transformer_test.go b/pkg/transformers/literal_string_transformer_test.go
index 16aefa0..d81d5fc 100644
--- a/pkg/transformers/literal_string_transformer_test.go
+++ b/pkg/transformers/literal_string_transformer_test.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -76,7 +77,7 @@ func TestLiteralStringTransformer_Transform(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			got, err := lst.Transform(Value{TransformValue: tc.input})
+			got, err := lst.Transform(context.Background(), Value{TransformValue: tc.input})
 			require.ErrorIs(t, err, tc.wantErr)
 			if tc.wantErr != nil {
 				return
diff --git a/pkg/transformers/masking_transformer_test.go b/pkg/transformers/masking_transformer_test.go
index 1c00c8f..19735e0 100644
--- a/pkg/transformers/masking_transformer_test.go
+++ b/pkg/transformers/masking_transformer_test.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -218,7 +219,7 @@ func TestMaskingTransformer_Transform(t *testing.T) {
 			t.Parallel()
 			mt, err := NewMaskingTransformer(tt.params)
 			require.NoError(t, err)
-			got, err := mt.Transform(Value{TransformValue: tt.input})
+			got, err := mt.Transform(context.Background(), Value{TransformValue: tt.input})
 			require.ErrorIs(t, err, tt.wantErr)
 			if tt.wantErr != nil {
 				return
diff --git a/pkg/transformers/neosync/neosync_email_transformer_test.go b/pkg/transformers/neosync/neosync_email_transformer_test.go
index 159ee8b..47984fe 100644
--- a/pkg/transformers/neosync/neosync_email_transformer_test.go
+++ b/pkg/transformers/neosync/neosync_email_transformer_test.go
@@ -3,6 +3,7 @@
 package neosync
 
 import (
+	"context"
 	"strings"
 	"testing"
 
@@ -184,7 +185,7 @@ func TestEmailTransformer_Transform(t *testing.T) {
 			}
 			transformer, err := NewEmailTransformer(params)
 			require.NoError(t, err)
-			got, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.input})
 			require.ErrorIs(t, err, tc.wantErr)
 			require.NotNil(t, got)
 			val, ok := got.(string)
diff --git a/pkg/transformers/neosync/neosync_firstname_transformer_test.go b/pkg/transformers/neosync/neosync_firstname_transformer_test.go
index 1afb5a4..1dfdc29 100644
--- a/pkg/transformers/neosync/neosync_firstname_transformer_test.go
+++ b/pkg/transformers/neosync/neosync_firstname_transformer_test.go
@@ -3,6 +3,7 @@
 package neosync
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -75,7 +76,7 @@ func TestFirstnameTransformer_Transform(t *testing.T) {
 				return
 			}
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tc.value})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			require.Equal(t, tc.wantName, got)
 		})
diff --git a/pkg/transformers/neosync/neosync_lastname_transformer_test.go b/pkg/transformers/neosync/neosync_lastname_transformer_test.go
index 7144d12..77475d2 100644
--- a/pkg/transformers/neosync/neosync_lastname_transformer_test.go
+++ b/pkg/transformers/neosync/neosync_lastname_transformer_test.go
@@ -3,6 +3,7 @@
 package neosync
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -67,7 +68,7 @@ func TestNewLastnameTransformer(t *testing.T) {
 			}
 			require.NoError(t, err)
 			require.NotNil(t, lst)
-			got, _ := lst.Transform(transformers.Value{TransformValue: tc.input})
+			got, _ := lst.Transform(context.Background(), transformers.Value{TransformValue: tc.input})
 			require.Equal(t, tc.wantName, got)
 		})
 	}
diff --git a/pkg/transformers/neosync/neosync_string_transformer_test.go b/pkg/transformers/neosync/neosync_string_transformer_test.go
index 0e2cf91..17a96c2 100644
--- a/pkg/transformers/neosync/neosync_string_transformer_test.go
+++ b/pkg/transformers/neosync/neosync_string_transformer_test.go
@@ -3,6 +3,7 @@
 package neosync
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -86,7 +87,7 @@ func TestStringTransformer_Transform(t *testing.T) {
 				return
 			}
 
-			got, err := transformer.Transform(transformers.Value{TransformValue: tc.value})
+			got, err := transformer.Transform(context.Background(), transformers.Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			require.Equal(t, tc.wantString, got)
 		})
diff --git a/pkg/transformers/phone_number_transformer_test.go b/pkg/transformers/phone_number_transformer_test.go
index defc423..be19134 100644
--- a/pkg/transformers/phone_number_transformer_test.go
+++ b/pkg/transformers/phone_number_transformer_test.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -86,7 +87,7 @@ func TestPhoneNumberTransformer_Transform(t *testing.T) {
 			}
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(Value{TransformValue: tc.value})
+			got, err := transformer.Transform(context.Background(), Value{TransformValue: tc.value})
 			require.NoError(t, err)
 
 			gotStr, ok := got.(string)
diff --git a/pkg/transformers/string_transformer_test.go b/pkg/transformers/string_transformer_test.go
index 8fc269b..b43fde3 100644
--- a/pkg/transformers/string_transformer_test.go
+++ b/pkg/transformers/string_transformer_test.go
@@ -3,6 +3,7 @@
 package transformers
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -47,7 +48,7 @@ func TestStringTransformer_Transform(t *testing.T) {
 
 			st, err := NewStringTransformer(nil)
 			require.NoError(t, err)
-			got, err := st.Transform(Value{TransformValue: tc.value})
+			got, err := st.Transform(context.Background(), Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			if tc.wantErr != nil {
 				return
diff --git a/pkg/wal/processor/transformer/wal_transformer_validator_test.go b/pkg/wal/processor/transformer/wal_postgres_transformer_parser_test.go
similarity index 95%
rename from pkg/wal/processor/transformer/wal_transformer_validator_test.go
rename to pkg/wal/processor/transformer/wal_postgres_transformer_parser_test.go
index a43f4d1..96aa28f 100644
--- a/pkg/wal/processor/transformer/wal_transformer_validator_test.go
+++ b/pkg/wal/processor/transformer/wal_postgres_transformer_parser_test.go
@@ -13,9 +13,10 @@ import (
 	"github.com/stretchr/testify/require"
 	pglib "github.com/xataio/pgstream/internal/postgres"
 	pgmocks "github.com/xataio/pgstream/internal/postgres/mocks"
+	"github.com/xataio/pgstream/pkg/transformers/builder"
 )
 
-func TestPostgresTransformerValidator(t *testing.T) {
+func TestPostgresTransformerParser_ParseAndValidate(t *testing.T) {
 	t.Parallel()
 	testSchemaTable := "\"public\".\"test\""
 	testQuerier := &pgmocks.Querier{
@@ -41,7 +42,8 @@ func TestPostgresTransformerValidator(t *testing.T) {
 	}
 
 	testPGValidator := PostgresTransformerParser{
-		conn: testQuerier,
+		conn:    testQuerier,
+		builder: builder.NewTransformerBuilder(),
 	}
 
 	tests := []struct {
diff --git a/pkg/wal/processor/transformer/wal_transformer_parser_test.go b/pkg/wal/processor/transformer/wal_transformer_parser_test.go
new file mode 100644
index 0000000..f619f73
--- /dev/null
+++ b/pkg/wal/processor/transformer/wal_transformer_parser_test.go
@@ -0,0 +1,100 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformer
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/pkg/transformers"
+	transformermocks "github.com/xataio/pgstream/pkg/transformers/mocks"
+)
+
+func TestTransformerParser_parse(t *testing.T) {
+	t.Parallel()
+
+	testSchema := "test_schema"
+	testTable := "test_table"
+	testTransformer, err := transformers.NewStringTransformer(nil)
+	require.NoError(t, err)
+	testKey := "\"test_schema\".\"test_table\""
+
+	mockBuilder := &transformermocks.TransformerBuilder{
+		NewFn: func(cfg *transformers.Config) (transformers.Transformer, error) {
+			if cfg.Name == "string" {
+				return testTransformer, nil
+			}
+			return nil, transformers.ErrUnsupportedTransformer
+		},
+	}
+
+	tests := []struct {
+		name  string
+		rules []TableRules
+
+		wantTransformerMap map[string]ColumnTransformers
+		wantErr            error
+	}{
+		{
+			name: "ok",
+			rules: []TableRules{
+				{
+					Schema: testSchema,
+					Table:  testTable,
+					ColumnRules: map[string]TransformerRules{
+						"column_1": {
+							Name: "string",
+						},
+						"column_2": {
+							Name: "string",
+						},
+					},
+				},
+			},
+
+			wantTransformerMap: map[string]ColumnTransformers{
+				testKey: {
+					"column_1": testTransformer,
+					"column_2": testTransformer,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name:  "ok - no rules",
+			rules: []TableRules{},
+
+			wantTransformerMap: map[string]ColumnTransformers{},
+			wantErr:            nil,
+		},
+		{
+			name: "error - invalid transformer rules",
+			rules: []TableRules{
+				{
+					Schema: testSchema,
+					Table:  testTable,
+					ColumnRules: map[string]TransformerRules{
+						"column_1": {
+							Name: "invalid",
+						},
+					},
+				},
+			},
+
+			wantTransformerMap: nil,
+			wantErr:            transformers.ErrUnsupportedTransformer,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tp := newTransformerParser(mockBuilder)
+
+			transformerMap, err := tp.parse(tc.rules)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantTransformerMap, transformerMap)
+		})
+	}
+}
diff --git a/pkg/wal/processor/transformer/wal_transformer_test.go b/pkg/wal/processor/transformer/wal_transformer_test.go
index c00b049..67fcb08 100644
--- a/pkg/wal/processor/transformer/wal_transformer_test.go
+++ b/pkg/wal/processor/transformer/wal_transformer_test.go
@@ -23,6 +23,12 @@ func TestTransformer_New(t *testing.T) {
 	testTransformer, err := transformers.NewStringTransformer(nil)
 	require.NoError(t, err)
 
+	mockBuilder := &transformermocks.TransformerBuilder{
+		NewFn: func(_ *transformers.Config) (transformers.Transformer, error) {
+			return testTransformer, nil
+		},
+	}
+
 	tests := []struct {
 		name   string
 		config *Config
@@ -71,7 +77,7 @@ func TestTransformer_New(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			transformer, err := New(context.Background(), tc.config, mockProcessor)
+			transformer, err := New(context.Background(), tc.config, mockProcessor, mockBuilder)
 			require.ErrorIs(t, err, tc.wantErr)
 			require.Equal(t, tc.wantTransformerMap, transformer.transformerMap)
 			require.Equal(t, mockProcessor, transformer.processor)
@@ -240,81 +246,3 @@ func TestTransformer_ProcessWALEvent(t *testing.T) {
 		})
 	}
 }
-
-func Test_transformerMapFromRules(t *testing.T) {
-	t.Parallel()
-
-	testSchema := "test_schema"
-	testTable := "test_table"
-	testTransformer, err := transformers.NewStringTransformer(nil)
-	require.NoError(t, err)
-	testKey := "\"test_schema\".\"test_table\""
-
-	tests := []struct {
-		name  string
-		rules []TableRules
-
-		wantTransformerMap map[string]ColumnTransformers
-		wantErr            error
-	}{
-		{
-			name: "ok",
-			rules: []TableRules{
-				{
-					Schema: testSchema,
-					Table:  testTable,
-					ColumnRules: map[string]TransformerRules{
-						"column_1": {
-							Name: "string",
-						},
-						"column_2": {
-							Name: "string",
-						},
-					},
-				},
-			},
-
-			wantTransformerMap: map[string]ColumnTransformers{
-				testKey: {
-					"column_1": testTransformer,
-					"column_2": testTransformer,
-				},
-			},
-			wantErr: nil,
-		},
-		{
-			name:  "ok - no rules",
-			rules: []TableRules{},
-
-			wantTransformerMap: map[string]ColumnTransformers{},
-			wantErr:            nil,
-		},
-		{
-			name: "error - invalid transformer rules",
-			rules: []TableRules{
-				{
-					Schema: testSchema,
-					Table:  testTable,
-					ColumnRules: map[string]TransformerRules{
-						"column_1": {
-							Name: "invalid",
-						},
-					},
-				},
-			},
-
-			wantTransformerMap: nil,
-			wantErr:            transformers.ErrUnsupportedTransformer,
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			transformerMap, err := transformerMapFromRules(tc.rules)
-			require.ErrorIs(t, err, tc.wantErr)
-			require.Equal(t, tc.wantTransformerMap, transformerMap)
-		})
-	}
-}