diff --git a/src/common/meta/src/key.rs b/src/common/meta/src/key.rs
index c1d1b54ee3..0b5e7bb43f 100644
--- a/src/common/meta/src/key.rs
+++ b/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;
diff --git a/src/common/meta/src/key/schema_name.rs b/src/common/meta/src/key/schema_name.rs
index c162e31b16..5da6bdb2d3 100644
--- a/src/common/meta/src/key/schema_name.rs
+++ b/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]
diff --git a/src/frontend/src/instance/distributed.rs b/src/frontend/src/instance/distributed.rs
index bccbb32161..ef7b9adf8d 100644
--- a/src/frontend/src/instance/distributed.rs
+++ b/src/frontend/src/instance/distributed.rs
@@ -67,9 +67,8 @@ use crate::catalog::FrontendCatalogManager;
 use crate::error::{
     self, AlterExprToRequestSnafu, CatalogSnafu, ColumnDataTypeSnafu, ColumnNotFoundSnafu,
     DeserializePartitionSnafu, InvokeDatanodeSnafu, NotSupportedSnafu, ParseSqlSnafu,
-    RequestDatanodeSnafu, RequestMetaSnafu, Result, SchemaExistsSnafu, SchemaNotFoundSnafu,
-    TableAlreadyExistSnafu, TableMetadataManagerSnafu, TableNotFoundSnafu, TableSnafu,
-    UnrecognizedTableOptionSnafu,
+    RequestDatanodeSnafu, RequestMetaSnafu, Result, SchemaExistsSnafu, TableAlreadyExistSnafu,
+    TableMetadataManagerSnafu, TableNotFoundSnafu, TableSnafu, UnrecognizedTableOptionSnafu,
 };
 use crate::expr_factory;
 use crate::instance::distributed::deleter::DistDeleter;
@@ -106,7 +105,7 @@ impl DistInstance {
     ) -> Result<TableRef> {
         let _timer = common_telemetry::timer!(crate::metrics::DIST_CREATE_TABLE);
         // 1. get schema info
-        let schema = self
+        let schema_value = self
             .catalog_manager
             .table_metadata_manager_ref()
             .schema_manager()
@@ -117,13 +116,6 @@ impl DistInstance {
             .await
             .context(TableMetadataManagerSnafu)?;
 
-        let Some(schema_opts) = schema else {
-            return SchemaNotFoundSnafu {
-                schema_info: &create_table.schema_name,
-            }
-            .fail();
-        };
-
         let table_name = TableName::new(
             &create_table.catalog_name,
             &create_table.schema_name,
@@ -132,7 +124,7 @@ impl DistInstance {
 
         let (partitions, partition_cols) = parse_partitions(create_table, partitions)?;
 
-        let mut table_info = create_table_info(create_table, partition_cols, schema_opts)?;
+        let mut table_info = create_table_info(create_table, partition_cols, schema_value)?;
 
         let resp = self
             .create_table_procedure(create_table, partitions, table_info.clone())
@@ -795,7 +787,7 @@ fn create_partitions_stmt(partitions: Vec<PartitionInfo>) -> Result<Option<Parti
 fn create_table_info(
     create_table: &CreateTableExpr,
     partition_columns: Vec<String>,
-    schema_opts: SchemaNameValue,
+    schema_opts: Option<SchemaNameValue>,
 ) -> Result<RawTableInfo> {
     let mut column_schemas = Vec::with_capacity(create_table.column_defs.len());
     let mut column_name_to_index_map = HashMap::new();
@@ -881,8 +873,11 @@ fn create_table_info(
     Ok(table_info)
 }
 
-fn merge_options(mut table_opts: TableOptions, schema_opts: SchemaNameValue) -> TableOptions {
-    table_opts.ttl = table_opts.ttl.or(schema_opts.ttl);
+fn merge_options(
+    mut table_opts: TableOptions,
+    schema_opts: Option<SchemaNameValue>,
+) -> TableOptions {
+    table_opts.ttl = table_opts.ttl.or(schema_opts.and_then(|s| s.ttl));
     table_opts
 }