support VERSION() for 1.x/2.x

Cargo.toml

@@ -4,30 +4,23 @@ version = "0.2.0"
 authors = ["OceanBase OBKV Developers", "CeresDB Authors <ceresdbservice@gmail.com>"]
 edition = "2021"
 
+[workspace]
+members = ["ycsb-rs"]
+
 [workspace.package]
 version = "0.2.0"
 authors = ["OceanBase OBKV Developers", "CeresDB Authors <ceresdbservice@gmail.com>"]
 edition = "2021"
 
-[workspace]
-members = [
-    "ycsb-rs",
-]
-
-[lib]
-name = "obkv"
-
-[[bench]]
-name = "concurrent_insert"
-harness = false
-path = "benches/concurrent_insert/mod.rs"
-
 [workspace.dependencies]
 anyhow = "1.0"
 log = "0.4"
-toml = "0.7.3"
 prometheus-client = "0.21"
 tokio = { version = "1", features = ["full"] }
+toml = "0.7.3"
+
+[lib]
+name = "obkv"
 
 [dependencies]
 anyhow = { workspace = true }
@@ -71,8 +64,13 @@ serial_test = "2.0"
 serial_test_derive = "2.0"
 tempfile = "3.0"
 test-log = "0.2"
-time = "0.3"
+time = "0.3.36"
 tokio-test = "0.4"
 
+[[bench]]
+name = "concurrent_insert"
+harness = false
+path = "benches/concurrent_insert/mod.rs"
+
 [profile.release]
 debug = true

src/client/filter.rs

@@ -107,7 +107,7 @@ impl ObTableFilterList {
     where
         I: IntoIterator<Item = Filter>,
     {
-        self.filters.extend(filters.into_iter())
+        self.filters.extend(filters)
     }
 }
 

src/client/query.rs

