Use int32 for scale and precision values

clients/bigquery/dialect/dialect_test.go

@@ -102,15 +102,15 @@ func TestBigQueryDialect_KindForDataType(t *testing.T) {
 		kd, err := dialect.KindForDataType("numeric(5, 2)", "")
 		assert.NoError(t, err)
 		assert.Equal(t, typing.EDecimal.Kind, kd.Kind)
-		assert.Equal(t, 5, *kd.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, kd.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(5), *kd.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), kd.ExtendedDecimalDetails.Scale())
 	}
 	{
 		kd, err := dialect.KindForDataType("bignumeric(5, 2)", "")
 		assert.NoError(t, err)
 		assert.Equal(t, typing.EDecimal.Kind, kd.Kind)
-		assert.Equal(t, 5, *kd.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, kd.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(5), *kd.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), kd.ExtendedDecimalDetails.Scale())
 	}
 }
 

clients/mssql/dialect/dialect_test.go

@@ -85,8 +85,8 @@ func TestMSSQLDialect_KindForDataType(t *testing.T) {
 		kd, err := dialect.KindForDataType("numeric(5, 2)", "")
 		assert.NoError(t, err)
 		assert.Equal(t, typing.EDecimal.Kind, kd.Kind)
-		assert.Equal(t, 5, *kd.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, kd.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(5), *kd.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), kd.ExtendedDecimalDetails.Scale())
 	}
 	{
 		kd, err := dialect.KindForDataType("char", "5")

clients/redshift/dialect/dialect_test.go

@@ -145,8 +145,8 @@ func TestRedshiftDialect_KindForDataType(t *testing.T) {
 		kd, err := dialect.KindForDataType("numeric(5,2)", "")
 		assert.NoError(t, err)
 		assert.Equal(t, typing.EDecimal.Kind, kd.Kind)
-		assert.Equal(t, 5, *kd.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, kd.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(5), *kd.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), kd.ExtendedDecimalDetails.Scale())
 	}
 }
 

clients/snowflake/dialect/dialect_test.go

@@ -84,15 +84,15 @@ func TestSnowflakeDialect_KindForDataType_Floats(t *testing.T) {
 		kd, err := SnowflakeDialect{}.KindForDataType("NUMERIC(38, 2)", "")
 		assert.NoError(t, err)
 		assert.Equal(t, typing.EDecimal.Kind, kd.Kind)
-		assert.Equal(t, 38, *kd.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, kd.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(38), *kd.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), kd.ExtendedDecimalDetails.Scale())
 	}
 	{
 		kd, err := SnowflakeDialect{}.KindForDataType("NUMBER(38, 2)", "")
 		assert.NoError(t, err)
 		assert.Equal(t, typing.EDecimal.Kind, kd.Kind)
-		assert.Equal(t, 38, *kd.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, kd.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(38), *kd.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), kd.ExtendedDecimalDetails.Scale())
 	}
 	{
 		kd, err := SnowflakeDialect{}.KindForDataType("DECIMAL", "")

lib/cdc/util/optional_schema_test.go

@@ -67,35 +67,35 @@ func TestGetOptionalSchema(t *testing.T) {
 				"bit_test":     typing.Boolean,
 				"numeric_test": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(decimal.PrecisionNotSpecified), decimal.DefaultScale),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(decimal.PrecisionNotSpecified), decimal.DefaultScale),
 				},
 				"numeric_5": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(5), 0),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(5), 0),
 				},
 				"numeric_5_2": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(5), 2),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(5), 2),
 				},
 				"numeric_5_6": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(5), 6),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(5), 6),
 				},
 				"numeric_5_0": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(5), 0),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(5), 0),
 				},
 				"numeric_39_0": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(39), 0),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(39), 0),
 				},
 				"numeric_39_2": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(39), 2),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(39), 2),
 				},
 				"numeric_39_6": {
 					Kind:                   typing.EDecimal.Kind,
-					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt(39), 6),
+					ExtendedDecimalDetails: decimal.NewDecimalDetails(ptr.ToInt32(39), 6),
 				},
 			},
 		},

lib/debezium/schema.go

@@ -61,20 +61,20 @@ type Field struct {
 	Parameters   map[string]any        `json:"parameters"`
 }
 
