Updates mysql and postgres processors to handle null strings in json columns (#3305)

Author: alishakawaguchi

internal/database-record-mapper/mysql/mapper.go

@@ -61,6 +61,8 @@ func parseMysqlRowValues(values []any, columnNames, columnDbTypes []string) (map
 		col := columnNames[i]
 		colDataType := columnDbTypes[i]
 		switch t := v.(type) {
+		case nil:
+			jObj[col] = t
 		case time.Time:
 			dt, err := neosynctypes.NewDateTimeFromMysql(t)
 			if err != nil {
@@ -69,6 +71,10 @@ func parseMysqlRowValues(values []any, columnNames, columnDbTypes []string) (map
 			jObj[col] = dt
 		case []byte:
 			if strings.EqualFold(colDataType, "json") {
+				if string(t) == "null" {
+					jObj[col] = string(t)
+					continue
+				}
 				var js any
 				if err := json.Unmarshal(t, &js); err != nil {
 					return nil, err

internal/database-record-mapper/mysql/mapper_test.go

@@ -53,20 +53,21 @@ func Test_parseMysqlRowValues(t *testing.T) {
 	})
 
 	t.Run("JSON Columns", func(t *testing.T) {
-		values := []any{[]byte(`"Hello"`), []byte(`true`), []byte(`null`), []byte(`42`), []byte(`{"items": ["book", "pen"], "count": 2, "in_stock": true}`), []byte(`[1,2,3]`)}
-		columnNames := []string{"text_col", "bool_col", "null_col", "int_col", "json_col", "array_col"}
-		cTypes := []string{"json", "json", "json", "json", "json", "json"}
+		values := []any{[]byte(`"Hello"`), []byte(`true`), []byte(`null`), []byte(`42`), []byte(`{"items": ["book", "pen"], "count": 2, "in_stock": true}`), []byte(`[1,2,3]`), nil}
+		columnNames := []string{"text_col", "bool_col", "null_col", "int_col", "json_col", "array_col", "nil_col"}
+		cTypes := []string{"json", "json", "json", "json", "json", "json", "json"}
 
 		result, err := parseMysqlRowValues(values, columnNames, cTypes)
 		require.NoError(t, err)
 
 		expected := map[string]any{
 			"text_col":  "Hello",
 			"bool_col":  true,
-			"null_col":  nil,
+			"null_col":  "null",
 			"int_col":   float64(42),
 			"json_col":  map[string]any{"items": []any{"book", "pen"}, "count": float64(2), "in_stock": true},
 			"array_col": []any{float64(1), float64(2), float64(3)},
+			"nil_col":   nil,
 		}
 		require.Equal(t, expected, result)
 	})

internal/database-record-mapper/postgres/mapper.go

@@ -109,6 +109,10 @@ func parsePgRowValues(values []any, columnNames, columnTypes []string) (map[stri
 			}
 			jObj[col] = val
 		case *NullableJSON:
+			if t.Valid && string(t.RawMessage) == "null" {
+				jObj[col] = string(t.RawMessage)
+				continue
+			}
 			js, err := t.Unmarshal()
 			if err != nil {
 				return nil, fmt.Errorf("failed to unmarshal JSON: %w", err)

internal/database-record-mapper/postgres/mapper_test.go

@@ -220,9 +220,11 @@ func Test_parsePgRowValues(t *testing.T) {
 			&NullableJSON{RawMessage: json.RawMessage(`42`), Valid: true},
 			&NullableJSON{RawMessage: json.RawMessage(`{"items": ["book", "pen"], "count": 2, "in_stock": true}`), Valid: true},
 			&NullableJSON{RawMessage: json.RawMessage(`[1,2,3]`), Valid: true},
+			nil,
+			&NullableJSON{RawMessage: json.RawMessage(`null`), Valid: true},
 		}
-		columnNames := []string{"text_col", "bool_col", "null_col", "int_col", "json_col", "array_col"}
-		columnTypes := []string{"json", "json", "json", "json", "json", "_json"}
+		columnNames := []string{"text_col", "bool_col", "null_col", "int_col", "json_col", "array_col", "nil_col", "null_json"}
+		columnTypes := []string{"json", "json", "json", "json", "json", "_json", "json", "json"}
 
 		result, err := parsePgRowValues(values, columnNames, columnTypes)
 		require.NoError(t, err)
@@ -234,6 +236,8 @@ func Test_parsePgRowValues(t *testing.T) {
 			"int_col":   float64(42),
 			"json_col":  map[string]any{"items": []any{"book", "pen"}, "count": float64(2), "in_stock": true},
 			"array_col": []any{float64(1), float64(2), float64(3)},
+			"nil_col":   nil,
+			"null_json": "null",
 		}
 		require.Equal(t, expected, result)
 	})

worker/pkg/benthos/sql/processor_neosync_mysql.go

@@ -148,6 +148,9 @@ func getMysqlValue(value any, colDefaults *neosync_benthos.ColumnDefaultProperti
 			}
 			return validJson, nil
 		}
+		if value == "null" {
+			return value, nil
+		}
 		bits, err := json.Marshal(value)
 		if err != nil {
 			return nil, fmt.Errorf("unable to marshal mysql json to bits: %w", err)

worker/pkg/benthos/sql/processor_neosync_pgx.go

@@ -145,6 +145,9 @@ func getPgxValue(value any, colDefaults *neosync_benthos.ColumnDefaultProperties
 
 	switch {
 	case strings.EqualFold(datatype, "json") || strings.EqualFold(datatype, "jsonb"):
+		if value == "null" {
+			return value, nil
+		}
 		bits, err := json.Marshal(value)
 		if err != nil {
 			return nil, fmt.Errorf("unable to marshal postgres json to bits: %w", err)