feat: add binary to decimal and binary to hex transformations; update test cases and configuration

Author: flarco

core/dbio/database/database_bigquery.go

@@ -601,6 +601,7 @@ func (conn *BigQueryConn) importViaLocalStorage(tableFName string, df *iop.Dataf
 
 	go func() {
 		config := iop.LoaderStreamConfig(true)
+		config.TargetType = conn.GetType()
 		_, err = fs.WriteDataflowReady(df, localPath, fileReadyChn, config)
 
 		if err != nil {
@@ -692,6 +693,7 @@ func (conn *BigQueryConn) importViaGoogleStorage(tableFName string, df *iop.Data
 
 	go func() {
 		config := iop.LoaderStreamConfig(true)
+		config.TargetType = conn.GetType()
 		_, err = fs.WriteDataflowReady(df, gcsPath, fileReadyChn, config)
 
 		if err != nil {

core/dbio/database/database_snowflake.go

@@ -785,6 +785,7 @@ func (conn *SnowflakeConn) CopyViaStage(table Table, df *iop.Dataflow) (count ui
 		}
 
 		config := iop.LoaderStreamConfig(true)
+		config.TargetType = conn.GetType()
 		_, err = fs.WriteDataflowReady(df, folderPath, fileReadyChn, config)
 
 		if err != nil {

core/dbio/iop/stream_processor.go

@@ -1028,6 +1028,8 @@ func (sp *StreamProcessor) CastToString(i int, val interface{}, valType ...Colum
 			return tVal.Format("2006-01-02 15:04:05.999999999") + " +00"
 		}
 		return tVal.Format("2006-01-02 15:04:05.999999999 -07")
+	case typ.IsBinary() && g.In(sp.Config.TargetType, dbio.TypeDbSnowflake, dbio.TypeDbBigQuery):
+		return Transforms.BinaryToHex(cast.ToString(val))
 	default:
 		strVal := cast.ToString(val)
 		if !utf8.ValidString(strVal) {

core/dbio/iop/transforms.go

@@ -44,6 +44,8 @@ func init() {
 		TransformHashSha256,
 		TransformHashSha512,
 		TransformParseBit,
+		TransformBinaryToDecimal,
+		TransformBinaryToHex,
 		TransformParseFix,
 		TransformParseUuid,
 		TransformParseMsUuid,
@@ -250,6 +252,20 @@ var (
 		},
 	}
 
+	TransformBinaryToDecimal = Transform{
+		Name: "binary_to_decimal",
+		FuncString: func(sp *StreamProcessor, val string) (string, error) {
+			return Transforms.BinaryToDecimal(sp, val)
+		},
+	}
+
+	TransformBinaryToHex = Transform{
+		Name: "binary_to_hex",
+		FuncString: func(sp *StreamProcessor, val string) (string, error) {
+			return Transforms.BinaryToHex(val), nil
+		},
+	}
+
 	TransformParseFix = Transform{
 		Name: "parse_fix",
 		FuncString: func(sp *StreamProcessor, val string) (string, error) {
@@ -494,6 +510,51 @@ func (t transformsNS) ParseBit(sp *StreamProcessor, val string) (string, error)
 	return val, nil
 }
 
+func (t transformsNS) BinaryToDecimal(sp *StreamProcessor, val string) (string, error) {
+	// Handle MySQL BIT type which can be 1 to 64 bits (1 to 8 bytes)
+	// Convert binary data to decimal representation for better compatibility
+	if len(val) > 0 && len(val) <= 8 {
+		// Check if it's binary data (all bytes are either printable or control chars)
+		isBinary := true
+		for _, b := range []byte(val) {
+			// If we have high bit values or control characters, treat as binary
+			if b > 127 || (b < 32 && b != 9 && b != 10 && b != 13) {
+				isBinary = true
+				break
+			}
+			// If we have regular ASCII text, don't treat as binary
+			if b >= 32 && b <= 126 {
+				isBinary = false
+			}
+		}
+
+		if isBinary {
+			// Convert binary data to uint64 (big-endian)
+			var result uint64
+			for i, b := range []byte(val) {
+				result |= uint64(b) << (8 * (len(val) - 1 - i))
+			}
+			return fmt.Sprintf("%d", result), nil
+		}
+	}
+	return val, nil
+}
+
+func (t transformsNS) BinaryToHex(val string) string {
+	// Convert binary data to hexadecimal representation for Snowflake COPY
+	if len(val) == 0 {
+		return ""
+	}
+
+	// Convert each byte to hex and concatenate
+	hexStr := ""
+	for _, b := range []byte(val) {
+		hexStr += fmt.Sprintf("%02X", b)
+	}
+
+	return hexStr
+}
+
 func (t transformsNS) Replace0x00(sp *StreamProcessor, val string) (string, error) {
 	return strings.ReplaceAll(strings.ReplaceAll(val, "\x00", ""), "\\u0000", "u-0000"), nil // replace the NUL character
 }

core/dbio/iop/transforms_test.go

@@ -68,3 +68,135 @@ func TestTransformMsUUID(t *testing.T) {
 	val, _ := Transforms.ParseMsUUID(sp, cast.ToString(uuidBytes))
 	assert.Equal(t, "12345678-1234-1234-1234-123456789abc", val)
 }
+
+func TestBinaryToDecimal(t *testing.T) {
+	sp := NewStreamProcessor()
+
+	// Test cases for various BIT sizes
+	testCases := []struct {
+		name     string
+		input    []byte
+		expected string
+	}{
+		{
+			name:     "BIT(1) - 0",
+			input:    []byte{0x00},
+			expected: "0",
+		},
+		{
+			name:     "BIT(1) - 1",
+			input:    []byte{0x01},
+			expected: "1",
+		},
+		{
+			name:     "BIT(8) - 255",
+			input:    []byte{0xFF},
+			expected: "255",
+		},
+		{
+			name:     "BIT(16) - 65535",
+			input:    []byte{0xFF, 0xFF},
+			expected: "65535",
+		},
+		{
+			name:     "BIT(24) - 16777215",
+			input:    []byte{0xFF, 0xFF, 0xFF},
+			expected: "16777215",
+		},
+		{
+			name:     "BIT(32) - 4294967295",
+			input:    []byte{0xFF, 0xFF, 0xFF, 0xFF},
+			expected: "4294967295",
+		},
+		{
+			name:     "BIT(64) - max value",
+			input:    []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF},
+			expected: "18446744073709551615",
+		},
+		{
+			name:     "BIT(8) - binary 10101010",
+			input:    []byte{0xAA}, // binary 10101010
+			expected: "170",
+		},
+		{
+			name:     "BIT(16) - binary pattern",
+			input:    []byte{0x12, 0x34}, // 0x1234 = 4660
+			expected: "4660",
+		},
+		{
+			name:     "Regular text should not be converted",
+			input:    []byte("hello"),
+			expected: "hello",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			val, err := Transforms.BinaryToDecimal(sp, string(tc.input))
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expected, val, "Failed for test case: %s", tc.name)
+		})
+	}
+}
+
+func TestBinaryToHex(t *testing.T) {
+	// Test cases for ToHex transform
+	testCases := []struct {
+		name     string
+		input    []byte
+		expected string
+	}{
+		{
+			name:     "Empty input",
+			input:    []byte{},
+			expected: "",
+		},
+		{
+			name:     "Single byte - 0x00",
+			input:    []byte{0x00},
+			expected: "00",
+		},
+		{
+			name:     "Single byte - 0x01",
+			input:    []byte{0x01},
+			expected: "01",
+		},
+		{
+			name:     "Single byte - 0xFF",
+			input:    []byte{0xFF},
+			expected: "FF",
+		},
+		{
+			name:     "Two bytes - 0x1234",
+			input:    []byte{0x12, 0x34},
+			expected: "1234",
+		},
+		{
+			name:     "Four bytes - 0xDEADBEEF",
+			input:    []byte{0xDE, 0xAD, 0xBE, 0xEF},
+			expected: "DEADBEEF",
+		},
+		{
+			name:     "Text - Hello",
+			input:    []byte("Hello"),
+			expected: "48656C6C6F",
+		},
+		{
+			name:     "Eight bytes - all 0xFF",
+			input:    []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF},
+			expected: "FFFFFFFFFFFFFFFF",
+		},
+		{
+			name:     "Alternating pattern",
+			input:    []byte{0xAA, 0x55, 0xAA, 0x55},
+			expected: "AA55AA55",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := Transforms.BinaryToHex(string(tc.input))
+			assert.Equal(t, tc.expected, result, "Failed for test case: %s", tc.name)
+		})
+	}
+}

core/dbio/scripts/test.sh

@@ -8,7 +8,7 @@ go test -v -run 'TestConnection'
 cd -
 
 cd iop
-go test -timeout 5m -v -run 'TestParseDate|TestDetectDelimiter|TestFIX|TestConstraints|TestDuckDb|TestParquetDuckDb|TestIcebergReader|TestDeltaReader|TestPartition|TestExtractPartitionTimeValue|TestGetLowestPartTimeUnit|TestMatchedPartitionMask|TestGeneratePartURIsFromRange|TestDataset|TestValidateNames|TestExcelDateToTime'
+go test -timeout 5m -v -run 'TestParseDate|TestDetectDelimiter|TestFIX|TestConstraints|TestDuckDb|TestParquetDuckDb|TestIcebergReader|TestDeltaReader|TestPartition|TestExtractPartitionTimeValue|TestGetLowestPartTimeUnit|TestMatchedPartitionMask|TestGeneratePartURIsFromRange|TestDataset|TestValidateNames|TestExcelDateToTime|TestBinaryToHex|TestBinaryToDecimal'
 cd -
 
 cd database

core/dbio/templates/types_native_to_general.tsv

@@ -155,7 +155,7 @@ duckdb	hugeint	bigint	col_hugeint hugeint	TRUE	TRUE
 duckdb	interval	string	col_interval interval	TRUE	TRUE	
 mariadb	bigint	bigint	col_bigint bigint	TRUE	TRUE	A large integer
 mariadb	binary	binary	col_binary binary(100)	TRUE	TRUE	A fixed-length binary string
-mariadb	bit	smallint	col_bit bit	FALSE	FALSE	A bit field
+mariadb	bit	binary	col_bit bit	FALSE	FALSE	A bit field
 mariadb	blob	text	col_blob blob	TRUE	TRUE	A small BLOB
 mariadb	char	string	col_char char(17)	TRUE	TRUE	A fixed-length nonbinary (character) string
 mariadb	date	date	col_date date	TRUE	TRUE	A date value in CCYY-MM-DD format
@@ -227,7 +227,7 @@ motherduck	hugeint	bigint	col_hugeint hugeint	TRUE	TRUE
 motherduck	interval	string	col_interval interval	TRUE	TRUE	
 mysql	bigint	bigint	col_bigint bigint	TRUE	TRUE	A large integer
 mysql	binary	binary	col_binary binary(100)	TRUE	TRUE	A fixed-length binary string
-mysql	bit	smallint	col_bit bit	FALSE	FALSE	A bit field
+mysql	bit	binary	col_bit bit	FALSE	FALSE	A bit field
 mysql	blob	text	col_blob blob	TRUE	TRUE	A small BLOB
 mysql	char	string	col_char char(17)	TRUE	TRUE	A fixed-length nonbinary (character) string
 mysql	date	date	col_date date	TRUE	TRUE	A date value in CCYY-MM-DD format

core/sling/config.go

@@ -136,17 +136,8 @@ func (cfg *Config) SetDefault() {
 		}
 	}
 
-	// set default transforms
-	switch cfg.SrcConn.Type {
-	case dbio.TypeDbMySQL, dbio.TypeDbMariaDB, dbio.TypeDbStarRocks:
-		// parse_bit for MySQL
-		cfg.extraTransforms = append(cfg.extraTransforms, "parse_bit")
-	}
-
 	// set default metadata
 	switch {
-	case g.In(cfg.TgtConn.Type, dbio.TypeDbStarRocks):
-		cfg.extraTransforms = append(cfg.extraTransforms, "parse_bit")
 	case g.In(cfg.TgtConn.Type, dbio.TypeDbBigQuery):
 		cfg.Target.Options.DatetimeFormat = "2006-01-02 15:04:05.000000-07"
 	}