Add new packages, dependencies to Cargo.lock, and modify Cargo.toml.

src/meta/Cargo.toml

@@ -56,6 +56,7 @@ risingwave_sqlparser = { workspace = true }
 scopeguard = "1.2.0"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+sqlx = { version = "0.7", features = ["runtime-tokio", "postgres", "mysql", "sqlite", "chrono"] }
 sync-point = { path = "../utils/sync-point" }
 thiserror = "1"
 tokio = { version = "0.2", package = "madsim-tokio", features = [

src/meta/src/error.rs

@@ -20,6 +20,7 @@ use risingwave_common::error::BoxedError;
 use risingwave_connector::sink::SinkError;
 use risingwave_pb::PbFieldNotFound;
 use risingwave_rpc_client::error::RpcError;
+use sqlx::Error;
 
 use crate::hummock::error::Error as HummockError;
 use crate::manager::WorkerId;
@@ -62,7 +63,7 @@ enum MetaErrorInner {
     Unavailable(String),
 
     #[error("Election failed: {0}")]
-    Election(etcd_client::Error),
+    Election(String),
 
     #[error("Cancelled: {0}")]
     Cancelled(String),
@@ -73,7 +74,7 @@ enum MetaErrorInner {
     #[error("Sink error: {0}")]
     Sink(SinkError),
 
-    #[error("AWS SDK error: {}", DisplayErrorContext(&**.0))]
+    #[error("AWS SDK error: {}", DisplayErrorContext(& * *.0))]
     Aws(BoxedError),
 
     #[error(transparent)]
@@ -165,7 +166,13 @@ impl From<HummockError> for MetaError {
 
 impl From<etcd_client::Error> for MetaError {
     fn from(e: etcd_client::Error) -> Self {
-        MetaErrorInner::Election(e).into()
+        MetaErrorInner::Election(e.to_string()).into()
+    }
+}
+
+impl From<sqlx::Error> for MetaError {
+    fn from(value: Error) -> Self {
+        MetaErrorInner::Election(value.to_string()).into()
     }
 }
 

src/meta/src/rpc/election_client.rs → src/meta/src/rpc/election/etcd.rs

@@ -18,33 +18,15 @@ use std::time::Duration;
 
 use etcd_client::{ConnectOptions, Error, GetOptions, LeaderKey, ResignOptions};
 use risingwave_common::bail;
-use serde::Serialize;
 use tokio::sync::watch::Receiver;
 use tokio::sync::{oneshot, watch};
 use tokio::time;
 use tokio_stream::StreamExt;
 
+use crate::rpc::election::{ElectionClient, ElectionMember, META_ELECTION_KEY};
 use crate::storage::WrappedEtcdClient;
 use crate::MetaResult;
 
-const META_ELECTION_KEY: &str = "__meta_election_";
-
-#[derive(Debug, Serialize)]
-pub struct ElectionMember {
-    pub id: String,
-    pub is_leader: bool,
-}
-
-#[async_trait::async_trait]
-pub trait ElectionClient: Send + Sync + 'static {
-    fn id(&self) -> MetaResult<String>;
-    async fn run_once(&self, ttl: i64, stop: Receiver<()>) -> MetaResult<()>;
-    fn subscribe(&self) -> Receiver<bool>;
-    async fn leader(&self) -> MetaResult<Option<ElectionMember>>;
-    async fn get_members(&self) -> MetaResult<Vec<ElectionMember>>;
-    async fn is_leader(&self) -> bool;
-}
-
 pub struct EtcdElectionClient {
     id: String,
     is_leader_sender: watch::Sender<bool>,

src/meta/src/rpc/election/mod.rs

@@ -0,0 +1,38 @@
+// Copyright 2023 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 etcd;
+pub mod sql;
+
+use serde::Serialize;
+use tokio::sync::watch::Receiver;
+
+use crate::MetaResult;
+
+const META_ELECTION_KEY: &str = "__meta_election_";
+
+#[derive(Debug, Serialize)]
+pub struct ElectionMember {
+    pub id: String,
+    pub is_leader: bool,
+}
+
+#[async_trait::async_trait]
+pub trait ElectionClient: Send + Sync + 'static {
+    fn id(&self) -> MetaResult<String>;
+    async fn run_once(&self, ttl: i64, stop: Receiver<()>) -> MetaResult<()>;
+    fn subscribe(&self) -> Receiver<bool>;
+    async fn leader(&self) -> MetaResult<Option<ElectionMember>>;
+    async fn get_members(&self) -> MetaResult<Vec<ElectionMember>>;
+    async fn is_leader(&self) -> bool;
+}