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

fix(json2): standardize widening and projection cast semantics (#8661)

Browse Source 
* fix(json2): standardize widening and projection cast semantics

Signed-off-by: fys <fengys1996@gmail.com>

* fix: cargo clippy

Signed-off-by: fys <fengys1996@gmail.com>

* fix: typos

Signed-off-by: fys <fengys1996@gmail.com>

* fix: unit test

Signed-off-by: fys <fengys1996@gmail.com>

* feat: add fast-path

Signed-off-by: fys <fengys1996@gmail.com>

* remove unsed code

Signed-off-by: fys <fengys1996@gmail.com>

* refactor(json2): project nested json_get paths with JsonArray

Signed-off-by: fys <fengys1996@gmail.com>

* fix(query): reject projecting whole JSON2 columns

Signed-off-by: fys <fengys1996@gmail.com>

* fix(query): reject whole-column JSON2 reads

Signed-off-by: fys <fengys1996@gmail.com>

* test(json2): organize limitation sqlness cases

Signed-off-by: fys <fengys1996@gmail.com>

* fix: sqlness test

Signed-off-by: fys <fengys1996@gmail.com>

---------

Signed-off-by: fys <fengys1996@gmail.com>
This commit is contained in:
fys
2026-07-30 12:08:53 +08:00
committed by GitHub
parent 47ca5c362e
commit 3a904f332f
20 changed files with 1455 additions and 572 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/src/helper.rs
+1 -1
View File
@@ -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
);
src/common/function/src/scalars/json/json_get.rs
+48 -149
View File
@@ -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<Vec<u8>> {
}
}
enum JsonResultValue<'a> {
Jsonb(Vec<u8>),
#[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<Box<dyn JsonGetRes
struct StringResultBuilder(StringViewBuilder);
impl JsonGetResultBuilder for StringResultBuilder {
fn append_value(&mut self, value: JsonResultValue<'_>) -> 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::<Float64Type>()
.is_valid(i)
.then(|| column.as_primitive::<Float64Type>().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<ArrayRef> {
let path = path.trim_start_matches("$");
let segments = path
.trim_start_matches("$")
.split('.')
.filter(|segment| !segment.is_empty())
.collect::<Vec<_>>();
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<F>(input: &ArrayRef, with_type: &DataType, value_mapper: F) -> Result<ArrayRef>
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<a: Struct<b: Int64>>`.
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),
src/common/recordbatch/src/recordbatch.rs
+6 -37
View File
@@ -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<ArrayRef>,
) -> Result<Vec<ArrayRef>> {
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<VectorRef> = 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);
}
}
src/datatypes/src/types/json_type.rs
+7 -12
View File
@@ -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#""<Number>""#,
);
for (mut this, other) in [
(JsonNativeType::u64(), JsonNativeType::i64()),
(JsonNativeType::u64(), JsonNativeType::f64()),
(JsonNativeType::i64(), JsonNativeType::f64()),
] {
test(other, &mut this, r#""<Variant>""#);
}
// Object merge should preserve existing fields and append missing fields.
test(
src/datatypes/src/types/primitive_type.rs
+18 -3
View File
@@ -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::<i32>().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]
src/datatypes/src/vectors/json/array.rs
+349 -85
View File
@@ -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<ArrayRef> {
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<ArrayRef> {
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::<i32>();
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<ArrayRef> {
/// 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<ArrayRef> {
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::<Result<Vec<_>>>()?;
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::<Vec<_>>();
Ok(Arc::new(StringViewArray::from(values)))
.fail()
}
fn encode_variant(&self) -> Result<ArrayRef> {
@@ -264,45 +257,184 @@ impl JsonArray<'_> {
Ok(Arc::new(builder.finish()))
}
fn decode_variant(&self, to_type: &DataType) -> Result<ArrayRef> {
/// 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<ArrayRef> {
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::<i32>();
let item_projected =
JsonArray::from(list_array.values()).project_to(target_item.data_type())?;
Ok(Arc::new(
GenericListArray::<i32>::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<ArrayRef> {
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::<Result<Vec<_>>>()?;
decode_json_values(values, to_type)
project_json_values(values, to_type)
}
}
fn decode_json_values(values: Vec<Value>, to_type: &DataType) -> Result<ArrayRef> {
/// 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<Value>, to_type: &DataType) -> Result<ArrayRef> {
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<ArrayRef> {
let item_aligned = match (expect_item.data_type(), array_item.data_type()) {
fn project_json_value_to_type(value: Value, to_type: &ConcreteDataType) -> Result<GreptimeValue> {
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::<Result<Vec<_>>>()?;
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::<Result<Vec<_>>>()?;
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<ArrayRef> {
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::<i32>();
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::<i32>::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}}),
src/datatypes/src/vectors/json/builder.rs
+1 -107
View File
@@ -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<Value> {
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::<Result<Vec<_>>>()?;
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::<Result<Vec<_>>>()?;
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<Value>) -> Result<Value> {
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 { .. }));
}
}
src/mito2/src/memtable/bulk/json_align.rs
+48 -2
View File
@@ -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()]);
src/mito2/src/read/compat.rs
+1 -1
View File
@@ -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())
src/mito2/src/read/flat_projection.rs
+1 -1
View File
@@ -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)?;
}
src/mito2/src/sst/parquet/json_align/stream.rs
+1 -1
View File
@@ -175,7 +175,7 @@ fn align_array(array: &ArrayRef, field: &FieldRef) -> Result<ArrayRef> {
if is_structured_json_field(field) {
return JsonArray::from(array)
.try_align(field.data_type())
.project_to(field.data_type())
.context(DataTypeMismatchSnafu);
}
src/query/src/optimizer/json_type_concretize.rs
+271 -4
View File
@@ -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<Transformed<LogicalPlan>> {
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<HashMap<String, JsonNativeType>> {
let mut json_types = HashMap::<String, JsonNativeType>::new();
@@ -148,16 +248,21 @@ fn deduce_json_type(expr: &Expr) -> Result<Option<(String, JsonNativeType)>> {
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<DataType>) -> Result<Expr> {
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<DummyTableProvider>, 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<Expr>) -> Result<(Arc<DummyTableProvider>, 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::<Expr>::new())?
.aggregate(Vec::<Expr>::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::<Expr>::new())?
.aggregate(Vec::<Expr>::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::<Expr>::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(
tests/cases/standalone/common/types/json/json2.result
+1 -104
View File
@@ -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
tests/cases/standalone/common/types/json/json2.sql
-65
View File
@@ -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;
tests/cases/standalone/common/types/json/json2_cast.result
+268
View File
@@ -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
tests/cases/standalone/common/types/json/json2_cast.sql
+108
View File
@@ -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;
tests/cases/standalone/common/types/json/json2_limit.result
+174
View File
@@ -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
tests/cases/standalone/common/types/json/json2_limit.sql
+110
View File
@@ -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;
tests/cases/standalone/common/types/json/json2_type_hints.result
+24
View File
@@ -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,
tests/cases/standalone/common/types/json/json2_type_hints.sql
+18
View File
@@ -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,