feat: impl `key encode` clause for sink #16377

tabVersion · 2024-04-18T08:39:41Z

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

What's changed and what's your intention?

as title, introducing key encode clause in sink statement.

the whole statement looks like:

CREATE SINK [ IF NOT EXISTS ] sink_name
[FROM sink_from | AS select_query]
WITH (
   connector='kafka',
   connector_parameter = 'value', ...
)
FORMAT data_format ENCODE data_encode [ (
    key = 'value'
) ]
[KEY ENCODE key_encode [(...)]]

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.

adding the key encode at the end of the sink statement.

The key encode can only be TEXT now, and the primary key specified in SQL should be one and only one of the types (varchar, bool, small int, int, big int)

gitguardian · 2024-04-29T06:58:57Z

⚠️ GitGuardian has uncovered 1 secret following the scan of your pull request.

Please consider investigating the findings and remediating the incidents. Failure to do so may lead to compromising the associated services or software components.

🔎 Detected hardcoded secret in your pull request

GitGuardian id	GitGuardian status	Secret	Commit	Filename
9425213	Triggered	Generic Password	`3ba944b`	e2e_test/source/cdc/cdc.validate.postgres.slt	View secret

🛠 Guidelines to remediate hardcoded secrets

Understand the implications of revoking this secret by investigating where it is used in your code.
Replace and store your secret safely. Learn here the best practices.
Revoke and rotate this secret.
If possible, rewrite git history. Rewriting git history is not a trivial act. You might completely break other contributing developers' workflow and you risk accidentally deleting legitimate data.

To avoid such incidents in the future consider

following these best practices for managing and storing secrets including API keys and other credentials
install secret detection on pre-commit to catch secret before it leaves your machine and ease remediation.

^{🦉 GitGuardian detects secrets in your source code to help developers and security teams secure the modern development process. You are seeing this because you or someone else with access to this repository has authorized GitGuardian to scan your pull request.}

tabVersion · 2024-05-03T06:55:15Z

wait for #16575 merge first.

fuyufjh

Overall LGTM

fuyufjh · 2024-05-07T05:13:04Z

src/connector/src/sink/formatter/mod.rs

 UpsertTemplate(UpsertFormatter<TemplateEncoder, TemplateEncoder>),
+ UpsertTextTemplate(UpsertFormatter<TextEncoder, TemplateEncoder>),


So here we expand the cartesian product of

{Append-only, Upsert} x {Text (key) or not} x {Json, Proto, Template}

Is it possible to separate them?

Possibly. The caller of formatter does not need to know whether it is AvroEncoder or JsonEncoder, but it still need to know whether the output of an encoder SerTo a (utf8) String or not (Vec<u8>). So it may be possible to reduce into {Append-only, Upsert} x {String, Vec<u8>} x {String, Vec<u8>} (or possibly {Append-only, Upsert} x {String, Vec<u8>} x {Vec<u8>} if all MQs allow payload to be non-utf8 bytes). But the refactor wont fit into this PR. I will give it a try after this. #16628

src/frontend/src/optimizer/plan_node/stream_sink.rs

fuyufjh · 2024-05-07T05:19:00Z

src/frontend/src/optimizer/plan_node/stream_sink.rs

