Add the snapshot configuration field (#28)

* Add the snapshot configuration field

* Return records with the different mode as records with the different operation (#29)

* Return records with the different mode as records with the different operation

* Add unit tests for iterator's position

* Delete unnecessary integration tests, refactoring (#30)

* Delete unnecessary integration tests

* Update comments of methods

* Update codeowners file

* Update CODEOWNERS (#31)

* Remove  configuration parameter (#36)

* Fix: copyright year doesn't have to be the current year

* review + refactor

* fixes

* upgrade conduit sdk and protocol

* go mod updates

---------

Co-authored-by: Haris Osmanagić <haris.osmanagic@gmail.com>
Co-authored-by: maha-hajja <maha.d.hajja@gmail.com>

.github/CODEOWNERS

@@ -1,3 +1,3 @@
 # Define code owners (individuals or teams that are responsible for code in this repository)
 # More about code owners at https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners
-*       @conduitio-labs/conduit-core
+*       @conduitio-labs/conduit-core

.github/workflows/lint.yml

@@ -11,10 +11,9 @@ jobs:
     steps:
       - uses: actions/setup-go@v4
         with:
-          go-version: "1.20"
+          go-version: '1.20'
       - uses: actions/checkout@v3
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3.7.0
+        uses: golangci/golangci-lint-action@v3.6.0
         with:
-          version: v1.51.2
-          args: --timeout 2m
+          version: v1.53.3

.goreleaser.yml

@@ -11,12 +11,13 @@ builds:
 checksum:
   name_template: checksums.txt
 archives:
-  - replacements:
-      darwin: Darwin
-      linux: Linux
-      windows: Windows
-      386: i386
-      amd64: x86_64
+  - name_template: >-
+      {{ .ProjectName }}_
+      {{- .Version }}_
+      {{- title .Os }}_
+      {{- if eq .Arch "amd64" }}x86_64
+      {{- else if eq .Arch "386" }}i386
+      {{- else }}{{ .Arch }}{{ end }}
 changelog:
   sort: asc
   use: github
@@ -26,4 +27,4 @@ changelog:
       - '^test:'
       - '^go.mod:'
       - '^.github:'
-      - Merge branch
+      - Merge branch

Makefile

@@ -1,4 +1,4 @@
-.PHONY: build test
+.PHONY: build test lint dep mockgen
 
 VERSION=$(shell git describe --tags --dirty --always)
 

README.md

@@ -10,7 +10,7 @@ Connector uses [Golang SQL database driver](https://github.com/ClickHouse/clickh
 ## Prerequisites
 
 - [Go](https://go.dev/) 1.20
-- (optional) [golangci-lint](https://github.com/golangci/golangci-lint) 1.51.2
+- (optional) [golangci-lint](https://github.com/golangci/golangci-lint) 1.53.3
 
 ## How to build it
 
@@ -58,8 +58,8 @@ configuration.
 | `url`            | [DSN](https://github.com/ClickHouse/clickhouse-go#dsn) to connect to the database.                                                                                  | **true** | `http://username:password@host1:8123/database` |
 | `table`          | Name of the table that the connector should read.                                                                                                                   | **true** | `table_name`                                   |
 | `orderingColumn` | Column name that the connector will use for ordering rows. Column must contain unique values and suitable for sorting, otherwise the snapshot won't work correctly. | **true** | `id`                                           |
+| `snapshot`       | whether to take a snapshot of the entire table before starting CDC mode or not. The default is `true`.                                                              | false    | `false`, `true`                                |
 | `keyColumns`     | Comma-separated list of column names to build the `sdk.Record.Key`. See more: [key handling](#key-handling).                                                        | false    | `id,name`                                      |
-| `columns`        | Comma-separated list of column names that should be included in each payload of the `sdk.Record`. By default includes all columns.                                  | false    | `id,name,age`                                  |
 | `batchSize`      | Size of rows batch. Min is 1 and max is 100000. The default is 1000.                                                                                                | false    | `100`                                          |
 
 #### Key handling

acceptance_test.go

@@ -29,21 +29,30 @@ import (
 	"github.com/matryer/is"
 )
 
+const (
+	// driverName is a database driver name.
+	driverName = "clickhouse"
+	// envNameURL is a ClickHouse url environment name.
+	envNameURL = "CLICKHOUSE_URL"
+	// metadataFieldTable is a name of a record metadata field that stores a ClickHouse table name.
+	metadataFieldTable = "clickhouse.table"
+)
+
 type driver struct {
 	sdk.ConfigurableAcceptanceTestDriver
 
 	id int32
 }
 
 // GenerateRecord generates a random sdk.Record.
-func (d *driver) GenerateRecord(t *testing.T, operation sdk.Operation) sdk.Record {
+func (d *driver) GenerateRecord(_ *testing.T, operation sdk.Operation) sdk.Record {
 	atomic.AddInt32(&d.id, 1)
 
 	return sdk.Record{
 		Position:  nil,
 		Operation: operation,
 		Metadata: map[string]string{
-			"clickhouse.table": d.Config.SourceConfig[config.Table],
+			metadataFieldTable: d.Config.SourceConfig[config.Table],
 		},
 		Key: sdk.StructuredData{
 			"Int32Type": d.id,
@@ -81,9 +90,9 @@ func TestAcceptance(t *testing.T) {
 // prepareConfig receives the connection URL from the environment variable
 // and prepares configuration map.
 func prepareConfig(t *testing.T) map[string]string {
-	url := os.Getenv("CLICKHOUSE_URL")
+	url := os.Getenv(envNameURL)
 	if url == "" {
-		t.Skip("CLICKHOUSE_URL env var must be set")
+		t.Skipf("%s env var must be set", envNameURL)
 
 		return nil
 	}
@@ -93,12 +102,14 @@ func prepareConfig(t *testing.T) map[string]string {
 		config.Table:          fmt.Sprintf("CONDUIT_TEST_%s", randString(6)),
 		config.KeyColumns:     "Int32Type",
 		config.OrderingColumn: "Int32Type",
+		config.Snapshot:       "true",
+		config.BatchSize:      "1000",
 	}
 }
 
 // createTable creates test table.
 func createTable(url, table string) error {
-	db, err := sqlx.Open("clickhouse", url)
+	db, err := sqlx.Open(driverName, url)
 	if err != nil {
 		return fmt.Errorf("open connection: %w", err)
 	}
@@ -119,7 +130,7 @@ func createTable(url, table string) error {
 
 // dropTables drops test table and tracking test table if exists.
 func dropTables(url, table string) error {
-	db, err := sqlx.Open("clickhouse", url)
+	db, err := sqlx.Open(driverName, url)
 	if err != nil {
 		return fmt.Errorf("open connection: %w", err)
 	}

config/config.go

@@ -0,0 +1,35 @@
+// Copyright © 2023 Meroxa, Inc. & Yalantis
+//
+// 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 config
+
+const (
+	// URL is the configuration name of the url.
+	URL = "url"
+	// Table is the configuration name of the table.
+	Table = "table"
+	// KeyColumns is the configuration name of key column names (for the DestConfig),
+	// or of the names of the columns to build the record.Key (for the SourceConfig), separated by commas.
+	KeyColumns = "keyColumns"
+)
+
+// Config is the configuration needed to connect to ClickHouse database.
+type Config struct {
+	// URL is the configuration of the connection string to connect to ClickHouse database.
+	URL string `json:"url" validate:"required"`
+	// Table is the configuration of the table name.
+	Table string `validate:"required"`
+	// KeyColumns is the configuration of key column names, separated by commas.
+	KeyColumns []string
+}