Update destination keyColumns logic, rename general config to config in errors

config/configuration.go

@@ -29,7 +29,7 @@ const (
 	KeyColumns = "keyColumns"
 )
 
-// Configuration is the general configurations needed to connect to ClickHouse database.
+// Configuration is the configuration needed to connect to ClickHouse database.
 type Configuration struct {
 	// URL is the configuration of the connection string to connect to ClickHouse database.
 	URL string `validate:"required"`
@@ -39,7 +39,7 @@ type Configuration struct {
 	KeyColumns []string
 }
 
-// parses a general configuration.
+// parses a configuration.
 func parseConfiguration(cfg map[string]string) (Configuration, error) {
 	config := Configuration{
 		URL:   strings.TrimSpace(cfg[URL]),
@@ -48,7 +48,7 @@ func parseConfiguration(cfg map[string]string) (Configuration, error) {
 
 	err := validate(config)
 	if err != nil {
-		return Configuration{}, fmt.Errorf("validate general config: %w", err)
+		return Configuration{}, fmt.Errorf("validate config: %w", err)
 	}
 
 	if cfg[KeyColumns] == "" {

config/configuration_test.go

@@ -27,7 +27,7 @@ const (
 	testTable = "test_table"
 )
 
-func TestParseGeneral(t *testing.T) {
+func TestParseConfiguration(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
@@ -170,19 +170,19 @@ func TestParseGeneral(t *testing.T) {
 			in: map[string]string{
 				Table: testTable,
 			},
-			err: fmt.Errorf("validate general config: %w", fmt.Errorf("%q must be set", URL)),
+			err: fmt.Errorf("validate config: %w", fmt.Errorf("%q must be set", URL)),
 		},
 		{
 			name: "failure_required_table",
 			in: map[string]string{
 				URL: testURL,
 			},
-			err: fmt.Errorf("validate general config: %w", fmt.Errorf("%q must be set", Table)),
+			err: fmt.Errorf("validate config: %w", fmt.Errorf("%q must be set", Table)),
 		},
 		{
 			name: "failure_required_url_and_table",
 			in:   map[string]string{},
-			err: fmt.Errorf("validate general config: %w",
+			err: fmt.Errorf("validate config: %w",
 				multierr.Combine(fmt.Errorf("%q must be set", URL), fmt.Errorf("%q must be set", Table))),
 		},
 	}

config/destination.go

@@ -27,7 +27,7 @@ type Destination struct {
 func ParseDestination(cfg map[string]string) (Destination, error) {
 	config, err := parseConfiguration(cfg)
 	if err != nil {
-		return Destination{}, fmt.Errorf("parse general config: %w", err)
+		return Destination{}, fmt.Errorf("parse config: %w", err)
 	}
 
 	destinationConfig := Destination{

config/source.go

@@ -47,7 +47,7 @@ type Source struct {
 func ParseSource(cfg map[string]string) (Source, error) {
 	config, err := parseConfiguration(cfg)
 	if err != nil {
-		return Source{}, fmt.Errorf("parse general config: %w", err)
+		return Source{}, fmt.Errorf("parse source config: %w", err)
 	}
 
 	sourceConfig := Source{

config/validator.go

@@ -43,7 +43,7 @@ func validate(s interface{}) error {
 	validationErr := Get().Struct(s)
 	if validationErr != nil {
 		if _, ok := validationErr.(*v.InvalidValidationError); ok {
-			return fmt.Errorf("validate general config struct: %w", validationErr)
+			return fmt.Errorf("validate struct: %w", validationErr)
 		}
 
 		for _, e := range validationErr.(v.ValidationErrors) {

destination/destination_test.go

@@ -57,7 +57,7 @@ func TestDestination_Configure_Fail(t *testing.T) {
 		config.URL: testURL,
 	})
 	is.Equal(err.Error(),
-		`parse destination config: parse general config: validate general config: "table" must be set`)
+		`parse destination config: parse config: validate config: "table" must be set`)
 }
 
 func TestDestination_Write_Success(t *testing.T) {

destination/writer/writer.go

@@ -142,9 +142,12 @@ func (w *Writer) Update(ctx context.Context, record sdk.Record) error {
 		key = make(sdk.StructuredData)
 
 		for i := range w.keyColumns {
-			if val, ok := payload[w.keyColumns[i]]; ok {
-				key[w.keyColumns[i]] = val
+			val, ok := payload[w.keyColumns[i]]
+			if !ok {
+				return fmt.Errorf("key column %q not found", w.keyColumns[i])
 			}
+
+			key[w.keyColumns[i]] = val
 		}
 	}