From 3a904f332f34f7e86ed2f2be2f8894ee66b97c8f Mon Sep 17 00:00:00 2001 From: fys <40801205+fengys1996@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:08:53 +0800 Subject: [PATCH] fix(json2): standardize widening and projection cast semantics (#8661) * fix(json2): standardize widening and projection cast semantics Signed-off-by: fys * fix: cargo clippy Signed-off-by: fys * fix: typos Signed-off-by: fys * fix: unit test Signed-off-by: fys * feat: add fast-path Signed-off-by: fys * remove unsed code Signed-off-by: fys * refactor(json2): project nested json_get paths with JsonArray Signed-off-by: fys * fix(query): reject projecting whole JSON2 columns Signed-off-by: fys * fix(query): reject whole-column JSON2 reads Signed-off-by: fys * test(json2): organize limitation sqlness cases Signed-off-by: fys * fix: sqlness test Signed-off-by: fys --------- Signed-off-by: fys --- src/api/src/helper.rs | 2 +- .../function/src/scalars/json/json_get.rs | 197 ++------ src/common/recordbatch/src/recordbatch.rs | 43 +- src/datatypes/src/types/json_type.rs | 19 +- src/datatypes/src/types/primitive_type.rs | 21 +- src/datatypes/src/vectors/json/array.rs | 434 ++++++++++++++---- src/datatypes/src/vectors/json/builder.rs | 108 +---- src/mito2/src/memtable/bulk/json_align.rs | 50 +- src/mito2/src/read/compat.rs | 2 +- src/mito2/src/read/flat_projection.rs | 2 +- .../src/sst/parquet/json_align/stream.rs | 2 +- .../src/optimizer/json_type_concretize.rs | 275 ++++++++++- .../standalone/common/types/json/json2.result | 105 +---- .../standalone/common/types/json/json2.sql | 65 --- .../common/types/json/json2_cast.result | 268 +++++++++++ .../common/types/json/json2_cast.sql | 108 +++++ .../common/types/json/json2_limit.result | 174 +++++++ .../common/types/json/json2_limit.sql | 110 +++++ .../common/types/json/json2_type_hints.result | 24 + .../common/types/json/json2_type_hints.sql | 18 + 20 files changed, 1455 insertions(+), 572 deletions(-) create mode 100644 tests/cases/standalone/common/types/json/json2_cast.result create mode 100644 tests/cases/standalone/common/types/json/json2_cast.sql create mode 100644 tests/cases/standalone/common/types/json/json2_limit.result create mode 100644 tests/cases/standalone/common/types/json/json2_limit.sql diff --git a/src/api/src/helper.rs b/src/api/src/helper.rs index 1ff437f127..0a8c999461 100644 --- a/src/api/src/helper.rs +++ b/src/api/src/helper.rs @@ -1900,7 +1900,7 @@ mod tests { })), }; assert_eq!( - JsonNativeType::Array(Box::new(JsonNativeType::f64())), + JsonNativeType::Array(Box::new(JsonNativeType::Variant)), decode_json_value_parts(&proto).1 ); diff --git a/src/common/function/src/scalars/json/json_get.rs b/src/common/function/src/scalars/json/json_get.rs index d081ad917b..15e8542bfe 100644 --- a/src/common/function/src/scalars/json/json_get.rs +++ b/src/common/function/src/scalars/json/json_get.rs @@ -16,7 +16,6 @@ use std::sync::Arc; use arrow::array::{ArrayRef, BinaryViewArray, new_null_array}; use arrow::compute; -use arrow::datatypes::Float64Type; use arrow_schema::Field; use datafusion_common::arrow::array::{ Array, AsArray, BinaryViewBuilder, BooleanBuilder, Float64Builder, Int64Builder, @@ -25,10 +24,8 @@ use datafusion_common::arrow::array::{ use datafusion_common::arrow::datatypes::DataType; use datafusion_common::{DataFusionError, Result, ScalarValue, exec_datafusion_err, exec_err}; use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, Signature, Volatility}; -use datatypes::arrow_array::{int_array_value_at_index, string_array_value_at_index}; use datatypes::vectors::json::array::JsonArray; use derive_more::Display; -use serde_json::Value; use crate::function::{Function, extract_args}; use crate::helper; @@ -48,15 +45,8 @@ fn get_json_by_path(json: &[u8], path: &str) -> Option> { } } -enum JsonResultValue<'a> { - Jsonb(Vec), - #[expect(unused)] - JsonStructByColumn(&'a ArrayRef, usize), - JsonStructByValue(&'a Value), -} - trait JsonGetResultBuilder { - fn append_value(&mut self, value: JsonResultValue<'_>) -> Result<()>; + fn append_value(&mut self, value: &[u8]) -> Result<()>; fn append_null(&mut self); @@ -83,25 +73,8 @@ fn result_builder(len: usize, with_type: &DataType) -> Result) -> Result<()> { - match value { - JsonResultValue::Jsonb(value) => self.0.append_option(jsonb::to_str(&value).ok()), - JsonResultValue::JsonStructByColumn(column, i) => { - if let Some(v) = string_array_value_at_index(column, i) { - self.0.append_value(v); - } else { - self.0 - .append_value(arrow_cast::display::array_value_to_string(column, i)?); - } - } - JsonResultValue::JsonStructByValue(value) => { - if let Some(s) = value.as_str() { - self.0.append_value(s) - } else { - self.0.append_value(value.to_string()) - } - } - } + fn append_value(&mut self, value: &[u8]) -> Result<()> { + self.0.append_option(jsonb::to_str(value).ok()); Ok(()) } @@ -145,14 +118,8 @@ impl Function for JsonGetString { struct IntResultBuilder(Int64Builder); impl JsonGetResultBuilder for IntResultBuilder { - fn append_value(&mut self, value: JsonResultValue<'_>) -> Result<()> { - match value { - JsonResultValue::Jsonb(value) => self.0.append_option(jsonb::to_i64(&value).ok()), - JsonResultValue::JsonStructByColumn(column, i) => { - self.0.append_option(int_array_value_at_index(column, i)) - } - JsonResultValue::JsonStructByValue(value) => self.0.append_option(value.as_i64()), - } + fn append_value(&mut self, value: &[u8]) -> Result<()> { + self.0.append_option(jsonb::to_i64(value).ok()); Ok(()) } @@ -196,22 +163,8 @@ impl Function for JsonGetInt { struct FloatResultBuilder(Float64Builder); impl JsonGetResultBuilder for FloatResultBuilder { - fn append_value(&mut self, value: JsonResultValue<'_>) -> Result<()> { - match value { - JsonResultValue::Jsonb(value) => self.0.append_option(jsonb::to_f64(&value).ok()), - JsonResultValue::JsonStructByColumn(column, i) => { - let result = if column.data_type() == &DataType::Float64 { - column - .as_primitive::() - .is_valid(i) - .then(|| column.as_primitive::().value(i)) - } else { - None - }; - self.0.append_option(result); - } - JsonResultValue::JsonStructByValue(value) => self.0.append_option(value.as_f64()), - } + fn append_value(&mut self, value: &[u8]) -> Result<()> { + self.0.append_option(jsonb::to_f64(value).ok()); Ok(()) } @@ -255,22 +208,8 @@ impl Function for JsonGetFloat { struct BoolResultBuilder(BooleanBuilder); impl JsonGetResultBuilder for BoolResultBuilder { - fn append_value(&mut self, value: JsonResultValue<'_>) -> Result<()> { - match value { - JsonResultValue::Jsonb(value) => self.0.append_option(jsonb::to_bool(&value).ok()), - JsonResultValue::JsonStructByColumn(column, i) => { - let result = if column.data_type() == &DataType::Boolean { - column - .as_boolean() - .is_valid(i) - .then(|| column.as_boolean().value(i)) - } else { - None - }; - self.0.append_option(result); - } - JsonResultValue::JsonStructByValue(value) => self.0.append_option(value.as_bool()), - } + fn append_value(&mut self, value: &[u8]) -> Result<()> { + self.0.append_option(jsonb::to_bool(value).ok()); Ok(()) } @@ -324,7 +263,7 @@ fn jsonb_get( _ => None, }; if let Some(v) = result { - builder.append_value(JsonResultValue::Jsonb(v))?; + builder.append_value(&v)?; } else { builder.append_null(); } @@ -333,89 +272,49 @@ fn jsonb_get( } fn json_struct_get(array: &ArrayRef, path: &str, with_type: &DataType) -> Result { - let path = path.trim_start_matches("$"); + let segments = path + .trim_start_matches("$") + .split('.') + .filter(|segment| !segment.is_empty()) + .collect::>(); + let mut curr = array.clone(); - // Fast path: if the JSON array fields can be directly indexed into by the `path`, simply get - // the sub-array (`column_by_name`). - let mut direct = true; - let mut current = array; - for segment in path.split(".").filter(|s| !s.is_empty()) { - if matches!(current.data_type(), DataType::Binary) { - direct = false; - break; + for (idx, segment) in segments.iter().enumerate() { + if curr.data_type().is_binary() { + let target = nested_projection_type(&segments[idx..], with_type); + curr = JsonArray::from(&curr) + .project_to(&target) + .map_err(|e| exec_datafusion_err!("{e}"))?; } - let Some(json) = current.as_struct_opt() else { - return exec_err!("unknown JSON array datatype: {}", current.data_type()); + let Some(json) = curr.as_struct_opt() else { + return exec_err!("unknown JSON array datatype: {}", curr.data_type()); }; let Some(sub_json) = json.column_by_name(segment) else { return Ok(new_null_array(with_type, array.len())); }; - current = sub_json; + curr = sub_json.clone(); } - // Build the result array with optional value mapper. - fn build_with(input: &ArrayRef, with_type: &DataType, value_mapper: F) -> Result - where - for<'a> F: Fn(&'a Value) -> Option<&'a Value>, - { - let json_array = JsonArray::from(input); - - let mut builder = result_builder(input.len(), with_type)?; - for i in 0..input.len() { - if input.is_null(i) { - builder.append_null(); - continue; - } - - let value = json_array - .try_get_value(i) - .map_err(|e| exec_datafusion_err!("{e}"))?; - let value = value_mapper(&value); - - if let Some(value) = value { - builder.append_value(JsonResultValue::JsonStructByValue(value))?; - } else { - builder.append_null(); - } - } - Ok(builder.build()) + if curr.data_type() == with_type { + Ok(curr) + } else { + JsonArray::from(&curr) + .project_to(with_type) + .map_err(|e| exec_datafusion_err!("{e}")) } +} - if direct { - let casted = if current.data_type() != with_type { - match (current.data_type(), with_type) { - (DataType::Binary, _) => { - // Fall back to the slow path if the found JSON sub-array is serialized to bytes - // (because of JSON type conflicting) - build_with(current, with_type, |v| Some(v))? - } - (DataType::List(_) | DataType::Struct(_), with_type) if with_type.is_string() => { - // Special handle for wanted array is string (Arrow cast is not working here if - // the datatype is list or struct), because it could be used in displaying the - // result. - build_with(current, with_type, |v| Some(v))? - } - (_, with_type) if with_type.is_string() => { - // Same special handle for wanted array is string as above, except for simply - // casting by Arrow is more desirable. - arrow_cast::cast(current.as_ref(), with_type)? - } - _ => new_null_array(with_type, current.len()), - } - } else { - current.clone() - }; - return Ok(casted); - } - - // Slow path: reconstruct the JSON array from serialized representation of conflicting JSON - // values: `serde_json::Value`. - let mut pointer = path.replace(".", "/"); - if !pointer.starts_with("/") { - pointer = format!("/{}", pointer); - } - build_with(array, with_type, |value| value.pointer(&pointer)) +/// Builds a nested struct type for projecting the remaining JSON path. +/// +/// For example, path `["a", "b"]` with an `Int64` leaf produces +/// `Struct>`. +fn nested_projection_type(path: &[&str], leaf_type: &DataType) -> DataType { + path.iter() + .rev() + .fold(leaf_type.clone(), |data_type, name| { + DataType::Struct(vec![Arc::new(Field::new(*name, data_type, true))].into()) + }) } /// This function is mostly called as `json_get(value, 'attr')::type` and rewritten by @@ -681,8 +580,8 @@ mod tests { ("$.c", None), ("$.kind", None), ("$.payload.code", Some(404)), - ("$.payload.success", None), - ("$.payload.result.time_cost", None), + ("$.payload.success", Some(0)), + ("$.payload.result.time_cost", Some(1)), ("$.payload.not-exists", None), ("$.not-exists", None), ("$", None), @@ -749,8 +648,8 @@ mod tests { ("$.a", Some(4.4)), ("$.c", None), ("$.kind", None), - ("$.payload.code", None), - ("$.payload.success", None), + ("$.payload.code", Some(404.0)), + ("$.payload.success", Some(0.0)), ("$.payload.result.time_cost", Some(1.234)), ("$.payload.not-exists", None), ("$.not-exists", None), @@ -818,9 +717,9 @@ mod tests { ("$.a", Some(false)), ("$.c", None), ("$.kind", None), - ("$.payload.code", None), + ("$.payload.code", Some(true)), ("$.payload.success", Some(false)), - ("$.payload.result.time_cost", None), + ("$.payload.result.time_cost", Some(true)), ("$.payload.not-exists", None), ("$.not-exists", None), ("$", None), diff --git a/src/common/recordbatch/src/recordbatch.rs b/src/common/recordbatch/src/recordbatch.rs index 77a36836d6..6257aabf82 100644 --- a/src/common/recordbatch/src/recordbatch.rs +++ b/src/common/recordbatch/src/recordbatch.rs @@ -21,10 +21,8 @@ use datafusion_common::arrow::array::ArrayRef; use datafusion_common::arrow::compute; use datafusion_common::arrow::datatypes::{DataType as ArrowDataType, SchemaRef as ArrowSchemaRef}; use datatypes::arrow::array::{Array, AsArray, RecordBatchOptions}; -use datatypes::extension::json::is_structured_json_field; use datatypes::prelude::DataType; use datatypes::schema::SchemaRef; -use datatypes::vectors::json::array::JsonArray; use datatypes::vectors::{Helper, VectorRef}; use serde::ser::{Error, SerializeStruct}; use serde::{Serialize, Serializer}; @@ -61,8 +59,6 @@ impl RecordBatch { // TODO(LFC): Remove the casting here once `Batch` is no longer used. let arrow_arrays = Self::cast_view_arrays(schema.arrow_schema(), arrow_arrays)?; - let arrow_arrays = maybe_align_json_array_with_schema(schema.arrow_schema(), arrow_arrays)?; - let df_record_batch = DfRecordBatch::try_new(schema.arrow_schema().clone(), arrow_arrays) .context(error::NewDfRecordBatchSnafu)?; @@ -88,8 +84,6 @@ impl RecordBatch { // TODO(LFC): Remove the casting here once `Batch` is no longer used. let arrow_arrays = Self::cast_view_arrays(&arrow_schema, arrow_arrays)?; - let arrow_arrays = maybe_align_json_array_with_schema(&arrow_schema, arrow_arrays)?; - let df_record_batch = DfRecordBatch::try_new(arrow_schema, arrow_arrays) .context(error::NewDfRecordBatchSnafu)?; @@ -377,36 +371,11 @@ pub fn merge_record_batches(schema: SchemaRef, batches: &[RecordBatch]) -> Resul Ok(RecordBatch::from_df_record_batch(schema, record_batch)) } -fn maybe_align_json_array_with_schema( - schema: &ArrowSchemaRef, - arrays: Vec, -) -> Result> { - if schema.fields().iter().all(|f| !is_structured_json_field(f)) { - return Ok(arrays); - } - - let mut aligned = Vec::with_capacity(arrays.len()); - for (field, array) in schema.fields().iter().zip(arrays) { - if !is_structured_json_field(field) { - aligned.push(array); - continue; - } - - let json_array = JsonArray::from(&array) - .try_align(field.data_type()) - .context(DataTypesSnafu)?; - aligned.push(json_array); - } - Ok(aligned) -} - #[cfg(test)] mod tests { use std::sync::Arc; - use datatypes::arrow::array::{ - AsArray, BinaryArray, StringArray, StringViewArray, UInt32Array, - }; + use datatypes::arrow::array::{AsArray, StringArray, StringViewArray, UInt32Array}; use datatypes::arrow::datatypes::{DataType, Field, Schema as ArrowSchema, UInt32Type}; use datatypes::data_type::ConcreteDataType; use datatypes::extension::json::{JsonExtensionType, JsonMetadata}; @@ -629,10 +598,10 @@ mod tests { .with_extension_type(JsonExtensionType::new(Arc::new(JsonMetadata::default()))); let arrow_schema = Arc::new(ArrowSchema::new(vec![field])); let schema = Arc::new(Schema::try_from(arrow_schema).unwrap()); - let arrays = - vec![Arc::new(BinaryArray::from(vec![Some(br#"{"a":1}"#.as_slice())])) as ArrayRef]; - - let aligned = maybe_align_json_array_with_schema(schema.arrow_schema(), arrays).unwrap(); - assert_eq!(aligned[0].data_type(), &DataType::Binary); + let columns: Vec = vec![Arc::new(BinaryVector::from(vec![Some( + br#"{"a":1}"#.to_vec(), + )]))]; + let batch = RecordBatch::new(schema, columns).unwrap(); + assert_eq!(batch.column(0).data_type(), &DataType::Binary); } } diff --git a/src/datatypes/src/types/json_type.rs b/src/datatypes/src/types/json_type.rs index 378c7b3d6a..4bab0da52f 100644 --- a/src/datatypes/src/types/json_type.rs +++ b/src/datatypes/src/types/json_type.rs @@ -116,13 +116,6 @@ impl JsonNativeType { (this, JsonNativeType::Null) => this, (this, that) if this == *that => this, - (JsonNativeType::Number(x), JsonNativeType::Number(y)) => { - JsonNativeType::Number(match (x, y) { - (x, y) if x == *y => x, - (JsonNumberType::F64, _) | (_, JsonNumberType::F64) => JsonNumberType::F64, - _ => JsonNumberType::I64, - }) - } _ => JsonNativeType::Variant, }; } @@ -746,11 +739,13 @@ mod tests { ); // Conflicting number categories should be lifted to Variant. - test( - JsonNativeType::f64(), - &mut JsonNativeType::i64(), - r#""""#, - ); + for (mut this, other) in [ + (JsonNativeType::u64(), JsonNativeType::i64()), + (JsonNativeType::u64(), JsonNativeType::f64()), + (JsonNativeType::i64(), JsonNativeType::f64()), + ] { + test(other, &mut this, r#""""#); + } // Object merge should preserve existing fields and append missing fields. test( diff --git a/src/datatypes/src/types/primitive_type.rs b/src/datatypes/src/types/primitive_type.rs index b861b4c158..8bf9190a9c 100644 --- a/src/datatypes/src/types/primitive_type.rs +++ b/src/datatypes/src/types/primitive_type.rs @@ -384,11 +384,11 @@ impl DataType for Int32Type { Value::Int8(v) => num::cast::cast(v).map(Value::Int32), Value::Int16(v) => num::cast::cast(v).map(Value::Int32), Value::Int32(v) => Some(Value::Int32(v)), - Value::Int64(v) => num::cast::cast(v).map(Value::Int64), + Value::Int64(v) => num::cast::cast(v).map(Value::Int32), Value::UInt8(v) => num::cast::cast(v).map(Value::Int32), Value::UInt16(v) => num::cast::cast(v).map(Value::Int32), - Value::UInt32(v) => num::cast::cast(v).map(Value::UInt32), - Value::UInt64(v) => num::cast::cast(v).map(Value::UInt64), + Value::UInt32(v) => num::cast::cast(v).map(Value::Int32), + Value::UInt64(v) => num::cast::cast(v).map(Value::Int32), Value::Float32(v) => num::cast::cast(v).map(Value::Int32), Value::Float64(v) => num::cast::cast(v).map(Value::Int32), Value::String(v) => v.as_utf8().parse::().map(Value::Int32).ok(), @@ -522,6 +522,21 @@ mod tests { ConcreteDataType::int64_datatype(), Value::Int64(12345) ); + assert_primitive_cast!( + Value::Int64(12345), + ConcreteDataType::int32_datatype(), + Value::Int32(12345) + ); + assert_primitive_cast!( + Value::UInt32(12345), + ConcreteDataType::int32_datatype(), + Value::Int32(12345) + ); + assert_primitive_cast!( + Value::UInt64(12345), + ConcreteDataType::int32_datatype(), + Value::Int32(12345) + ); } #[test] diff --git a/src/datatypes/src/vectors/json/array.rs b/src/datatypes/src/vectors/json/array.rs index 3b52064299..afe9f55305 100644 --- a/src/datatypes/src/vectors/json/array.rs +++ b/src/datatypes/src/vectors/json/array.rs @@ -15,27 +15,23 @@ use std::cmp::Ordering; use std::sync::Arc; -use arrow::compute; -use arrow::util::display::{ArrayFormatter, FormatOptions}; +use arrow::compute::{can_cast_types, cast}; use arrow_array::cast::AsArray; use arrow_array::types::{Float64Type, Int64Type, UInt64Type}; use arrow_array::{Array, ArrayRef, GenericListArray, ListArray, StructArray, new_null_array}; use arrow_schema::{DataType, FieldRef}; +use common_telemetry::trace; use serde_json::Value; use snafu::{OptionExt, ResultExt}; -use crate::arrow_array::{ - MutableBinaryArray, StringViewArray, binary_array_value, string_array_value, -}; +use crate::arrow_array::{MutableBinaryArray, binary_array_value, string_array_value}; use crate::data_type::ConcreteDataType; use crate::error::{ AlignJsonArraySnafu, ArrowComputeSnafu, InvalidJsonSnafu, InvalidJsonbSnafu, Result, }; -use crate::json::value::{JsonVariant, decode_json_variant, encode_serde_json_as_jsonb}; -use crate::prelude::DataType as _; -use crate::vectors::json::builder::{ - json_variant_into_projected_value, null_on_json_type_mismatch, -}; +use crate::json::value::{decode_json_variant, encode_serde_json_as_jsonb}; +use crate::prelude::{DataType as _, Value as GreptimeValue}; +use crate::value::{ListValue, StructValue}; pub struct JsonArray<'a> { inner: &'a ArrayRef, @@ -96,27 +92,28 @@ impl JsonArray<'_> { Ok(value) } - /// Align a JSON array to the `expect` data type. The alignment mostly does three things: + /// Normalizes a JSON2 array to the wider `expect` data type without losing + /// information. /// - /// 1. set the missing fields with null arrays; - /// 2. discard the fields that are not in the `expect` data type; - /// 3. cast the fields to the ones with same names in the `expect` if their data types are not - /// matched. - pub fn try_align(&self, expect: &DataType) -> Result { - if self.inner.data_type() == expect { + /// This is mainly used for write/flush-time JSON2 schema alignment: + /// - fields missing from the source are filled with typed null arrays; + /// - fields present in the source must also exist in `expect`; + /// - fields present in both are widened recursively when their types differ. + /// + /// Narrowing conversions and any other conversions that may lose information + /// are rejected. + pub fn widen_to(&self, expect: &DataType) -> Result { + let data_type = self.inner.data_type(); + + if data_type == expect { return Ok(self.inner.clone()); } - common_telemetry::trace!( + trace!( "Try aligning JSON array {} to data type {}", - self.inner.data_type(), - expect + data_type, expect ); - if self.inner.data_type().is_binary() && matches!(expect, DataType::Struct(_)) { - return self.decode_variant(expect); - } - let struct_array = self.inner.as_struct_opt().context(AlignJsonArraySnafu { reason: "expect struct array", })?; @@ -151,13 +148,13 @@ impl JsonArray<'_> { let array_type = array_field.data_type(); let array = match (expect_type, array_type) { (DataType::Struct(_), DataType::Struct(_)) => { - JsonArray::from(&array_columns[j]).try_align(expect_type)? + JsonArray::from(&array_columns[j]).widen_to(expect_type)? } (DataType::List(expect_item), DataType::List(array_item)) => { let list_array = array_columns[j].as_list::(); - try_align_list(list_array, expect_item, array_item)? + widen_list(list_array, array_item, expect_item)? } - _ => JsonArray::from(&array_columns[j]).try_cast(expect_type)?, + _ => JsonArray::from(&array_columns[j]).widen_scalar_to(expect_type)?, }; aligned.push(array); } @@ -169,20 +166,36 @@ impl JsonArray<'_> { i += 1; } Ordering::Greater => { - j += 1; + return AlignJsonArraySnafu { + reason: format!( + "source field {} does not exist in target schema", + array_field.name() + ), + } + .fail(); } } } + if j < array_fields.len() { + return AlignJsonArraySnafu { + reason: format!( + "source field {} does not exist in target schema", + array_fields[j].name() + ), + } + .fail(); + } if i < expect_fields.len() { for field in &expect_fields[i..] { aligned.push(new_null_array(field.data_type(), struct_array.len())); } } - let json_array = StructArray::try_new( + let json_array = StructArray::try_new_with_length( expect_fields.clone(), aligned, struct_array.nulls().cloned(), + struct_array.len(), ) .map_err(|e| { AlignJsonArraySnafu { @@ -193,52 +206,32 @@ impl JsonArray<'_> { Ok(Arc::new(json_array)) } - fn try_cast(&self, to_type: &DataType) -> Result { + /// Widens an array to the merged JSON2 physical type without losing information. + /// + /// Supported conversions: + /// - identical types are returned unchanged; + /// - null arrays become typed null arrays; + /// - concrete JSON values are encoded as JSONB when the target type is binary. + /// + /// All other conversions are rejected. + fn widen_scalar_to(&self, to_type: &DataType) -> Result { let from_type = self.inner.data_type(); if from_type == to_type { return Ok(self.inner.clone()); } - if to_type == &DataType::Utf8View { - let values = (0..self.inner.len()) - .map(|i| { - if self.inner.is_null(i) { - return Ok(None); - } - let value = match self.try_get_value(i)? { - Value::Null => return Ok(None), - Value::String(value) => value, - value => value.to_string(), - }; - Ok(Some(value)) - }) - .collect::>>()?; - return Ok(Arc::new(StringViewArray::from(values)) as ArrayRef); - } - - if from_type.is_binary() && !to_type.is_binary() { - return self.decode_variant(to_type); + if from_type == &DataType::Null { + return Ok(new_null_array(to_type, self.inner.len())); } if !from_type.is_binary() && to_type.is_binary() { return self.encode_variant(); } - if compute::can_cast_types(from_type, to_type) { - return compute::cast(&self.inner, to_type).context(ArrowComputeSnafu); + AlignJsonArraySnafu { + reason: format!("unable to widen {from_type} to {to_type}"), } - - let formatter = ArrayFormatter::try_new(&self.inner, &FormatOptions::default()) - .context(ArrowComputeSnafu)?; - - let values = (0..self.inner.len()) - .map(|i| { - self.inner - .is_valid(i) - .then(|| formatter.value(i).to_string()) - }) - .collect::>(); - Ok(Arc::new(StringViewArray::from(values))) + .fail() } fn encode_variant(&self) -> Result { @@ -264,45 +257,184 @@ impl JsonArray<'_> { Ok(Arc::new(builder.finish())) } - fn decode_variant(&self, to_type: &DataType) -> Result { + /// Projects this JSON array to `target` for query evaluation. + /// + /// Unlike [`Self::widen_to`], projection tolerates lossy conversions: + /// - source fields not present in `target` are discarded; + /// - fields missing from the source are filled with typed null arrays; + /// - values incompatible with the target type become NULL. + /// + /// Projection is applied recursively to structs and lists. Input nulls + /// remain NULL. Errors unrelated to type incompatibility, such as invalid + /// JSONB, are returned. + pub fn project_to(&self, target: &DataType) -> Result { + if self.inner.data_type() == target { + return Ok(self.inner.clone()); + } + + match (self.inner.data_type(), target) { + (DataType::Struct(_), DataType::Struct(target_fields)) => { + let struct_array = self.inner.as_struct(); + let mut columns = Vec::with_capacity(target_fields.len()); + for target_field in target_fields { + let column = struct_array + .column_by_name(target_field.name()) + .map(|column| JsonArray::from(column).project_to(target_field.data_type())) + .transpose()? + .unwrap_or_else(|| { + new_null_array(target_field.data_type(), self.inner.len()) + }); + columns.push(column); + } + let projected = StructArray::try_new_with_length( + target_fields.clone(), + columns, + struct_array.nulls().cloned(), + struct_array.len(), + ) + .context(ArrowComputeSnafu)?; + Ok(Arc::new(projected)) + } + (DataType::List(_), DataType::List(target_item)) => { + let list_array = self.inner.as_list::(); + let item_projected = + JsonArray::from(list_array.values()).project_to(target_item.data_type())?; + Ok(Arc::new( + GenericListArray::::try_new( + target_item.clone(), + list_array.offsets().clone(), + item_projected, + list_array.nulls().cloned(), + ) + .context(ArrowComputeSnafu)?, + )) + } + _ => self.project_values_to(target), + } + } + + fn project_values_to(&self, to_type: &DataType) -> Result { + let from_type = self.inner.data_type(); + if can_fast_cast_types(from_type, to_type) { + return cast(self.inner.as_ref(), to_type).context(ArrowComputeSnafu); + } + let values = (0..self.inner.len()) .map(|i| self.try_get_value(i)) .collect::>>()?; - decode_json_values(values, to_type) + project_json_values(values, to_type) } } -fn decode_json_values(values: Vec, to_type: &DataType) -> Result { +/// Returns whether Arrow can cast between the types without JSON-aware projection. +/// Binary and nested types require JSONB decoding or recursive projection. +fn can_fast_cast_types(from_type: &DataType, to_type: &DataType) -> bool { + let is_scalar = |data_type: &DataType| { + data_type.is_numeric() || data_type.is_string() || data_type == &DataType::Boolean + }; + + is_scalar(from_type) && is_scalar(to_type) && can_cast_types(from_type, to_type) +} + +fn project_json_values(values: Vec, to_type: &DataType) -> Result { let concrete_type = ConcreteDataType::from_arrow_type(to_type); let mut builder = concrete_type.create_mutable_vector(values.len()); for value in values { - let value = null_on_json_type_mismatch(json_variant_into_projected_value( - JsonVariant::from(value), - &concrete_type, - ))?; + let value = project_json_value_to_type(value, &concrete_type)?; builder.try_push_value_ref(&value.as_value_ref())?; } Ok(builder.to_vector().to_arrow_array()) } -fn try_align_list( - list_array: &ListArray, - expect_item: &FieldRef, - array_item: &FieldRef, -) -> Result { - let item_aligned = match (expect_item.data_type(), array_item.data_type()) { +fn project_json_value_to_type(value: Value, to_type: &ConcreteDataType) -> Result { + if value.is_null() { + return Ok(GreptimeValue::Null); + } + + if to_type.is_string() { + let value = match value { + Value::String(value) => value, + value => value.to_string(), + }; + return Ok(GreptimeValue::String(value.into())); + } + + if matches!(to_type, ConcreteDataType::Binary(_)) { + return Ok(GreptimeValue::Binary( + encode_serde_json_as_jsonb(value).into(), + )); + } + + if let Some(struct_type) = to_type.as_struct() { + let Value::Object(mut object) = value else { + return Ok(GreptimeValue::Null); + }; + let values = struct_type + .fields() + .iter() + .map(|field| { + object + .remove(field.name()) + .map(|value| project_json_value_to_type(value, field.data_type())) + .transpose() + .map(|value| value.unwrap_or(GreptimeValue::Null)) + }) + .collect::>>()?; + return Ok(GreptimeValue::Struct(StructValue::new( + values, + struct_type.clone(), + ))); + } + + if let Some(list_type) = to_type.as_list() { + let Value::Array(values) = value else { + return Ok(GreptimeValue::Null); + }; + let item_type = list_type.item_type().clone(); + let values = values + .into_iter() + .map(|value| project_json_value_to_type(value, &item_type)) + .collect::>>()?; + return Ok(GreptimeValue::List(ListValue::new( + values, + Arc::new(item_type), + ))); + } + + let value = match value { + Value::Bool(value) => GreptimeValue::Boolean(value), + Value::Number(value) => { + if let Some(value) = value.as_i64() { + GreptimeValue::Int64(value) + } else if let Some(value) = value.as_u64() { + GreptimeValue::UInt64(value) + } else if let Some(value) = value.as_f64() { + GreptimeValue::Float64(value.into()) + } else { + GreptimeValue::Null + } + } + Value::String(value) => GreptimeValue::String(value.into()), + Value::Array(_) | Value::Object(_) => GreptimeValue::Null, + Value::Null => GreptimeValue::Null, + }; + Ok(to_type.try_cast(value).unwrap_or(GreptimeValue::Null)) +} + +fn widen_list(list_array: &ListArray, actual: &FieldRef, expected: &FieldRef) -> Result { + let item_aligned = match (actual.data_type(), expected.data_type()) { (DataType::Struct(_), DataType::Struct(_)) => { - JsonArray::from(list_array.values()).try_align(expect_item.data_type())? + JsonArray::from(list_array.values()).widen_to(expected.data_type())? } - (DataType::List(expect_item), DataType::List(array_item)) => { + (DataType::List(actual), DataType::List(expected)) => { let list_array = list_array.values().as_list::(); - try_align_list(list_array, expect_item, array_item)? + widen_list(list_array, actual, expected)? } - _ => JsonArray::from(list_array.values()).try_cast(expect_item.data_type())?, + _ => JsonArray::from(list_array.values()).widen_scalar_to(expected.data_type())?, }; Ok(Arc::new( GenericListArray::::try_new( - expect_item.clone(), + expected.clone(), list_array.offsets().clone(), item_aligned, list_array.nulls().cloned(), @@ -322,6 +454,7 @@ mod test { use arrow_array::types::Int64Type; use arrow_array::{ BinaryArray, BooleanArray, Float64Array, Int32Array, Int64Array, ListArray, StringArray, + UInt64Array, }; use arrow_schema::{Field, Fields}; use serde_json::json; @@ -421,7 +554,7 @@ mod test { None, ])); - let casted = JsonArray::from(&variants).try_cast(&DataType::Utf8View)?; + let casted = JsonArray::from(&variants).project_to(&DataType::Utf8View)?; let casted = casted.as_string_view(); assert!(casted.is_null(0)); assert_eq!(casted.value(1), r#"{"value":1}"#); @@ -431,6 +564,109 @@ mod test { Ok(()) } + #[test] + fn test_project_plain_scalars() -> Result<()> { + let integers: ArrayRef = Arc::new(Int64Array::from(vec![Some(42), Some(i64::MAX), None])); + let projected = JsonArray::from(&integers).project_to(&DataType::Int32)?; + let expected: ArrayRef = Arc::new(Int32Array::from(vec![Some(42), None, None])); + assert_eq!(&expected, &projected); + + let booleans: ArrayRef = Arc::new(BooleanArray::from(vec![Some(true), Some(false), None])); + let projected = JsonArray::from(&booleans).project_to(&DataType::Float64)?; + let expected: ArrayRef = Arc::new(Float64Array::from(vec![Some(1.0), Some(0.0), None])); + assert_eq!(&expected, &projected); + + let strings: ArrayRef = Arc::new(StringArray::from(vec![Some("42"), Some("bad"), None])); + let projected = JsonArray::from(&strings).project_to(&DataType::UInt64)?; + let expected: ArrayRef = Arc::new(UInt64Array::from(vec![Some(42), None, None])); + assert_eq!(&expected, &projected); + + Ok(()) + } + + #[test] + fn test_widen_null_to_any_type() -> Result<()> { + let nulls = new_null_array(&DataType::Null, 2); + let target_types = [ + DataType::Boolean, + DataType::UInt64, + DataType::Utf8View, + DataType::Binary, + DataType::List(Arc::new(Field::new_list_field(DataType::Int64, true))), + DataType::Struct(Fields::from(vec![Field::new( + "value", + DataType::Int64, + true, + )])), + ]; + + for target_type in target_types { + let widened = JsonArray::from(&nulls).widen_scalar_to(&target_type)?; + assert_eq!(&target_type, widened.data_type()); + assert_eq!(2, widened.len()); + assert_eq!(2, widened.null_count()); + } + + Ok(()) + } + + #[test] + fn test_widen_non_null_to_utf8_view_fails() { + let bools: ArrayRef = Arc::new(BooleanArray::from(vec![true])); + let err = JsonArray::from(&bools) + .widen_scalar_to(&DataType::Utf8View) + .unwrap_err(); + + assert_eq!( + "Failed to align JSON array, reason: unable to widen Boolean to Utf8View", + err.to_string() + ); + } + + #[test] + fn test_widen_variant_to_non_binary_fails() { + let value = jsonb::parse_value(b"true").unwrap().to_vec(); + let variants: ArrayRef = Arc::new(BinaryArray::from(vec![value.as_slice()])); + let err = JsonArray::from(&variants) + .widen_scalar_to(&DataType::Boolean) + .unwrap_err(); + + assert_eq!( + "Failed to align JSON array, reason: unable to widen Binary to Boolean", + err.to_string() + ); + } + + #[test] + fn test_widen_between_number_types_fails() { + let values: ArrayRef = Arc::new(UInt64Array::from(vec![1])); + let err = JsonArray::from(&values) + .widen_scalar_to(&DataType::Int64) + .unwrap_err(); + + assert_eq!( + "Failed to align JSON array, reason: unable to widen UInt64 to Int64", + err.to_string() + ); + } + + #[test] + fn test_widen_numbers_to_variant_preserves_values() -> Result<()> { + let cases: [(ArrayRef, Value); 3] = [ + (Arc::new(UInt64Array::from(vec![u64::MAX])), json!(u64::MAX)), + (Arc::new(Int64Array::from(vec![i64::MIN])), json!(i64::MIN)), + (Arc::new(Float64Array::from(vec![1.25])), json!(1.25)), + ]; + + for (values, expected) in cases { + let widened = JsonArray::from(&values).widen_scalar_to(&DataType::Binary)?; + assert_eq!(&DataType::Binary, widened.data_type()); + assert_eq!(expected, JsonArray::from(&widened).try_get_value(0)?); + } + + Ok(()) + } + #[test] fn test_align_json_array() -> Result<()> { struct TestCase { @@ -454,7 +690,7 @@ mod test { } fn test(self) -> Result<()> { - let result = JsonArray::from(&self.json_array).try_align(&self.schema_type); + let result = JsonArray::from(&self.json_array).widen_to(&self.schema_type); match (result, self.expected) { (Ok(json_array), Ok(expected)) => assert_eq!(&json_array, &expected), (Ok(json_array), Err(e)) => { @@ -573,6 +809,34 @@ mod test { ) .test()?; + // Source fields that do not exist in the target schema must not be discarded. + TestCase::new( + StructArray::from(vec![( + Arc::new(Field::new("a", DataType::Boolean, true)), + Arc::new(BooleanArray::from(vec![true])) as ArrayRef, + )]), + Fields::from(vec![Field::new("b", DataType::Boolean, true)]), + Err( + "Failed to align JSON array, reason: source field a does not exist in target schema" + .to_string(), + ), + ) + .test()?; + + // Trailing source fields must also be rejected after all target fields are processed. + TestCase::new( + StructArray::from(vec![( + Arc::new(Field::new("b", DataType::Boolean, true)), + Arc::new(BooleanArray::from(vec![true])) as ArrayRef, + )]), + Fields::from(vec![Field::new("a", DataType::Boolean, true)]), + Err( + "Failed to align JSON array, reason: source field b does not exist in target schema" + .to_string(), + ), + ) + .test()?; + Ok(()) } @@ -599,7 +863,7 @@ mod test { true, )])); - let aligned = JsonArray::from(&variants).try_align(&expected_type)?; + let aligned = JsonArray::from(&variants).project_to(&expected_type)?; assert_eq!(&expected_type, aligned.data_type()); assert_eq!( json!({ @@ -638,7 +902,7 @@ mod test { true, )])); - let aligned = JsonArray::from(&input).try_align(&expected_type)?; + let aligned = JsonArray::from(&input).project_to(&expected_type)?; assert_eq!(&expected_type, aligned.data_type()); assert_eq!( json!({"nested": {"flag": true, "value": 42}}), diff --git a/src/datatypes/src/vectors/json/builder.rs b/src/datatypes/src/vectors/json/builder.rs index 2b9d15f002..e6ceb6d6a5 100644 --- a/src/datatypes/src/vectors/json/builder.rs +++ b/src/datatypes/src/vectors/json/builder.rs @@ -18,7 +18,7 @@ use std::sync::Arc; use arrow_schema::DataType; use crate::data_type::ConcreteDataType; -use crate::error::{Error, Result, TryFromValueSnafu, UnexpectedSnafu, UnsupportedOperationSnafu}; +use crate::error::{Result, TryFromValueSnafu, UnexpectedSnafu, UnsupportedOperationSnafu}; use crate::json::value::{JsonNumber, JsonVariant, encode_json_variant}; use crate::prelude::{ValueRef, Vector, VectorRef}; use crate::types::StructType; @@ -171,65 +171,6 @@ fn json_variant_into_value(value: JsonVariant, expected_type: &ConcreteDataType) Ok(value) } -/// Projects a JSON value to `expected_type`, discarding object fields that are not expected. -/// -/// This is used when aligning a dynamically typed JSON value to a projected Arrow schema. -pub(crate) fn json_variant_into_projected_value( - value: JsonVariant, - expected_type: &ConcreteDataType, -) -> Result { - match (value, expected_type) { - (JsonVariant::Null, _) => Ok(Value::Null), - (JsonVariant::String(value), ConcreteDataType::String(_)) => { - Ok(Value::String(value.into())) - } - (value, ConcreteDataType::String(_)) => Ok(Value::String(value.to_string().into())), - (JsonVariant::Object(mut object), ConcreteDataType::Struct(struct_type)) => { - let values = struct_type - .fields() - .iter() - .map(|field| { - object - .remove(field.name()) - .map(|value| { - // A dynamically typed JSON field may not match the corresponding field - // type in the projected Arrow schema. Treat only the mismatched field as - // SQL NULL so the enclosing value remains readable. Errors other than - // type mismatches are preserved. - null_on_json_type_mismatch(json_variant_into_projected_value( - value, - field.data_type(), - )) - }) - .transpose() - .map(|value| value.unwrap_or(Value::Null)) - }) - .collect::>>()?; - Ok(Value::Struct(StructValue::new(values, struct_type.clone()))) - } - (JsonVariant::Array(array), ConcreteDataType::List(list_type)) => { - let item_type = list_type.item_type().clone(); - let values = array - .into_iter() - .map(|value| { - // Apply the same mismatch-to-NULL rule to each list item. - null_on_json_type_mismatch(json_variant_into_projected_value(value, &item_type)) - }) - .collect::>>()?; - Ok(Value::List(ListValue::new(values, Arc::new(item_type)))) - } - (value, expected_type) => json_variant_into_value(value, expected_type), - } -} - -pub(crate) fn null_on_json_type_mismatch(result: Result) -> Result { - match result { - Ok(value) => Ok(value), - Err(Error::TryFromValue { .. }) => Ok(Value::Null), - Err(error) => Err(error), - } -} - impl MutableVector for JsonVectorBuilder { fn data_type(&self) -> ConcreteDataType { ConcreteDataType::json2(self.merged_type.clone()) @@ -510,51 +451,4 @@ mod tests { ); Ok(()) } - - #[test] - fn test_projected_value_nulls_nested_type_mismatches() { - let struct_type = ConcreteDataType::struct_datatype(StructType::new(Arc::new(vec![ - StructField::new( - "value".to_string(), - ConcreteDataType::uint64_datatype(), - true, - ), - StructField::new( - "items".to_string(), - ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::uint64_datatype())), - true, - ), - StructField::new( - "payload".to_string(), - ConcreteDataType::binary_datatype(), - true, - ), - ]))); - - let invalid_field = JsonVariant::from([("value", JsonVariant::from("invalid"))]); - let Value::Struct(value) = json_variant_into_projected_value(invalid_field, &struct_type) - .expect("type mismatch should become null") - else { - unreachable!() - }; - assert_eq!(value.items()[0], Value::Null); - - let invalid_item = JsonVariant::from([( - "items", - JsonVariant::Array(vec![JsonVariant::from("invalid")]), - )]); - let Value::Struct(value) = json_variant_into_projected_value(invalid_item, &struct_type) - .expect("type mismatch should become null") - else { - unreachable!() - }; - let Value::List(items) = &value.items()[1] else { - unreachable!() - }; - assert_eq!(items.items()[0], Value::Null); - - let invalid_json = JsonVariant::from([("payload", JsonVariant::from(f64::NAN))]); - let error = json_variant_into_projected_value(invalid_json, &struct_type).unwrap_err(); - assert!(matches!(error, Error::InvalidJson { .. })); - } } diff --git a/src/mito2/src/memtable/bulk/json_align.rs b/src/mito2/src/memtable/bulk/json_align.rs index 4ce9bb6d14..030de4cb5b 100644 --- a/src/mito2/src/memtable/bulk/json_align.rs +++ b/src/mito2/src/memtable/bulk/json_align.rs @@ -136,7 +136,7 @@ impl Json2Aligner { for (idx, expected_type) in &self.json_columns { if batch.schema_ref().field(*idx).data_type() != expected_type { cols[*idx] = JsonArray::from(batch.column(*idx)) - .try_align(expected_type) + .widen_to(expected_type) .context(ConvertValueSnafu)?; } } @@ -189,10 +189,11 @@ mod tests { use std::sync::Arc; use datatypes::arrow::array::{ - Array, ArrayRef, Int64Array, StringViewArray, StructArray, UInt64Array, + Array, ArrayRef, AsArray, Int64Array, StringViewArray, StructArray, UInt64Array, }; use datatypes::arrow::datatypes::{DataType, Field, Fields, Schema}; use datatypes::extension::json::{JsonExtensionType, JsonMetadata}; + use serde_json::json; use super::*; @@ -338,6 +339,51 @@ mod tests { assert_eq!("bob", name_values.value(1)); } + #[test] + fn test_align_conflicting_number_types_as_variant() { + let u64_fields = Fields::from(vec![Arc::new(Field::new("value", DataType::UInt64, true))]); + let i64_fields = Fields::from(vec![Arc::new(Field::new("value", DataType::Int64, true))]); + let u64_schema = schema_with_json_field(json_field("data", u64_fields.clone())); + let i64_schema = schema_with_json_field(json_field("data", i64_fields.clone())); + let u64_batch = RecordBatch::try_new( + u64_schema.clone(), + vec![ + Arc::new(Int64Array::from_iter_values([1])) as ArrayRef, + struct_array( + u64_fields, + vec![Arc::new(UInt64Array::from_iter_values([u64::MAX])) as ArrayRef], + ), + ], + ) + .unwrap(); + let i64_batch = RecordBatch::try_new( + i64_schema.clone(), + vec![ + Arc::new(Int64Array::from_iter_values([2])) as ArrayRef, + struct_array( + i64_fields, + vec![Arc::new(Int64Array::from_iter_values([i64::MIN])) as ArrayRef], + ), + ], + ) + .unwrap(); + + let aligner = Json2Aligner::try_new([u64_schema, i64_schema]).unwrap(); + let DataType::Struct(fields) = aligner.schema().field(1).data_type() else { + panic!("expected JSON2 field to be a struct"); + }; + assert_eq!(&DataType::Binary, fields[0].data_type()); + + for (batch, expected) in [(u64_batch, json!(u64::MAX)), (i64_batch, json!(i64::MIN))] { + let aligned = aligner.align_batch(batch).unwrap(); + let data = aligned.column(1).as_struct(); + assert_eq!( + expected, + JsonArray::from(data.column(0)).try_get_value(0).unwrap() + ); + } + } + #[test] fn test_wrap_iter_aligns_each_batch() { let id_fields = Fields::from(vec![id_field()]); diff --git a/src/mito2/src/read/compat.rs b/src/mito2/src/read/compat.rs index 3f58b3bd43..561a34a52e 100644 --- a/src/mito2/src/read/compat.rs +++ b/src/mito2/src/read/compat.rs @@ -274,7 +274,7 @@ impl FlatCompatBatch { && json_type.is_json2() { JsonArray::from(old_column) - .try_align(&json_type.as_arrow_type()) + .project_to(&json_type.as_arrow_type()) .context(ConvertValueSnafu)? } else { datatypes::arrow::compute::cast(old_column, &ty.as_arrow_type()) diff --git a/src/mito2/src/read/flat_projection.rs b/src/mito2/src/read/flat_projection.rs index 2990f4670f..df674d0389 100644 --- a/src/mito2/src/read/flat_projection.rs +++ b/src/mito2/src/read/flat_projection.rs @@ -367,7 +367,7 @@ impl FlatProjectionMapper { let field = &self.output_schema.arrow_schema().fields()[output_idx]; if is_structured_json_field(field) { array = JsonArray::from(&array) - .try_align(field.data_type()) + .project_to(field.data_type()) .context(DataTypesSnafu)?; } diff --git a/src/mito2/src/sst/parquet/json_align/stream.rs b/src/mito2/src/sst/parquet/json_align/stream.rs index 8eb2b10499..62d3c67ec0 100644 --- a/src/mito2/src/sst/parquet/json_align/stream.rs +++ b/src/mito2/src/sst/parquet/json_align/stream.rs @@ -175,7 +175,7 @@ fn align_array(array: &ArrayRef, field: &FieldRef) -> Result { if is_structured_json_field(field) { return JsonArray::from(array) - .try_align(field.data_type()) + .project_to(field.data_type()) .context(DataTypeMismatchSnafu); } diff --git a/src/query/src/optimizer/json_type_concretize.rs b/src/query/src/optimizer/json_type_concretize.rs index 9c7e764efe..809c85adaa 100644 --- a/src/query/src/optimizer/json_type_concretize.rs +++ b/src/query/src/optimizer/json_type_concretize.rs @@ -18,9 +18,11 @@ use arrow_schema::DataType; use common_function::scalars::json::json_get::JsonGetWithType; use datafusion::datasource::DefaultTableSource; use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; -use datafusion_common::{Result, plan_datafusion_err, plan_err}; +use datafusion_common::{ExprSchema, Result, plan_datafusion_err, plan_err}; +use datafusion_expr::utils::merge_schema; use datafusion_expr::{Expr, LogicalPlan}; use datafusion_optimizer::{OptimizerConfig, OptimizerRule}; +use datatypes::extension::json::is_structured_json_field; use datatypes::types::json_type::{JsonNativeType, JsonObjectType}; use crate::dummy_catalog::DummyTableProvider; @@ -41,6 +43,8 @@ impl OptimizerRule for JsonTypeConcretizeRule { plan: LogicalPlan, _config: &dyn OptimizerConfig, ) -> Result> { + ensure_no_whole_json2_read(&plan)?; + let json_types = deduce_json_types(&plan)?; if json_types.is_empty() { return Ok(Transformed::no(plan)); @@ -72,6 +76,102 @@ impl OptimizerRule for JsonTypeConcretizeRule { } } +/// Rejects unsupported whole-column JSON2 reads in a logical plan. +fn ensure_no_whole_json2_read(plan: &LogicalPlan) -> Result<()> { + // Reject whole JSON2 columns in the final query output, including `SELECT *`. + for field in plan.schema().fields() { + if is_structured_json_field(field) { + return plan_err!( + "Querying the whole JSON2 column '{}' is currently not supported; use json_get to select its fields", + field.name() + ); + } + } + + // Reject whole JSON2 columns consumed by intermediate expressions, for example: + // `SELECT count(*) FROM (SELECT j FROM t GROUP BY j)`. + plan.apply(|plan| { + let input_schema = merge_schema(&plan.inputs()); + for expr in plan.expressions() { + // A bare column in an intermediate projection is only passed through, not consumed. + if matches!(plan, LogicalPlan::Projection(_)) && is_passthrough_column(&expr) { + continue; + } + + expr.apply(|expr| { + // For JSON2, `json_get` is allowed only with a non-empty path; skip its arguments + // after validation. + if let Expr::ScalarFunction(function) = expr + && function.name().eq_ignore_ascii_case(JsonGetWithType::NAME) + { + let Some(Expr::Column(col)) = function.args.first() else { + return Ok(TreeNodeRecursion::Jump); + }; + let Some(path) = function + .args + .get(1) + .and_then(Expr::as_literal) + .and_then(|value| value.try_as_str()) + .flatten() + else { + return Ok(TreeNodeRecursion::Jump); + }; + let reads_whole_column = path + .trim_start_matches('$') + .split('.') + .all(str::is_empty); + if !reads_whole_column { + return Ok(TreeNodeRecursion::Jump); + } + + let Ok(field) = input_schema + .field_from_column(col) + .or_else(|_| plan.schema().field_from_column(col)) + else { + return Ok(TreeNodeRecursion::Jump); + }; + if is_structured_json_field(field) { + return plan_err!( + "Querying the whole JSON2 column '{}' is currently not supported; use json_get to select its fields", + col.name + ); + } + return Ok(TreeNodeRecursion::Jump); + } + + // Any remaining JSON2 column reference is a whole-column read. + let Expr::Column(col) = expr else { + return Ok(TreeNodeRecursion::Continue); + }; + let Ok(field) = input_schema + .field_from_column(col) + .or_else(|_| plan.schema().field_from_column(col)) + else { + return Ok(TreeNodeRecursion::Continue); + }; + if is_structured_json_field(field) { + return plan_err!( + "Querying the whole JSON2 column '{}' is currently not supported; use json_get to select its fields", + col.name + ); + } + Ok(TreeNodeRecursion::Continue) + })?; + } + Ok(TreeNodeRecursion::Continue) + })?; + + Ok(()) +} + +fn is_passthrough_column(expr: &Expr) -> bool { + match expr { + Expr::Column(_) => true, + Expr::Alias(alias) => is_passthrough_column(&alias.expr), + _ => false, + } +} + fn deduce_json_types(plan: &LogicalPlan) -> Result> { let mut json_types = HashMap::::new(); @@ -148,16 +248,21 @@ fn deduce_json_type(expr: &Expr) -> Result> { mod tests { use std::sync::Arc; + use api::v1::SemanticType; use common_function::scalars::udf::create_udf; use datafusion::datasource::provider_as_source; + use datafusion::functions_aggregate::expr_fn::count; use datafusion_common::{Column, ScalarValue}; use datafusion_expr::expr::ScalarFunction; - use datafusion_expr::{LogicalPlanBuilder, col}; + use datafusion_expr::{LogicalPlanBuilder, col, lit}; use datafusion_optimizer::OptimizerContext; - use store_api::storage::RegionId; + use datatypes::extension::json::{JsonExtensionType, JsonMetadata}; + use datatypes::schema::ColumnSchema; + use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder}; + use store_api::storage::{ConcreteDataType, RegionId}; use super::*; - use crate::optimizer::test_util::mock_table_provider; + use crate::optimizer::test_util::{MetaRegionEngine, mock_table_provider}; fn json_get_expr(base: Expr, path: Expr, with_type: Option) -> Result { let json_get = Arc::new(create_udf(Arc::new(JsonGetWithType::default()))); @@ -183,6 +288,45 @@ mod tests { Ok((provider, plan)) } + fn build_json2_scan() -> Result<(Arc, LogicalPlanBuilder)> { + let region_id = RegionId::new(1024, 2); + let mut builder = RegionMetadataBuilder::new(region_id); + let mut json_column = ColumnSchema::new( + "j", + ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())), + true, + ); + json_column + .with_extension_type(&JsonExtensionType::new(Arc::new(JsonMetadata::default()))) + .unwrap(); + builder + .push_column_metadata(ColumnMetadata { + column_schema: json_column, + semantic_type: SemanticType::Field, + column_id: 1, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 2, + }); + let metadata = Arc::new(builder.build().unwrap()); + let engine = Arc::new(MetaRegionEngine::with_metadata(metadata.clone())); + let provider = Arc::new(DummyTableProvider::new(region_id, engine, metadata)); + let plan = LogicalPlanBuilder::scan("t", provider_as_source(provider.clone()), None)?; + Ok((provider, plan)) + } + + fn build_json2_plan(exprs: Vec) -> Result<(Arc, LogicalPlan)> { + let (provider, plan) = build_json2_scan()?; + let plan = plan.project(exprs)?.build()?; + Ok((provider, plan)) + } + #[test] fn test_json_type_concretize_rule_rewrite() -> Result<()> { let exprs = vec![ @@ -253,6 +397,129 @@ mod tests { Ok(()) } + #[test] + fn test_reject_whole_json2_projection() -> Result<()> { + for (exprs, output_name) in [ + (vec![col("j")], "j"), + (vec![col("j").alias("json"), col("ts")], "json"), + ] { + let (_, plan) = build_json2_plan(exprs)?; + let err = JsonTypeConcretizeRule + .rewrite(plan, &OptimizerContext::default()) + .unwrap_err(); + assert!(err.to_string().contains(&format!( + "Querying the whole JSON2 column '{output_name}' is currently not supported" + ))); + } + Ok(()) + } + + #[test] + fn test_reject_whole_json2_output_without_projection() -> Result<()> { + let (_, plan) = build_json2_scan()?; + let plan = plan.sort(vec![col("ts").sort(true, false)])?.build()?; + + let err = JsonTypeConcretizeRule + .rewrite(plan, &OptimizerContext::default()) + .unwrap_err(); + assert!( + err.to_string() + .contains("Querying the whole JSON2 column 'j' is currently not supported") + ); + Ok(()) + } + + #[test] + fn test_reject_whole_json2_use_in_intermediate_plan() -> Result<()> { + let (_, plan) = build_json2_scan()?; + let plan = plan + .aggregate(vec![col("j")], Vec::::new())? + .aggregate(Vec::::new(), vec![count(lit(1))])? + .build()?; + + let err = JsonTypeConcretizeRule + .rewrite(plan, &OptimizerContext::default()) + .unwrap_err(); + assert!( + err.to_string() + .contains("Querying the whole JSON2 column 'j' is currently not supported") + ); + Ok(()) + } + + #[test] + fn test_allow_json2_path_use_in_intermediate_plan() -> Result<()> { + let json_get = json_get_expr(col("j"), path_expr("a"), Some(DataType::Int64))?; + let (provider, plan) = build_json2_scan()?; + let plan = plan + .aggregate(vec![json_get], Vec::::new())? + .aggregate(Vec::::new(), vec![count(lit(1))])? + .build()?; + + assert!( + JsonTypeConcretizeRule + .rewrite(plan, &OptimizerContext::default())? + .transformed + ); + assert!(provider.scan_request().json_type_hint.contains_key("j")); + Ok(()) + } + + #[test] + fn test_allow_json2_passthrough_for_later_projection() -> Result<()> { + let json_get = json_get_expr(col("j"), path_expr("a"), Some(DataType::Int64))?; + let (provider, plan) = build_json2_scan()?; + let plan = plan + .project(vec![json_get.alias("__common_expr"), col("j")])? + .aggregate(Vec::::new(), vec![count(lit(1))])? + .build()?; + + assert!( + JsonTypeConcretizeRule + .rewrite(plan, &OptimizerContext::default())? + .transformed + ); + assert!(provider.scan_request().json_type_hint.contains_key("j")); + Ok(()) + } + + #[test] + fn test_allow_json2_projection_by_path() -> Result<()> { + let expr = json_get_expr(col("j"), path_expr("a"), Some(DataType::Int64))?; + let (provider, plan) = build_json2_plan(vec![expr])?; + + assert!( + JsonTypeConcretizeRule + .rewrite(plan, &OptimizerContext::default())? + .transformed + ); + assert_eq!( + Some(&JsonNativeType::Object(JsonObjectType::from([( + "a".to_string(), + JsonNativeType::i64(), + )]))), + provider.scan_request().json_type_hint.get("j") + ); + Ok(()) + } + + #[test] + fn test_reject_json2_projection_with_empty_path() -> Result<()> { + for path in ["", "$", ".", "$."] { + let expr = json_get_expr(col("j"), path_expr(path), Some(DataType::Utf8View))?; + let (_, plan) = build_json2_plan(vec![expr])?; + + let err = JsonTypeConcretizeRule + .rewrite(plan, &OptimizerContext::default()) + .unwrap_err(); + assert!( + err.to_string() + .contains("Querying the whole JSON2 column 'j' is currently not supported") + ); + } + Ok(()) + } + #[test] fn test_deduce_json_type_with_non_column_base() -> Result<()> { let expr = json_get_expr( diff --git a/tests/cases/standalone/common/types/json/json2.result b/tests/cases/standalone/common/types/json/json2.result index 56e2f16af1..10b288fc58 100644 --- a/tests/cases/standalone/common/types/json/json2.result +++ b/tests/cases/standalone/common/types/json/json2.result @@ -8,30 +8,6 @@ create table json2_table ( Affected Rows: 0 -insert into json2_table (ts, j) values (101, '[1, 2, 3]'); - -Error: 1001(Unsupported), Non-object json is not supported currently - -insert into json2_table (ts, j) values (102, '"hello"'); - -Error: 1001(Unsupported), Non-object json is not supported currently - -insert into json2_table (ts, j) values (103, '42'); - -Error: 1001(Unsupported), Non-object json is not supported currently - -insert into json2_table (ts, j) values (104, 'true'); - -Error: 1001(Unsupported), Non-object json is not supported currently - -insert into json2_table (ts, j) values (105, 'null'); - -Error: 1001(Unsupported), Non-object json is not supported currently - -insert into json2_table (ts, j) values (1, '{}'); - -Error: 1004(InvalidArguments), Invalid InsertRequest, reason: empty json object is not supported, consider adding a dummy field - insert into json2_table (ts, j) values (1, '{"a": {"b": 1}, "c": "s1", "d": [{"e": {"f": 0.1}}]}'), (2, '{"a": {"b": -2}, "c": "s2", "d": [{"e": {"f": 0.2}}]}'); @@ -155,7 +131,7 @@ select j.a, j.a.x from json2_table order by ts; | {"b":"s7"} | | | {"b":8} | | | {"b":null,"x":true} | true | -| {"b":10,"x":null} | null | +| {"b":10,"x":null} | | +-----------------------------------+-------------------------------------+ select j.c, j.y from json2_table order by ts; @@ -175,14 +151,6 @@ select j.c, j.y from json2_table order by ts; | | false | +-----------------------------------+-----------------------------------+ -select j from json2_table order by ts; - -Error: 3001(EngineExecuteQuery), Failed to align JSON array, reason: Invalid argument error: use StructArray::try_new_with_length or StructArray::new_empty_fields to create a struct array with no fields so that the length can be set correctly - -select * from json2_table order by ts; - -Error: 3001(EngineExecuteQuery), Failed to align JSON array, reason: Invalid argument error: use StructArray::try_new_with_length or StructArray::new_empty_fields to create a struct array with no fields so that the length can be set correctly - select j.a.b + 1 from json2_table order by ts; +------------------------------------------------------------+ @@ -289,74 +257,3 @@ drop table json2_variant_null; Affected Rows: 0 -create table json2_without_append_mode ( - ts timestamp time index, - j json2 -); - -Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' - -create table json2_append_mode_false ( - ts timestamp time index, - j json2 -) with ( - 'append_mode' = 'false' -); - -Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' - -create table json2_alter_non_append ( - ts timestamp time index -); - -Affected Rows: 0 - -alter table json2_alter_non_append add column j json2; - -Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' - -drop table json2_alter_non_append; - -Affected Rows: 0 - -create table json2_default_null_ok ( - ts timestamp time index, - j json2( - a int64 null default null - ) -) with ( - 'append_mode' = 'true' -); - -Affected Rows: 0 - -drop table json2_default_null_ok; - -Affected Rows: 0 - -create table json2_default_null_check ( - ts timestamp time index, - j json2( - a int64 not null default null - ) -); - -Error: 2000(InvalidSyntax), Invalid SQL, error: invalid DEFAULT for JSON2 type hint 'a': Default value should not be null for non null column - -create table json2_set_append_mode_false ( - ts timestamp time index, - j json2 -) with ( - 'append_mode' = 'true' -); - -Affected Rows: 0 - -alter table json2_set_append_mode_false set 'append_mode' = 'false'; - -Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' - -drop table json2_set_append_mode_false; - -Affected Rows: 0 - diff --git a/tests/cases/standalone/common/types/json/json2.sql b/tests/cases/standalone/common/types/json/json2.sql index 39390bf592..d6d8cb4df1 100644 --- a/tests/cases/standalone/common/types/json/json2.sql +++ b/tests/cases/standalone/common/types/json/json2.sql @@ -6,18 +6,6 @@ create table json2_table ( 'sst_format' = 'flat', ); -insert into json2_table (ts, j) values (101, '[1, 2, 3]'); - -insert into json2_table (ts, j) values (102, '"hello"'); - -insert into json2_table (ts, j) values (103, '42'); - -insert into json2_table (ts, j) values (104, 'true'); - -insert into json2_table (ts, j) values (105, 'null'); - -insert into json2_table (ts, j) values (1, '{}'); - insert into json2_table (ts, j) values (1, '{"a": {"b": 1}, "c": "s1", "d": [{"e": {"f": 0.1}}]}'), (2, '{"a": {"b": -2}, "c": "s2", "d": [{"e": {"f": 0.2}}]}'); @@ -58,10 +46,6 @@ select j.a, j.a.x from json2_table order by ts; select j.c, j.y from json2_table order by ts; -select j from json2_table order by ts; - -select * from json2_table order by ts; - select j.a.b + 1 from json2_table order by ts; select abs(j.a.b) from json2_table order by ts; @@ -99,52 +83,3 @@ from json2_variant_null order by ts; drop table json2_variant_null; - -create table json2_without_append_mode ( - ts timestamp time index, - j json2 -); - -create table json2_append_mode_false ( - ts timestamp time index, - j json2 -) with ( - 'append_mode' = 'false' -); - -create table json2_alter_non_append ( - ts timestamp time index -); - -alter table json2_alter_non_append add column j json2; - -drop table json2_alter_non_append; - -create table json2_default_null_ok ( - ts timestamp time index, - j json2( - a int64 null default null - ) -) with ( - 'append_mode' = 'true' -); - -drop table json2_default_null_ok; - -create table json2_default_null_check ( - ts timestamp time index, - j json2( - a int64 not null default null - ) -); - -create table json2_set_append_mode_false ( - ts timestamp time index, - j json2 -) with ( - 'append_mode' = 'true' -); - -alter table json2_set_append_mode_false set 'append_mode' = 'false'; - -drop table json2_set_append_mode_false; diff --git a/tests/cases/standalone/common/types/json/json2_cast.result b/tests/cases/standalone/common/types/json/json2_cast.result new file mode 100644 index 0000000000..541ceb025c --- /dev/null +++ b/tests/cases/standalone/common/types/json/json2_cast.result @@ -0,0 +1,268 @@ +-- Read-time scalar casts. json_get(..., path)::type is rewritten to typed json_get. +-- Keep each value in a separate SST so mixed JSON2 physical layouts are exercised. +CREATE TABLE json2_cast_scalar ( + ts TIMESTAMP TIME INDEX, + j JSON2 +) WITH ( + 'append_mode' = 'true', + 'sst_format' = 'flat' +); + +Affected Rows: 0 + +INSERT INTO json2_cast_scalar VALUES (1, '{"a":42}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (2, '{"a":"42"}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (3, '{"a":"bad"}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (4, '{"a":3.14}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (5, '{"a":true}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (6, '{"a":null}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (7, '{"a":{"b":1}}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (8, '{"a":[1,2]}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_scalar VALUES (9, '{"z":0}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_scalar'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_scalar') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +SELECT + ts, + json_get(j, 'a')::BIGINT AS a_bigint, + json_get(j, 'a')::DOUBLE AS a_double, + json_get(j, 'a')::BOOLEAN AS a_bool, + json_get(j, 'a')::STRING AS a_string +FROM json2_cast_scalar +ORDER BY ts; + ++-------------------------+----------+----------+--------+----------+ +| ts | a_bigint | a_double | a_bool | a_string | ++-------------------------+----------+----------+--------+----------+ +| 1970-01-01T00:00:00.001 | 42 | 42.0 | true | 42 | +| 1970-01-01T00:00:00.002 | 42 | 42.0 | | 42 | +| 1970-01-01T00:00:00.003 | | | | bad | +| 1970-01-01T00:00:00.004 | 3 | 3.14 | true | 3.14 | +| 1970-01-01T00:00:00.005 | 1 | 1.0 | true | true | +| 1970-01-01T00:00:00.006 | | | | | +| 1970-01-01T00:00:00.007 | | | | {"b":1} | +| 1970-01-01T00:00:00.008 | | | | [1,2] | +| 1970-01-01T00:00:00.009 | | | | | ++-------------------------+----------+----------+--------+----------+ + +DROP TABLE json2_cast_scalar; + +Affected Rows: 0 + +-- Read-time parent projection while only child leaves are stored. +CREATE TABLE json2_cast_parent ( + ts TIMESTAMP TIME INDEX, + j JSON2 +) WITH ( + 'append_mode' = 'true', + 'sst_format' = 'flat' +); + +Affected Rows: 0 + +INSERT INTO json2_cast_parent VALUES (1, '{"a":{"b":1,"c":2}}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_parent'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_parent') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_parent VALUES (2, '{"a":{"b":"1","c":"2"}}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_parent'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_parent') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +INSERT INTO json2_cast_parent VALUES (3, '{"a":{"b":true,"c":false}}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_parent'); + ++----------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_parent') | ++----------------------------------------+ +| 0 | ++----------------------------------------+ + +SELECT + ts, + json_get(j, 'a')::STRING AS a_string, + json_get(j, 'a')::INT AS a_int, + json_get(j, 'a.b')::BIGINT AS ab_bigint, + json_get(j, 'a.c')::STRING AS ac_string +FROM json2_cast_parent +ORDER BY ts; + ++-------------------------+----------------------+-------+-----------+-----------+ +| ts | a_string | a_int | ab_bigint | ac_string | ++-------------------------+----------------------+-------+-----------+-----------+ +| 1970-01-01T00:00:00.001 | {"b":1,"c":2} | | 1 | 2 | +| 1970-01-01T00:00:00.002 | {"b":"1","c":"2"} | | 1 | 2 | +| 1970-01-01T00:00:00.003 | {"b":true,"c":false} | | 1 | false | ++-------------------------+----------------------+-------+-----------+-----------+ + +DROP TABLE json2_cast_parent; + +Affected Rows: 0 + +-- Write-time schema alignment across different ingest batches. +CREATE TABLE json2_cast_write_alignment ( + ts TIMESTAMP TIME INDEX, + j JSON2 +) WITH ( + 'append_mode' = 'true', + 'sst_format' = 'flat' +); + +Affected Rows: 0 + +INSERT INTO json2_cast_write_alignment VALUES + (1, '{"s":null,"b":null,"n":null,"o":null,"l":null,"ui":18446744073709551615,"uf":9007199254740993,"if":-9007199254740993}'); + +Affected Rows: 1 + +INSERT INTO json2_cast_write_alignment VALUES + (2, '{"s":"text","b":true,"n":42,"o":{"x":1},"l":[1,2],"ui":-1,"uf":1.5,"if":1.5}'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_cast_write_alignment'); + ++-------------------------------------------------+ +| ADMIN FLUSH_TABLE('json2_cast_write_alignment') | ++-------------------------------------------------+ +| 0 | ++-------------------------------------------------+ + +SELECT + ts, + json_get(j, 's')::STRING AS null_to_string, + json_get(j, 'b')::BOOLEAN AS null_to_bool, + json_get(j, 'n')::UINT64 AS null_to_uint, + json_get(j, 'o')::STRING AS null_to_object, + json_get(j, 'l')::STRING AS null_to_list, + json_get(j, 'ui')::STRING AS uint_int_variant, + json_get(j, 'uf')::STRING AS uint_float_variant, + json_get(j, 'if')::STRING AS int_float_variant +FROM json2_cast_write_alignment +ORDER BY ts; + ++-------------------------+----------------+--------------+--------------+----------------+--------------+----------------------+--------------------+-------------------+ +| ts | null_to_string | null_to_bool | null_to_uint | null_to_object | null_to_list | uint_int_variant | uint_float_variant | int_float_variant | ++-------------------------+----------------+--------------+--------------+----------------+--------------+----------------------+--------------------+-------------------+ +| 1970-01-01T00:00:00.001 | | | | | | 18446744073709551615 | 9007199254740993 | -9007199254740993 | +| 1970-01-01T00:00:00.002 | text | true | 42 | {"x":1} | [1,2] | -1 | 1.5 | 1.5 | ++-------------------------+----------------+--------------+--------------+----------------+--------------+----------------------+--------------------+-------------------+ + +DROP TABLE json2_cast_write_alignment; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/types/json/json2_cast.sql b/tests/cases/standalone/common/types/json/json2_cast.sql new file mode 100644 index 0000000000..2ed133be06 --- /dev/null +++ b/tests/cases/standalone/common/types/json/json2_cast.sql @@ -0,0 +1,108 @@ +-- Read-time scalar casts. json_get(..., path)::type is rewritten to typed json_get. +-- Keep each value in a separate SST so mixed JSON2 physical layouts are exercised. +CREATE TABLE json2_cast_scalar ( + ts TIMESTAMP TIME INDEX, + j JSON2 +) WITH ( + 'append_mode' = 'true', + 'sst_format' = 'flat' +); + +INSERT INTO json2_cast_scalar VALUES (1, '{"a":42}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (2, '{"a":"42"}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (3, '{"a":"bad"}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (4, '{"a":3.14}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (5, '{"a":true}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (6, '{"a":null}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (7, '{"a":{"b":1}}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (8, '{"a":[1,2]}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +INSERT INTO json2_cast_scalar VALUES (9, '{"z":0}'); +ADMIN FLUSH_TABLE('json2_cast_scalar'); + +SELECT + ts, + json_get(j, 'a')::BIGINT AS a_bigint, + json_get(j, 'a')::DOUBLE AS a_double, + json_get(j, 'a')::BOOLEAN AS a_bool, + json_get(j, 'a')::STRING AS a_string +FROM json2_cast_scalar +ORDER BY ts; + +DROP TABLE json2_cast_scalar; + +-- Read-time parent projection while only child leaves are stored. +CREATE TABLE json2_cast_parent ( + ts TIMESTAMP TIME INDEX, + j JSON2 +) WITH ( + 'append_mode' = 'true', + 'sst_format' = 'flat' +); + +INSERT INTO json2_cast_parent VALUES (1, '{"a":{"b":1,"c":2}}'); +ADMIN FLUSH_TABLE('json2_cast_parent'); + +INSERT INTO json2_cast_parent VALUES (2, '{"a":{"b":"1","c":"2"}}'); +ADMIN FLUSH_TABLE('json2_cast_parent'); + +INSERT INTO json2_cast_parent VALUES (3, '{"a":{"b":true,"c":false}}'); +ADMIN FLUSH_TABLE('json2_cast_parent'); + +SELECT + ts, + json_get(j, 'a')::STRING AS a_string, + json_get(j, 'a')::INT AS a_int, + json_get(j, 'a.b')::BIGINT AS ab_bigint, + json_get(j, 'a.c')::STRING AS ac_string +FROM json2_cast_parent +ORDER BY ts; + +DROP TABLE json2_cast_parent; + +-- Write-time schema alignment across different ingest batches. +CREATE TABLE json2_cast_write_alignment ( + ts TIMESTAMP TIME INDEX, + j JSON2 +) WITH ( + 'append_mode' = 'true', + 'sst_format' = 'flat' +); + +INSERT INTO json2_cast_write_alignment VALUES + (1, '{"s":null,"b":null,"n":null,"o":null,"l":null,"ui":18446744073709551615,"uf":9007199254740993,"if":-9007199254740993}'); + +INSERT INTO json2_cast_write_alignment VALUES + (2, '{"s":"text","b":true,"n":42,"o":{"x":1},"l":[1,2],"ui":-1,"uf":1.5,"if":1.5}'); + +ADMIN FLUSH_TABLE('json2_cast_write_alignment'); + +SELECT + ts, + json_get(j, 's')::STRING AS null_to_string, + json_get(j, 'b')::BOOLEAN AS null_to_bool, + json_get(j, 'n')::UINT64 AS null_to_uint, + json_get(j, 'o')::STRING AS null_to_object, + json_get(j, 'l')::STRING AS null_to_list, + json_get(j, 'ui')::STRING AS uint_int_variant, + json_get(j, 'uf')::STRING AS uint_float_variant, + json_get(j, 'if')::STRING AS int_float_variant +FROM json2_cast_write_alignment +ORDER BY ts; + +DROP TABLE json2_cast_write_alignment; diff --git a/tests/cases/standalone/common/types/json/json2_limit.result b/tests/cases/standalone/common/types/json/json2_limit.result new file mode 100644 index 0000000000..9302db37f6 --- /dev/null +++ b/tests/cases/standalone/common/types/json/json2_limit.result @@ -0,0 +1,174 @@ +create table json2_disable_non_object_insert ( + ts timestamp time index, + j json2 +) +with ( + 'append_mode' = 'true' +); + +Affected Rows: 0 + +insert into json2_disable_non_object_insert values (1, '[1, 2, 3]'); + +Error: 1001(Unsupported), Non-object json is not supported currently + +insert into json2_disable_non_object_insert values (2, '"hello"'); + +Error: 1001(Unsupported), Non-object json is not supported currently + +insert into json2_disable_non_object_insert values (3, '42'); + +Error: 1001(Unsupported), Non-object json is not supported currently + +insert into json2_disable_non_object_insert values (4, 'true'); + +Error: 1001(Unsupported), Non-object json is not supported currently + +insert into json2_disable_non_object_insert values (5, 'null'); + +Error: 1001(Unsupported), Non-object json is not supported currently + +insert into json2_disable_non_object_insert values (6, '{}'); + +Error: 1004(InvalidArguments), Invalid InsertRequest, reason: empty json object is not supported, consider adding a dummy field + +drop table json2_disable_non_object_insert; + +Affected Rows: 0 + +create table json2_disable_whole_column_read ( + ts timestamp time index, + j json2 +) +with ( + 'append_mode' = 'true' +); + +Affected Rows: 0 + +insert into json2_disable_whole_column_read values + (1, '{"a": {"b": 1}}'), + (2, '{"a": {"b": 2}}'); + +Affected Rows: 2 + +-- Whole JSON2 uses are unsupported (case 1): direct projection. +select j from json2_disable_whole_column_read order by ts; + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +-- Whole JSON2 uses are unsupported (case 2): wildcard projection. +select * from json2_disable_whole_column_read order by ts; + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +-- Whole JSON2 uses are unsupported (case 3): json_get with an empty path. +select json_get(j, '') from json2_disable_whole_column_read; + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +select json_get(j, '$') from json2_disable_whole_column_read; + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +select json_get(j, '.') from json2_disable_whole_column_read; + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +select json_get(j, '$.') from json2_disable_whole_column_read; + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +-- Whole JSON2 uses are unsupported (case 4): use in an intermediate plan node. +select count(*) +from ( + select j + from json2_disable_whole_column_read + group by j +); + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +-- JSON2 field projection remains supported (case 5): use in an intermediate plan node. +select json_get(j, 'a.b'), count(*) +from json2_disable_whole_column_read +group by json_get(j, 'a.b') +order by json_get(j, 'a.b'); + ++---------------------------------------------------------+----------+ +| json_get(json2_disable_whole_column_read.j,Utf8("a.b")) | count(*) | ++---------------------------------------------------------+----------+ +| 1 | 1 | +| 2 | 1 | ++---------------------------------------------------------+----------+ + +-- Whole JSON2 uses are unsupported (case 6): output after an intermediate projection. +select ts, j +from ( + select ts, j + from json2_disable_whole_column_read +) +order by ts; + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +-- Whole JSON2 uses are unsupported (case 7): DISTINCT in an intermediate plan node. +select count(*) +from ( + select distinct j + from json2_disable_whole_column_read +); + +Error: 3001(EngineExecuteQuery), Error during planning: Querying the whole JSON2 column 'j' is currently not supported; use json_get to select its fields + +drop table json2_disable_whole_column_read; + +Affected Rows: 0 + +create table json2_without_append_mode ( + ts timestamp time index, + j json2 +); + +Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' + +create table json2_append_mode_false ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'false' +); + +Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' + +create table json2_alter_non_append ( + ts timestamp time index +); + +Affected Rows: 0 + +alter table json2_alter_non_append add column j json2; + +Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' + +drop table json2_alter_non_append; + +Affected Rows: 0 + +create table json2_set_append_mode_false ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'true' +); + +Affected Rows: 0 + +alter table json2_set_append_mode_false set 'append_mode' = 'false'; + +Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' + +drop table json2_set_append_mode_false; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/types/json/json2_limit.sql b/tests/cases/standalone/common/types/json/json2_limit.sql new file mode 100644 index 0000000000..d792d335f5 --- /dev/null +++ b/tests/cases/standalone/common/types/json/json2_limit.sql @@ -0,0 +1,110 @@ +create table json2_disable_non_object_insert ( + ts timestamp time index, + j json2 +) +with ( + 'append_mode' = 'true' +); + +insert into json2_disable_non_object_insert values (1, '[1, 2, 3]'); + +insert into json2_disable_non_object_insert values (2, '"hello"'); + +insert into json2_disable_non_object_insert values (3, '42'); + +insert into json2_disable_non_object_insert values (4, 'true'); + +insert into json2_disable_non_object_insert values (5, 'null'); + +insert into json2_disable_non_object_insert values (6, '{}'); + +drop table json2_disable_non_object_insert; + +create table json2_disable_whole_column_read ( + ts timestamp time index, + j json2 +) +with ( + 'append_mode' = 'true' +); + +insert into json2_disable_whole_column_read values + (1, '{"a": {"b": 1}}'), + (2, '{"a": {"b": 2}}'); + +-- Whole JSON2 uses are unsupported (case 1): direct projection. +select j from json2_disable_whole_column_read order by ts; + +-- Whole JSON2 uses are unsupported (case 2): wildcard projection. +select * from json2_disable_whole_column_read order by ts; + +-- Whole JSON2 uses are unsupported (case 3): json_get with an empty path. +select json_get(j, '') from json2_disable_whole_column_read; + +select json_get(j, '$') from json2_disable_whole_column_read; + +select json_get(j, '.') from json2_disable_whole_column_read; + +select json_get(j, '$.') from json2_disable_whole_column_read; + +-- Whole JSON2 uses are unsupported (case 4): use in an intermediate plan node. +select count(*) +from ( + select j + from json2_disable_whole_column_read + group by j +); + +-- JSON2 field projection remains supported (case 5): use in an intermediate plan node. +select json_get(j, 'a.b'), count(*) +from json2_disable_whole_column_read +group by json_get(j, 'a.b') +order by json_get(j, 'a.b'); + +-- Whole JSON2 uses are unsupported (case 6): output after an intermediate projection. +select ts, j +from ( + select ts, j + from json2_disable_whole_column_read +) +order by ts; + +-- Whole JSON2 uses are unsupported (case 7): DISTINCT in an intermediate plan node. +select count(*) +from ( + select distinct j + from json2_disable_whole_column_read +); + +drop table json2_disable_whole_column_read; + +create table json2_without_append_mode ( + ts timestamp time index, + j json2 +); + +create table json2_append_mode_false ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'false' +); + +create table json2_alter_non_append ( + ts timestamp time index +); + +alter table json2_alter_non_append add column j json2; + +drop table json2_alter_non_append; + +create table json2_set_append_mode_false ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'true' +); + +alter table json2_set_append_mode_false set 'append_mode' = 'false'; + +drop table json2_set_append_mode_false; diff --git a/tests/cases/standalone/common/types/json/json2_type_hints.result b/tests/cases/standalone/common/types/json/json2_type_hints.result index c79607f8f9..785fa76942 100644 --- a/tests/cases/standalone/common/types/json/json2_type_hints.result +++ b/tests/cases/standalone/common/types/json/json2_type_hints.result @@ -88,6 +88,30 @@ CREATE TABLE json2_type_hints_timestamp ( Error: 2000(InvalidSyntax), Invalid SQL, error: unsupported JSON2 type hint data type: TIMESTAMP +CREATE TABLE json2_default_null_ok ( + ts TIMESTAMP TIME INDEX, + j JSON2 ( + a BIGINT NULL DEFAULT NULL + ) +) WITH ( + 'append_mode' = 'true' +); + +Affected Rows: 0 + +DROP TABLE json2_default_null_ok; + +Affected Rows: 0 + +CREATE TABLE json2_default_null_check ( + ts TIMESTAMP TIME INDEX, + j JSON2 ( + a BIGINT NOT NULL DEFAULT NULL + ) +); + +Error: 2000(InvalidSyntax), Invalid SQL, error: invalid DEFAULT for JSON2 type hint 'a': Default value should not be null for non null column + -- A type hint at the maximum supported depth is accepted. CREATE TABLE json2_type_hint_depth_50 ( ts TIMESTAMP TIME INDEX, diff --git a/tests/cases/standalone/common/types/json/json2_type_hints.sql b/tests/cases/standalone/common/types/json/json2_type_hints.sql index 9cdd25295b..741684f79c 100644 --- a/tests/cases/standalone/common/types/json/json2_type_hints.sql +++ b/tests/cases/standalone/common/types/json/json2_type_hints.sql @@ -48,6 +48,24 @@ CREATE TABLE json2_type_hints_timestamp ( ) ); +CREATE TABLE json2_default_null_ok ( + ts TIMESTAMP TIME INDEX, + j JSON2 ( + a BIGINT NULL DEFAULT NULL + ) +) WITH ( + 'append_mode' = 'true' +); + +DROP TABLE json2_default_null_ok; + +CREATE TABLE json2_default_null_check ( + ts TIMESTAMP TIME INDEX, + j JSON2 ( + a BIGINT NOT NULL DEFAULT NULL + ) +); + -- A type hint at the maximum supported depth is accepted. CREATE TABLE json2_type_hint_depth_50 ( ts TIMESTAMP TIME INDEX,