fix : insert lineprotocol error

cnosdb · Jul 26, 2023 · 8a8b09a · 8a8b09a
1 parent 9a42b66
commit 8a8b09a
Show file tree

Hide file tree

Showing 6 changed files with 76 additions and 17 deletions.
diff --git a/coordinator/src/writer.rs b/coordinator/src/writer.rs
@@ -395,17 +395,27 @@ impl PointWriter {
                 span_recorder.span_ctx(),
             )
             .await;
-        if let Err(err @ CoordinatorError::FailoverNode { .. }) = result {
-            debug!(
-                "write data to remote {}({}) failed; write to hinted handoff!",
-                node_id, vnode_id
-            );
-
-            span_recorder.error(err.to_string());
 
-            return self
-                .write_to_handoff(vnode_id, node_id, tenant, precision, data)
-                .await;
+        if let Err(ref err) = result {
+            let meta_retry = MetaError::Retry {
+                msg: "default".to_string(),
+            };
+            let tskv_memory = tskv::Error::MemoryExhausted;
+            if matches!(*err, CoordinatorError::FailoverNode { .. })
+                || err.error_code().to_string() == meta_retry.error_code().to_string()
+                || err.error_code().to_string() == tskv_memory.error_code().to_string()
+            {
+                debug!(
+                    "write data to remote {}({}) failed; write to hinted handoff!",
+                    node_id, vnode_id
+                );
+
+                span_recorder.error(err.to_string());
+
+                return self
+                    .write_to_handoff(vnode_id, node_id, tenant, precision, data)
+                    .await;
+            }
         }
 
         debug!(

diff --git a/meta/src/error.rs b/meta/src/error.rs
@@ -116,9 +116,12 @@ pub enum MetaError {
     #[error_code(code = 24)]
     RequestLimit { kind: RequestLimiterKind },
 
-    #[snafu(display("An error occurred while processing the data. Please try again"))]
+    #[snafu(display(
+        "An error occurred while processing the data: {}. Please try again",
+        msg
+    ))]
     #[error_code(code = 25)]
-    Retry,
+    Retry { msg: String },
 
     #[snafu(display("{}", msg))]
     ObjectLimit { msg: String },

diff --git a/meta/src/model/meta_tenant.rs b/meta/src/model/meta_tenant.rs
@@ -17,7 +17,7 @@ use store::command;
 use trace::info;
 
 use crate::error::{MetaError, MetaResult};
-use crate::store::command::EntryLog;
+use crate::store::command::{EntryLog, ReadCommand};
 use crate::store::key_path;
 use crate::{client, store};
 
@@ -468,6 +468,24 @@ impl TenantMeta {
         Ok(None)
     }
 
+    pub async fn get_tskv_table_schema_by_meta(
+        &self,
+        db: &str,
+        table: &str,
+    ) -> MetaResult<Option<Arc<TskvTableSchema>>> {
+        let req = ReadCommand::TableSchema(
+            self.cluster.clone(),
+            self.tenant.name().to_string(),
+            db.to_string(),
+            table.to_string(),
+        );
+        let rsp = self.client.read::<Option<TableSchema>>(&req).await?;
+        if let Some(TableSchema::TsKvTableSchema(val)) = rsp {
+            return Ok(Some(val));
+        }
+        Ok(None)
+    }
+
     pub fn get_external_table_schema(
         &self,
         db: &str,

diff --git a/meta/src/store/command.rs b/meta/src/store/command.rs
@@ -144,6 +144,8 @@ pub enum ReadCommand {
     Tenant(String, String),
     // cluster
     Tenants(String),
+    // cluster, tenant, db, table
+    TableSchema(String, String, String, String),
 }
 
 pub const ENTRY_LOG_TYPE_SET: i32 = 1;

diff --git a/meta/src/store/state_machine.rs b/meta/src/store/state_machine.rs
@@ -406,6 +406,10 @@ impl StateMachine {
                 response_encode(self.get_struct::<Tenant>(&path))
             }
             ReadCommand::Tenants(cluster) => response_encode(self.process_read_tenants(cluster)),
+            ReadCommand::TableSchema(cluster, tenant_name, db_name, table_name) => {
+                let path = KeyPath::tenant_schema_name(cluster, tenant_name, db_name, table_name);
+                response_encode(self.get_struct::<TableSchema>(&path))
+            }
         }
     }
 

diff --git a/tskv/src/schema/schemas.rs b/tskv/src/schema/schemas.rs
@@ -211,8 +211,19 @@ impl DBschemas {
                     let schema_get = self
                         .client
                         .get_tskv_table_schema(&schema.db, &schema.name)
-                        .map_err(|_| MetaError::Retry)?
-                        .ok_or(MetaError::Retry)?;
+                        .map_err(|e| MetaError::Retry { msg: e.to_string() })?;
+                    let schema_get = match schema_get {
+                        None => self
+                            .client
+                            .get_tskv_table_schema_by_meta(&schema.db, &schema.name)
+                            .await
+                            .map_err(|e| MetaError::Retry { msg: e.to_string() })?
+                            .ok_or(MetaError::Retry {
+                                msg: "Table not found after TableAlreadyExists".to_string(),
+                            })?,
+                        Some(schema) => schema,
+                    };
+
                     if schema.tenant == schema_get.tenant
                         && schema.db == schema_get.db
                         && schema.columns() == schema_get.columns()
@@ -223,8 +234,19 @@ impl DBschemas {
                             let schema_get = self
                                 .client
                                 .get_tskv_table_schema(&schema.db, &schema.name)
-                                .map_err(|_| MetaError::Retry)?
-                                .ok_or(MetaError::Retry)?;
+                                .map_err(|e| MetaError::Retry { msg: e.to_string() })?;
+                            let schema_get = match schema_get {
+                                None => self
+                                    .client
+                                    .get_tskv_table_schema_by_meta(&schema.db, &schema.name)
+                                    .await
+                                    .map_err(|e| MetaError::Retry { msg: e.to_string() })?
+                                    .ok_or(MetaError::Retry {
+                                        msg: "Table not found when retry update table schema"
+                                            .to_string(),
+                                    })?,
+                                Some(schema) => schema,
+                            };
                             let mut schema = schema.as_ref().clone();
                             schema.schema_id = schema_get.schema_id + 1;
                             let schema = Arc::new(schema);