refactor(postgres-cdc): refactor postgres_row_to_owned_row #16714

KeXiangWang · 2024-05-13T04:11:36Z

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

What's changed and what's your intention?

Try refactoring to resolve #16416 (comment)

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

KeXiangWang · 2024-05-13T05:41:33Z

Before this PR, NULL::TYPE[] is not well handled, it will be parsed as {} instead of NULL. The bug is due to the following style of codes.

let mut builder = dtype.create_array_builder(0);
// do something
if let Some(v) = val {
    v.into_iter().for_each(|val| {
        builder.append(
            val.map(|v| ScalarImpl::from(v.into_boxed_slice())),
        )
    });
}
// do something
Some(ScalarImpl::from(ListValue::new(builder.finish())))

The correct way to handle NULL::TYPE[] is to check if the value is None before creating the array builder.

if let Some(v) = val {
    let mut builder = dtype.create_array_builder(0);
    v.into_iter().for_each(|val| {
        builder.append(
            val.map(|v| ScalarImpl::from(v.into_boxed_slice())),
        )
    });
    Some(ScalarImpl::from(ListValue::new(builder.finish())))
} else {
    None
}

KeXiangWang · 2024-05-13T05:45:20Z

An open question: Do you think we should also add UuidList(Vec<Option<uuid::Uuid>>) and EnumList(Vec<Option<EnumString>>) in ScalarAdapter?

src/connector/src/parser/scalar_adapter.rs

KeXiangWang · 2024-05-21T03:13:25Z

After refactoring, multi-dismension list should trivial to support. Will add it in next PR.

During refactoring, I find an interesting thing: while testing, I setup a postgres as data source and I run the test multiple times. The second round of test for the postgres will usually fail but the first and following tests will succeed🤣. Will continue to investigate.

xiangjinwu · 2024-05-21T03:29:58Z

After refactoring, multi-dismension list should trivial to support. Will add it in next PR.

https://docs.rs/postgres-types/latest/postgres_types/trait.FromSql.html#arrays
https://docs.rs/postgres-types/latest/postgres_types/trait.ToSql.html#arrays

FromSql is implemented for Vec<T>, Box<[T]> and [T; N] where T implements FromSql, and corresponds to one-dimensional Postgres arrays.
ToSql is implemented for [u8; N], Vec<T>, &[T], Box<[T]> and [T; N] where T implements ToSql and N is const usize, and corresponds to one-dimensional Postgres arrays with an index offset of 1.

Related: #3811 (comment) #15614

KeXiangWang · 2024-05-23T04:14:18Z

During this refactoring, I found three another bugs of Debezium CDC.

#16895
Early timestamp is not handled correctly, so I have to use a timestamp close to now to avoid this error.
#16880
When enum in PG is recreated (delete and create with the same name), the type registry won't keep up with upstream so that Debezium cannot handle the enum correctly, so that lead to a NULL for this enum in debezium message. I have no solution to resolve, so temporarily didn't add test for this.
#16882
INTERVAL_ARRAY is not supported. Resolved by not allowing INTERVAL_ARRAY's into_scalar

BTW, besides these three, I suspect Debezium may not support "inf/-inf/nan" for some types. Will check later.

StrikeW · 2024-05-24T03:03:07Z

src/connector/src/parser/scalar_adapter.rs

