refatoring

Author: ilyario

api/v1/Untitled

@@ -0,0 +1 @@
+protocol

api/v1/dataflow_checkpoint.go

@@ -0,0 +1,48 @@
+/*
+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 (
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const defaultCheckpointSaveInterval = 30 * time.Second
+
+// CheckpointPersistenceEnabled reports whether checkpoint persistence is enabled (default true).
+func CheckpointPersistenceEnabled(spec *DataFlowSpec) bool {
+	return spec == nil || spec.CheckpointPersistence == nil || *spec.CheckpointPersistence
+}
+
+// CheckpointSyncOnAckEnabled reports whether checkpoint should flush after each sink batch ack.
+func CheckpointSyncOnAckEnabled(spec *DataFlowSpec) bool {
+	return spec != nil && spec.CheckpointSyncOnAck != nil && *spec.CheckpointSyncOnAck
+}
+
+// CheckpointSaveIntervalDuration returns the debounce/sync coalesce interval (default 30s).
+func CheckpointSaveIntervalDuration(spec *DataFlowSpec) time.Duration {
+	if spec != nil && spec.CheckpointSaveInterval != nil && spec.CheckpointSaveInterval.Duration > 0 {
+		return spec.CheckpointSaveInterval.Duration
+	}
+	return defaultCheckpointSaveInterval
+}
+
+// CheckpointSaveIntervalOrDefault returns the spec interval pointer or a default metav1.Duration.
+func CheckpointSaveIntervalOrDefault(spec *DataFlowSpec) metav1.Duration {
+	return metav1.Duration{Duration: CheckpointSaveIntervalDuration(spec)}
+}

api/v1/dataflow_checkpoint_test.go

@@ -0,0 +1,54 @@
+/*
+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 (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestCheckpointPersistenceEnabled(t *testing.T) {
+	t.Parallel()
+
+	trueVal := true
+	falseVal := false
+	assert.True(t, CheckpointPersistenceEnabled(nil))
+	assert.True(t, CheckpointPersistenceEnabled(&DataFlowSpec{}))
+	assert.True(t, CheckpointPersistenceEnabled(&DataFlowSpec{CheckpointPersistence: &trueVal}))
+	assert.False(t, CheckpointPersistenceEnabled(&DataFlowSpec{CheckpointPersistence: &falseVal}))
+}
+
+func TestCheckpointSyncOnAckEnabled(t *testing.T) {
+	t.Parallel()
+
+	trueVal := true
+	assert.False(t, CheckpointSyncOnAckEnabled(nil))
+	assert.False(t, CheckpointSyncOnAckEnabled(&DataFlowSpec{}))
+	assert.True(t, CheckpointSyncOnAckEnabled(&DataFlowSpec{CheckpointSyncOnAck: &trueVal}))
+}
+
+func TestCheckpointSaveIntervalDuration(t *testing.T) {
+	t.Parallel()
+
+	assert.Equal(t, 30*time.Second, CheckpointSaveIntervalDuration(nil))
+	assert.Equal(t, 5*time.Second, CheckpointSaveIntervalDuration(&DataFlowSpec{
+		CheckpointSaveInterval: &metav1.Duration{Duration: 5 * time.Second},
+	}))
+}

api/v1/dataflow_types.go

@@ -76,6 +76,18 @@ type DataFlowSpec struct {
 	// +optional
 	CheckpointPersistence *bool `json:"checkpointPersistence,omitempty"`
 
+	// CheckpointSyncOnAck persists checkpoint to the ConfigMap immediately after each sink batch ack.
+	// Shrinks the duplicate window on pod crash from the debounce interval to roughly one batch.
+	// Default: false (debounced save only). Recommended for migration and cron workloads.
+	// +optional
+	CheckpointSyncOnAck *bool `json:"checkpointSyncOnAck,omitempty"`
+
+	// CheckpointSaveInterval controls how often pending checkpoints are flushed to the ConfigMap.
+	// Also used as the minimum coalesce interval when checkpointSyncOnAck is true.
+	// Default: 30s.
+	// +optional
+	CheckpointSaveInterval *metav1.Duration `json:"checkpointSaveInterval,omitempty"`
+
 	// ChannelBufferSize is the buffer size for message channels between source, processor, and sink.
 	// Larger values reduce blocking when sink is slower than source (e.g. high Kafka throughput).
 	// Default: 100. Recommended for high throughput: 500–1000.
@@ -733,6 +745,17 @@ type PostgreSQLSinkSpec struct {
 	// +optional
 	ConflictKey *string `json:"conflictKey,omitempty"`
 
+	// UpsertVersionColumn is the column used to compare row versions when UpsertStrategy is ifNewer.
+	// On conflict, the existing row is updated only when EXCLUDED.<column> > target.<column>.
+	// +optional
+	UpsertVersionColumn *string `json:"upsertVersionColumn,omitempty"`
+
+	// UpsertStrategy controls conflict resolution: always (default) updates on every conflict;
+	// ifNewer updates only when the incoming version is newer (requires upsertVersionColumn).
+	// +optional
+	// +kubebuilder:validation:Enum=always;ifNewer
+	UpsertStrategy *string `json:"upsertStrategy,omitempty"`
+
 	// SoftDeleteColumn specifies column for soft delete (e.g. "deleted_at"). If set, DELETE operations will UPDATE this column instead of physical delete.
 	// +optional
 	SoftDeleteColumn *string `json:"softDeleteColumn,omitempty"`
@@ -785,6 +808,16 @@ type TrinoSinkSpec struct {
 	// +optional
 	AutoCreateTable *bool `json:"autoCreateTable,omitempty"`
 
+	// UpsertMode enables idempotent writes via MERGE (Iceberg catalog).
+	// Requires conflictKey. Retries and replays update existing rows instead of creating duplicates.
+	// +optional
+	UpsertMode *bool `json:"upsertMode,omitempty"`
+
+	// ConflictKey specifies the column used to match rows in MERGE ON clause.
+	// Required when upsertMode is true.
+	// +optional
+	ConflictKey *string `json:"conflictKey,omitempty"`
+
 	// RawMode when true, creates table with data VARCHAR column (JSON storage).
 	// When false (default), uses columnar format matching message keys to table columns.
 	// +optional
@@ -833,6 +866,27 @@ type ClickHouseSinkSpec struct {
 	// +optional
 	AutoCreateTable *bool `json:"autoCreateTable,omitempty"`
 
+	// UpsertMode enables idempotent writes via ReplacingMergeTree deduplication.
+	// Inserts remain INSERT; duplicates are resolved on merge using conflictKey ORDER BY.
+	// +optional
+	UpsertMode *bool `json:"upsertMode,omitempty"`
+
+	// ConflictKey specifies the deduplication key column for ORDER BY when upsertMode is true.
+	// If not set, the first message column (or created_at in raw mode) is used.
+	// +optional
+	ConflictKey *string `json:"conflictKey,omitempty"`
+
+	// TableEngine selects the MergeTree engine variant (default: MergeTree).
+	// When upsertMode is true and tableEngine is unset, ReplacingMergeTree is used for auto-created tables.
+	// +optional
+	// +kubebuilder:validation:Enum=MergeTree;ReplacingMergeTree
+	TableEngine *string `json:"tableEngine,omitempty"`
+
+	// ReplacingVersionColumn is the version column for ReplacingMergeTree(engine).
+	// Rows with the highest version value are kept during background merges.
+	// +optional
+	ReplacingVersionColumn *string `json:"replacingVersionColumn,omitempty"`
+
 	// RawMode when true, creates table with data String and created_at columns (JSON storage).
 	// When false (default), creates table from message structure (columnar, replicates source schema).
 	// +optional

api/v1/dataflow_validation.go

@@ -479,6 +479,26 @@ func validatePostgreSQLSink(p *PostgreSQLSinkSpec, f *field.Path) field.ErrorLis
 	if p.TableSecretRef != nil {
 		all = append(all, validateSecretRef(p.TableSecretRef, f.Child("tableSecretRef"))...)
 	}
+	if p.UpsertMode != nil && *p.UpsertMode {
+		if err := validateSQLIdentifier(resolveConflictKeyForValidation(p.ConflictKey), f.Child("conflictKey")); err != nil {
+			all = append(all, err)
+		}
+	}
+	if p.UpsertStrategy != nil && *p.UpsertStrategy == "ifNewer" {
+		if p.UpsertVersionColumn == nil || strings.TrimSpace(*p.UpsertVersionColumn) == "" {
+			all = append(all, field.Required(f.Child("upsertVersionColumn"), "upsertVersionColumn is required when upsertStrategy is ifNewer"))
+		} else if err := validateSQLIdentifier(*p.UpsertVersionColumn, f.Child("upsertVersionColumn")); err != nil {
+			all = append(all, err)
+		}
+	}
+	if p.UpsertVersionColumn != nil && strings.TrimSpace(*p.UpsertVersionColumn) != "" {
+		if err := validateSQLIdentifier(*p.UpsertVersionColumn, f.Child("upsertVersionColumn")); err != nil {
+			all = append(all, err)
+		}
+	}
+	if p.UpsertStrategy != nil && *p.UpsertStrategy != "" && *p.UpsertStrategy != "always" && *p.UpsertStrategy != "ifNewer" {
+		all = append(all, field.Invalid(f.Child("upsertStrategy"), *p.UpsertStrategy, "must be always or ifNewer"))
+	}
 	all = append(all, validateFlattenMetadataSpec(p.RawMode, p.FlattenMetadataColumns, p.FlattenMetadataColumnsPrefix, f)...)
 	return all
 }
@@ -513,6 +533,19 @@ func validateTrinoSink(t *TrinoSinkSpec, f *field.Path) field.ErrorList {
 	if t.TableSecretRef != nil {
 		all = append(all, validateSecretRef(t.TableSecretRef, f.Child("tableSecretRef"))...)
 	}
+	if t.UpsertMode != nil && *t.UpsertMode {
+		conflictKey := resolveConflictKeyForValidation(t.ConflictKey)
+		if conflictKey == "" {
+			all = append(all, field.Required(f.Child("conflictKey"), "conflictKey is required when upsertMode is enabled"))
+		} else if err := validateSQLIdentifier(conflictKey, f.Child("conflictKey")); err != nil {
+			all = append(all, err)
+		}
+		catalog := strings.ToLower(t.Catalog)
+		if catalog != "" && !strings.Contains(catalog, "iceberg") {
+			all = append(all, field.Invalid(f.Child("catalog"), t.Catalog,
+				"upsertMode requires an Iceberg catalog (catalog name should contain \"iceberg\")"))
+		}
+	}
 	all = append(all, validateFlattenMetadataSpec(t.RawMode, t.FlattenMetadataColumns, t.FlattenMetadataColumnsPrefix, f)...)
 	return all
 }
@@ -558,10 +591,32 @@ func validateClickHouseSink(c *ClickHouseSinkSpec, f *field.Path) field.ErrorLis
 	if c.TableSecretRef != nil {
 		all = append(all, validateSecretRef(c.TableSecretRef, f.Child("tableSecretRef"))...)
 	}
+	if c.UpsertMode != nil && *c.UpsertMode {
+		if c.ConflictKey != nil && strings.TrimSpace(*c.ConflictKey) != "" {
+			if err := validateSQLIdentifier(*c.ConflictKey, f.Child("conflictKey")); err != nil {
+				all = append(all, err)
+			}
+		}
+	}
+	if c.TableEngine != nil && *c.TableEngine != "" && *c.TableEngine != "MergeTree" && *c.TableEngine != "ReplacingMergeTree" {
+		all = append(all, field.Invalid(f.Child("tableEngine"), *c.TableEngine, "must be MergeTree or ReplacingMergeTree"))
+	}
+	if c.ReplacingVersionColumn != nil && strings.TrimSpace(*c.ReplacingVersionColumn) != "" {
+		if err := validateSQLIdentifier(*c.ReplacingVersionColumn, f.Child("replacingVersionColumn")); err != nil {
+			all = append(all, err)
+		}
+	}
 	all = append(all, validateFlattenMetadataSpec(c.RawMode, c.FlattenMetadataColumns, c.FlattenMetadataColumnsPrefix, f)...)
 	return all
 }
 
+func resolveConflictKeyForValidation(conflictKey *string) string {
+	if conflictKey == nil {
+		return ""
+	}
+	return strings.TrimSpace(*conflictKey)
+}
+
 func validateSecretRef(r *SecretRef, f *field.Path) field.ErrorList {
 	var all field.ErrorList
 	if r == nil {

api/v1/dataflow_validation_upsert_test.go

@@ -0,0 +1,76 @@
+/*
+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 (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+)
+
+func TestValidatePostgreSQLSink_ifNewerRequiresVersionColumn(t *testing.T) {
+	strategy := "ifNewer"
+	spec := &PostgreSQLSinkSpec{
+		ConnectionString: "postgres://localhost/db",
+		Table:            "t",
+		UpsertStrategy:   &strategy,
+	}
+	errs := validatePostgreSQLSink(spec, field.NewPath("sink"))
+	assert.NotEmpty(t, errs)
+}
+
+func TestValidateTrinoSink_upsertRequiresIcebergCatalog(t *testing.T) {
+	upsertMode := true
+	conflictKey := "id"
+	spec := &TrinoSinkSpec{
+		ServerURL:   "http://trino:8080",
+		Catalog:     "hive",
+		Schema:      "default",
+		Table:       "t",
+		UpsertMode:  &upsertMode,
+		ConflictKey: &conflictKey,
+	}
+	errs := validateTrinoSink(spec, field.NewPath("sink"))
+	assert.NotEmpty(t, errs)
+}
+
+func TestValidateTrinoSink_upsertValidIceberg(t *testing.T) {
+	upsertMode := true
+	conflictKey := "id"
+	spec := &TrinoSinkSpec{
+		ServerURL:   "http://trino:8080",
+		Catalog:     "nessie_iceberg",
+		Schema:      "default",
+		Table:       "t",
+		UpsertMode:  &upsertMode,
+		ConflictKey: &conflictKey,
+	}
+	errs := validateTrinoSink(spec, field.NewPath("sink"))
+	assert.Empty(t, errs)
+}
+
+func TestValidateClickHouseSink_upsertEngineFields(t *testing.T) {
+	badEngine := "SummingMergeTree"
+	spec := &ClickHouseSinkSpec{
+		ConnectionString: "clickhouse://localhost:9000",
+		Table:            "t",
+		TableEngine:      &badEngine,
+	}
+	errs := validateClickHouseSink(spec, field.NewPath("sink"))
+	assert.NotEmpty(t, errs)
+}