Remove configuration parameter (#36)

conduitio-labs · Dec 20, 2022 · 4e357e6 · 4e357e6
1 parent 53c93b8
commit 4e357e6
Show file tree

Hide file tree

Showing 6 changed files with 2 additions and 108 deletions.
diff --git a/README.md b/README.md
@@ -59,7 +59,6 @@ configuration.
 | `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`                                           |
 | `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

diff --git a/config/configuration.go b/config/configuration.go
@@ -74,7 +74,6 @@ func getKeyName(fieldName string) string {
 		"Table":          Table,
 		"KeyColumns":     KeyColumns,
 		"OrderingColumn": OrderingColumn,
-		"Columns":        Columns,
 		"BatchSize":      BatchSize,
 	}[fieldName]
 }
diff --git a/config/source.go b/config/source.go
@@ -17,16 +17,13 @@ package config
 import (
 	"fmt"
 	"strconv"
-	"strings"
 )
 
 const (
 	// OrderingColumn is a config name for an ordering column.
 	OrderingColumn = "orderingColumn"
 	// Snapshot is the configuration name for the Snapshot field.
 	Snapshot = "snapshot"
-	// Columns is the config name for a list of columns, separated by commas.
-	Columns = "columns"
 	// BatchSize is the config name for a batch size.
 	BatchSize = "batchSize"
 
@@ -45,8 +42,6 @@ type Source struct {
 	// Snapshot is the configuration that determines whether the connector
 	// will take a snapshot of the entire table before starting cdc mode.
 	Snapshot bool
-	// Columns list of column names that should be included in each Record's payload.
-	Columns []string
 	// BatchSize is a size of rows batch.
 	BatchSize int `validate:"gte=1,lte=100000"`
 }
@@ -74,17 +69,6 @@ func ParseSource(cfg map[string]string) (Source, error) {
 		sourceConfig.Snapshot = snapshot
 	}
 
-	if cfg[Columns] != "" {
-		columnsSl := strings.Split(strings.ReplaceAll(cfg[Columns], " ", ""), ",")
-		for i := range columnsSl {
-			if columnsSl[i] == "" {
-				return Source{}, fmt.Errorf("invalid %q", Columns)
-			}
-
-			sourceConfig.Columns = append(sourceConfig.Columns, columnsSl[i])
-		}
-	}
-
 	if cfg[BatchSize] != "" {
 		sourceConfig.BatchSize, err = strconv.Atoi(cfg[BatchSize])
 		if err != nil {
@@ -97,24 +81,5 @@ func ParseSource(cfg map[string]string) (Source, error) {
 		return Source{}, err
 	}
 
-	if len(sourceConfig.Columns) == 0 {
-		return sourceConfig, nil
-	}
-
-	columnsMap := make(map[string]struct{}, len(sourceConfig.Columns))
-	for i := 0; i < len(sourceConfig.Columns); i++ {
-		columnsMap[sourceConfig.Columns[i]] = struct{}{}
-	}
-
-	if _, ok := columnsMap[sourceConfig.OrderingColumn]; !ok {
-		return Source{}, fmt.Errorf("columns must include %q", OrderingColumn)
-	}
-
-	for i := range sourceConfig.KeyColumns {
-		if _, ok := columnsMap[sourceConfig.KeyColumns[i]]; !ok {
-			return Source{}, fmt.Errorf("columns must include all %q", KeyColumns)
-		}
-	}
-
 	return sourceConfig, nil
 }
diff --git a/config/source_test.go b/config/source_test.go
@@ -124,7 +124,6 @@ func TestParseSource(t *testing.T) {
 				URL:            testURL,
 				Table:          testTable,
 				OrderingColumn: "id",
-				Columns:        "id",
 			},
 			want: Source{
 				Configuration: Configuration{
@@ -134,7 +133,6 @@ func TestParseSource(t *testing.T) {
 				OrderingColumn: "id",
 				Snapshot:       defaultSnapshot,
 				BatchSize:      defaultBatchSize,
-				Columns:        []string{"id"},
 			},
 		},
 		{
@@ -143,7 +141,6 @@ func TestParseSource(t *testing.T) {
 				URL:            testURL,
 				Table:          testTable,
 				OrderingColumn: "id",
-				Columns:        "id,name",
 			},
 			want: Source{
 				Configuration: Configuration{
@@ -153,7 +150,6 @@ func TestParseSource(t *testing.T) {
 				OrderingColumn: "id",
 				Snapshot:       defaultSnapshot,
 				BatchSize:      defaultBatchSize,
-				Columns:        []string{"id", "name"},
 			},
 		},
 		{
@@ -162,7 +158,6 @@ func TestParseSource(t *testing.T) {
 				URL:            testURL,
 				Table:          testTable,
 				OrderingColumn: "id",
-				Columns:        "id, name",
 			},
 			want: Source{
 				Configuration: Configuration{
@@ -172,7 +167,6 @@ func TestParseSource(t *testing.T) {
 				OrderingColumn: "id",
 				Snapshot:       defaultSnapshot,
 				BatchSize:      defaultBatchSize,
-				Columns:        []string{"id", "name"},
 			},
 		},
 		{
@@ -181,7 +175,6 @@ func TestParseSource(t *testing.T) {
 				URL:            testURL,
 				Table:          testTable,
 				OrderingColumn: "id",
-				Columns:        "id,name ",
 			},
 			want: Source{
 				Configuration: Configuration{
@@ -191,7 +184,6 @@ func TestParseSource(t *testing.T) {
 				OrderingColumn: "id",
 				Snapshot:       defaultSnapshot,
 				BatchSize:      defaultBatchSize,
-				Columns:        []string{"id", "name"},
 			},
 		},
 		{
@@ -200,7 +192,6 @@ func TestParseSource(t *testing.T) {
 				URL:            testURL,
 				Table:          testTable,
 				OrderingColumn: "id",
-				Columns:        " id,name",
 			},
 			want: Source{
 				Configuration: Configuration{
@@ -210,7 +201,6 @@ func TestParseSource(t *testing.T) {
 				OrderingColumn: "id",
 				Snapshot:       defaultSnapshot,
 				BatchSize:      defaultBatchSize,
-				Columns:        []string{"id", "name"},
 			},
 		},
 		{
@@ -219,7 +209,6 @@ func TestParseSource(t *testing.T) {
 				URL:            testURL,
 				Table:          testTable,
 				OrderingColumn: "id",
-				Columns:        "id ,name",
 			},
 			want: Source{
 				Configuration: Configuration{
@@ -229,7 +218,6 @@ func TestParseSource(t *testing.T) {
 				OrderingColumn: "id",
 				Snapshot:       defaultSnapshot,
 				BatchSize:      defaultBatchSize,
-				Columns:        []string{"id", "name"},
 			},
 		},
 		{
@@ -238,7 +226,6 @@ func TestParseSource(t *testing.T) {
 				URL:            testURL,
 				Table:          testTable,
 				OrderingColumn: "id",
-				Columns:        "id,  name",
 			},
 			want: Source{
 				Configuration: Configuration{
@@ -248,15 +235,13 @@ func TestParseSource(t *testing.T) {
 				OrderingColumn: "id",
 				Snapshot:       defaultSnapshot,
 				BatchSize:      defaultBatchSize,
-				Columns:        []string{"id", "name"},
 			},
 		},
 		{
 			name: "failure_required_orderingColumn",
 			in: map[string]string{
-				URL:     testURL,
-				Table:   testTable,
-				Columns: "id,name,age",
+				URL:   testURL,
+				Table: testTable,
 			},
 			err: fmt.Errorf("%q must be set", OrderingColumn),
 		},
@@ -280,27 +265,6 @@ func TestParseSource(t *testing.T) {
 			},
 			err: fmt.Errorf(`invalid %q: strconv.Atoi: parsing "a": invalid syntax`, BatchSize),
 		},
-		{
-			name: "failure_missed_orderingColumn_in_columns",
-			in: map[string]string{
-				URL:            testURL,
-				Table:          testTable,
-				OrderingColumn: "id",
-				Columns:        "name,age",
-			},
-			err: fmt.Errorf("columns must include %q", OrderingColumn),
-		},
-		{
-			name: "failure_missed_keyColumn_in_columns",
-			in: map[string]string{
-				URL:            testURL,
-				Table:          testTable,
-				OrderingColumn: "id",
-				KeyColumns:     "name",
-				Columns:        "id,age",
-			},
-			err: fmt.Errorf("columns must include all %q", KeyColumns),
-		},
 		{
 			name: "failure_batchSize_is_too_big",
 			in: map[string]string{
@@ -331,26 +295,6 @@ func TestParseSource(t *testing.T) {
 			},
 			err: fmt.Errorf("%w", fmt.Errorf("%q is out of range", BatchSize)),
 		},
-		{
-			name: "failure_columns_ends_with_comma",
-			in: map[string]string{
-				URL:            testURL,
-				Table:          testTable,
-				OrderingColumn: "id",
-				Columns:        "id,name,",
-			},
-			err: fmt.Errorf("invalid %q", Columns),
-		},
-		{
-			name: "failure_columns_starts_with_comma",
-			in: map[string]string{
-				URL:            testURL,
-				Table:          testTable,
-				OrderingColumn: "id",
-				Columns:        ",id,name",
-			},
-			err: fmt.Errorf("invalid %q", Columns),
-		},
 	}
 
 	for _, tt := range tests {

diff --git a/source/iterator/iterator.go b/source/iterator/iterator.go
@@ -43,9 +43,6 @@ type Iterator struct {
 	keyColumns []string
 	// name of the column that iterator will use for sorting data
 	orderingColumn string
-	// list of table's columns for record payload.
-	// if empty - will get all columns
-	columns []string
 	// size of batch
 	batchSize int
 }
@@ -209,10 +206,6 @@ func (iter *Iterator) loadRows(ctx context.Context) error {
 		OrderBy(iter.orderingColumn).
 		Limit(iter.batchSize)
 
-	if len(iter.columns) > 0 {
-		sb.Select(iter.columns...)
-	}
-
 	if iter.position.LastProcessedValue != nil {
 		sb.Where(sb.GreaterThan(iter.orderingColumn, iter.position.LastProcessedValue))
 	}

diff --git a/source/source.go b/source/source.go
@@ -75,12 +75,6 @@ func (s *Source) Parameters() map[string]sdk.Parameter {
 			Required:    false,
 			Description: "Whether the connector will take a snapshot of the entire table before starting cdc mode.",
 		},
-		config.Columns: {
-			Default:  "",
-			Required: false,
-			Description: "Comma-separated list of column names that should be included in each payload of the " +
-				"sdk.Record. By default includes all columns.",
-		},
 		config.BatchSize: {
 			Default:     "1000",
 			Required:    false,