refactor: explicitly define json struct to ingest jsonbench data (#7462)

ingest jsonbench data Signed-off-by: luofucong <luofc@foxmail.com>
2026-05-30 11:50:38 +00:00 · 2025-12-24 15:30:22 +08:00
parent 2d9967b981
commit dc9f3a702e
31 changed files with 795 additions and 481 deletions
--- a/src/common/sql/Cargo.toml
+++ b/src/common/sql/Cargo.toml
@@ -5,10 +5,12 @@ edition.workspace = true
 license.workspace = true

 [dependencies]
+arrow-schema.workspace = true
 common-base.workspace = true
 common-decimal.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
+common-telemetry.workspace = true
 common-time.workspace = true
 datafusion-sql.workspace = true
 datatypes.workspace = true
--- a/src/common/sql/src/convert.rs
+++ b/src/common/sql/src/convert.rs
@@ -14,11 +14,12 @@

 use std::str::FromStr;

+use arrow_schema::extension::ExtensionType;
 use common_time::Timestamp;
 use common_time::timezone::Timezone;
-use datatypes::json::JsonStructureSettings;
+use datatypes::extension::json::JsonExtensionType;
 use datatypes::prelude::ConcreteDataType;
-use datatypes::schema::ColumnDefaultConstraint;
+use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema};
 use datatypes::types::{JsonFormat, parse_string_to_jsonb, parse_string_to_vector_type_value};
 use datatypes::value::{OrderedF32, OrderedF64, Value};
 use snafu::{OptionExt, ResultExt, ensure};
@@ -124,13 +125,14 @@ pub(crate) fn sql_number_to_value(data_type: &ConcreteDataType, n: &str) -> Resu
 /// If `auto_string_to_numeric` is true, tries to cast the string value to numeric values,
 /// and returns error if the cast fails.
 pub fn sql_value_to_value(
-    column_name: &str,
-    data_type: &ConcreteDataType,
+    column_schema: &ColumnSchema,
    sql_val: &SqlValue,
    timezone: Option<&Timezone>,
    unary_op: Option<UnaryOperator>,
    auto_string_to_numeric: bool,
 ) -> Result<Value> {
+    let column_name = &column_schema.name;
+    let data_type = &column_schema.data_type;
    let mut value = match sql_val {
        SqlValue::Number(n, _) => sql_number_to_value(data_type, n)?,
        SqlValue::Null => Value::Null,
@@ -146,13 +148,9 @@ pub fn sql_value_to_value(

            (*b).into()
        }
-        SqlValue::DoubleQuotedString(s) | SqlValue::SingleQuotedString(s) => parse_string_to_value(
-            column_name,
-            s.clone(),
-            data_type,
-            timezone,
-            auto_string_to_numeric,
-        )?,
+        SqlValue::DoubleQuotedString(s) | SqlValue::SingleQuotedString(s) => {
+            parse_string_to_value(column_schema, s.clone(), timezone, auto_string_to_numeric)?
+        }
        SqlValue::HexStringLiteral(s) => {
            // Should not directly write binary into json column
            ensure!(
@@ -244,12 +242,12 @@ pub fn sql_value_to_value(
 }

 pub(crate) fn parse_string_to_value(
-    column_name: &str,
+    column_schema: &ColumnSchema,
    s: String,
-    data_type: &ConcreteDataType,
    timezone: Option<&Timezone>,
    auto_string_to_numeric: bool,
 ) -> Result<Value> {
+    let data_type = &column_schema.data_type;
    if auto_string_to_numeric && let Some(value) = auto_cast_to_numeric(&s, data_type)? {
        return Ok(value);
    }
@@ -257,7 +255,7 @@ pub(crate) fn parse_string_to_value(
    ensure!(
        data_type.is_stringifiable(),
        ColumnTypeMismatchSnafu {
-            column_name,
+            column_name: column_schema.name.clone(),
            expect: data_type.clone(),
            actual: ConcreteDataType::string_datatype(),
        }
@@ -303,23 +301,21 @@ pub(crate) fn parse_string_to_value(
            }
        }
        ConcreteDataType::Binary(_) => Ok(Value::Binary(s.as_bytes().into())),
-        ConcreteDataType::Json(j) => {
-            match &j.format {
-                JsonFormat::Jsonb => {
-                    let v = parse_string_to_jsonb(&s).context(DatatypeSnafu)?;
-                    Ok(Value::Binary(v.into()))
-                }
-                JsonFormat::Native(_inner) => {
-                    // Always use the structured version at this level.
-                    let serde_json_value =
-                        serde_json::from_str(&s).context(DeserializeSnafu { json: s })?;
-                    let json_structure_settings = JsonStructureSettings::Structured(None);
-                    json_structure_settings
-                        .encode(serde_json_value)
-                        .context(DatatypeSnafu)
-                }
+        ConcreteDataType::Json(j) => match &j.format {
+            JsonFormat::Jsonb => {
+                let v = parse_string_to_jsonb(&s).context(DatatypeSnafu)?;
+                Ok(Value::Binary(v.into()))
            }
-        }
+            JsonFormat::Native(_) => {
+                let extension_type: Option<JsonExtensionType> =
+                    column_schema.extension_type().context(DatatypeSnafu)?;
+                let json_structure_settings = extension_type
+                    .and_then(|x| x.metadata().json_structure_settings.clone())
+                    .unwrap_or_default();
+                let v = serde_json::from_str(&s).context(DeserializeSnafu { json: s })?;
+                json_structure_settings.encode(v).context(DatatypeSnafu)
+            }
+        },
        ConcreteDataType::Vector(d) => {
            let v = parse_string_to_vector_type_value(&s, Some(d.dim)).context(DatatypeSnafu)?;
            Ok(Value::Binary(v.into()))
@@ -417,305 +413,265 @@ mod test {

    use super::*;

+    macro_rules! call_parse_string_to_value {
+        ($column_name: expr, $input: expr, $data_type: expr) => {
+            call_parse_string_to_value!($column_name, $input, $data_type, None)
+        };
+        ($column_name: expr, $input: expr, $data_type: expr, timezone = $timezone: expr) => {
+            call_parse_string_to_value!($column_name, $input, $data_type, Some($timezone))
+        };
+        ($column_name: expr, $input: expr, $data_type: expr, $timezone: expr) => {{
+            let column_schema = ColumnSchema::new($column_name, $data_type, true);
+            parse_string_to_value(&column_schema, $input, $timezone, true)
+        }};
+    }
+
    #[test]
-    fn test_string_to_value_auto_numeric() {
+    fn test_string_to_value_auto_numeric() -> Result<()> {
        // Test string to boolean with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "true".to_string(),
-            &ConcreteDataType::boolean_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::boolean_datatype()
+        )?;
        assert_eq!(Value::Boolean(true), result);

        // Test invalid string to boolean with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_a_boolean".to_string(),
-            &ConcreteDataType::boolean_datatype(),
-            None,
-            true,
+            ConcreteDataType::boolean_datatype()
        );
        assert!(result.is_err());

        // Test string to int8
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "42".to_string(),
-            &ConcreteDataType::int8_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::int8_datatype()
+        )?;
        assert_eq!(Value::Int8(42), result);

        // Test invalid string to int8 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_an_int8".to_string(),
-            &ConcreteDataType::int8_datatype(),
-            None,
-            true,
+            ConcreteDataType::int8_datatype()
        );
        assert!(result.is_err());

        // Test string to int16
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "1000".to_string(),
-            &ConcreteDataType::int16_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::int16_datatype()
+        )?;
        assert_eq!(Value::Int16(1000), result);

        // Test invalid string to int16 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_an_int16".to_string(),
-            &ConcreteDataType::int16_datatype(),
-            None,
-            true,
+            ConcreteDataType::int16_datatype()
        );
        assert!(result.is_err());

        // Test string to int32
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "100000".to_string(),
-            &ConcreteDataType::int32_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::int32_datatype()
+        )?;
        assert_eq!(Value::Int32(100000), result);

        // Test invalid string to int32 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_an_int32".to_string(),
-            &ConcreteDataType::int32_datatype(),
-            None,
-            true,
+            ConcreteDataType::int32_datatype()
        );
        assert!(result.is_err());

        // Test string to int64
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "1000000".to_string(),
-            &ConcreteDataType::int64_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::int64_datatype()
+        )?;
        assert_eq!(Value::Int64(1000000), result);

        // Test invalid string to int64 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_an_int64".to_string(),
-            &ConcreteDataType::int64_datatype(),
-            None,
-            true,
+            ConcreteDataType::int64_datatype()
        );
        assert!(result.is_err());

        // Test string to uint8
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "200".to_string(),
-            &ConcreteDataType::uint8_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::uint8_datatype()
+        )?;
        assert_eq!(Value::UInt8(200), result);

        // Test invalid string to uint8 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_a_uint8".to_string(),
