Repository: xataio/pgstream
Pull request #339 (2025-05-13 19:43:07)

Add template transformer
This PR adds template transformer which gets a template during creation, then executes it when transforming data. This transformer also supports dynamic values using other column values.

Other than the functionalities supported by default by templates, this PR introduces two methods, namely `GetValue` and `GetDynamicValue`, allowing users to build up more complex templates using all the column values in a row.

Considering adding support for wide range of helper functions supported [here](https://github.com/eminano/greenmask/blob/main/pkg/toolkit/template_functions.go#L71-L199) in greenmask, in a followup PR.
Skipping updating the docs for now, will do it once we're done with templates.

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

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 85.83% (**+0.63%**) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/builder | 29.03% (**-0.97%**) | :thumbsdown: |

---

<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 | 29.03% (**-0.97%**) | 31 (+1) | 9 | 22 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/template_transformer.go | 82.35% (**+82.35%**) | 17 (+17) | 14 (+14) | 3 (+3) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 98.36% (**+2.06%**) | 61 (+7) | 60 (+8) | 1 (-1) | :thumbsup: |

_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/template_transformer_test.go

</details>


Patch:
diff --git a/pkg/transformers/builder/transformer_builder.go b/pkg/transformers/builder/transformer_builder.go
index c61279f..d318138 100644
--- a/pkg/transformers/builder/transformer_builder.go
+++ b/pkg/transformers/builder/transformer_builder.go
@@ -78,6 +78,8 @@ func (b *TransformerBuilder) New(cfg *transformers.Config) (t transformers.Trans
 		return transformers.NewPhoneNumberTransformer(cfg.Parameters)
 	case transformers.Masking:
 		return transformers.NewMaskingTransformer(cfg.Parameters)
+	case transformers.Template:
+		return transformers.NewTemplateTransformer(cfg.Parameters)
 	default:
 		return nil, fmt.Errorf("%w: unexpected transformer name '%s'", transformers.ErrUnsupportedTransformer, cfg.Name)
 	}
diff --git a/pkg/transformers/template_transformer.go b/pkg/transformers/template_transformer.go
new file mode 100644
index 0000000..84d2bf4
--- /dev/null
+++ b/pkg/transformers/template_transformer.go
@@ -0,0 +1,57 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"text/template"
+)
+
+type TemplateTransformer struct {
+	template *template.Template
+}
+
+var (
+	templateTransformerParams = []string{"template"}
+	errTemplateMustBeProvided = errors.New("template_transformer: template parameter must be provided")
+)
+
+func NewTemplateTransformer(params Parameters) (*TemplateTransformer, error) {
+	if err := ValidateParameters(params, templateTransformerParams); err != nil {
+		return nil, err
+	}
+	templateStr, found, err := FindParameter[string](params, "template")
+	if err != nil {
+		return nil, fmt.Errorf("template_transformer: template must be a string: %w", err)
+	}
+	if !found {
+		return nil, errTemplateMustBeProvided
+	}
+
+	tmpl, err := template.New("").Parse(templateStr)
+	if err != nil {
+		return nil, fmt.Errorf("template_transformer: error parsing template: %w", err)
+	}
+	return &TemplateTransformer{template: tmpl}, nil
+}
+
+func (t *TemplateTransformer) Transform(_ context.Context, value Value) (any, error) {
+	var buf strings.Builder
+	if err := t.template.Execute(&buf, &value); err != nil {
+		return nil, fmt.Errorf("template_transformer: error executing template: %w", err)
+	}
+	return buf.String(), nil
+}
+
+func (t *TemplateTransformer) CompatibleTypes() []SupportedDataType {
+	return []SupportedDataType{
+		AllDataTypes,
+	}
+}
+
+func (t *TemplateTransformer) Type() TransformerType {
+	return Template
+}
diff --git a/pkg/transformers/transformer.go b/pkg/transformers/transformer.go
index 7ea9485..77c0b43 100644
--- a/pkg/transformers/transformer.go
+++ b/pkg/transformers/transformer.go
@@ -51,6 +51,7 @@ const (
 	GreenmaskDate          TransformerType = "greenmask_date"
 	GreenmaskUTCTimestamp  TransformerType = "greenmask_utc_timestamp"
 	Masking                TransformerType = "masking"
+	Template               TransformerType = "template"
 )
 
 type SupportedDataType string
@@ -204,3 +205,18 @@ func ValidateParameters(provided map[string]any, expected []string) error {
 
 	return nil
 }
+
+func (v *Value) GetValue() any {
+	return v.TransformValue
+}
+
+func (v *Value) GetDynamicValue(name string) (any, error) {
+	if v.DynamicValues == nil {
+		return nil, fmt.Errorf("dynamic values are nil")
+	}
+	dynValue, found := v.DynamicValues[name]
+	if !found {
+		return nil, fmt.Errorf("dynamic value '%s' not found", name)
+	}
+	return dynValue, nil
+}


