feat(connector): SinkExecutor init + MySQLSink mvp #2969

nanderstabel · 2022-06-02T22:22:55Z

What's changed and what's your intention?

Intention is to create a simple start of the implementation of a the SinkExecutor MVP.

Please explain IN DETAIL what the changes are in this PR and why they are needed:

Sink trait is added:

pub trait Sink {
    fn write_batch(&mut self);
}

As well as a simple SinkExecutor struct:

pub struct SinkExecutor<S: Sink> {
    child: BoxedExecutor,
    external_sink: S,
    identity: String
}

As well as an initial MySQLSink struct:

pub struct MySQLSink {
    endpoint: String,
    table: String,
    database: Option<String>,
    user: Option<String>,
    password: Option<String>,
}

For MySQLSink is a mvp version of Fn write_batch implemented.

Checklist

I have written necessary docs and comments
I have added necessary unit tests and integration tests
All checks passed in ./risedev check (or alias, ./risedev c)

neverchanje · 2022-06-03T01:32:20Z

I suggest you should create a new crate for the sink code. Because we may not want the MySQL and redis libs bundled in the streaming crate.

tabVersion · 2022-06-03T05:17:20Z

I suggest you should create a new crate for the sink code.

is sink on the same level of source? we may reuse the connector's code later?

nanderstabel · 2022-06-03T07:25:44Z

I suggest you should create a new crate for the sink code.

is sink on the same level of source? we may reuse the connector's code later?

I was thinking it would make sense to move sink into the connector crate and have a separate sink and source directory there. What do you think @neverchanje ?

neverchanje · 2022-06-06T07:27:35Z

I was thinking it would make sense to move sink into the connector crate and have a separate sink and source directory there. What do you think @neverchanje ?

I partly agree, we can go iteratively and leave the sink code in connector crate in the first version.

github-actions

license-eye has totally checked 832 files.

Valid	Invalid	Ignored	Fixed
830	1	1	0

Click to see the invalid file list

src/frontend/test_runner/tests/gen/testcases.rs

src/frontend/test_runner/tests/gen/testcases.rs

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

…data/risingwave-dev into ns-sink-write-batch

neverchanje · 2022-06-12T05:49:02Z

src/frontend/test_runner/tests/gen/testcases.rs

