Repository: xataio/pgstream
Pull request #288 (2025-04-17 14:12:33)

Add noop transformers support and Strict validation mode
Single PR for both adding noop transformer support and strict validation mode.
Parsing the config, if no validation mode is not provided, the default will be `relaxed` mode. If validation mode is set to `strict` we'll be validating that all columns for that particular table are present in the transformer config, by comparing them to the column list we get from postgres.

fixes: #259 
fixes: #260 

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 81.82% (**-0.29%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.95% (**+0.25%**) | 63 (+6) | 51 (+5) | 12 (+1) | :thumbsup: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 83.33% (**-0.88%**) | 36 (-2) | 30 (-2) | 6 | :thumbsdown: |

_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/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 81.82% (**-0.29%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.95% (**+0.25%**) | 63 (+6) | 51 (+5) | 12 (+1) | :thumbsup: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 83.33% (**-0.88%**) | 36 (-2) | 30 (-2) | 6 | :thumbsdown: |

_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/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 81.82% (**-0.29%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.95% (**+0.25%**) | 63 (+6) | 51 (+5) | 12 (+1) | :thumbsup: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 83.33% (**-0.88%**) | 36 (-2) | 30 (-2) | 6 | :thumbsdown: |

_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/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 82.03% (**-1.81%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 80.39% (**-1.71%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 89.44% (ø) | 161 | 144 | 17 |  |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 84.25% (**-4.90%**) | 146 (+17) | 123 (+8) | 23 (+9) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.00% (**-0.70%**) | 65 (+8) | 52 (+6) | 13 (+2) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 0.00% (ø) | 0 | 0 | 0 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 81.08% (**-3.13%**) | 37 (-1) | 30 (-2) | 7 (+1) | :thumbsdown: |

_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/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 82.03% (**-1.81%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 80.39% (**-1.71%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 89.44% (ø) | 161 | 144 | 17 |  |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 84.25% (**-4.90%**) | 146 (+17) | 123 (+8) | 23 (+9) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.00% (**-0.70%**) | 65 (+8) | 52 (+6) | 13 (+2) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 0.00% (ø) | 0 | 0 | 0 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 81.08% (**-3.13%**) | 37 (-1) | 30 (-2) | 7 (+1) | :thumbsdown: |

_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/stream/integration/helper_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 82.03% (**-1.81%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 80.39% (**-1.71%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 89.44% (ø) | 161 | 144 | 17 |  |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 84.25% (**-4.90%**) | 146 (+17) | 123 (+8) | 23 (+9) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.00% (**-0.70%**) | 65 (+8) | 52 (+6) | 13 (+2) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 0.00% (ø) | 0 | 0 | 0 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 81.08% (**-3.13%**) | 37 (-1) | 30 (-2) | 7 (+1) | :thumbsdown: |

_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/stream/integration/helper_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 82.03% (**-1.81%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 81.37% (**-0.73%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 89.44% (ø) | 161 | 144 | 17 |  |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 84.25% (**-4.90%**) | 146 (+17) | 123 (+8) | 23 (+9) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.00% (**-0.70%**) | 65 (+8) | 52 (+6) | 13 (+2) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 0.00% (ø) | 0 | 0 | 0 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 83.78% (**-0.43%**) | 37 (-1) | 31 (-1) | 6 | :thumbsdown: |

_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/stream/integration/helper_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 82.03% (**-1.81%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 81.37% (**-0.73%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 89.44% (ø) | 161 | 144 | 17 |  |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 84.25% (**-4.90%**) | 146 (+17) | 123 (+8) | 23 (+9) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.00% (**-0.70%**) | 65 (+8) | 52 (+6) | 13 (+2) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 0.00% (ø) | 0 | 0 | 0 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 83.78% (**-0.43%**) | 37 (-1) | 31 (-1) | 6 | :thumbsdown: |

_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/stream/integration/helper_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 81.71% (**-1.83%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 78.85% (**-0.54%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 89.03% (ø) | 155 | 138 | 17 |  |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 84.25% (**-4.90%**) | 146 (+17) | 123 (+8) | 23 (+9) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 80.00% (**-0.70%**) | 65 (+8) | 52 (+6) | 13 (+2) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 0.00% (ø) | 0 | 0 | 0 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 76.92% (**-0.58%**) | 39 (-1) | 30 (-1) | 9 | :thumbsdown: |

_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/stream/integration/helper_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 81.71% (**-1.83%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/stream | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 74.11% (**-5.27%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 89.03% (ø) | 155 | 138 | 17 |  |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 84.25% (**-4.90%**) | 146 (+17) | 123 (+8) | 23 (+9) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/stream/stream_run.go | 0.00% (ø) | 169 | 0 | 169 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer.go | 81.36% (**+0.65%**) | 59 (+2) | 48 (+2) | 11 | :thumbsup: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_rules.go | 0.00% (ø) | 0 | 0 | 0 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 66.04% (**-11.46%**) | 53 (+13) | 35 (+4) | 18 (+9) | :skull:  |

_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/stream/integration/helper_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_test.go
- github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator_test.go

</details>


Patch:
diff --git a/cmd/config/config_env.go b/cmd/config/config_env.go
index 06376b5..7b0e6e6 100644
--- a/cmd/config/config_env.go
+++ b/cmd/config/config_env.go
@@ -424,7 +424,7 @@ func parseTransformerConfig() (*transformer.Config, error) {
 		return nil, fmt.Errorf("in file %q: %w", filename, err)
 	}
 
-	return yamlConfig.Transformations.parseTransformationConfig(), nil
+	return yamlConfig.Transformations.parseTransformationConfig()
 }
 
 func parseFilterConfig() *filter.Config {
diff --git a/cmd/config/config_yaml.go b/cmd/config/config_yaml.go
index 0e50580..0a43988 100644
--- a/cmd/config/config_yaml.go
+++ b/cmd/config/config_yaml.go
@@ -212,12 +212,15 @@ type FilterConfig struct {
 	ExcludeTables []string `mapstructure:"exclude_tables" yaml:"exclude_tables"`
 }
 
-type TransformationsConfig []TransformationConfig
-
-type TransformationConfig struct {
-	Schema      string                              `mapstructure:"schema" yaml:"schema"`
-	Table       string                              `mapstructure:"table" yaml:"table"`
-	ColumnRules map[string]ColumnTransformersConfig `mapstructure:"column_transformers" yaml:"column_transformers"`
+type TransformationsConfig struct {
+	TransformerRules []TableTransformersConfig `mapstructure:"table_transformers" yaml:"table_transformers"`
+	ValidationMode   string                    `mapstructure:"validation_mode" yaml:"validation_mode"`
+}
+type TableTransformersConfig struct {
+	Schema         string                              `mapstructure:"schema" yaml:"schema"`
+	Table          string                              `mapstructure:"table" yaml:"table"`
+	ColumnRules    map[string]ColumnTransformersConfig `mapstructure:"column_transformers" yaml:"column_transformers"`
+	ValidationMode string                              `mapstructure:"validation_mode" yaml:"validation_mode"`
 }
 
 type ColumnTransformersConfig struct {
@@ -252,14 +255,24 @@ const (
 	opensearchEngine    = "opensearch"
 )
 
+// transformer validation modes
+const (
+	strictValidationMode     = "strict"
+	tableLevelValidationMode = "table_level"
+	relaxedValidationMode    = "relaxed"
+)
+
 var (
-	errUnsupportedSchemaSnapshotMode = errors.New("unsupported schema snapshot mode, must be one of 'pgdump_pgrestore' or 'schemalog'")
-	errUnsupportedSnapshotMode       = errors.New("unsupported snapshot mode, must be one of 'full', 'schema' or 'data'")
-	errUnsupportedPostgresSourceMode = errors.New("unsupported postgres source mode, must be one of 'replication', 'snapshot' or 'snapshot_and_replication'")
-	errUnsupportedSearchEngine       = errors.New("unsupported search engine, must be one of 'opensearch' or 'elasticsearch'")
-	errInvalidPgdumpPgrestoreConfig  = errors.New("pgdump_pgrestore snapshot mode requires target postgres config")
-	errInvalidInjectorConfig         = errors.New("injector config can't infer schemalog url from source postgres url, schemalog_url must be provided")
-	errInvalidSnapshotRecorderConfig = errors.New("snapshot recorder config requires a postgres url")
+	errUnsupportedSchemaSnapshotMode           = errors.New("unsupported schema snapshot mode, must be one of 'pgdump_pgrestore' or 'schemalog'")
+	errUnsupportedSnapshotMode                 = errors.New("unsupported snapshot mode, must be one of 'full', 'schema' or 'data'")
+	errUnsupportedPostgresSourceMode           = errors.New("unsupported postgres source mode, must be one of 'replication', 'snapshot' or 'snapshot_and_replication'")
+	errUnsupportedTransformationValidationMode = errors.New("unsupported transformation validation mode, must be one of 'strict', 'table_level' or 'relaxed'")
+	errUnsupportedTableValidationMode          = errors.New("unsupported table level validation mode, must be either 'strict' or 'relaxed'")
+	errInvalidTableValidationConfig            = errors.New("table level validation mode should be used when transformation validation mode is set to 'table_level'")
+	errUnsupportedSearchEngine                 = errors.New("unsupported search engine, must be one of 'opensearch' or 'elasticsearch'")
+	errInvalidPgdumpPgrestoreConfig            = errors.New("pgdump_pgrestore snapshot mode requires target postgres config")
+	errInvalidInjectorConfig                   = errors.New("injector config can't infer schemalog url from source postgres url, schemalog_url must be provided")
+	errInvalidSnapshotRecorderConfig           = errors.New("snapshot recorder config requires a postgres url")
 )
 
 func (c *YAMLConfig) toStreamConfig() (*stream.Config, error) {
@@ -292,11 +305,10 @@ func (c *YAMLConfig) parseListenerConfig() (stream.ListenerConfig, error) {
 
 func (c *YAMLConfig) parseProcessorConfig() (stream.ProcessorConfig, error) {
 	streamCfg := stream.ProcessorConfig{
-		Kafka:       c.parseKafkaProcessorConfig(),
-		Postgres:    c.parsePostgresProcessorConfig(),
-		Webhook:     c.parseWebhookProcessorConfig(),
-		Transformer: c.parseTransformationConfig(),
-		Filter:      c.parseFilterConfig(),
+		Kafka:    c.parseKafkaProcessorConfig(),
+		Postgres: c.parsePostgresProcessorConfig(),
+		Webhook:  c.parseWebhookProcessorConfig(),
+		Filter:   c.parseFilterConfig(),
 	}
 
 	var err error
@@ -310,6 +322,11 @@ func (c *YAMLConfig) parseProcessorConfig() (stream.ProcessorConfig, error) {
 		return stream.ProcessorConfig{}, err
 	}
 
+	streamCfg.Transformer, err = c.parseTransformationConfig()
+	if err != nil {
+		return stream.ProcessorConfig{}, err
+	}
+
 	return streamCfg, nil
 }
 
@@ -546,9 +563,9 @@ func (c *YAMLConfig) parseWebhookProcessorConfig() *stream.WebhookProcessorConfi
 	return streamCfg
 }
 
-func (c *YAMLConfig) parseTransformationConfig() *transformer.Config {
-	if c.Modifiers.Transformations == nil {
-		return nil
+func (c *YAMLConfig) parseTransformationConfig() (*transformer.Config, error) {
+	if c.Modifiers.Transformations.TransformerRules == nil {
+		return nil, nil
 	}
 
 	return c.Modifiers.Transformations.parseTransformationConfig()
@@ -564,12 +581,39 @@ func (c YAMLConfig) parseFilterConfig() *filter.Config {
 	}
 }
 
-func (c TransformationsConfig) parseTransformationConfig() *transformer.Config {
-	if len(c) == 0 {
-		return nil
+func (c TransformationsConfig) parseTransformationConfig() (*transformer.Config, error) {
+	if len(c.TransformerRules) == 0 {
+		return nil, nil
+	}
+
+	var globalValidationMode string
+	switch c.ValidationMode {
+	case strictValidationMode, relaxedValidationMode, tableLevelValidationMode:
+		globalValidationMode = c.ValidationMode
+	case "":
+		globalValidationMode = relaxedValidationMode
+	default:
+		return nil, errUnsupportedTransformationValidationMode
 	}
-	rules := make([]transformer.TableRules, 0, len(c))
-	for _, t := range c {
+
+	rules := make([]transformer.TableRules, 0, len(c.TransformerRules))
+	for _, t := range c.TransformerRules {
+		var tableValidationMode string
+		switch t.ValidationMode {
+		case strictValidationMode, relaxedValidationMode:
+			if globalValidationMode != tableLevelValidationMode {
+				return nil, errInvalidTableValidationConfig
+			}
+			tableValidationMode = t.ValidationMode
+		case "":
+			if globalValidationMode == tableLevelValidationMode {
+				return nil, errInvalidTableValidationConfig
+			}
+			tableValidationMode = globalValidationMode
+		default:
+			return nil, errUnsupportedTableValidationMode
+		}
+
 		columnRules := make(map[string]transformer.TransformerRules, len(t.ColumnRules))
 		for column, cr := range t.ColumnRules {
 			columnRules[column] = transformer.TransformerRules{
@@ -579,15 +623,16 @@ func (c TransformationsConfig) parseTransformationConfig() *transformer.Config {
 			}
 		}
 		rules = append(rules, transformer.TableRules{
-			Schema:      t.Schema,
-			Table:       t.Table,
-			ColumnRules: columnRules,
+			Schema:         t.Schema,
+			Table:          t.Table,
+			ColumnRules:    columnRules,
+			ValidationMode: tableValidationMode,
 		})
 	}
 
 	return &transformer.Config{
 		TransformerRules: rules,
-	}
+	}, nil
 }
 
 func (c *KafkaConfig) parseKafkaListenerConfig() *stream.KafkaListenerConfig {
diff --git a/pkg/stream/stream_run.go b/pkg/stream/stream_run.go
index b86a645..400bd5b 100644
--- a/pkg/stream/stream_run.go
+++ b/pkg/stream/stream_run.go
@@ -233,12 +233,12 @@ func Run(ctx context.Context, logger loglib.Logger, config *Config, instrumentat
 			pgURL = config.Listener.Snapshot.Generator.URL
 		}
 		if pgURL != "" {
-			pgValidator, err := transformer.NewPostgresTransformerValidator(ctx, pgURL)
+			pgParser, err := transformer.NewPostgresTransformerParser(ctx, pgURL)
 			if err != nil {
 				return fmt.Errorf("error creating postgres transformer validator: %w", err)
 			}
-			defer pgValidator.Close()
-			opts = append(opts, transformer.WithValidator(pgValidator.Validate))
+			defer pgParser.Close()
+			opts = append(opts, transformer.WithParser(pgParser.ParseAndValidate))
 		}
 		transformer, err := transformer.New(ctx, config.Processor.Transformer, processor, opts...)
 		if err != nil {
diff --git a/pkg/wal/processor/transformer/wal_transformer.go b/pkg/wal/processor/transformer/wal_transformer.go
index 5afb81f..48fbc0d 100644
--- a/pkg/wal/processor/transformer/wal_transformer.go
+++ b/pkg/wal/processor/transformer/wal_transformer.go
@@ -4,6 +4,7 @@ package transformer
 
 import (
 	"context"
+	"errors"
 
 	pglib "github.com/xataio/pgstream/internal/postgres"
 	loglib "github.com/xataio/pgstream/pkg/log"
@@ -19,10 +20,10 @@ type Transformer struct {
 	logger         loglib.Logger
 	processor      processor.Processor
 	transformerMap map[string]ColumnTransformers
-	validator      ValidatorFn
+	parser         ParseFn
 }
 
-type ValidatorFn func(ctx context.Context, transformerMap map[string]ColumnTransformers) error
+type ParseFn func(rules []TableRules) (map[string]ColumnTransformers, error)
 
 type ColumnTransformers map[string]transformers.Transformer
 
@@ -32,28 +33,27 @@ type Config struct {
 
 type Option func(t *Transformer)
 
+const validationModeStrict = "strict"
+
+var errValidatorRequiredForStrictMode = errors.New("strict validation mode requires a validator function")
+
 // 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) {
-	transformerMap, err := transformerMapFromRules(cfg.TransformerRules)
-	if err != nil {
-		return nil, err
-	}
-
 	t := &Transformer{
-		logger:         loglib.NewNoopLogger(),
-		transformerMap: transformerMap,
-		processor:      processor,
+		logger:    loglib.NewNoopLogger(),
+		processor: processor,
+		parser:    transformerMapFromRules,
 	}
 
 	for _, opt := range opts {
 		opt(t)
 	}
 
-	if t.validator != nil {
-		if err := t.validator(ctx, t.transformerMap); err != nil {
-			return nil, err
-		}
+	var err error
+	t.transformerMap, err = t.parser(cfg.TransformerRules)
+	if err != nil {
+		return nil, err
 	}
 
 	return t, nil
@@ -67,9 +67,9 @@ func WithLogger(l loglib.Logger) Option {
 	}
 }
 
-func WithValidator(validator ValidatorFn) Option {
+func WithParser(parser ParseFn) Option {
 	return func(in *Transformer) {
-		in.validator = validator
+		in.parser = parser
 	}
 }
 
@@ -148,11 +148,18 @@ func transformerMapFromRules(rules []TableRules) (map[string]ColumnTransformers,
 	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 {
-			schemaTableTransformers[colName], err = builder.New(transformerRulesToConfig(transformerRules))
-			if err != nil {
+			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
 			}
 		}
diff --git a/pkg/wal/processor/transformer/wal_transformer_rules.go b/pkg/wal/processor/transformer/wal_transformer_rules.go
index 29b01df..95b9941 100644
--- a/pkg/wal/processor/transformer/wal_transformer_rules.go
+++ b/pkg/wal/processor/transformer/wal_transformer_rules.go
@@ -7,9 +7,10 @@ type Rules struct {
 }
 
 type TableRules struct {
-	Schema      string                      `yaml:"schema"`
-	Table       string                      `yaml:"table"`
-	ColumnRules map[string]TransformerRules `yaml:"column_transformers"`
+	Schema         string                      `yaml:"schema"`
+	Table          string                      `yaml:"table"`
+	ColumnRules    map[string]TransformerRules `yaml:"column_transformers"`
+	ValidationMode string                      `yaml:"validation_mode"`
 }
 
 type TransformerRules struct {
diff --git a/pkg/wal/processor/transformer/wal_transformer_validator.go b/pkg/wal/processor/transformer/wal_transformer_validator.go
index 75deed8..9c493eb 100644
--- a/pkg/wal/processor/transformer/wal_transformer_validator.go
+++ b/pkg/wal/processor/transformer/wal_transformer_validator.go
@@ -10,65 +10,88 @@ 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 PostgresTransformerValidator struct {
+type PostgresTransformerParser struct {
 	conn pglib.Querier
 }
 
 const fieldDescriptionsQuery = "SELECT * FROM %s LIMIT 0"
 
-func NewPostgresTransformerValidator(ctx context.Context, pgURL string) (*PostgresTransformerValidator, error) {
+func NewPostgresTransformerParser(ctx context.Context, pgURL string) (*PostgresTransformerParser, error) {
 	pool, err := pglib.NewConnPool(ctx, pgURL)
 	if err != nil {
 		return nil, err
 	}
-	return &PostgresTransformerValidator{
+	return &PostgresTransformerParser{
 		conn: pool,
 	}, nil
 }
 
-func (v *PostgresTransformerValidator) Validate(ctx context.Context, transformerMap map[string]ColumnTransformers) error {
-	for schemaTable, columnTransformers := range transformerMap {
-		fieldDescriptions, err := v.getFieldDescriptions(ctx, schemaTable)
+func (v *PostgresTransformerParser) ParseAndValidate(rules []TableRules) (map[string]ColumnTransformers, error) {
+	transformerMap := map[string]ColumnTransformers{}
+	for _, table := range rules {
+		tableKey := schemaTableKey(table.Schema, table.Table)
+		fieldDescriptions, err := v.getFieldDescriptions(context.Background(), tableKey)
 		if err != nil {
-			return err
+			return nil, err
 		}
-		// TODO: maybe error out if len(fieldDescriptions) != len(columnTransformers)
-		// if we start requiring a transformer for every column (noop transformers)
 
-		// map column names to column pg type OIDs, skip columns that don't have a transformer
-		mappedColumns := make(map[string]uint32, len(fieldDescriptions))
+		// map column names to column pg type OIDs
+		mappedColumnTypes := make(map[string]uint32, len(fieldDescriptions))
 		for _, desc := range fieldDescriptions {
-			if _, found := columnTransformers[string(desc.Name)]; !found {
+			if _, found := table.ColumnRules[string(desc.Name)]; !found {
+				// column is not configured in rules, error out if strict validation mode is enabled
+				if table.ValidationMode == validationModeStrict {
+					return nil, fmt.Errorf("column %s of table %s has no transformer configured", desc.Name, tableKey)
+				}
 				continue
 			}
-
-			mappedColumns[string(desc.Name)] = desc.DataTypeOID
+			mappedColumnTypes[string(desc.Name)] = desc.DataTypeOID
 		}
 
-		// check that all column transformers are compatible with corresponding column types
-		for colName, tr := range columnTransformers {
-			datatype, found := mappedColumns[colName]
+		schemaTableTransformers := make(map[string]transformers.Transformer)
+		transformerMap[tableKey] = schemaTableTransformers
+		for colName, transformerRules := range table.ColumnRules {
+			// get the data type so that we can later validate if it's compatible with the configured transformer
+			datatype, found := mappedColumnTypes[colName]
 			if !found {
-				// validate that all column in the rules are present in the table
-				return fmt.Errorf("column %s not found in table %s", colName, schemaTable)
+				// validate that the column in the rules is present in the table
+				return nil, fmt.Errorf("column %s not found in table %s", colName, tableKey)
+			}
+
+			cfg := transformerRulesToConfig(transformerRules)
+
+			// skip if noop transformer
+			if cfg.Name == "" || cfg.Name == "noop" {
+				continue
 			}
-			if !pgTypeCompatibleWithTransformerType(tr.CompatibleTypes(), datatype) {
-				return fmt.Errorf("transformer specified for column '%s' in table %s does not support pg data type with oid: %d", colName, schemaTable, datatype)
+
+			// build the transformer
+			transformer, err := builder.New(cfg)
+			if err != nil {
+				return nil, err
+			}
+
+			// validate that the transformer is compatible with the column type
+			if !pgTypeCompatibleWithTransformerType(transformer.CompatibleTypes(), datatype) {
+				return nil, fmt.Errorf("transformer specified for column '%s' in table %s does not support pg data type with oid: %d", colName, tableKey, datatype)
 			}
+
+			// add the transformer to the map
+			schemaTableTransformers[colName] = transformer
 		}
 	}
-
-	return nil
+	return transformerMap, nil
 }
 
-func (v *PostgresTransformerValidator) Close() error {
+func (v *PostgresTransformerParser) Close() error {
 	return v.conn.Close(context.Background())
 }
 
-func (v *PostgresTransformerValidator) getFieldDescriptions(ctx context.Context, schemaTable string) ([]pgconn.FieldDescription, error) {
+func (v *PostgresTransformerParser) getFieldDescriptions(ctx context.Context, schemaTable string) ([]pgconn.FieldDescription, error) {
 	query := fmt.Sprintf(fieldDescriptionsQuery, schemaTable)
 	rows, err := v.conn.Query(ctx, query)
 	if err != nil {


Test patch:
diff --git a/cmd/config/test/test_config.yaml b/cmd/config/test/test_config.yaml
index 142bb5a..d3f1cf9 100644
--- a/cmd/config/test/test_config.yaml
+++ b/cmd/config/test/test_config.yaml
@@ -110,11 +110,13 @@ modifiers:
      - "excluded_schema.test"
      - "another_excluded_schema.*"
   transformations:
-  - schema: public
-    table: test
-    column_transformers:
-      name:
-        name: greenmask_firstname
-        dynamic_parameters:
-          gender:
-            column: sex
+    validation_mode: relaxed
+    table_transformers:
+      - schema: public
+        table: test
+        column_transformers:
+          name:
+            name: greenmask_firstname
+            dynamic_parameters:
+              gender:
+                column: sex
diff --git a/cmd/config/test/test_transformer_rules.yaml b/cmd/config/test/test_transformer_rules.yaml
index 891cbe7..0580c41 100644
--- a/cmd/config/test/test_transformer_rules.yaml
+++ b/cmd/config/test/test_transformer_rules.yaml
@@ -1,9 +1,11 @@
 transformations:
-  - schema: public
-    table: test
-    column_transformers:
-      name:
-        name: greenmask_firstname
-        dynamic_parameters:
-          gender:
-            column: sex
+  validation_mode: relaxed
+  table_transformers:
+    - schema: public
+      table: test
+      column_transformers:
+        name:
+          name: greenmask_firstname
+          dynamic_parameters:
+            gender:
+              column: sex
diff --git a/pkg/stream/integration/helper_test.go b/pkg/stream/integration/helper_test.go
index 856247b..ff37379 100644
--- a/pkg/stream/integration/helper_test.go
+++ b/pkg/stream/integration/helper_test.go
@@ -283,9 +283,11 @@ func testKafkaCfg() kafkalib.ConnConfig {
 func testTransformationRules() []transformer.TableRules {
 	return []transformer.TableRules{
 		{
-			Schema: "public",
-			Table:  "pg2pg_integration_transformer_test",
+			Schema:         "public",
+			Table:          "pg2pg_integration_transformer_test",
+			ValidationMode: "strict",
 			ColumnRules: map[string]transformer.TransformerRules{
+				"id": {},
 				"name": {
 					Name: "neosync_firstname",
 					Parameters: map[string]any{
diff --git a/pkg/wal/processor/transformer/wal_transformer_test.go b/pkg/wal/processor/transformer/wal_transformer_test.go
index 208e2ea..c00b049 100644
--- a/pkg/wal/processor/transformer/wal_transformer_test.go
+++ b/pkg/wal/processor/transformer/wal_transformer_test.go
@@ -27,8 +27,8 @@ func TestTransformer_New(t *testing.T) {
 		name   string
 		config *Config
 
-		wantTransformer *Transformer
-		wantErr         error
+		wantTransformerMap map[string]ColumnTransformers
+		wantErr            error
 	}{
 		{
 			name: "ok",
@@ -55,16 +55,12 @@ func TestTransformer_New(t *testing.T) {
 				},
 			},
 
-			wantTransformer: &Transformer{
-				logger:    log.NewNoopLogger(),
-				processor: mockProcessor,
-				transformerMap: map[string]ColumnTransformers{
-					"\"public\".\"test1\"": {
-						"column_1": testTransformer,
-					},
-					"\"test\".\"test2\"": {
-						"column_2": testTransformer,
-					},
+			wantTransformerMap: map[string]ColumnTransformers{
+				"\"public\".\"test1\"": {
+					"column_1": testTransformer,
+				},
+				"\"test\".\"test2\"": {
+					"column_2": testTransformer,
 				},
 			},
 			wantErr: nil,
@@ -77,7 +73,9 @@ func TestTransformer_New(t *testing.T) {
 
 			transformer, err := New(context.Background(), tc.config, mockProcessor)
 			require.ErrorIs(t, err, tc.wantErr)
-			require.Equal(t, tc.wantTransformer, transformer)
+			require.Equal(t, tc.wantTransformerMap, transformer.transformerMap)
+			require.Equal(t, mockProcessor, transformer.processor)
+			require.Equal(t, log.NewNoopLogger(), transformer.logger)
 		})
 	}
 }
diff --git a/pkg/wal/processor/transformer/wal_transformer_validator_test.go b/pkg/wal/processor/transformer/wal_transformer_validator_test.go
index 7a0ddf6..a43f4d1 100644
--- a/pkg/wal/processor/transformer/wal_transformer_validator_test.go
+++ b/pkg/wal/processor/transformer/wal_transformer_validator_test.go
@@ -13,8 +13,6 @@ 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"
-	"github.com/xataio/pgstream/pkg/transformers/mocks"
 )
 
 func TestPostgresTransformerValidator(t *testing.T) {
@@ -34,38 +32,6 @@ func TestPostgresTransformerValidator(t *testing.T) {
 							Name:        "name",
 							DataTypeOID: pgtype.TextOID,
 						},
-						{
-							Name:        "age",
-							DataTypeOID: pgtype.Int2OID,
-						},
-						{
-							Name:        "birth_date",
-							DataTypeOID: pgtype.DateOID,
-						},
-						{
-							Name:        "postcode",
-							DataTypeOID: pgtype.Int4OID,
-						},
-						{
-							Name:        "customer_id",
-							DataTypeOID: pgtype.UUIDOID,
-						},
-						{
-							Name:        "total_purchases",
-							DataTypeOID: pgtype.Float8OID,
-						},
-						{
-							Name:        "total_discounts",
-							DataTypeOID: pgtype.Float4OID,
-						},
-						{
-							Name:        "is_active",
-							DataTypeOID: pgtype.BoolOID,
-						},
-						{
-							Name:        "created_at",
-							DataTypeOID: pgtype.TimestamptzOID,
-						},
 					}
 				},
 				CloseFn: func() {},
@@ -74,125 +40,86 @@ func TestPostgresTransformerValidator(t *testing.T) {
 		},
 	}
 
-	testPGValidator := PostgresTransformerValidator{
+	testPGValidator := PostgresTransformerParser{
 		conn: testQuerier,
 	}
 
 	tests := []struct {
-		name           string
-		wantErr        error
-		transformerMap map[string]ColumnTransformers
+		name             string
+		transformerRules []TableRules
+
+		wantErr             error
+		wantTransformersFor []string
 	}{
 		{
-			name: "ok - no error, few columns",
-			transformerMap: map[string]ColumnTransformers{
-				testSchemaTable: {
-					"id": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.Integer64DataType,
-							}
-						},
-					},
-					"name": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.StringDataType,
-							}
+			name: "ok - no error, relaxed mode",
+			transformerRules: []TableRules{
+				{
+					Schema:         "public",
+					Table:          "test",
+					ValidationMode: "relaxed",
+					ColumnRules: map[string]TransformerRules{
+						"id": {
+							Name: "noop",
+						},
+						"name": {
+							Name: "string",
 						},
 					},
 				},
 			},
 
-			wantErr: nil,
+			wantTransformersFor: []string{"name"},
+			wantErr:             nil,
 		},
 		{
-			name: "ok - no error, all columns",
-			transformerMap: map[string]ColumnTransformers{
-				testSchemaTable: {
-					"id": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.Integer64DataType,
-							}
-						},
-					},
-					"name": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.StringDataType,
-							}
-						},
-					},
-					"age": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.Integer16DataType,
-							}
-						},
-					},
-					"birth_date": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.DateDataType,
-							}
-						},
-					},
-					"postcode": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.Integer32DataType,
-							}
-						},
-					},
-					"customer_id": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.UInt8ArrayOf16DataType,
-							}
-						},
-					},
-					"total_purchases": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.Float64DataType,
-							}
-						},
-					},
-					"total_discounts": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.Float32DataType,
-							}
+			name: "ok - no error for missing column, relaxed mode",
+			transformerRules: []TableRules{
+				{
+					Schema:         "public",
+					Table:          "test",
+					ValidationMode: "relaxed",
+					ColumnRules: map[string]TransformerRules{
+						"name": {
+							Name: "string",
 						},
 					},
-					"is_active": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.BooleanDataType,
-							}
-						},
-					},
-					"created_at": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.DatetimeDataType,
-							}
+				},
+			},
+
+			wantTransformersFor: []string{"name"},
+			wantErr:             nil,
+		},
+		{
+			name: "error - missing column for strict validation",
+			transformerRules: []TableRules{
+				{
+					Schema:         "public",
+					Table:          "test",
+					ValidationMode: "strict",
+					ColumnRules: map[string]TransformerRules{
+						"name": {
+							Name: "string",
 						},
 					},
 				},
 			},
-			wantErr: nil,
+
+			wantErr: fmt.Errorf("column id of table %s has no transformer configured", testSchemaTable),
 		},
 		{
 			name: "error - invalid column type",
-			transformerMap: map[string]ColumnTransformers{
-				testSchemaTable: {
-					"id": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.BooleanDataType,
-							}
+			transformerRules: []TableRules{
+				{
+					Schema:         "public",
+					Table:          "test",
+					ValidationMode: "relaxed",
+					ColumnRules: map[string]TransformerRules{
+						"id": {
+							Name: "string",
+						},
+						"name": {
+							Name: "string",
 						},
 					},
 				},
@@ -201,13 +128,17 @@ func TestPostgresTransformerValidator(t *testing.T) {
 		},
 		{
 			name: "error - column not found in table",
-			transformerMap: map[string]ColumnTransformers{
-				testSchemaTable: {
-					"unknown_column": &mocks.Transformer{
-						CompatibleTypesFn: func() []transformers.SupportedDataType {
-							return []transformers.SupportedDataType{
-								transformers.StringDataType,
-							}
+			transformerRules: []TableRules{
+				{
+					Schema:         "public",
+					Table:          "test",
+					ValidationMode: "relaxed",
+					ColumnRules: map[string]TransformerRules{
+						"unknown_column": {
+							Name: "string",
+						},
+						"name": {
+							Name: "string",
 						},
 					},
 				},
@@ -219,9 +150,22 @@ func TestPostgresTransformerValidator(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			err := testPGValidator.Validate(context.Background(), tc.transformerMap)
-			if !errors.Is(err, tc.wantErr) {
-				require.Contains(t, err.Error(), tc.wantErr.Error())
+			transformerMap, err := testPGValidator.ParseAndValidate(tc.transformerRules)
+			if tc.wantErr != nil {
+				require.Error(t, err)
+				if !errors.Is(err, tc.wantErr) {
+					require.Contains(t, err.Error(), tc.wantErr.Error())
+				}
+				return
+			}
+			require.NoError(t, err)
+
+			columnTransformers, ok := transformerMap[testSchemaTable]
+			require.True(t, ok)
+
+			require.Equal(t, len(tc.wantTransformersFor), len(columnTransformers))
+			for _, col := range tc.wantTransformersFor {
+				require.Contains(t, columnTransformers, col)
 			}
 		})
 	}