+ | DataType::Int16
+ | DataType::Int32
+ | DataType::Int64 => {}
+ _ => {


Please explicitly exhaust all data types here. It helps to you to think about all the data types, and also reminds the other people to maintain this code when new data type is introduced.

And I think many data types are missing, such as Decimal, Int256, Timestamp, etc

Refer to #16377 (comment) for a list of all data types, and why we want to be conservative on some data types initially.

src/connector/src/sink/formatter/mod.rs

xiangjinwu · 2024-05-07T04:51:45Z

src/connector/src/sink/formatter/mod.rs

+ (F::AppendOnly, E::Json, Some(E::Text)) => {
+ Impl::AppendOnlyTextJson(build(p).await?)
+ }
+ (F::AppendOnly, E::Json, None | Some(_)) => Impl::AppendOnlyJson(build(p).await?),


Some(_) shall not be ignored silently, although it probably already checked in frontend.

After a second thought, I only keep (F::AppendOnly, E::Json, None) for append-only json. Other key encode config is invalid and should be rejected (although it is done in the frontend).

src/connector/src/sink/encoder/text.rs

xiangjinwu · 2024-05-07T06:14:48Z

src/connector/src/sink/encoder/text.rs

+ pub fn new(schema: Schema, col_indices: Option<Vec<usize>>) -> Self {
+ Self {


Prefer to restrict the data type here:

allowed: bool, int16, int32, int64, varchar, serial, int256

denied

float32, float64

1e6::float4 can be either 1e+06 or 1000000

interval, struct, list, bytea

PostgreSQL ToText unfriendly to downstream

jsonb

feat(frontend): ban jsonb in aggregation stream key #14442

maybe

decimal

1.0 and 1.00 are equal as decimal but unequal as text

date

-0001-01-01 is same as 0002-01-01 BC

time, timestamptz

12:34:56.020 is same as 12:34:56.02 (i.e. pad to 3/6/9 or not)

timestamp

2024-01-01 12:34:56 is neither ISO 8601 nor RFC 3339 but a mixture of both.

src/sqlparser/src/ast/statement.rs

xiangjinwu · 2024-05-07T06:42:38Z

src/frontend/src/optimizer/plan_node/stream_sink.rs

+ if let Some(format_desc) = &format_desc
+ && let Some(key_encode) = &format_desc.key_encode


Encoder-specific check in EncoderBuild::build() and ::new(). This file can be untouched.

The differences between new and build:

New is supposed to be the minimal parameter required by the encoder itself. So it shall accept a single col_index: usize.

Build is supposed to adapt the common input to the ones required by new. So it checks whether pk_indices contains exactly one column.

xiangjinwu · 2024-05-07T06:44:59Z

e2e_test/sink/kafka/create_sink.slt

+ primary_key = 'id')
+format plain encode json (
+ force_append_only='true'
+) key encode text ;


Ingest it back with include key as and compare it with expected value?

already included in the test.

Signed-off-by: tabVersion <[email protected]>

e2e_test/sink/kafka/create_sink.slt

src/sqlparser/src/ast/legacy_source.rs

src/sqlparser/src/ast/statement.rs

src/sqlparser/src/parser.rs

src/connector/src/sink/formatter/mod.rs

xiangjinwu · 2024-05-08T05:02:17Z

src/connector/src/sink/encoder/text.rs

+ let datum = row.datum_at(col_index);
+ result = datum.to_text_with_type(&self.schema.fields[col_index].data_type);


Second point of #16377 (comment) unresolved:

Let's make an exception for bool to return true/false rather than t/f used by PostgreSQL ToText. (Similarly in rw-expr and pg-proc)

Oh I thought it is to_text_with_type 's default behavior. Let me fix.

Signed-off-by: tabVersion <[email protected]>

fuyufjh · 2024-05-08T10:54:17Z

src/sqlparser/src/ast/statement.rs

+ Canal,
+ // Keyword::UPSERT
+ Upsert,
+ // Keyword::PLAIN


Is it intended? Or just a formatter issue?

It seems a RustRover's bug, it has its own formatter instead of using the rust toolchain one.

I remember there is an option like "use reustfmt instead of the built-in formatter". Btw, don't forget to fix this

Fixed already

src/sqlparser/src/ast/legacy_source.rs

src/sqlparser/src/ast/statement.rs

Signed-off-by: tabVersion <[email protected]>

xiangjinwu

Rest LGTM

xiangjinwu · 2024-05-08T12:37:48Z

src/connector/src/sink/encoder/text.rs

+ let data_type = &self.schema.fields[col_index].data_type;
+ if data_type == &DataType::Boolean {
+ result = datum
+ .unwrap_or(ScalarRefImpl::Bool(false))


Good point ... I do not feel outputting false is right when the input is null. I just checked ToText for Datum and it is outputting NULL. Maybe keep consistent here.

As for PostgreSQL, there is no text or binary format for null. It is always indicated by a side channel.

The typoutput procedure for each pg_type always return null::varchar.

In psql it is empty string by default but customizable via \pset null.

The typoutput of an array element (rather than the whole array) outputs NULL.

The typoutput of a record field (rather than the whole record) outputs empty string.

fixed. And thanks for explaining the pg behavior. Then why we make an exception for the boolean type? Directly using the ToText trait meets the behavior above and is consistent with other types.

why we make an exception for the boolean type

PostgreSQL ToText PostgreSQL v::varchar Popular in other programming languages TextEncoder

bool t/f true/false true/false follow v::varchar

list {1,2,3} {1,2,3} [1,2,3] disallowed

int 49 49 49 49 (rather than 0x31)

So actually we only allowed unambiguous data types 😂 Or we could say PostgreSQL also made an exception for bool in v::varchar which we follow, while it is unfortunate there is no such trait ...

src/connector/src/sink/formatter/mod.rs

Signed-off-by: tabVersion <[email protected]>

fuyufjh

LGTM!

fuyufjh · 2024-05-08T16:06:32Z

src/sqlparser/src/ast/statement.rs

+ Canal,
+ // Keyword::UPSERT
+ Upsert,
+ // Keyword::PLAIN


I remember there is an option like "use reustfmt instead of the built-in formatter". Btw, don't forget to fix this

update sql parser

ba5e1c7

github-actions bot added the type/feature label Apr 18, 2024

tabVersion added 5 commits April 19, 2024 15:17

stash

ebc7ec3

stash

e029373

stash

8e743e6

fix

ceb6a29

Merge branch 'main' into tab/key-encode

3ba944b

tabVersion added 2 commits April 30, 2024 16:11

impl text encoder

04df211

impl sink dispatch

81ee330

tabVersion marked this pull request as ready for review April 30, 2024 08:38

tabVersion requested review from chenzl25, xiangjinwu and xxhZs April 30, 2024 08:38

tabVersion added 6 commits April 30, 2024 16:45

fix

5aa02a8

fix

69bd77e

fix

0507809

Merge branch 'main' into tab/key-encode

2d845b0

Merge remote-tracking branch 'origin' into tab/key-encode

46222d5

add e2e test

c8c3979

tabVersion added the user-facing-changes Contains changes that are visible to users label Apr 30, 2024

drop sink

f4233f6

tabVersion added 2 commits May 6, 2024 19:09

Merge branch 'main' into tab/key-encode

15924d4

fix

313216c

fuyufjh self-requested a review May 7, 2024 03:40

fuyufjh reviewed May 7, 2024

View reviewed changes

xiangjinwu reviewed May 7, 2024

View reviewed changes

xiangjinwu mentioned this pull request May 7, 2024

refactor(sink): TextEncoder can be a special case of TemplateEncoder #16605

Open

resolve comments

250ef64

tabVersion added 7 commits May 7, 2024 13:19

resolve comments

687ea90

migrate sink schema check inside build

1870ba8

Signed-off-by: tabVersion <[email protected]>

fix test

08d58e1

Signed-off-by: tabVersion <[email protected]>

add drop table

5c0bf16

Signed-off-by: tabVersion <[email protected]>

fix

02cb6d1

Signed-off-by: tabVersion <[email protected]>

rerun

8a68348

Signed-off-by: tabVersion <[email protected]>

create topic in advance

5416c79

Signed-off-by: tabVersion <[email protected]>

xiangjinwu mentioned this pull request May 8, 2024

refactor(sink): reduce SinkFormatterImpl branches #16628

Open

xiangjinwu reviewed May 8, 2024

View reviewed changes

tabVersion added 2 commits May 8, 2024 10:20

Merge remote-tracking branch 'origin' into tab/key-encode

3f0697d

resolve comments

09caf0a

Signed-off-by: tabVersion <[email protected]>

fuyufjh reviewed May 8, 2024

View reviewed changes

make an exception for bool

2ffdc53

Signed-off-by: tabVersion <[email protected]>

xiangjinwu reviewed May 8, 2024

View reviewed changes

tabVersion added 2 commits May 8, 2024 14:59

output null for when we get NULL in bool

07c7587

Signed-off-by: tabVersion <[email protected]>

format manually

fed0cc2

Signed-off-by: tabVersion <[email protected]>

fuyufjh approved these changes May 8, 2024

View reviewed changes

xiangjinwu approved these changes May 8, 2024

View reviewed changes

tabVersion added this pull request to the merge queue May 8, 2024

github-merge-queue bot removed this pull request from the merge queue due to failed status checks May 8, 2024

Merge branch 'main' into tab/key-encode

0128d55

tabVersion enabled auto-merge May 8, 2024 21:32

tabVersion added this pull request to the merge queue May 8, 2024

Merged via the queue into main with commit 926ba1b May 8, 2024
27 of 30 checks passed

tabVersion deleted the tab/key-encode branch May 8, 2024 22:05

neverchanje mentioned this pull request May 8, 2024

Document: feat: impl key encode clause for sink risingwavelabs/risingwave-docs#2148

Closed

WanYixian mentioned this pull request May 29, 2024

Add key encode clause for sink risingwavelabs/risingwave-docs#2215

Merged

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: impl `key encode` clause for sink #16377

feat: impl `key encode` clause for sink #16377

tabVersion commented Apr 18, 2024 •

edited

Loading

gitguardian bot commented Apr 29, 2024 •

edited

Loading

tabVersion commented May 3, 2024

fuyufjh left a comment

fuyufjh May 7, 2024

xiangjinwu May 7, 2024 •

edited

Loading

fuyufjh May 7, 2024 •

edited

Loading

xiangjinwu May 7, 2024

xiangjinwu May 7, 2024

tabVersion May 7, 2024

xiangjinwu May 7, 2024

xiangjinwu May 7, 2024 •

edited

Loading

xiangjinwu May 7, 2024

tabVersion May 7, 2024

xiangjinwu May 8, 2024

tabVersion May 8, 2024

fuyufjh May 8, 2024 •

edited

Loading

tabVersion May 8, 2024

fuyufjh May 8, 2024

tabVersion May 8, 2024

xiangjinwu left a comment

xiangjinwu May 8, 2024

tabVersion May 8, 2024

xiangjinwu May 8, 2024

fuyufjh left a comment

fuyufjh May 8, 2024

		UpsertTemplate(UpsertFormatter<TemplateEncoder, TemplateEncoder>),
		UpsertTextTemplate(UpsertFormatter<TextEncoder, TemplateEncoder>),

		pub fn new(schema: Schema, col_indices: Option<Vec<usize>>) -> Self {
		Self {

		if let Some(format_desc) = &format_desc
		&& let Some(key_encode) = &format_desc.key_encode

		let datum = row.datum_at(col_index);
		result = datum.to_text_with_type(&self.schema.fields[col_index].data_type);

	PostgreSQL `ToText`	PostgreSQL `v::varchar`	Popular in other programming languages	`TextEncoder`
bool	`t`/`f`	`true`/`false`	`true`/`false`	follow `v::varchar`
list	`{1,2,3}`	`{1,2,3}`	`[1,2,3]`	disallowed
int	`49`	`49`	`49`	`49` (rather than `0x31`)

feat: impl key encode clause for sink #16377

feat: impl key encode clause for sink #16377

Conversation

tabVersion commented Apr 18, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

gitguardian bot commented Apr 29, 2024 • edited Loading

⚠️ GitGuardian has uncovered 1 secret following the scan of your pull request.

tabVersion commented May 3, 2024

fuyufjh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiangjinwu May 7, 2024 • edited Loading

Choose a reason for hiding this comment

fuyufjh May 7, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiangjinwu May 7, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fuyufjh May 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiangjinwu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fuyufjh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

feat: impl `key encode` clause for sink #16377

feat: impl `key encode` clause for sink #16377

tabVersion commented Apr 18, 2024 •

edited

Loading

gitguardian bot commented Apr 29, 2024 •

edited

Loading

xiangjinwu May 7, 2024 •

edited

Loading

fuyufjh May 7, 2024 •

edited

Loading

xiangjinwu May 7, 2024 •

edited

Loading

fuyufjh May 8, 2024 •

edited

Loading