refactor(source): move protobuf to codec crate, and refactor tests (#…

…18507) Signed-off-by: xxchan <[email protected]>
risingwavelabs · Sep 17, 2024 · 7354d30 · 7354d30
1 parent 4c1649b
commit 7354d30
Show file tree

Hide file tree

Showing 36 changed files with 1,238 additions and 1,147 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/ci/scripts/e2e-source-test.sh b/ci/scripts/e2e-source-test.sh
@@ -45,7 +45,6 @@ risedev ci-kill
 echo "--- Prepare data"
 cp src/connector/src/test_data/simple-schema.avsc ./avro-simple-schema.avsc
 cp src/connector/src/test_data/complex-schema.avsc ./avro-complex-schema.avsc
-cp src/connector/src/test_data/complex-schema ./proto-complex-schema
 cp src/connector/src/test_data/complex-schema.json ./json-complex-schema
 
 

diff --git a/e2e_test/sink/kafka/protobuf.slt b/e2e_test/sink/kafka/protobuf.slt
@@ -1,14 +1,17 @@
 statement ok
 set sink_decouple = false;
 
+system ok
+rpk topic create test-rw-sink-append-only-protobuf
+
 statement ok
 create table from_kafka with (
   connector = 'kafka',
   topic = 'test-rw-sink-append-only-protobuf',
   properties.bootstrap.server = 'message_queue:29092')
 format plain encode protobuf (
-  schema.location = 'file:///risingwave/proto-recursive',
-  message = 'recursive.AllTypes');
+  schema.location = 'file:///risingwave/src/connector/codec/tests/test_data/all-types.pb',
+  message = 'all_types.AllTypes');
 
 statement ok
 create table from_kafka_csr_trivial with (
@@ -73,8 +76,8 @@ create sink sink0 from into_kafka with (
   properties.bootstrap.server = 'message_queue:29092')
 format plain encode protobuf (
   force_append_only = true,
-  schema.location = 'file:///risingwave/proto-recursive',
-  message = 'recursive.AllTypes');
+  schema.location = 'file:///risingwave/src/connector/codec/tests/test_data/all-types.pb',
+  message = 'all_types.AllTypes');
 
 statement ok
 create sink sink_csr_trivial as select string_field as field_a from into_kafka with (
@@ -103,8 +106,8 @@ create sink sink_upsert from into_kafka with (
   properties.bootstrap.server = 'message_queue:29092',
   primary_key = 'string_field')
 format upsert encode protobuf (
-  schema.location = 'file:///risingwave/proto-recursive',
-  message = 'recursive.AllTypes');
+  schema.location = 'file:///risingwave/src/connector/codec/tests/test_data/all-types.pb',
+  message = 'all_types.AllTypes');
 ----
 db error: ERROR: Failed to run the query
 
@@ -122,8 +125,8 @@ create sink sink_upsert from into_kafka with (
   properties.bootstrap.server = 'message_queue:29092',
   primary_key = 'string_field')
 format upsert encode protobuf (
-  schema.location = 'file:///risingwave/proto-recursive',
-  message = 'recursive.AllTypes')
+  schema.location = 'file:///risingwave/src/connector/codec/tests/test_data/all-types.pb',
+  message = 'all_types.AllTypes')
 key encode text;
 
 # Shall be ignored by force_append_only sinks but processed by upsert sinks.
@@ -178,7 +181,7 @@ create sink sink_err from into_kafka with (
 format plain encode protobuf (
   force_append_only = true,
   schema.location = 'file:///risingwave/proto-recursiv',
-  message = 'recursive.AllTypes');
+  message = 'all_types.AllTypes');
 
 statement error field not in proto
 create sink sink_err as select 1 as extra_column with (
@@ -187,8 +190,8 @@ create sink sink_err as select 1 as extra_column with (
   properties.bootstrap.server = 'message_queue:29092')
 format plain encode protobuf (
   force_append_only = true,
-  schema.location = 'file:///risingwave/proto-recursive',
-  message = 'recursive.AllTypes');
+  schema.location = 'file:///risingwave/src/connector/codec/tests/test_data/all-types.pb',
+  message = 'all_types.AllTypes');
 
 statement error s3 URL not supported yet
 create sink sink_err from into_kafka with (
@@ -197,8 +200,8 @@ create sink sink_err from into_kafka with (
   properties.bootstrap.server = 'message_queue:29092')
 format plain encode protobuf (
   force_append_only = true,
-  schema.location = 's3:///risingwave/proto-recursive',
-  message = 'recursive.AllTypes');
+  schema.location = 's3:///risingwave/src/connector/codec/tests/test_data/all-types.pb',
+  message = 'all_types.AllTypes');
 
 statement ok
 drop table from_kafka cascade;

