cloudquery · kodiakhq · Jul 2, 2025 · Jun 24, 2025 · Jun 26, 2025 · Jul 1, 2025
@@ -21,31 +21,10 @@ jobs:
     name: 'plugins/destination/clickhouse'
     runs-on: ubuntu-latest
     timeout-minutes: 30
-    env:
-      DB_USER: cq
-      DB_PASSWORD: test
-      DB_NAME: cloudquery
     defaults:
       run:
         working-directory: plugins/destination/clickhouse
-    services:
-      clickhouse:
-        image: clickhouse/clickhouse-server:24.8.1
-        env:
-          CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
-          CLICKHOUSE_PASSWORD: ${{ env.DB_PASSWORD }}
-          CLICKHOUSE_USER: ${{ env.DB_USER }}
-          CLICKHOUSE_DB: ${{ env.DB_NAME }}
-        ports:
-          - 8123:8123
-          - 9000:9000
-        options: >-
-          --ulimit nofile=262144:262144
-          --health-cmd "wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1"
-          --health-interval 60s
-          --health-timeout 30s
-          --health-start-period 20s
-          --health-retries 5
+
     steps:
       - uses: actions/checkout@v4
         with:
@@ -70,9 +49,13 @@ jobs:
         run: test "$(git status -s | wc -l)" -eq 0 || (git status -s; exit 1)
       - name: Build
         run: go build .
+
+      - uses: hoverkraft-tech/compose-action@ad8e0e414a8244c238d7071359bbf04d1e50cd79
+        with:
+          compose-file: docker-compose.yaml
+          up-flags: "--wait"
+          cwd: plugins/destination/clickhouse
       - name: Test ClickHouse
-        env:
-          CQ_DEST_CH_TEST_CONN: 'clickhouse://${{ env.DB_USER }}:${{ env.DB_PASSWORD }}@localhost:9000/${{ env.DB_NAME }}'
         run: make test
 
   deploy:

@@ -2,7 +2,7 @@
 test:
 # we clean the cache to avoid scenarios when we change something in the db and we want to retest without noticing nothing run
 	go clean -testcache
-	go test -race -timeout 3m ./...
+	go test -v -race -timeout 3m ./...
 
 .PHONY: lint
 lint:

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"strconv"
 	"testing"
 	"time"
 
@@ -20,7 +21,9 @@ import (
 	sdkTypes "github.com/cloudquery/plugin-sdk/v4/types"
 	"github.com/goccy/go-json"
 	"github.com/google/uuid"
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
 )
 
 func getTestConnection() string {
@@ -228,3 +231,85 @@ func TestMigrateNewArrayAndMapColumns(t *testing.T) {
 		t.Fatal(fmt.Errorf("failed to insert record: %w", err))
 	}
 }
+
+func TestConcurrentSyncsSameTable(t *testing.T) {
+	const syncConcurrency = 2000
+	ctx := context.Background()
+	group, _ := errgroup.WithContext(ctx)
+	randomUUIDString := uuid.New().String()
+	tableName := "k8s_core_namespaces_" + randomUUIDString
+	table := &schema.Table{
+		Name: tableName,
+		Columns: []schema.Column{
+			schema.CqIDColumn,
+			schema.CqSourceNameColumn,
+			schema.CqSyncTimeColumn,
+		},
+	}
+	// Create the table
+	p := plugin.NewPlugin("clickhouse",
+		internalPlugin.Version,
+		New,
+		plugin.WithJSONSchema(spec.JSONSchema),
+	)
+	s := &spec.Spec{ConnectionString: getTestConnection()}
+	b, err := json.Marshal(s)
+	require.NoError(t, err)
+	err = p.Init(ctx, b, plugin.NewClientOptions{})
+	require.NoError(t, err)
+	if err := p.WriteAll(ctx, []message.WriteMessage{&message.WriteMigrateTable{Table: table}}); err != nil {
+		t.Fatal(fmt.Errorf("failed to create table: %w", err))
+	}
+
+	for i := range syncConcurrency {
+		group.Go(func() error {
+			// Simulate a sync job against the same table
+			syncContext := context.Background()
+			p := plugin.NewPlugin("clickhouse",
+				internalPlugin.Version,
+				New,
+				plugin.WithJSONSchema(spec.JSONSchema),
+			)
+			p.SetLogger(zerolog.New(zerolog.NewTestWriter(t)).Level(zerolog.WarnLevel))
+			s := &spec.Spec{ConnectionString: getTestConnection()}
+			b, err := json.Marshal(s)
+			require.NoError(t, err)
+			err = p.Init(syncContext, b, plugin.NewClientOptions{})
+			require.NoError(t, err)
+			if err := p.WriteAll(syncContext, []message.WriteMessage{&message.WriteMigrateTable{Table: table}}); err != nil {
+				t.Fatal(fmt.Errorf("failed to create table: %w", err))
+			}
+
+			jobIndexAsString := strconv.Itoa(i)
+			randomUUIDStringWithLastCharacterReplaced := randomUUIDString[:len(randomUUIDString)-len(jobIndexAsString)] + jobIndexAsString
+			bldr := array.NewRecordBuilder(memory.DefaultAllocator, table.ToArrowSchema())
+			bldr.Field(0).(*sdkTypes.UUIDBuilder).Append(uuid.MustParse(randomUUIDStringWithLastCharacterReplaced))
+			bldr.Field(1).(*array.StringBuilder).Append("source")
+			bldr.Field(2).(*array.TimestampBuilder).Append(arrow.Timestamp(time.Now().UnixMicro()))
+			record := bldr.NewRecord()
+
+			if err := p.WriteAll(syncContext, []message.WriteMessage{&message.WriteInsert{
+				Record: record,
+			}}); err != nil {
+				t.Fatal(fmt.Errorf("failed to insert record: %w", err))
+			}
+			return nil
+		})
+	}
+
+	require.NoError(t, group.Wait())
+
+	ch := make(chan arrow.Record)
+	go func() {
+		defer close(ch)
+		err = p.Read(ctx, table, ch)
+	}()
+
+	numRows := 0
+	for record := range ch {
+		numRows += int(record.NumRows())
+	}
+
+	require.Equal(t, syncConcurrency, numRows)
+	require.NoError(t, err)
+}
@@ -18,7 +18,7 @@ func (c *Client) DeleteStale(ctx context.Context, messages message.WriteDeleteSt
 	}
 
 	for _, msg := range messages {
-		if err := c.conn.Exec(ctx, generateDeleteForDeleteStale(msg), msg.SourceName, msg.SyncTime); err != nil {
+		if err := retryExec(ctx, c.logger, c.conn, generateDeleteForDeleteStale(msg), msg.SourceName, msg.SyncTime); err != nil {
 			return err
 		}
 	}
@@ -50,7 +50,7 @@ func (c *Client) DeleteRecord(ctx context.Context, messages message.WriteDeleteR
 			return err
 		}
 
-		if err = c.conn.Exec(ctx, sql, params...); err != nil {
+		if err = retryExec(ctx, c.logger, c.conn, sql, params...); err != nil {
 			return err
 		}
 	}