-func (f Field) GetScaleAndPrecision() (int, *int, error) {
-	scale, scaleErr := maputil.GetIntegerFromMap(f.Parameters, "scale")
+func (f Field) GetScaleAndPrecision() (int32, *int32, error) {
+	scale, scaleErr := maputil.GetInt32FromMap(f.Parameters, "scale")
 	if scaleErr != nil {
 		return 0, nil, scaleErr
 	}
 
-	var precisionPtr *int
+	var precisionPtr *int32
 	if _, isOk := f.Parameters[KafkaDecimalPrecisionKey]; isOk {
-		precision, precisionErr := maputil.GetIntegerFromMap(f.Parameters, KafkaDecimalPrecisionKey)
+		precision, precisionErr := maputil.GetInt32FromMap(f.Parameters, KafkaDecimalPrecisionKey)
 		if precisionErr != nil {
 			return 0, nil, precisionErr
 		}
 
-		precisionPtr = ptr.ToInt(precision)
+		precisionPtr = ptr.ToInt32(precision)
 	}
 
 	return scale, precisionPtr, nil
@@ -106,7 +106,7 @@ func (f Field) ToKindDetails() typing.KindDetails {
 		// This is because scale is not specified at the column level, rather at the row level
 		// It shouldn't matter much anyway since the column type we are creating is `TEXT` to avoid boundary errors.
 		eDecimal := typing.EDecimal
-		eDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt(decimal.PrecisionNotSpecified), decimal.DefaultScale)
+		eDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt32(decimal.PrecisionNotSpecified), decimal.DefaultScale)
 		return eDecimal
 	}
 

lib/debezium/schema_test.go

@@ -19,8 +19,8 @@ func TestField_GetScaleAndPrecision(t *testing.T) {
 		name              string
 		parameters        map[string]any
 		expectedErr       string
-		expectedScale     int
-		expectedPrecision *int
+		expectedScale     int32
+		expectedPrecision *int32
 	}{
 		{
 			name:        "Test Case 1: Empty Parameters",
@@ -41,7 +41,7 @@ func TestField_GetScaleAndPrecision(t *testing.T) {
 				KafkaDecimalPrecisionKey: 10,
 			},
 			expectedScale:     5,
-			expectedPrecision: ptr.ToInt(10),
+			expectedPrecision: ptr.ToInt32(10),
 		},
 		{
 			name: "Test Case 4: Invalid Scale Type",
@@ -89,10 +89,10 @@ func TestField_ToKindDetails(t *testing.T) {
 	}
 
 	eDecimal := typing.EDecimal
-	eDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt(decimal.PrecisionNotSpecified), decimal.DefaultScale)
+	eDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt32(decimal.PrecisionNotSpecified), decimal.DefaultScale)
 
 	kafkaDecimalType := typing.EDecimal
