avoid converting json null to sql null (#3593)

Author: serprex

flow/connectors/postgres/cdc.go

@@ -323,19 +323,23 @@ func (p *PostgresCDCSource) decodeColumnData(
 				p.logger.Error("[pg_cdc] failed to unmarshal json", slog.Any("error", err))
 				return nil, fmt.Errorf("failed to unmarshal json: %w", err)
 			}
+			if parsedData == nil {
+				// avoid confusing SQL null & JSON null by using struct{} as json null sentinel
+				parsedData = model.JsonNull{}
+			}
 			return p.parseFieldFromPostgresOID(dataType, typmod, true, protos.DBType_DBTYPE_UNKNOWN,
 				parsedData, customTypeMapping, p.internalVersion)
 		}
 		return types.QValueNull(types.QValueKindJSON), nil
 	} else if dataType == pgtype.JSONArrayOID || dataType == pgtype.JSONBArrayOID {
-		textArr := &pgtype.FlatArray[pgtype.Text]{}
-		if err := p.typeMap.Scan(dataType, formatCode, data, textArr); err != nil {
+		textArr := pgtype.FlatArray[pgtype.Text]{}
+		if err := p.typeMap.Scan(dataType, formatCode, data, &textArr); err != nil {
 			p.logger.Error("[pg_cdc] failed to scan json array", slog.Any("error", err))
 			return nil, fmt.Errorf("failed to scan json array: %w", err)
 		}
 
-		arr := make([]any, len(*textArr))
-		for j, text := range *textArr {
+		arr := make([]any, len(textArr))
+		for j, text := range textArr {
 			if text.Valid {
 				if err := p.jsonApi.UnmarshalFromString(text.String, &arr[j]); err != nil {
 					p.logger.Error("[pg_cdc] failed to unmarshal json array element", slog.Any("error", err))

flow/connectors/postgres/qrep_query_executor.go

@@ -378,6 +378,10 @@ func (qe *QRepQueryExecutor) mapRowToQRecord(
 				qe.logger.Error("[pg_query_executor] failed to unmarshal json", slog.Any("error", err))
 				return nil, fmt.Errorf("failed to unmarshal json: %w", err)
 			}
+			if values[i] == nil {
+				// avoid confusing SQL null & JSON null by using struct{} as json null sentinel
+				values[i] = model.JsonNull{}
+			}
 		case pgtype.JSONArrayOID, pgtype.JSONBArrayOID:
 			var textArr pgtype.FlatArray[pgtype.Text]
 			if err := qe.conn.TypeMap().Scan(fd.DataTypeOID, fd.Format, buf, &textArr); err != nil {

flow/e2e/clickhouse_test.go

@@ -1388,6 +1388,88 @@ func (s ClickHouseSuite) Test_InfiniteTimestamp() {
 	RequireEnvCanceled(s.t, env)
 }
 
+func (s ClickHouseSuite) Test_JSON_Null() {
+	if _, ok := s.source.(*PostgresSource); !ok {
+		s.t.Skip("only applies to postgres")
+	}
+
+	srcTableName := "test_json_null"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	dstTableName := "test_json_null"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id INT PRIMARY KEY,
+			j_null JSON NULL,
+			j_notnull JSON NOT NULL,
+			j_sqlnull JSON NULL,
+			jb_null JSONB NULL,
+			jb_notnull JSONB NOT NULL,
+			jb_sqlnull JSONB NULL
+		);
+	`, srcFullName)))
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id,j_null,j_notnull,j_sqlnull,jb_null,jb_notnull,jb_sqlnull)
+		VALUES (1,'null'::json,'null'::json,NULL,'null'::jsonb,'null'::jsonb,NULL)`,
+			srcFullName)))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      s.attachSuffix("ch_json_null"),
+		TableNameMapping: map[string]string{srcFullName: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+	flowConnConfig.Env = map[string]string{"PEERDB_NULLABLE": "true"}
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTableName, dstTableName, "id")
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id,j_null,j_notnull,j_sqlnull,jb_null,jb_notnull,jb_sqlnull)
+		VALUES (2,'null'::json,'null'::json,NULL,'null'::jsonb,'null'::jsonb,NULL)`,
+			srcFullName)))
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc", srcTableName, dstTableName, "id")
+
+	ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig())
+	require.NoError(s.t, err)
+	rows, err := ch.Query(s.t.Context(),
+		fmt.Sprintf("select id,j_null,j_notnull,j_sqlnull,jb_null,jb_notnull,jb_sqlnull from %s order by id", dstTableName))
+	require.NoError(s.t, err)
+	defer rows.Close()
+	numRows := 0
+	for rows.Next() {
+		numRows++
+		var id int32
+		var jNull *string
+		var jbNull *string
+		var jNotNull string
+		var jbNotNull string
+		var jSqlNull *string
+		var jbSqlNull *string
+		require.NoError(s.t, rows.Scan(&id, &jNull, &jNotNull, &jSqlNull, &jbNull, &jbNotNull, &jbSqlNull))
+		s.t.Log(id, jNull, jbNull, jNotNull, jbNotNull)
+		require.NotNil(s.t, jNull)
+		require.NotNil(s.t, jbNull)
+		require.Equal(s.t, "null", *jNull)
+		require.Equal(s.t, "null", jNotNull)
+		require.Equal(s.t, "null", *jbNull)
+		require.Equal(s.t, "null", jbNotNull)
+		require.Nil(s.t, jSqlNull)
+		require.Nil(s.t, jbSqlNull)
+	}
+	require.NoError(s.t, rows.Err())
+	require.Equal(s.t, 2, numRows)
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 func (s ClickHouseSuite) Test_JSON_CH() {
 	if mySource, ok := s.source.(*MySqlSource); ok && mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
 		s.t.Skip("skip maria, where JSON is not a supported data type")

flow/model/model.go

@@ -13,6 +13,12 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )
 
+type JsonNull struct{}
+
+func (JsonNull) MarshalJSON() ([]byte, error) {
+	return []byte("null"), nil
+}
+
 type NameAndExclude struct {
 	Exclude map[string]struct{}
 	Name    string