Repository: xataio/pgstream
Pull request #240 (2025-04-03 10:41:45)

[Transformers] Support dynamic parameters
This PR updates the transformer interface to be able to support dynamic parameters when providing a transformation. The transformer interface now receives the value for the column to transform along with the map of all other row values so that dynamic transformations can be applied. For now, the dynamic parameters offer basic support to provide a column, but should be extended as we include templates.

Example of implementation can be found for the greenmask first name transformer, where the gender can be dynamically provided by a different column.

Example of yaml transformer rules, assuming a `test` table with columns `name` and `sex`:
```
transformations:
  - schema: public
    table: test
    column_transformers:
      name:
        name: greenmask_firstname
        dynamic_parameters:
          gender:
            column: sex
```

Commits have been split for ease of reviewing.


github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 2 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 75.45% (**-16.68%**) | :skull:  |
| github.com/xataio/pgstream/pkg/transformers/builder | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask | 91.84% (**+0.37%**) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/mocks | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/neosync | 94.38% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 84.13% (**+1.67%**) | :thumbsup: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/transformers/builder/transformer_builder.go | 0.00% (ø) | 18 | 0 | 18 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_boolean_transformer.go | 93.33% (ø) | 15 | 14 | 1 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_choice_transformer.go | 91.30% (ø) | 23 | 21 | 2 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_date_transformer.go | 82.86% (ø) | 35 | 29 | 6 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_firstname_transformer.go | 100.00% (**+5.00%**) | 29 (+9) | 29 (+10) | 0 (-1) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_float_transformer.go | 96.55% (ø) | 29 | 28 | 1 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_integer_transformer.go | 91.11% (ø) | 45 | 41 | 4 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_string_transformer.go | 95.45% (ø) | 22 | 21 | 1 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_timestamp_transformer.go | 97.14% (ø) | 35 | 34 | 1 |  |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_transformer.go | 85.00% (**-4.29%**) | 20 (-8) | 17 (-8) | 3 | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_uuid_transformer.go | 88.89% (ø) | 18 | 16 | 2 |  |
| github.com/xataio/pgstream/pkg/transformers/masking_transformer.go | 100.00% (ø) | 25 | 25 | 0 |  |
| github.com/xataio/pgstream/pkg/transformers/mocks/mock_transformer.go | 0.00% (ø) | 1 | 0 | 1 |  |
| github.com/xataio/pgstream/pkg/transformers/neosync/neosync_transformer.go | 93.55% (ø) | 31 | 29 | 2 |  |
| github.com/xataio/pgstream/pkg/transformers/phone_number_transformer.go | 82.14% (**+2.73%**) | 28 (-6) | 23 (-4) | 5 (-2) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/string_transformer.go | 100.00% (ø) | 10 | 10 | 0 |  |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 53.19% (**-46.81%**) | 47 (+27) | 25 (+5) | 22 (+22) | :skull: :skull: :skull: :skull:  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 85.45% (**+1.78%**) | 55 (+6) | 47 (+6) | 8 | :thumbsup: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 75.00% (ø) | 8 | 6 | 2 |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_choice_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_date_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_firstname_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_float_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_integer_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_string_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_uuid_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/masking_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/neosync/neosync_email_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/neosync/neosync_firstname_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/neosync/neosync_string_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/phone_number_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/string_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_test.go

</details>


Patch:
diff --git a/go.mod b/go.mod
index 24a0a76..c8caa8c 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,7 @@
 module github.com/xataio/pgstream
 
 go 1.24.1