Test patch:
diff --git a/pkg/transformers/template_transformer_test.go b/pkg/transformers/template_transformer_test.go
new file mode 100644
index 0000000..3e9b96c
--- /dev/null
+++ b/pkg/transformers/template_transformer_test.go
@@ -0,0 +1,223 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewTemplateTransformer(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		params  Parameters
+		wantErr error
+	}{
+		{
+			name: "ok - template",
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} first {{- else -}} second {{- end -}}",
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - invalid parameter",
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} first {{- else -}} second {{- end -}}",
+				"invalid":  "invalid",
+			},
+			wantErr: ErrUnknownParameter,
+		},
+		{
+			name:    "error - template not provided",
+			params:  Parameters{},
+			wantErr: errTemplateMustBeProvided,
+		},
+		{
+			name: "error - template cannot be parsed",
+			params: Parameters{
+				"template": "{{- if eq syntaxerror",
+			},
+			wantErr: errors.New("template_transformer: error parsing template"),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tt, err := NewTemplateTransformer(tc.params)
+			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)
+			require.NotNil(t, tt)
+		})
+	}
+}
+
+func TestTemplateTransformer_Transform(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name   string
+		value  any
+		params Parameters
+
+		wantOutput string
+		wantErr    error
+	}{
+		{
+			name:  "ok - basic template",
+			value: "hello",
+			params: Parameters{
+				"template": "hello world",
+			},
+			wantOutput: "hello world",
+			wantErr:    nil,
+		},
+		{
+			name:  "ok - GetValue with if statement",
+			value: "hello",
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} first {{- else -}} second {{- end -}}",
+			},
+
+			wantOutput: "first",
+			wantErr:    nil,
+		},
+		{
+			name:  "ok - GetValue with if statement - else",
+			value: "world",
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} first {{- else -}} second {{- end -}}",
+			},
+
+			wantOutput: "second",
+			wantErr:    nil,
+		},
+		{
+			name:  "incompatible types for comparison",
+			value: 1,
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} first {{- else -}} second {{- end -}}",
+			},
+			wantOutput: "",
+			wantErr:    errors.New("incompatible types for comparison"),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tt, err := NewTemplateTransformer(tc.params)
+			require.NoError(t, err)
+			got, err := tt.Transform(context.Background(), Value{TransformValue: tc.value})
+
+			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)
+
+			require.Equal(t, tc.wantOutput, got)
+		})
+	}
+}
+
+func TestTemplateTransformer_Transform_WithDynamicValues(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name          string
+		value         any
+		dynamicValues map[string]any
+		params        Parameters
+		wantOutput    string
+		wantErr       error
+	}{
+		{
+			name:  "ok - GetValue with dynamic values",
+			value: "hello",
+			dynamicValues: map[string]any{
+				"value1": "first",
+				"value2": "second",
+			},
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} {{.GetDynamicValue \"value1\" }} {{- else -}} {{.GetDynamicValue \"value2\" }} {{- end -}}",
+			},
+			wantOutput: "first",
+			wantErr:    nil,
+		},
+		{
+			name:  "ok - GetValue with dynamic values - else",
+			value: "world",
+			dynamicValues: map[string]any{
+				"value1": "first",
+				"value2": "second",
+			},
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} {{.GetDynamicValue \"value1\" }} {{- else -}} {{.GetDynamicValue \"value2\" }} {{- end -}}",
+			},
+			wantOutput: "second",
+			wantErr:    nil,
+		},
+		{
+			name:          "error - no dynamic values",
+			value:         "hello",
+			dynamicValues: nil,
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} {{.GetDynamicValue \"value1\" }} {{- else -}} {{.GetDynamicValue \"value2\" }} {{- end -}}",
+			},
+			wantOutput: "",
+			wantErr:    errors.New("dynamic values are nil"),
+		},
+		{
+			name:  "error - dynamic value not found",
+			value: "world",
+			dynamicValues: map[string]any{
+				"value1": "first",
+			},
+			params: Parameters{
+				"template": "{{- if eq .GetValue \"hello\" -}} {{.GetDynamicValue \"value1\" }} {{- else -}} {{.GetDynamicValue \"value2\" }} {{- end -}}",
+			},
+			wantOutput: "",
+			wantErr:    errors.New("dynamic value 'value2' not found"),
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tt, err := NewTemplateTransformer(tc.params)
+			require.NoError(t, err)
+			got, err := tt.Transform(context.Background(), Value{
+				TransformValue: tc.value,
+				DynamicValues:  tc.dynamicValues,
+			})
+
+			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)
+
+			require.Equal(t, tc.wantOutput, got)
+		})
+	}
+}