ISSUE-883 continue cluster management with namespace

common/clickhouse-srv/src/protocols/protocol_query.rs

@@ -106,6 +106,7 @@ impl QueryClientInfo {
     }
 }
 
+#[allow(dead_code)]
 #[derive(Default, Debug)]
 pub struct QueryRequest {
     pub(crate) query_id: String,

common/clickhouse-srv/src/types/mod.rs

@@ -117,6 +117,7 @@ impl fmt::Debug for ServerInfo {
 }
 
 #[derive(Clone)]
+#[allow(dead_code)]
 pub(crate) struct Context {
     pub(crate) server_info: ServerInfo,
     pub(crate) hostname: String,

common/exception/src/exception.rs

@@ -77,6 +77,15 @@ impl ErrorCode {
         }
     }
 
+    pub fn add_message_back(self, msg: impl AsRef<str>) -> Self {
+        Self {
+            code: self.code(),
+            display_text: format!("{}{}", self.display_text, msg.as_ref()),
+            cause: self.cause,
+            backtrace: self.backtrace,
+        }
+    }
+
     pub fn backtrace(&self) -> Option<ErrorCodeBacktrace> {
         self.backtrace.clone()
     }
@@ -168,6 +177,7 @@ build_exceptions! {
     AuthenticateFailure(51),
     TLSConfigurationFailure(52),
     UnknownSession(53),
+    UnexpectedError(54),
 
     // uncategorized
     UnexpectedResponseType(600),

common/functions/src/aggregates/aggregate_avg.rs

@@ -59,7 +59,7 @@ where T: std::ops::Add<Output = T> + DFPrimitiveType
 #[derive(Clone)]
 pub struct AggregateAvgFunction<T, SumT> {
     display_name: String,
-    arguments: Vec<DataField>,
+    _arguments: Vec<DataField>,
     t: PhantomData<T>,
     sum_t: PhantomData<SumT>,
 }
@@ -170,7 +170,7 @@ where
     ) -> Result<AggregateFunctionRef> {
         Ok(Arc::new(Self {
             display_name: display_name.to_string(),
-            arguments,
+            _arguments: arguments,
             t: PhantomData,
             sum_t: PhantomData,
         }))

common/functions/src/aggregates/aggregate_stddev_pop.rs

@@ -71,7 +71,7 @@ impl AggregateStddevPopState {
 #[derive(Clone)]
 pub struct AggregateStddevPopFunction<T> {
     display_name: String,
-    arguments: Vec<DataField>,
+    _arguments: Vec<DataField>,
     t: PhantomData<T>,
 }
 
@@ -202,7 +202,7 @@ where T: DFPrimitiveType + AsPrimitive<f64>
     ) -> Result<AggregateFunctionRef> {
         Ok(Arc::new(Self {
             display_name: display_name.to_string(),
-            arguments,
+            _arguments: arguments,
             t: PhantomData,
         }))
     }

common/functions/src/aggregates/aggregate_sum.rs

@@ -61,7 +61,7 @@ where
 #[derive(Clone)]
 pub struct AggregateSumFunction<T, SumT> {
     display_name: String,
-    arguments: Vec<DataField>,
+    _arguments: Vec<DataField>,
     t: PhantomData<T>,
     sum_t: PhantomData<SumT>,
 }
@@ -185,7 +185,7 @@ where
     ) -> Result<AggregateFunctionRef> {
         Ok(Arc::new(Self {
             display_name: display_name.to_owned(),
-            arguments,
+            _arguments: arguments,
             t: PhantomData,
             sum_t: PhantomData,
         }))

common/functions/src/aggregates/aggregate_window_funnel.rs

@@ -162,7 +162,7 @@ where T: Ord
 #[derive(Clone)]
 pub struct AggregateWindowFunnelFunction<T> {
     display_name: String,
-    arguments: Vec<DataField>,
+    _arguments: Vec<DataField>,
     event_size: usize,
     window: u64,
     t: PhantomData<T>,
@@ -300,7 +300,7 @@ where
         let window = params[0].as_u64()?;
         Ok(Arc::new(Self {
             display_name: display_name.to_owned(),
-            arguments,
+            _arguments: arguments,
             event_size,
             window,
             t: PhantomData,

common/functions/src/scalars/conditionals/if.rs

@@ -24,13 +24,13 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct IfFunction {
-    display_name: String,
+    _display_name: String,
 }
 
 impl IfFunction {
     pub fn try_create_func(display_name: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(IfFunction {
-            display_name: display_name.to_string(),
+            _display_name: display_name.to_string(),
         }))
     }
 }

common/functions/src/scalars/expressions/cast.rs

@@ -24,15 +24,15 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct CastFunction {
-    display_name: String,
+    _display_name: String,
     /// The data type to cast to
     cast_type: DataType,
 }
 
 impl CastFunction {
     pub fn create(display_name: String, cast_type: DataType) -> Result<Box<dyn Function>> {
         Ok(Box::new(Self {
-            display_name,
+            _display_name: display_name,
             cast_type,
         }))
     }

common/functions/src/scalars/function_column.rs

@@ -26,14 +26,14 @@ use crate::scalars::Function;
 #[derive(Clone, Debug)]
 pub struct ColumnFunction {
     value: String,
-    saved: Option<DataValue>,
+    _saved: Option<DataValue>,
 }
 
 impl ColumnFunction {
     pub fn try_create(value: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(ColumnFunction {
             value: value.to_string(),
-            saved: None,
+            _saved: None,
         }))
     }
 }

common/functions/src/scalars/nullables/is_not_null.rs

@@ -24,13 +24,13 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct IsNotNullFunction {
-    display_name: String,
+    _display_name: String,
 }
 
 impl IsNotNullFunction {
     pub fn try_create_func(display_name: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(IsNotNullFunction {
-            display_name: display_name.to_string(),
+            _display_name: display_name.to_string(),
         }))
     }
 }

common/functions/src/scalars/nullables/is_null.rs

@@ -24,13 +24,13 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct IsNullFunction {
-    display_name: String,
+    _display_name: String,
 }
 
 impl IsNullFunction {
     pub fn try_create_func(display_name: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(IsNullFunction {
-            display_name: display_name.to_string(),
+            _display_name: display_name.to_string(),
         }))
     }
 }

common/functions/src/scalars/strings/substring.rs

@@ -24,13 +24,13 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct SubstringFunction {
-    display_name: String,
+    _display_name: String,
 }
 
 impl SubstringFunction {
     pub fn try_create(display_name: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(SubstringFunction {
-            display_name: display_name.to_string(),
+            _display_name: display_name.to_string(),
         }))
     }
 }

common/functions/src/scalars/udfs/crash_me.rs

@@ -24,13 +24,13 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct CrashMeFunction {
-    display_name: String,
+    _display_name: String,
 }
 
 impl CrashMeFunction {
     pub fn try_create(display_name: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(CrashMeFunction {
-            display_name: display_name.to_string(),
+            _display_name: display_name.to_string(),
         }))
     }
 }

common/functions/src/scalars/udfs/to_type_name.rs

@@ -25,13 +25,13 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct ToTypeNameFunction {
-    display_name: String,
+    _display_name: String,
 }
 
 impl ToTypeNameFunction {
     pub fn try_create(display_name: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(ToTypeNameFunction {
-            display_name: display_name.to_string(),
+            _display_name: display_name.to_string(),
         }))
     }
 }

