feat(sink): add Google Pub/Sub support #16363

jetjinser · 2024-04-17T09:27:05Z

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

What's changed and what's your intention?

Add new sink Goolge Pub/Sub with only support Plain => Json in append-only mode.

e.g.

CREATE TABLE IF NOT EXISTS personnel (id integer, name varchar);

CREATE SINK ggps_sink
FROM
  personnel
WITH
(
    connector = 'google_pubsub',
    endpoint = 'localhost:8900',
    emulator_host = 'localhost:8900',
    project_id = 'demo',
    topic = 'test',
    type = 'append-only',
) FORMAT PLAIN ENCODE JSON (
    force_append_only='true',
);

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

Support new sink google_pubsub.

xiangjinwu · 2024-04-19T09:05:25Z

src/connector/src/sink/google_pubsub.rs

+                    DateHandlingMode::FromCe,
+                    TimestampHandlingMode::Milli,
+                    timestamptz_mode,
+                    TimeHandlingMode::Milli,


Why not use their string form by default? It is quite confusing to see "date_of_birth": 730119, "daily_meeting_time": 39600000.

Or another question, why not reuse SinkFormatterImpl::new?

src/connector/src/sink/google_pubsub.rs

neverchanje · 2024-04-24T09:31:38Z

@tabVersion Please take a look. I hope that this can be included in the 1.9 version.

stdrc · 2024-04-29T05:43:52Z

src/connector/src/sink/google_pubsub.rs