+
 require (
 	github.com/bytedance/sonic v1.13.2
 	github.com/cenkalti/backoff/v4 v4.3.0
@@ -37,7 +38,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.35.0
 	golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c
 	golang.org/x/sync v0.12.0
-	gopkg.in/yaml.v2 v2.4.0
+	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
@@ -159,5 +160,4 @@ require (
 	golang.org/x/time v0.9.0 // indirect
 	google.golang.org/protobuf v1.36.3 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/go.sum b/go.sum
index 5d85991..3216544 100644
--- a/go.sum
+++ b/go.sum
@@ -578,8 +578,6 @@ gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYs
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
-gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
diff --git a/pkg/transformers/builder/transformer_builder.go b/pkg/transformers/builder/transformer_builder.go
index 91f1a1f..8802ff3 100644
--- a/pkg/transformers/builder/transformer_builder.go
+++ b/pkg/transformers/builder/transformer_builder.go
@@ -13,7 +13,7 @@ func New(cfg *transformers.Config) (transformers.Transformer, error) {
 	case transformers.GreenmaskString:
 		return greenmask.NewStringTransformer(cfg.Parameters)
 	case transformers.GreenmaskFirstName:
-		return greenmask.NewFirstNameTransformer(cfg.Parameters)
+		return greenmask.NewFirstNameTransformer(cfg.Parameters, cfg.DynamicParameters)
 	case transformers.GreenmaskInteger:
 		return greenmask.NewIntegerTransformer(cfg.Parameters)
 	case transformers.GreenmaskFloat:
diff --git a/pkg/transformers/greenmask/greenmask_boolean_transformer.go b/pkg/transformers/greenmask/greenmask_boolean_transformer.go
index c787d88..f230bf3 100644
--- a/pkg/transformers/greenmask/greenmask_boolean_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_boolean_transformer.go
@@ -21,9 +21,9 @@ func NewBooleanTransformer(params transformers.Parameters) (*BooleanTransformer,
 	}, nil
 }
 
-func (bt *BooleanTransformer) Transform(value any) (any, error) {
+func (bt *BooleanTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case bool:
 		if val {
 			toTransform = []byte{1}
diff --git a/pkg/transformers/greenmask/greenmask_choice_transformer.go b/pkg/transformers/greenmask/greenmask_choice_transformer.go
index b1bda92..801ae72 100644
--- a/pkg/transformers/greenmask/greenmask_choice_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_choice_transformer.go
@@ -44,9 +44,9 @@ func NewChoiceTransformer(params transformers.Parameters) (*ChoiceTransformer, e
 	}, nil
 }
 
-func (t *ChoiceTransformer) Transform(value any) (any, error) {
+func (t *ChoiceTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case []byte:
 		toTransform = val
 	case string:
diff --git a/pkg/transformers/greenmask/greenmask_date_transformer.go b/pkg/transformers/greenmask/greenmask_date_transformer.go
index e86f4e9..f1599bc 100644
--- a/pkg/transformers/greenmask/greenmask_date_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_date_transformer.go
@@ -61,9 +61,9 @@ func NewDateTransformer(params transformers.Parameters) (*DateTransformer, error
 	}, nil
 }
 
-func (t *DateTransformer) Transform(value any) (any, error) {
+func (t *DateTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case time.Time:
 		toTransform = []byte(val.Format(time.DateOnly))
 	case []byte:
diff --git a/pkg/transformers/greenmask/greenmask_firstname_transformer.go b/pkg/transformers/greenmask/greenmask_firstname_transformer.go
index fa83360..d7f3f5f 100644
--- a/pkg/transformers/greenmask/greenmask_firstname_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_firstname_transformer.go
@@ -10,29 +10,37 @@ import (
 )
 
 type FirstNameTransformer struct {
-	transformer *greenmasktransformers.RandomPersonTransformer
+	transformer   *greenmasktransformers.RandomPersonTransformer
+	dynamicParams map[string]*transformers.DynamicParameter
 }
 
-func NewFirstNameTransformer(params transformers.Parameters) (*FirstNameTransformer, error) {
-	gender, err := findParameter(params, "gender", greenmasktransformers.AnyGenderName)
+const genderParam = "gender"
+
+func NewFirstNameTransformer(params, dynamicParams transformers.Parameters) (*FirstNameTransformer, error) {
+	gender, err := transformers.FindParameterWithDefault(params, genderParam, greenmasktransformers.AnyGenderName)
 	if err != nil {
-		return nil, fmt.Errorf("greenmask_string: symbols must be a string: %w", err)
+		return nil, fmt.Errorf("greenmask_firstname: gender must be a string: %w", err)
 	}
 
-	t := greenmasktransformers.NewRandomPersonTransformer(toGreenmaskGender(gender), nil)
+	dynamicParamMap, err := transformers.ParseDynamicParameters(dynamicParams)
+	if err != nil {
+		return nil, err
+	}
 
+	t := greenmasktransformers.NewRandomPersonTransformer(toGreenmaskGender(gender), nil)
 	if err := setGenerator(t, params); err != nil {
 		return nil, err
 	}
 
 	return &FirstNameTransformer{
-		transformer: t,
+		transformer:   t,
+		dynamicParams: dynamicParamMap,
 	}, nil
 }
 
-func (fnt *FirstNameTransformer) Transform(value any) (any, error) {
+func (fnt *FirstNameTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case string:
 		toTransform = []byte(val)
 	case []byte:
@@ -41,7 +49,18 @@ func (fnt *FirstNameTransformer) Transform(value any) (any, error) {
 		return nil, transformers.ErrUnsupportedValueType
 	}
 
-	ret, err := fnt.transformer.GetFullName("", toTransform)
+	gender := ""
+	// If there's a dynamic parameter defined for the gender, get the value and
+	// pass it to the transformer
+	if genderDynamicParam := fnt.dynamicParams[genderParam]; genderDynamicParam != nil {
+		var err error
+		gender, err = transformers.FindDynamicValue(genderDynamicParam, value.DynamicValues, gender)
+		if err != nil {
+			return nil, fmt.Errorf("invalid value type for dynamic parameter %q", genderParam)
+		}
+	}
+
+	ret, err := fnt.transformer.GetFullName(gender, toTransform)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/transformers/greenmask/greenmask_float_transformer.go b/pkg/transformers/greenmask/greenmask_float_transformer.go
index 21567ac..32a315b 100644
--- a/pkg/transformers/greenmask/greenmask_float_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_float_transformer.go
@@ -49,9 +49,9 @@ func NewFloatTransformer(params transformers.Parameters) (*FloatTransformer, err
 	}, nil
 }
 
-func (ft *FloatTransformer) Transform(value any) (any, error) {
+func (ft *FloatTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case float32:
 		toTransform = getBytesForFloat(float64(val))
 	case float64:
diff --git a/pkg/transformers/greenmask/greenmask_integer_transformer.go b/pkg/transformers/greenmask/greenmask_integer_transformer.go
index 3422431..7a95962 100644
--- a/pkg/transformers/greenmask/greenmask_integer_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_integer_transformer.go
@@ -69,9 +69,9 @@ 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 any) (any, error) {
+func (t *IntegerTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case int:
 		toTransform = generators.BuildBytesFromUint64(uint64(val))
 	case int8:
diff --git a/pkg/transformers/greenmask/greenmask_string_transformer.go b/pkg/transformers/greenmask/greenmask_string_transformer.go
index 2a38d94..3f262dd 100644
--- a/pkg/transformers/greenmask/greenmask_string_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_string_transformer.go
@@ -45,9 +45,9 @@ func NewStringTransformer(params transformers.Parameters) (*StringTransformer, e
 	}, nil
 }
 
-func (st *StringTransformer) Transform(value any) (any, error) {
+func (st *StringTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case string:
 		toTransform = []byte(val)
 	case []byte:
diff --git a/pkg/transformers/greenmask/greenmask_timestamp_transformer.go b/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
index 3febf25..c75ca51 100644
--- a/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
@@ -67,9 +67,9 @@ func NewUTCTimestampTransformer(params transformers.Parameters) (*UTCTimestampTr
 	}, nil
 }
 
-func (t *UTCTimestampTransformer) Transform(value any) (any, error) {
+func (t *UTCTimestampTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case time.Time:
 		toTransform = []byte(val.Format(time.RFC3339))
 	case []byte:
diff --git a/pkg/transformers/greenmask/greenmask_transformer.go b/pkg/transformers/greenmask/greenmask_transformer.go
index 6836930..f6a2e4a 100644
--- a/pkg/transformers/greenmask/greenmask_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_transformer.go
@@ -44,18 +44,7 @@ func getGeneratorType(params transformers.Parameters) (string, error) {
 }
 
 func findParameter[T any](params transformers.Parameters, name string, defaultVal T) (T, error) {
-	var found bool
-	var err error
-
-	var val T
-	val, found, err = transformers.FindParameter[T](params, name)
-	if err != nil {
-		return val, err
-	}
-	if !found {
-		return defaultVal, nil
-	}
-	return val, nil
+	return transformers.FindParameterWithDefault(params, name, defaultVal)
 }
 
 func findParameterArray[T any](params transformers.Parameters, name string, defaultVal []T) ([]T, error) {
diff --git a/pkg/transformers/greenmask/greenmask_uuid_transformer.go b/pkg/transformers/greenmask/greenmask_uuid_transformer.go
index c08ee49..ee2b452 100644
--- a/pkg/transformers/greenmask/greenmask_uuid_transformer.go
+++ b/pkg/transformers/greenmask/greenmask_uuid_transformer.go
@@ -22,9 +22,9 @@ func NewUUIDTransformer(params transformers.Parameters) (*UUIDTransformer, error
 	}, nil
 }
 
-func (ut *UUIDTransformer) Transform(value any) (any, error) {
+func (ut *UUIDTransformer) Transform(value transformers.Value) (any, error) {
 	var toTransform []byte
-	switch val := value.(type) {
+	switch val := value.TransformValue.(type) {
 	case string:
 		parsed, err := uuid.Parse(val)
 		if err != nil {
diff --git a/pkg/transformers/masking_transformer.go b/pkg/transformers/masking_transformer.go
index d95db39..b763832 100644
--- a/pkg/transformers/masking_transformer.go
+++ b/pkg/transformers/masking_transformer.go
@@ -76,8 +76,8 @@ 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 any) (any, error) {
-	switch val := value.(type) {
+func (t *MaskingTransformer) Transform(value Value) (any, error) {
+	switch val := value.TransformValue.(type) {
 	case string:
 		return t.maskingFunction(val), nil
 	case []byte:
diff --git a/pkg/transformers/mocks/mock_transformer.go b/pkg/transformers/mocks/mock_transformer.go
index b9de9c9..3a6e2e2 100644
--- a/pkg/transformers/mocks/mock_transformer.go
+++ b/pkg/transformers/mocks/mock_transformer.go
@@ -2,10 +2,12 @@
 
 package mocks
 
+import "github.com/xataio/pgstream/pkg/transformers"
+
 type Transformer struct {
-	TransformFn func(any) (any, error)
+	TransformFn func(transformers.Value) (any, error)
 }
 
-func (m *Transformer) Transform(val any) (any, error) {
+func (m *Transformer) Transform(val transformers.Value) (any, error) {
 	return m.TransformFn(val)
 }
diff --git a/pkg/transformers/neosync/neosync_transformer.go b/pkg/transformers/neosync/neosync_transformer.go
index 5547a4d..b17c8be 100644
--- a/pkg/transformers/neosync/neosync_transformer.go
+++ b/pkg/transformers/neosync/neosync_transformer.go
@@ -25,8 +25,8 @@ func New[T any](t neosyncTransformer, opts any) *transformer[T] {
 	}
 }
 
-func (t *transformer[T]) Transform(value any) (any, error) {
-	retPtr, err := t.neosyncTransformer.Transform(value, t.opts)
+func (t *transformer[T]) Transform(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 b4fe2ee..e51d84f 100644
--- a/pkg/transformers/phone_number_transformer.go
+++ b/pkg/transformers/phone_number_transformer.go
@@ -16,29 +16,20 @@ type PhoneNumberTransformer struct {
 }
 
 func NewPhoneNumberTransformer(params Parameters) (*PhoneNumberTransformer, error) {
-	prefix, found, err := FindParameter[string](params, "prefix")
+	prefix, err := FindParameterWithDefault(params, "prefix", "")
 	if err != nil {
 		return nil, fmt.Errorf("phone_number: prefix must be a string: %w", err)
 	}
-	if !found {
-		prefix = ""
-	}
 
-	maxLength, found, err := FindParameter[int](params, "max_length")
+	maxLength, err := FindParameterWithDefault(params, "max_length", 10)
 	if err != nil {
 		return nil, fmt.Errorf("phone_number: max_length must be an integer: %w", err)
 	}
-	if !found {
-		maxLength = 10
-	}
 
-	minLength, found, err := FindParameter[int](params, "min_length")
+	minLength, err := FindParameterWithDefault(params, "min_length", 6)
 	if err != nil {
 		return nil, fmt.Errorf("phone_number: min_length must be an integer: %w", err)
 	}
-	if !found {
-		minLength = 6
-	}
 
 	if minLength < 0 {
 		return nil, fmt.Errorf("phone_number: min_length must be greater than 0")
@@ -75,8 +66,8 @@ func NewPhoneNumberTransformer(params Parameters) (*PhoneNumberTransformer, erro
 	}, nil
 }
 
-func (t *PhoneNumberTransformer) Transform(value any) (any, error) {
-	switch v := value.(type) {
+func (t *PhoneNumberTransformer) Transform(value Value) (any, error) {
+	switch v := value.TransformValue.(type) {
 	case string:
 		return t.transform([]byte(v))
 	case []byte:
diff --git a/pkg/transformers/string_transformer.go b/pkg/transformers/string_transformer.go
index 32ad991..bc77392 100644
--- a/pkg/transformers/string_transformer.go
+++ b/pkg/transformers/string_transformer.go
@@ -18,8 +18,8 @@ func NewStringTransformer(params Parameters) (*StringTransformer, error) {
 	return &StringTransformer{}, nil
 }
 
-func (st *StringTransformer) Transform(v any) (any, error) {
-	switch str := v.(type) {
+func (st *StringTransformer) Transform(v Value) (any, error) {
+	switch str := v.TransformValue.(type) {
 	case string:
 		return st.transform(str), nil
 	case []byte:
diff --git a/pkg/transformers/transformer.go b/pkg/transformers/transformer.go
index fbc4d06..bec6e9b 100644
--- a/pkg/transformers/transformer.go
+++ b/pkg/transformers/transformer.go
@@ -8,12 +8,22 @@ import (
 )
 
 type Transformer interface {
-	Transform(any) (any, error)
+	Transform(Value) (any, error)
+}
+
+type Value struct {
+	TransformValue any
+	DynamicValues  map[string]any
+}
+
+type DynamicParameter struct {
+	Column string
 }
 
 type Config struct {
-	Name       TransformerType
-	Parameters Parameters
+	Name              TransformerType
+	Parameters        Parameters
+	DynamicParameters Parameters
 }
 
 type TransformerType string
@@ -37,15 +47,27 @@ const (
 	Masking                TransformerType = "masking"
 )
 
+const (
+	columnDynamicParam = "column"
+)
+
 type Parameters map[string]any
 
 var (
-	ErrUnsupportedValueType   = errors.New("unsupported value type for transformer")
-	ErrUnsupportedGenerator   = errors.New("transformer doesn't support the configured generator")
-	ErrUnsupportedTransformer = errors.New("unsupported transformer config")
-	ErrInvalidParameters      = errors.New("invalid transformer parameters")
+	ErrUnsupportedValueType     = errors.New("unsupported value type for transformer")
+	ErrUnsupportedGenerator     = errors.New("transformer doesn't support the configured generator")
+	ErrUnsupportedTransformer   = errors.New("unsupported transformer config")
+	ErrInvalidParameters        = errors.New("invalid transformer parameters")
+	ErrInvalidDynamicParameters = errors.New("invalid transformer dynamic parameters")
 )
 
+func NewValue(transformValue any, dynamicValues map[string]any) Value {
+	return Value{
+		TransformValue: transformValue,
+		DynamicValues:  dynamicValues,
+	}
+}
+
 func FindParameter[T any](params Parameters, name string) (T, bool, error) {
 	valAny, found := params[name]
 	if !found {
@@ -60,6 +82,17 @@ func FindParameter[T any](params Parameters, name string) (T, bool, error) {
 	return val, true, nil
 }
 
+func FindParameterWithDefault[T any](params Parameters, name string, defaultValue T) (T, error) {
+	val, found, err := FindParameter[T](params, name)
+	if err != nil {
+		return val, err
+	}
+	if !found {
+		return defaultValue, nil
+	}
+	return val, nil
+}
+
 func FindParameterArray[T any](params Parameters, name string) ([]T, bool, error) {
 	// first check if the array is of the expected type
 	arrValue, found, err := FindParameter[[]T](params, name)
@@ -84,3 +117,43 @@ func FindParameterArray[T any](params Parameters, name string) ([]T, bool, error
 
 	return valArray, true, nil
 }
+
+func ParseDynamicParameters(params Parameters) (map[string]*DynamicParameter, error) {
+	dynamicParamMap := make(map[string]*DynamicParameter, len(params))
+	for param, anyVal := range params {
+		if param == "" {
+			return nil, ErrInvalidDynamicParameters
+		}
+
+		dynamicParam, ok := anyVal.(map[string]any)
+		if !ok {
+			return nil, ErrInvalidDynamicParameters
+		}
+
+		column, found, err := FindParameter[string](dynamicParam, columnDynamicParam)
+		if err != nil {
+			return nil, fmt.Errorf("dynamic parameter column must be of type string: %w", err)
+		}
+		if !found {
+			return nil, fmt.Errorf("dynamic parameter must have column field: %w", ErrInvalidDynamicParameters)
+		}
+
+		dynamicParamMap[param] = &DynamicParameter{
+			Column: column,
+		}
+	}
+
+	return dynamicParamMap, nil
+}
+
+func FindDynamicValue[T any](param *DynamicParameter, dynamicValues map[string]any, defaultValue T) (T, error) {
+	dynValue, found, err := FindParameter[T](dynamicValues, param.Column)
+	if err != nil {
+		return dynValue, err
+	}
+	if !found {
+		return defaultValue, nil
+	}
+
+	return dynValue, nil
+}
diff --git a/pkg/wal/processor/transformer/wal_transformer.go b/pkg/wal/processor/transformer/wal_transformer.go
index 964f8b8..d611c74 100644
--- a/pkg/wal/processor/transformer/wal_transformer.go
+++ b/pkg/wal/processor/transformer/wal_transformer.go
@@ -101,7 +101,7 @@ func (t *Transformer) applyTransformations(event *wal.Event) error {
 			continue
 		}
 
-		newValue, err := columnTransformer.Transform(col.Value)
+		newValue, err := columnTransformer.Transform(t.getTransformValue(&col, event.Data.Columns))
 		if err != nil {
 			t.logger.Error(err, "transforming column", loglib.Fields{
 				"severity":    "DATALOSS",
@@ -118,6 +118,17 @@ func (t *Transformer) applyTransformations(event *wal.Event) error {
 	return nil
 }
 
+func (t *Transformer) getTransformValue(column *wal.Column, columns []wal.Column) transformers.Value {
+	values := make(map[string]any, len(columns)-1)
+	for _, col := range columns {
+		if col.Name == column.Name {
+			continue
+		}
+		values[col.Name] = col.Value
+	}
+	return transformers.NewValue(column.Value, values)
+}
+
 func schemaTableKey(schema, table string) string {
 	return schema + "/" + table
 }
@@ -140,7 +151,8 @@ func transformerMapFromRules(rules *Rules) (map[string]columnTransformers, error
 
 func transformerRulesToConfig(rules TransformerRules) *transformers.Config {
 	return &transformers.Config{
-		Name:       transformers.TransformerType(rules.Name),
-		Parameters: rules.Parameters,
+		Name:              transformers.TransformerType(rules.Name),
+		Parameters:        rules.Parameters,
+		DynamicParameters: rules.DynamicParameters,
 	}
 }
diff --git a/pkg/wal/processor/transformer/wal_transformer_rules.go b/pkg/wal/processor/transformer/wal_transformer_rules.go
index 5dcadb6..7d5dd6c 100644
--- a/pkg/wal/processor/transformer/wal_transformer_rules.go
+++ b/pkg/wal/processor/transformer/wal_transformer_rules.go
@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"os"
 
-	"gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v3"
 )
 
 type RulesConfig struct{}
@@ -22,8 +22,9 @@ type TableRules struct {
 }
 
 type TransformerRules struct {
-	Name       string         `yaml:"name"`
-	Parameters map[string]any `yaml:"parameters"`
+	Name              string         `yaml:"name"`
+	Parameters        map[string]any `yaml:"parameters"`
+	DynamicParameters map[string]any `yaml:"dynamic_parameters"`
 }
 
 func readRulesFromFile(filePath string) (*Rules, error) {
diff --git a/transformer_rules.yaml b/transformer_rules.yaml
index 536112e..891cbe7 100644
--- a/transformer_rules.yaml
+++ b/transformer_rules.yaml
@@ -3,7 +3,7 @@ transformations:
     table: test
     column_transformers:
       name:
-        name: neosync_firstname
-        parameters:
-          preserve_length: false
-          max_length: 5
+        name: greenmask_firstname
+        dynamic_parameters:
+          gender:
+            column: sex


Test patch:
diff --git a/pkg/stream/integration/config/integration_test_transformer_rules.yaml b/pkg/stream/integration/config/integration_test_transformer_rules.yaml
index 9e2cce1..0b13a1c 100644
--- a/pkg/stream/integration/config/integration_test_transformer_rules.yaml
+++ b/pkg/stream/integration/config/integration_test_transformer_rules.yaml
@@ -44,8 +44,8 @@ transformations:
       birth_date:
         name: greenmask_date
         parameters:
-          min_value: 1990-01-01
-          max_value: 2000-12-31
+          min_value: "1990-01-01"
+          max_value: "2000-12-31"
       is_active:
         name: greenmask_boolean
       created_at:
@@ -56,8 +56,8 @@ transformations:
       updated_at:
         name: greenmask_utc_timestamp
         parameters:
-          min_timestamp: 2022-01-01T00:00:00Z
-          max_timestamp: 2024-01-01T23:59:59Z
+          min_timestamp: "2022-01-01T00:00:00Z"
+          max_timestamp: "2024-01-01T23:59:59Z"
       gender:
         name: greenmask_choice
         parameters:
diff --git a/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go b/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go
index 0fe1e5f..cbd1e64 100644
--- a/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_boolean_transformer_test.go
@@ -96,7 +96,7 @@ func Test_BooleanTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(tc.input)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
 			require.ErrorIs(t, err, tc.wantErr)
 			if err != nil {
 				return
@@ -107,7 +107,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(tc.input)
+				gotAgain, err := transformer.Transform(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 e21b294..38eb55b 100644
--- a/pkg/transformers/greenmask/greenmask_choice_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_choice_transformer_test.go
@@ -106,7 +106,7 @@ func TestChoiceTransformer_Transform(t *testing.T) {
 			transformer, err := NewChoiceTransformer(tt.params)
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
-			got, err := transformer.Transform(tt.input)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
 			require.Equal(t, tt.wantErr, err)
 			if err != nil {
 				return
@@ -120,7 +120,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(tt.input)
+				gotAgain, err := transformer.Transform(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 546627a..5387a22 100644
--- a/pkg/transformers/greenmask/greenmask_date_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_date_transformer_test.go
@@ -123,7 +123,7 @@ func TestDateTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(tt.input)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
 			require.ErrorIs(t, err, tt.wantErr)
 			if err != nil {
 				return
@@ -143,7 +143,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(tt.input)
+				gotAgain, err := transformer.Transform(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 0dc4cd0..79330c7 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 (
+	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -13,8 +14,9 @@ func Test_NewFirstNameTransformer(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		name   string
-		params transformers.Parameters
+		name          string
+		params        transformers.Parameters
+		dynamicParams transformers.Parameters
 
 		wantErr error
 	}{
@@ -62,13 +64,24 @@ func Test_NewFirstNameTransformer(t *testing.T) {
 
 			wantErr: transformers.ErrUnsupportedGenerator,
 		},
+		{
+			name:   "error - parsing dynamic parameters",
+			params: map[string]any{},
+			dynamicParams: map[string]any{
+				"gender": map[string]any{
+					"invalid": "blob",
+				},
+			},
+
+			wantErr: transformers.ErrInvalidDynamicParameters,
+		},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			_, err := NewFirstNameTransformer(tc.params)
+			_, err := NewFirstNameTransformer(tc.params, tc.dynamicParams)
 			require.ErrorIs(t, err, tc.wantErr)
 		})
 	}
@@ -78,9 +91,11 @@ func TestFirstNameTransformer_Transform(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		name   string
-		value  any
-		params transformers.Parameters
+		name          string
+		value         any
+		dynamicValues map[string]any
+		params        transformers.Parameters
+		dynamicParams transformers.Parameters
 
 		wantName string
 		wantErr  error
@@ -116,6 +131,61 @@ func TestFirstNameTransformer_Transform(t *testing.T) {
 			wantName: "Pearlie",
 			wantErr:  nil,
 		},
+		{
+			name:  "ok - deterministic with dynamic parameter",
+			value: "bob",
+			dynamicValues: map[string]any{
+				"sex": "Male",
+			},
+			params: map[string]any{
+				"generator": deterministic,
+				"gender":    "female",
+			},
+			dynamicParams: map[string]any{
+				"gender": map[string]any{
+					"column": "sex",
+				},
+			},
+
+			wantName: "Domingo",
+			wantErr:  nil,
+		},
+		{
+			name:  "error - invalid dynamic value type",
+			value: "bob",
+			dynamicValues: map[string]any{
+				"sex": 1,
+			},
+			params: map[string]any{
+				"generator": deterministic,
+				"gender":    "female",
+			},
+			dynamicParams: map[string]any{
+				"gender": map[string]any{
+					"column": "sex",
+				},
+			},
+
+			wantErr: errors.New("invalid value type for dynamic parameter \"gender\""),
+		},
+		{
+			name:  "error - invalid dynamic value",
+			value: "bob",
+			dynamicValues: map[string]any{
+				"sex": "Banana",
+			},
+			params: map[string]any{
+				"generator": deterministic,
+				"gender":    "female",
+			},
+			dynamicParams: map[string]any{
+				"gender": map[string]any{
+					"column": "sex",
+				},
+			},
+
+			wantErr: errors.New("unable to match gender \"Banana\""),
+		},
 		{
 			name:  "error - unsupported value type",
 			value: 1,
@@ -132,11 +202,11 @@ func TestFirstNameTransformer_Transform(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			transformer, err := NewFirstNameTransformer(tc.params)
+			transformer, err := NewFirstNameTransformer(tc.params, tc.dynamicParams)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(tc.value)
-			require.ErrorIs(t, err, tc.wantErr)
+			got, err := transformer.Transform(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 5419814..f62d904 100644
--- a/pkg/transformers/greenmask/greenmask_float_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_float_transformer_test.go
@@ -184,7 +184,7 @@ func TestFloatTransformer_Transform(t *testing.T) {
 			transformer, err := NewFloatTransformer(tc.params)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(tc.value)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			if err != nil {
 				return
@@ -207,7 +207,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(tc.value)
+				gotAgain, err := transformer.Transform(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 78bc298..f26bc6b 100644
--- a/pkg/transformers/greenmask/greenmask_integer_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_integer_transformer_test.go
@@ -255,7 +255,7 @@ func TestIntegerTransformer_Transform(t *testing.T) {
 			transformer, err := NewIntegerTransformer(tt.params)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(tt.input)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
 			require.ErrorIs(t, err, tt.wantErr)
 			if err != nil {
 				return
@@ -280,7 +280,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(tt.input)
+				gotAgain, err := transformer.Transform(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 7230a64..a14373e 100644
--- a/pkg/transformers/greenmask/greenmask_string_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_string_transformer_test.go
@@ -161,7 +161,7 @@ func TestStringTransformer_Transform(t *testing.T) {
 			transformer, err := NewStringTransformer(tc.params)
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(tc.value)
+			got, err := transformer.Transform(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 25cdf00..76e4d82 100644
--- a/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
@@ -181,7 +181,7 @@ func TestUTCTimestampTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(tt.input)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
 			require.ErrorIs(t, err, tt.wantErr)
 			if err != nil {
 				return
@@ -234,7 +234,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(tt.input)
+				gotAgain, err := transformer.Transform(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 61fa681..8e54ca3 100644
--- a/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_unix_timestamp_transformer_test.go
@@ -116,7 +116,7 @@ func TestUnixTimestampTransformer_Transform(t *testing.T) {
 			t.Parallel()
 			transformer, err := NewUnixTimestampTransformer(tt.params)
 			require.NoError(t, err)
-			got, err := transformer.Transform(tt.input)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tt.input})
 			require.NoError(t, err)
 			require.NotNil(t, got)
 
@@ -135,7 +135,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(tt.input)
+				gotAgain, err := transformer.Transform(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 5509ca5..93ab85c 100644
--- a/pkg/transformers/greenmask/greenmask_uuid_transformer_test.go
+++ b/pkg/transformers/greenmask/greenmask_uuid_transformer_test.go
@@ -109,7 +109,7 @@ func Test_UUIDTransformer_Transform(t *testing.T) {
 			require.NoError(t, err)
 			require.NotNil(t, transformer)
 
-			got, err := transformer.Transform(tc.input)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
 			if !errors.Is(err, tc.wantErr) {
 				require.Error(t, err, tc.wantErr.Error())
 			}
@@ -120,7 +120,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(tc.input)
+				gotAgain, err := transformer.Transform(transformers.Value{TransformValue: tc.input})
 				require.NoError(t, err)
 				require.Equal(t, got, gotAgain)
 			}
diff --git a/pkg/transformers/masking_transformer_test.go b/pkg/transformers/masking_transformer_test.go
index ccce4a5..a28a515 100644
--- a/pkg/transformers/masking_transformer_test.go
+++ b/pkg/transformers/masking_transformer_test.go
@@ -170,7 +170,7 @@ func TestMaskingTransformer_Transform(t *testing.T) {
 			t.Parallel()
 			mt, err := NewMaskingTransformer(tt.params)
 			require.NoError(t, err)
-			got, err := mt.Transform(tt.input)
+			got, err := mt.Transform(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 f8617b0..159ee8b 100644
--- a/pkg/transformers/neosync/neosync_email_transformer_test.go
+++ b/pkg/transformers/neosync/neosync_email_transformer_test.go
@@ -184,7 +184,7 @@ func TestEmailTransformer_Transform(t *testing.T) {
 			}
 			transformer, err := NewEmailTransformer(params)
 			require.NoError(t, err)
-			got, err := transformer.Transform(tc.input)
+			got, err := transformer.Transform(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 72796c2..1afb5a4 100644
--- a/pkg/transformers/neosync/neosync_firstname_transformer_test.go
+++ b/pkg/transformers/neosync/neosync_firstname_transformer_test.go
@@ -75,7 +75,7 @@ func TestFirstnameTransformer_Transform(t *testing.T) {
 				return
 			}
 
-			got, err := transformer.Transform(tc.value)
+			got, err := transformer.Transform(transformers.Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			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 23d4bf5..0e2cf91 100644
--- a/pkg/transformers/neosync/neosync_string_transformer_test.go
+++ b/pkg/transformers/neosync/neosync_string_transformer_test.go
@@ -86,7 +86,7 @@ func TestStringTransformer_Transform(t *testing.T) {
 				return
 			}
 
-			got, err := transformer.Transform(tc.value)
+			got, err := transformer.Transform(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 a739b2f..defc423 100644
--- a/pkg/transformers/phone_number_transformer_test.go
+++ b/pkg/transformers/phone_number_transformer_test.go
@@ -86,7 +86,7 @@ func TestPhoneNumberTransformer_Transform(t *testing.T) {
 			}
 			require.NoError(t, err)
 
-			got, err := transformer.Transform(tc.value)
+			got, err := transformer.Transform(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 8811d86..8fc269b 100644
--- a/pkg/transformers/string_transformer_test.go
+++ b/pkg/transformers/string_transformer_test.go
@@ -47,7 +47,7 @@ func TestStringTransformer_Transform(t *testing.T) {
 
 			st, err := NewStringTransformer(nil)
 			require.NoError(t, err)
-			got, err := st.Transform(tc.value)
+			got, err := st.Transform(Value{TransformValue: tc.value})
 			require.ErrorIs(t, err, tc.wantErr)
 			if tc.wantErr != nil {
 				return
diff --git a/pkg/transformers/transformer_test.go b/pkg/transformers/transformer_test.go
index 3646016..0276822 100644
--- a/pkg/transformers/transformer_test.go
+++ b/pkg/transformers/transformer_test.go
@@ -67,6 +67,64 @@ func Test_FindParameter(t *testing.T) {
 	}
 }
 
+func Test_FindParameterWithDefault(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name         string
+		params       Parameters
+		paramName    string
+		defaultValue int
+
+		wantValue int
+		wantErr   error
+	}{
+		{
+			name: "ok - parameter found",
+			params: map[string]any{
+				"test": 42,
+			},
+			paramName:    "test",
+			defaultValue: 0,
+
+			wantValue: 42,
+			wantErr:   nil,
+		},
+		{
+			name: "ok - parameter not found, use default",
+			params: map[string]any{
+				"test": 42,
+			},
+			paramName:    "another",
+			defaultValue: 99,
+
+			wantValue: 99,
+			wantErr:   nil,
+		},
+		{
+			name: "error - invalid parameter type",
+			params: map[string]any{
+				"test": "invalid",
+			},
+			paramName:    "test",
+			defaultValue: 0,
+
+			wantValue: 0,
+			wantErr:   ErrInvalidParameters,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := FindParameterWithDefault(tc.params, tc.paramName, tc.defaultValue)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantValue, got)
+		})
+	}
+}
+
 func Test_FindParameterArray(t *testing.T) {
 	t.Parallel()
 
@@ -147,3 +205,139 @@ func Test_FindParameterArray(t *testing.T) {
 		})
 	}
 }
+
+func Test_ParseDynamicParameters(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		params  Parameters
+		want    map[string]*DynamicParameter
+		wantErr error
+	}{
+		{
+			name: "ok - valid dynamic parameters",
+			params: map[string]any{
+				"param1": map[string]any{
+					"column": "col1",
+				},
+				"param2": map[string]any{
+					"column": "col2",
+				},
+			},
+			want: map[string]*DynamicParameter{
+				"param1": {Column: "col1"},
+				"param2": {Column: "col2"},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - empty parameter name",
+			params: map[string]any{
+				"": map[string]any{
+					"column": "col1",
+				},
+			},
+			want:    nil,
+			wantErr: ErrInvalidDynamicParameters,
+		},
+		{
+			name: "error - invalid parameter type",
+			params: map[string]any{
+				"param1": "invalid",
+			},
+			want:    nil,
+			wantErr: ErrInvalidDynamicParameters,
+		},
+		{
+			name: "error - missing column field",
+			params: map[string]any{
+				"param1": map[string]any{},
+			},
+			want:    nil,
+			wantErr: ErrInvalidDynamicParameters,
+		},
+		{
+			name: "error - column field wrong type",
+			params: map[string]any{
+				"param1": map[string]any{
+					"column": 123,
+				},
+			},
+			want:    nil,
+			wantErr: ErrInvalidParameters,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := ParseDynamicParameters(tc.params)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.want, got)
+		})
+	}
+}
+
+func Test_FindDynamicValue(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name          string
+		param         *DynamicParameter
+		dynamicValues map[string]any
+		defaultValue  int
+
+		wantValue int
+		wantErr   error
+	}{
+		{
+			name: "ok - value found",
+			param: &DynamicParameter{
+				Column: "col1",
+			},
+			dynamicValues: map[string]any{
+				"col1": 42,
+			},
+			defaultValue: 0,
+
+			wantValue: 42,
+			wantErr:   nil,
+		},
+		{
+			name: "ok - value not found, use default",
+			param: &DynamicParameter{
+				Column: "col1",
+			},
+			dynamicValues: map[string]any{},
+			defaultValue:  99,
+
+			wantValue: 99,
+			wantErr:   nil,
+		},
+		{
+			name: "error - invalid value type",
+			param: &DynamicParameter{
+				Column: "col1",
+			},
+			dynamicValues: map[string]any{
+				"col1": "invalid",
+			},
+			defaultValue: 0,
+
+			wantValue: 0,
+			wantErr:   ErrInvalidParameters,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := FindDynamicValue(tc.param, tc.dynamicValues, tc.defaultValue)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantValue, got)
+		})
+	}
+}
diff --git a/pkg/wal/processor/transformer/wal_transformer_test.go b/pkg/wal/processor/transformer/wal_transformer_test.go
index 2cedba9..e0f0396 100644
--- a/pkg/wal/processor/transformer/wal_transformer_test.go
+++ b/pkg/wal/processor/transformer/wal_transformer_test.go
@@ -138,8 +138,8 @@ func TestTransformer_ProcessWALEvent(t *testing.T) {
 			transformerMap: map[string]columnTransformers{
 				testKey: {
 					"column_1": &transformermocks.Transformer{
-						TransformFn: func(a any) (any, error) {
-							aStr, ok := a.(string)
+						TransformFn: func(a transformers.Value) (any, error) {
+							aStr, ok := a.TransformValue.(string)
 							require.True(t, ok)
 							require.Equal(t, "one", aStr)
 							return "two", nil
@@ -169,7 +169,7 @@ func TestTransformer_ProcessWALEvent(t *testing.T) {
 			transformerMap: map[string]columnTransformers{
 				testKey: {
 					"column_1": &transformermocks.Transformer{
-						TransformFn: func(a any) (any, error) {
+						TransformFn: func(a transformers.Value) (any, error) {
 							return nil, errors.New("TransformFn: should not be called")
 						},
 					},
@@ -197,7 +197,7 @@ func TestTransformer_ProcessWALEvent(t *testing.T) {
 			transformerMap: map[string]columnTransformers{
 				testKey: {
 					"column_1": &transformermocks.Transformer{
-						TransformFn: func(a any) (any, error) {
+						TransformFn: func(a transformers.Value) (any, error) {
 							return nil, errTest
 						},
 					},