common/functions/src/scalars/udfs/version.rs

@@ -24,13 +24,13 @@ use crate::scalars::Function;
 
 #[derive(Clone)]
 pub struct VersionFunction {
-    display_name: String,
+    _display_name: String,
 }
 
 impl VersionFunction {
     pub fn try_create(display_name: &str) -> Result<Box<dyn Function>> {
         Ok(Box::new(VersionFunction {
-            display_name: display_name.to_string(),
+            _display_name: display_name.to_string(),
         }))
     }
 }

common/management/Cargo.toml

@@ -24,6 +24,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sha2 = "0.9.8"
 sha1 = "0.6.0"
+tempfile = "3.2.0"
 
 [dev-dependencies]
 tempfile = "3.2.0"

common/management/src/lib.rs

@@ -16,6 +16,9 @@
 mod namespace;
 mod user;
 
+pub use namespace::NamespaceApi;
+pub use namespace::NamespaceMgr;
+pub use namespace::NodeInfo;
 pub use user::user_api::AuthType;
 pub use user::user_api::UserInfo;
 pub use user::user_api::UserMgrApi;

common/management/src/namespace/mod.rs

@@ -13,11 +13,12 @@
 // limitations under the License.
 //
 
-pub use namespace_api::NamespaceApi;
-pub use namespace_api::NodeInfo;
-
 #[cfg(test)]
 mod namespace_mgr_test;
 
 mod namespace_api;
 mod namespace_mgr;
+
+pub use namespace_api::NamespaceApi;
+pub use namespace_api::NodeInfo;
+pub use namespace_mgr::NamespaceMgr;

common/management/src/namespace/namespace_api.rs

@@ -17,20 +17,17 @@ use std::convert::TryFrom;
 
 use common_exception::ErrorCode;
 use common_exception::Result;
-use common_metatypes::SeqValue;
 
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
 pub struct NodeInfo {
     #[serde(default)]
     pub id: String,
     #[serde(default)]
-    pub cpu_nums: u32,
+    pub cpu_nums: u64,
     #[serde(default)]
     pub version: u32,
     #[serde(default)]
-    pub ip: String,
-    #[serde(default)]
-    pub port: u32,
+    pub flight_address: String,
 }
 
 impl TryFrom<Vec<u8>> for NodeInfo {
@@ -47,33 +44,28 @@ impl TryFrom<Vec<u8>> for NodeInfo {
     }
 }
 
-pub trait NamespaceApi {
+impl NodeInfo {
+    pub fn create(id: String, cpu_nums: u64, flight_address: String) -> NodeInfo {
+        NodeInfo {
+            id,
+            cpu_nums,
+            version: 0,
+            flight_address,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+pub trait NamespaceApi: Sync + Send {
     // Add a new node info to /tenant/namespace/node-name.
-    fn add_node(&self, tenant_id: String, namespace_id: String, node: NodeInfo) -> Result<u64>;
+    async fn add_node(&self, node: NodeInfo) -> Result<u64>;
 
     // Get the tenant's namespace all nodes.
-    fn get_nodes(
-        &self,
-        tenant_id: String,
-        namespace_id: String,
-        seq: Option<u64>,
-    ) -> Result<Vec<SeqValue<NodeInfo>>>;
-
-    // Update the tenant's namespace node.
-    fn update_node(
-        &self,
-        tenant_id: String,
-        namespace_id: String,
-        node: NodeInfo,
-        seq: Option<u64>,
-    ) -> Result<Option<u64>>;
+    async fn get_nodes(&self) -> Result<Vec<NodeInfo>>;
 
     // Drop the tenant's namespace one node by node.id.
-    fn drop_node(
-        &self,
-        tenant_id: String,
-        namespace_id: String,
-        node_id: String,
-        seq: Option<u64>,
-    ) -> Result<()>;
+    async fn drop_node(&self, node_id: String, seq: Option<u64>) -> Result<()>;
+
+    // Keep the tenant's namespace node alive.
+    async fn heartbeat(&self, node_id: String, seq: Option<u64>) -> Result<u64>;
 }