Merge branch 'main' into bz2/multi-meta-etcd

risingwavelabs · Nov 17, 2022 · c67bbdc · c67bbdc
2 parents 71c5197 + f88b680
commit c67bbdc
Show file tree

Hide file tree

Showing 165 changed files with 2,599 additions and 1,943 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -44,6 +44,24 @@ members = [
   "src/workspace-hack",
 ]
 
+[workspace.package]
+version = "0.2.0-alpha"
+edition = "2021"
+homepage = "https://github.com/risingwavelabs/risingwave"
+keywords = ["sql", "database", "streaming"]
+license = "Apache-2.0"
+repository = "https://github.com/risingwavelabs/risingwave"
+
+[workspace.dependencies]
+aws-config = { version = "0.51", default-features = false, features = ["rt-tokio", "native-tls"] }
+aws-sdk-kinesis = { version = "0.21", default-features = false, features = ["rt-tokio", "native-tls"] }
+aws-sdk-s3 = { version = "0.21", default-features = false, features = ["rt-tokio","native-tls"] }
+aws-sdk-sqs = { version = "0.21", default-features = false, features = ["rt-tokio", "native-tls"] }
+aws-smithy-http = "0.51"
+aws-smithy-types = "0.51"
+aws-endpoint = "0.51"
+aws-types = { version = "0.51", features = ["hardcoded-credentials"] }
+
 [profile.dev]
 lto = 'off'
 