+impl GooglePubSubConfig {
+    fn from_hashmap(values: HashMap<String, String>) -> Result<Self> {
+        serde_json::from_value::<GooglePubSubConfig>(
+            serde_json::to_value(values).expect("impossible"),


"impossible" is not a good expect reason. May just use unwrap(), no much difference.

stdrc · 2024-04-29T05:45:25Z

src/connector/src/sink/google_pubsub.rs

+    pub(crate) fn initialize_env(&self) {
+        tracing::debug!("setting pubsub environment variables");
+        if let Some(emulator_host) = &self.emulator_host {
+            std::env::set_var("PUBSUB_EMULATOR_HOST", emulator_host);
+        }
+        if let Some(credentials) = &self.credentials {
+            std::env::set_var("GOOGLE_APPLICATION_CREDENTIALS_JSON", credentials);
+        }
+    }


What if first create a pubsub sink with PUBSUB_EMULATOR_HOST and then create another without PUBSUB_EMULATOR_HOST?

I guess we should clean existing such env vars before setting them.

Btw, could you plz add some comment to explain why we set these env vars here? Any doc links?

Seems that this part is copied from source/google_pubsub/mod.rs. The source and sink code can also conflict. You can fix the original code together.

I think you can also try set the emulator host and credentials via ClientConfig instead of env var, to eliminate the possibility of env var conflict.

+1, it is quite strange that we set env variable for a sink job. Can you share more on why we need this ENV?

In order to set the environment in ClientConfig, we need to introduce another direct dependency (now indirect): https://docs.rs/google-cloud-gax/0.17.0/google_cloud_gax/conn/enum.Environment.html. google-cloud-pubsub dont re-export it (still) :|
If it's ok, I'll set it up via this.

No problem.

tabVersion · 2024-04-25T04:26:49Z

src/connector/Cargo.toml

+google-cloud-googleapis = { version = "0.12", features = ["pubsub"] }
 google-cloud-pubsub = "0.24"


Is there any overlap between the deps? Shall we just keep one of them?

No, google-cloud-pubsub depends on google-cloud-googleapis but does not re-export, which is required for publishing messages.

tabVersion · 2024-04-29T06:22:08Z

src/connector/src/sink/google_pubsub.rs

+    pub(crate) fn initialize_env(&self) {
+        tracing::debug!("setting pubsub environment variables");
+        if let Some(emulator_host) = &self.emulator_host {
+            std::env::set_var("PUBSUB_EMULATOR_HOST", emulator_host);
+        }
+        if let Some(credentials) = &self.credentials {
+            std::env::set_var("GOOGLE_APPLICATION_CREDENTIALS_JSON", credentials);
+        }
+    }


+1, it is quite strange that we set env variable for a sink job. Can you share more on why we need this ENV?

tabVersion · 2024-04-29T06:26:36Z

src/connector/src/sink/google_pubsub.rs

+        if !matches!(self.format_desc.encode, SinkEncode::Json) {
+            return Err(SinkError::GooglePubSub(anyhow!(
+                "Google Pub/Sub sink only support `Json` sink encode"
+            )));
+        }


please impl this in CONNECTORS_COMPATIBLE_FORMATS (src/frontend/src/handler/create_sink.rs)

tabVersion · 2024-04-29T06:27:21Z

src/connector/src/sink/google_pubsub.rs

+
+    async fn validate(&self) -> Result<()> {
+        if !self.is_append_only {
+            return Err(SinkError::Nats(anyhow!(


why NatsError?

support append-only mode and JSON encoding

because we have already checked the encode in `CONNECTORS_COMPATIBLE_FORMATS` (src/frontend/src/handler/create_sink.rs)

to avoid the need to set environment variables for authentication

stdrc

LGTM

WanYixian · 2024-05-22T03:56:19Z

@tabVersion Please take a look. I hope that this can be included in the 1.9 version.

Is this included in 1.9 now? Seems not in the comparing changes.

@lmatz @neverchanje @jetjinser PTAL, thanks!

neverchanje · 2024-05-22T04:57:37Z

@WanYixian It's not cherrypicked in v1.9. I'll let users try out with the nightly version.

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

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

jetjinser marked this pull request as ready for review April 19, 2024 08:46

jetjinser requested a review from a team as a code owner April 19, 2024 08:46

xiangjinwu reviewed Apr 19, 2024

View reviewed changes

jetjinser force-pushed the jinser/google-pubsub-sink branch from a0fe7dc to bd7907f Compare April 19, 2024 09:07

jetjinser requested a review from xiangjinwu April 23, 2024 00:40

neverchanje reviewed Apr 23, 2024

View reviewed changes

src/connector/src/sink/google_pubsub.rs Outdated Show resolved Hide resolved

neverchanje mentioned this pull request Apr 23, 2024

Sink to BigQuery with google pubsub sink #15306

Closed

neverchanje linked an issue Apr 23, 2024 that may be closed by this pull request

Sink to BigQuery with google pubsub sink #15306

Closed

neverchanje requested a review from tabVersion April 24, 2024 09:30

stdrc self-requested a review April 29, 2024 05:28

stdrc reviewed Apr 29, 2024

View reviewed changes

tabVersion reviewed Apr 29, 2024

View reviewed changes

jetjinser added 9 commits April 29, 2024 21:32

feat(sink): add Google Pub/Sub sink

35e690a

support append-only mode and JSON encoding

feat(sink): make google_pubsub sink config more consistent

25272cd

test: update with_options_sink test for google pubsub sink

986a7a1

feat(sink): use formatter for google pubsub sink

c02bfb0

chore: remove unnecessary comments

aa85312

fix: change error type to GooglePubSub for Google Pub/Sub sink

164a18b

refactor: remove encode check for google pubsub sink

aeeb127

because we have already checked the encode in `CONNECTORS_COMPATIBLE_FORMATS` (src/frontend/src/handler/create_sink.rs)

refactor: use unwrap instead of expect

1a4df2c

refactor: use google-cloud-gax for authentication

94896a4

to avoid the need to set environment variables for authentication

jetjinser force-pushed the jinser/google-pubsub-sink branch from 14aeb03 to 94896a4 Compare April 29, 2024 13:35

tabVersion approved these changes May 2, 2024

View reviewed changes

stdrc approved these changes May 7, 2024

View reviewed changes

jetjinser added this pull request to the merge queue May 7, 2024

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

refactor: use context to add context to errors

00ca033

jetjinser added this pull request to the merge queue May 7, 2024

Merged via the queue into main with commit a4adc5e May 8, 2024
27 of 28 checks passed

jetjinser deleted the jinser/google-pubsub-sink branch May 8, 2024 00:30

jetjinser mentioned this pull request May 16, 2024

fix: remove unused r#type field in GooglePubSubConfig #16785

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(sink): add Google Pub/Sub support #16363

feat(sink): add Google Pub/Sub support #16363

jetjinser commented Apr 17, 2024 •

edited

Loading

xiangjinwu Apr 19, 2024

neverchanje commented Apr 24, 2024

stdrc Apr 29, 2024

stdrc Apr 29, 2024

stdrc Apr 29, 2024

stdrc Apr 29, 2024

tabVersion Apr 29, 2024

jetjinser Apr 29, 2024

stdrc Apr 29, 2024

tabVersion Apr 25, 2024

jetjinser Apr 29, 2024

tabVersion Apr 29, 2024

tabVersion Apr 29, 2024

tabVersion Apr 29, 2024

stdrc left a comment

WanYixian commented May 22, 2024

neverchanje commented May 22, 2024 •

edited

Loading

		google-cloud-googleapis = { version = "0.12", features = ["pubsub"] }
		google-cloud-pubsub = "0.24"

feat(sink): add Google Pub/Sub support #16363

feat(sink): add Google Pub/Sub support #16363

Conversation

jetjinser commented Apr 17, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

Choose a reason for hiding this comment

neverchanje commented Apr 24, 2024

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stdrc left a comment

Choose a reason for hiding this comment

WanYixian commented May 22, 2024

neverchanje commented May 22, 2024 • edited Loading

jetjinser commented Apr 17, 2024 •

edited

Loading

neverchanje commented May 22, 2024 •

edited

Loading