refactor(json2): support JSON2 storage layout settings in DDL (#8895)

* refactor(json2): add JSON2 storage layout settings

Signed-off-by: luofucong <luofc@foxmail.com>

* resolve PR comments

Signed-off-by: luofucong <luofc@foxmail.com>

---------

Signed-off-by: luofucong <luofc@foxmail.com>

src/datatypes/src/error.rs

@@ -196,6 +196,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Invalid JSON2 layout: {reason}"))]
+    InvalidJson2Layout {
+        reason: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Invalid Vector: {}", msg))]
     InvalidVector {
         msg: String,
@@ -318,6 +325,7 @@ impl ErrorExt for Error {
             | InvalidTimestampPrecision { .. }
             | InvalidPrecisionOrScale { .. }
             | InvalidJson { .. }
+            | InvalidJson2Layout { .. }
             | InvalidJsonb { .. }
             | InvalidVector { .. }
             | InvalidFulltextOption { .. }

src/datatypes/src/extension/json.rs

@@ -22,6 +22,7 @@ use arrow_schema::{ArrowError, DataType, Field};
 use serde::{Deserialize, Serialize};
 use snafu::ResultExt;
 
+pub use crate::json::JSON2_REMAINDER_FIELD_NAME;
 use crate::json::JsonSettings;
 
 const LEGACY_JSON_STRUCTURE_SETTINGS_KEY: &str = "json_structure_settings";

src/datatypes/src/json.rs

@@ -29,21 +29,47 @@ use serde_json::{Map, Value as Json};
 use snafu::ResultExt;
 
 use crate::data_type::ConcreteDataType;
-use crate::error::{self, Result, UnsupportedJsonTypeSnafu};
+use crate::error::{self, InvalidJson2LayoutSnafu, Result, UnsupportedJsonTypeSnafu};
 use crate::json::value::{JsonValue, JsonVariant, encode_serde_json_as_jsonb};
 use crate::schema::ColumnDefaultConstraint;
-use crate::types::json_type::JsonNativeType;
+use crate::types::json_type::{JsonNativeType, JsonObjectType};
 use crate::value::{ListValue, StructValue, Value};
 
 /// Maximum number of JSON container levels represented as nested Arrow types.
 pub const JSON2_MAX_STRUCTURED_DEPTH: usize = 50;
+/// Reserved physical field containing unexpanded JSON2 paths.
+pub const JSON2_REMAINDER_FIELD_NAME: &str = "!__remainder__!";
 
 /// JSON2 settings stored in column schema metadata and represented through
 /// Arrow extension metadata.
-#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
 pub struct JsonSettings {
     #[serde(default)]
-    pub type_hints: Vec<JsonTypeHint>,
+    type_hints: Vec<JsonTypeHint>,
+    /// Maximum number of unhinted JSON paths expanded into Arrow fields.
+    ///
+    /// `None` preserves the legacy unlimited layout.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    max_auto_expanded_paths: Option<u32>,
+}
+
+#[derive(Deserialize)]
+struct JsonSettingsSerde {
+    #[serde(default)]
+    type_hints: Vec<JsonTypeHint>,
+    #[serde(default)]
+    max_auto_expanded_paths: Option<u32>,
+}
+
+impl<'de> Deserialize<'de> for JsonSettings {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let settings = JsonSettingsSerde::deserialize(deserializer)?;
+        Self::try_new(settings.type_hints, settings.max_auto_expanded_paths)
+            .map_err(serde::de::Error::custom)
+    }
 }
 
 /// Declares selected JSON2 subpaths as typed fields.
@@ -79,8 +105,28 @@ pub struct JsonContext<'a> {
 }
 
 impl JsonSettings {
-    pub fn new(type_hints: Vec<JsonTypeHint>) -> Self {
-        Self { type_hints }
+    /// Creates and validates JSON2 settings.
+    pub fn try_new(
+        type_hints: Vec<JsonTypeHint>,
+        max_auto_expanded_paths: Option<u32>,
+    ) -> Result<Self> {
+        validate_type_hints(&type_hints)?;
+        Ok(Self {
+            type_hints,
+            max_auto_expanded_paths,
+        })
+    }
+
+    pub fn type_hints(&self) -> &[JsonTypeHint] {
+        &self.type_hints
+    }
+
+    pub fn max_auto_expanded_paths(&self) -> Option<u32> {
+        self.max_auto_expanded_paths
+    }
+
+    pub fn into_parts(self) -> (Vec<JsonTypeHint>, Option<u32>) {
+        (self.type_hints, self.max_auto_expanded_paths)
     }
 
     /// Decode an encoded StructValue back into a serde_json::Value.
@@ -102,6 +148,87 @@ impl JsonSettings {
     }
 }
 
+fn validate_type_hints(type_hints: &[JsonTypeHint]) -> Result<()> {
+    let mut object = JsonObjectType::new();
+    for hint in type_hints {
+        if hint.path.len() > JSON2_MAX_STRUCTURED_DEPTH {
+            return InvalidJson2LayoutSnafu {
+                reason: format!(
+                    "JSON2 type hint path cannot exceed {JSON2_MAX_STRUCTURED_DEPTH} segments"
+                ),
+            }
+            .fail();
+        }
+        if hint
+            .path
+            .first()
+            .is_some_and(|x| x == JSON2_REMAINDER_FIELD_NAME)
+        {
+            return InvalidJson2LayoutSnafu {
+                reason: format!(
+                    "JSON2 type hint path cannot be rooted at reserved field '{JSON2_REMAINDER_FIELD_NAME}'"
+                ),
+            }
+            .fail();
+        }
+        let data_type = match &hint.data_type {
+            ConcreteDataType::Boolean(_)
+            | ConcreteDataType::UInt8(_)
+            | ConcreteDataType::UInt16(_)
+            | ConcreteDataType::UInt32(_)
+            | ConcreteDataType::UInt64(_)
+            | ConcreteDataType::Int8(_)
+            | ConcreteDataType::Int16(_)
+            | ConcreteDataType::Int32(_)
+            | ConcreteDataType::Int64(_)
+            | ConcreteDataType::Float32(_)
+            | ConcreteDataType::Float64(_)
+            | ConcreteDataType::String(_) => (&hint.data_type).into(),
+            data_type => {
+                return InvalidJson2LayoutSnafu {
+                    reason: format!("unsupported JSON2 type hint data type: {data_type}"),
+                }
+                .fail();
+            }
+        };
+        validate_type_hint(&mut object, &hint.path, data_type)?;
+    }
+    Ok(())
+}
+
+fn validate_type_hint(
+    object: &mut JsonObjectType,
+    path: &[String],
+    data_type: JsonNativeType,
+) -> Result<()> {
+    let Some((name, path)) = path.split_first() else {
+        return InvalidJson2LayoutSnafu {
+            reason: "JSON2 type hint path must not be empty".to_string(),
+        }
+        .fail();
+    };
+    if path.is_empty() {
+        if object.insert(name.clone(), data_type).is_some() {
+            return InvalidJson2LayoutSnafu {
+                reason: format!("duplicate JSON2 type hint path '{name}'"),
+            }
+            .fail();
+        }
+        return Ok(());
+    }
+
+    let child = object
+        .entry(name.clone())
+        .or_insert_with(|| JsonNativeType::Object(JsonObjectType::new()));
+    let JsonNativeType::Object(child) = child else {
+        return InvalidJson2LayoutSnafu {
+            reason: format!("conflicting JSON2 type hint path at '{name}'"),
+        }
+        .fail();
+    };
+    validate_type_hint(child, path, data_type)
+}
+
 impl<'a> JsonContext<'a> {
     fn type_hint(&self) -> Option<&'a JsonTypeHint> {
         self.settings
@@ -483,7 +610,8 @@ mod tests {
     }
 
     #[test]
-    fn test_json_settings_forward_compatibility() {
+    fn test_json_settings_forward_compatibility()
+    -> std::result::Result<(), Box<dyn std::error::Error>> {
         let json_str = r#"{
             "type_hints": [
                 {
@@ -512,11 +640,10 @@ mod tests {
             ]
         }"#;
 
-        let deserialized = serde_json::from_str::<JsonSettings>(json_str).unwrap();
+        let deserialized = serde_json::from_str::<JsonSettings>(json_str)?;
 
-        assert_eq!(
-            deserialized,
-            JsonSettings::new(vec![
+        let expected = JsonSettings::try_new(
+            vec![
                 JsonTypeHint {
                     path: vec!["user".to_string(), "age".to_string()],
                     data_type: ConcreteDataType::int64_datatype(),
@@ -531,13 +658,16 @@ mod tests {
                     default_constraint: None,
                     inverted_index: false,
                 },
-            ])
-        );
+            ],
+            None,
+        )?;
+        assert_eq!(deserialized, expected);
+        Ok(())
     }
 
     #[test]
-    fn test_json_settings_ser_de() {
-        let settings = JsonSettings::new(vec![
+    fn test_json_settings_ser_de() -> std::result::Result<(), Box<dyn std::error::Error>> {
+        let type_hints = vec![
             JsonTypeHint {
                 path: vec!["user".to_string(), "age".to_string()],
                 data_type: ConcreteDataType::int64_datatype(),
@@ -552,12 +682,76 @@ mod tests {
                 default_constraint: None,
                 inverted_index: false,
             },
-        ]);
+        ];
 
-        let serialized = serde_json::to_string(&settings).unwrap();
-        let deserialized = serde_json::from_str::<JsonSettings>(&serialized).unwrap();
+        for settings in [
+            JsonSettings {
+                type_hints: type_hints.clone(),
+                max_auto_expanded_paths: None,
+            },
+            JsonSettings {
+                type_hints,
+                max_auto_expanded_paths: Some(1),
+            },
+        ] {
+            let serialized = serde_json::to_string(&settings)?;
+            let deserialized = serde_json::from_str::<JsonSettings>(&serialized)?;
+            assert_eq!(settings, deserialized);
+        }
+        Ok(())
+    }
 
-        assert_eq!(settings, deserialized);
+    #[test]
+    fn test_json_settings_reject_invalid_type_hint_layout() {
+        for type_hints in [
+            json!([{"path": [], "type": {"Int64": {}}, "nullable": true, "inverted_index": false}]),
+            json!([
+                {"path": ["a"], "type": {"Int64": {}}, "nullable": true, "inverted_index": false},
+                {"path": ["a", "b"], "type": {"Int64": {}}, "nullable": true, "inverted_index": false}
+            ]),
+            json!([{"path": [JSON2_REMAINDER_FIELD_NAME], "type": {"Int64": {}}, "nullable": true, "inverted_index": false}]),
+        ] {
+            let settings = json!({"type_hints": type_hints});
+            assert!(serde_json::from_value::<JsonSettings>(settings).is_err());
+        }
+
+        let hint = |path, data_type| JsonTypeHint {
+            path,
+            data_type,
+            nullable: true,
+            default_constraint: None,
+            inverted_index: false,
+        };
+        assert!(
+            JsonSettings::try_new(
+                vec![hint(
+                    vec!["nested".to_string(), JSON2_REMAINDER_FIELD_NAME.to_string()],
+                    ConcreteDataType::string_datatype(),
+                )],
+                None,
+            )
+            .is_ok()
+        );
+        assert!(
+            JsonSettings::try_new(
+                vec![hint(
+                    vec!["nested".to_string(); JSON2_MAX_STRUCTURED_DEPTH + 1],
+                    ConcreteDataType::string_datatype(),
+                )],
+                None,
+            )
+            .is_err()
+        );
+        assert!(
+            JsonSettings::try_new(
+                vec![hint(
+                    vec!["date".to_string()],
+                    ConcreteDataType::date_datatype(),
+                )],
+                None,
+            )
+            .is_err()
+        );
     }
 
     #[test]
@@ -702,14 +896,15 @@ mod tests {
     }
 
     #[test]
-    fn test_encode_json_respects_type_hint() {
-        let settings = JsonSettings::new(vec![JsonTypeHint {
+    fn test_encode_json_respects_type_hint() -> Result<()> {
+        let type_hints = vec![JsonTypeHint {
             path: vec!["age".to_string()],
             data_type: ConcreteDataType::int64_datatype(),
             nullable: false,
             default_constraint: None,
             inverted_index: false,
-        }]);
+        }];
+        let settings = JsonSettings::try_new(type_hints, None)?;
 
         let result = settings
             .encode(json!({
@@ -731,17 +926,19 @@ mod tests {
             }))
             .unwrap_err();
         assert!(err.to_string().contains("does not match JSON2 type hint"));
+        Ok(())
     }
 
     #[test]
-    fn test_encode_json_respects_unsigned_type_hint() {
-        let settings = JsonSettings::new(vec![JsonTypeHint {
+    fn test_encode_json_respects_unsigned_type_hint() -> Result<()> {
+        let type_hints = vec![JsonTypeHint {
             path: vec!["count".to_string()],
             data_type: ConcreteDataType::uint64_datatype(),
             nullable: false,
             default_constraint: None,
             inverted_index: false,
-        }]);
+        }];
+        let settings = JsonSettings::try_new(type_hints, None)?;
 
         let result = settings
             .encode(json!({
@@ -765,17 +962,19 @@ mod tests {
             }))
             .unwrap_err();
         assert!(err.to_string().contains("does not match JSON2 type hint"));
+        Ok(())
     }
 
     #[test]
-    fn test_encode_json_fills_missing_type_hint_with_default() {
-        let settings = JsonSettings::new(vec![JsonTypeHint {
+    fn test_encode_json_fills_missing_type_hint_with_default() -> Result<()> {
+        let type_hints = vec![JsonTypeHint {
             path: vec!["user".to_string(), "age".to_string()],
             data_type: ConcreteDataType::int64_datatype(),
             nullable: false,
             default_constraint: Some(ColumnDefaultConstraint::Value(Value::Int64(7))),
             inverted_index: false,
-        }]);
+        }];
+        let settings = JsonSettings::try_new(type_hints, None)?;
 
         let result = settings
             .encode(json!({}))
@@ -790,17 +989,19 @@ mod tests {
             panic!("Expected user Struct value");
         };
         assert_eq!(struct_field_value(user, "age"), &Value::Int64(7));
+        Ok(())
     }
 
     #[test]
-    fn test_encode_json_fills_missing_nullable_type_hint_with_null() {
-        let settings = JsonSettings::new(vec![JsonTypeHint {
+    fn test_encode_json_fills_missing_nullable_type_hint_with_null() -> Result<()> {
+        let type_hints = vec![JsonTypeHint {
             path: vec!["user".to_string(), "name".to_string()],
             data_type: ConcreteDataType::string_datatype(),
             nullable: true,
             default_constraint: None,
             inverted_index: false,
-        }]);
+        }];
+        let settings = JsonSettings::try_new(type_hints, None)?;
 
         let result = settings
             .encode(json!({ "user": {} }))
@@ -815,28 +1016,31 @@ mod tests {
             panic!("Expected user Struct value");
         };
         assert_eq!(struct_field_value(user, "name"), &Value::Null);
+        Ok(())
     }
 
     #[test]
-    fn test_encode_json_rejects_missing_non_null_type_hint() {
-        let settings = JsonSettings::new(vec![JsonTypeHint {
+    fn test_encode_json_rejects_missing_non_null_type_hint() -> Result<()> {
+        let type_hints = vec![JsonTypeHint {
             path: vec!["user".to_string(), "age".to_string()],
             data_type: ConcreteDataType::int64_datatype(),
             nullable: false,
             default_constraint: None,
             inverted_index: false,
-        }]);
+        }];
+        let settings = JsonSettings::try_new(type_hints, None)?;
 
         let err = settings.encode(json!({})).unwrap_err();
         assert!(
             err.to_string()
                 .contains("missing non-null JSON2 type hint path user.age")
         );
+        Ok(())
     }
 
     #[test]
-    fn test_encode_json_merges_missing_type_hint_prefix() {
-        let settings = JsonSettings::new(vec![
+    fn test_encode_json_merges_missing_type_hint_prefix() -> Result<()> {
+        let type_hints = vec![
             JsonTypeHint {
                 path: vec!["user".to_string(), "age".to_string()],
                 data_type: ConcreteDataType::int64_datatype(),
@@ -853,13 +1057,10 @@ mod tests {
                 ))),
                 inverted_index: false,
             },
-        ]);
+        ];
+        let settings = JsonSettings::try_new(type_hints, None)?;
 
-        let result = settings
-            .encode(json!({}))
-            .unwrap()
-            .into_json_inner()
-            .unwrap();
+        let result = settings.encode(json!({}))?.into_json_inner().unwrap();
 
         let Value::Struct(root) = result else {
             panic!("Expected Struct value");
@@ -872,6 +1073,7 @@ mod tests {
             struct_field_value(user, "name"),
             &Value::String("unknown".into())
         );
+        Ok(())
     }
 
     #[test]

src/sql/src/error.rs

@@ -339,6 +339,13 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(transparent)]
+    Datatypes {
+        #[snafu(implicit)]
+        location: Location,
+        source: datatypes::error::Error,
+    },
 }
 
 impl ErrorExt for Error {
@@ -384,7 +391,9 @@ impl ErrorExt for Error {
             #[cfg(feature = "enterprise")]
             InvalidTriggerWebhookOption { .. } => StatusCode::InvalidArguments,
 
-            SerializeColumnDefaultConstraint { source, .. } => source.status_code(),
+            SerializeColumnDefaultConstraint { source, .. } | Datatypes { source, .. } => {
+                source.status_code()
+            }
 
             ConvertToGrpcDataType { source, .. } => source.status_code(),
             SqlCommon { source, .. } => source.status_code(),

src/sql/src/parsers/create_parser.rs

@@ -711,8 +711,8 @@ impl<'a> ParserContext<'a> {
         let mut extensions = ColumnExtensions::default();
 
         let data_type =
-            if let Some((data_type, type_hints)) = json::parse_json2_type_and_hints(parser)? {
-                extensions.json_type_hints = type_hints;
+            if let Some((data_type, options)) = json::parse_json2_type_and_options(parser)? {
+                extensions.json2_options = options;
                 data_type
             } else {
                 parser.parse_data_type().context(SyntaxSnafu)?

src/sql/src/parsers/create_parser/json.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use datatypes::extension::json::JSON2_REMAINDER_FIELD_NAME;
 use datatypes::json::JSON2_MAX_STRUCTURED_DEPTH;
 use snafu::{ResultExt, ensure};
 use sqlparser::ast::{DataType, ExactNumberInfo, Expr, ObjectName, UnaryOperator};
@@ -22,14 +23,15 @@ use sqlparser::tokenizer::Token;
 use crate::ast::Ident;
 use crate::error::{InvalidSqlSnafu, Result, SyntaxSnafu};
 use crate::parsers::create_parser::{INVERTED, SKIPPING};
-use crate::statements::create::JsonTypeHint;
+use crate::statements::create::{Json2Options, JsonTypeHint};
 use crate::statements::transform::type_alias::get_type_by_alias;
 
 const JSON2_TYPE_NAME: &str = "JSON2";
+const MAX_AUTO_EXPANDED_PATHS: &str = "max_auto_expanded_paths";
 
-pub(super) fn parse_json2_type_and_hints(
+pub(super) fn parse_json2_type_and_options(
     parser: &mut Parser<'_>,
-) -> Result<Option<(DataType, Vec<JsonTypeHint>)>> {
+) -> Result<Option<(DataType, Option<Json2Options>)>> {
     let token = parser.peek_token();
     let Token::Word(word) = &token.token else {
         return Ok(None);
@@ -41,26 +43,62 @@ pub(super) fn parse_json2_type_and_hints(
 
     parser.next_token();
     let data_type = DataType::Custom(ObjectName::from(vec![Ident::new(JSON2_TYPE_NAME)]), vec![]);
-    let type_hints = if parser.consume_token(&Token::LParen) {
-        parse_json2_type_hints(parser)?
+    let options = if parser.consume_token(&Token::LParen) {
+        parse_json2_options(parser)?
     } else {
-        vec![]
+        None
     };
 
-    Ok(Some((data_type, type_hints)))
+    Ok(Some((data_type, options)))
 }
 
-fn parse_json2_type_hints(parser: &mut Parser<'_>) -> Result<Vec<JsonTypeHint>> {
-    let mut hints = Vec::new();
-
+fn parse_json2_options(parser: &mut Parser<'_>) -> Result<Option<Json2Options>> {
     if parser.consume_token(&Token::RParen) {
-        return Ok(hints);
+        return Ok(None);
     }
 
+    let mut max_auto_expanded_paths = None;
+    let mut type_hints = Vec::new();
     loop {
-        let hint = parse_json2_type_hint(parser)?;
-        ensure_no_path_conflict(&hints, &hint.path)?;
-        hints.push(hint);
+        let token = parser.peek_token();
+        let is_max_auto_expanded_paths = matches!(
+            &token.token,
+            Token::Word(word)
+                if word.quote_style.is_none()
+                    && word.value.eq_ignore_ascii_case(MAX_AUTO_EXPANDED_PATHS)
+        );
+        if is_max_auto_expanded_paths {
+            parser.next_token();
+            ensure!(
+                max_auto_expanded_paths.is_none(),
+                InvalidSqlSnafu {
+                    msg: format!("duplicated JSON2 option '{MAX_AUTO_EXPANDED_PATHS}'")
+                }
+            );
+            parser.expect_token(&Token::Eq).context(SyntaxSnafu)?;
+
+            let token = parser.next_token();
+            let Token::Number(value, _) = token.token else {
+                return InvalidSqlSnafu {
+                    msg: format!(
+                        "JSON2 option '{MAX_AUTO_EXPANDED_PATHS}' expects a non-negative integer"
+                    ),
+                }
+                .fail();
+            };
+            max_auto_expanded_paths = Some(value.parse::<u32>().map_err(|_| {
+                InvalidSqlSnafu {
+                    msg: format!(
+                        "JSON2 option '{MAX_AUTO_EXPANDED_PATHS}' expects a non-negative integer"
+                    ),
+                }
+                .build()
+            })?);
+        } else {
+            let hint = parse_json2_type_hint(parser)?;
+            ensure_no_path_conflict(&type_hints, &hint.path)?;
+            type_hints.push(hint);
+        }
 
         if parser.consume_token(&Token::Comma) {
             if parser.consume_token(&Token::RParen) {
@@ -72,11 +110,22 @@ fn parse_json2_type_hints(parser: &mut Parser<'_>) -> Result<Vec<JsonTypeHint>>
         }
     }
 
-    Ok(hints)
+    Ok(Some(Json2Options {
+        max_auto_expanded_paths,
+        type_hints,
+    }))
 }
 
 fn parse_json2_type_hint(parser: &mut Parser<'_>) -> Result<JsonTypeHint> {
     let path = parse_json2_path(parser)?;
+    ensure!(
+        path.first().is_none_or(|x| x != JSON2_REMAINDER_FIELD_NAME),
+        InvalidSqlSnafu {
+            msg: format!(
+                "JSON2 type hint path cannot be rooted at reserved field '{JSON2_REMAINDER_FIELD_NAME}'"
+            )
+        }
+    );
     ensure!(
         path.len() <= JSON2_MAX_STRUCTURED_DEPTH,
         InvalidSqlSnafu {
@@ -309,7 +358,7 @@ CREATE TABLE traces (
             column.column_def.data_type,
             DataType::Custom(_, _)
         ));
-        let hints = column.extensions.json_type_hints;
+        let hints = column.extensions.json2_options.unwrap().type_hints;
         assert_eq!(hints.len(), 4);
 
         assert_eq!(hints[0].path, vec!["service.name"]);
@@ -339,6 +388,71 @@ CREATE TABLE traces (
         assert!(hints[3].nullable);
     }
 
+    #[test]
+    fn test_parse_json2_max_auto_expanded_paths() {
+        let column = parse_json2_column(
+            r#"
+CREATE TABLE traces (
+    log_json_data JSON2 (
+        http.method STRING,
+        max_auto_expanded_paths = 0
+    ),
+    ts TIMESTAMP TIME INDEX,
+)"#,
+        );
+
+        let options = column.extensions.json2_options.unwrap();
+        assert_eq!(options.max_auto_expanded_paths, Some(0));
+        assert_eq!(options.type_hints.len(), 1);
+
+        let empty = parse_json2_column(
+            r#"
+CREATE TABLE traces (
+    log_json_data JSON2 (),
+    ts TIMESTAMP TIME INDEX,
+)"#,
+        );
+        assert!(empty.extensions.json2_options.is_none());
+
+        let quoted = parse_json2_column(
+            r#"
+CREATE TABLE traces (
+    log_json_data JSON2 (
+        "max_auto_expanded_paths" STRING,
+        nested."!__remainder__!" STRING
+    ),
+    ts TIMESTAMP TIME INDEX,
+)"#,
+        );
+        let options = quoted.extensions.json2_options.unwrap();
+        assert_eq!(options.max_auto_expanded_paths, None);
+        assert_eq!(options.type_hints.len(), 2);
+    }
+
+    #[test]
+    fn test_parse_json2_max_auto_expanded_paths_rejects_invalid_options() {
+        for options in [
+            "max_auto_expanded_paths = 0, max_auto_expanded_paths = 1",
+            "max_auto_expanded_paths = -1",
+            "max_auto_expanded_paths = 1.5",
+            "max_auto_expanded_paths = 4294967296",
+            r#""!__remainder__!".value STRING"#,
+        ] {
+            let sql = format!(
+                "CREATE TABLE traces (log_json_data JSON2 ({options}), ts TIMESTAMP TIME INDEX)"
+            );
+            assert!(
+                ParserContext::create_with_dialect(
+                    &sql,
+                    &GreptimeDbDialect {},
+                    ParseOptions::default()
+                )
+                .is_err(),
+                "{options}"
+            );
+        }
+    }
+
     #[test]
     fn test_parse_json2_type_hint_default_nullable() {
         let column = parse_json2_column(
@@ -349,7 +463,7 @@ CREATE TABLE traces (
 )"#,
         );
 
-        let hints = column.extensions.json_type_hints;
+        let hints = column.extensions.json2_options.unwrap().type_hints;
         assert_eq!(hints.len(), 1);
         assert!(hints[0].nullable);
     }
@@ -369,7 +483,7 @@ CREATE TABLE traces (
 )"#,
         );
 
-        let hints = column.extensions.json_type_hints;
+        let hints = column.extensions.json2_options.unwrap().type_hints;
         assert_eq!(hints.len(), 4);
         assert_eq!(hints[0].path, vec!["a", "b"]);
         assert_eq!(hints[1].path, vec!["x", "y"]);
@@ -402,7 +516,7 @@ CREATE TABLE traces (
 )"#,
         );
 
-        let hints = column.extensions.json_type_hints;
+        let hints = column.extensions.json2_options.unwrap().type_hints;
         assert_eq!(hints.len(), 14);
         for hint in hints.iter().take(6) {
             assert_eq!(hint.data_type, DataType::BigInt(None));
@@ -428,7 +542,7 @@ CREATE TABLE traces (
 )"#,
         );
 
-        let hints = column.extensions.json_type_hints;
+        let hints = column.extensions.json2_options.unwrap().type_hints;
         assert_eq!(hints.len(), 2);
         assert_eq!(
             hints[0]

src/sql/src/statements.rs

@@ -718,11 +718,7 @@ mod tests {
                         "true".to_string(),
                     ),
                 ])),
-                vector_options: None,
-                skipping_index_options: None,
-                inverted_index_options: None,
-                json_type_hints: vec![],
-                vector_index_options: None,
+                ..Default::default()
             },
         };
 
@@ -749,17 +745,13 @@ mod tests {
                 options: vec![],
             },
             extensions: ColumnExtensions {
-                fulltext_index_options: None,
-                vector_options: None,
-                skipping_index_options: None,
-                inverted_index_options: None,
-                json_type_hints: vec![],
                 vector_index_options: Some(OptionMap::from([
                     ("metric".to_string(), "cosine".to_string()),
                     ("connectivity".to_string(), "32".to_string()),
                     ("expansion_add".to_string(), "200".to_string()),
                     ("expansion_search".to_string(), "100".to_string()),
                 ])),
+                ..Default::default()
             },
         };
 
@@ -790,12 +782,8 @@ mod tests {
                 options: vec![],
             },
             extensions: ColumnExtensions {
-                fulltext_index_options: None,
-                vector_options: None,
-                skipping_index_options: None,
-                inverted_index_options: None,
-                json_type_hints: vec![],
                 vector_index_options: Some(OptionMap::default()),
+                ..Default::default()
             },
         };
 

src/sql/src/statements/create.rs

@@ -136,7 +136,28 @@ pub struct ColumnExtensions {
     pub inverted_index_options: Option<OptionMap>,
     /// Vector index options for HNSW-based vector similarity search.
     pub vector_index_options: Option<OptionMap>,
-    pub json_type_hints: Vec<JsonTypeHint>,
+    /// JSON2-specific column options.
+    pub json2_options: Option<Json2Options>,
+}
+
+/// JSON2-specific options represented in the SQL AST.
+#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
+pub struct Json2Options {
+    /// Maximum number of unhinted JSON2 paths expanded into Arrow fields.
+    pub(crate) max_auto_expanded_paths: Option<u32>,
+    /// Paths stored as explicitly typed JSON2 fields.
+    pub(crate) type_hints: Vec<JsonTypeHint>,
+}
+
+impl Display for Json2Options {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let mut options = Vec::with_capacity(self.type_hints.len() + 1);
+        if let Some(max) = self.max_auto_expanded_paths {
+            options.push(format!("max_auto_expanded_paths = {max}"));
+        }
+        options.extend(self.type_hints.iter().map(format_json_type_hint));
+        write!(f, "(\n    {}\n  )", options.iter().join(",\n    "))
+    }
 }
 
 #[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
@@ -180,12 +201,8 @@ impl Display for Column {
         }
 
         write!(f, "{} {}", self.column_def.name, self.column_def.data_type)?;
-        if !self.extensions.json_type_hints.is_empty() {
-            write!(
-                f,
-                "{}",
-                format_json_type_hints(&self.extensions.json_type_hints)
-            )?;
+        if let Some(options) = &self.extensions.json2_options {
+            write!(f, "{options}")?;
         }
         for option in &self.column_def.options {
             write!(f, " {option}")?;
@@ -335,29 +352,30 @@ impl ColumnExtensions {
     }
 
     pub fn build_json_settings(&self) -> Result<Option<JsonSettings>> {
-        if self.json_type_hints.is_empty() {
+        let Some(options) = &self.json2_options else {
             return Ok(None);
-        }
+        };
 
-        Ok(Some(JsonSettings::new(
-            self.json_type_hints
-                .iter()
-                .map(|hint| {
-                    Ok(datatypes::json::JsonTypeHint {
-                        path: hint.path.clone(),
-                        data_type: json_type_hint_concrete_data_type(&hint.data_type)?,
-                        nullable: hint.nullable,
-                        default_constraint: build_json_type_hint_default_constraint(hint)?,
-                        inverted_index: hint.inverted_index,
-                    })
+        let type_hints = options
+            .type_hints
+            .iter()
+            .map(|hint| {
+                Ok(datatypes::json::JsonTypeHint {
+                    path: hint.path.clone(),
+                    data_type: json_type_hint_concrete_data_type(&hint.data_type)?,
+                    nullable: hint.nullable,
+                    default_constraint: build_json_type_hint_default_constraint(hint)?,
+                    inverted_index: hint.inverted_index,
                 })
-                .collect::<Result<Vec<_>>>()?,
-        )))
+            })
+            .collect::<Result<Vec<_>>>()?;
+        let settings = JsonSettings::try_new(type_hints, options.max_auto_expanded_paths)?;
+        Ok(Some(settings))
     }
 
     pub fn set_json_settings(&mut self, settings: JsonSettings) -> Result<()> {
-        self.json_type_hints = settings
-            .type_hints
+        let (type_hints, max_auto_expanded_paths) = settings.into_parts();
+        let type_hints = type_hints
             .into_iter()
             .map(|hint| {
                 let data_type = json_type_hint_sql_data_type(&hint.data_type)?;
@@ -374,6 +392,11 @@ impl ColumnExtensions {
                 })
             })
             .collect::<Result<Vec<_>>>()?;
+        self.json2_options = (max_auto_expanded_paths.is_some() || !type_hints.is_empty())
+            .then_some(Json2Options {
+                max_auto_expanded_paths,
+                type_hints,
+            });
         Ok(())
     }
 }
@@ -487,13 +510,6 @@ fn format_json_type_hint(hint: &JsonTypeHint) -> String {
     )
 }
 
-fn format_json_type_hints(hints: &[JsonTypeHint]) -> String {
-    format!(
-        "(\n    {}\n  )",
-        hints.iter().map(format_json_type_hint).join(",\n    ")
-    )
-}
-
 fn format_json_path_segment(segment: &str) -> String {
     format!("\"{}\"", segment.replace('"', "\"\""))
 }
@@ -786,6 +802,7 @@ mod tests {
     use datatypes::schema::ColumnDefaultConstraint;
     use datatypes::value::Value;
 
+    use super::*;
     use crate::dialect::GreptimeDbDialect;
     use crate::error::Error;
     use crate::parser::{ParseOptions, ParserContext};
@@ -1066,7 +1083,7 @@ ENGINE=mito
             .build_json_settings()
             .unwrap()
             .unwrap();
-        let hints = settings.type_hints;
+        let hints = settings.type_hints();
 
         assert_eq!(hints[0].data_type, ConcreteDataType::int64_datatype());
         assert_eq!(
@@ -1118,10 +1135,10 @@ ENGINE=mito
     }
 
     #[test]
-    fn test_set_json_settings_normalizes_type_hint_sql_types() {
+    fn test_set_json_settings_normalizes_type_hint_sql_types() -> Result<()> {
         let mut extensions = super::ColumnExtensions::default();
-        extensions
-            .set_json_settings(JsonSettings::new(vec![
+        let settings = JsonSettings::try_new(
+            vec![
                 DatatypeJsonTypeHint {
                     path: vec!["i".to_string()],
                     data_type: ConcreteDataType::int32_datatype(),
@@ -1157,36 +1174,51 @@ ENGINE=mito
                     default_constraint: None,
                     inverted_index: false,
                 },
-            ]))
-            .unwrap();
+            ],
+            None,
+        )?;
+        extensions.set_json_settings(settings)?;
 
         assert_eq!(
             extensions
-                .json_type_hints
+                .json2_options
+                .unwrap()
+                .type_hints
                 .iter()
                 .map(|hint| hint.data_type.to_string())
                 .collect::<Vec<_>>(),
             vec!["BIGINT", "DOUBLE", "BIGINT UNSIGNED", "STRING", "BOOLEAN"]
         );
+        Ok(())
     }
 
     #[test]
-    fn test_set_json_settings_rejects_unsupported_type_hint_type() {
-        let mut extensions = super::ColumnExtensions::default();
-        let err = extensions
-            .set_json_settings(JsonSettings::new(vec![DatatypeJsonTypeHint {
+    fn test_set_json_settings_rejects_unsupported_type_hint_type() -> Result<()> {
+        let err = JsonSettings::try_new(
+            vec![DatatypeJsonTypeHint {
                 path: vec!["u".to_string()],
                 data_type: ConcreteDataType::date_datatype(),
                 nullable: true,
                 default_constraint: None,
                 inverted_index: false,
-            }]))
-            .unwrap_err();
+            }],
+            None,
+        )
+        .unwrap_err();
 
         assert!(
             err.to_string()
                 .contains("unsupported JSON2 type hint data type")
         );
+        Ok(())
+    }
+
+    #[test]
+    fn test_set_empty_json_settings_omits_json2_options() -> Result<()> {
+        let mut extensions = ColumnExtensions::default();
+        extensions.set_json_settings(JsonSettings::default())?;
+        assert!(extensions.json2_options.is_none());
+        Ok(())
     }
 
     #[test]
@@ -1364,15 +1396,11 @@ AS SELECT number FROM numbers_input where number > 10"#,
 
         // Test zero connectivity should fail
         let extensions = ColumnExtensions {
-            fulltext_index_options: None,
-            vector_options: None,
-            skipping_index_options: None,
-            inverted_index_options: None,
-            json_type_hints: vec![],
             vector_index_options: Some(OptionMap::from([(
                 "connectivity".to_string(),
                 "0".to_string(),
             )])),
+            ..Default::default()
         };
         let result = extensions.build_vector_index_options();
         assert!(result.is_err());
@@ -1385,15 +1413,11 @@ AS SELECT number FROM numbers_input where number > 10"#,
 
         // Test zero expansion_add should fail
         let extensions = ColumnExtensions {
-            fulltext_index_options: None,
-            vector_options: None,
-            skipping_index_options: None,
-            inverted_index_options: None,
-            json_type_hints: vec![],
             vector_index_options: Some(OptionMap::from([(
                 "expansion_add".to_string(),
                 "0".to_string(),
             )])),
+            ..Default::default()
         };
         let result = extensions.build_vector_index_options();
         assert!(result.is_err());
@@ -1406,15 +1430,11 @@ AS SELECT number FROM numbers_input where number > 10"#,
 
         // Test zero expansion_search should fail
         let extensions = ColumnExtensions {
-            fulltext_index_options: None,
-            vector_options: None,
-            skipping_index_options: None,
-            inverted_index_options: None,
-            json_type_hints: vec![],
             vector_index_options: Some(OptionMap::from([(
                 "expansion_search".to_string(),
                 "0".to_string(),
             )])),
+            ..Default::default()
         };
         let result = extensions.build_vector_index_options();
         assert!(result.is_err());
@@ -1427,16 +1447,12 @@ AS SELECT number FROM numbers_input where number > 10"#,
 
         // Test valid values should succeed
         let extensions = ColumnExtensions {
-            fulltext_index_options: None,
-            vector_options: None,
-            skipping_index_options: None,
-            inverted_index_options: None,
-            json_type_hints: vec![],
             vector_index_options: Some(OptionMap::from([
                 ("connectivity".to_string(), "32".to_string()),
                 ("expansion_add".to_string(), "200".to_string()),
                 ("expansion_search".to_string(), "100".to_string()),
             ])),
+            ..Default::default()
         };
         let result = extensions.build_vector_index_options();
         assert!(result.is_ok());

tests/compatibility/cases/json2_storage_layout_settings/case.toml

@@ -0,0 +1,8 @@
+name = "json2_storage_layout_settings"
+reason = "Verify JSON2 storage layout settings in Arrow extension metadata survive restarts and upgrades."
+introduced_by = "PR #8895"
+topologies = ["distributed", "standalone"]
+from_range = [">=v1.3.0"]
+to_range = [">=v1.3.0"]
+features = ["json2", "table", "metadata", "read"]
+owner = "query"

tests/compatibility/cases/json2_storage_layout_settings/setup.sql

@@ -0,0 +1,11 @@
+CREATE TABLE t_json2_storage_layout_settings (
+  ts TIMESTAMP TIME INDEX,
+  j JSON2(max_auto_expanded_paths = 0)
+) WITH (
+  'append_mode' = 'true'
+);
+
+INSERT INTO t_json2_storage_layout_settings VALUES
+  ('2026-08-17 00:00:00+0000', '{"a": 1}');
+
+ADMIN FLUSH_TABLE('t_json2_storage_layout_settings');

tests/compatibility/cases/json2_storage_layout_settings/verify.result

@@ -0,0 +1,27 @@
+SELECT j.a AS a
+FROM t_json2_storage_layout_settings;
+
++---+
+| a |
++---+
+| 1 |
++---+
+
+SHOW CREATE TABLE t_json2_storage_layout_settings;
+
++---------------------------------+----------------------------------------------------------------+
+| Table                           | Create Table                                                   |
++---------------------------------+----------------------------------------------------------------+
+| t_json2_storage_layout_settings | CREATE TABLE IF NOT EXISTS "t_json2_storage_layout_settings" ( |
+|                                 |   "ts" TIMESTAMP(3) NOT NULL,                                  |
+|                                 |   "j" JSON2(                                                   |
+|                                 |     max_auto_expanded_paths = 0                                |
+|                                 |   ) NULL,                                                      |
+|                                 |   TIME INDEX ("ts")                                            |
+|                                 | )                                                              |
+|                                 |                                                                |
+|                                 | ENGINE=mito                                                    |
+|                                 | WITH(                                                          |
+|                                 |   append_mode = 'true'                                         |
+|                                 | )                                                              |
++---------------------------------+----------------------------------------------------------------+

tests/compatibility/cases/json2_storage_layout_settings/verify.sql

@@ -0,0 +1,4 @@
+SELECT j.a AS a
+FROM t_json2_storage_layout_settings;
+
+SHOW CREATE TABLE t_json2_storage_layout_settings;