Main kvector hybrid 202405

ds/src/mem/ring_slice.rs

@@ -85,6 +85,9 @@ impl RingSlice {
     #[inline]
     pub fn try_str_num(&self, r: impl Range) -> Option<usize> {
         let (start, end) = r.range(self);
+        if start == end {
+            return None;
+        }
         let mut num = 0usize;
         for i in start..end {
             if !self[i].is_ascii_digit() {

endpoint/src/topo.rs

@@ -36,6 +36,7 @@ procs::topology_dispatcher! {
 
     pub trait Topology : Endpoint + Hash{
         fn exp_sec(&self) -> u32 {86400}
+        fn has_attach(&self) -> bool {false}
     } => where P:Protocol, E:Endpoint<Item = R>, R:Request, Topologies<E, P>: Endpoint
 
     trait Inited {

endpoint/src/vector/batch.rs

@@ -0,0 +1,162 @@
+use super::strategy::Postfix;
+use chrono::{Datelike, NaiveDate};
+use chrono_tz::Tz;
+use core::fmt::Write;
+use ds::RingSlice;
+use protocol::Error;
+use sharding::{distribution::DBRange, hash::Hasher};
+
+#[derive(Clone, Debug)]
+pub struct Batch {
+    db_prefix: String,
+    table_prefix: String,
+    table_postfix: Postfix,
+    hasher: Hasher,
+    distribution: DBRange,
+    keys_name: Vec<String>,
+    si_cols: Vec<String>,
+    si: Si,
+}
+
+impl Batch {
+    pub fn new_with_db(
+        db_prefix: String,
+        table_prefix: String,
+        db_count: u32,
+        shards: u32,
+        table_postfix: Postfix,
+        keys_name: Vec<String>,
+        si_cols: Vec<String>,
+        si_db_prefix: String,
+        si_db_count: u32,
+        si_table_prefix: String,
+        si_table_count: u32,
+        si_shards: u32,
+    ) -> Self {
+        Self {
+            db_prefix,
+            table_prefix,
+            table_postfix,
+            distribution: DBRange::new(db_count as usize, 1usize, shards as usize),
+            hasher: Hasher::from("crc32"),
+            keys_name,
+            si_cols,
+            si: Si::new(
+                si_db_prefix,
+                si_db_count,
+                si_table_prefix,
+                si_table_count,
+                si_shards,
+            ),
+        }
+    }
+
+    pub fn distribution(&self) -> &DBRange {
+        &self.distribution
+    }
+
+    pub fn si_distribution(&self) -> &DBRange {
+        self.si.distribution()
+    }
+
+    pub fn hasher(&self) -> &Hasher {
+        &self.hasher
+    }
+
+    pub fn get_date(&self, _: &[RingSlice]) -> Result<NaiveDate, Error> {
+        let now = chrono::Utc::now().with_timezone(&Tz::Asia__Shanghai);
+        Ok(NaiveDate::from_ymd_opt(now.year(), now.month(), now.day()).unwrap())
+    }
+
+    pub fn write_dname_with_hash(&self, buf: &mut impl Write, hash: i64) {
+        let db_idx: usize = self.distribution.db_idx(hash);
+        let _ = write!(buf, "{}_{}", self.db_prefix, db_idx);
+    }
+
+    pub fn write_tname_with_date(&self, buf: &mut impl Write, date: &NaiveDate) {
+        let (mut year, month, day) = (date.year(), date.month(), date.day());
+        year %= 100;
+        match self.table_postfix {
+            Postfix::YYMM => {
+                let _ = write!(buf, "{}_{:02}{:02}", &self.table_prefix, year, month);
+            }
+            //Postfix::YYMMDD
+            _ => {
+                let _ = write!(
+                    buf,
+                    "{}_{:02}{:02}{:02}",
+                    &self.table_prefix, year, month, day
+                );
+            }
+        }
+    }
+
+    pub fn write_database_table(&self, buf: &mut impl Write, date: &NaiveDate, hash: i64) {
+        self.write_dname_with_hash(buf, hash);
+        let _ = buf.write_char('.');
+        self.write_tname_with_date(buf, date)
+    }
+
+    pub(crate) fn write_si_database_table(&self, buf: &mut impl Write, hash: i64) {
+        self.si.write_database_table(buf, hash)
+    }
+
+    pub(crate) fn condition_keys(&self) -> Box<dyn Iterator<Item = Option<&String>> + '_> {
+        Box::new(self.keys_name.iter().map(|x| Some(x)))
+    }
+
+    pub(crate) fn keys(&self) -> &[String] {
+        &self.keys_name
+    }
+
+    // pub(crate) fn get_next_date(&self, year: u16, month: u8) -> NaiveDate {
+    //     if month == 1 {
+    //         return NaiveDate::from_ymd_opt((year - 1).into(), 12, 1).unwrap();
+    //     } else {
+    //         return NaiveDate::from_ymd_opt(year.into(), (month - 1).into(), 1).unwrap();
+    //     }
+    // }
+
+    pub(crate) fn batch(&self, limit: u64, _: &protocol::vector::VectorCmd) -> u64 {
+        limit
+    }
+
+    pub(crate) fn si_cols(&self) -> &[String] {
+        &self.si_cols
+    }
+}
+
+#[derive(Clone, Debug)]
+struct Si {
+    db_prefix: String,
+    table_prefix: String,
+    distribution: DBRange,
+}
+
+impl Si {
+    fn new(
+        db_prefix: String,
+        db_count: u32,
+        table_prefix: String,
+        table_count: u32,
+        shards: u32,
+    ) -> Self {
+        Self {
+            db_prefix: db_prefix,
+            table_prefix: table_prefix,
+            distribution: DBRange::new(db_count as usize, table_count as usize, shards as usize),
+        }
+    }
+    fn distribution(&self) -> &DBRange {
+        &self.distribution
+    }
+    fn write_database_table(&self, buf: &mut impl Write, hash: i64) {
+        let db_idx = self.distribution.db_idx(hash);
+        let table_idx = self.distribution.table_idx(hash);
+        let _ = write!(
+            buf,
+            "{}_{}.{}_{}",
+            self.db_prefix, db_idx, self.table_prefix, table_idx
+        );
+    }
+}

endpoint/src/vector/config.rs

@@ -14,6 +14,8 @@ pub struct VectorNamespace {
     pub(crate) backends_flaten: Vec<String>,
     #[serde(default)]
     pub(crate) backends: HashMap<Years, Vec<String>>,
+    #[serde(default)]
+    pub(crate) si_backends: Vec<String>,
 }
 
 #[derive(Debug, Clone, Default, Deserialize, Serialize)]
@@ -44,6 +46,20 @@ pub struct Basic {
     pub(crate) user: String,
     #[serde(default)]
     pub(crate) region_enabled: bool,
+    #[serde(default)]
+    pub(crate) si_db_name: String,
+    #[serde(default)]
+    pub(crate) si_table_name: String,
+    #[serde(default)]
+    pub(crate) si_db_count: u32,
+    #[serde(default)]
+    pub(crate) si_table_count: u32,
+    #[serde(default)]
+    pub(crate) si_user: String,
+    #[serde(default)]
+    pub(crate) si_password: String,
+    #[serde(default)]
+    pub(crate) si_cols: Vec<String>,
 }
 
 impl VectorNamespace {
@@ -66,7 +82,7 @@ impl VectorNamespace {
                     }
                     last_year = year.1;
                 }
-                match ns.decrypt_password() {
+                match ns.decrypt_password(ns.basic.password.as_bytes()) {
                     Ok(password) => ns.basic.password = password,
                     Err(e) => {
                         log::warn!("failed to decrypt password, e:{}", e);
@@ -77,6 +93,19 @@ impl VectorNamespace {
                     init.extend_from_slice(b.1);
                     init
                 });
+                if ns.basic.si_password.len() > 0 {
+                    match ns.decrypt_password(ns.basic.si_password.as_bytes()) {
+                        Ok(password) => ns.basic.si_password = password,
+                        Err(e) => {
+                            log::warn!("failed to decrypt si password, e:{}", e);
+                            return None;
+                        }
+                    }
+                }
+                if ns.si_backends.len() > 0 {
+                    ns.backends_flaten
+                        .extend_from_slice(ns.si_backends.as_slice());
+                }
                 Some(ns)
             }
             Err(e) => {
@@ -87,9 +116,9 @@ impl VectorNamespace {
     }
 
     #[inline]
-    fn decrypt_password(&self) -> Result<String, Box<dyn std::error::Error>> {
+    fn decrypt_password(&self, data: &[u8]) -> Result<String, Box<dyn std::error::Error>> {
         let key_pem = fs::read_to_string(&context::get().key_path)?;
-        let encrypted_data = general_purpose::STANDARD.decode(self.basic.password.as_bytes())?;
+        let encrypted_data = general_purpose::STANDARD.decode(data)?;
         let decrypted_data = ds::decrypt::decrypt_password(&key_pem, &encrypted_data)?;
         let decrypted_string = String::from_utf8(decrypted_data)?;
         Ok(decrypted_string)

endpoint/src/vector/mod.rs

@@ -1,3 +1,4 @@
+mod batch;
 pub(crate) mod config;
 mod strategy;
 pub mod topo;