feat: add ReplaceField transformation support

- Introduced ReplaceFieldTransformation to rename fields and filter by include/exclude criteria.
- Updated TransformationSpec to include ReplaceField as a valid transformation type.
- Implemented validation logic for ReplaceField transformation configuration, ensuring required fields and mutually exclusive options.
- Added unit tests for ReplaceField transformation creation and validation scenarios.
- Enhanced existing tests to cover new transformation functionality.
- Updated CRDs to reflect the new ReplaceField transformation type in documentation.

Author: ilyario

api/v1/dataflow_types.go

@@ -1342,7 +1342,7 @@ type SASLConfig struct {
 // TransformationSpec defines a transformation to apply (type + config).
 // +kubebuilder:pruning:PreserveUnknownFields
 type TransformationSpec struct {
-	// Type of transformation: timestamp, flatten, filter, mask, router, select, remove, snakeCase, camelCase, debeziumUnwrap
+	// Type of transformation: timestamp, flatten, filter, mask, router, select, remove, snakeCase, camelCase, debeziumUnwrap, replaceField
 	Type string `json:"type"`
 
 	// Config holds transformation configuration. Structure depends on type.
@@ -1401,6 +1401,11 @@ func (t *TransformationSpec) GetDebeziumUnwrapConfig() (*DebeziumUnwrapTransform
 	return getTypedConfig[DebeziumUnwrapTransformation](t.Config)
 }
 
+// GetReplaceFieldConfig returns ReplaceField transformation config.
+func (t *TransformationSpec) GetReplaceFieldConfig() (*ReplaceFieldTransformation, error) {
+	return getTypedConfig[ReplaceFieldTransformation](t.Config)
+}
+
 // TimestampTransformation adds a timestamp field
 type TimestampTransformation struct {
 	// FieldName is the name of the timestamp field (default: created_at)
@@ -1494,6 +1499,23 @@ type DebeziumUnwrapTransformation struct {
 	SnapshotOperation string `json:"snapshotOperation,omitempty"`
 }
 
+// ReplaceFieldTransformation renames fields and optionally filters by include/exclude.
+// Unlike select, include preserves nested structure (no key flattening).
+// Include and Exclude are mutually exclusive. At least one of Renames, Include, or Exclude is required.
+type ReplaceFieldTransformation struct {
+	// Renames is a list of oldPath:newPath mappings (JSONPath without requiring $.).
+	// +optional
+	Renames []string `json:"renames,omitempty"`
+
+	// Include keeps only the listed field paths, preserving nesting.
+	// +optional
+	Include []string `json:"include,omitempty"`
+
+	// Exclude removes the listed field paths.
+	// +optional
+	Exclude []string `json:"exclude,omitempty"`
+}
+
 // DataFlowStatus defines the observed state of DataFlow
 type DataFlowStatus struct {
 	// Phase represents the current phase of the data flow

api/v1/dataflow_validation.go

@@ -1032,11 +1032,41 @@ func validateTransformations(transformations []TransformationSpec, f *field.Path
 			} else {
 				all = append(all, field.Required(idx.Child("config"), "debeziumUnwrap transformation configuration is required"))
 			}
+		case transformtypes.ReplaceField:
+			if hasConfig {
+				var cfg ReplaceFieldTransformation
+				if err := json.Unmarshal(t.Config.Raw, &cfg); err != nil {
+					all = append(all, field.Invalid(idx.Child("config"), string(t.Config.Raw), "invalid replaceField config: "+err.Error()))
+				} else {
+					if len(cfg.Include) > 0 && len(cfg.Exclude) > 0 {
+						all = append(all, field.Invalid(idx.Child("config"), string(t.Config.Raw), "include and exclude are mutually exclusive"))
+					}
+					if len(cfg.Renames) == 0 && len(cfg.Include) == 0 && len(cfg.Exclude) == 0 {
+						all = append(all, field.Required(idx.Child("config"), "at least one of renames, include, or exclude is required"))
+					}
+					for j, rename := range cfg.Renames {
+						if !isValidReplaceFieldRename(rename) {
+							all = append(all, field.Invalid(idx.Child("config", "renames").Index(j), rename, "rename must be in oldPath:newPath format"))
+						}
+					}
+				}
+			} else {
+				all = append(all, field.Required(idx.Child("config"), "replaceField transformation configuration is required"))
+			}
 		}
 	}
 	return all
 }
 
+// isValidReplaceFieldRename reports whether rename is in oldPath:newPath form with non-empty sides.
+func isValidReplaceFieldRename(rename string) bool {
+	oldPath, newPath, ok := strings.Cut(rename, ":")
+	if !ok {
+		return false
+	}
+	return strings.TrimSpace(oldPath) != "" && strings.TrimSpace(newPath) != ""
+}
+
 func validateResources(r *corev1.ResourceRequirements, f *field.Path) field.ErrorList {
 	var all field.ErrorList
 	if r == nil {

api/v1/dataflow_validation_transformations_test.go

@@ -86,3 +86,106 @@ func TestValidateTransformations_DebeziumUnwrap(t *testing.T) {
 		}
 	})
 }
