Merge branch 'vector_continue' into kv2vector_topo

Author: parabala

Cargo.lock

@@ -368,6 +368,7 @@ impl RetrieveAttach {
         self.body.push(body_data);
         self.body_token_count += rows * columns;
         self.left_count = self.left_count.saturating_sub(rows);
+        log::debug!("left_count: {}", self.left_count);
     }
 
     #[inline]
@@ -413,6 +414,7 @@ impl RetrieveAttach {
                 }
             }
         }
+        log::debug!("si len: {},content: {:?}", self.si.len(), self.si);
         self.si.len() > 0
     }
     #[inline]

protocol/src/vector/attachment.rs

@@ -200,13 +200,18 @@ impl Protocol for Vector {
 
         match vec_attach.attch_type() {
             AttachType::Retrieve => {
-                // 如果header为none，说明查询结果为空，直接返回false
+                let attach = vec_attach.retrieve_attach_mut();
+                // 如果header为none，说明当前查询结果为空
+                // 如果有si，则看是否还有后续请求
+                // 如果没有si，直接返回false
                 if response.header.is_none() {
+                    if attach.has_si() {
+                        return attach.left_count != 0;
+                    }
                     return false;
                 }
 
                 assert!(response.header.is_some(), "rsp:{}", response);
-                let attach = vec_attach.retrieve_attach_mut();
                 let body_data = response.data().0.to_vec();
                 let header = response.header.as_mut().expect("rsp header");
 

protocol/src/vector/mod.rs

@@ -25,6 +25,7 @@ tokio.workspace = true
 ctor = "0.1.23"
 mysql_async = "0.31.3"
 chrono = "0.4"
+chrono-tz = { version = "0.5", default-features = false }
 lazy_static = "*"
 
 mysql = "*"

tests/Cargo.toml

@@ -1,5 +1,7 @@
 use std::{fs::File, io::BufRead};
 
+use chrono::TimeZone;
+use endpoint::kv::uuid::Uuid;
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -16,12 +18,12 @@ const TCP_LOG_PREFIX_LENT: usize =
 
 #[test]
 fn parse_like_by_me() {
-    let file_name = "datas/consistency.log";
+    let file_name = "tmp_data/vector.log";
     let file = File::open(file_name).unwrap();
     let reader = std::io::BufReader::new(file);
 
     let mut mesh_ids = Vec::new();
-    let mut tcp_ids = Vec::new();
+    let mut _tcp_ids = Vec::new();
     let mut count = 0;
     const MAX_COUNT: usize = 1;
 
@@ -43,32 +45,33 @@ fn parse_like_by_me() {
             // println!("mesh_ids: {:?}", mesh_ids);
             continue;
         } else {
-            tcp_ids = ids;
+            _tcp_ids = ids;
             // println!("tcp_ids: {:?}", tcp_ids);
 
             let diff_in_mesh: Vec<usize> = mesh_ids
                 .clone()
                 .into_iter()
-                .filter(|id| !tcp_ids.contains(id))
+                .filter(|id| !_tcp_ids.contains(id))
                 .collect();
-            let diff_in_tcp: Vec<usize> = tcp_ids
+            let diff_in_tcp: Vec<usize> = _tcp_ids
                 .clone()
                 .into_iter()
                 .filter(|id| !mesh_ids.contains(id))
                 .collect();
 
             if diff_in_mesh.len() > 0 || diff_in_tcp.len() > 0 {
-                println!("raw tcp line:{}", line);
+                println!("mesh/tcp count:{}/{}", mesh_ids.len(), _tcp_ids.len());
                 println!("mesh more ids:{:?}", diff_in_mesh);
                 println!("tcp more ids: {:?}", diff_in_tcp);
-
+                println!("mesh:{:?}", mesh_ids);
+                println!(" tcp:{:?}", _tcp_ids);
                 count += 1;
                 if count >= MAX_COUNT {
                     break;
                 }
 
                 mesh_ids.clear();
-                tcp_ids.clear();
+                _tcp_ids.clear();
             }
             continue;
         }
@@ -83,3 +86,18 @@ fn parse_line(line: &str) -> Vec<usize> {
     }
     ids
 }
+
+#[test]
+fn parse_time() {
+    // use endpoint::kv::uuid;
+    let uuid = 5100423841317698 as i64;
+    let seconds = uuid.unix_secs();
+
+    use chrono_tz::Asia::Shanghai;
+    let t = chrono::Utc
+        .timestamp_opt(seconds, 0)
+        .unwrap()
+        .with_timezone(&Shanghai)
+        .naive_local();
+    println!("{}", t);
+}