merge aggregate #496

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

hustfisher wants to merge 59 commits into main_kvector_hybrid_202405 from main_kvector_hybrid_202408

Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

endpoint/src/kv/kvtime.rs

-Original file line number
+Diff line change
@@ -1,8 +1,9 @@
-    use super::{strategy::Postfix, uuid::*};
+    use super::uuid::*;
     use chrono::{Datelike, NaiveDate};
     use core::fmt::Write;
     use ds::RingSlice;
     use protocol::kv::Strategy;
+    use protocol::vector::Postfix;
     use sharding::hash::Hash;
     use sharding::{distribution::DBRange, hash::Hasher};
@@ Expand Down @@

endpoint/src/kv/strategy.rs

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -5,25 +5,10 @@ use super::kvtime::KVTime;
  
    use ds::RingSlice;

    use protocol::kv::Strategy;

    use protocol::vector::Postfix;

    use sharding::distribution::DBRange;

    use sharding::hash::Hasher;

    #[derive(Debug, Clone)]

    pub enum Postfix {

        YYMM,

        YYMMDD,

        INDEX,

    }

    impl Into<Postfix> for &str {

        fn into(self) -> Postfix {

            match self.to_lowercase().as_str() {

                "yymm" => Postfix::YYMM,

                _ => Postfix::YYMMDD,

            }

        }

    }

    #[derive(Debug, Clone)]

    pub enum Strategist {

        KVTime(KVTime),

    @@ -75,13 +60,14 @@ impl Default for Strategist {
  
    }

    impl Strategist {

        pub fn try_from(ns: &KvNamespace) -> Self {

            Self::KVTime(KVTime::new(

        pub fn try_from(ns: &KvNamespace) -> Option<Self> {

            let table_postfix = ns.basic.table_postfix.as_str().try_into().ok()?;

            Some(Self::KVTime(KVTime::new(

                ns.basic.db_name.clone(),

                ns.basic.db_count,

                //此策略默认所有年都有同样的shard，basic也只配置了一项，也暗示了这个默认

                ns.backends.iter().next().unwrap().1.len() as u32,

                ns.basic.table_postfix.as_str().into(),

            ))

                table_postfix,

            )))

        }

    }

endpoint/src/kv/topo.rs

-Original file line number
+Diff line change
@@ Expand Up / @@ -134,7 +134,7 @@ where @@
                         (ctx.idx as usize, &shard.master)
                     }
                 };
-                ctx.idx = idx as u16;
+                ctx.idx = idx as u8;
                 ctx.runs += 1;
                 // 只重试一次，重试次数过多，可能会导致雪崩。如果不重试，现在的配额策略在当前副本也只会连续发送四次请求，问题也不大
                 let try_next = ctx.runs == 1;
@@ Expand Down Expand Up / @@ -163,7 +163,10 @@ where @@
         }
         fn update(&mut self, namespace: &str, cfg: &str) {
             if let Some(ns) = KvNamespace::try_from(cfg) {
-                self.strategist = Strategist::try_from(&ns);
+                let Some(strategist) = Strategist::try_from(&ns) else {
+                    return;
+                };
+                self.strategist = strategist;
                 self.cfg.update(namespace, ns);
             }
         }
@@ Expand Down @@

endpoint/src/vector/batch.rs

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -1,13 +1,14 @@
  
    use super::strategy::Postfix;

    use chrono::{Datelike, NaiveDate};

    use chrono_tz::Tz;

    use core::fmt::Write;

    use ds::RingSlice;

    use protocol::Error;

    use protocol::{

        vector::{CommandType, KeysType, Postfix},

        Error, DATE_YYMM, DATE_YYMMDD,

    };

    use sharding::{distribution::DBRange, hash::Hasher};

    #[derive(Clone, Debug)]

    pub struct Batch {

    pub struct Aggregation {

        db_prefix: String,

        table_prefix: String,

        table_postfix: Postfix,

    @@ -18,7 +19,7 @@ pub struct Batch {
  
        si: Si,

    }

    impl Batch {

    impl Aggregation {

        pub fn new_with_db(

            db_prefix: String,

            table_prefix: String,

    @@ -63,11 +64,6 @@ impl Batch {
  
            &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);

    @@ -101,14 +97,44 @@ impl Batch {
  
            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 fn get_date(&self, cmd: CommandType, keys: &[RingSlice]) -> Result<NaiveDate, Error> {

            match cmd {

                CommandType::VRange | CommandType::VCard | CommandType::VRangeSi => {

                    Ok(NaiveDate::default())

                }

                //相比vrange多了一个日期key

                _ => {

                    let date = keys.last().unwrap();

                    let ymd = match self.keys_name.last().unwrap().as_str() {

                        DATE_YYMM => (

                            date.try_str_num(0..0 + 2)

                                .ok_or(Error::RequestProtocolInvalid)? as u16

                                + 2000,

                            date.try_str_num(2..2 + 2)

                                .ok_or(Error::RequestProtocolInvalid)? as u16,

                            1,

                        ),

                        DATE_YYMMDD => (

                            date.try_str_num(0..0 + 2)

                                .ok_or(Error::RequestProtocolInvalid)? as u16

                                + 2000,

                            date.try_str_num(2..2 + 2)

                                .ok_or(Error::RequestProtocolInvalid)? as u16,

                            date.try_str_num(4..4 + 2)

                                .ok_or(Error::RequestProtocolInvalid)? as u16,

                        ),

                        _ => (0, 0, 0),

                    };

                    NaiveDate::from_ymd_opt(ymd.0.into(), ymd.1.into(), ymd.2.into())

                        .ok_or(Error::RequestProtocolInvalid)

                }

            }

        }

        // 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();

    @@ -124,6 +150,51 @@ impl Batch {
  
        pub(crate) fn si_cols(&self) -> &[String] {

            &self.si_cols

        }

        pub(crate) fn keys_with_type(&self) -> Box<dyn Iterator<Item = KeysType> + '_> {

            Box::new(

                self.keys_name

                    .iter()

                    .map(|key_name| match key_name.as_str() {

                        DATE_YYMM | DATE_YYMMDD => KeysType::Time,

                        &_ => KeysType::Keys(key_name),

                    }),

            )

        }

        //校验动态信息，后续有重复校验的后面延迟校验

        //上行、xx.timeline请求多带一个日期key，日期key固定在最后

        pub(crate) fn check_vector_cmd(

            &self,

            vcmd: &protocol::vector::VectorCmd,

        ) -> protocol::Result<()> {

            match vcmd.cmd {

                CommandType::VRange | CommandType::VCard | CommandType::VRangeSi => {

                    if vcmd.keys.len() != self.keys().len() - 1 {

                        return Err(Error::RequestProtocolInvalid);

                    }

                    Ok(())

                }

                //相比vrange多了一个日期key

                CommandType::VGet

                | CommandType::VAdd

                | CommandType::VDel

                | CommandType::VUpdate

                | CommandType::VAddTimeline

                | CommandType::VDelTimeline

                | CommandType::VRangeTimeline

                | CommandType::VUpdateTimeline

                | CommandType::VAddSi

                | CommandType::VUpdateSi

                | CommandType::VDelSi => {

                    if vcmd.keys.len() != self.keys().len() {

                        return Err(Error::RequestProtocolInvalid);

                    }

                    Ok(())

                }

                CommandType::Unknown => panic!("not sup {:?}", vcmd.cmd),

            }

        }

    }

    #[derive(Clone, Debug)]

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

merge aggregate #496

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

merge aggregate #496

Are you sure you want to change the base?

Uh oh!

merge aggregate #496

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!