Add debeziumUnwrap transformers

Author: ilyario

api/v1/dataflow_types.go

@@ -808,7 +808,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
+	// Type of transformation: timestamp, flatten, filter, mask, router, select, remove, snakeCase, camelCase, debeziumUnwrap
 	Type string `json:"type"`
 
 	// Config holds transformation configuration. Structure depends on type.
@@ -862,6 +862,11 @@ func (t *TransformationSpec) GetCamelCaseConfig() (*CamelCaseTransformation, err
 	return getTypedConfig[CamelCaseTransformation](t.Config)
 }
 
+// GetDebeziumUnwrapConfig returns DebeziumUnwrap transformation config.
+func (t *TransformationSpec) GetDebeziumUnwrapConfig() (*DebeziumUnwrapTransformation, error) {
+	return getTypedConfig[DebeziumUnwrapTransformation](t.Config)
+}
+
 // TimestampTransformation adds a timestamp field
 type TimestampTransformation struct {
 	// FieldName is the name of the timestamp field (default: created_at)
@@ -940,6 +945,21 @@ type CamelCaseTransformation struct {
 	Deep bool `json:"deep,omitempty"`
 }
 
+// DebeziumUnwrapTransformation unwraps Debezium envelope messages into row payloads.
+type DebeziumUnwrapTransformation struct {
+	// InferDeleteFromTombstone converts Kafka tombstone records into operation=delete messages using metadata.key JSON.
+	// +optional
+	InferDeleteFromTombstone bool `json:"inferDeleteFromTombstone,omitempty"`
+
+	// IncludeSourceInMetadata copies payload.source fields into metadata with source_ prefix.
+	// +optional
+	IncludeSourceInMetadata bool `json:"includeSourceInMetadata,omitempty"`
+
+	// SnapshotOperation defines operation for Debezium snapshot records (op="r"): insert (default) or update.
+	// +optional
+	SnapshotOperation string `json:"snapshotOperation,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

@@ -540,6 +540,17 @@ func validateTransformations(transformations []TransformationSpec, f *field.Path
 			} else {
 				all = append(all, field.Required(idx.Child("config"), "camelCase transformation configuration is required"))
 			}
+		case transformtypes.DebeziumUnwrap:
+			if hasConfig {
+				var cfg DebeziumUnwrapTransformation
+				if err := json.Unmarshal(t.Config.Raw, &cfg); err != nil {
+					all = append(all, field.Invalid(idx.Child("config"), string(t.Config.Raw), "invalid debeziumUnwrap config: "+err.Error()))
+				} else if cfg.SnapshotOperation != "" && cfg.SnapshotOperation != "insert" && cfg.SnapshotOperation != "update" {
+					all = append(all, field.NotSupported(idx.Child("config", "snapshotOperation"), cfg.SnapshotOperation, []string{"insert", "update"}))
+				}
+			} else {
+				all = append(all, field.Required(idx.Child("config"), "debeziumUnwrap transformation configuration is required"))
+			}
 		}
 	}
 	return all

api/v1/dataflow_validation_transformations_test.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+func mustRawConfigForValidation(v interface{}) *runtime.RawExtension {
+	b, _ := json.Marshal(v)
+	return &runtime.RawExtension{Raw: b}
+}
+
+func TestValidateTransformations_DebeziumUnwrap(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 config", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{
+			Type: "debeziumUnwrap",
+			Config: mustRawConfigForValidation(DebeziumUnwrapTransformation{
+				InferDeleteFromTombstone: true,
+				IncludeSourceInMetadata:  true,
+				SnapshotOperation:        "update",
+			}),
+		}}
+		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: "debeziumUnwrap"}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) == 0 {
+			t.Fatal("expected validation error for missing config")
+		}
+		if !strings.Contains(errs.ToAggregate().Error(), "debeziumUnwrap transformation configuration is required") {
+			t.Fatalf("unexpected error: %v", errs.ToAggregate())
+		}
+	})
+
+	t.Run("invalid snapshot operation", func(t *testing.T) {
+		spec := baseSpec
+		spec.Transformations = []TransformationSpec{{
+			Type: "debeziumUnwrap",
+			Config: mustRawConfigForValidation(DebeziumUnwrapTransformation{
+				SnapshotOperation: "noop",
+			}),
+		}}
+		errs := ValidateDataFlowSpec(&spec)
+		if len(errs) == 0 {
+			t.Fatal("expected validation error for snapshotOperation")
+		}
+		if !strings.Contains(errs.ToAggregate().Error(), "Unsupported value") {
+			t.Fatalf("unexpected error: %v", errs.ToAggregate())
+		}
+	})
+}

api/v1/dataflow_webhook_test.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/dataflow-operator/dataflow/pkg/transformtypes"
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -98,3 +99,29 @@ func TestDataFlow_ValidateDelete(t *testing.T) {
 		t.Errorf("ValidateDelete: unexpected warnings: %v", warnings)
 	}
 }
+
+func TestDataFlow_ValidateCreate_DebeziumUnwrapInvalidSnapshotOperation(t *testing.T) {
+	df := &DataFlow{}
+	df.Spec = DataFlowSpec{
+		Source: SourceSpec{
+			Type:   "kafka",
+			Config: mustConfig(KafkaSourceSpec{Brokers: []string{"b"}, Topic: "t"}),
+		},
+		Sink: SinkSpec{
+			Type:   "kafka",
+			Config: mustConfig(KafkaSinkSpec{Brokers: []string{"b"}, Topic: "out"}),
+		},
+		Transformations: []TransformationSpec{
+			{
+				Type: transformtypes.DebeziumUnwrap,
+				Config: mustConfig(DebeziumUnwrapTransformation{
+					SnapshotOperation: "delete",
+				}),
+			},
+		},
+	}
+	_, err := df.ValidateCreate(context.Background(), df)
+	if err == nil {
+		t.Fatal("ValidateCreate: expected validation error for invalid snapshotOperation")
+	}
+}

api/v1/zz_generated.deepcopy.go

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

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

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

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

@@ -0,0 +1,26 @@
+apiVersion: dataflow.dataflow.io/v1
+kind: DataFlow
+metadata:
+  name: kafka-debezium-to-postgres
+spec:
+  source:
+    type: kafka
+    config:
+      brokers:
+        - kafka:9092
+      topic: dbserver1.inventory.customers
+      consumerGroup: dataflow-debezium-group
+  transformations:
+    - type: debeziumUnwrap
+      config:
+        inferDeleteFromTombstone: true
+        includeSourceInMetadata: true
+        snapshotOperation: insert
+  sink:
+    type: postgresql
+    config:
+      connectionString: "postgres://dataflow:dataflow@postgres:5432/dataflow?sslmode=disable"
+      table: customers
+      upsertMode: true
+      conflictKey: id
+      softDeleteColumn: deleted_at