-            &ConcreteDataType::uint8_datatype(),
-            None,
-            true,
+            ConcreteDataType::uint8_datatype()
        );
        assert!(result.is_err());

        // Test string to uint16
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "60000".to_string(),
-            &ConcreteDataType::uint16_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::uint16_datatype()
+        )?;
        assert_eq!(Value::UInt16(60000), result);

        // Test invalid string to uint16 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_a_uint16".to_string(),
-            &ConcreteDataType::uint16_datatype(),
-            None,
-            true,
+            ConcreteDataType::uint16_datatype()
        );
        assert!(result.is_err());

        // Test string to uint32
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "4000000000".to_string(),
-            &ConcreteDataType::uint32_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::uint32_datatype()
+        )?;
        assert_eq!(Value::UInt32(4000000000), result);

        // Test invalid string to uint32 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_a_uint32".to_string(),
-            &ConcreteDataType::uint32_datatype(),
-            None,
-            true,
+            ConcreteDataType::uint32_datatype()
        );
        assert!(result.is_err());

        // Test string to uint64
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "18446744073709551615".to_string(),
-            &ConcreteDataType::uint64_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::uint64_datatype()
+        )?;
        assert_eq!(Value::UInt64(18446744073709551615), result);

        // Test invalid string to uint64 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_a_uint64".to_string(),
-            &ConcreteDataType::uint64_datatype(),
-            None,
-            true,
+            ConcreteDataType::uint64_datatype()
        );
        assert!(result.is_err());

        // Test string to float32
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "3.5".to_string(),
-            &ConcreteDataType::float32_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::float32_datatype()
+        )?;
        assert_eq!(Value::Float32(OrderedF32::from(3.5)), result);

        // Test invalid string to float32 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_a_float32".to_string(),
-            &ConcreteDataType::float32_datatype(),
-            None,
-            true,
+            ConcreteDataType::float32_datatype()
        );
        assert!(result.is_err());

        // Test string to float64
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "3.5".to_string(),
-            &ConcreteDataType::float64_datatype(),
-            None,
-            true,
-        )
-        .unwrap();
+            ConcreteDataType::float64_datatype()
+        )?;
        assert_eq!(Value::Float64(OrderedF64::from(3.5)), result);

        // Test invalid string to float64 with auto cast
-        let result = parse_string_to_value(
+        let result = call_parse_string_to_value!(
            "col",
            "not_a_float64".to_string(),
-            &ConcreteDataType::float64_datatype(),
-            None,
-            true,
+            ConcreteDataType::float64_datatype()
        );
        assert!(result.is_err());
+        Ok(())
    }

