feat: create table with new json datatype (#7128)

* feat: create table with new json datatype

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/sql/Cargo.toml

@@ -29,6 +29,7 @@ datafusion-expr.workspace = true
 datafusion-physical-expr.workspace = true
 datafusion-sql.workspace = true
 datatypes.workspace = true
+either.workspace = true
 hex = "0.4"
 humantime.workspace = true
 iso8601 = "0.6.1"

src/sql/src/error.rs

@@ -332,6 +332,14 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Failed to set JSON structure settings: {value}"))]
+    SetJsonStructureSettings {
+        value: String,
+        source: datatypes::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {
@@ -377,7 +385,9 @@ impl ErrorExt for Error {
             #[cfg(feature = "enterprise")]
             InvalidTriggerWebhookOption { .. } => StatusCode::InvalidArguments,
 
-            SerializeColumnDefaultConstraint { source, .. } => source.status_code(),
+            SerializeColumnDefaultConstraint { source, .. }
+            | SetJsonStructureSettings { source, .. } => source.status_code(),
+
             ConvertToGrpcDataType { source, .. } => source.status_code(),
             SqlCommon { source, .. } => source.status_code(),
             ConvertToDfStatement { .. } => StatusCode::Internal,

src/sql/src/statements.rs

@@ -49,8 +49,8 @@ use crate::ast::{
 };
 use crate::error::{
     self, ConvertToGrpcDataTypeSnafu, ConvertValueSnafu, Result,
-    SerializeColumnDefaultConstraintSnafu, SetFulltextOptionSnafu, SetSkippingIndexOptionSnafu,
-    SqlCommonSnafu,
+    SerializeColumnDefaultConstraintSnafu, SetFulltextOptionSnafu, SetJsonStructureSettingsSnafu,
+    SetSkippingIndexOptionSnafu, SqlCommonSnafu,
 };
 use crate::statements::create::Column;
 pub use crate::statements::option_map::OptionMap;
@@ -144,6 +144,18 @@ pub fn column_to_schema(
 
     column_schema.set_inverted_index(column.extensions.inverted_index_options.is_some());
 
+    if matches!(column.data_type(), SqlDataType::JSON) {
+        let settings = column
+            .extensions
+            .build_json_structure_settings()?
+            .unwrap_or_default();
+        column_schema
+            .with_json_structure_settings(&settings)
+            .with_context(|_| SetJsonStructureSettingsSnafu {
+                value: format!("{settings:?}"),
+            })?;
+    }
+
     Ok(column_schema)
 }
 

src/sql/src/statements/create.rs

@@ -32,6 +32,7 @@ use crate::error::{
 use crate::statements::OptionMap;
 use crate::statements::statement::Statement;
 use crate::statements::tql::Tql;
+use crate::util::OptionValue;
 
 const LINE_SEP: &str = ",\n";
 const COMMA_SEP: &str = ", ";
@@ -166,7 +167,20 @@ impl Display for Column {
             return Ok(());
         }
 
-        write!(f, "{}", self.column_def)?;
+        write!(f, "{} {}", self.column_def.name, self.column_def.data_type)?;
+        if let Some(options) = &self.extensions.json_datatype_options {
+            write!(
+                f,
+                "({})",
+                options
+                    .entries()
+                    .map(|(k, v)| format!("{k} = {v}"))
+                    .join(COMMA_SEP)
+            )?;
+        }
+        for option in &self.column_def.options {
+            write!(f, " {option}")?;
+        }
 
         if let Some(fulltext_options) = &self.extensions.fulltext_index_options {
             if !fulltext_options.is_empty() {
@@ -251,6 +265,34 @@ impl ColumnExtensions {
             })
             .transpose()
     }
+
+    pub fn set_json_structure_settings(&mut self, settings: JsonStructureSettings) {
+        let mut map = OptionMap::default();
+
+        let format = match settings {
+            JsonStructureSettings::Structured(_) => JSON_FORMAT_FULL_STRUCTURED,
+            JsonStructureSettings::PartialUnstructuredByKey { .. } => JSON_FORMAT_PARTIAL,
+            JsonStructureSettings::UnstructuredRaw => JSON_FORMAT_RAW,
+        };
+        map.insert(JSON_OPT_FORMAT.to_string(), format.to_string());
+
+        if let JsonStructureSettings::PartialUnstructuredByKey {
+            fields: _,
+            unstructured_keys,
+        } = settings
+        {
+            let value = OptionValue::from(
+                unstructured_keys
+                    .iter()
+                    .map(|x| x.as_str())
+                    .sorted()
+                    .collect::<Vec<_>>(),
+            );
+            map.insert_options(JSON_OPT_UNSTRUCTURED_KEYS, value);
+        }
+
+        self.json_datatype_options = Some(map);
+    }
 }
 
 /// Partition on columns or values.

src/sql/src/statements/option_map.rs

@@ -16,6 +16,7 @@ use std::collections::{BTreeMap, HashMap};
 use std::ops::ControlFlow;
 
 use common_base::secrets::{ExposeSecret, ExposeSecretMut, SecretString};
+use either::Either;
 use serde::Serialize;
 use sqlparser::ast::{Visit, VisitMut, Visitor, VisitorMut};
 
@@ -56,6 +57,17 @@ impl OptionMap {
         }
     }
 
+    pub fn insert_options(&mut self, key: &str, value: OptionValue) {
+        if REDACTED_OPTIONS.contains(&key) {
+            self.secrets.insert(
+                key.to_string(),
+                SecretString::new(Box::new(value.to_string())),
+            );
+        } else {
+            self.options.insert(key.to_string(), value);
+        }
+    }
+
     pub fn get(&self, k: &str) -> Option<&str> {
         if let Some(value) = self.options.get(k) {
             value.as_string()
@@ -130,6 +142,18 @@ impl OptionMap {
         }
         result
     }
+
+    pub fn entries(&self) -> impl Iterator<Item = (&str, Either<&OptionValue, &str>)> {
+        let options = self
+            .options
+            .iter()
+            .map(|(k, v)| (k.as_str(), Either::Left(v)));
+        let secrets = self
+            .secrets
+            .keys()
+            .map(|k| (k.as_str(), Either::Right("******")));
+        std::iter::chain(options, secrets)
+    }
 }
 
 impl<I: IntoIterator<Item = (String, String)>> From<I> for OptionMap {

src/sql/src/util.rs

@@ -15,6 +15,7 @@
 use std::collections::HashSet;
 use std::fmt::{Display, Formatter};
 
+use itertools::Itertools;
 use serde::Serialize;
 use snafu::ensure;
 use sqlparser::ast::{
@@ -131,6 +132,22 @@ impl From<Vec<&str>> for OptionValue {
     }
 }
 
+impl Display for OptionValue {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        if let Some(s) = self.as_string() {
+            write!(f, "'{s}'")
+        } else if let Some(s) = self.as_list() {
+            write!(
+                f,
+                "[{}]",
+                s.into_iter().map(|x| format!("'{x}'")).join(", ")
+            )
+        } else {
+            write!(f, "'{}'", self.0)
+        }
+    }
+}
+
 pub fn parse_option_string(option: SqlOption) -> Result<(String, OptionValue)> {
     let SqlOption::KeyValue { key, value } = option else {
         return InvalidSqlSnafu {