Merge branch 'main' of https://github.com/risingwavelabs/risingwave i…

…nto li0k/grafana_fix_lock_metrics
risingwavelabs · Apr 2, 2024 · 4c57200 · 4c57200
2 parents 6593b1c + 689a1f5
commit 4c57200
Show file tree

Hide file tree

Showing 16 changed files with 175 additions and 65 deletions.
diff --git a/.github/labeler.yml b/.github/labeler.yml
@@ -24,6 +24,10 @@ labels:
   title: "^doc.*"
 - label: "type/deprecate"
   title: "^deprecate.*"
+- label: "cherry-pick"
+  title: "^cherry-pick.*"
+- label: "cherry-pick"
+  title: "^cherry pick.*"
 
 - label: "breaking-change"
   body: '- \[x\] My PR contains breaking changes'
diff --git a/.github/pr-title-checker-config.json b/.github/pr-title-checker-config.json
@@ -4,7 +4,7 @@
     "color": "B60205"
   },
   "CHECKS": {
-    "regexp": "^(feat|fix|test|refactor|chore|style|doc|perf|build|ci|revert|deprecate)(\\(.*\\))?:.*",
+    "regexp": "^(feat|fix|test|refactor|chore|style|doc|perf|build|ci|revert|deprecate|cherry pick|cherry-pick)(\\(.*\\))?:.*",
     "ignoreLabels" : ["ignore-title"]
   },
   "MESSAGES": {

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/src/cmd_all/Cargo.toml b/src/cmd_all/Cargo.toml
@@ -37,7 +37,7 @@ risingwave_frontend = { workspace = true }
 risingwave_meta_node = { workspace = true }
 risingwave_rt = { workspace = true }
 shell-words = "1.1.0"
-strum = "0.25"
+strum = "0.26"
 strum_macros = "0.26"
 tempfile = "3"
 tikv-jemallocator = { workspace = true, features = [

diff --git a/src/connector/src/source/kafka/enumerator/client.rs b/src/connector/src/source/kafka/enumerator/client.rs
@@ -312,7 +312,8 @@ impl KafkaSplitEnumerator {
         for elem in offsets.elements_for_topic(self.topic.as_str()) {
             match elem.offset() {
                 Offset::Offset(offset) => {
-                    result.insert(elem.partition(), Some(offset));
+                    // XXX(rc): currently in RW source, `offset` means the last consumed offset, so we need to subtract 1
+                    result.insert(elem.partition(), Some(offset - 1));
                 }
                 _ => {
                     let (_, high_watermark) = self

diff --git a/src/frontend/src/session.rs b/src/frontend/src/session.rs
@@ -71,7 +71,6 @@ use risingwave_rpc_client::{ComputeClientPool, ComputeClientPoolRef, MetaClient}
 use risingwave_sqlparser::ast::{ObjectName, Statement};
 use risingwave_sqlparser::parser::Parser;
 use thiserror::Error;
-use thiserror_ext::AsReport;
 use tokio::runtime::Builder;
 use tokio::sync::oneshot::Sender;
 use tokio::sync::watch;
@@ -885,9 +884,7 @@ impl SessionImpl {
         formats: Vec<Format>,
     ) -> std::result::Result<PgResponse<PgResponseStream>, BoxedError> {
         // Parse sql.
-        let mut stmts = Parser::parse_sql(&sql).inspect_err(
-            |e| tracing::error!(error = %e.as_report(), %sql, "failed to parse sql"),
-        )?;
+        let mut stmts = Parser::parse_sql(&sql)?;
         if stmts.is_empty() {
             return Ok(PgResponse::empty_result(
                 pgwire::pg_response::StatementType::EMPTY,
@@ -901,9 +898,7 @@ impl SessionImpl {
             );
         }
         let stmt = stmts.swap_remove(0);
-        let rsp = handle(self, stmt, sql.clone(), formats).await.inspect_err(
-            |e| tracing::error!(error = %e.as_report(), %sql, "failed to handle sql"),
-        )?;
+        let rsp = handle(self, stmt, sql.clone(), formats).await?;
         Ok(rsp)
     }
 
@@ -1093,11 +1088,7 @@ impl Session for SessionImpl {
         let string = stmt.to_string();
         let sql_str = string.as_str();
         let sql: Arc<str> = Arc::from(sql_str);
-        let rsp = handle(self, stmt, sql.clone(), vec![format])
-            .await
-            .inspect_err(
-                |e| tracing::error!(error = %e.as_report(), %sql, "failed to handle sql"),
-            )?;
+        let rsp = handle(self, stmt, sql.clone(), vec![format]).await?;
         Ok(rsp)
     }
 
@@ -1140,9 +1131,7 @@ impl Session for SessionImpl {
         self: Arc<Self>,
         portal: Portal,
     ) -> std::result::Result<PgResponse<PgResponseStream>, BoxedError> {
-        let rsp = handle_execute(self, portal)
-            .await
-            .inspect_err(|e| tracing::error!(error=%e.as_report(), "failed to handle execute"))?;
+        let rsp = handle_execute(self, portal).await?;
         Ok(rsp)
     }
 

diff --git a/src/meta/Cargo.toml b/src/meta/Cargo.toml
@@ -65,7 +65,7 @@ scopeguard = "1.2.0"
 sea-orm = { workspace = true }
 serde = { version = "1.0.196", features = ["derive"] }
 serde_json = "1.0.113"
-strum = { version = "0.25", features = ["derive"] }
+strum = { version = "0.26", features = ["derive"] }
 sync-point = { path = "../utils/sync-point" }
 thiserror = "1"
 thiserror-ext = { workspace = true }

diff --git a/src/object_store/Cargo.toml b/src/object_store/Cargo.toml
@@ -30,7 +30,7 @@ madsim = "0.2.22"
 opendal = "0.45.1"
 prometheus = { version = "0.13", features = ["process"] }
 risingwave_common = { workspace = true }
-rustls = "0.22.3"
+rustls = "0.23.4"
 spin = "0.9"
 thiserror = "1"
 thiserror-ext = { workspace = true }