Fixes Postgres jsonb in CLI sync (#2954)

Author: alishakawaguchi

cli/internal/cmds/neosync/sync/sync.go

@@ -238,7 +238,7 @@ func newCliSyncFromCmd(
 		return nil, err
 	}
 	cmd.SilenceUsage = true
-	logger = logger.With("accountId", cmdCfg.AccountId)
+	logger = logger.With("accountId", *cmdCfg.AccountId)
 
 	logger.Info("Starting sync")
 

internal/postgres/utils.go

@@ -135,6 +135,9 @@ func IsPgArrayColumnDataType(colDataType string) bool {
 }
 
 func pgArrayToGoSlice(array *PgxArray[any]) any {
+	if array.Elements == nil {
+		return nil
+	}
 	goSlice := convertArrayToGoType(array)
 
 	dim := array.Dimensions()

worker/pkg/benthos/sql/json_processor.go

@@ -84,13 +84,12 @@ func (p *jsonToSqlProcessor) transform(path string, root any) any {
 			return v
 		}
 		if isPgArray(datatype) {
-			var pgarray []any
-			err := json.Unmarshal(v, &pgarray)
+			pgarray, err := processPgArray(v, datatype)
 			if err != nil {
-				p.logger.Errorf("unable to unmarshal pg array: %w", err)
+				p.logger.Errorf("unable to process PG Array: %w", err)
 				return v
 			}
-			return pq.Array(pgarray)
+			return pgarray
 		}
 		switch datatype {
 		case "bit":
@@ -100,7 +99,7 @@ func (p *jsonToSqlProcessor) transform(path string, root any) any {
 				return v
 			}
 			return bit
-		case "json":
+		case "json", "jsonb":
 			validJson, err := getValidJson(v)
 			if err != nil {
 				p.logger.Errorf("unable to get valid json: %w", err)
@@ -117,6 +116,24 @@ func (p *jsonToSqlProcessor) transform(path string, root any) any {
 	}
 }
 
+func processPgArray(bits []byte, datatype string) (any, error) {
+	var pgarray []any
+	err := json.Unmarshal(bits, &pgarray)
+	if err != nil {
+		return nil, err
+	}
+	switch datatype {
+	case "json[]", "jsonb[]":
+		jsonArray, err := stringifyJsonArray(pgarray)
+		if err != nil {
+			return nil, err
+		}
+		return pq.Array(jsonArray), nil
+	default:
+		return pq.Array(pgarray), nil
+	}
+}
+
 func isPgArray(datatype string) bool {
 	return strings.HasSuffix(datatype, "[]")
 }
@@ -135,6 +152,18 @@ func getValidJson(jsonData []byte) ([]byte, error) {
 	return quotedData, nil
 }
 
+func stringifyJsonArray(pgarray []any) ([]string, error) {
+	jsonArray := make([]string, len(pgarray))
+	for i, item := range pgarray {
+		bytes, err := json.Marshal(item)
+		if err != nil {
+			return nil, err
+		}
+		jsonArray[i] = string(bytes)
+	}
+	return jsonArray, nil
+}
+
 func convertStringToBit(bitString string) ([]byte, error) {
 	val, err := strconv.ParseUint(bitString, 2, len(bitString))
 	if err != nil {