feat(json2): support altering JSON2 settings (#9029)

* feat(sql): support alter syntax for JSON2 columns

Signed-off-by: fys <fengys1996@gmail.com>

* fix(json2): preserve rows on type hint mismatch during compaction

* refactor(json2): simplify alter settings handling

* fix(json2): preserve coerced values during compaction

* chore: remove unnecessary clone

* chor: reduce memory allocations

* fix: cargo clippy

* chore: update greptime-proto to main branch

* refactor(datatypes): unify string handling with other JSON type hints

* fix: cr

---------

Signed-off-by: fys <fengys1996@gmail.com>

src/datatypes/src/extension/json.rs

@@ -24,9 +24,10 @@ use parquet_variant_compute::VariantType;
 use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, ensure};
 
-use crate::error::InvalidJson2LayoutSnafu;
+use crate::error::{InvalidJson2LayoutSnafu, SerializeSnafu};
 pub use crate::json::JSON2_REMAINDER_FIELD_NAME;
 use crate::json::JsonSettings;
+use crate::schema::Metadata;
 
 /// Aligns JSON2 field types with their built arrays while preserving field and schema metadata.
 pub fn align_schema_with_json_array(schema: SchemaRef, columns: &[ArrayRef]) -> SchemaRef {
@@ -151,7 +152,7 @@ pub struct JsonMetadata {
 }
 
 impl JsonMetadata {
-    /// Creates metadata for the JSON2 layout version 2.
+    /// Creates metadata for the latest JSON2 layout (currently V2).
     pub fn new(json_settings: JsonSettings) -> Self {
         Self {
             json_settings,
@@ -300,6 +301,27 @@ impl ExtensionType for Json2ExtensionType {
     }
 }
 
+/// Returns JSON2 column metadata with updated settings and the latest layout.
+///
+/// Existing SSTs retain their own physical layout metadata.
+pub fn json2_metadata_with_updated_settings(
+    current_metadata: &Metadata,
+    settings: JsonSettings,
+) -> crate::error::Result<Metadata> {
+    let json_metadata = JsonMetadata::new(settings);
+
+    let mut metadata = current_metadata.clone();
+    metadata.insert(
+        EXTENSION_TYPE_NAME_KEY.to_string(),
+        Json2ExtensionType::NAME.to_string(),
+    );
+    metadata.insert(
+        EXTENSION_TYPE_METADATA_KEY.to_string(),
+        serde_json::to_string(&json_metadata).context(SerializeSnafu)?,
+    );
+    Ok(metadata)
+}
+
 /// Checks whether this field is either a legacy JSONB or JSON2 extension type.
 pub fn is_any_json_extension_type<T: AsRef<Field>>(field: T) -> bool {
     let name = field.as_ref().extension_type_name();
@@ -442,6 +464,35 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_json2_metadata_with_updated_settings_upgrades_layout() -> crate::error::Result<()> {
+        for (name, json) in [
+            (JsonExtensionType::NAME, r#"{"json_settings":{}}"#),
+            (Json2ExtensionType::NAME, r#"{"json_settings":{}}"#),
+            (
+                Json2ExtensionType::NAME,
+                r#"{"json_settings":{},"layout_version":2}"#,
+            ),
+        ] {
+            let metadata = HashMap::from([
+                (EXTENSION_TYPE_NAME_KEY.to_string(), name.to_string()),
+                (EXTENSION_TYPE_METADATA_KEY.to_string(), json.to_string()),
+                ("other".to_string(), "kept".to_string()),
+            ]);
+            let settings = JsonSettings::try_new(vec![], Some(10))?;
+            let updated = json2_metadata_with_updated_settings(&metadata, settings.clone())?;
+            assert_eq!(Some("kept"), updated.get("other").map(String::as_str));
+            assert_eq!(
+                Some(Json2ExtensionType::NAME),
+                updated.get(EXTENSION_TYPE_NAME_KEY).map(String::as_str)
+            );
+            let json_metadata: JsonMetadata =
+                serde_json::from_str(updated.get(EXTENSION_TYPE_METADATA_KEY).unwrap()).unwrap();
+            assert_eq!(JsonMetadata::new(settings), json_metadata);
+        }
+        Ok(())
+    }
+
     #[test]
     fn test_parse_json2_physical_layout() -> crate::error::Result<()> {
         let legacy = Field::new("data", DataType::Struct(Fields::empty()), true)

src/datatypes/src/json.rs

@@ -23,6 +23,7 @@ pub mod value;
 
 use std::collections::BTreeMap;
 use std::collections::btree_map::Entry;
+use std::sync::Arc;
 
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value as Json};
@@ -31,6 +32,7 @@ use snafu::ResultExt;
 use crate::data_type::ConcreteDataType;
 use crate::error::{self, InvalidJson2SettingsSnafu, Result, UnsupportedJsonTypeSnafu};
 use crate::json::value::{JsonValue, JsonVariant, encode_serde_json_as_jsonb};
+use crate::prelude::DataType as _;
 use crate::types::json_type::{JsonNativeType, JsonObjectType};
 use crate::value::{ListValue, StructValue, Value};
 
@@ -93,6 +95,15 @@ pub struct JsonTypeHint {
     pub inverted_index: bool,
 }
 
+/// Specifies how encoding handles a value that does not match its JSON2 type hint.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TypeHintMismatchPolicy {
+    /// Returns an error for a type hint mismatch.
+    Reject,
+    /// Coerces to the hint type, encoding unconvertible values as null.
+    CoerceOrNull,
+}
+
 /// Context for JSON encoding/decoding that tracks the current key path.
 #[derive(Clone, Debug)]
 pub struct JsonContext<'a> {
@@ -135,6 +146,30 @@ impl JsonSettings {
         (self.type_hints, self.max_auto_expanded_paths)
     }
 
+    /// Returns whether the settings are equivalent, ignoring type hint order.
+    pub fn equivalent(&self, other: &Self) -> bool {
+        let Self {
+            type_hints,
+            max_auto_expanded_paths,
+        } = self;
+        let Self {
+            type_hints: other_hints,
+            max_auto_expanded_paths: other_max_auto_expanded_paths,
+        } = other;
+
+        if max_auto_expanded_paths != other_max_auto_expanded_paths
+            || type_hints.len() != other_hints.len()
+        {
+            return false;
+        }
+
+        let mut hints = type_hints.iter().collect::<Vec<_>>();
+        let mut other_hints = other_hints.iter().collect::<Vec<_>>();
+        hints.sort_unstable_by(|a, b| a.path.cmp(&b.path));
+        other_hints.sort_unstable_by(|a, b| a.path.cmp(&b.path));
+        hints == other_hints
+    }
+
     /// Decode an encoded StructValue back into a serde_json::Value.
     pub fn decode(&self, value: Value) -> Result<Json> {
         let mut context = JsonContext {
@@ -146,6 +181,15 @@ impl JsonSettings {
 
     /// Encode a serde_json::Value into a Value::Json using current settings.
     pub fn encode(&self, json: Json) -> Result<Value> {
+        self.encode_with_type_hint_mismatch_policy(json, TypeHintMismatchPolicy::Reject)
+    }
+
+    /// Encodes a serde_json::Value using the given type hint mismatch policy.
+    pub fn encode_with_type_hint_mismatch_policy(
+        &self,
+        json: Json,
+        policy: TypeHintMismatchPolicy,
+    ) -> Result<Value> {
         if let Json::Object(object) = &json
             && object.contains_key(JSON2_REMAINDER_FIELD_NAME)
         {
@@ -160,7 +204,7 @@ impl JsonSettings {
             path: Vec::new(),
             settings: self,
         };
-        encode_json_with_context(json, &mut context).map(|v| Value::Json(Box::new(v)))
+        encode_json_with_context(json, &mut context, policy).map(|v| Value::Json(Box::new(v)))
     }
 }
 
@@ -272,36 +316,41 @@ fn with_key_context<T>(
 }
 
 /// Main encoding function with key path tracking
-fn encode_json_with_context(json: Json, context: &mut JsonContext) -> Result<JsonValue> {
+fn encode_json_with_context(
+    json: Json,
+    context: &mut JsonContext,
+    policy: TypeHintMismatchPolicy,
+) -> Result<JsonValue> {
     if context.path.is_empty() && !matches!(json, Json::Object(_)) {
         return UnsupportedJsonTypeSnafu.fail();
     }
 
     match json {
-        Json::Object(json_object) => encode_json_object_with_context(json_object, context),
-        Json::Array(json_array) => encode_json_array_with_context(json_array, context),
-        _ => encode_json_value_with_context(json, context),
+        Json::Object(json_object) => encode_json_object_with_context(json_object, context, policy),
+        Json::Array(json_array) => encode_json_array_with_context(json_array, context, policy),
+        _ => encode_json_value_with_context(json, context, policy),
     }
 }
 
 fn encode_json_object_with_context<'a>(
     json_object: Map<String, Json>,
     context: &mut JsonContext<'a>,
+    policy: TypeHintMismatchPolicy,
 ) -> Result<JsonValue> {
     let mut object = BTreeMap::new();
     for (key, value) in json_object {
         let value = with_key_context(context, &key, |context| {
             if let Some(hint) = context.type_hint() {
-                encode_json_value_with_hint(value, hint, context)
+                encode_json_value_with_hint(value, hint, context, policy)
             } else {
-                encode_json_value_with_context(value, context)
+                encode_json_value_with_context(value, context, policy)
             }
         })?;
 
         object.insert(key, value.into_variant());
     }
 
-    fill_missing_type_hints(&mut object, context)?;
+    fill_missing_type_hints(&mut object, context, policy)?;
 
     Ok(JsonValue::new(JsonVariant::Object(object)))
 }
@@ -309,13 +358,14 @@ fn encode_json_object_with_context<'a>(
 fn fill_missing_type_hints(
     object: &mut BTreeMap<String, JsonVariant>,
     context: &mut JsonContext,
+    policy: TypeHintMismatchPolicy,
 ) -> Result<()> {
     for hint in &context.settings.type_hints {
         if hint.path.len() > context.path.len() && hint.path.starts_with(&context.path) {
             let depth = context.path.len();
             let key = &hint.path[depth];
             with_key_context(context, key, |context| {
-                insert_missing_type_hint(object, context, hint, depth)
+                insert_missing_type_hint(object, context, hint, depth, policy)
             })?;
         }
     }
@@ -327,6 +377,7 @@ fn insert_missing_type_hint(
     field_context: &mut JsonContext,
     hint: &JsonTypeHint,
     depth: usize,
+    policy: TypeHintMismatchPolicy,
 ) -> Result<()> {
     let key = &hint.path[depth];
     let is_leaf = depth + 1 == hint.path.len();
@@ -341,7 +392,11 @@ fn insert_missing_type_hint(
     match object.entry(key.clone()) {
         Entry::Occupied(mut entry) => match entry.get_mut() {
             JsonVariant::Object(child) => {
-                insert_missing_type_hint(child, field_context, hint, depth + 1)
+                insert_missing_type_hint(child, field_context, hint, depth + 1, policy)
+            }
+            _ if policy == TypeHintMismatchPolicy::CoerceOrNull => {
+                entry.insert(JsonValue::null().into_variant());
+                Ok(())
             }
             _ => error::InvalidJsonSnafu {
                 value: format!(
@@ -354,7 +409,7 @@ fn insert_missing_type_hint(
         },
         Entry::Vacant(entry) => {
             let mut child = BTreeMap::new();
-            insert_missing_type_hint(&mut child, field_context, hint, depth + 1)?;
+            insert_missing_type_hint(&mut child, field_context, hint, depth + 1, policy)?;
             entry.insert(JsonVariant::Object(child));
             Ok(())
         }
@@ -365,65 +420,152 @@ fn encode_json_value_with_hint(
     json: Json,
     hint: &JsonTypeHint,
     context: &mut JsonContext,
+    policy: TypeHintMismatchPolicy,
 ) -> Result<JsonValue> {
     if json.is_null() {
         return Ok(JsonValue::null());
     }
 
-    let invalid_type = || {
-        error::InvalidJsonSnafu {
+    let json = match (&hint.data_type, json) {
+        (ConcreteDataType::String(_), Json::String(value)) => {
+            return Ok(value.into());
+        }
+        (
+            ConcreteDataType::Int8(_)
+            | ConcreteDataType::Int16(_)
+            | ConcreteDataType::Int32(_)
+            | ConcreteDataType::Int64(_),
+            Json::Number(value),
+        ) => {
+            if let Some(value) = value.as_i64() {
+                return Ok(value.into());
+            }
+            Json::Number(value)
+        }
+        (
+            ConcreteDataType::UInt8(_)
+            | ConcreteDataType::UInt16(_)
+            | ConcreteDataType::UInt32(_)
+            | ConcreteDataType::UInt64(_),
+            Json::Number(value),
+        ) => {
+            if let Some(value) = value.as_u64() {
+                return Ok(value.into());
+            }
+            Json::Number(value)
+        }
+        (ConcreteDataType::Float32(_) | ConcreteDataType::Float64(_), Json::Number(value)) => {
+            if let Some(value) = value.as_f64() {
+                return Ok(value.into());
+            }
+            Json::Number(value)
+        }
+        (ConcreteDataType::Boolean(_), Json::Bool(value)) => {
+            return Ok(value.into());
+        }
+        (_, json) => json,
+    };
+
+    match policy {
+        TypeHintMismatchPolicy::Reject => error::InvalidJsonSnafu {
             value: format!(
                 "JSON value at {} does not match JSON2 type hint {}",
                 context.path.join("."),
                 hint.data_type
             ),
         }
-        .fail()
-    };
+        .fail(),
+        TypeHintMismatchPolicy::CoerceOrNull => {
+            let value = coerce_json_value_to_type(json, &hint.data_type);
+            let value = Json::try_from(value).map_err(|error| {
+                error::InvalidJsonSnafu {
+                    value: error.to_string(),
+                }
+                .build()
+            })?;
+            Ok(JsonValue::new(value.into()))
+        }
+    }
+}
 
-    match (&hint.data_type, json) {
-        (ConcreteDataType::String(_), Json::String(v)) => Ok(v.into()),
-        (
-            ConcreteDataType::Int8(_)
-            | ConcreteDataType::Int16(_)
-            | ConcreteDataType::Int32(_)
-            | ConcreteDataType::Int64(_),
-            Json::Number(v),
-        ) => match v.as_i64() {
-            Some(v) => Ok(v.into()),
-            None => invalid_type(),
-        },
-        (
-            ConcreteDataType::UInt8(_)
-            | ConcreteDataType::UInt16(_)
-            | ConcreteDataType::UInt32(_)
-            | ConcreteDataType::UInt64(_),
-            Json::Number(v),
-        ) => match v.as_u64() {
-            Some(v) => Ok(v.into()),
-            None => invalid_type(),
-        },
-        (ConcreteDataType::Float32(_) | ConcreteDataType::Float64(_), Json::Number(v)) => {
-            match v.as_f64() {
-                Some(v) => Ok(v.into()),
-                None => invalid_type(),
+/// Coerces a JSON value using the same semantics as JSON2 query projection.
+/// Values that cannot be converted become null.
+pub(crate) fn coerce_json_value_to_type(value: Json, to_type: &ConcreteDataType) -> Value {
+    if value.is_null() {
+        return Value::Null;
+    }
+
+    if to_type.is_string() {
+        let value = match value {
+            Json::String(value) => value,
+            value => value.to_string(),
+        };
+        return Value::String(value.into());
+    }
+
+    if matches!(to_type, ConcreteDataType::Binary(_)) {
+        return Value::Binary(encode_serde_json_as_jsonb(value).into());
+    }
+
+    if let Some(struct_type) = to_type.as_struct() {
+        let Json::Object(mut object) = value else {
+            return Value::Null;
+        };
+        let values = struct_type
+            .fields()
+            .iter()
+            .map(|field| {
+                object
+                    .remove(field.name())
+                    .map(|value| coerce_json_value_to_type(value, field.data_type()))
+                    .unwrap_or(Value::Null)
+            })
+            .collect::<Vec<_>>();
+        return Value::Struct(StructValue::new(values, struct_type.clone()));
+    }
+
+    if let Some(list_type) = to_type.as_list() {
+        let Json::Array(values) = value else {
+            return Value::Null;
+        };
+        let item_type = list_type.item_type().clone();
+        let values = values
+            .into_iter()
+            .map(|value| coerce_json_value_to_type(value, &item_type))
+            .collect::<Vec<_>>();
+        return Value::List(ListValue::new(values, Arc::new(item_type)));
+    }
+
+    let value = match value {
+        Json::Bool(value) => Value::Boolean(value),
+        Json::Number(value) => {
+            if let Some(value) = value.as_i64() {
+                Value::Int64(value)
+            } else if let Some(value) = value.as_u64() {
+                Value::UInt64(value)
+            } else if let Some(value) = value.as_f64() {
+                Value::Float64(value.into())
+            } else {
+                Value::Null
             }
         }
-        (ConcreteDataType::Boolean(_), Json::Bool(v)) => Ok(v.into()),
-        _ => invalid_type(),
-    }
+        Json::String(value) => Value::String(value.into()),
+        Json::Array(_) | Json::Object(_) | Json::Null => Value::Null,
+    };
+    to_type.try_cast(value).unwrap_or(Value::Null)
 }
 
 fn encode_json_array_with_context<'a>(
     json_array: Vec<Json>,
     context: &mut JsonContext<'a>,
+    policy: TypeHintMismatchPolicy,
 ) -> Result<JsonValue> {
     let json_array_len = json_array.len();
     let mut items = Vec::with_capacity(json_array_len);
 
     for (index, value) in json_array.into_iter().enumerate() {
         let item_value = with_key_context(context, &index.to_string(), |context| {
-            encode_json_value_with_context(value, context)
+            encode_json_value_with_context(value, context, policy)
         })?;
         items.push(item_value);
     }
@@ -458,7 +600,11 @@ fn encode_json_array_with_context<'a>(
 }
 
 /// Helper function to encode a JSON value to a Value and determine its ConcreteDataType with context
-fn encode_json_value_with_context(json: Json, context: &mut JsonContext) -> Result<JsonValue> {
+fn encode_json_value_with_context(
+    json: Json,
+    context: &mut JsonContext,
+    policy: TypeHintMismatchPolicy,
+) -> Result<JsonValue> {
     if context.path.len() >= JSON2_MAX_STRUCTURED_DEPTH
         && matches!(&json, Json::Object(_) | Json::Array(_))
     {
@@ -487,8 +633,8 @@ fn encode_json_value_with_context(json: Json, context: &mut JsonContext) -> Resu
             }
         }
         Json::String(s) => Ok(s.into()),
-        Json::Array(arr) => encode_json_array_with_context(arr, context),
-        Json::Object(obj) => encode_json_object_with_context(obj, context),
+        Json::Array(arr) => encode_json_array_with_context(arr, context, policy),
+        Json::Object(obj) => encode_json_object_with_context(obj, context, policy),
     }
 }
 
@@ -597,6 +743,50 @@ mod tests {
         &struct_value.items()[index]
     }
 
+    #[test]
+    fn test_json_settings_equivalent() -> Result<()> {
+        let hints = vec![
+            JsonTypeHint {
+                path: vec!["user".to_string(), "name".to_string()],
+                data_type: ConcreteDataType::string_datatype(),
+                inverted_index: false,
+            },
+            JsonTypeHint {
+                path: vec!["count".to_string()],
+                data_type: ConcreteDataType::int64_datatype(),
+                inverted_index: false,
+            },
+        ];
+        let settings = JsonSettings::try_new(hints.clone(), Some(10))?;
+        let mut reversed = hints.clone();
+        reversed.reverse();
+        let reordered = JsonSettings::try_new(reversed, Some(10))?;
+        assert_ne!(settings, reordered);
+        assert!(settings.equivalent(&reordered));
+        assert!(reordered.equivalent(&settings));
+        assert!(settings.equivalent(&settings));
+        assert_eq!(hints, settings.type_hints());
+        assert!(JsonSettings::default().equivalent(&JsonSettings::default()));
+
+        for limit in [None, Some(0), Some(11)] {
+            assert!(!settings.equivalent(&JsonSettings::try_new(hints.clone(), limit)?));
+        }
+        assert!(!settings.equivalent(&JsonSettings::try_new(vec![hints[0].clone()], Some(10))?));
+
+        let mut changed = hints.clone();
+        changed[0].path = vec!["user".to_string(), "id".to_string()];
+        assert!(!settings.equivalent(&JsonSettings::try_new(changed, Some(10))?));
+
+        let mut changed = hints.clone();
+        changed[0].data_type = ConcreteDataType::int64_datatype();
+        assert!(!settings.equivalent(&JsonSettings::try_new(changed, Some(10))?));
+
+        let mut changed = hints;
+        changed[0].inverted_index = true;
+        assert!(!settings.equivalent(&JsonSettings::try_new(changed, Some(10))?));
+        Ok(())
+    }
+
     #[test]
     fn test_json_settings_deserializes_legacy_type_hint_constraints()
     -> std::result::Result<(), Box<dyn std::error::Error>> {
@@ -755,6 +945,104 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_coerce_or_null_on_type_hint_mismatch() -> Result<()> {
+        let settings = JsonSettings::try_new(
+            vec![JsonTypeHint {
+                path: vec!["kind".to_string()],
+                data_type: ConcreteDataType::int64_datatype(),
+                inverted_index: false,
+            }],
+            None,
+        )?;
+
+        let result = settings
+            .encode_with_type_hint_mismatch_policy(
+                json!({"kind": "invalid", "message": "kept"}),
+                TypeHintMismatchPolicy::CoerceOrNull,
+            )?
+            .into_json_inner()
+            .unwrap();
+        let Value::Struct(root) = result else {
+            panic!("Expected Struct value");
+        };
+        assert_eq!(struct_field_value(&root, "kind"), &Value::Null);
+        assert_eq!(
+            struct_field_value(&root, "message"),
+            &Value::String("kept".into())
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_coerce_or_null_converts_type_hint_values() -> Result<()> {
+        let settings = JsonSettings::try_new(
+            vec![
+                JsonTypeHint {
+                    path: vec!["to_string".to_string()],
+                    data_type: ConcreteDataType::string_datatype(),
+                    inverted_index: false,
+                },
+                JsonTypeHint {
+                    path: vec!["to_int".to_string()],
+                    data_type: ConcreteDataType::int64_datatype(),
+                    inverted_index: false,
+                },
+            ],
+            None,
+        )?;
+
+        let result = settings
+            .encode_with_type_hint_mismatch_policy(
+                json!({"to_string": 123, "to_int": "456", "invalid": "value"}),
+                TypeHintMismatchPolicy::CoerceOrNull,
+            )?
+            .into_json_inner()
+            .unwrap();
+        let Value::Struct(root) = result else {
+            panic!("Expected Struct value");
+        };
+        assert_eq!(
+            struct_field_value(&root, "to_string"),
+            &Value::String("123".into())
+        );
+        assert_eq!(struct_field_value(&root, "to_int"), &Value::Int64(456));
+        assert_eq!(
+            struct_field_value(&root, "invalid"),
+            &Value::String("value".into())
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_coerce_or_null_on_nested_type_hint_structure_mismatch() -> Result<()> {
+        let settings = JsonSettings::try_new(
+            vec![JsonTypeHint {
+                path: vec!["user".to_string(), "age".to_string()],
+                data_type: ConcreteDataType::int64_datatype(),
+                inverted_index: false,
+            }],
+            None,
+        )?;
+
+        let result = settings
+            .encode_with_type_hint_mismatch_policy(
+                json!({"user": "invalid", "message": "kept"}),
+                TypeHintMismatchPolicy::CoerceOrNull,
+            )?
+            .into_json_inner()
+            .unwrap();
+        let Value::Struct(root) = result else {
+            panic!("Expected Struct value");
+        };
+        assert_eq!(struct_field_value(&root, "user"), &Value::Null);
+        assert_eq!(
+            struct_field_value(&root, "message"),
+            &Value::String("kept".into())
+        );
+        Ok(())
+    }
+
     #[test]
     fn test_encode_rejects_reserved_remainder_field() -> Result<()> {
         let settings = JsonSettings::default();

src/datatypes/src/vectors/json/array.rs

@@ -26,15 +26,13 @@ use serde_json::Value;
 use snafu::{OptionExt, ResultExt};
 
 use crate::arrow_array::{binary_array_value, string_array_value};
-use crate::data_type::ConcreteDataType;
+use crate::data_type::{ConcreteDataType, DataType as _};
 use crate::error::{
     AlignJsonArraySnafu, ArrowComputeSnafu, InvalidJsonSnafu, InvalidJsonbSnafu, Result,
 };
 use crate::extension::json::{JSON2_REMAINDER_FIELD_NAME, json2_remainder_field};
-use crate::json::JsonSettings;
-use crate::json::value::{decode_json_variant, encode_serde_json_as_jsonb};
-use crate::prelude::{DataType as _, Value as GreptimeValue};
-use crate::value::{ListValue, StructValue};
+use crate::json::value::decode_json_variant;
+use crate::json::{JsonSettings, TypeHintMismatchPolicy, coerce_json_value_to_type};
 use crate::vectors::MutableVector;
 use crate::vectors::json::builder::{JsonVectorBuilder, json2_physical_data_type};
 use crate::vectors::json::variant::variant_to_json_values;
@@ -123,6 +121,23 @@ impl JsonArray<'_> {
         field: &Field,
         logical_settings: &JsonSettings,
         target_layout: &JsonSettings,
+    ) -> Result<ArrayRef> {
+        self.rewrite_to_v2_with_type_hint_mismatch_policy(
+            field,
+            logical_settings,
+            target_layout,
+            TypeHintMismatchPolicy::Reject,
+        )
+    }
+
+    /// Rewrites a JSON2 array to the specified v2 physical layout using the
+    /// given type hint mismatch policy.
+    pub fn rewrite_to_v2_with_type_hint_mismatch_policy(
+        &self,
+        field: &Field,
+        logical_settings: &JsonSettings,
+        target_layout: &JsonSettings,
+        policy: TypeHintMismatchPolicy,
     ) -> Result<ArrayRef> {
         let is_v2 = json2_remainder_field(field)?.is_some();
         if is_v2 && self.inner.data_type() == &json2_physical_data_type(target_layout) {
@@ -141,7 +156,8 @@ impl JsonArray<'_> {
             if value.is_null() {
                 builder.push_null();
             } else {
-                let value = logical_settings.encode(value)?;
+                let value =
+                    logical_settings.encode_with_type_hint_mismatch_policy(value, policy)?;
                 builder.try_push_value_ref(&value.as_value_ref())?;
             }
         }
@@ -319,87 +335,12 @@ fn project_json_values(values: Vec<Value>, to_type: &DataType) -> Result<ArrayRe
     let concrete_type = ConcreteDataType::from_arrow_type(to_type);
     let mut builder = concrete_type.create_mutable_vector(values.len());
     for value in values {
-        let value = project_json_value_to_type(value, &concrete_type)?;
+        let value = coerce_json_value_to_type(value, &concrete_type);
         builder.try_push_value_ref(&value.as_value_ref())?;
     }
     Ok(builder.to_vector().to_arrow_array())
 }
 
-fn project_json_value_to_type(value: Value, to_type: &ConcreteDataType) -> Result<GreptimeValue> {
-    if value.is_null() {
-        return Ok(GreptimeValue::Null);
-    }
-
-    if to_type.is_string() {
-        let value = match value {
-            Value::String(value) => value,
-            value => value.to_string(),
-        };
-        return Ok(GreptimeValue::String(value.into()));
-    }
-
-    if matches!(to_type, ConcreteDataType::Binary(_)) {
-        return Ok(GreptimeValue::Binary(
-            encode_serde_json_as_jsonb(value).into(),
-        ));
-    }
-
-    if let Some(struct_type) = to_type.as_struct() {
-        let Value::Object(mut object) = value else {
-            return Ok(GreptimeValue::Null);
-        };
-        let values = struct_type
-            .fields()
-            .iter()
-            .map(|field| {
-                object
-                    .remove(field.name())
-                    .map(|value| project_json_value_to_type(value, field.data_type()))
-                    .transpose()
-                    .map(|value| value.unwrap_or(GreptimeValue::Null))
-            })
-            .collect::<Result<Vec<_>>>()?;
-        return Ok(GreptimeValue::Struct(StructValue::new(
-            values,
-            struct_type.clone(),
-        )));
-    }
-
-    if let Some(list_type) = to_type.as_list() {
-        let Value::Array(values) = value else {
-            return Ok(GreptimeValue::Null);
-        };
-        let item_type = list_type.item_type().clone();
-        let values = values
-            .into_iter()
-            .map(|value| project_json_value_to_type(value, &item_type))
-            .collect::<Result<Vec<_>>>()?;
-        return Ok(GreptimeValue::List(ListValue::new(
-            values,
-            Arc::new(item_type),
-        )));
-    }
-
-    let value = match value {
-        Value::Bool(value) => GreptimeValue::Boolean(value),
-        Value::Number(value) => {
-            if let Some(value) = value.as_i64() {
-                GreptimeValue::Int64(value)
-            } else if let Some(value) = value.as_u64() {
-                GreptimeValue::UInt64(value)
-            } else if let Some(value) = value.as_f64() {
-                GreptimeValue::Float64(value.into())
-            } else {
-                GreptimeValue::Null
-            }
-        }
-        Value::String(value) => GreptimeValue::String(value.into()),
-        Value::Array(_) | Value::Object(_) => GreptimeValue::Null,
-        Value::Null => GreptimeValue::Null,
-    };
-    Ok(to_type.try_cast(value).unwrap_or(GreptimeValue::Null))
-}
-
 impl<'a> From<&'a ArrayRef> for JsonArray<'a> {
     fn from(inner: &'a ArrayRef) -> Self {
         Self { inner }