diff --git a/clients/shared/merge.go b/clients/shared/merge.go index 2aadf03be..2d4fb20d2 100644 --- a/clients/shared/merge.go +++ b/clients/shared/merge.go @@ -151,8 +151,6 @@ func Merge(ctx context.Context, dwh destination.DataWarehouse, tableData *optimi return fmt.Errorf("failed to generate merge statements: %w", err) } - fmt.Println("mergeStatements", mergeStatements) - if err = destination.ExecStatements(dwh, mergeStatements); err != nil { return fmt.Errorf("failed to execute merge statements: %w", err) } diff --git a/lib/debezium/converters/basic.go b/lib/debezium/converters/basic.go index 60e4ccb97..ff4a9de3f 100644 --- a/lib/debezium/converters/basic.go +++ b/lib/debezium/converters/basic.go @@ -87,3 +87,17 @@ func (Float64) Convert(value any) (any, error) { return nil, fmt.Errorf("unexpected type %T", value) } } + +type Array struct{} + +func (Array) ToKindDetails() typing.KindDetails { + return typing.Array +} + +func (Array) Convert(value any) (any, error) { + if fmt.Sprint(value) == fmt.Sprintf("[%s]", constants.ToastUnavailableValuePlaceholder) { + return constants.ToastUnavailableValuePlaceholder, nil + } + + return value, nil +} diff --git a/lib/debezium/converters/basic_test.go b/lib/debezium/converters/basic_test.go index 95d5d130e..6747b7e59 100644 --- a/lib/debezium/converters/basic_test.go +++ b/lib/debezium/converters/basic_test.go @@ -3,6 +3,8 @@ package converters import ( "testing" + "github.com/artie-labs/transfer/lib/config/constants" + "github.com/stretchr/testify/assert" ) @@ -85,3 +87,27 @@ func TestFloat64_Convert(t *testing.T) { } } } + +func TestArray_Convert(t *testing.T) { + { + // Irrelevant data type + value, err := Array{}.Convert([]int{1, 2, 3, 4}) + assert.NoError(t, err) + assert.Equal(t, []int{1, 2, 3, 4}, value) + } + { + // TOASTED data + { + // As []any + value, err := Array{}.Convert([]any{constants.ToastUnavailableValuePlaceholder}) + assert.NoError(t, err) + assert.Equal(t, constants.ToastUnavailableValuePlaceholder, value) + } + { + // As []string + value, err := Array{}.Convert([]string{constants.ToastUnavailableValuePlaceholder}) + assert.NoError(t, err) + assert.Equal(t, constants.ToastUnavailableValuePlaceholder, value) + } + } +} diff --git a/lib/debezium/schema.go b/lib/debezium/schema.go index 5702cb753..539a5bf5b 100644 --- a/lib/debezium/schema.go +++ b/lib/debezium/schema.go @@ -137,6 +137,8 @@ func (f Field) ToValueConverter() (converters.ValueConverter, error) { } switch f.Type { + case Array: + return converters.Array{}, nil case Double, Float: return converters.Float64{}, nil } @@ -167,8 +169,6 @@ func (f Field) ToKindDetails() (typing.KindDetails, error) { return typing.Struct, nil case Boolean: return typing.Boolean, nil - case Array: - return typing.Array, nil default: return typing.Invalid, fmt.Errorf("unhandled field type %q", f.Type) } diff --git a/models/event/event.go b/models/event/event.go index a56125040..9dc1ba7b6 100644 --- a/models/event/event.go +++ b/models/event/event.go @@ -224,22 +224,8 @@ func (e *Event) Save(cfg config.Config, inMemDB *models.DatabaseData, tc kafkali toastedCol = true } } - - valArray, isOk := val.([]any) - if isOk { - if len(valArray) == 1 { - if _, isOk = valArray[0].(string); isOk { - if valArray[0] == constants.ToastUnavailableValuePlaceholder { - val = constants.ToastUnavailableValuePlaceholder - toastedCol = true - } - } - } - } } - fmt.Println("toastedCol", toastedCol, "val", val) - if toastedCol { err := inMemoryColumns.UpsertColumn(newColName, columns.UpsertColumnArg{ ToastCol: typing.ToPtr(true),