diff --git a/Makefile.toml b/Makefile.toml
@@ -390,7 +390,7 @@ script = """
 
 set -e
 
-for tool in cargo-llvm-cov cargo-nextest cargo-udeps cargo-hakari cargo-sort cargo-make typos-cli
+for tool in cargo-llvm-cov cargo-nextest cargo-udeps cargo-hakari cargo-sort cargo-make cargo-upgrades typos-cli
 do
   echo "install: $(tput setaf 4)$tool$(tput sgr0)"
   cargo install $tool --locked

diff --git a/docs/relational_table/relational-table-schema.md b/docs/relational_table/relational-table-schema.md
@@ -7,7 +7,7 @@ In this doc, we will take HashAgg with extreme state (`max`, `min`) or value sta
 [Code](https://github.com/risingwavelabs/risingwave/blob/7f9ad2240712aa0cfe3edffb4535d43b42f32cc5/src/frontend/src/optimizer/plan_node/logical_agg.rs#L144)
 
 ## Table id
-For all relational table states, the keyspace must start with `table_id`. This is a globally unique id allocated in meta. Meta is responsible for traversing the Plan Tree and calculating the total number of Relational Tables needed. For example, the Hash Join Operator needs 2, one for the left table and one for the right table. The number of tables needed for Agg depends on the number of agg calls.
+`table_id` is a globally unique id allocated in meta for each relational table object. Meta is responsible for traversing the Plan Tree and calculating the total number of Relational Tables needed. For example, the Hash Join Operator needs 2, one for the left table and one for the right table. The number of tables needed for Agg depends on the number of agg calls.
 
 ## Value State (Sum, Count)
 Query example:

diff --git a/docs/state-store-overview.md b/docs/state-store-overview.md
@@ -28,11 +28,11 @@ Hummock consists of a manager service on the meta node, clients on worker nodes
 
 The streaming state store has distinguished workload characteristics.
 
-* Every streaming executor will only ***read and write its own portion of data***, which are multiple consecutive non-overlapping ranges of keys (we call it ***key space***).
+* Every streaming executor will only ***read and write its own portion of data***.
 * Data (generally) ***won’t be shared across nodes***, so every worker node will only read and write its own data. Therefore, every Hummock API, like `get` or `scan`, only guarantees that writes on one node can be immediately read from the same node. In some cases, if we want to read data written from other nodes, we will need to ***wait for the epoch***.
 * Streaming data are ***committed in serial***. Based on the [barrier-based checkpoint algorithm](https://en.wikipedia.org/wiki/Chandy%E2%80%93Lamport_algorithm), the states are persisted epoch by epoch. We can tailor the write path specifically for the epoch-based checkpoint workload.
 
-This leads to the design of Hummock, the cloud-native KV-based streaming state store. We’ll explain concepts like “epoch”, “key space” and “barrier” in the following chapters.
+This leads to the design of Hummock, the cloud-native KV-based streaming state store. We’ll explain concepts like “epoch” and “barrier” in the following chapters.
 
 ## The Hummock User API
 
@@ -119,8 +119,8 @@ For `scan`, we simply select by overlapping key range. For point get, we will fi
 Hummock implements the following iterators:
 - `BlockIterator`: iterates a block of an SSTable.
 - `SSTableIterator`: iterates an SSTable.
-- `ConcatIterator`: iterates SSTables with non-overlapping keyspaces.
-- `MergeIterator`: iterates SSTables with overlapping keyspaces.
+- `ConcatIterator`: iterates SSTables with non-overlapping key ranges.
+- `MergeIterator`: iterates SSTables with overlapping key ranges.
 - `UserIterator`: wraps internal iterators and outputs user key-value with epoch <= read epoch.
 
 [iterators source code](https://github.com/risingwavelabs/risingwave/tree/main/src/storage/src/hummock/iterator)

diff --git a/e2e_test/nexmark/create_sources.slt.part b/e2e_test/nexmark/create_sources.slt.part
@@ -1,5 +1,5 @@
 statement ok
-CREATE SOURCE person (
+CREATE MATERIALIZED SOURCE person (
     "id" BIGINT,
     "name" VARCHAR,
     "email_address" VARCHAR,
@@ -17,7 +17,7 @@ CREATE SOURCE person (
 ) ROW FORMAT JSON;
 
 statement ok
-CREATE SOURCE auction (
+CREATE MATERIALIZED SOURCE auction (
     "id" BIGINT,
     "item_name" VARCHAR,
     "description" VARCHAR,

diff --git a/proto/cdc_service.proto b/proto/cdc_service.proto
@@ -0,0 +1,47 @@
+syntax = "proto3";
+
+package cdc_service;
+
+option optimize_for = SPEED;
+
+// Notes: This proto needs to be self-contained
+message Status {
+  enum Code {
+    UNSPECIFIED = 0;
+    OK = 1;
+  }
+  Code code = 1;
+  string message = 2;
+}
+
+message DbConnectorProperties {
+  string database_host = 1;
+  string database_port = 2;
+  string database_user = 3;
+  string database_password = 4;
+  string database_name = 5;
+  string table_name = 6;
+  string partition = 7;
+  string start_offset = 8;
+  bool include_schema_events = 9;
+}
+
+message CdcMessage {
+  string payload = 1;
+  string partition = 2;
+  string offset = 3;
+}
+
+message GetEventStreamRequest {
+  uint64 source_id = 1;
+  DbConnectorProperties properties = 2;
+}
+
+message GetEventStreamResponse {
+  uint64 source_id = 1;
+  repeated CdcMessage events = 2;
+}
+
+service CdcService {
+  rpc GetEventStream(GetEventStreamRequest) returns (stream GetEventStreamResponse);
+}
diff --git a/proto/hummock.proto b/proto/hummock.proto
@@ -197,9 +197,12 @@ message UnpinSnapshotBeforeResponse {
   common.Status status = 1;
 }
 
+// When `right_exclusive=false`, it represents [left, right], of which both boundary are open. When `right_exclusive=true`,
+// it represents [left, right), of which right is exclusive.
 message KeyRange {
   bytes left = 1;
   bytes right = 2;
+  bool right_exclusive = 3;
 }
 
 message TableOption {

diff --git a/risedev.yml b/risedev.yml
@@ -577,6 +577,9 @@ template:
     # If `enable-tiered-cache` is true, hummock will use data directory as file cache.
     enable-tiered-cache: false
 
+    # RPC endpoint for source connector node
+    connector-source-endpoint: "127.0.0.1:61261"
+
     # Minio instances used by this compute node
     provide-minio: "minio*"
 

diff --git a/src/batch/Cargo.toml b/src/batch/Cargo.toml
@@ -1,7 +1,11 @@
 [package]
-edition = "2021"
 name = "risingwave_batch"
-version = "0.2.0-alpha"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+keywords = { workspace = true }
+license = { workspace = true }
+repository = { workspace = true }
 
 [dependencies]
 anyhow = "1"
@@ -11,7 +15,7 @@ async-stream = "0.3"
 async-trait = "0.1"
 byteorder = "1"
 bytes = "1"
-chrono = { version = "0.4", default-features = false, features = ["clock", "std"] }
+chrono = { version = "0.4.22", default-features = false, features = ["clock", "std"] }
 crc32fast = "1"
 either = "1"
 farmhash = "1"
@@ -60,10 +64,10 @@ url = "2"
 uuid = "1"
 
 [target.'cfg(not(madsim))'.dependencies]
-workspace-hack = { version = "0.2.0-alpha", path = "../workspace-hack" }
+workspace-hack = { path = "../workspace-hack" }
 
 [dev-dependencies]
-criterion = { version = "0.3.5", features = ["async_tokio", "async"] }
+criterion = { version = "0.4", features = ["async_tokio", "async"] }
 rand = "0.8"
 tempfile = "3"
 tikv-jemallocator = "0.5"

diff --git a/src/batch/src/executor/row_seq_scan.rs b/src/batch/src/executor/row_seq_scan.rs
@@ -405,136 +405,3 @@ impl<S: StateStore> RowSeqScanExecutor<S> {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::collections::Bound::Unbounded;
-
-    use futures::StreamExt;
-    use risingwave_common::catalog::{ColumnDesc, ColumnId, TableId, TableOption};
-    use risingwave_common::row::Row;
-    use risingwave_common::types::DataType;
-    use risingwave_common::util::epoch::EpochPair;
-    use risingwave_common::util::sort_util::OrderType;
-    use risingwave_storage::memory::MemoryStateStore;
-    use risingwave_storage::table::batch_table::storage_table::StorageTable;
-    use risingwave_storage::table::streaming_table::state_table::StateTable;
-    use risingwave_storage::table::Distribution;
-
-    use crate::executor::{Executor, RowSeqScanExecutor, ScanRange};
-
-    #[tokio::test]
-    async fn test_row_seq_scan() {
-        let state_store = MemoryStateStore::new();
-        let column_ids = vec![ColumnId::from(0), ColumnId::from(1), ColumnId::from(2)];
-        let column_descs = vec![
-            ColumnDesc::unnamed(column_ids[0], DataType::Int32),
-            ColumnDesc::unnamed(column_ids[1], DataType::Int32),
-            ColumnDesc::unnamed(column_ids[2], DataType::Int32),
-        ];
-        let pk_indices = vec![0_usize, 1_usize];
-        let order_types = vec![OrderType::Ascending, OrderType::Descending];
-        let mut state = StateTable::new_without_distribution(
-            state_store.clone(),
-            TableId::from(0x42),
-            column_descs.clone(),
-            order_types.clone(),
-            pk_indices.clone(),
-        )
-        .await;
-        let column_ids_partial = vec![ColumnId::from(1), ColumnId::from(2)];
-        let value_indices: Vec<usize> = vec![0, 1, 2];
-        let table = StorageTable::new_partial(
-            state_store.clone(),
-            TableId::from(0x42),
-            column_descs.clone(),
-            column_ids_partial,
-            order_types.clone(),
-            pk_indices,
-            Distribution::fallback(),
-            TableOption::default(),
-            value_indices,
-        );
-        let epoch = EpochPair::new_test_epoch(1);
-        state.init_epoch(epoch);
-        epoch.inc();
-
-        state.insert(Row(vec![
-            Some(1_i32.into()),
-            Some(11_i32.into()),
-            Some(111_i32.into()),
-        ]));
-        state.insert(Row(vec![
-            Some(2_i32.into()),
-            Some(22_i32.into()),
-            Some(222_i32.into()),
-        ]));
-        state.insert(Row(vec![
-            Some(3_i32.into()),
-            Some(33_i32.into()),
-            Some(333_i32.into()),
-        ]));
-
-        state.commit_for_test(epoch).await.unwrap();
-
-        let scan_range1 = ScanRange {
-            pk_prefix: Row(vec![Some(1_i32.into()), Some(11_i32.into())]),
-            next_col_bounds: (Unbounded, Unbounded),
-        };
-
-        let scan_range2 = ScanRange {
-            pk_prefix: Row(vec![Some(2_i32.into()), Some(22_i32.into())]),
-            next_col_bounds: (Unbounded, Unbounded),
-        };
-
-        let row_seq_scan_exec = RowSeqScanExecutor::new(
-            table.clone(),
-            vec![scan_range1, scan_range2],
-            epoch.curr,
-            1024,
-            "row_seq_scan_exec".to_string(),
-            None,
-        );
-
-        let point_get_row_seq_scan_exec = Box::new(row_seq_scan_exec);
-
-        let mut stream = point_get_row_seq_scan_exec.execute();
-        let chunk = stream.next().await.unwrap().unwrap();
-
-        assert_eq!(
-            chunk.row_at(0).0.to_owned_row(),
-            Row(vec![Some(11_i32.into()), Some(111_i32.into())])
-        );
-        assert_eq!(
-            chunk.row_at(1).0.to_owned_row(),
-            Row(vec![Some(22_i32.into()), Some(222_i32.into())])
-        );
-
-        let full_row_seq_scan_exec = RowSeqScanExecutor::new(
-            table,
-            vec![ScanRange::full()],
-            epoch.curr,
-            1024,
-            "row_seq_scan_exec".to_string(),
-            None,
-        );
-
-        let row_seq_scan_exec = Box::new(full_row_seq_scan_exec);
-
-        let mut stream = row_seq_scan_exec.execute();
-        let chunk = stream.next().await.unwrap().unwrap();
-
-        assert_eq!(
-            chunk.row_at(0).0.to_owned_row(),
-            Row(vec![Some(11_i32.into()), Some(111_i32.into())])
-        );
-        assert_eq!(
-            chunk.row_at(1).0.to_owned_row(),
-            Row(vec![Some(22_i32.into()), Some(222_i32.into())])
-        );
-        assert_eq!(
-            chunk.row_at(2).0.to_owned_row(),
-            Row(vec![Some(33_i32.into()), Some(333_i32.into())])
-        );
-    }
-}
diff --git a/src/batch/src/task/task_execution.rs b/src/batch/src/task/task_execution.rs
@@ -319,8 +319,7 @@ impl<C: BatchTaskContext> BatchTaskExecution<C> {
                     })
                     .await
                 {
-                    // Prints the entire backtrace of error.
-                    error!("Execution failed [{:?}]: {:?}", &task_id, &e);
+                    error!("Execution failed [{:?}]: {}", &task_id, e);
                     let err_str = e.to_string();
                     *failure.lock() = Some(e);
                     if let Err(_e) = t_1

diff --git a/src/bench/Cargo.toml b/src/bench/Cargo.toml
@@ -1,14 +1,17 @@
 [package]
 name = "risingwave_bench"
-version = "0.2.0-alpha"
-edition = "2021"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+keywords = { workspace = true }
+license = { workspace = true }
+repository = { workspace = true }
 
 [dependencies]
 async-trait = "0.1"
-aws-config = { version = "0.49", default-features = false, features = ["rt-tokio", "native-tls"] }
-aws-creds = "0.34"
-aws-sdk-s3 = { version = "0.19", default-features = false, features = ["rt-tokio", "native-tls"] }
-aws-smithy-http = "0.49"
+aws-config = { workspace = true }
+aws-sdk-s3 = { workspace = true }
+aws-smithy-http = { workspace = true }
 bcc = { version = "0.0.33", optional = true }
 bytes = "1"
 bytesize = { version = "1", features = ["serde"] }
@@ -21,7 +24,7 @@ libc = "0.2"
 minitrace = "0.4"
 minitrace-jaeger = "0.4"
 moka = { version = "0.9", features = ["future"] }
-nix = { version = "0.24.1", features = ["fs", "mman"] }
+nix = { version = "0.25", features = ["fs", "mman"] }
 opentelemetry = { version = "0.17", optional = true, features = ["rt-tokio"] }
 opentelemetry-jaeger = { version = "0.16", optional = true, features = [
     "rt-tokio",
@@ -55,7 +58,7 @@ tracing-opentelemetry = { version = "0.17", optional = true }
 tracing-subscriber = "0.3.16"
 
 [target.'cfg(not(madsim))'.dependencies]
-workspace-hack = { version = "0.2.0-alpha", path = "../workspace-hack" }
+workspace-hack = { path = "../workspace-hack" }
 
 [[bin]]
 name = "file-cache-bench"