SF JSON (Variant): Remove escaped quotes

flow/connectors/snowflake/qrep_avro_sync.go

@@ -329,6 +329,10 @@ func (c *SnowflakeConnector) GetCopyTransformation(
 		case "NUMBER":
 			transformations = append(transformations,
 				fmt.Sprintf("$1:\"%s\" AS %s", avroColName, normalizedColName))
+		case "VARIANT":
+			transformations = append(transformations,
+				fmt.Sprintf("PARSE_JSON($1:\"%s\") AS %s", avroColName, normalizedColName))
+
 		default:
 			transformations = append(transformations,
 				fmt.Sprintf("($1:\"%s\")::%s AS %s", avroColName, colType, normalizedColName))

flow/connectors/snowflake/snowflake.go

@@ -850,6 +850,10 @@ func (c *SnowflakeConnector) generateAndExecuteMergeStatement(
 			flattenedCastsSQLArray = append(flattenedCastsSQLArray,
 				fmt.Sprintf("TO_GEOMETRY(CAST(%s:\"%s\" AS STRING),true) AS %s,",
 					toVariantColumnName, columnName, targetColumnName))
+		case qvalue.QValueKindJSON:
+			flattenedCastsSQLArray = append(flattenedCastsSQLArray,
+				fmt.Sprintf("PARSE_JSON(CAST(%s:\"%s\" AS STRING)) AS %s,",
+					toVariantColumnName, columnName, targetColumnName))
 		// TODO: https://github.com/PeerDB-io/peerdb/issues/189 - handle time types and interval types
 		// case model.ColumnTypeTime:
 		// 	flattenedCastsSQLArray = append(flattenedCastsSQLArray, fmt.Sprintf("TIME_FROM_PARTS(0,0,0,%s:%s:"+

flow/connectors/sql/query_executor.go

@@ -410,7 +410,10 @@ func toQValue(kind qvalue.QValueKind, val interface{}) (qvalue.QValue, error) {
 
 	case qvalue.QValueKindJSON:
 		vraw := val.(*interface{})
-		vstring := (*vraw).(string)
+		vstring, ok := (*vraw).(string)
+		if !ok {
+			slog.Warn("A parsed JSON value was not a string. Likely a null field value")
+		}
 
 		if strings.HasPrefix(vstring, "[") {
 			// parse the array

flow/e2e/snowflake/peer_flow_sf_test.go

@@ -749,6 +749,11 @@ func (s PeerFlowE2ETestSuiteSF) Test_Types_SF() {
 	if err != nil {
 		s.t.Log(err)
 	}
+
+	// check if JSON on snowflake side is a good JSON
+	err = s.checkJSONValue(dstTableName, "c17", "sai", "1")
+	require.NoError(s.t, err)
+
 	// Make sure that there are no nulls
 	s.Equal(noNulls, true)
 

flow/e2e/snowflake/qrep_flow_sf_test.go

@@ -23,6 +23,21 @@ func (s PeerFlowE2ETestSuiteSF) compareTableContentsSF(tableName, selector strin
 	s.compareTableContentsWithDiffSelectorsSF(tableName, selector, selector, false)
 }
 
+func (s PeerFlowE2ETestSuiteSF) checkJSONValue(tableName, colName, fieldName, value string) error {
+	res, err := s.sfHelper.ExecuteAndProcessQuery(fmt.Sprintf(
+		"SELECT %s:%s FROM %s;",
+		colName, fieldName, tableName))
+	if err != nil {
+		return fmt.Errorf("json value check failed: %v", err)
+	}
+
+	jsonVal := res.Records[0].Entries[0].Value
+	if jsonVal != value {
+		return fmt.Errorf("bad json value in field %s of column %s: %v. expected: %v", fieldName, colName, jsonVal, value)
+	}
+	return nil
+}
+
 func (s PeerFlowE2ETestSuiteSF) compareTableContentsWithDiffSelectorsSF(tableName, pgSelector, sfSelector string,
 	tableCaseSensitive bool,
 ) {
@@ -89,6 +104,9 @@ func (s PeerFlowE2ETestSuiteSF) Test_Complete_QRep_Flow_Avro_SF() {
 	sel := e2e.GetOwnersSelectorStringsSF()
 	s.compareTableContentsWithDiffSelectorsSF(tblName, sel[0], sel[1], false)
 
+	err = s.checkJSONValue(dstSchemaQualified, "f7", "key", "\"value\"")
+	require.NoError(s.t, err)
+
 	env.AssertExpectations(s.t)
 }
 
@@ -247,7 +265,6 @@ func (s PeerFlowE2ETestSuiteSF) Test_Complete_QRep_Flow_Avro_SF_S3_Integration()
 		s.attachSchemaSuffix(tblName),
 		dstSchemaQualified,
 		query,
-
 		sfPeer,
 		"",
 		false,