feat(cdc): support mapping json of pg to jsonb in rw in pg-cdc

[KeXiangWang]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

As titled

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• My PR contains critical fixes that are necessary to be merged into the latest release.

Documentation

• My PR needs documentation updates.

Release note

[xiangjinwu]

Note that json and jsonb have different semantics in PostgreSQL:

test=# select jsonb_each('{"foo": 1, "bar": 2, "foo": 3}');
 jsonb_each 
------------
 (bar,2)
 (foo,3)
(2 rows)

test=# select json_each('{"foo": 1, "bar": 2, "foo": 3}');
 json_each 
-----------
 (foo,1)
 (bar,2)
 (foo,3)
(3 rows)

Similarly, char(8) is different from varchar. But for PostgreSQL CDC source, it may be acceptable as long as we clarify this in doc.

[xiangjinwu]

Let's acknowledge in comments that treating JSON as JSONB can be wrong in special cases.

[KeXiangWang]

Added. Will also add it in doc.

[xiangjinwu]

How is this used in RisingWave? I see CHAR_ARRAY and TIMETZ_ARRAY here which do not exist in RisingWave.

[KeXiangWang]

This part of codes are used to load the schema of the table to be connected via source or sink.
timestamptz[] should be supported in RW?

[xiangjinwu]

I meant it should be either:

    // mapping into RisingWave supported data types
    SeaType::Char(_) | SeaType::Varchar(_) => Ok(PgType::VARCHAR),
    SeaType::Json | SeaType::JsonBinary => Ok(PgType::JSONB)

or

    // mapping into PostgreSQL data types, regardless of RisingWave support
    SeaType::Varchar(_) => Ok(PgType::VARCHAR),
    SeaType::Char(_) => Ok(PgType::CHAR),
    SeaType::JsonBinary => Ok(PgType::JSONB),
    SeaType::Json => Ok(PgType::JSON),

But after this PR it is:

    SeaType::Varchar(_) => Ok(PgType::VARCHAR),
    SeaType::Char(_) => Ok(PgType::CHAR),
    SeaType::Json | SeaType::JsonBinary => Ok(PgType::JSONB)

That's why I am asking how is this function used / what is its responsibility.