-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
aee91c0
commit 0a8abaf
Showing
6 changed files
with
262 additions
and
287 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,84 +1,77 @@ | ||
| package converters | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
| "fmt" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/debezium" | ||
| "github.com/cockroachdb/apd/v3" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/debezium" | ||
| ) | ||
|
|
||
| type decimalConverter struct { | ||
| scale uint16 | ||
| precision *int | ||
| scale uint16 | ||
| precision *int | ||
| } | ||
|
|
||
| func NewDecimalConverter(scale uint16, precision *int) decimalConverter { | ||
| return decimalConverter{scale: scale, precision: precision} | ||
| return decimalConverter{scale: scale, precision: precision} | ||
| } | ||
|
|
||
| func (d decimalConverter) ToField(name string) debezium.Field { | ||
| field := debezium.Field{ | ||
| FieldName: name, | ||
| Type: debezium.Bytes, | ||
| DebeziumType: debezium.KafkaDecimalType, | ||
| Parameters: map[string]any{ | ||
| "scale": fmt.Sprint(d.scale), | ||
| }, | ||
| } | ||
|
|
||
| if d.precision != nil { | ||
| field.Parameters[debezium.KafkaDecimalPrecisionKey] = fmt.Sprint(*d.precision) | ||
| } | ||
|
|
||
| return field | ||
| field := debezium.Field{ | ||
| FieldName: name, | ||
| Type: debezium.Bytes, | ||
| DebeziumType: debezium.KafkaDecimalType, | ||
| Parameters: map[string]any{ | ||
| "scale": fmt.Sprint(d.scale), | ||
| }, | ||
| } | ||
|
|
||
| if d.precision != nil { | ||
| field.Parameters[debezium.KafkaDecimalPrecisionKey] = fmt.Sprint(*d.precision) | ||
| } | ||
|
|
||
| return field | ||
| } | ||
|
|
||
| func (d decimalConverter) Convert(value any) (any, error) { | ||
| castValue, isOk := value.(string) | ||
| if !isOk { | ||
| return nil, fmt.Errorf("expected string got %T with value: %v", value, value) | ||
| } | ||
| return debezium.EncodeDecimal(castValue, d.scale) | ||
| } | ||
|
|
||
| func getScale(value string) uint16 { | ||
| // Find the index of the decimal point | ||
| i := strings.IndexRune(value, '.') | ||
| stringValue, isOk := value.(string) | ||
| if !isOk { | ||
| return nil, fmt.Errorf("expected string got %T with value: %v", value, value) | ||
| } | ||
|
|
||
| if i == -1 { | ||
| // No decimal point: scale is 0 | ||
| return 0 | ||
| } | ||
| decimal, _, err := apd.NewFromString(stringValue) | ||
| if err != nil { | ||
| return nil, fmt.Errorf(`unable to use %q as a decimal: %w`, stringValue, err) | ||
| } | ||
|
|
||
| // The scale is the number of digits after the decimal point | ||
| return uint16(len(value[i+1:])) | ||
| return debezium.EncodeDecimalWithScale(decimal, int32(d.scale)), nil | ||
| } | ||
|
|
||
| type VariableNumericConverter struct{} | ||
|
|
||
| func (VariableNumericConverter) ToField(name string) debezium.Field { | ||
| return debezium.Field{ | ||
| FieldName: name, | ||
| Type: debezium.Struct, | ||
| DebeziumType: debezium.KafkaVariableNumericType, | ||
| } | ||
| return debezium.Field{ | ||
| FieldName: name, | ||
| Type: debezium.Struct, | ||
| DebeziumType: debezium.KafkaVariableNumericType, | ||
| } | ||
| } | ||
|
|
||
| func (VariableNumericConverter) Convert(value any) (any, error) { | ||
| stringValue, ok := value.(string) | ||
| if !ok { | ||
| return nil, fmt.Errorf("expected string got %T with value: %v", value, value) | ||
| } | ||
|
|
||
| scale := getScale(stringValue) | ||
|
|
||
| bytes, err := debezium.EncodeDecimal(stringValue, scale) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return map[string]any{ | ||
| "scale": int32(scale), | ||
| "value": bytes, | ||
| }, nil | ||
| stringValue, ok := value.(string) | ||
| if !ok { | ||
| return nil, fmt.Errorf("expected string got %T with value: %v", value, value) | ||
| } | ||
|
|
||
| decimal, _, err := apd.NewFromString(stringValue) | ||
| if err != nil { | ||
| return nil, fmt.Errorf(`unable to use %q as a decimal: %w`, stringValue, err) | ||
| } | ||
|
|
||
| bytes, scale := debezium.EncodeDecimal(decimal) | ||
| return map[string]any{ | ||
| "scale": scale, | ||
| "value": bytes, | ||
| }, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,133 +1,102 @@ | ||
| package converters | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "testing" | ||
| "fmt" | ||
| "testing" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/debezium" | ||
| "github.com/artie-labs/transfer/lib/ptr" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/artie-labs/transfer/lib/debezium" | ||
| "github.com/artie-labs/transfer/lib/ptr" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestDecimalConverter_ToField(t *testing.T) { | ||
| { | ||
| // Without precision | ||
| converter := NewDecimalConverter(2, nil) | ||
| expected := debezium.Field{ | ||
| Type: "bytes", | ||
| FieldName: "col", | ||
| DebeziumType: "org.apache.kafka.connect.data.Decimal", | ||
| Parameters: map[string]any{ | ||
| "scale": "2", | ||
| }, | ||
| } | ||
| assert.Equal(t, expected, converter.ToField("col")) | ||
| } | ||
| { | ||
| // With precision | ||
| converter := NewDecimalConverter(2, ptr.ToInt(3)) | ||
| expected := debezium.Field{ | ||
| Type: "bytes", | ||
| FieldName: "col", | ||
| DebeziumType: "org.apache.kafka.connect.data.Decimal", | ||
| Parameters: map[string]any{ | ||
| "connect.decimal.precision": "3", | ||
| "scale": "2", | ||
| }, | ||
| } | ||
| assert.Equal(t, expected, converter.ToField("col")) | ||
| } | ||
| { | ||
| // Without precision | ||
| converter := NewDecimalConverter(2, nil) | ||
| expected := debezium.Field{ | ||
| Type: "bytes", | ||
| FieldName: "col", | ||
| DebeziumType: "org.apache.kafka.connect.data.Decimal", | ||
| Parameters: map[string]any{ | ||
| "scale": "2", | ||
| }, | ||
| } | ||
| assert.Equal(t, expected, converter.ToField("col")) | ||
| } | ||
| { | ||
| // With precision | ||
| converter := NewDecimalConverter(2, ptr.ToInt(3)) | ||
| expected := debezium.Field{ | ||
| Type: "bytes", | ||
| FieldName: "col", | ||
| DebeziumType: "org.apache.kafka.connect.data.Decimal", | ||
| Parameters: map[string]any{ | ||
| "connect.decimal.precision": "3", | ||
| "scale": "2", | ||
| }, | ||
| } | ||
| assert.Equal(t, expected, converter.ToField("col")) | ||
| } | ||
| } | ||
|
|
||
| func TestDecimalConverter_Convert(t *testing.T) { | ||
| converter := NewDecimalConverter(2, nil) | ||
| { | ||
| // Malformed value - empty string. | ||
| _, err := converter.Convert("") | ||
| assert.ErrorContains(t, err, `unable to use "" as a floating-point number`) | ||
| } | ||
| { | ||
| // Malformed value - not a floating-point. | ||
| _, err := converter.Convert("11qwerty00") | ||
| assert.ErrorContains(t, err, `unable to use "11qwerty00" as a floating-point number`) | ||
| } | ||
| { | ||
| // Happy path. | ||
| converted, err := converter.Convert("1.23") | ||
| assert.NoError(t, err) | ||
| bytes, ok := converted.([]byte) | ||
| assert.True(t, ok) | ||
| actualValue, err := converter.ToField("").DecodeDecimal(bytes) | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, "1.23", fmt.Sprint(actualValue)) | ||
| } | ||
| } | ||
|
|
||
| func TestGetScale(t *testing.T) { | ||
| type _testCase struct { | ||
| name string | ||
| value string | ||
| expectedScale uint16 | ||
| } | ||
|
|
||
| testCases := []_testCase{ | ||
| { | ||
| name: "0 scale", | ||
| value: "5", | ||
| expectedScale: 0, | ||
| }, | ||
| { | ||
| name: "2 scale", | ||
| value: "9.99", | ||
| expectedScale: 2, | ||
| }, | ||
| { | ||
| name: "5 scale", | ||
| value: "9.12345", | ||
| expectedScale: 5, | ||
| }, | ||
| } | ||
|
|
||
| for _, testCase := range testCases { | ||
| actualScale := getScale(testCase.value) | ||
| assert.Equal(t, testCase.expectedScale, actualScale, testCase.name) | ||
| } | ||
| converter := NewDecimalConverter(2, nil) | ||
| { | ||
| // Malformed value - empty string. | ||
| _, err := converter.Convert("") | ||
| assert.ErrorContains(t, err, `unable to use "" as a decimal: parse mantissa:`) | ||
| } | ||
| { | ||
| // Malformed value - not a floating-point. | ||
| _, err := converter.Convert("11qwerty00") | ||
| assert.ErrorContains(t, err, `unable to use "11qwerty00" as a decimal: parse exponent:`) | ||
| } | ||
| { | ||
| // Happy path. | ||
| converted, err := converter.Convert("1.23") | ||
| assert.NoError(t, err) | ||
| bytes, ok := converted.([]byte) | ||
| assert.True(t, ok) | ||
| actualValue, err := converter.ToField("").DecodeDecimal(bytes) | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, "1.23", fmt.Sprint(actualValue)) | ||
| } | ||
| } | ||
|
|
||
| func TestVariableNumericConverter_ToField(t *testing.T) { | ||
| converter := VariableNumericConverter{} | ||
| expected := debezium.Field{ | ||
| FieldName: "col", | ||
| Type: "struct", | ||
| DebeziumType: "io.debezium.data.VariableScaleDecimal", | ||
| } | ||
| assert.Equal(t, expected, converter.ToField("col")) | ||
| converter := VariableNumericConverter{} | ||
| expected := debezium.Field{ | ||
| FieldName: "col", | ||
| Type: "struct", | ||
| DebeziumType: "io.debezium.data.VariableScaleDecimal", | ||
| } | ||
| assert.Equal(t, expected, converter.ToField("col")) | ||
| } | ||
|
|
||
| func TestVariableNumericConverter_Convert(t *testing.T) { | ||
| converter := VariableNumericConverter{} | ||
| { | ||
| // Wrong type | ||
| _, err := converter.Convert(1234) | ||
| assert.ErrorContains(t, err, "expected string got int with value: 1234") | ||
| } | ||
| { | ||
| // Malformed value - empty string. | ||
| _, err := converter.Convert("") | ||
| assert.ErrorContains(t, err, `unable to use "" as a floating-point number`) | ||
| } | ||
| { | ||
| // Malformed value - not a floating point. | ||
| _, err := converter.Convert("malformed") | ||
| assert.ErrorContains(t, err, `unable to use "malformed" as a floating-point number`) | ||
| } | ||
| { | ||
| // Happy path | ||
| converted, err := converter.Convert("12.34") | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, map[string]any{"scale": int32(2), "value": []byte{0x4, 0xd2}}, converted) | ||
| actualValue, err := converter.ToField("").DecodeDebeziumVariableDecimal(converted) | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, "12.34", actualValue.String()) | ||
| } | ||
| converter := VariableNumericConverter{} | ||
| { | ||
| // Wrong type | ||
| _, err := converter.Convert(1234) | ||
| assert.ErrorContains(t, err, "expected string got int with value: 1234") | ||
| } | ||
| { | ||
| // Malformed value - empty string. | ||
| _, err := converter.Convert("") | ||
| assert.ErrorContains(t, err, `unable to use "" as a decimal: parse mantissa:`) | ||
| } | ||
| { | ||
| // Malformed value - not a floating point. | ||
| _, err := converter.Convert("malformed") | ||
| assert.ErrorContains(t, err, `unable to use "malformed" as a decimal: parse exponent`) | ||
| } | ||
| { | ||
| // Happy path | ||
| converted, err := converter.Convert("12.34") | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, map[string]any{"scale": int32(2), "value": []byte{0x4, 0xd2}}, converted) | ||
| actualValue, err := converter.ToField("").DecodeDebeziumVariableDecimal(converted) | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, "12.34", actualValue.String()) | ||
| } | ||
| } |
Oops, something went wrong.