rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-13 08:52:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: optimize json2 write (#8393)

Browse Source 
refactor: optimize json2 write performance

Signed-off-by: luofucong <luofc@foxmail.com>
This commit is contained in:
LFC
2026-07-09 07:08:57 +00:00
committed by GitHub
parent 59e4e7efda
commit 6ae687dc8e
12 changed files with 496 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/datatypes/src/data_type.rs
+1 -1
View File
@@ -685,7 +685,7 @@ impl ConcreteDataType {
}
pub fn json2(native_type: JsonNativeType) -> ConcreteDataType {
ConcreteDataType::Json(JsonType::new_json2(native_type))
ConcreteDataType::Json(JsonType::json2(Arc::new(native_type)))
}
}
src/datatypes/src/error.rs
+10 -9
View File
@@ -264,13 +264,6 @@ pub enum Error {
location: Location,
},
#[snafu(display("Failed to merge JSON datatype: {reason}"))]
MergeJsonDatatype {
reason: String,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to parse or serialize arrow metadata"))]
ArrowMetadata {
#[snafu(source)]
@@ -298,6 +291,13 @@ pub enum Error {
#[snafu(implicit)]
location: Location,
},
#[snafu(display("unexpected: {reason}"))]
Unexpected {
reason: String,
#[snafu(implicit)]
location: Location,
},
}
impl ErrorExt for Error {
@@ -321,8 +321,7 @@ impl ErrorExt for Error {
| InvalidJsonb { .. }
| InvalidVector { .. }
| InvalidFulltextOption { .. }
| InvalidSkippingIndexOption { .. }
| MergeJsonDatatype { .. } => StatusCode::InvalidArguments,
| InvalidSkippingIndexOption { .. } => StatusCode::InvalidArguments,
ValueExceedsPrecision { .. }
| CastType { .. }
@@ -344,6 +343,8 @@ impl ErrorExt for Error {
| ArrowMetadata { .. }
| AlignJsonValue { .. }
| AlignJsonArray { .. } => StatusCode::Internal,
Unexpected { .. } => StatusCode::Unexpected,
}
}
src/datatypes/src/json.rs
+55 -45
View File
@@ -82,34 +82,24 @@ impl JsonSettings {
/// Decode an encoded StructValue back into a serde_json::Value.
pub fn decode(&self, value: Value) -> Result<Json> {
let context = JsonContext {
let mut context = JsonContext {
path: Vec::new(),
settings: self,
};
decode_value_with_context(value, &context)
decode_value_with_context(value, &mut context)
}
/// Encode a serde_json::Value into a Value::Json using current settings.
pub fn encode(&self, json: Json) -> Result<Value> {
let context = JsonContext {
let mut context = JsonContext {
path: Vec::new(),
settings: self,
};
encode_json_with_context(json, &context).map(|v| Value::Json(Box::new(v)))
encode_json_with_context(json, &mut context).map(|v| Value::Json(Box::new(v)))
}
}
impl<'a> JsonContext<'a> {
/// Create a new context with an updated key path
pub fn with_key(&self, key: &str) -> JsonContext<'a> {
let mut path = self.path.clone();
path.push(key.to_string());
JsonContext {
path,
settings: self.settings,
}
}
fn type_hint(&self) -> Option<&'a JsonTypeHint> {
self.settings
.type_hints
@@ -118,8 +108,19 @@ impl<'a> JsonContext<'a> {
}
}
fn with_key_context<T>(
context: &mut JsonContext,
key: &str,
f: impl FnOnce(&mut JsonContext) -> Result<T>,
) -> Result<T> {
context.path.push(key.to_string());
let result = f(context);
context.path.pop();
result
}
/// Main encoding function with key path tracking
pub fn encode_json_with_context<'a>(json: Json, context: &JsonContext<'a>) -> Result<JsonValue> {
fn encode_json_with_context(json: Json, context: &mut JsonContext) -> Result<JsonValue> {
if context.path.is_empty() && !matches!(json, Json::Object(_)) {
return UnsupportedJsonTypeSnafu.fail();
}
@@ -133,17 +134,17 @@ pub fn encode_json_with_context<'a>(json: Json, context: &JsonContext<'a>) -> Re
fn encode_json_object_with_context<'a>(
json_object: Map<String, Json>,
context: &JsonContext<'a>,
context: &mut JsonContext<'a>,
) -> Result<JsonValue> {
let mut object = BTreeMap::new();
for (key, value) in json_object {
let field_context = context.with_key(&key);
let value = if let Some(hint) = field_context.type_hint() {
encode_json_value_with_hint(value, hint, &field_context)?
} else {
encode_json_value_with_context(value, &field_context)?
};
let value = with_key_context(context, &key, |context| {
if let Some(hint) = context.type_hint() {
encode_json_value_with_hint(value, hint, context)
} else {
encode_json_value_with_context(value, context)
}
})?;
object.insert(key, value.into_variant());
}
@@ -155,11 +156,15 @@ fn encode_json_object_with_context<'a>(
fn apply_missing_type_hints(
object: &mut BTreeMap<String, JsonVariant>,
context: &JsonContext,
context: &mut JsonContext,
) -> Result<()> {
for hint in &context.settings.type_hints {
if hint.path.len() > context.path.len() && hint.path.starts_with(&context.path) {
insert_missing_type_hint(object, context, hint, context.path.len())?;
let depth = context.path.len();
let key = &hint.path[depth];
with_key_context(context, key, |context| {
insert_missing_type_hint(object, context, hint, depth)
})?;
}
}
Ok(())
@@ -167,17 +172,16 @@ fn apply_missing_type_hints(
fn insert_missing_type_hint(
object: &mut BTreeMap<String, JsonVariant>,
context: &JsonContext,
field_context: &mut JsonContext,
hint: &JsonTypeHint,
depth: usize,
) -> Result<()> {
let key = &hint.path[depth];
let field_context = context.with_key(key);
let is_leaf = depth + 1 == hint.path.len();
if is_leaf {
if !object.contains_key(key) {
let value = encode_missing_type_hint_value(hint, &field_context)?;
let value = encode_missing_type_hint_value(hint, field_context)?;
object.insert(key.clone(), value.into_variant());
}
return Ok(());
@@ -186,7 +190,7 @@ 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)
}
_ => error::InvalidJsonSnafu {
value: format!(
@@ -199,14 +203,17 @@ 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)?;
entry.insert(JsonVariant::Object(child));
Ok(())
}
}
}
fn encode_missing_type_hint_value(hint: &JsonTypeHint, context: &JsonContext) -> Result<JsonValue> {
fn encode_missing_type_hint_value(
hint: &JsonTypeHint,
context: &mut JsonContext,
) -> Result<JsonValue> {
if let Some(default_constraint) = &hint.default_constraint {
let value = default_constraint.create_default(&hint.data_type, hint.nullable)?;
let json = decode_primitive_value(value)?;
@@ -229,7 +236,7 @@ fn encode_missing_type_hint_value(hint: &JsonTypeHint, context: &JsonContext) ->
fn encode_json_value_with_hint(
json: Json,
hint: &JsonTypeHint,
context: &JsonContext,
context: &mut JsonContext,
) -> Result<JsonValue> {
if json.is_null() {
return if hint.nullable {
@@ -291,14 +298,15 @@ fn encode_json_value_with_hint(
fn encode_json_array_with_context<'a>(
json_array: Vec<Json>,
context: &JsonContext<'a>,
context: &mut JsonContext<'a>,
) -> 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 array_context = context.with_key(&index.to_string());
let item_value = encode_json_value_with_context(value, &array_context)?;
let item_value = with_key_context(context, &index.to_string(), |context| {
encode_json_value_with_context(value, context)
})?;
items.push(item_value);
}
@@ -310,10 +318,10 @@ fn encode_json_array_with_context<'a>(
let merged_item_type = if let Some((first, rests)) = items.split_first() {
let mut merged = first.json_type().clone();
for rest in rests.iter().map(|x| x.json_type()) {
if matches!(merged.native_type(), JsonNativeType::Variant) {
if matches!(merged, JsonNativeType::Variant) {
break;
}
merged.merge(rest)?;
merged.merge(rest);
}
Some(merged)
} else {
@@ -332,7 +340,7 @@ 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<'a>(json: Json, context: &JsonContext<'a>) -> Result<JsonValue> {
fn encode_json_value_with_context(json: Json, context: &mut JsonContext) -> Result<JsonValue> {
match json {
Json::Null => Ok(JsonValue::null()),
Json::Bool(b) => Ok(b.into()),
@@ -359,7 +367,7 @@ fn encode_json_value_with_context<'a>(json: Json, context: &JsonContext<'a>) ->
}
/// Main decoding function with key path tracking
pub fn decode_value_with_context(value: Value, context: &JsonContext) -> Result<Json> {
fn decode_value_with_context(value: Value, context: &mut JsonContext) -> Result<Json> {
match value {
Value::Struct(struct_value) => decode_struct_with_context(struct_value, context),
Value::List(list_value) => decode_list_with_context(list_value, context),
@@ -370,15 +378,16 @@ pub fn decode_value_with_context(value: Value, context: &JsonContext) -> Result<
/// Decode a structured value to JSON object
fn decode_struct_with_context<'a>(
struct_value: StructValue,
context: &JsonContext<'a>,
context: &mut JsonContext<'a>,
) -> Result<Json> {
let mut json_object = Map::with_capacity(struct_value.len());
let (items, fields) = struct_value.into_parts();
for (field, field_value) in fields.fields().iter().zip(items) {
let field_context = context.with_key(field.name());
let json_value = decode_value_with_context(field_value, &field_context)?;
let json_value = with_key_context(context, field.name(), |context| {
decode_value_with_context(field_value, context)
})?;
json_object.insert(field.name().to_string(), json_value);
}
@@ -386,14 +395,15 @@ fn decode_struct_with_context<'a>(
}
/// Decode a list value to JSON array
fn decode_list_with_context(list_value: ListValue, context: &JsonContext) -> Result<Json> {
fn decode_list_with_context(list_value: ListValue, context: &mut JsonContext) -> Result<Json> {
let mut json_array = Vec::with_capacity(list_value.len());
let data_items = list_value.take_items();
for (index, item) in data_items.into_iter().enumerate() {
let array_context = context.with_key(&index.to_string());
let json_value = decode_value_with_context(item, &array_context)?;
let json_value = with_key_context(context, &index.to_string(), |context| {
decode_value_with_context(item, context)
})?;
json_array.push(json_value);
}
src/datatypes/src/json/value.rs
+75 -99
View File
@@ -26,9 +26,9 @@ use snafu::{OptionExt, ensure};
use crate::Result;
use crate::data_type::ConcreteDataType;
use crate::error::{AlignJsonValueSnafu, InvalidJsonSnafu, InvalidJsonbSnafu};
use crate::types::json_type::{JsonNativeType, JsonNumberType};
use crate::types::{JsonType, StructField, StructType};
use crate::value::{ListValue, ListValueRef, StructValue, StructValueRef, Value, ValueRef};
use crate::types::json_type::{JsonNativeType, JsonNumberType, is_include};
use crate::types::{StructField, StructType};
use crate::value::{ListValue, StructValue, Value};
/// Number in json, can be a positive integer, a negative integer, or a floating number.
/// Each of which is represented as `u64`, `i64` and `f64`.
@@ -160,8 +160,14 @@ impl JsonVariant {
}
}
fn json_type(&self) -> JsonType {
JsonType::new_json2(self.native_type())
fn contains_empty_object(&self) -> bool {
match self {
JsonVariant::Array(array) => array.iter().any(JsonVariant::contains_empty_object),
JsonVariant::Object(object) => {
object.is_empty() || object.values().any(JsonVariant::contains_empty_object)
}
_ => false,
}
}
fn as_ref(&self) -> JsonVariantRef<'_> {
@@ -295,7 +301,7 @@ impl Display for JsonVariant {
#[derive(Debug, Eq, Serialize, Deserialize)]
pub struct JsonValue {
#[serde(skip)]
json_type: OnceLock<JsonType>,
json_type: OnceLock<Arc<JsonNativeType>>,
json_variant: JsonVariant,
}
@@ -311,12 +317,21 @@ impl JsonValue {
}
}
pub(crate) fn data_type(&self) -> ConcreteDataType {
ConcreteDataType::Json(self.json_type().clone())
pub(crate) fn new_with(json_variant: JsonVariant, json_type: Arc<JsonNativeType>) -> Self {
Self {
json_type: OnceLock::from(json_type),
json_variant,
}
}
pub fn json_type(&self) -> &JsonType {
self.json_type.get_or_init(|| self.json_variant.json_type())
pub(crate) fn data_type(&self) -> ConcreteDataType {
ConcreteDataType::json2(self.json_type().clone())
}
pub(crate) fn json_type(&self) -> &JsonNativeType {
self.json_type
.get_or_init(|| Arc::new(self.json_variant.native_type()))
.as_ref()
}
pub(crate) fn is_null(&self) -> bool {
@@ -368,6 +383,10 @@ impl JsonValue {
self.json_variant
}
pub(crate) fn variant(&self) -> &JsonVariant {
&self.json_variant
}
pub(crate) fn into_value(self) -> Value {
fn helper(v: JsonVariant) -> Value {
match v {
@@ -415,14 +434,13 @@ impl JsonValue {
/// - `Null` aligns to any type, and any value aligns to `Null` as `Null`.
/// - Numbers are converted only within compatible number categories.
/// - Arrays align each element recursively to the expected item type.
/// - Objects require `expected` to contain all fields from the current value. Missing expected
/// fields are filled with `Null`.
/// - Objects require `expected` to have all fields from the current value.
/// - `Variant` preserves the original JSON payload as serialized bytes.
///
/// Returns an error if the value cannot be aligned without losing existing object fields or
/// when a scalar type conversion is incompatible.
pub(crate) fn try_align(&mut self, expected: &JsonType) -> Result<()> {
if self.json_type() == expected {
pub(crate) fn try_align(&mut self, expected: &JsonNativeType) -> Result<()> {
if is_include(expected, self.json_type()) && !self.json_variant.contains_empty_object() {
return Ok(());
}
@@ -455,7 +473,6 @@ impl JsonValue {
.collect::<Result<_>>()?,
),
(JsonVariant::Object(kvs), _) if kvs.is_empty() => JsonVariant::Null,
(JsonVariant::Object(mut kvs), JsonNativeType::Object(expected)) => {
ensure!(
expected.keys().len() >= kvs.keys().len()
@@ -469,15 +486,12 @@ impl JsonValue {
}
);
let mut object = BTreeMap::new();
for (field, field_type) in expected {
if let Some((k, v)) = kvs.remove_entry(field) {
object.insert(k, helper(v, field_type)?);
} else {
object.insert(field.clone(), JsonVariant::Null);
kvs.insert(k, helper(v, field_type)?);
}
}
JsonVariant::Object(object)
JsonVariant::Object(kvs)
}
(v, JsonNativeType::Variant) => JsonVariant::Variant(encode_json_variant(v)?),
@@ -498,7 +512,7 @@ impl JsonValue {
let x = std::mem::take(&mut self.json_variant);
self.json_variant = helper(x, expected.native_type())?;
self.json_variant = helper(x, expected)?;
self.json_type = OnceLock::new();
Ok(())
}
@@ -692,10 +706,6 @@ impl JsonVariantRef<'_> {
JsonVariantRef::Variant(_) => JsonNativeType::Variant,
}
}
fn json_type(&self) -> JsonType {
JsonType::new_json2(self.native_type())
}
}
fn json_array_native_type<I>(items: I) -> JsonNativeType
@@ -780,19 +790,17 @@ impl<'a> From<BTreeMap<&'a str, JsonVariantRef<'a>>> for JsonVariantRef<'a> {
}
}
impl From<JsonVariantRef<'_>> for JsonVariant {
fn from(v: JsonVariantRef) -> Self {
impl From<&JsonVariantRef<'_>> for JsonVariant {
fn from(v: &JsonVariantRef) -> Self {
match v {
JsonVariantRef::Null => Self::Null,
JsonVariantRef::Bool(x) => Self::Bool(x),
JsonVariantRef::Number(x) => Self::Number(x),
JsonVariantRef::Bool(x) => Self::Bool(*x),
JsonVariantRef::Number(x) => Self::Number(*x),
JsonVariantRef::String(x) => Self::String(x.to_string()),
JsonVariantRef::Array(array) => {
Self::Array(array.into_iter().map(Into::into).collect())
}
JsonVariantRef::Array(array) => Self::Array(array.iter().map(Into::into).collect()),
JsonVariantRef::Object(object) => Self::Object(
object
.into_iter()
.iter()
.map(|(k, v)| (k.to_string(), v.into()))
.collect(),
),
@@ -811,7 +819,7 @@ impl<'a> From<&'a [u8]> for JsonVariantRef<'a> {
#[derive(Debug, Serialize)]
pub struct JsonValueRef<'a> {
#[serde(skip)]
json_type: OnceLock<JsonType>,
json_type: OnceLock<Arc<JsonNativeType>>,
json_variant: JsonVariantRef<'a>,
}
@@ -821,11 +829,13 @@ impl<'a> JsonValueRef<'a> {
}
pub(crate) fn data_type(&self) -> ConcreteDataType {
ConcreteDataType::Json(self.json_type().clone())
ConcreteDataType::json2(self.json_type().as_ref().clone())
}
pub(crate) fn json_type(&self) -> &JsonType {
self.json_type.get_or_init(|| self.json_variant.json_type())
pub(crate) fn json_type(&self) -> Arc<JsonNativeType> {
self.json_type
.get_or_init(|| Arc::new(self.json_variant.native_type()))
.clone()
}
pub fn into_variant(self) -> JsonVariantRef<'a> {
@@ -854,48 +864,6 @@ impl<'a> JsonValueRef<'a> {
}
}
pub fn as_value_ref(&self) -> ValueRef<'_> {
fn helper<'a>(v: &'a JsonVariantRef) -> ValueRef<'a> {
match v {
JsonVariantRef::Null => ValueRef::Null,
JsonVariantRef::Bool(x) => ValueRef::Boolean(*x),
JsonVariantRef::Number(x) => match x {
JsonNumber::PosInt(i) => ValueRef::UInt64(*i),
JsonNumber::NegInt(i) => ValueRef::Int64(*i),
JsonNumber::Float(f) => ValueRef::Float64(*f),
},
JsonVariantRef::String(x) => ValueRef::String(x),
JsonVariantRef::Array(array) => {
let val = array.iter().map(helper).collect::<Vec<_>>();
let item_datatype = if let Some(first) = val.first() {
first.data_type()
} else {
ConcreteDataType::null_datatype()
};
ValueRef::List(ListValueRef::RefList {
val,
item_datatype: Arc::new(item_datatype),
})
}
JsonVariantRef::Object(object) => {
let mut fields = Vec::with_capacity(object.len());
let mut val = Vec::with_capacity(object.len());
for (k, v) in object.iter() {
let v = helper(v);
fields.push(StructField::new(k.to_string(), v.data_type(), true));
val.push(v);
}
ValueRef::Struct(StructValueRef::RefList {
val,
fields: StructType::new(Arc::new(fields)),
})
}
JsonVariantRef::Variant(x) => ValueRef::Binary(x),
}
}
helper(&self.json_variant)
}
pub(crate) fn data_size(&self) -> usize {
size_of_val(self)
}
@@ -918,7 +886,7 @@ impl From<JsonValueRef<'_>> for JsonValue {
fn from(v: JsonValueRef<'_>) -> Self {
Self {
json_type: OnceLock::new(),
json_variant: v.json_variant.into(),
json_variant: JsonVariant::from(&v.json_variant),
}
}
}
@@ -967,17 +935,14 @@ mod tests {
// Root type can be aligned to Null, and the cached json_type must be refreshed.
let mut value = JsonValue::from(true);
assert_eq!(
value.json_type(),
&JsonType::new_json2(JsonNativeType::Bool)
);
value.try_align(&JsonType::null())?;
assert_eq!(value.json_type(), &JsonNativeType::Bool);
value.try_align(&JsonNativeType::Null)?;
assert_eq!(value, JsonValue::null());
assert_eq!(value.json_type(), &JsonType::null());
assert_eq!(value.json_type(), &JsonNativeType::Null);
// Object alignment now requires the expected type to be a superset of the
// value fields, while still filling missing expected fields with null.
let expected = JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([
let expected = JsonNativeType::Object(JsonObjectType::from([
("extra".to_string(), JsonNativeType::u64()),
(
"items".to_string(),
@@ -988,7 +953,7 @@ mod tests {
])))),
),
("name".to_string(), JsonNativeType::String),
])));
]));
let mut value = parse_json_value(r#"{"items":[{"id":1,"payload":{"k":"v"}}],"extra":1}"#);
assert_ne!(value.json_type(), &expected);
value.try_align(&expected)?;
@@ -1000,25 +965,40 @@ mod tests {
"items".to_string(),
JsonVariant::Array(vec![JsonVariant::Object(BTreeMap::from([
("id".to_string(), JsonVariant::from(1_u64)),
("note".to_string(), JsonVariant::Null),
(
"payload".to_string(),
JsonVariant::Variant(jsonb_bytes(r#"{"k":"v"}"#)),
),
]))]),
),
("name".to_string(), JsonVariant::Null),
])))
);
// Empty objects have native type Null, but the value still needs alignment
// before converting into a typed struct value.
let expected = JsonNativeType::Object(JsonObjectType::from([(
"empty".to_string(),
JsonNativeType::Null,
)]));
let mut value = parse_json_value(r#"{"empty":{}}"#);
assert_eq!(value.json_type(), &expected);
value.try_align(&expected)?;
assert_eq!(
value,
JsonValue::from(JsonVariant::Object(BTreeMap::from([(
"empty".to_string(),
JsonVariant::Null,
)])))
);
// Object alignment should fail if the expected type misses any field from the value.
let expected = JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
let expected = JsonNativeType::Object(JsonObjectType::from([(
"items".to_string(),
JsonNativeType::Array(Box::new(JsonNativeType::Object(JsonObjectType::from([
("id".to_string(), JsonNativeType::u64()),
("payload".to_string(), JsonNativeType::Variant),
])))),
)])));
)]));
let mut value =
parse_json_value(r#"{"items":[{"id":1,"payload":{"k":"v"},"extra":true}]}"#);
let err = value.try_align(&expected).unwrap_err();
@@ -1029,7 +1009,7 @@ mod tests {
// Root-level Variant alignment should preserve the original JSON payload.
let mut value = parse_json_value(r#"{"foo":[1,true,null]}"#);
value.try_align(&JsonType::new_json2(JsonNativeType::Variant))?;
value.try_align(&JsonNativeType::Variant)?;
assert_eq!(
value,
JsonValue::from(JsonVariant::Variant(jsonb_bytes(
@@ -1039,18 +1019,14 @@ mod tests {
// Incompatible scalar alignment should fail instead of coercing the value.
let mut value = JsonValue::from("hello");
let err = value
.try_align(&JsonType::new_json2(JsonNativeType::Bool))
.unwrap_err();
let err = value.try_align(&JsonNativeType::Bool).unwrap_err();
assert_eq!(
err.to_string(),
r#"Failed to align JSON value, reason: unable to align 'hello' of type "<String>" to type "<Bool>""#
);
let mut value = JsonValue::from(f64::NAN);
let err = value
.try_align(&JsonType::new_json2(JsonNativeType::Variant))
.unwrap_err();
let err = value.try_align(&JsonNativeType::Variant).unwrap_err();
assert_eq!(
err.to_string(),
"Invalid JSON: NaN is not a valid JSON number"
src/datatypes/src/types/json_type.rs
+63 -119
View File
@@ -27,13 +27,11 @@ use snafu::ResultExt;
use crate::Error;
use crate::data_type::DataType;
use crate::error::{
DeserializeSnafu, InvalidJsonSnafu, InvalidJsonbSnafu, MergeJsonDatatypeSnafu, Result,
UnsupportedArrowTypeSnafu,
DeserializeSnafu, InvalidJsonSnafu, InvalidJsonbSnafu, Result, UnsupportedArrowTypeSnafu,
};
use crate::prelude::ConcreteDataType;
use crate::scalars::ScalarVectorBuilder;
use crate::type_id::LogicalTypeId;
use crate::types::StructType;
use crate::value::Value;
use crate::vectors::json::builder::JsonVectorBuilder;
use crate::vectors::{BinaryVectorBuilder, MutableVector};
@@ -82,6 +80,10 @@ impl JsonNativeType {
Self::Number(JsonNumberType::F64)
}
fn object() -> Self {
Self::Object(JsonObjectType::new())
}
/// Merge other [JsonNativeType] into this.
/// Conflicting fields will be resolved to the "Variant" type.
pub fn merge(&mut self, other: &JsonNativeType) {
@@ -262,7 +264,7 @@ impl Display for JsonNativeType {
pub enum JsonFormat {
#[default]
Jsonb,
Json2(Box<JsonNativeType>),
Json2(Arc<JsonNativeType>),
}
/// JsonType is a data type for JSON data. It is stored as binary data of jsonb format.
@@ -277,9 +279,9 @@ impl JsonType {
Self { format }
}
pub(crate) fn new_json2(native: JsonNativeType) -> Self {
pub(crate) fn json2(json_type: Arc<JsonNativeType>) -> Self {
Self {
format: JsonFormat::Json2(Box::new(native)),
format: JsonFormat::Json2(json_type),
}
}
@@ -295,41 +297,7 @@ impl JsonType {
}
pub fn null() -> Self {
Self {
format: JsonFormat::Json2(Box::new(JsonNativeType::Null)),
}
}
pub(crate) fn as_struct_type(&self) -> StructType {
match &self.format {
JsonFormat::Json2(native_type) => match native_type.as_arrow_type() {
// TODO(LFC): Direct use Arrow's Struct datatype here.
ArrowDataType::Struct(fields) => StructType::from(&fields),
// FIXME(fys): Since writing with a non-object root is currently
// not supported, this temporarily returns default.
_ => StructType::default(),
},
JsonFormat::Jsonb => StructType::default(),
}
}
/// Try to merge this json type with others, error on datatype conflict.
pub fn merge(&mut self, other: &JsonType) -> Result<()> {
if self == other {
return Ok(());
}
match (&mut self.format, &other.format) {
(JsonFormat::Jsonb, JsonFormat::Jsonb) => Ok(()),
(JsonFormat::Json2(this), JsonFormat::Json2(that)) => {
this.merge(that);
Ok(())
}
_ => MergeJsonDatatypeSnafu {
reason: "json format not match",
}
.fail(),
}
Self::json2(Arc::new(JsonNativeType::Null))
}
/// Check if it includes all fields in `other` json type.
@@ -342,7 +310,7 @@ impl JsonType {
}
}
fn is_include(this: &JsonNativeType, that: &JsonNativeType) -> bool {
pub(crate) fn is_include(this: &JsonNativeType, that: &JsonNativeType) -> bool {
fn is_include_object(this: &JsonObjectType, that: &JsonObjectType) -> bool {
for (type_name, that_type) in that {
let Some(this_type) = this.get(type_name) else {
@@ -366,12 +334,6 @@ fn is_include(this: &JsonNativeType, that: &JsonNativeType) -> bool {
}
}
impl From<&ArrowDataType> for JsonType {
fn from(t: &ArrowDataType) -> Self {
JsonType::new_json2(JsonNativeType::from(&ConcreteDataType::from_arrow_type(t)))
}
}
impl DataType for JsonType {
fn name(&self) -> String {
match &self.format {
@@ -391,9 +353,13 @@ impl DataType for JsonType {
}
fn as_arrow_type(&self) -> ArrowDataType {
match self.format {
match &self.format {
JsonFormat::Jsonb => ArrowDataType::Binary,
JsonFormat::Json2(_) => self.as_struct_type().as_arrow_type(),
JsonFormat::Json2(x) => {
let mut object = JsonNativeType::object();
object.merge(x.as_ref());
object.as_arrow_type()
}
}
}
@@ -752,122 +718,100 @@ mod tests {
}
#[test]
fn test_merge_json_type() -> Result<()> {
fn test(
other: JsonType,
json_type: &mut JsonType,
expected: std::result::Result<&str, &str>,
) -> Result<()> {
let result = json_type.merge(&other);
match (result, expected) {
(Ok(()), Ok(expected)) => {
assert_eq!(json_type.native_type().to_string(), expected);
}
(Err(err), Err(expected)) => {
assert_eq!(err.to_string(), expected);
}
_ => unreachable!(),
}
Ok(())
fn test_merge_json_type() {
fn test(other: JsonNativeType, json_type: &mut JsonNativeType, expected: &str) {
json_type.merge(&other);
assert_eq!(json_type.to_string(), expected);
}
// Null should be absorbed by a concrete scalar type.
test(
JsonType::new_json2(JsonNativeType::Bool),
&mut JsonType::null(),
Ok(r#""<Bool>""#),
)?;
JsonNativeType::Bool,
&mut JsonNativeType::Null,
r#""<Bool>""#,
);
// Merging a null value into an existing concrete type should keep the type unchanged.
test(
JsonType::null(),
&mut JsonType::new_json2(JsonNativeType::Bool),
Ok(r#""<Bool>""#),
)?;
JsonNativeType::Null,
&mut JsonNativeType::Bool,
r#""<Bool>""#,
);
// Identical number categories should stay as Number.
test(
JsonType::new_json2(JsonNativeType::i64()),
&mut JsonType::new_json2(JsonNativeType::i64()),
Ok(r#""<Number>""#),
)?;
JsonNativeType::i64(),
&mut JsonNativeType::i64(),
r#""<Number>""#,
);
// Conflicting number categories should be lifted to Variant.
test(
JsonType::new_json2(JsonNativeType::f64()),
&mut JsonType::new_json2(JsonNativeType::i64()),
Ok(r#""<Number>""#),
)?;
JsonNativeType::f64(),
&mut JsonNativeType::i64(),
r#""<Number>""#,
);
// Object merge should preserve existing fields and append missing fields.
test(
JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
JsonNativeType::Object(JsonObjectType::from([(
"foo".to_string(),
JsonNativeType::String,
)]))),
&mut JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
)])),
&mut JsonNativeType::Object(JsonObjectType::from([(
"bar".to_string(),
JsonNativeType::i64(),
)]))),
Ok(r#"{"bar":"<Number>","foo":"<String>"}"#),
)?;
)])),
r#"{"bar":"<Number>","foo":"<String>"}"#,
);
// Conflicting object field types should only lift that field to Variant.
test(
JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
JsonNativeType::Object(JsonObjectType::from([(
"foo".to_string(),
JsonNativeType::i64(),
)]))),
&mut JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
)])),
&mut JsonNativeType::Object(JsonObjectType::from([(
"foo".to_string(),
JsonNativeType::Bool,
)]))),
Ok(r#"{"foo":"<Variant>"}"#),
)?;
)])),
r#"{"foo":"<Variant>"}"#,
);
// Nested objects should merge recursively.
test(
JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
JsonNativeType::Object(JsonObjectType::from([(
"nested".to_string(),
JsonNativeType::Object(JsonObjectType::from([(
"foo".to_string(),
JsonNativeType::String,
)])),
)]))),
&mut JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
)])),
&mut JsonNativeType::Object(JsonObjectType::from([(
"nested".to_string(),
JsonNativeType::Object(JsonObjectType::from([(
"bar".to_string(),
JsonNativeType::Bool,
)])),
)]))),
Ok(r#"{"nested":{"bar":"<Bool>","foo":"<String>"}}"#),
)?;
)])),
r#"{"nested":{"bar":"<Bool>","foo":"<String>"}}"#,
);
// Arrays should merge their element types recursively.
test(
JsonType::new_json2(JsonNativeType::Array(Box::new(JsonNativeType::String))),
&mut JsonType::new_json2(JsonNativeType::Array(Box::new(JsonNativeType::u64()))),
Ok(r#"["<Variant>"]"#),
)?;
JsonNativeType::Array(Box::new(JsonNativeType::String)),
&mut JsonNativeType::Array(Box::new(JsonNativeType::u64())),
r#"["<Variant>"]"#,
);
// Root-level incompatible types should be lifted to Variant.
test(
JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
JsonNativeType::Object(JsonObjectType::from([(
"foo".to_string(),
JsonNativeType::String,
)]))),
&mut JsonType::new_json2(JsonNativeType::Bool),
Ok(r#""<Variant>""#),
)?;
// Jsonb and Json2 should not be mergeable.
test(
JsonType::new_json2(JsonNativeType::Bool),
&mut JsonType::new(JsonFormat::Jsonb),
Err("Failed to merge JSON datatype: json format not match"),
)?;
Ok(())
)])),
&mut JsonNativeType::Bool,
r#""<Variant>""#,
);
}
}
src/datatypes/src/vectors/json/builder.rs
+202 -30
View File
@@ -13,18 +13,22 @@
// limitations under the License.
use std::any::Any;
use std::sync::Arc;
use arrow_schema::DataType;
use crate::data_type::ConcreteDataType;
use crate::error::{Result, TryFromValueSnafu, UnsupportedOperationSnafu};
use crate::json::value::{JsonValue, JsonVariant};
use crate::error::{Result, TryFromValueSnafu, UnexpectedSnafu, UnsupportedOperationSnafu};
use crate::json::value::{JsonNumber, JsonValue, JsonVariant};
use crate::prelude::{ValueRef, Vector, VectorRef};
use crate::types::JsonType;
use crate::types::json_type::{JsonFormat, JsonNativeType};
use crate::types::StructType;
use crate::types::json_type::{JsonNativeType, is_include};
use crate::value::{ListValueRef, StructValueRef};
use crate::vectors::{MutableVector, StructVectorBuilder};
#[derive(Clone)]
pub(crate) struct JsonVectorBuilder {
merged_type: JsonType,
merged_type: JsonNativeType,
values: Vec<JsonValue>,
}
@@ -35,31 +39,111 @@ impl JsonVectorBuilder {
JsonNativeType::Object(_) | JsonNativeType::Null
));
Self {
merged_type: JsonType::new_json2(initial_native_type),
merged_type: initial_native_type,
values: Vec::with_capacity(capacity),
}
}
fn try_build(&mut self) -> Result<VectorRef> {
let mut builder = StructVectorBuilder::with_type_and_capacity(
self.merged_type.as_struct_type(),
self.values.len(),
);
let DataType::Struct(fields) = self.merged_type.as_arrow_type() else {
return UnexpectedSnafu {
reason: "merged JSON2 type must map to Arrow Struct in JsonVectorBuilder",
}
.fail();
};
// TODO(LFC): Direct use Arrow's Struct datatype here.
let struct_type = StructType::from(&fields);
let mut builder =
StructVectorBuilder::with_type_and_capacity(struct_type.clone(), self.values.len());
for value in self.values.iter_mut() {
value.try_align(&self.merged_type)?;
if value.is_null() {
builder.push_null();
continue;
}
builder.try_push_value_ref(&value.as_ref().as_value_ref())?;
value.try_align(&self.merged_type)?;
let value_ref = json_variant_to_struct_value_ref(value.variant(), struct_type.clone())?;
builder.push_struct_value_ref(value_ref)?;
}
Ok(builder.to_vector())
}
}
fn json_variant_to_struct_value_ref(
value: &JsonVariant,
struct_type: StructType,
) -> Result<StructValueRef<'_>> {
let JsonVariant::Object(object) = value else {
return TryFromValueSnafu {
reason: format!("expected json object value, got {value:?}"),
}
.fail();
};
let values = struct_type
.fields()
.iter()
.map(|field| {
object
.get(field.name())
.map(|v| json_variant_to_value_ref(v, field.data_type()))
.unwrap_or(Ok(ValueRef::Null))
})
.collect::<Result<Vec<_>>>()?;
Ok(StructValueRef::RefList {
val: values,
fields: struct_type,
})
}
fn json_variant_to_value_ref<'a>(
value: &'a JsonVariant,
expected_type: &ConcreteDataType,
) -> Result<ValueRef<'a>> {
let value = match value {
JsonVariant::Null => ValueRef::Null,
JsonVariant::Bool(x) => ValueRef::Boolean(*x),
JsonVariant::Number(x) => match x {
JsonNumber::PosInt(i) => ValueRef::UInt64(*i),
JsonNumber::NegInt(i) => ValueRef::Int64(*i),
JsonNumber::Float(f) => ValueRef::Float64(*f),
},
JsonVariant::String(x) => ValueRef::String(x),
JsonVariant::Array(array) => {
let item_type = match expected_type {
ConcreteDataType::List(list_type) => list_type.item_type().clone(),
_ => ConcreteDataType::null_datatype(),
};
let values = array
.iter()
.map(|v| json_variant_to_value_ref(v, &item_type))
.collect::<Result<Vec<_>>>()?;
ValueRef::List(ListValueRef::RefList {
val: values,
item_datatype: Arc::new(item_type),
})
}
JsonVariant::Object(_) => {
let ConcreteDataType::Struct(struct_type) = expected_type else {
return TryFromValueSnafu {
reason: format!("expected struct type, got {expected_type}"),
}
.fail();
};
ValueRef::Struct(json_variant_to_struct_value_ref(
value,
struct_type.clone(),
)?)
}
JsonVariant::Variant(x) => ValueRef::Binary(x),
};
Ok(value)
}
impl MutableVector for JsonVectorBuilder {
fn data_type(&self) -> ConcreteDataType {
ConcreteDataType::Json(self.merged_type.clone())
ConcreteDataType::json2(self.merged_type.clone())
}
fn len(&self) -> usize {
@@ -90,21 +174,18 @@ impl MutableVector for JsonVectorBuilder {
.fail();
};
let json_type = value.json_type();
if !matches!(
json_type.format,
JsonFormat::Json2(ref native_type)
if matches!(native_type.as_ref(), JsonNativeType::Object(_) | JsonNativeType::Null)
) {
let json_type = json_type.as_ref();
if !matches!(json_type, JsonNativeType::Object(_) | JsonNativeType::Null) {
return TryFromValueSnafu {
reason: format!("expected json object value, got {value:?}"),
}
.fail();
}
if !self.merged_type.is_include(json_type) {
self.merged_type.merge(json_type)?;
if !is_include(&self.merged_type, json_type) {
self.merged_type.merge(json_type);
}
let value = JsonValue::new(JsonVariant::from(value.variant().clone()));
let value = JsonValue::new_with(JsonVariant::from(value.variant()), value.json_type());
self.values.push(value);
Ok(())
}
@@ -124,12 +205,15 @@ impl MutableVector for JsonVectorBuilder {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use common_base::bytes::Bytes;
use super::*;
use crate::data_type::ConcreteDataType;
use crate::types::StructField;
use crate::types::json_type::JsonObjectType;
use crate::value::{StructValue, Value, ValueRef};
use crate::value::{ListValue, StructValue, Value, ValueRef};
#[test]
fn test_json_vector_builder() -> Result<()> {
@@ -151,17 +235,20 @@ mod tests {
builder.push_null();
builder.try_push_value_ref(&second.as_value_ref())?;
let merged_type = JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([
let merged_type = JsonNativeType::Object(JsonObjectType::from([
("extra".to_string(), JsonNativeType::Bool),
("id".to_string(), JsonNativeType::i64()),
("payload".to_string(), JsonNativeType::Variant),
])));
]));
assert_eq!(
builder.data_type(),
ConcreteDataType::Json(merged_type.clone())
ConcreteDataType::json2(merged_type.clone())
);
let merged_struct_type = merged_type.as_struct_type();
let DataType::Struct(fields) = merged_type.as_arrow_type() else {
unreachable!()
};
let merged_struct_type = StructType::from(&fields);
let vector = builder.to_vector();
assert_eq!(vector.len(), 3);
assert_eq!(
@@ -195,16 +282,19 @@ mod tests {
inferred_builder.push_null();
inferred_builder.try_push_value_ref(&inferred_value.as_value_ref())?;
let inferred_type = JsonType::new_json2(JsonNativeType::Object(JsonObjectType::from([(
let inferred_type = JsonNativeType::Object(JsonObjectType::from([(
"id".to_string(),
JsonNativeType::i64(),
)])));
)]));
assert_eq!(
inferred_builder.data_type(),
ConcreteDataType::Json(inferred_type.clone())
ConcreteDataType::json2(inferred_type.clone())
);
let inferred_struct_type = inferred_type.as_struct_type();
let DataType::Struct(fields) = inferred_type.as_arrow_type() else {
unreachable!()
};
let inferred_struct_type = StructType::from(&fields);
let vector = inferred_builder.to_vector();
assert_eq!(vector.get(0), Value::Null);
assert_eq!(
@@ -240,4 +330,86 @@ mod tests {
Ok(())
}
#[test]
fn test_json_variant_to_struct_value_ref() -> Result<()> {
let item_type =
ConcreteDataType::struct_datatype(StructType::new(Arc::new(vec![StructField::new(
"id".to_string(),
ConcreteDataType::int64_datatype(),
true,
)])));
let struct_type = StructType::new(Arc::new(vec![
StructField::new(
"items".to_string(),
ConcreteDataType::list_datatype(Arc::new(item_type.clone())),
true,
),
StructField::new(
"meta".to_string(),
ConcreteDataType::struct_datatype(StructType::new(Arc::new(vec![
StructField::new(
"name".to_string(),
ConcreteDataType::string_datatype(),
true,
),
]))),
true,
),
]));
let variant = JsonVariant::from([
(
"items",
JsonVariant::Array(vec![
JsonVariant::from([("id", JsonVariant::from(1i64))]),
JsonVariant::from([("id", JsonVariant::from(2i64))]),
]),
),
(
"meta",
JsonVariant::from([("name", JsonVariant::from("foo"))]),
),
]);
let value_ref = json_variant_to_struct_value_ref(&variant, struct_type.clone())?;
let value = value_ref.to_value();
assert_eq!(
value,
Value::Struct(StructValue::new(
vec![
Value::List(ListValue::new(
vec![
Value::Struct(StructValue::new(
vec![Value::Int64(1)],
StructType::new(Arc::new(vec![StructField::new(
"id".to_string(),
ConcreteDataType::int64_datatype(),
true,
)]))
)),
Value::Struct(StructValue::new(
vec![Value::Int64(2)],
StructType::new(Arc::new(vec![StructField::new(
"id".to_string(),
ConcreteDataType::int64_datatype(),
true,
)]))
)),
],
Arc::new(item_type),
)),
Value::Struct(StructValue::new(
vec![Value::String("foo".into())],
StructType::new(Arc::new(vec![StructField::new(
"name".to_string(),
ConcreteDataType::string_datatype(),
true,
)])),
)),
],
struct_type,
))
);
Ok(())
}
}
src/datatypes/src/vectors/struct_vector.rs
+64 -12
View File
@@ -317,6 +317,37 @@ impl StructVectorBuilder {
Ok(())
}
pub(crate) fn push_struct_value_ref(&mut self, struct_value: StructValueRef<'_>) -> Result<()> {
match struct_value {
StructValueRef::Indexed { vector, idx } => match vector.get(idx).as_struct()? {
Some(struct_value) => self.push_struct_value(struct_value)?,
None => self.push_null_struct_value(),
},
StructValueRef::Ref(value) => self.push_struct_value(value)?,
StructValueRef::RefList { val, fields } => {
ensure!(
val.len() == self.value_builders.len(),
InconsistentStructFieldsAndItemsSnafu {
field_len: self.value_builders.len(),
item_len: val.len(),
}
);
ensure!(
fields.fields().len() == self.value_builders.len(),
InconsistentStructFieldsAndItemsSnafu {
field_len: self.value_builders.len(),
item_len: fields.fields().len(),
}
);
for (builder, value) in self.value_builders.iter_mut().zip(val) {
builder.try_push_value_ref(&value)?;
}
self.null_buffer.append_non_null();
}
}
Ok(())
}
fn push_null_struct_value(&mut self) {
for builder in &mut self.value_builders {
builder.push_null();
@@ -352,18 +383,7 @@ impl MutableVector for StructVectorBuilder {
fn try_push_value_ref(&mut self, value: &ValueRef) -> Result<()> {
if let Some(struct_ref) = value.try_into_struct()? {
match struct_ref {
StructValueRef::Indexed { vector, idx } => match vector.get(idx).as_struct()? {
Some(struct_value) => self.push_struct_value(struct_value)?,
None => self.push_null(),
},
StructValueRef::Ref(val) => self.push_struct_value(val)?,
StructValueRef::RefList { val, fields } => {
let struct_value =
StructValue::try_new(val.into_iter().map(Value::from).collect(), fields)?;
self.push_struct_value(&struct_value)?;
}
}
self.push_struct_value_ref(struct_ref)?;
} else {
self.push_null();
}
@@ -490,6 +510,38 @@ mod tests {
}
}
#[test]
fn test_struct_vector_builder_push_ref_list() {
let struct_type = StructType::new(Arc::new(vec![
StructField::new("id".to_string(), ConcreteDataType::int64_datatype(), true),
StructField::new(
"name".to_string(),
ConcreteDataType::string_datatype(),
true,
),
]));
let mut builder = StructVectorBuilder::with_type_and_capacity(struct_type.clone(), 2);
builder
.push_struct_value_ref(StructValueRef::RefList {
val: vec![ValueRef::Int64(1), ValueRef::String("foo")],
fields: struct_type.clone(),
})
.unwrap();
builder.push_null();
let vector = builder.finish();
assert_eq!(vector.len(), 2);
assert_eq!(vector.null_count(), 1);
assert_eq!(
vector.get(0),
Value::Struct(StructValue::new(
vec![Value::Int64(1), Value::String("foo".into())],
struct_type,
))
);
assert_eq!(vector.get(1), Value::Null);
}
#[test]
fn test_deep_nested_struct_list() {
// level 1: struct
src/mito2/src/memtable/bulk/json_align.rs
+10 -7
View File
@@ -17,9 +17,8 @@ use std::sync::Arc;
use datatypes::arrow::datatypes::{DataType as ArrowDataType, Schema, SchemaRef};
use datatypes::arrow::record_batch::RecordBatch;
use datatypes::data_type::DataType;
use datatypes::extension::json::is_structured_json_field;
use datatypes::types::JsonType;
use datatypes::types::json_type::JsonNativeType;
use datatypes::vectors::json::array::JsonArray;
use snafu::{OptionExt, ResultExt};
@@ -57,13 +56,17 @@ impl Json2Aligner {
})?;
// Init merged types from base schema.
let mut merged_types: HashMap<usize, JsonType> = base_schema
let mut merged_types = base_schema
.fields()
.iter()
.enumerate()
.filter(|&(_idx, field)| is_structured_json_field(field))
.map(|(idx, field)| (idx, JsonType::from(field.data_type())))
.collect();
.map(|(idx, field)| {
let json_type =
JsonNativeType::try_from(field.data_type()).context(DataTypeMismatchSnafu)?;
Ok((idx, json_type))
})
.collect::<Result<HashMap<usize, JsonNativeType>>>()?;
// No JSON2 columns, no alignment needed.
if merged_types.is_empty() {
@@ -83,9 +86,9 @@ impl Json2Aligner {
if *idx >= schema.fields().len() {
continue;
}
merged
.merge(&JsonType::from(schema.field(*idx).data_type()))
let json_type = JsonNativeType::try_from(schema.field(*idx).data_type())
.context(DataTypeMismatchSnafu)?;
merged.merge(&json_type);
}
}
src/pipeline/src/etl/transform/transformer/greptime.rs
+8 -1
View File
@@ -692,6 +692,7 @@ fn resolve_value(
VrlValue::Array(_) | VrlValue::Object(_) => {
let is_json2 = schema_info
.find_column_schema_in_table(&column_name)
// TODO(LFC): Default to JSON2 for auto-created tables.
.is_some_and(|x| {
matches!(
&x.column_schema.data_type,
@@ -714,7 +715,13 @@ fn resolve_value(
})?;
let value = settings.encode(value)?;
resolve_schema(index, p_ctx, &column_name, &value.data_type(), schema_info)?;
resolve_schema(
index,
p_ctx,
&column_name,
&ConcreteDataType::json2(Default::default()),
schema_info,
)?;
let Value::Json(value) = value else {
unreachable!()
src/sql/src/statements.rs
+1 -1
View File
@@ -294,7 +294,7 @@ pub fn sql_data_type_to_concrete_data_type(data_type: &SqlDataType) -> Result<Co
JSON2_TYPE_NAME if args.is_empty() => {
// Currently, JSON2 is not inferred as any native type initially.
// TODO(fys): infer it later from type hints.
let format = JsonFormat::Json2(Box::new(JsonNativeType::Null));
let format = JsonFormat::Json2(Arc::new(JsonNativeType::Null));
Ok(ConcreteDataType::Json(JsonType::new(format)))
}
_ => error::SqlTypeNotSupportedSnafu {