test(stream): add ci for split change mutation in source (risingwavel…

…abs#3039) * stage Signed-off-by: tabVersion <[email protected]> * stage Signed-off-by: tabVersion <[email protected]> * add test Signed-off-by: tabVersion <[email protected]> * change e2e to datagen Signed-off-by: tabVersion <[email protected]> * stage Signed-off-by: tabVersion <[email protected]> * some bug to fix Signed-off-by: tabVersion <[email protected]> * fix async issue Signed-off-by: tabVersion <[email protected]> * add assert Signed-off-by: tabVersion <[email protected]>
Little-Wallace · Jun 20, 2022 · 5d2bb42 · 5d2bb42
1 parent 04fe6d6
commit 5d2bb42
Show file tree

Hide file tree

Showing 8 changed files with 255 additions and 73 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/src/compute/Cargo.toml b/src/compute/Cargo.toml
@@ -21,6 +21,7 @@ itertools = "0.10"
 lazy_static = "1"
 log = "0.4"
 madsim = "=0.2.0-alpha.3"
+maplit = "1.0.2"
 memcomparable = { path = "../utils/memcomparable" }
 num-traits = "0.2"
 paste = "1"
@@ -29,6 +30,7 @@ prost = "0.10"
 rdkafka = { version = "0.28", features = ["cmake-build"] }
 risingwave_batch = { path = "../batch" }
 risingwave_common = { path = "../common" }
+risingwave_connector = { path = "../connector" }
 risingwave_pb = { path = "../prost" }
 risingwave_rpc_client = { path = "../rpc_client" }
 risingwave_source = { path = "../source" }

diff --git a/src/connector/src/datagen/source/reader.rs b/src/connector/src/datagen/source/reader.rs
@@ -104,10 +104,20 @@ impl SplitReader for DatagenSplitReader {
                 .get(&random_seed_key)
                 .map(|s| s.to_string())
             {
-                Some(seed) => seed.parse::<u64>().unwrap_or(split_index),
+                Some(seed) => {
+                    match seed.parse::<u64>() {
+                        // we use given seed xor split_index to make sure every split has different
+                        // seed
+                        Ok(seed) => seed ^ split_index,
+                        Err(e) => {
+                            log::warn!("cannot parse {:?} to u64 due to {:?}, will use {:?} as random seed", seed, e, split_index);
+                            split_index
+                        }
+                    }
+                }
                 None => split_index,
             };
-            match column.data_type{
+            match column.data_type {
                 DataType::Timestamp => {
                 let max_past_key = format!("fields.{}.max_past", name);
                 let max_past_value =

diff --git a/src/connector/src/datagen/split.rs b/src/connector/src/datagen/split.rs
@@ -20,9 +20,9 @@ use crate::base::SplitMetaData;
 
 #[derive(Clone, Serialize, Deserialize, Debug, Default, PartialEq, Hash)]
 pub struct DatagenSplit {
-    pub(crate) split_index: i32,
-    pub(crate) split_num: i32,
-    pub(crate) start_offset: Option<u64>,
+    pub split_index: i32,
+    pub split_num: i32,
+    pub start_offset: Option<u64>,
 }
 
 impl SplitMetaData for DatagenSplit {

diff --git a/src/connector/src/lib.rs b/src/connector/src/lib.rs
@@ -37,9 +37,9 @@
 extern crate core;
 
 pub mod base;
-mod datagen;
+pub mod datagen;
 mod filesystem;
-mod kafka;
+pub mod kafka;
 pub mod kinesis;
 mod nexmark;
 mod pulsar;

diff --git a/src/stream/Cargo.toml b/src/stream/Cargo.toml
@@ -24,6 +24,7 @@ itertools = "0.10"
 lazy_static = "1"
 log = "0.4"
 madsim = "=0.2.0-alpha.3"
+maplit = "1.0.2"
 memcomparable = { path = "../utils/memcomparable" }
 num-traits = "0.2"
 parking_lot = "0.12"

diff --git a/src/stream/src/executor/mod.rs b/src/stream/src/executor/mod.rs
@@ -167,8 +167,8 @@ pub trait ExprFn = Fn(&DataChunk) -> Result<Bitmap> + Send + Sync + 'static;
 
 #[derive(Debug, PartialEq, Clone, Default)]
 pub struct AddOutput {
-    map: HashMap<(ActorId, DispatcherId), Vec<ActorInfo>>,
-    splits: HashMap<ActorId, Vec<SplitImpl>>,
+    pub map: HashMap<(ActorId, DispatcherId), Vec<ActorInfo>>,
+    pub splits: HashMap<ActorId, Vec<SplitImpl>>,
 }
 
 #[derive(Debug, Clone, PartialEq)]