fix: json compatibility to null (#2287)

* fix: existing null value for schema name value

* chore: fix null check

* fix: change catalognamevalue and schemanamevalue to option

* fix: fix null case

src/common/meta/src/key.rs

@@ -494,15 +494,35 @@ macro_rules! impl_table_meta_value {
     }
 }
 
+#[macro_export]
+macro_rules! impl_optional_meta_value {
+    ($($val_ty: ty), *) => {
+        $(
+            impl $val_ty {
+                pub fn try_from_raw_value(raw_value: &[u8]) -> Result<Option<Self>> {
+                    serde_json::from_slice(raw_value).context(SerdeJsonSnafu)
+                }
+
+                pub fn try_as_raw_value(&self) -> Result<Vec<u8>> {
+                    serde_json::to_vec(self).context(SerdeJsonSnafu)
+                }
+            }
+        )*
+    }
+}
+
 impl_table_meta_value! {
-    CatalogNameValue,
-    SchemaNameValue,
     TableNameValue,
     TableInfoValue,
     DatanodeTableValue,
     TableRouteValue
 }
 
+impl_optional_meta_value! {
+    CatalogNameValue,
+    SchemaNameValue
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::BTreeMap;

src/common/meta/src/key/schema_name.rs

@@ -163,7 +163,7 @@ impl SchemaManager {
         let raw_key = schema.as_raw_key();
         let value = self.kv_backend.get(&raw_key).await?;
         value
-            .map(|v| SchemaNameValue::try_from_raw_value(v.value.as_ref()))
+            .and_then(|v| SchemaNameValue::try_from_raw_value(v.value.as_ref()).transpose())
             .transpose()
     }
 
@@ -206,7 +206,11 @@ mod tests {
         assert_eq!(value, from_value);
 
         let parsed = SchemaNameValue::try_from_raw_value("{\"ttl\":\"10s\"}".as_bytes()).unwrap();
-        assert_eq!(value, parsed);
+        assert_eq!(Some(value), parsed);
+        let none = SchemaNameValue::try_from_raw_value("null".as_bytes()).unwrap();
+        assert!(none.is_none());
+        let err_empty = SchemaNameValue::try_from_raw_value("".as_bytes());
+        assert!(err_empty.is_err());
     }
 
     #[tokio::test]