@@ -386,7 +386,7 @@ impl Drop for QueryStreamResult {
 pub enum QueryResultSet {
     #[default]
     None,
-    Some(QueryStreamResult),
+    Some(Box<QueryStreamResult>),
 }
 
 impl QueryResultSet {
@@ -403,7 +403,7 @@ impl QueryResultSet {
     }
 
     pub fn from_stream_result(stream_result: QueryStreamResult) -> Self {
-        QueryResultSet::Some(stream_result)
+        QueryResultSet::Some(Box::new(stream_result))
     }
 
     pub fn cache_size(&self) -> usize {

src/client/table_client.rs

@@ -18,7 +18,6 @@
 use std::{
     borrow::Borrow,
     collections::HashMap,
-    isize,
     sync::{
         atomic::{AtomicBool, AtomicI64, AtomicIsize, AtomicUsize, Ordering},
         Arc, Mutex, RwLock,
@@ -155,7 +154,7 @@ impl ServerRoster {
             return;
         }
 
-        let mut priority: isize = isize::min_value();
+        let mut priority: isize = isize::MIN;
 
         for addr in self.roster.rl().iter() {
             if addr.priority() > priority {
@@ -169,13 +168,7 @@ impl ServerRoster {
     pub fn max_priority(&self) -> isize {
         let p = self.max_priority.load(Ordering::Acquire);
 
-        if p < -MAX_PRIORITY {
-            -MAX_PRIORITY
-        } else if p > MAX_PRIORITY {
-            MAX_PRIORITY
-        } else {
-            p
-        }
+        p.clamp(-MAX_PRIORITY, MAX_PRIORITY)
     }
 }
 
@@ -679,6 +672,7 @@ impl ObTableClientInner {
         result
     }
 
+    #[allow(clippy::mutable_key_type)]
     fn add_ob_table_to_roster(
         &self,
         addr: &ObServerAddr,
@@ -2607,9 +2601,8 @@ impl ObTableAggregation {
     pub async fn execute(mut self) -> Result<HashMap<String, Value>> {
         // In order to get cache size.
         self.table_query = self.table_query.select(self.aggregation_operations);
-        let mut query_set = self.table_query.execute().await.map_err(|e| {
+        let mut query_set = self.table_query.execute().await.inspect_err(|_e| {
             error!("fail to execute aggregate");
-            e
         })?;
         let aggregate_option = query_set.next().await;
         match aggregate_option {

src/error.rs

@@ -179,7 +179,7 @@ impl Error {
         } else if let Error::Common(CommonErrCode::ConnPool, message) = self {
             // conn_pool will produced this error if all connection to a server is shutdown
             // which means we need refresh
-            return message.ends_with("are all removed");
+            message.ends_with("are all removed")
         } else {
             false
         }

src/location/mod.rs

@@ -632,6 +632,7 @@ impl ObTableLocation {
         }
     }
 
+    #[allow(clippy::mutable_key_type)]
     fn get_pool_from_cache(
         &self,
         pools: &HashMap<ObServerAddr, HashMap<String, Arc<my::Pool>>>,
@@ -645,6 +646,7 @@ impl ObTableLocation {
     }
 
     /// Invalidate expired mysql pools.
+    #[allow(clippy::mutable_key_type)]
     pub fn invalidate_mysql_pools(&self, valid_addrs: &[ObServerAddr]) {
         let valid_addrs: HashSet<ObServerAddr> = valid_addrs.iter().cloned().collect();
         self.mysql_pools

src/location/util.rs

@@ -17,9 +17,7 @@
 
 use std::{
     collections::HashMap,
-    i32, i64,
     sync::{atomic::AtomicUsize, Arc},
-    u8,
 };
 
 use chrono::Utc;
@@ -53,8 +51,35 @@ pub struct LocationUtil {}
 impl LocationUtil {
     pub fn get_ob_version_from_server(conn: &mut my::PooledConn) -> Result<()> {
         if ob_vsn_major() == 0 {
-            let sql = "SELECT /*+READ_CONSISTENCY(WEAK)*/ OB_VERSION() AS CLUSTER_VERSION";
-            for row in conn.query::<Row, &str>(sql)? {
+            let sql_4x = "SELECT /*+READ_CONSISTENCY(WEAK)*/ OB_VERSION() AS CLUSTER_VERSION";
+            let sql_2x = "SELECT /*+READ_CONSISTENCY(WEAK)*/ VERSION() AS CLUSTER_VERSION";
+
+            // try sql_4x first
+            let result = conn.query::<Row, &str>(sql_4x);
+            let rows = match result {
+                Ok(rows) => rows,
+                Err(_) => {
+                    // if sql_4x failed, try sql_2x
+                    match conn.query::<Row, &str>(sql_2x) {
+                        Ok(rows) => rows,
+                        Err(e) => {
+                            error!(
+                                "Both OB_VERSION() and VERSION() are not supported, err:{}",
+                                e
+                            );
+                            return Err(CommonErr(
+                                CommonErrCode::ConvertFailed,
+                                format!(
+                                    "Both OB_VERSION() and VERSION() are not supported, err:{e}"
+                                ),
+                            ));
+                        }
+                    }
+                }
+            };
+
+            // process the query results
+            for row in rows {
                 let cluster_version = match my::from_row_opt(row) {
                     Ok(version) => version,
                     Err(e) => {

src/rpc/conn_pool.rs

@@ -19,7 +19,6 @@ use std::{
     cmp,
     sync::{Arc, Condvar, Mutex},
     time::{Duration, Instant},
-    u32,
 };
 
 use tokio::time::sleep;

src/rpc/protocol/lsop.rs

@@ -458,7 +458,7 @@ impl ObTableSingleOpEntity {
     ) -> Result<()> {
         bit_map.clear();
         column_names.clear();
-        let bm_len = (len + 7) / 8;
+        let bm_len = len.div_ceil(8);
         for idx in 0..bm_len {
             bit_map.push(decode_u8(src)?);
             for bit_idx in 0usize..8 {

src/serde_obkv/de.rs

@@ -54,7 +54,7 @@ impl<'de> Deserializer<'de> {
     }
 }
 
-impl<'de, 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
+impl<'de> de::Deserializer<'de> for &mut Deserializer<'de> {
     type Error = Error;
 
     forward_to_deserialize_any! {

src/serde_obkv/error.rs

@@ -76,6 +76,6 @@ use std::io;
 
 impl From<Error> for io::Error {
     fn from(e: Error) -> io::Error {
-        io::Error::new(io::ErrorKind::Other, format!("cause: {e}"))
+        io::Error::other(format!("cause: {e}"))
     }
 }

src/serde_obkv/ser.rs

@@ -362,9 +362,9 @@ impl<'a> ser::SerializeStruct for SerializeStruct<'a> {
     type Ok = ();
 
     /// encode table entity properties
-    fn serialize_field<V: ?Sized>(&mut self, key: &'static str, value: &V) -> Result<()>
+    fn serialize_field<V>(&mut self, key: &'static str, value: &V) -> Result<()>
     where
-        V: ser::Serialize,
+        V: ?Sized + ser::Serialize,
     {
         if self.first {
             match self.ser {
@@ -393,9 +393,9 @@ impl<'a> ser::SerializeMap for SerializeStruct<'a> {
     type Error = Error;
     type Ok = ();
 
-    fn serialize_key<T: ?Sized>(&mut self, key: &T) -> Result<()>
+    fn serialize_key<T>(&mut self, key: &T) -> Result<()>
     where
-        T: Serialize,
+        T: ?Sized + Serialize,
     {
         match self.ser {
             Serializer::Output(_, ref mut raw_key) => *raw_key = true,
@@ -410,18 +410,18 @@ impl<'a> ser::SerializeMap for SerializeStruct<'a> {
         ret
     }
 
-    fn serialize_value<T: ?Sized>(&mut self, value: &T) -> Result<()>
+    fn serialize_value<T>(&mut self, value: &T) -> Result<()>
     where
-        T: Serialize,
+        T: ?Sized + Serialize,
     {
         value.serialize(&mut *self.ser)
     }
 
     /// encode table entity properties
-    fn serialize_entry<K: ?Sized, V: ?Sized>(&mut self, key: &K, value: &V) -> Result<()>
+    fn serialize_entry<K, V>(&mut self, key: &K, value: &V) -> Result<()>
     where
-        K: Serialize,
-        V: Serialize,
+        K: ?Sized + Serialize,
+        V: ?Sized + Serialize,
     {
         if self.first {
             match self.ser {
@@ -450,9 +450,9 @@ impl<'a> ser::SerializeSeq for SerializeArray<'a> {
     type Error = Error;
     type Ok = ();
 
-    fn serialize_element<T: ?Sized>(&mut self, elem: &T) -> Result<()>
+    fn serialize_element<T>(&mut self, elem: &T) -> Result<()>
     where
-        T: ser::Serialize,
+        T: ?Sized + ser::Serialize,
     {
         elem.serialize(&mut *self.ser)?;
         Ok(())
@@ -467,9 +467,9 @@ impl<'a> ser::SerializeTuple for SerializeArray<'a> {
     type Error = Error;
     type Ok = ();
 
-    fn serialize_element<T: ?Sized>(&mut self, elem: &T) -> Result<()>
+    fn serialize_element<T>(&mut self, elem: &T) -> Result<()>
     where
-        T: ser::Serialize,
+        T: ?Sized + ser::Serialize,
     {
         ser::SerializeSeq::serialize_element(self, elem)
     }
@@ -483,9 +483,9 @@ impl<'a> ser::SerializeTupleStruct for SerializeArray<'a> {
     type Error = Error;
     type Ok = ();
 
-    fn serialize_field<V: ?Sized>(&mut self, value: &V) -> Result<()>
+    fn serialize_field<V>(&mut self, value: &V) -> Result<()>
     where
-        V: ser::Serialize,
+        V: ?Sized + ser::Serialize,
     {
         ser::SerializeSeq::serialize_element(self, value)
     }