+            (ScalarRefImpl::List(list), &Type::NUMERIC_ARRAY, _) => {
+                let mut vec = vec![];
+                for scalar in list.iter() {
+                    vec.push(match scalar {


The scalar should rename to datum, because Datum is Option<ScalarImpl>. And I am curious why not vec.push( datum.map(|scalar| ScalarAdapter::from_scalar(scalar, &Type::NUMERIC) ) ) to call the from_scalar recursively? Just add one more branch to handle (ScalarRefImpl::Decimal, &Type::NUMERIC) as base case.

The scalar should rename to datum, because Datum is Option

updated.

And I am curious why not vec.push( datum.map(|scalar| ScalarAdapter::from_scalar(scalar, &Type::NUMERIC) ) ) to call the from_scalar recursively? Just add one more branch to handle (ScalarRefImpl::Decimal, &Type::NUMERIC) as base case.

For two reason:

from_scalar return ScalarAdapter directly, but ScalarAdapter::NumericList requires Vec<Option<PgNumeric>> actually.

If we don't handle (ScalarRefImpl::Decimal, &Type::NUMERIC), ScalarRefImpl::Decimal will be builtin. We have implement ToSql for rw's Decimal, so it acceptale by PG. If we handle (ScalarRefImpl::Decimal, &Type::NUMERIC), we have to first convert it to string and then parse as PgNumeric, seems unnecessary.

If we handle (ScalarRefImpl::Decimal, &Type::NUMERIC), we have to first convert it to string and then parse as PgNumeric, seems unnecessary.

Your are right, I got it.

src/connector/src/parser/scalar_adapter.rs

StrikeW

LGTM!

src/connector/src/parser/scalar_adapter.rs

src/connector/src/parser/postgres.rs

e2e_test/source/cdc/cdc.check_new_rows.slt

StrikeW · 2024-05-25T03:48:58Z

src/common/src/types/from_sql.rs

+    Date, Interval, JsonbVal, ScalarImpl, Time, Timestamp, Timestamptz,
+};
+
+impl<'a> FromSql<'a> for ScalarImpl {


These FromSql implementations LGTM, @xiangjinwu please also take a look

xiangjinwu

Thanks for the efforts to test all these tricky cases and help to get them organized!

Just a reminder to all of us: rather than going too far away to be bug-compatible with debezium, we may:

Disallow ingesting such data type at all, for example interval[]
Spend more time on fixing debezium. IIRC there is dark magic in java class loader to replace a certain library class without forking the project

src/common/src/types/jsonb.rs

xiangjinwu · 2024-05-27T05:56:53Z

src/common/src/types/to_sql.rs


-impl ToSql for ScalarRefImpl<'_> {
+impl ToSql for ScalarImpl {


Why do we change from ScalarRefImpl to ScalarImpl? The former is always more general and there is as_scalar_ref. (That is, to for a ref type and from for an owned type.)

Most of our types that implement FromSql are scalar types instead of scalar_ref. So it's easier to implement FromSql for ScalarImpl then ScalarRefImpl. Besides, as the postgres_cell_to_scalar_impl's return type is ScalarImpl, ScalarImpl would be more natural.

src/common/src/types/from_sql.rs

src/connector/src/parser/scalar_adapter.rs

xiangjinwu · 2024-05-27T07:15:40Z

src/connector/src/parser/scalar_adapter.rs

+                    ScalarAdapter::List(vec)
+                }
+            },
+            _ => ScalarAdapter::Builtin(scalar.into_scalar_impl()),


.into_scalar_impl() contains a clone and is the original reason ScalarAdapter contains ScalarRefImpl. But given it is only used as primary key during backfill at this time, I guess cloning is okay here.

src/connector/src/parser/postgres.rs

KeXiangWang · 2024-05-27T19:48:50Z

Spend more time on fixing debezium.

Agree. I've noted down the bugs found as issues. But they may not be urgent, will fix when requested.

IIRC there is dark magic in java class loader to replace a certain library class without forking the project

Should be ClassLoader.

src/common/src/types/from_sql.rs

refactor(postgres-cdc): refactor postgres_row_to_owned_row

ed5494e

github-actions bot added the type/refactor label May 13, 2024

fix

1f064a4

KeXiangWang requested review from StrikeW and xiangjinwu May 13, 2024 06:01

github-actions bot added the breaking-change label May 13, 2024

StrikeW reviewed May 14, 2024

View reviewed changes

src/connector/src/parser/scalar_adapter.rs Outdated Show resolved Hide resolved

src/connector/src/parser/scalar_adapter.rs Show resolved Hide resolved

KeXiangWang added 7 commits May 20, 2024 17:31

impl fromsql

8dd7dd7

tmp

287ae9d

tmp

ca6a368

tmp

d6ae522

tmp

06a86c1

fix

da18139

Merge branch 'main' into wkx/refactor-postgres-row-to-owned-row

b7bce43

KeXiangWang added 3 commits May 21, 2024 22:32

fix interval_array

07a0370

change test to a recent timestamp

9e0c6f1

update result

20d2288

KeXiangWang requested a review from StrikeW May 23, 2024 04:14

StrikeW reviewed May 24, 2024

View reviewed changes

small refactor

530c04c

StrikeW approved these changes May 25, 2024

View reviewed changes

KeXiangWang added 4 commits May 26, 2024 19:41

fix and refactor test

5b70eb7

fix fmt

8f3d3e3

fix test

81d23ec

fix test

dc0fe7b

KeXiangWang added 4 commits May 26, 2024 21:00

fix test

6bb96f9

fix test

0c8678a

Merge branch 'main' into wkx/refactor-postgres-row-to-owned-row

d93c228

fix test

0f357f5

xiangjinwu reviewed May 27, 2024

View reviewed changes

small refactor

360a6af

KeXiangWang force-pushed the wkx/refactor-postgres-row-to-owned-row branch from efabf6b to 360a6af Compare May 27, 2024 19:19

xiangjinwu approved these changes May 28, 2024

View reviewed changes

src/common/src/types/from_sql.rs Outdated Show resolved Hide resolved

KeXiangWang added 2 commits May 28, 2024 22:45

Merge branch 'main' into wkx/refactor-postgres-row-to-owned-row

ea29485

small fix

fa21534

KeXiangWang enabled auto-merge May 28, 2024 14:47

KeXiangWang added 2 commits May 29, 2024 03:13

Merge branch 'main' into wkx/refactor-postgres-row-to-owned-row

ce630c2

fix test

78eb45c

KeXiangWang added this pull request to the merge queue May 28, 2024

Merged via the queue into main with commit 12f5c0d May 28, 2024
29 of 30 checks passed

KeXiangWang deleted the wkx/refactor-postgres-row-to-owned-row branch May 28, 2024 21:13

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(postgres-cdc): refactor postgres_row_to_owned_row #16714

refactor(postgres-cdc): refactor postgres_row_to_owned_row #16714

KeXiangWang commented May 13, 2024 •

edited

KeXiangWang commented May 13, 2024 •

edited

KeXiangWang commented May 13, 2024 •

edited

KeXiangWang commented May 21, 2024

xiangjinwu commented May 21, 2024

KeXiangWang commented May 23, 2024 •

edited

StrikeW May 24, 2024

KeXiangWang May 24, 2024

StrikeW May 25, 2024

StrikeW left a comment

StrikeW May 25, 2024

xiangjinwu left a comment

xiangjinwu May 27, 2024

KeXiangWang May 27, 2024

xiangjinwu May 27, 2024

KeXiangWang commented May 27, 2024


		impl ToSql for ScalarRefImpl<'_> {
		impl ToSql for ScalarImpl {

refactor(postgres-cdc): refactor postgres_row_to_owned_row #16714

refactor(postgres-cdc): refactor postgres_row_to_owned_row #16714

Conversation

KeXiangWang commented May 13, 2024 • edited

What's changed and what's your intention?

Checklist

Documentation

Release note

KeXiangWang commented May 13, 2024 • edited

KeXiangWang commented May 13, 2024 • edited

KeXiangWang commented May 21, 2024

xiangjinwu commented May 21, 2024

KeXiangWang commented May 23, 2024 • edited

StrikeW May 24, 2024

Choose a reason for hiding this comment

KeXiangWang May 24, 2024

Choose a reason for hiding this comment

StrikeW May 25, 2024

Choose a reason for hiding this comment

StrikeW left a comment

Choose a reason for hiding this comment

StrikeW May 25, 2024

Choose a reason for hiding this comment

xiangjinwu left a comment

Choose a reason for hiding this comment

xiangjinwu May 27, 2024

Choose a reason for hiding this comment

KeXiangWang May 27, 2024

Choose a reason for hiding this comment

xiangjinwu May 27, 2024

Choose a reason for hiding this comment

KeXiangWang commented May 27, 2024

KeXiangWang commented May 13, 2024 •

edited

KeXiangWang commented May 13, 2024 •

edited

KeXiangWang commented May 13, 2024 •

edited

KeXiangWang commented May 23, 2024 •

edited