-    #[test]
-    fn test_sql_value_to_value() {
-        let sql_val = SqlValue::Null;
-        assert_eq!(
-            Value::Null,
-            sql_value_to_value(
-                "a",
-                &ConcreteDataType::float64_datatype(),
-                &sql_val,
-                None,
+    macro_rules! call_sql_value_to_value {
+        ($column_name: expr, $data_type: expr, $sql_value: expr) => {
+            call_sql_value_to_value!($column_name, $data_type, $sql_value, None, None, false)
+        };
+        ($column_name: expr, $data_type: expr, $sql_value: expr, timezone = $timezone: expr) => {
+            call_sql_value_to_value!(
+                $column_name,
+                $data_type,
+                $sql_value,
+                Some($timezone),
                None,
                false
            )
-            .unwrap()
+        };
+        ($column_name: expr, $data_type: expr, $sql_value: expr, unary_op = $unary_op: expr) => {
+            call_sql_value_to_value!(
+                $column_name,
+                $data_type,
+                $sql_value,
+                None,
+                Some($unary_op),
+                false
+            )
+        };
+        ($column_name: expr, $data_type: expr, $sql_value: expr, auto_string_to_numeric) => {
+            call_sql_value_to_value!($column_name, $data_type, $sql_value, None, None, true)
+        };
+        ($column_name: expr, $data_type: expr, $sql_value: expr, $timezone: expr, $unary_op: expr, $auto_string_to_numeric: expr) => {{
+            let column_schema = ColumnSchema::new($column_name, $data_type, true);
+            sql_value_to_value(
+                &column_schema,
+                $sql_value,
+                $timezone,
+                $unary_op,
+                $auto_string_to_numeric,
+            )
+        }};
+    }
+
+    #[test]
+    fn test_sql_value_to_value() -> Result<()> {
+        let sql_val = SqlValue::Null;
+        assert_eq!(
+            Value::Null,
+            call_sql_value_to_value!("a", ConcreteDataType::float64_datatype(), &sql_val)?
        );

        let sql_val = SqlValue::Boolean(true);
        assert_eq!(
            Value::Boolean(true),
-            sql_value_to_value(
-                "a",
-                &ConcreteDataType::boolean_datatype(),
-                &sql_val,
-                None,
-                None,
-                false
-            )
-            .unwrap()
+            call_sql_value_to_value!("a", ConcreteDataType::boolean_datatype(), &sql_val)?
        );

        let sql_val = SqlValue::Number("3.0".to_string(), false);
        assert_eq!(
            Value::Float64(OrderedFloat(3.0)),
-            sql_value_to_value(
-                "a",
-                &ConcreteDataType::float64_datatype(),
-                &sql_val,
-                None,
-                None,
-                false
-            )
-            .unwrap()
+            call_sql_value_to_value!("a", ConcreteDataType::float64_datatype(), &sql_val)?
        );

        let sql_val = SqlValue::Number("3.0".to_string(), false);
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::boolean_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        );
+        let v = call_sql_value_to_value!("a", ConcreteDataType::boolean_datatype(), &sql_val);
        assert!(v.is_err());
        assert!(format!("{v:?}").contains("Failed to parse number '3.0' to boolean column type"));

        let sql_val = SqlValue::Boolean(true);
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::float64_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        );
+        let v = call_sql_value_to_value!("a", ConcreteDataType::float64_datatype(), &sql_val);
        assert!(v.is_err());
        assert!(
            format!("{v:?}").contains(
@@ -725,41 +681,18 @@ mod test {
        );

        let sql_val = SqlValue::HexStringLiteral("48656c6c6f20776f726c6421".to_string());
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::binary_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        )
-        .unwrap();
+        let v = call_sql_value_to_value!("a", ConcreteDataType::binary_datatype(), &sql_val)?;
        assert_eq!(Value::Binary(Bytes::from(b"Hello world!".as_slice())), v);

        let sql_val = SqlValue::DoubleQuotedString("MorningMyFriends".to_string());
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::binary_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        )
-        .unwrap();
+        let v = call_sql_value_to_value!("a", ConcreteDataType::binary_datatype(), &sql_val)?;
        assert_eq!(
            Value::Binary(Bytes::from(b"MorningMyFriends".as_slice())),
            v
        );

        let sql_val = SqlValue::HexStringLiteral("9AF".to_string());
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::binary_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        );
+        let v = call_sql_value_to_value!("a", ConcreteDataType::binary_datatype(), &sql_val);
        assert!(v.is_err());
        assert!(
            format!("{v:?}").contains("odd number of digits"),
@@ -767,38 +700,16 @@ mod test {
        );

        let sql_val = SqlValue::HexStringLiteral("AG".to_string());
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::binary_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        );
+        let v = call_sql_value_to_value!("a", ConcreteDataType::binary_datatype(), &sql_val);
        assert!(v.is_err());
        assert!(format!("{v:?}").contains("invalid character"), "v is {v:?}",);

        let sql_val = SqlValue::DoubleQuotedString("MorningMyFriends".to_string());
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::json_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        );
+        let v = call_sql_value_to_value!("a", ConcreteDataType::json_datatype(), &sql_val);
        assert!(v.is_err());

        let sql_val = SqlValue::DoubleQuotedString(r#"{"a":"b"}"#.to_string());
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::json_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        )
-        .unwrap();
+        let v = call_sql_value_to_value!("a", ConcreteDataType::json_datatype(), &sql_val)?;
        assert_eq!(
            Value::Binary(Bytes::from(
                jsonb::parse_value(r#"{"a":"b"}"#.as_bytes())
@@ -808,16 +719,15 @@ mod test {
            )),
            v
        );
+        Ok(())
    }

    #[test]
    fn test_parse_json_to_jsonb() {
-        match parse_string_to_value(
+        match call_parse_string_to_value!(
            "json_col",
            r#"{"a": "b"}"#.to_string(),
-            &ConcreteDataType::json_datatype(),
-            None,
-            false,
+            ConcreteDataType::json_datatype()
        ) {
            Ok(Value::Binary(b)) => {
                assert_eq!(
@@ -833,12 +743,10 @@ mod test {
        }

        assert!(
-            parse_string_to_value(
+            call_parse_string_to_value!(
                "json_col",
                r#"Nicola Kovac is the best rifler in the world"#.to_string(),
-                &ConcreteDataType::json_datatype(),
-                None,
-                false,
+                ConcreteDataType::json_datatype()
            )
            .is_err()
        )
@@ -878,13 +786,10 @@ mod test {

    #[test]
    fn test_parse_date_literal() {
-        let value = sql_value_to_value(
+        let value = call_sql_value_to_value!(
            "date",
-            &ConcreteDataType::date_datatype(),
-            &SqlValue::DoubleQuotedString("2022-02-22".to_string()),
-            None,
-            None,
-            false,
+            ConcreteDataType::date_datatype(),
+            &SqlValue::DoubleQuotedString("2022-02-22".to_string())
        )
        .unwrap();
        assert_eq!(ConcreteDataType::date_datatype(), value.data_type());
@@ -895,13 +800,11 @@ mod test {
        }

        // with timezone
-        let value = sql_value_to_value(
+        let value = call_sql_value_to_value!(
            "date",
-            &ConcreteDataType::date_datatype(),
+            ConcreteDataType::date_datatype(),
            &SqlValue::DoubleQuotedString("2022-02-22".to_string()),
-            Some(&Timezone::from_tz_string("+07:00").unwrap()),
-            None,
-            false,
+            timezone = &Timezone::from_tz_string("+07:00").unwrap()
        )
        .unwrap();
        assert_eq!(ConcreteDataType::date_datatype(), value.data_type());
@@ -913,16 +816,12 @@ mod test {
    }

    #[test]
-    fn test_parse_timestamp_literal() {
-        match parse_string_to_value(
+    fn test_parse_timestamp_literal() -> Result<()> {
+        match call_parse_string_to_value!(
            "timestamp_col",
            "2022-02-22T00:01:01+08:00".to_string(),
-            &ConcreteDataType::timestamp_millisecond_datatype(),
-            None,
-            false,
-        )
-        .unwrap()
-        {
+            ConcreteDataType::timestamp_millisecond_datatype()
+        )? {
            Value::Timestamp(ts) => {
                assert_eq!(1645459261000, ts.value());
                assert_eq!(TimeUnit::Millisecond, ts.unit());
@@ -932,15 +831,11 @@ mod test {
            }
        }

-        match parse_string_to_value(
+        match call_parse_string_to_value!(
            "timestamp_col",
            "2022-02-22T00:01:01+08:00".to_string(),
-            &ConcreteDataType::timestamp_datatype(TimeUnit::Second),
-            None,
-            false,
-        )
-        .unwrap()
-        {
+            ConcreteDataType::timestamp_datatype(TimeUnit::Second)
+        )? {
            Value::Timestamp(ts) => {
                assert_eq!(1645459261, ts.value());
                assert_eq!(TimeUnit::Second, ts.unit());
@@ -950,15 +845,11 @@ mod test {
            }
        }

-        match parse_string_to_value(
+        match call_parse_string_to_value!(
            "timestamp_col",
            "2022-02-22T00:01:01+08:00".to_string(),
-            &ConcreteDataType::timestamp_datatype(TimeUnit::Microsecond),
-            None,
-            false,
-        )
-        .unwrap()
-        {
+            ConcreteDataType::timestamp_datatype(TimeUnit::Microsecond)
+        )? {
            Value::Timestamp(ts) => {
                assert_eq!(1645459261000000, ts.value());
                assert_eq!(TimeUnit::Microsecond, ts.unit());
@@ -968,15 +859,11 @@ mod test {
            }
        }

-        match parse_string_to_value(
+        match call_parse_string_to_value!(
            "timestamp_col",
            "2022-02-22T00:01:01+08:00".to_string(),
-            &ConcreteDataType::timestamp_datatype(TimeUnit::Nanosecond),
-            None,
-            false,
-        )
-        .unwrap()
-        {
+            ConcreteDataType::timestamp_datatype(TimeUnit::Nanosecond)
+        )? {
            Value::Timestamp(ts) => {
                assert_eq!(1645459261000000000, ts.value());
                assert_eq!(TimeUnit::Nanosecond, ts.unit());
@@ -987,26 +874,21 @@ mod test {
        }

        assert!(
-            parse_string_to_value(
+            call_parse_string_to_value!(
                "timestamp_col",
                "2022-02-22T00:01:01+08".to_string(),
-                &ConcreteDataType::timestamp_datatype(TimeUnit::Nanosecond),
-                None,
-                false,
+                ConcreteDataType::timestamp_datatype(TimeUnit::Nanosecond)
            )
            .is_err()
        );

        // with timezone
-        match parse_string_to_value(
+        match call_parse_string_to_value!(
            "timestamp_col",
            "2022-02-22T00:01:01".to_string(),
-            &ConcreteDataType::timestamp_datatype(TimeUnit::Nanosecond),
-            Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap()),
-            false,
-        )
-        .unwrap()
-        {
+            ConcreteDataType::timestamp_datatype(TimeUnit::Nanosecond),
+            timezone = &Timezone::from_tz_string("Asia/Shanghai").unwrap()
+        )? {
            Value::Timestamp(ts) => {
                assert_eq!(1645459261000000000, ts.value());
                assert_eq!("2022-02-21 16:01:01+0000", ts.to_iso8601_string());
@@ -1016,51 +898,42 @@ mod test {
                unreachable!()
            }
        }
+        Ok(())
    }

    #[test]
    fn test_parse_placeholder_value() {
        assert!(
-            sql_value_to_value(
+            call_sql_value_to_value!(
                "test",
-                &ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+                &SqlValue::Placeholder("default".into())
+            )
+            .is_err()
+        );
+        assert!(
+            call_sql_value_to_value!(
+                "test",
+                ConcreteDataType::string_datatype(),
                &SqlValue::Placeholder("default".into()),
-                None,
-                None,
-                false
+                unary_op = UnaryOperator::Minus
            )
            .is_err()
        );
        assert!(
-            sql_value_to_value(
+            call_sql_value_to_value!(
                "test",
-                &ConcreteDataType::string_datatype(),
-                &SqlValue::Placeholder("default".into()),
-                None,
-                Some(UnaryOperator::Minus),
-                false
-            )
-            .is_err()
-        );
-        assert!(
-            sql_value_to_value(
-                "test",
-                &ConcreteDataType::uint16_datatype(),
+                ConcreteDataType::uint16_datatype(),
                &SqlValue::Number("3".into(), false),
-                None,
-                Some(UnaryOperator::Minus),
-                false
+                unary_op = UnaryOperator::Minus
            )
            .is_err()
        );
        assert!(
-            sql_value_to_value(
+            call_sql_value_to_value!(
                "test",
-                &ConcreteDataType::uint16_datatype(),
-                &SqlValue::Number("3".into(), false),
-                None,
-                None,
-                false
+                ConcreteDataType::uint16_datatype(),
+                &SqlValue::Number("3".into(), false)
            )
            .is_ok()
        );
@@ -1070,77 +943,60 @@ mod test {
    fn test_auto_string_to_numeric() {
        // Test with auto_string_to_numeric=true
        let sql_val = SqlValue::SingleQuotedString("123".to_string());
-        let v = sql_value_to_value(
+        let v = call_sql_value_to_value!(
            "a",
-            &ConcreteDataType::int32_datatype(),
+            ConcreteDataType::int32_datatype(),
            &sql_val,
-            None,
-            None,
-            true,
+            auto_string_to_numeric
        )
        .unwrap();
        assert_eq!(Value::Int32(123), v);

        // Test with a float string
        let sql_val = SqlValue::SingleQuotedString("3.5".to_string());
-        let v = sql_value_to_value(
+        let v = call_sql_value_to_value!(
            "a",
-            &ConcreteDataType::float64_datatype(),
+            ConcreteDataType::float64_datatype(),
            &sql_val,
-            None,
-            None,
-            true,
+            auto_string_to_numeric
        )
        .unwrap();
        assert_eq!(Value::Float64(OrderedFloat(3.5)), v);

        // Test with auto_string_to_numeric=false
        let sql_val = SqlValue::SingleQuotedString("123".to_string());
-        let v = sql_value_to_value(
-            "a",
-            &ConcreteDataType::int32_datatype(),
-            &sql_val,
-            None,
-            None,
-            false,
-        );
+        let v = call_sql_value_to_value!("a", ConcreteDataType::int32_datatype(), &sql_val);
        assert!(v.is_err());

        // Test with an invalid numeric string but auto_string_to_numeric=true
        // Should return an error now with the new auto_cast_to_numeric behavior
        let sql_val = SqlValue::SingleQuotedString("not_a_number".to_string());
-        let v = sql_value_to_value(
+        let v = call_sql_value_to_value!(
            "a",
-            &ConcreteDataType::int32_datatype(),
+            ConcreteDataType::int32_datatype(),
            &sql_val,
-            None,
-            None,
-            true,
+            auto_string_to_numeric
        );
        assert!(v.is_err());

        // Test with boolean type
        let sql_val = SqlValue::SingleQuotedString("true".to_string());
-        let v = sql_value_to_value(
+        let v = call_sql_value_to_value!(
            "a",
-            &ConcreteDataType::boolean_datatype(),
+            ConcreteDataType::boolean_datatype(),
            &sql_val,
-            None,
-            None,
-            true,
+            auto_string_to_numeric
        )
        .unwrap();
        assert_eq!(Value::Boolean(true), v);

        // Non-numeric types should still be handled normally
        let sql_val = SqlValue::SingleQuotedString("hello".to_string());
-        let v = sql_value_to_value(
+        let v = call_sql_value_to_value!(
            "a",
-            &ConcreteDataType::string_datatype(),
+            ConcreteDataType::string_datatype(),
            &sql_val,
-            None,
-            None,
-            true,
+            auto_string_to_numeric
        );
        assert!(v.is_ok());
    }
--- a/src/common/sql/src/default_constraint.rs
+++ b/src/common/sql/src/default_constraint.rs
@@ -14,8 +14,8 @@

 use common_time::timezone::Timezone;
 use datatypes::prelude::ConcreteDataType;
-use datatypes::schema::ColumnDefaultConstraint;
 use datatypes::schema::constraint::{CURRENT_TIMESTAMP, CURRENT_TIMESTAMP_FN};
+use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema};
 use snafu::ensure;
 use sqlparser::ast::ValueWithSpan;
 pub use sqlparser::ast::{
@@ -47,9 +47,12 @@ pub fn parse_column_default_constraint(
        );

        let default_constraint = match &opt.option {
-            ColumnOption::Default(Expr::Value(v)) => ColumnDefaultConstraint::Value(
-                sql_value_to_value(column_name, data_type, &v.value, timezone, None, false)?,
-            ),
+            ColumnOption::Default(Expr::Value(v)) => {
+                let schema = ColumnSchema::new(column_name, data_type.clone(), true);
+                ColumnDefaultConstraint::Value(sql_value_to_value(
+                    &schema, &v.value, timezone, None, false,
+                )?)
+            }
            ColumnOption::Default(Expr::Function(func)) => {
                let mut func = format!("{func}").to_lowercase();
                // normalize CURRENT_TIMESTAMP to CURRENT_TIMESTAMP()
@@ -80,8 +83,7 @@ pub fn parse_column_default_constraint(

                if let Expr::Value(v) = &**expr {
                    let value = sql_value_to_value(
-                        column_name,
-                        data_type,
+                        &ColumnSchema::new(column_name, data_type.clone(), true),
                        &v.value,
                        timezone,
                        Some(*op),
--- a/src/datatypes/src/json.rs
+++ b/src/datatypes/src/json.rs
@@ -26,9 +26,9 @@ use std::sync::Arc;

 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value as Json};
-use snafu::{ResultExt, ensure};
+use snafu::{OptionExt, ResultExt, ensure};

-use crate::error::{self, Error};
+use crate::error::{self, InvalidJsonSnafu, Result, SerializeSnafu};
 use crate::json::value::{JsonValue, JsonVariant};
 use crate::types::json_type::{JsonNativeType, JsonNumberType, JsonObjectType};
 use crate::types::{StructField, StructType};
@@ -71,7 +71,7 @@ impl JsonStructureSettings {
    pub const RAW_FIELD: &'static str = "_raw";

    /// Decode an encoded StructValue back into a serde_json::Value.
-    pub fn decode(&self, value: Value) -> Result<Json, Error> {
+    pub fn decode(&self, value: Value) -> Result<Json> {
        let context = JsonContext {
            key_path: String::new(),
            settings: self,
@@ -82,7 +82,7 @@ impl JsonStructureSettings {
    /// Decode a StructValue that was encoded with current settings back into a fully structured StructValue.
    /// This is useful for reconstructing the original structure from encoded data, especially when
    /// unstructured encoding was used for some fields.
-    pub fn decode_struct(&self, struct_value: StructValue) -> Result<StructValue, Error> {
+    pub fn decode_struct(&self, struct_value: StructValue) -> Result<StructValue> {
        let context = JsonContext {
            key_path: String::new(),
            settings: self,
@@ -91,7 +91,11 @@ impl JsonStructureSettings {
    }

    /// Encode a serde_json::Value into a Value::Json using current settings.
-    pub fn encode(&self, json: Json) -> Result<Value, Error> {
+    pub fn encode(&self, json: Json) -> Result<Value> {
+        if let Some(json_struct) = self.json_struct() {
+            return encode_by_struct(json_struct, json);
+        }
+
        let context = JsonContext {
            key_path: String::new(),
            settings: self,
@@ -104,13 +108,21 @@ impl JsonStructureSettings {
        &self,
        json: Json,
        data_type: Option<&JsonNativeType>,
-    ) -> Result<Value, Error> {
+    ) -> Result<Value> {
        let context = JsonContext {
            key_path: String::new(),
            settings: self,
        };
        encode_json_with_context(json, data_type, &context).map(|v| Value::Json(Box::new(v)))
    }
+
+    fn json_struct(&self) -> Option<&StructType> {
+        match &self {
+            JsonStructureSettings::Structured(fields) => fields.as_ref(),
+            JsonStructureSettings::PartialUnstructuredByKey { fields, .. } => fields.as_ref(),
+            _ => None,
+        }
+    }
 }

 impl Default for JsonStructureSettings {
@@ -144,12 +156,54 @@ impl<'a> JsonContext<'a> {
    }
 }

+fn encode_by_struct(json_struct: &StructType, mut json: Json) -> Result<Value> {
+    let Some(json_object) = json.as_object_mut() else {
+        return InvalidJsonSnafu {
+            value: "expect JSON object when struct is provided",
+        }
+        .fail();
+    };
+    let mut encoded = BTreeMap::new();
+
+    fn extract_field(json_object: &mut Map<String, Json>, field: &str) -> Result<Option<Json>> {
+        let (first, rest) = field.split_once('.').unwrap_or((field, ""));
+
+        if rest.is_empty() {
+            Ok(json_object.remove(first))
+        } else {
+            let Some(value) = json_object.get_mut(first) else {
+                return Ok(None);
+            };
+            let json_object = value.as_object_mut().with_context(|| InvalidJsonSnafu {
+                value: format!(r#"expect "{}" an object"#, first),
+            })?;
+            extract_field(json_object, rest)
+        }
+    }
+
+    let fields = json_struct.fields();
+    for field in fields.iter() {
+        let Some(field_value) = extract_field(json_object, field.name())? else {
+            continue;
+        };
+        let field_type: JsonNativeType = field.data_type().into();
+        let field_value = try_convert_to_expected_type(field_value, &field_type)?;
+        encoded.insert(field.name().to_string(), field_value);
+    }
+
+    let rest = serde_json::to_string(json_object).context(SerializeSnafu)?;
+    encoded.insert(JsonStructureSettings::RAW_FIELD.to_string(), rest.into());
+
+    let value: JsonValue = encoded.into();
+    Ok(Value::Json(Box::new(value)))
+}
+
 /// Main encoding function with key path tracking
 pub fn encode_json_with_context<'a>(
    json: Json,
    data_type: Option<&JsonNativeType>,
    context: &JsonContext<'a>,
-) -> Result<JsonValue, Error> {
+) -> Result<JsonValue> {
    // Check if the entire encoding should be unstructured
    if matches!(context.settings, JsonStructureSettings::UnstructuredRaw) {
        let json_string = json.to_string();
@@ -215,7 +269,7 @@ fn encode_json_object_with_context<'a>(
    mut json_object: Map<String, Json>,
    fields: Option<&JsonObjectType>,
    context: &JsonContext<'a>,
-) -> Result<JsonValue, Error> {
+) -> Result<JsonValue> {
    let mut object = BTreeMap::new();
    // First, process fields from the provided schema in their original order
    if let Some(fields) = fields {
@@ -248,7 +302,7 @@ fn encode_json_array_with_context<'a>(
    json_array: Vec<Json>,
    item_type: Option<&JsonNativeType>,
    context: &JsonContext<'a>,
-) -> Result<JsonValue, Error> {
+) -> Result<JsonValue> {
    let json_array_len = json_array.len();
    let mut items = Vec::with_capacity(json_array_len);
    let mut element_type = item_type.cloned();
@@ -286,7 +340,7 @@ fn encode_json_value_with_context<'a>(
    json: Json,
    expected_type: Option<&JsonNativeType>,
    context: &JsonContext<'a>,
-) -> Result<JsonValue, Error> {
+) -> Result<JsonValue> {
    // Check if current key should be treated as unstructured
    if context.is_unstructured_key() {
        return Ok(json.to_string().into());
@@ -301,7 +355,7 @@ fn encode_json_value_with_context<'a>(
                if let Some(expected) = expected_type
                    && let Ok(value) = try_convert_to_expected_type(i, expected)
                {
-                    return Ok(value);
+                    return Ok(value.into());
                }
                Ok(i.into())
            } else if let Some(u) = n.as_u64() {
@@ -309,7 +363,7 @@ fn encode_json_value_with_context<'a>(
                if let Some(expected) = expected_type
                    && let Ok(value) = try_convert_to_expected_type(u, expected)
                {
-                    return Ok(value);
+                    return Ok(value.into());
                }
                if u <= i64::MAX as u64 {
                    Ok((u as i64).into())
@@ -321,7 +375,7 @@ fn encode_json_value_with_context<'a>(
                if let Some(expected) = expected_type
                    && let Ok(value) = try_convert_to_expected_type(f, expected)
                {
-                    return Ok(value);
+                    return Ok(value.into());
                }

                // Default to f64 for floating point numbers
@@ -335,7 +389,7 @@ fn encode_json_value_with_context<'a>(
            if let Some(expected) = expected_type
                && let Ok(value) = try_convert_to_expected_type(s.as_str(), expected)
            {
-                return Ok(value);
+                return Ok(value.into());
            }
            Ok(s.into())
        }
@@ -345,10 +399,7 @@ fn encode_json_value_with_context<'a>(
 }

 /// Main decoding function with key path tracking
-pub fn decode_value_with_context<'a>(
-    value: Value,
-    context: &JsonContext<'a>,
-) -> Result<Json, Error> {
+pub fn decode_value_with_context(value: Value, context: &JsonContext) -> Result<Json> {
    // Check if the entire decoding should be unstructured
    if matches!(context.settings, JsonStructureSettings::UnstructuredRaw) {
        return decode_unstructured_value(value);
@@ -370,7 +421,7 @@ pub fn decode_value_with_context<'a>(
 fn decode_struct_with_context<'a>(
    struct_value: StructValue,
    context: &JsonContext<'a>,
-) -> Result<Json, Error> {
+) -> Result<Json> {
    let mut json_object = Map::with_capacity(struct_value.len());

    let (items, fields) = struct_value.into_parts();
@@ -385,10 +436,7 @@ fn decode_struct_with_context<'a>(
 }

 /// Decode a list value to JSON array
-fn decode_list_with_context<'a>(
-    list_value: ListValue,
-    context: &JsonContext<'a>,
-) -> Result<Json, Error> {
+fn decode_list_with_context(list_value: ListValue, context: &JsonContext) -> Result<Json> {
    let mut json_array = Vec::with_capacity(list_value.len());

    let data_items = list_value.take_items();
@@ -403,7 +451,7 @@ fn decode_list_with_context<'a>(
 }

 /// Decode unstructured value (stored as string)
-fn decode_unstructured_value(value: Value) -> Result<Json, Error> {
+fn decode_unstructured_value(value: Value) -> Result<Json> {
    match value {
        // Handle expected format: StructValue with single _raw field
        Value::Struct(struct_value) => {
@@ -443,7 +491,7 @@ fn decode_unstructured_value(value: Value) -> Result<Json, Error> {
 }

 /// Decode primitive value to JSON
-fn decode_primitive_value(value: Value) -> Result<Json, Error> {
+fn decode_primitive_value(value: Value) -> Result<Json> {
    match value {
        Value::Null => Ok(Json::Null),
        Value::Boolean(b) => Ok(Json::Bool(b)),
@@ -487,7 +535,7 @@ fn decode_primitive_value(value: Value) -> Result<Json, Error> {
 fn decode_struct_with_settings<'a>(
    struct_value: StructValue,
    context: &JsonContext<'a>,
-) -> Result<StructValue, Error> {
+) -> Result<StructValue> {
    // Check if we can return the struct directly (Structured case)
    if matches!(context.settings, JsonStructureSettings::Structured(_)) {
        return Ok(struct_value);
@@ -567,7 +615,7 @@ fn decode_struct_with_settings<'a>(
 fn decode_list_with_settings<'a>(
    list_value: ListValue,
    context: &JsonContext<'a>,
-) -> Result<ListValue, Error> {
+) -> Result<ListValue> {
    let mut items = Vec::with_capacity(list_value.len());

    let (data_items, datatype) = list_value.into_parts();
@@ -592,7 +640,7 @@ fn decode_list_with_settings<'a>(
 }

 /// Helper function to decode a struct that was encoded with UnstructuredRaw settings
-fn decode_unstructured_raw_struct(struct_value: StructValue) -> Result<StructValue, Error> {
+fn decode_unstructured_raw_struct(struct_value: StructValue) -> Result<StructValue> {
    // For UnstructuredRaw, the struct must have exactly one field named "_raw"
    if struct_value.struct_type().fields().len() == 1 {
        let field = &struct_value.struct_type().fields()[0];
@@ -636,12 +684,9 @@ fn decode_unstructured_raw_struct(struct_value: StructValue) -> Result<StructVal
 }

 /// Helper function to try converting a value to an expected type
-fn try_convert_to_expected_type<T>(
-    value: T,
-    expected_type: &JsonNativeType,
-) -> Result<JsonValue, Error>
+fn try_convert_to_expected_type<T>(value: T, expected_type: &JsonNativeType) -> Result<JsonVariant>
 where
-    T: Into<JsonValue>,
+    T: Into<JsonVariant>,
 {
    let value = value.into();
    let cast_error = || {
@@ -650,7 +695,7 @@ where
        }
        .fail()
    };
-    let actual_type = value.json_type().native_type();
+    let actual_type = &value.native_type();
    match (actual_type, expected_type) {
        (x, y) if x == y => Ok(value),
        (JsonNativeType::Number(x), JsonNativeType::Number(y)) => match (x, y) {
@@ -691,6 +736,107 @@ mod tests {
    use crate::data_type::ConcreteDataType;
    use crate::types::ListType;

+    #[test]
+    fn test_encode_by_struct() {
+        let json_struct: StructType = [
+            StructField::new("s", ConcreteDataType::string_datatype(), true),
+            StructField::new("foo.i", ConcreteDataType::int64_datatype(), true),
+            StructField::new("x.y.z", ConcreteDataType::boolean_datatype(), true),
+        ]
+        .into();
+
+        let json = json!({
+            "s": "hello",
+            "t": "world",
+            "foo": {
+                "i": 1,
+                "j": 2
+            },
+            "x": {
+                "y": {
+                    "z": true
+                }
+            }
+        });
+        let value = encode_by_struct(&json_struct, json).unwrap();
+        assert_eq!(
+            value.to_string(),
+            r#"Json({ _raw: {"foo":{"j":2},"t":"world","x":{"y":{}}}, foo.i: 1, s: hello, x.y.z: true })"#
+        );
+
+        let json = json!({
+            "t": "world",
+            "foo": {
+                "i": 1,
+                "j": 2
+            },
+            "x": {
+                "y": {
+                    "z": true
+                }
+            }
+        });
+        let value = encode_by_struct(&json_struct, json).unwrap();
+        assert_eq!(
+            value.to_string(),
+            r#"Json({ _raw: {"foo":{"j":2},"t":"world","x":{"y":{}}}, foo.i: 1, x.y.z: true })"#
+        );
+
+        let json = json!({
+            "s": 1234,
+            "foo": {
+                "i": 1,
+                "j": 2
+            },
+            "x": {
+                "y": {
+                    "z": true
+                }
+            }
+        });
+        let value = encode_by_struct(&json_struct, json).unwrap();
+        assert_eq!(
+            value.to_string(),
+            r#"Json({ _raw: {"foo":{"j":2},"x":{"y":{}}}, foo.i: 1, s: 1234, x.y.z: true })"#
+        );
+
+        let json = json!({
+            "s": "hello",
+            "t": "world",
+            "foo": {
+                "i": "bar",
+                "j": 2
+            },
+            "x": {
+                "y": {
+                    "z": true
+                }
+            }
+        });
+        let result = encode_by_struct(&json_struct, json);
+        assert_eq!(
+            result.unwrap_err().to_string(),
+            "Cannot cast value bar to Number(I64)"
+        );
+
+        let json = json!({
+            "s": "hello",
+            "t": "world",
+            "foo": {
+                "i": 1,
+                "j": 2
+            },
+            "x": {
+                "y": "z"
+            }
+        });
+        let result = encode_by_struct(&json_struct, json);
+        assert_eq!(
+            result.unwrap_err().to_string(),
+            r#"Invalid JSON: expect "y" an object"#
+        );
+    }
+
    #[test]
    fn test_encode_json_null() {
        let json = Json::Null;
--- a/src/datatypes/src/json/value.rs
+++ b/src/datatypes/src/json/value.rs
@@ -82,6 +82,18 @@ impl From<f64> for JsonNumber {
    }
 }

+impl From<Number> for JsonNumber {
+    fn from(n: Number) -> Self {
+        if let Some(i) = n.as_i64() {
+            i.into()
+        } else if let Some(i) = n.as_u64() {
+            i.into()
+        } else {
+            n.as_f64().unwrap_or(f64::NAN).into()
+        }
+    }
+}
+
 impl Display for JsonNumber {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
@@ -109,7 +121,28 @@ pub enum JsonVariant {
 }

 impl JsonVariant {
-    fn native_type(&self) -> JsonNativeType {
+    pub(crate) fn as_i64(&self) -> Option<i64> {
+        match self {
+            JsonVariant::Number(n) => n.as_i64(),
+            _ => None,
+        }
+    }
+
+    pub(crate) fn as_u64(&self) -> Option<u64> {
+        match self {
+            JsonVariant::Number(n) => n.as_u64(),
+            _ => None,
+        }
+    }
+
+    pub(crate) fn as_f64(&self) -> Option<f64> {
+        match self {
+            JsonVariant::Number(n) => Some(n.as_f64()),
+            _ => None,
+        }
+    }
+
+    pub(crate) fn native_type(&self) -> JsonNativeType {
        match self {
            JsonVariant::Null => JsonNativeType::Null,
            JsonVariant::Bool(_) => JsonNativeType::Bool,
@@ -205,6 +238,32 @@ impl<K: Into<String>, V: Into<JsonVariant>, const N: usize> From<[(K, V); N]> fo
    }
 }

+impl From<serde_json::Value> for JsonVariant {
+    fn from(v: serde_json::Value) -> Self {
+        fn helper(v: serde_json::Value) -> JsonVariant {
+            match v {
+                serde_json::Value::Null => JsonVariant::Null,
+                serde_json::Value::Bool(b) => b.into(),
+                serde_json::Value::Number(n) => n.into(),
+                serde_json::Value::String(s) => s.into(),
+                serde_json::Value::Array(array) => {
+                    JsonVariant::Array(array.into_iter().map(helper).collect())
+                }
+                serde_json::Value::Object(object) => {
+                    JsonVariant::Object(object.into_iter().map(|(k, v)| (k, helper(v))).collect())
+                }
+            }
+        }
+        helper(v)
+    }
+}
+
+impl From<BTreeMap<String, JsonVariant>> for JsonVariant {
+    fn from(v: BTreeMap<String, JsonVariant>) -> Self {
+        Self::Object(v)
+    }
+}
+
 impl Display for JsonVariant {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
@@ -277,24 +336,11 @@ impl JsonValue {
    }

    pub(crate) fn as_i64(&self) -> Option<i64> {
-        match self.json_variant {
-            JsonVariant::Number(n) => n.as_i64(),
-            _ => None,
-        }
+        self.json_variant.as_i64()
    }

    pub(crate) fn as_u64(&self) -> Option<u64> {
-        match self.json_variant {
-            JsonVariant::Number(n) => n.as_u64(),
-            _ => None,
-        }
-    }
-
-    pub(crate) fn as_f64(&self) -> Option<f64> {
-        match self.json_variant {
-            JsonVariant::Number(n) => Some(n.as_f64()),
-            _ => None,
-        }
+        self.json_variant.as_u64()
    }

    pub(crate) fn as_f64_lossy(&self) -> Option<f64> {
--- a/src/datatypes/src/types/struct_type.rs
+++ b/src/datatypes/src/types/struct_type.rs
@@ -122,9 +122,9 @@ pub struct StructField {
 }

 impl StructField {
-    pub fn new(name: String, data_type: ConcreteDataType, nullable: bool) -> Self {
+    pub fn new<T: Into<String>>(name: T, data_type: ConcreteDataType, nullable: bool) -> Self {
        StructField {
-            name,
+            name: name.into(),
            data_type,
            nullable,
            metadata: BTreeMap::new(),
--- a/src/operator/src/req_convert/insert/stmt_to_region.rs
+++ b/src/operator/src/req_convert/insert/stmt_to_region.rs
@@ -410,8 +410,7 @@ fn sql_value_to_value(
        })?
    } else {
        common_sql::convert::sql_value_to_value(
-            column,
-            &column_schema.data_type,
+            column_schema,
            sql_val,
            timezone,
            None,
--- a/src/operator/src/statement.rs
+++ b/src/operator/src/statement.rs
@@ -52,6 +52,7 @@ use common_time::Timestamp;
 use common_time::range::TimestampRange;
 use datafusion_expr::LogicalPlan;
 use datatypes::prelude::ConcreteDataType;
+use datatypes::schema::ColumnSchema;
 use humantime::format_duration;
 use itertools::Itertools;
 use partition::manager::{PartitionRuleManager, PartitionRuleManagerRef};
@@ -644,11 +645,20 @@ impl StatementExecutor {
            })?
            .unit();

+        let start_column = ColumnSchema::new(
+            "range_start",
+            ConcreteDataType::timestamp_datatype(time_unit),
+            false,
+        );
+        let end_column = ColumnSchema::new(
+            "range_end",
+            ConcreteDataType::timestamp_datatype(time_unit),
+            false,
+        );
        let mut time_ranges = Vec::with_capacity(sql_values_time_range.len());
        for (start, end) in sql_values_time_range {
            let start = common_sql::convert::sql_value_to_value(
-                "range_start",
-                &ConcreteDataType::timestamp_datatype(time_unit),
+                &start_column,
                start,
                Some(&query_ctx.timezone()),
                None,
@@ -667,8 +677,7 @@ impl StatementExecutor {
            })?;

            let end = common_sql::convert::sql_value_to_value(
-                "range_end",
-                &ConcreteDataType::timestamp_datatype(time_unit),
+                &end_column,
                end,
                Some(&query_ctx.timezone()),
                None,
--- a/src/operator/src/statement/admin.rs
+++ b/src/operator/src/statement/admin.rs
@@ -242,8 +242,12 @@ fn values_to_vectors_by_exact_types(
    args.iter()
        .zip(exact_types.iter())
        .map(|(value, data_type)| {
-            let data_type = &ConcreteDataType::from_arrow_type(data_type);
-            let value = sql_value_to_value(DUMMY_COLUMN, data_type, value, tz, None, false)
+            let schema = ColumnSchema::new(
+                DUMMY_COLUMN,
+                ConcreteDataType::from_arrow_type(data_type),
+                true,
+            );
+            let value = sql_value_to_value(&schema, value, tz, None, false)
                .context(error::SqlCommonSnafu)?;

            Ok(value_to_vector(value))
@@ -260,10 +264,12 @@ fn values_to_vectors_by_valid_types(
    args.iter()
        .map(|value| {
            for data_type in valid_types {
-                let data_type = &ConcreteDataType::from_arrow_type(data_type);
-                if let Ok(value) =
-                    sql_value_to_value(DUMMY_COLUMN, data_type, value, tz, None, false)
-                {
+                let schema = ColumnSchema::new(
+                    DUMMY_COLUMN,
+                    ConcreteDataType::from_arrow_type(data_type),
+                    true,
+                );
+                if let Ok(value) = sql_value_to_value(&schema, value, tz, None, false) {
                    return Ok(value_to_vector(value));
                }
            }
--- a/src/operator/src/statement/ddl.rs
+++ b/src/operator/src/statement/ddl.rs
@@ -50,7 +50,7 @@ use common_time::{Timestamp, Timezone};
 use datafusion_common::tree_node::TreeNodeVisitor;
 use datafusion_expr::LogicalPlan;
 use datatypes::prelude::ConcreteDataType;
-use datatypes::schema::{RawSchema, Schema};
+use datatypes::schema::{ColumnSchema, RawSchema, Schema};
 use datatypes::value::Value;
 use partition::expr::{Operand, PartitionExpr, RestrictedOp};
 use partition::multi_dim::MultiDimPartitionRule;
@@ -2001,8 +2001,7 @@ fn convert_value(
    unary_op: Option<UnaryOperator>,
 ) -> Result<Value> {
    sql_value_to_value(
-        "<NONAME>",
-        &data_type,
+        &ColumnSchema::new("<NONAME>", data_type, true),
        value,
        Some(timezone),
        unary_op,
--- a/src/servers/src/mysql/helper.rs
+++ b/src/servers/src/mysql/helper.rs
@@ -22,6 +22,7 @@ use common_time::{Date, Timestamp};
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_expr::LogicalPlan;
 use datatypes::prelude::ConcreteDataType;
+use datatypes::schema::ColumnSchema;
 use datatypes::types::TimestampType;
 use datatypes::value::{self, Value};
 use itertools::Itertools;
@@ -254,9 +255,10 @@ pub fn convert_value(param: &ParamValue, t: &ConcreteDataType) -> Result<ScalarV
 /// Convert an MySQL expression to a scalar value.
 /// It automatically handles the conversion of strings to numeric values.
 pub fn convert_expr_to_scalar_value(param: &Expr, t: &ConcreteDataType) -> Result<ScalarValue> {
+    let column_schema = ColumnSchema::new("", t.clone(), true);
    match param {
        Expr::Value(v) => {
-            let v = sql_value_to_value("", t, &v.value, None, None, true);
+            let v = sql_value_to_value(&column_schema, &v.value, None, None, true);
            match v {
                Ok(v) => v
                    .try_to_scalar_value(t)
@@ -268,7 +270,7 @@ pub fn convert_expr_to_scalar_value(param: &Expr, t: &ConcreteDataType) -> Resul
            }
        }
        Expr::UnaryOp { op, expr } if let Expr::Value(v) = &**expr => {
-            let v = sql_value_to_value("", t, &v.value, None, Some(*op), true);
+            let v = sql_value_to_value(&column_schema, &v.value, None, Some(*op), true);
            match v {
                Ok(v) => v
                    .try_to_scalar_value(t)
--- a/src/sql/src/dialect.rs
+++ b/src/sql/src/dialect.rs
@@ -40,4 +40,8 @@ impl Dialect for GreptimeDbDialect {
    fn supports_filter_during_aggregation(&self) -> bool {
        true
    }
+
+    fn supports_struct_literal(&self) -> bool {
+        true
+    }
 }
--- a/src/sql/src/error.rs
+++ b/src/sql/src/error.rs
@@ -215,6 +215,13 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Invalid JSON structure setting, reason: {reason}"))]
+    InvalidJsonStructureSetting {
+        reason: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Failed to serialize column default constraint"))]
    SerializeColumnDefaultConstraint {
        #[snafu(implicit)]
@@ -374,6 +381,7 @@ impl ErrorExt for Error {

            InvalidColumnOption { .. }
            | InvalidExprAsOptionValue { .. }
+            | InvalidJsonStructureSetting { .. }
            | InvalidDatabaseName { .. }
            | InvalidDatabaseOption { .. }
            | ColumnTypeMismatch { .. }
--- a/src/sql/src/parsers/create_parser/json.rs
+++ b/src/sql/src/parsers/create_parser/json.rs
@@ -40,16 +40,17 @@ pub(super) fn parse_json_datatype_options(parser: &mut Parser<'_>) -> Result<Opt

 #[cfg(test)]
 mod tests {
-    use sqlparser::ast::DataType;
+    use sqlparser::ast::{DataType, Expr, Ident, StructField};

    use crate::dialect::GreptimeDbDialect;
    use crate::parser::{ParseOptions, ParserContext};
    use crate::statements::OptionMap;
    use crate::statements::create::{
-        Column, JSON_FORMAT_FULL_STRUCTURED, JSON_FORMAT_PARTIAL, JSON_FORMAT_RAW, JSON_OPT_FORMAT,
-        JSON_OPT_UNSTRUCTURED_KEYS,
+        Column, JSON_FORMAT_FULL_STRUCTURED, JSON_FORMAT_PARTIAL, JSON_FORMAT_RAW, JSON_OPT_FIELDS,
+        JSON_OPT_FORMAT, JSON_OPT_UNSTRUCTURED_KEYS,
    };
    use crate::statements::statement::Statement;
+    use crate::util::OptionValue;

    #[test]
    fn test_parse_json_datatype_options() {
@@ -77,6 +78,42 @@ mod tests {

        let sql = r#"
 CREATE TABLE json_data (
+    my_json JSON(format = "partial", fields = Struct<i Int, "o.a" String, "o.b" String, `x.y.z` Float64>),
+    ts TIMESTAMP TIME INDEX,
+)"#;
+        let options = parse(sql).unwrap();
+        assert_eq!(options.len(), 2);
+        let option = options.value(JSON_OPT_FIELDS);
+        let expected = OptionValue::try_new(Expr::Struct {
+            values: vec![],
+            fields: vec![
+                StructField {
+                    field_name: Some(Ident::new("i")),
+                    field_type: DataType::Int(None),
+                    options: None,
+                },
+                StructField {
+                    field_name: Some(Ident::with_quote('"', "o.a")),
+                    field_type: DataType::String(None),
+                    options: None,
+                },
+                StructField {
+                    field_name: Some(Ident::with_quote('"', "o.b")),
+                    field_type: DataType::String(None),
+                    options: None,
+                },
+                StructField {
+                    field_name: Some(Ident::with_quote('`', "x.y.z")),
+                    field_type: DataType::Float64,
+                    options: None,
+                },
+            ],
+        })
+        .ok();
+        assert_eq!(option, expected.as_ref());
+
+        let sql = r#"
+CREATE TABLE json_data (
    my_json JSON(format = "partial", unstructured_keys = ["k", "foo.bar", "a.b.c"]),
    ts TIMESTAMP TIME INDEX,
 )"#;
--- a/src/sql/src/statements.rs
+++ b/src/sql/src/statements.rs
@@ -40,6 +40,7 @@ use api::v1::SemanticType;
 use common_sql::default_constraint::parse_column_default_constraint;
 use common_time::timezone::Timezone;
 use datatypes::extension::json::{JsonExtensionType, JsonMetadata};
+use datatypes::json::JsonStructureSettings;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{COMMENT_KEY, ColumnDefaultConstraint, ColumnSchema};
 use datatypes::types::json_type::JsonNativeType;
@@ -281,8 +282,17 @@ pub fn sql_data_type_to_concrete_data_type(
            }
        },
        SqlDataType::JSON => {
-            let format = if column_extensions.json_datatype_options.is_some() {
-                JsonFormat::Native(Box::new(JsonNativeType::Null))
+            let format = if let Some(x) = column_extensions.build_json_structure_settings()? {
+                if let Some(fields) = match x {
+                    JsonStructureSettings::Structured(fields) => fields,
+                    JsonStructureSettings::UnstructuredRaw => None,
+                    JsonStructureSettings::PartialUnstructuredByKey { fields, .. } => fields,
+                } {
+                    let datatype = &ConcreteDataType::Struct(fields);
+                    JsonFormat::Native(Box::new(datatype.into()))
+                } else {
+                    JsonFormat::Native(Box::new(JsonNativeType::Null))
+                }
            } else {
                JsonFormat::Jsonb
            };
--- a/src/sql/src/statements/create.rs
+++ b/src/sql/src/statements/create.rs
@@ -14,27 +14,30 @@

 use std::collections::{HashMap, HashSet};
 use std::fmt::{Display, Formatter};
+use std::sync::Arc;

 use common_catalog::consts::FILE_ENGINE;
+use datatypes::data_type::ConcreteDataType;
 use datatypes::json::JsonStructureSettings;
 use datatypes::schema::{
    FulltextOptions, SkippingIndexOptions, VectorDistanceMetric, VectorIndexEngineType,
    VectorIndexOptions,
 };
+use datatypes::types::StructType;
 use itertools::Itertools;
 use serde::Serialize;
-use snafu::ResultExt;
+use snafu::{OptionExt, ResultExt};
 use sqlparser::ast::{ColumnOptionDef, DataType, Expr, Query};
 use sqlparser_derive::{Visit, VisitMut};

 use crate::ast::{ColumnDef, Ident, ObjectName, Value as SqlValue};
 use crate::error::{
-    InvalidFlowQuerySnafu, InvalidSqlSnafu, Result, SetFulltextOptionSnafu,
-    SetSkippingIndexOptionSnafu,
+    InvalidFlowQuerySnafu, InvalidJsonStructureSettingSnafu, InvalidSqlSnafu, Result,
+    SetFulltextOptionSnafu, SetSkippingIndexOptionSnafu,
 };
-use crate::statements::OptionMap;
 use crate::statements::statement::Statement;
 use crate::statements::tql::Tql;
+use crate::statements::{OptionMap, sql_data_type_to_concrete_data_type};
 use crate::util::OptionValue;

 const LINE_SEP: &str = ",\n";
@@ -44,6 +47,7 @@ pub const VECTOR_OPT_DIM: &str = "dim";

 pub const JSON_OPT_UNSTRUCTURED_KEYS: &str = "unstructured_keys";
 pub const JSON_OPT_FORMAT: &str = "format";
+pub(crate) const JSON_OPT_FIELDS: &str = "fields";
 pub const JSON_FORMAT_FULL_STRUCTURED: &str = "structured";
 pub const JSON_FORMAT_RAW: &str = "raw";
 pub const JSON_FORMAT_PARTIAL: &str = "partial";
@@ -346,14 +350,51 @@ impl ColumnExtensions {
            })
            .unwrap_or_default();

+        let fields = if let Some(value) = options.value(JSON_OPT_FIELDS) {
+            let fields = value
+                .as_struct_fields()
+                .context(InvalidJsonStructureSettingSnafu {
+                    reason: format!(r#"expect "{JSON_OPT_FIELDS}" a struct, actual: "{value}""#,),
+                })?;
+            let fields = fields
+                .iter()
+                .map(|field| {
+                    let name = field.field_name.as_ref().map(|x| x.value.clone()).context(
+                        InvalidJsonStructureSettingSnafu {
+                            reason: format!(r#"missing field name in "{field}""#),
+                        },
+                    )?;
+                    let datatype = sql_data_type_to_concrete_data_type(
+                        &field.field_type,
+                        &Default::default(),
+                    )?;
+                    Ok(datatypes::types::StructField::new(name, datatype, true))
+                })
+                .collect::<Result<_>>()?;
+            Some(StructType::new(Arc::new(fields)))
+        } else {
+            None
+        };
+
        options
            .get(JSON_OPT_FORMAT)
            .map(|format| match format {
-                JSON_FORMAT_FULL_STRUCTURED => Ok(JsonStructureSettings::Structured(None)),
-                JSON_FORMAT_PARTIAL => Ok(JsonStructureSettings::PartialUnstructuredByKey {
-                    fields: None,
-                    unstructured_keys,
-                }),
+                JSON_FORMAT_FULL_STRUCTURED => Ok(JsonStructureSettings::Structured(fields)),
+                JSON_FORMAT_PARTIAL => {
+                    let fields = fields.map(|fields| {
+                        let mut fields = Arc::unwrap_or_clone(fields.fields());
+                        fields.push(datatypes::types::StructField::new(
+                            JsonStructureSettings::RAW_FIELD.to_string(),
+                            ConcreteDataType::string_datatype(),
+                            true,
+                        ));
+                        StructType::new(Arc::new(fields))
+                    });
+                    Ok(JsonStructureSettings::PartialUnstructuredByKey {
+                        fields,
+                        unstructured_keys,
+                    })
+                }
                JSON_FORMAT_RAW => Ok(JsonStructureSettings::UnstructuredRaw),
                _ => InvalidSqlSnafu {
                    msg: format!("unknown JSON datatype 'format': {format}"),
--- a/src/sql/src/util.rs
+++ b/src/sql/src/util.rs
@@ -19,7 +19,8 @@ use itertools::Itertools;
 use serde::Serialize;
 use snafu::ensure;
 use sqlparser::ast::{
-    Array, Expr, Ident, ObjectName, SetExpr, SqlOption, TableFactor, Value, ValueWithSpan,
+    Array, Expr, Ident, ObjectName, SetExpr, SqlOption, StructField, TableFactor, Value,
+    ValueWithSpan,
 };
 use sqlparser_derive::{Visit, VisitMut};

@@ -52,9 +53,12 @@ pub fn format_raw_object_name(name: &ObjectName) -> String {
 pub struct OptionValue(Expr);

 impl OptionValue {
-    fn try_new(expr: Expr) -> Result<Self> {
+    pub(crate) fn try_new(expr: Expr) -> Result<Self> {
        ensure!(
-            matches!(expr, Expr::Value(_) | Expr::Identifier(_) | Expr::Array(_)),
+            matches!(
+                expr,
+                Expr::Value(_) | Expr::Identifier(_) | Expr::Array(_) | Expr::Struct { .. }
+            ),
            InvalidExprAsOptionValueSnafu {
                error: format!("{expr} not accepted")
            }
@@ -106,6 +110,13 @@ impl OptionValue {
            _ => None,
        }
    }
+
+    pub(crate) fn as_struct_fields(&self) -> Option<&[StructField]> {
+        match &self.0 {
+            Expr::Struct { fields, .. } => Some(fields),
+            _ => None,
+        }
+    }
 }

 impl From<String> for OptionValue {