refactor: json2 v2 storage layout (#8979)

* refactor: json2 v2 storage layout

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

* resolve PR comments

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

* fix ci

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

* rethinking when "needs_remainder"

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

* restore "ReadColumns"

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

* resolve PR comments

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

* fix ci

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

---------

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

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::{Json2ExtensionType, JsonMetadata};
+use datatypes::json::JsonSettings;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{COMMENT_KEY, ColumnDefaultConstraint, ColumnSchema};
 use datatypes::types::json_type::JsonNativeType;
@@ -163,7 +164,10 @@ pub fn column_to_schema(
         false
     };
     if is_json2_column {
-        let settings = column.extensions.build_json_settings()?.unwrap_or_default();
+        let settings = column
+            .extensions
+            .build_json_settings()?
+            .unwrap_or_else(JsonSettings::new_v2);
         let extension = Json2ExtensionType::new(Arc::new(JsonMetadata::new(settings)));
         column_schema.with_extension_type(&extension);
     }
@@ -643,6 +647,53 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_new_json2_column_uses_v2_layout() -> std::result::Result<(), Box<dyn std::error::Error>>
+    {
+        let column = Column {
+            column_def: ColumnDef {
+                name: "data".into(),
+                data_type: SqlDataType::Custom(
+                    sqlparser::ast::ObjectName::from(vec!["JSON2".into()]),
+                    vec![],
+                ),
+                options: vec![],
+            },
+            extensions: ColumnExtensions::default(),
+        };
+
+        let schema = column_to_schema(&column, "ts", None)?;
+        let metadata: serde_json::Value =
+            serde_json::from_str(schema.metadata().get("ARROW:extension:metadata").unwrap())?;
+        assert_eq!(Some(2), metadata["layout_version"].as_u64());
+        assert_eq!(
+            Some(100),
+            metadata["json_settings"]["max_auto_expanded_paths"].as_u64()
+        );
+
+        let mut hinted = column;
+        hinted
+            .extensions
+            .set_json_settings(datatypes::json::JsonSettings::try_new(
+                vec![datatypes::json::JsonTypeHint {
+                    path: vec!["kind".to_string()],
+                    data_type: ConcreteDataType::string_datatype(),
+                    nullable: true,
+                    default_constraint: None,
+                    inverted_index: false,
+                }],
+                None,
+            )?)?;
+        let schema = column_to_schema(&hinted, "ts", None)?;
+        let metadata: serde_json::Value =
+            serde_json::from_str(schema.metadata().get("ARROW:extension:metadata").unwrap())?;
+        assert_eq!(
+            Some(100),
+            metadata["json_settings"]["max_auto_expanded_paths"].as_u64()
+        );
+        Ok(())
+    }
+
     #[test]
     pub fn test_column_to_schema_timestamp_with_timezone() {
         let column = Column {

src/sql/src/statements/create.rs

@@ -17,7 +17,7 @@ use std::fmt::{Display, Formatter};
 
 use common_catalog::consts::FILE_ENGINE;
 use common_sql::default_constraint::parse_column_default_constraint;
-use datatypes::json::JsonSettings;
+use datatypes::json::{JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS, JsonSettings};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{
     ColumnDefaultConstraint, FulltextOptions, SkippingIndexOptions, VectorDistanceMetric,
@@ -369,7 +369,12 @@ impl ColumnExtensions {
                 })
             })
             .collect::<Result<Vec<_>>>()?;
-        let settings = JsonSettings::try_new(type_hints, options.max_auto_expanded_paths)?;
+        let settings = JsonSettings::try_new(
+            type_hints,
+            options
+                .max_auto_expanded_paths
+                .or(Some(JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS)),
+        )?;
         Ok(Some(settings))
     }
 
@@ -1017,6 +1022,31 @@ ENGINE=mito
         }
     }
 
+    #[test]
+    fn test_parse_json2_max_auto_expanded_paths_option() -> Result<()> {
+        let sql = r#"CREATE TABLE traces (
+            log_json_data JSON2 (
+                status_code INT64 NOT NULL,
+                max_auto_expanded_paths = 1
+            ),
+            ts TIMESTAMP TIME INDEX
+        )"#;
+        let result = ParserContext::create_with_dialect(
+            sql,
+            &GreptimeDbDialect {},
+            ParseOptions::default(),
+        )?;
+        let Statement::CreateTable(create_table) = &result[0] else {
+            unreachable!()
+        };
+        let settings = create_table.columns[0]
+            .extensions
+            .build_json_settings()?
+            .unwrap();
+        assert_eq!(settings.max_auto_expanded_paths(), Some(1));
+        Ok(())
+    }
+
     #[test]
     fn test_display_json2_type_hints_quotes_numeric_segments() {
         let sql = r#"CREATE TABLE traces (