+
+
+#[tokio::test]
+async fn test_expr() {


We don't have this file now. It was removed by some pr long ago.

neverchanje · 2022-06-12T05:49:56Z

src/connector/src/sink.rs

+pub struct RedisSink;
+
+#[async_trait]
+impl Sink for RedisSink {


/sink/mysql.rs
/sink/redis.rs
/sink/mod.rs

Personally recommend this way to organize the files.

neverchanje · 2022-06-12T05:53:36Z

src/connector/src/sink.rs

+    fn endpoint(&self) -> String;
+    fn table(&self) -> String;
+    fn database(&self) -> Option<String>;
+    fn user(&self) -> Option<String>;
+    fn password(&self) -> Option<String>; // TODO(nanderstabel): auth?


This is sink-specific. Not all sinks have the concepts of "table", "database", or "user". You need to generalize them into a "SinkConfig", which could be an enum.

pub enum SinkConfig { Mysql(MysqlConfig), Redis(RedisConfig) } impl Sink { fn new(cfg: SinkConfig) { match cfg { SinkConfig::Mysql(cfg) => ... } } }

Not 100% what you mean by impl Sink since Sink is a trait?

I solved it now like this:

pub enum SinkImpl { MySQL(MySQLSink), Redis(RedisSink), } impl SinkImpl { fn new(cfg: SinkConfig) -> Self { match cfg { SinkConfig::Mysql(cfg) => SinkImpl::MySQL(MySQLSink::new(cfg)), SinkConfig::Redis(cfg) => SinkImpl::Redis(RedisSink::new(cfg)), } } }

Sorry for not giving the right syntax. I'm indicating https://doc.rust-lang.org/book/ch10-02-traits.html. Anyway, SinkImpl also looks good to me.

neverchanje · 2022-06-12T06:00:38Z

The overall design looks nice to me, but I think the e2e testing would be a hard problem to solve.

…data/risingwave-dev into ns-sink-write-batch

skyzh · 2022-06-13T04:51:33Z

src/common/Cargo.toml

@@ -23,6 +23,7 @@ lru = { git = "https://github.com/singularity-data/lru-rs.git", rev = "e0e9ddaf8
 madsim = "=0.2.0-alpha.3"
 memcomparable = { path = "../utils/memcomparable" }
 more-asserts = "0.3"
+mysql_async = "0.30"


Don't add more dependencies to common crate 🤣

skyzh · 2022-06-13T04:51:48Z

src/common/src/error.rs

@@ -138,6 +139,8 @@ pub enum ErrorCode {
    },
    #[error("Invalid Parameter Value: {0}")]
    InvalidParameterValue(String),
+    #[error("MySQL error: {0}")]
+    MySQLError(MySQLError),


use BoxedError here so that we don't need to import mysql_async to common crate

codecov · 2022-06-13T11:32:51Z

Codecov Report

Merging #2969 (ca2a6e6) into main (1b749b3) will decrease coverage by 0.05%.
The diff coverage is 43.21%.

@@            Coverage Diff             @@
##             main    #2969      +/-   ##
==========================================
- Coverage   73.65%   73.59%   -0.06%     
==========================================
  Files         740      744       +4     
  Lines      101830   102029     +199     
==========================================
+ Hits        75001    75089      +88     
- Misses      26829    26940     +111

Flag	Coverage Δ
rust	`73.59% <43.21%> (-0.06%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/common/src/error.rs	`66.17% <0.00%> (-0.25%)`	⬇️
src/common/src/types/mod.rs	`67.91% <ø> (ø)`
src/connector/src/lib.rs	`100.00% <ø> (ø)`
src/connector/src/sink/mod.rs	`0.00% <0.00%> (ø)`
src/connector/src/sink/redis.rs	`0.00% <0.00%> (ø)`
src/stream/src/executor/mod.rs	`48.63% <ø> (ø)`
src/connector/src/sink/mysql.rs	`43.97% <43.97%> (ø)`
src/stream/src/executor/sink.rs	`64.86% <64.86%> (ø)`
src/common/src/types/ordered_float.rs	`24.70% <0.00%> (-0.20%)`	⬇️
src/common/src/types/chrono_wrapper.rs	`83.07% <0.00%> (+2.30%)`	⬆️

📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

neverchanje

Lgtm, great work!

SinkExecutor init

490216d

github-actions bot added the Invalid PR Title label Jun 2, 2022

nanderstabel changed the title ~~SinkExecutor init~~ feat(stream): SinkExecutor + Trait init Jun 2, 2022

github-actions bot added type/feature and removed Invalid PR Title labels Jun 2, 2022

nanderstabel added 5 commits June 6, 2022 14:44

Merge branch 'main' into ns-sink-write-batch

d499c02

MySQLSink init

7a7f5cf

lock

7c93c08

Merge branch 'main' into ns-sink-write-batch

21c77d9

add very rudimental implementation of MySQLSink write_batch()

9972c1e

github-actions bot reviewed Jun 7, 2022

View reviewed changes

src/frontend/test_runner/tests/gen/testcases.rs Outdated Show resolved Hide resolved

nanderstabel and others added 15 commits June 7, 2022 22:02

cargo fmt

7c3a5c3

Refactor MySQLSink::write_batch() and improve test

9f73fa2

Add MySQLType

2e38fda

Add DELETE to write_batch() and improve test

23d7037

Refactor

a56d36d

Improve test

ad0aec7

Integrate mysql_async::Error into RwError

07c2b0e

Implement Update operation in write_batch()

34a1c1f

Add closures

ec5ab5e

Add UPDATE to basic test

4265e6d

Fix merge conflict in error.rs

49c4f01

Fix ./risedev test

c1ce6a0

Update src/frontend/test_runner/tests/gen/testcases.rs

a257257

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

apply clippy

5d43580

Improve test

ebdd448

nanderstabel added 5 commits June 11, 2022 18:06

Add scalars and tests

e470558

Clean up

794fc89

Apply ./risedev check

2e22164

reintroduce MySQLError

6bd2399

Merge branch 'ns-sink-write-batch' of https://github.com/singularity-…

08a253e

…data/risingwave-dev into ns-sink-write-batch

nanderstabel assigned nanderstabel and unassigned nanderstabel Jun 11, 2022

neverchanje reviewed Jun 12, 2022

View reviewed changes

Merge branch 'main' into ns-sink-write-batch

f4cb487

nanderstabel added 6 commits June 12, 2022 17:57

Organize sink files

7be8f1c

Remove testcases.rs

9a27030

Merge branch 'ns-sink-write-batch' of https://github.com/singularity-…

f66e9b8

…data/risingwave-dev into ns-sink-write-batch

Merge branch 'main' into ns-sink-write-batch

d54aa52

Add SinkConfig

443ef35

impl Sink for SinkImpl

6a6be7b

skyzh reviewed Jun 13, 2022

View reviewed changes

nanderstabel added 4 commits June 13, 2022 12:01

Introduce SinkError

2237840

Merge branch 'main' into ns-sink-write-batch

b705609

Merge main

749ef6a

Merge branch 'main' into ns-sink-write-batch

c37239f

Merge branch 'main' into ns-sink-write-batch

ca2a6e6

nanderstabel marked this pull request as ready for review June 13, 2022 11:56

nanderstabel enabled auto-merge (squash) June 13, 2022 11:56

neverchanje approved these changes Jun 13, 2022

View reviewed changes

nanderstabel merged commit b7537bb into main Jun 13, 2022

nanderstabel deleted the ns-sink-write-batch branch June 13, 2022 12:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(connector): SinkExecutor init + MySQLSink mvp #2969

feat(connector): SinkExecutor init + MySQLSink mvp #2969

nanderstabel commented Jun 2, 2022 •

edited

neverchanje commented Jun 3, 2022

tabVersion commented Jun 3, 2022

nanderstabel commented Jun 3, 2022

neverchanje commented Jun 6, 2022

github-actions bot left a comment

neverchanje Jun 12, 2022

neverchanje Jun 12, 2022

neverchanje Jun 12, 2022 •

edited

nanderstabel Jun 12, 2022

neverchanje Jun 13, 2022

neverchanje commented Jun 12, 2022

skyzh Jun 13, 2022 •

edited

neverchanje Jun 13, 2022

skyzh Jun 13, 2022

codecov bot commented Jun 13, 2022 •

edited

neverchanje left a comment



		#[tokio::test]
		async fn test_expr() {

feat(connector): SinkExecutor init + MySQLSink mvp #2969

feat(connector): SinkExecutor init + MySQLSink mvp #2969

Conversation

nanderstabel commented Jun 2, 2022 • edited

What's changed and what's your intention?

Checklist

neverchanje commented Jun 3, 2022

tabVersion commented Jun 3, 2022

nanderstabel commented Jun 3, 2022

neverchanje commented Jun 6, 2022

github-actions bot left a comment

Choose a reason for hiding this comment

neverchanje Jun 12, 2022

Choose a reason for hiding this comment

neverchanje Jun 12, 2022

Choose a reason for hiding this comment

neverchanje Jun 12, 2022 • edited

Choose a reason for hiding this comment

nanderstabel Jun 12, 2022

Choose a reason for hiding this comment

neverchanje Jun 13, 2022

Choose a reason for hiding this comment

neverchanje commented Jun 12, 2022

skyzh Jun 13, 2022 • edited

Choose a reason for hiding this comment

neverchanje Jun 13, 2022

Choose a reason for hiding this comment

skyzh Jun 13, 2022

Choose a reason for hiding this comment

codecov bot commented Jun 13, 2022 • edited

Codecov Report

neverchanje left a comment

Choose a reason for hiding this comment

nanderstabel commented Jun 2, 2022 •

edited

neverchanje Jun 12, 2022 •

edited

skyzh Jun 13, 2022 •

edited

codecov bot commented Jun 13, 2022 •

edited