Repository: xataio/pgstream
Pull request #318 (2025-04-30 13:11:49)

Add literal string transformer
Adding `LiteralStringTransformer` which transforms anything into given string. This is expected to be used as an escape hatch, mostly for non supported types.

fixes: #311 

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

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 86.76% (**-0.32%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/builder | 21.05% (**-1.17%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 73.45% (**-0.66%**) | :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 | 21.05% (**-1.17%**) | 19 (+1) | 4 | 15 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/literal_string_transformer.go | 80.00% (**+80.00%**) | 10 (+10) | 8 (+8) | 2 (+2) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 96.30% (ø) | 54 | 52 | 2 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 64.81% (**-1.22%**) | 54 (+1) | 35 | 19 (+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/transformers/literal_string_transformer_test.go

</details>

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

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 86.76% (**-0.32%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/builder | 21.05% (**-1.17%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/wal/processor/transformer | 73.04% (**-1.06%**) | :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 | 21.05% (**-1.17%**) | 19 (+1) | 4 | 15 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/literal_string_transformer.go | 80.00% (**+80.00%**) | 10 (+10) | 8 (+8) | 2 (+2) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 96.30% (ø) | 54 | 52 | 2 |  |
| github.com/xataio/pgstream/pkg/wal/processor/transformer/wal_transformer_validator.go | 64.29% (**-1.75%**) | 56 (+3) | 36 (+1) | 20 (+2) | :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/transformers/literal_string_transformer_test.go

</details>


Patch:
diff --git a/pkg/transformers/builder/transformer_builder.go b/pkg/transformers/builder/transformer_builder.go
index 478a85e..f69581a 100644
--- a/pkg/transformers/builder/transformer_builder.go
+++ b/pkg/transformers/builder/transformer_builder.go
@@ -34,6 +34,8 @@ func New(cfg *transformers.Config) (transformers.Transformer, error) {
 		return greenmask.NewUTCTimestampTransformer(cfg.Parameters)
 	case transformers.String:
 		return transformers.NewStringTransformer(cfg.Parameters)
+	case transformers.LiteralString:
+		return transformers.NewLiteralStringTransformer(cfg.Parameters)
 	case transformers.NeosyncString:
 		return neosync.NewStringTransformer(cfg.Parameters)
 	case transformers.NeosyncFirstName:
diff --git a/pkg/transformers/literal_string_transformer.go b/pkg/transformers/literal_string_transformer.go
new file mode 100644
index 0000000..f944fa0
--- /dev/null
+++ b/pkg/transformers/literal_string_transformer.go
@@ -0,0 +1,45 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"errors"
+	"fmt"
+)
+
+type LiteralStringTransformer struct {
+	literal string
+}
+
+var (
+	literalStringTransformerParams = []string{"literal"}
+	errLiteralStringNotFound       = errors.New("literal_string_transformer: literal parameter not found")
+)
+
+func NewLiteralStringTransformer(params Parameters) (*LiteralStringTransformer, error) {
+	if err := ValidateParameters(params, literalStringTransformerParams); err != nil {
+		return nil, err
+	}
+
+	literal, found, err := FindParameter[string](params, "literal")
+	if err != nil {
+		return nil, fmt.Errorf("literal_string_transformer: literal must be a string: %w", err)
+	}
+	if !found {
+		return nil, errLiteralStringNotFound
+	}
+
+	return &LiteralStringTransformer{
+		literal: literal,
+	}, nil
+}
+
+func (lst *LiteralStringTransformer) Transform(value Value) (any, error) {
+	return lst.literal, nil
+}
+
+func (lst *LiteralStringTransformer) CompatibleTypes() []SupportedDataType {
+	return []SupportedDataType{
+		AllDataTypes,
+	}
+}
diff --git a/pkg/transformers/transformer.go b/pkg/transformers/transformer.go
index 83073dc..0aa68b5 100644
--- a/pkg/transformers/transformer.go
+++ b/pkg/transformers/transformer.go
@@ -31,6 +31,7 @@ type TransformerType string
 
 const (
 	String                 TransformerType = "string"
+	LiteralString          TransformerType = "literal_string"
 	PhoneNumber            TransformerType = "phone_number"
 	NeosyncString          TransformerType = "neosync_string"
 	GreenmaskString        TransformerType = "greenmask_string"
@@ -68,6 +69,8 @@ const (
 	UInt8ArrayOf16DataType SupportedDataType = "uint8_array_of_16"
 	DateDataType           SupportedDataType = "date"
 	DatetimeDataType       SupportedDataType = "datetime"
+	JSONDataType           SupportedDataType = "json"
+	AllDataTypes           SupportedDataType = "all"
 )
 
 const (
diff --git a/pkg/wal/processor/transformer/wal_transformer_validator.go b/pkg/wal/processor/transformer/wal_transformer_validator.go
index 9c493eb..d0b4861 100644
--- a/pkg/wal/processor/transformer/wal_transformer_validator.go
+++ b/pkg/wal/processor/transformer/wal_transformer_validator.go
@@ -102,6 +102,9 @@ func (v *PostgresTransformerParser) getFieldDescriptions(ctx context.Context, sc
 }
 
 func pgTypeCompatibleWithTransformerType(compatibleTypes []transformers.SupportedDataType, pgType uint32) bool {
+	if slices.Contains(compatibleTypes, transformers.AllDataTypes) {
+		return true
+	}
 	switch pgType {
 	case pgtype.TextOID, pgtype.VarcharOID, pgtype.BPCharOID:
 		return slices.Contains(compatibleTypes, transformers.StringDataType)
@@ -125,6 +128,8 @@ func pgTypeCompatibleWithTransformerType(compatibleTypes []transformers.Supporte
 		return slices.Contains(compatibleTypes, transformers.DateDataType)
 	case pgtype.TimestampOID, pgtype.TimestamptzOID:
 		return slices.Contains(compatibleTypes, transformers.DatetimeDataType)
+	case pgtype.JSONBOID, pgtype.JSONOID:
+		return slices.Contains(compatibleTypes, transformers.JSONDataType)
 	default:
 		return false
 	}


Test patch:
diff --git a/pkg/transformers/literal_string_transformer_test.go b/pkg/transformers/literal_string_transformer_test.go
new file mode 100644
index 0000000..16aefa0
--- /dev/null
+++ b/pkg/transformers/literal_string_transformer_test.go
@@ -0,0 +1,87 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLiteralStringTransformer(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name    string
+		params  Parameters
+		wantErr error
+	}{
+		{
+			name: "ok - valid",
+			params: Parameters{
+				"literal": "test",
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - invalid literal",
+			params: Parameters{
+				"literal": 123,
+			},
+			wantErr: ErrInvalidParameters,
+		},
+		{
+			name:    "error - empty literal",
+			params:  Parameters{},
+			wantErr: errLiteralStringNotFound,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			lst, err := NewLiteralStringTransformer(tc.params)
+			require.ErrorIs(t, err, tc.wantErr)
+			if tc.wantErr != nil {
+				return
+			}
+			require.NoError(t, err)
+			require.NotNil(t, lst)
+		})
+	}
+}
+
+func TestLiteralStringTransformer_Transform(t *testing.T) {
+	t.Parallel()
+	wantOutput := "{'output': 'testoutput'"
+	lst, err := NewLiteralStringTransformer(Parameters{"literal": wantOutput})
+	require.NoError(t, err)
+	tests := []struct {
+		name    string
+		params  Parameters
+		input   any
+		want    any
+		wantErr error
+	}{
+		{
+			name:    "ok - string",
+			input:   "testinput",
+			wantErr: nil,
+		},
+		{
+			name:    "ok - JSON",
+			input:   "{'json': 'jsoninput'}",
+			wantErr: nil,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := lst.Transform(Value{TransformValue: tc.input})
+			require.ErrorIs(t, err, tc.wantErr)
+			if tc.wantErr != nil {
+				return
+			}
+			require.Equal(t, wantOutput, got)
+		})
+	}
+}