+
+func TestValidateTransformations_ReplaceField(t *testing.T) {
+	baseSpec := DataFlowSpec{
+		Source: SourceSpec{
+			Type:   "kafka",
+			Config: mustRawConfigForValidation(KafkaSourceSpec{Brokers: []string{"broker:9092"}, Topic: "src"}),
+		},
+		Sink: SinkSpec{
+			Type:   "kafka",
+			Config: mustRawConfigForValidation(KafkaSinkSpec{Brokers: []string{"broker:9092"}, Topic: "dst"}),
+		},
+	}
+
+	t.Run("valid renames", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{
+			Type: "replaceField",
+			Config: mustRawConfigForValidation(ReplaceFieldTransformation{
+				Renames: []string{"old:new", "a.b:c"},
+			}),
+		}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) != 0 {
+			t.Fatalf("expected no validation errors, got %v", errs)
+		}
+	})
+
+	t.Run("valid include", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{
+			Type: "replaceField",
+			Config: mustRawConfigForValidation(ReplaceFieldTransformation{
+				Include: []string{"id", "user.name"},
+			}),
+		}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) != 0 {
+			t.Fatalf("expected no validation errors, got %v", errs)
+		}
+	})
+
+	t.Run("missing config", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{Type: "replaceField"}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) == 0 {
+			t.Fatal("expected validation error for missing config")
+		}
+		if !strings.Contains(errs.ToAggregate().Error(), "replaceField transformation configuration is required") {
+			t.Fatalf("unexpected error: %v", errs.ToAggregate())
+		}
+	})
+
+	t.Run("empty config", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{
+			Type:   "replaceField",
+			Config: mustRawConfigForValidation(ReplaceFieldTransformation{}),
+		}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) == 0 {
+			t.Fatal("expected validation error for empty replaceField config")
+		}
+		if !strings.Contains(errs.ToAggregate().Error(), "at least one of renames, include, or exclude is required") {
+			t.Fatalf("unexpected error: %v", errs.ToAggregate())
+		}
+	})
+
+	t.Run("include and exclude mutually exclusive", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{
+			Type: "replaceField",
+			Config: mustRawConfigForValidation(ReplaceFieldTransformation{
+				Include: []string{"a"},
+				Exclude: []string{"b"},
+			}),
+		}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) == 0 {
+			t.Fatal("expected validation error for include+exclude")
+		}
+		if !strings.Contains(errs.ToAggregate().Error(), "mutually exclusive") {
+			t.Fatalf("unexpected error: %v", errs.ToAggregate())
+		}
+	})
+
+	t.Run("invalid rename format", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{
+			Type: "replaceField",
+			Config: mustRawConfigForValidation(ReplaceFieldTransformation{
+				Renames: []string{"bad-format"},
+			}),
+		}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) == 0 {
+			t.Fatal("expected validation error for invalid rename")
+		}
+		if !strings.Contains(errs.ToAggregate().Error(), "oldPath:newPath") {
+			t.Fatalf("unexpected error: %v", errs.ToAggregate())
+		}
+	})
+}

api/v1/zz_generated.deepcopy.go

@@ -1266,7 +1266,8 @@ spec:
                       x-kubernetes-preserve-unknown-fields: true
                     type:
                       description: 'Type of transformation: timestamp, flatten, filter,
-                        mask, router, select, remove, snakeCase, camelCase, debeziumUnwrap'
+                        mask, router, select, remove, snakeCase, camelCase, debeziumUnwrap,
+                        replaceField'
                       type: string
                   required:
                   - type

config/crd/bases/dataflow.dataflow.io_dataflowcrons.yaml

@@ -1251,7 +1251,8 @@ spec:
                       x-kubernetes-preserve-unknown-fields: true
                     type:
                       description: 'Type of transformation: timestamp, flatten, filter,
-                        mask, router, select, remove, snakeCase, camelCase, debeziumUnwrap'
+                        mask, router, select, remove, snakeCase, camelCase, debeziumUnwrap,
+                        replaceField'
                       type: string
                   required:
                   - type

config/crd/bases/dataflow.dataflow.io_dataflows.yaml

@@ -45,6 +45,7 @@ var transformerRegistry = map[string]transformerEntry{
 			return NewDebeziumUnwrapTransformer(cfg)
 		}),
 	},
+	transformtypes.ReplaceField: {create: createTransformer[v1.ReplaceFieldTransformation](transformtypes.ReplaceField, func(cfg *v1.ReplaceFieldTransformation) Transformer { return NewReplaceFieldTransformer(cfg) })},
 }
 
 // createTransformer returns a factory function that unmarshals raw config into T and calls newFn.

internal/transformers/factory.go

@@ -259,6 +259,29 @@ func TestCreateTransformer_DebeziumUnwrap(t *testing.T) {
 	})
 }
 
+func TestCreateTransformer_ReplaceField(t *testing.T) {
+	runCreateTransformerTests(t, []transformerTestCase{
+		{
+			name: "valid replaceField transformation",
+			transformation: &v1.TransformationSpec{
+				Type: transformtypes.ReplaceField,
+				Config: mustConfig(v1.ReplaceFieldTransformation{
+					Renames: []string{"oldName:newName"},
+					Include: []string{"id", "name"},
+				}),
+			},
+		},
+		{
+			name: "replaceField without config",
+			transformation: &v1.TransformationSpec{
+				Type: transformtypes.ReplaceField,
+			},
+			wantErr:     true,
+			errContains: "replaceField transformation configuration is required",
+		},
+	})
+}
+
 func TestCreateTransformer_UnsupportedType(t *testing.T) {
 	transformation := &v1.TransformationSpec{
 		Type: "unsupported",