diff --git a/e2e_test/source/basic/kafka.slt b/e2e_test/source/basic/kafka.slt
@@ -187,17 +187,6 @@ create table s10 with (
   scan.startup.mode = 'earliest'
 ) FORMAT PLAIN ENCODE AVRO (schema.location = 'file:///risingwave/avro-complex-schema.avsc', with_deprecated_file_header = true);
 
-statement ok
-create table s11 with (
-  connector = 'kafka',
-  topic = 'proto_c_bin',
-  properties.bootstrap.server = 'message_queue:29092',
-  scan.startup.mode = 'earliest')
-FORMAT PLAIN ENCODE PROTOBUF (
-  message = 'test.User',
-  schema.location = 'file:///risingwave/proto-complex-schema'
-);
-
 statement ok
 CREATE TABLE s12(
     id int,
@@ -273,17 +262,6 @@ create table s16 (v1 int, v2 varchar) with (
   scan.startup.mode = 'latest'
 ) FORMAT PLAIN ENCODE JSON
 
-statement ok
-create source s17 with (
-  connector = 'kafka',
-  topic = 'proto_c_bin',
-  properties.bootstrap.server = 'message_queue:29092',
-  scan.startup.mode = 'earliest')
-FORMAT PLAIN ENCODE PROTOBUF (
-  message = 'test.User',
-  schema.location = 'file:///risingwave/proto-complex-schema'
-);
-
 statement ok
 create source s18 with (
   connector = 'kafka',
@@ -669,11 +647,6 @@ select id, code, timestamp, xfas, contacts, sex from s10;
 ----
 100 abc 1473305798 {"(0,200,10.0.0.1)","(1,400,10.0.0.2)"} ("{1xxx,2xxx}","{1xxx,2xxx}") MALE
 
-query ITITT
-select id, code, timestamp, xfas, contacts, sex from s11;
-----
-0 abc 1473305798 {"(0,200,127.0.0.1)","(1,400,127.0.0.2)"} ("{1xxx,2xxx}","{1xxx,2xxx}") MALE
-
 query ITITT
 select id, code, timestamp, xfas, contacts, jsonb from s12;
 ----
@@ -703,9 +676,6 @@ select count(*) from s16
 statement error Not supported: alter source with schema registry
 alter source s18 add column v10 int;
 
-statement error Not supported: alter source with schema registry
-alter source s17 add column v10 int;
-
 query III rowsort
 select * from s21;
 ----
@@ -848,9 +818,6 @@ drop table s9
 statement ok
 drop table s10
 
-statement ok
-drop table s11
-
 statement ok
 drop table s12
 
@@ -866,9 +833,6 @@ drop table s15
 statement ok
 drop table s16
 
-statement ok
-drop source s17
-
 statement ok
 drop source s18
 

diff --git a/e2e_test/source/basic/old_row_format_syntax/kafka.slt b/e2e_test/source/basic/old_row_format_syntax/kafka.slt
@@ -171,14 +171,6 @@ create table s10 with (
   scan.startup.mode = 'earliest'
 ) row format avro row schema location 'file:///risingwave/avro-complex-schema.avsc'
 
-statement ok
-create table s11 with (
-  connector = 'kafka',
-  topic = 'proto_c_bin',
-  properties.bootstrap.server = 'message_queue:29092',
-  scan.startup.mode = 'earliest'
-) row format protobuf message 'test.User' row schema location 'file:///risingwave/proto-complex-schema'
-
 statement ok
 CREATE TABLE s12(
     id int,
@@ -254,14 +246,6 @@ create table s16 (v1 int, v2 varchar) with (
   scan.startup.mode = 'latest'
 ) ROW FORMAT JSON
 
-statement ok
-create source s17 with (
-  connector = 'kafka',
-  topic = 'proto_c_bin',
-  properties.bootstrap.server = 'message_queue:29092',
-  scan.startup.mode = 'earliest'
-) row format protobuf message 'test.User' row schema location 'file:///risingwave/proto-complex-schema'
-
 statement error without schema registry
 create source s18 with (
   connector = 'kafka',
@@ -570,11 +554,6 @@ select id, first_name, last_name, email from s8_no_schema_field;
 # ----
 # 100 abc 1473305798 {"(0,200,10.0.0.1)","(1,400,10.0.0.2)"} ("{1xxx,2xxx}","{1xxx,2xxx}") MALE
 
-query ITITT
-select id, code, timestamp, xfas, contacts, sex from s11;
-----
-0 abc 1473305798 {"(0,200,127.0.0.1)","(1,400,127.0.0.2)"} ("{1xxx,2xxx}","{1xxx,2xxx}") MALE
-
 query ITITT
 select id, code, timestamp, xfas, contacts, jsonb from s12;
 ----
@@ -712,9 +691,6 @@ drop table s8_no_schema_field
 # statement ok
 # drop table s10
 
-statement ok
-drop table s11
-
 statement ok
 drop table s12
 
@@ -730,9 +706,6 @@ drop table s15
 statement ok
 drop table s16
 
-statement ok
-drop source s17
-
 # statement ok
 # drop source s18
 

diff --git a/src/connector/Cargo.toml b/src/connector/Cargo.toml
@@ -103,7 +103,6 @@ prometheus = { version = "0.13", features = ["process"] }
 prost = { workspace = true, features = ["no-recursion-limit"] }
 prost-reflect = { version = "0.14", features = ["serde"] }
 prost-types = "0.13"
-protobuf-native = "0.2.2"
 pulsar = { version = "6.3", default-features = false, features = [
     "tokio-runtime",
     "telemetry",
@@ -192,6 +191,7 @@ assert_matches = "1"
 criterion = { workspace = true, features = ["async_tokio", "async"] }
 deltalake = { workspace = true, features = ["datafusion"] }
 expect-test = "1"
+fs-err = "2"
 paste = "1"
 pretty_assertions = "1"
 quote = "1"
@@ -204,10 +204,6 @@ tracing-subscriber = "0.3"
 tracing-test = "0.2"
 walkdir = "2"
 
-[build-dependencies]
-prost-build = "0.12"
-protobuf-src = "1"
-
 [[bench]]
 name = "debezium_json_parser"
 harness = false

diff --git a/src/connector/codec/Cargo.toml b/src/connector/codec/Cargo.toml
@@ -26,6 +26,10 @@ itertools = { workspace = true }
 jsonbb = { workspace = true }
 jst = { package = 'jsonschema-transpiler', git = "https://github.com/mozilla/jsonschema-transpiler", rev = "c1a89d720d118843d8bcca51084deb0ed223e4b4" }
 num-bigint = "0.4"
+prost = { workspace = true, features = ["no-recursion-limit"] }
+prost-reflect = { version = "0.14", features = ["serde"] }
+prost-types = "0.13"
+protobuf-native = "0.2.2"
 risingwave_common = { workspace = true }
 risingwave_pb = { workspace = true }
 rust_decimal = "1"
@@ -37,7 +41,13 @@ tracing = "0.1"
 
 [dev-dependencies]
 expect-test = "1"
+fs-err = "2"
 hex = "0.4"
+tokio = { version = "0.2", package = "madsim-tokio" }
+
+[build-dependencies]
+prost-build = "0.12"
+protobuf-src = "1"
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../../workspace-hack" }

diff --git a/src/connector/build.rs → src/connector/codec/build.rs b/src/connector/build.rs → src/connector/codec/build.rs
@@ -13,17 +13,17 @@
 // limitations under the License.
 
 fn main() {
-    let proto_dir = "./src/test_data/proto_recursive";
+    let proto_dir = "./tests/test_data/";
 
     println!("cargo:rerun-if-changed={}", proto_dir);
 
-    let proto_files = ["recursive"];
+    let proto_files = ["recursive", "all-types"];
     let protos: Vec<String> = proto_files
         .iter()
         .map(|f| format!("{}/{}.proto", proto_dir, f))
         .collect();
     prost_build::Config::new()
-        .out_dir("./src/parser/protobuf")
+        .out_dir("./tests/integration_tests/protobuf")
         .compile_protos(&protos, &Vec::<String>::new())
         .unwrap();
 

diff --git a/src/connector/codec/src/common/mod.rs b/src/connector/codec/src/common/mod.rs
@@ -0,0 +1,15 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod protobuf;