-	kafkaDecimalType.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt(10), 5)
+	kafkaDecimalType.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt32(10), 5)
 
 	tcs := []_tc{
 		{

lib/debezium/types.go

@@ -236,7 +236,7 @@ func (f Field) DecodeDecimal(encoded []byte) (*decimal.Decimal, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to get scale and/or precision: %w", err)
 	}
-	_decimal := DecodeDecimal(encoded, int32(scale))
+	_decimal := DecodeDecimal(encoded, scale)
 	return decimal.NewDecimal(precision, _decimal), nil
 }
 
@@ -246,7 +246,7 @@ func (f Field) DecodeDebeziumVariableDecimal(value any) (*decimal.Decimal, error
 		return nil, fmt.Errorf("value is not map[string]any type")
 	}
 
-	scale, err := maputil.GetIntegerFromMap(valueStruct, "scale")
+	scale, err := maputil.GetInt32FromMap(valueStruct, "scale")
 	if err != nil {
 		return nil, err
 	}
@@ -260,6 +260,6 @@ func (f Field) DecodeDebeziumVariableDecimal(value any) (*decimal.Decimal, error
 	if err != nil {
 		return nil, err
 	}
-	_decimal := DecodeDecimal(bytes, int32(scale))
-	return decimal.NewDecimal(ptr.ToInt(decimal.PrecisionNotSpecified), _decimal), nil
+	_decimal := DecodeDecimal(bytes, scale)
+	return decimal.NewDecimal(ptr.ToInt32(decimal.PrecisionNotSpecified), _decimal), nil
 }

lib/debezium/types_test.go

@@ -447,9 +447,9 @@ func TestField_DecodeDecimal(t *testing.T) {
 		params  map[string]any
 
 		expectedValue         string
-		expectedPrecision     int
+		expectedPrecision     int32
 		expectNilPtrPrecision bool
-		expectedScale         int
+		expectedScale         int32
 		expectBigFloat        bool
 		expectedErr           string
 	}{
@@ -638,7 +638,7 @@ func TestField_DecodeDebeziumVariableDecimal(t *testing.T) {
 		value any
 
 		expectedValue string
-		expectedScale int
+		expectedScale int32
 		expectedErr   string
 	}
 
@@ -720,7 +720,7 @@ func TestField_DecodeDebeziumVariableDecimal(t *testing.T) {
 		// It should be a string instead.
 		_, isOk = dec.Value().(string)
 		assert.True(t, isOk, testCase.name)
-		assert.Equal(t, -1, *dec.Precision(), testCase.name)
+		assert.Equal(t, int32(-1), *dec.Precision(), testCase.name)
 		assert.Equal(t, testCase.expectedScale, dec.Scale(), testCase.name)
 		assert.Equal(t, testCase.expectedValue, dec.Value(), testCase.name)
 	}

lib/maputil/map.go

@@ -18,7 +18,7 @@ func GetKeyFromMap(obj map[string]any, key string, defaultValue any) any {
 	return val
 }
 
-func GetIntegerFromMap(obj map[string]any, key string) (int, error) {
+func GetInt32FromMap(obj map[string]any, key string) (int32, error) {
 	if len(obj) == 0 {
 		return 0, fmt.Errorf("object is empty")
 	}
@@ -28,10 +28,10 @@ func GetIntegerFromMap(obj map[string]any, key string) (int, error) {
 		return 0, fmt.Errorf("key: %s does not exist in object", key)
 	}
 
-	val, err := strconv.Atoi(fmt.Sprint(valInterface))
+	val, err := strconv.ParseInt(fmt.Sprint(valInterface), 10, 32)
 	if err != nil {
 		return 0, fmt.Errorf("key: %s is not type integer: %w", key, err)
 	}
 
-	return val, nil
+	return int32(val), nil
 }

lib/maputil/map_test.go

@@ -1,6 +1,7 @@
 package maputil
 
 import (
+	"math"
 	"reflect"
 	"testing"
 
@@ -27,21 +28,23 @@ func TestGetKeyFromMap(t *testing.T) {
 	assert.Equal(t, val, "robin55")
 }
 
-func TestGetIntegerFromMap(t *testing.T) {
+func TestGetInt32FromMap(t *testing.T) {
 	object := map[string]any{
 		"abc":          "123",
 		"abc (number)": 123,
 		"def":          true,
 		"ghi":          "hello",
 		"123":          "-321",
 		"123 (number)": -321,
+		"maxInt32":     math.MaxInt32,
+		"int64":        math.MaxInt32 + 1,
 	}
 
 	testCases := []struct {
 		name          string
 		obj           map[string]any
 		key           string
-		expectedValue int
+		expectedValue int32
 		expectedErr   string
 	}{
 		{
@@ -86,14 +89,26 @@ func TestGetIntegerFromMap(t *testing.T) {
 			key:           "123 (number)",
 			expectedValue: -321,
 		},
+		{
+			name:          "max int32",
+			obj:           object,
+			key:           "maxInt32",
+			expectedValue: int32(math.MaxInt32),
+		},
+		{
+			name:        "int64",
+			obj:         object,
+			key:         "int64",
+			expectedErr: "value out of range",
+		},
 	}
 
 	for _, testCase := range testCases {
-		value, err := GetIntegerFromMap(testCase.obj, testCase.key)
+		value, err := GetInt32FromMap(testCase.obj, testCase.key)
 		if testCase.expectedErr != "" {
 			assert.ErrorContains(t, err, testCase.expectedErr, testCase.name)
 		} else {
-			assert.Equal(t, reflect.Int, reflect.TypeOf(value).Kind())
+			assert.Equal(t, reflect.Int32, reflect.TypeOf(value).Kind())
 			assert.Equal(t, testCase.expectedValue, value)
 			assert.NoError(t, err, testCase.name)
 		}

lib/optimization/event_update_test.go

@@ -42,7 +42,7 @@ func TestTableData_UpdateInMemoryColumnsFromDestination(t *testing.T) {
 		tableDataCols.AddColumn(columns.NewColumn("ext_dec", typing.String))
 
 		extDecimalType := typing.EDecimal
-		extDecimalType.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt(22), 2)
+		extDecimalType.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt32(22), 2)
 		tableDataCols.AddColumn(columns.NewColumn("ext_dec_filled", extDecimalType))
 
 		tableDataCols.AddColumn(columns.NewColumn(strCol, typing.String))
@@ -121,31 +121,31 @@ func TestTableData_UpdateInMemoryColumnsFromDestination(t *testing.T) {
 		assert.Equal(t, typing.String, extDecCol.KindDetails)
 
 		extDecimal := typing.EDecimal
-		extDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt(30), 2)
+		extDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt32(30), 2)
 		assert.NoError(t, tableData.MergeColumnsFromDestination(columns.NewColumn("ext_dec", extDecimal)))
 		// Now it should be ext decimal type
 		extDecCol, isOk = tableData.inMemoryColumns.GetColumn("ext_dec")
 		assert.True(t, isOk)
 		assert.Equal(t, typing.EDecimal.Kind, extDecCol.KindDetails.Kind)
 		// Check precision and scale too.
-		assert.Equal(t, 30, *extDecCol.KindDetails.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, extDecCol.KindDetails.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(30), *extDecCol.KindDetails.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), extDecCol.KindDetails.ExtendedDecimalDetails.Scale())
 
 		// Testing ext_dec_filled since it's already filled out
 		extDecColFilled, isOk := tableData.inMemoryColumns.GetColumn("ext_dec_filled")
 		assert.True(t, isOk)
 		assert.Equal(t, typing.EDecimal.Kind, extDecColFilled.KindDetails.Kind)
 		// Check precision and scale too.
-		assert.Equal(t, 22, *extDecColFilled.KindDetails.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, extDecColFilled.KindDetails.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(22), *extDecColFilled.KindDetails.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), extDecColFilled.KindDetails.ExtendedDecimalDetails.Scale())
 
 		assert.NoError(t, tableData.MergeColumnsFromDestination(columns.NewColumn("ext_dec_filled", extDecimal)))
 		extDecColFilled, isOk = tableData.inMemoryColumns.GetColumn("ext_dec_filled")
 		assert.True(t, isOk)
 		assert.Equal(t, typing.EDecimal.Kind, extDecColFilled.KindDetails.Kind)
 		// Check precision and scale too.
-		assert.Equal(t, 22, *extDecColFilled.KindDetails.ExtendedDecimalDetails.Precision())
-		assert.Equal(t, 2, extDecColFilled.KindDetails.ExtendedDecimalDetails.Scale())
+		assert.Equal(t, int32(22), *extDecColFilled.KindDetails.ExtendedDecimalDetails.Precision())
+		assert.Equal(t, int32(2), extDecColFilled.KindDetails.ExtendedDecimalDetails.Scale())
 	}
 	{
 		tableDataCols := &columns.Columns{}

lib/parquetutil/parse_values_test.go

@@ -15,7 +15,7 @@ import (
 
 func TestParseValue(t *testing.T) {
 	eDecimal := typing.EDecimal
-	eDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt(30), 5)
+	eDecimal.ExtendedDecimalDetails = decimal.NewDecimalDetails(ptr.ToInt32(30), 5)
 
 	eTime := typing.ETime
 	eTime.ExtendedTimeDetails = &ext.Time
@@ -66,7 +66,7 @@ func TestParseValue(t *testing.T) {
 		},
 		{
 			name:          "decimal",
-			colVal:        decimal.NewDecimal(ptr.ToInt(30), numbers.MustParseDecimal("5000.22320")),
+			colVal:        decimal.NewDecimal(ptr.ToInt32(30), numbers.MustParseDecimal("5000.22320")),
 			colKind:       columns.NewColumn("", eDecimal),
 			expectedValue: "5000.22320",
 		},

lib/ptr/ptr.go

@@ -10,6 +10,10 @@ func ToInt(val int) *int {
 	return &val
 }
 
+func ToInt32(val int32) *int32 {
+	return &val
+}
+
 func ToInt64(val int64) *int64 {
 	return &val
 }