Merge branch 'main' into renovate/github.com-cloudquery-plugin-sdk-v4-4.x

erezrokah · web-flow · commit 755722ca09d5 · 2025-06-23T10:16:21.000+01:00
diff --git a/plugins/destination/clickhouse/client/client.go b/plugins/destination/clickhouse/client/client.go
@@ -22,7 +22,6 @@ type Client struct {
 	logger zerolog.Logger
 	writer *batchwriter.BatchWriter
 	plugin.UnimplementedSource
-	batchwriter.UnimplementedDeleteStale
 }
 
 var _ plugin.Client = (*Client)(nil)
diff --git a/plugins/destination/clickhouse/client/client_test.go b/plugins/destination/clickhouse/client/client_test.go
@@ -41,8 +41,7 @@ func TestPlugin(t *testing.T) {
 	plugin.TestWriterSuiteRunner(t,
 		p,
 		plugin.WriterTestSuiteTests{
-			SkipUpsert:      true,
-			SkipDeleteStale: true,
+			SkipUpsert: true,
 			SafeMigrations: plugin.SafeMigrations{
 				AddColumn:    true,
 				RemoveColumn: true,
diff --git a/plugins/destination/clickhouse/client/delete.go b/plugins/destination/clickhouse/client/delete.go
@@ -9,15 +9,42 @@ import (
 	"github.com/cloudquery/cloudquery/plugins/destination/clickhouse/v7/typeconv/ch/values"
 	"github.com/cloudquery/cloudquery/plugins/destination/clickhouse/v7/util"
 	"github.com/cloudquery/plugin-sdk/v4/message"
+	"github.com/cloudquery/plugin-sdk/v4/schema"
 )
 
+func (c *Client) DeleteStale(ctx context.Context, messages message.WriteDeleteStales) error {
+	if len(messages) == 0 {
+		return nil
+	}
+
+	for _, msg := range messages {
+		if err := c.conn.Exec(ctx, generateDeleteForDeleteStale(msg), msg.SourceName, msg.SyncTime); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func generateDeleteForDeleteStale(msg *message.WriteDeleteStale) string {
+	var sb strings.Builder
+	sb.WriteString("DELETE FROM ")
+	sb.WriteString(util.SanitizeID(msg.TableName))
+	sb.WriteString(" WHERE ")
+	sb.WriteString(util.SanitizeID(schema.CqSourceNameColumn.Name))
+	sb.WriteString(" = $1 AND toTimeZone(")
+	sb.WriteString(util.SanitizeID(schema.CqSyncTimeColumn.Name))
+	sb.WriteString(", 'UTC') < $2")
+	return sb.String()
+}
+
 func (c *Client) DeleteRecord(ctx context.Context, messages message.WriteDeleteRecords) error {
 	if len(messages) == 0 {
 		return nil
 	}
 
 	for _, msg := range messages {
-		sql := generateDelete(msg.DeleteRecord)
+		sql := generateDeleteForDeleteRecord(msg.DeleteRecord)
 		params, err := extractPredicateValues(msg.DeleteRecord.WhereClause)
 		if err != nil {
 			return err
@@ -62,7 +89,7 @@ func unpackArray(s any) []any {
 	return r
 }
 
-func generateDelete(msg message.DeleteRecord) string {
+func generateDeleteForDeleteRecord(msg message.DeleteRecord) string {
 	var sb strings.Builder
 
 	sb.WriteString("DELETE FROM ")

Original file line number	Diff line number	Diff line change
`@@ -22,7 +22,6 @@ type Client struct {`
`22`	`22`	`logger zerolog.Logger`
`23`	`23`	`writer *batchwriter.BatchWriter`
`24`	`24`	`plugin.UnimplementedSource`
`25`		`- batchwriter.UnimplementedDeleteStale`
`26`	`25`	`}`
`27`	`26`
`28`	`27`	`var _ plugin.Client = (*Client)(nil)`