Merge pull request #75 from dev-menon/fix/PS-json-empty-string-fix

Fix(Fivetran): Json conversion empty string to null

Author: nickvanw

cmd/internal/server/handlers/fivetran_value_converters.go

@@ -212,8 +212,20 @@ var converters = map[fivetransdk.DataType]ConverterFunc{
 				Inner: &fivetransdk.ValueType_Json{Json: CleanStringValue(string(geoJson))},
 			}, nil
 		}
+
+		// Empty strings are not valid JSON per RFC 8259. PlanetScale databases
+		// (MySQL-compatible) allow empty strings in JSON columns, but destinations
+		// like BigQuery reject them with "attempting to parse an empty input" errors.
+		// Convert empty strings to NULL for destination compatibility.
+		jsonString := CleanStringValue(value.ToString())
+		if jsonString == "" {
+			return &fivetransdk.ValueType{
+				Inner: &fivetransdk.ValueType_Null{},
+			}, nil
+		}
+
 		return &fivetransdk.ValueType{
-			Inner: &fivetransdk.ValueType_Json{Json: CleanStringValue(value.ToString())},
+			Inner: &fivetransdk.ValueType_Json{Json: jsonString},
 		}, nil
 	},
 }

cmd/internal/server/handlers/fivetran_value_converters_test.go

@@ -0,0 +1,39 @@
+package handlers
+
+import (
+	"testing"
+
+	fivetransdk "github.com/planetscale/fivetran-source/fivetran_sdk.v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"vitess.io/vitess/go/sqltypes"
+)
+
+func TestJSONConverter_EmptyString(t *testing.T) {
+	converter, err := GetConverter(fivetransdk.DataType_JSON)
+	require.NoError(t, err)
+
+	// Empty string should convert to NULL for BigQuery compatibility
+	emptyValue := sqltypes.NewVarChar("")
+	result, err := converter(emptyValue)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	_, ok := result.Inner.(*fivetransdk.ValueType_Null)
+	assert.True(t, ok, "empty JSON string should convert to NULL")
+}
+
+func TestJSONConverter_ValidJSON(t *testing.T) {
+	converter, err := GetConverter(fivetransdk.DataType_JSON)
+	require.NoError(t, err)
+
+	// Valid JSON should pass through
+	jsonValue := sqltypes.NewVarChar(`{"key": "value"}`)
+	result, err := converter(jsonValue)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	json, ok := result.Inner.(*fivetransdk.ValueType_Json)
+	require.True(t, ok)
+	assert.Equal(t, `{"key": "value"}`, json.Json)
+}