diff --git a/Cargo.lock b/Cargo.lock index 6ec2a2d2764..79e325b19b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14005,6 +14005,7 @@ version = "1.4.0-alpha.0" dependencies = [ "api", "arrow-buffer 59.2.0", + "arrow-schema 59.2.0", "chrono", "common-base", "common-catalog", diff --git a/src/mito2/src/read/compat.rs b/src/mito2/src/read/compat.rs index 0e566b7e078..acc05f0e9d2 100644 --- a/src/mito2/src/read/compat.rs +++ b/src/mito2/src/read/compat.rs @@ -19,7 +19,7 @@ use std::sync::Arc; use api::v1::SemanticType; use datatypes::arrow::array::{ - Array, ArrayRef, BinaryArray, BinaryBuilder, DictionaryArray, UInt32Array, + Array, ArrayRef, BinaryArray, BinaryBuilder, DictionaryArray, UInt32Array, new_null_array, }; use datatypes::arrow::compute::{TakeOptions, take}; use datatypes::arrow::datatypes::{FieldRef, Schema, SchemaRef}; @@ -27,7 +27,7 @@ use datatypes::arrow::record_batch::RecordBatch; use datatypes::data_type::ConcreteDataType; use datatypes::prelude::DataType; use datatypes::value::Value; -use datatypes::vectors::VectorRef; +use datatypes::vectors::{Helper, VectorRef}; use mito_codec::row_converter::{ CompositeValues, PrimaryKeyCodec, SortField, build_primary_key_codec, build_primary_key_codec_with_fields, @@ -192,20 +192,32 @@ impl FlatCompatBatch { }); } else { // Create a default vector with 1 element for that column. - let default_vector = expect_column - .column_schema - .create_default_vector(1) - .context(CreateDefaultSnafu { - region_id: expect_metadata.region_id, - column: &expect_column.column_schema.name, - })? - .with_context(|| CompatReaderSnafu { - region_id: expect_metadata.region_id, - reason: format!( - "column {} does not have a default value to read", - expect_column.column_schema.name - ), - })?; + let default_vector = if expect_data_type.is_json2() { + // JSON2 columns do not support defaults, so missing columns use NULL. + // `expect_data_type` may be a query-concretized JSON2 type, + // while region metadata still carries its unconcretized type. + // The default vector must match the target Arrow field above. + Helper::try_into_vector(new_null_array(&expect_data_type.as_arrow_type(), 1)) + .context(CreateDefaultSnafu { + region_id: expect_metadata.region_id, + column: &expect_column.column_schema.name, + })? + } else { + expect_column + .column_schema + .create_default_vector(1) + .context(CreateDefaultSnafu { + region_id: expect_metadata.region_id, + column: &expect_column.column_schema.name, + })? + .with_context(|| CompatReaderSnafu { + region_id: expect_metadata.region_id, + reason: format!( + "column {} does not have a default value to read", + expect_column.column_schema.name + ), + })? + }; index_or_defaults.push(IndexOrDefault::DefaultValue { default_vector, semantic_type: expect_column.semantic_type, diff --git a/src/operator/src/expr_helper.rs b/src/operator/src/expr_helper.rs index 1d06c24ff7f..02c1af46706 100644 --- a/src/operator/src/expr_helper.rs +++ b/src/operator/src/expr_helper.rs @@ -1192,6 +1192,7 @@ mod tests { use std::collections::HashMap; use api::v1::{SetDatabaseOptions, UnsetDatabaseOptions}; + use datatypes::extension::json::is_json2_extension_type; use datatypes::value::Value; use session::context::{QueryContext, QueryContextBuilder}; use sql::dialect::GreptimeDbDialect; @@ -1201,6 +1202,29 @@ mod tests { use super::*; + #[test] + fn test_to_alter_add_json2_column_preserves_extension() { + let sql = "ALTER TABLE monitor ADD COLUMN payload JSON2;"; + let stmt = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + .unwrap() + .pop() + .unwrap(); + let Statement::AlterTable(alter_table) = stmt else { + unreachable!() + }; + + let expr = to_alter_table_expr(alter_table, &QueryContext::arc()).unwrap(); + let AlterTableKind::AddColumns(AddColumns { add_columns, .. }) = expr.kind.unwrap() else { + unreachable!() + }; + let column_def = add_columns[0].column_def.as_ref().unwrap(); + let column_schema = try_as_column_schema(column_def).unwrap(); + let schema = Schema::new(vec![column_schema]); + + assert!(is_json2_extension_type(&schema.arrow_schema().fields()[0])); + } + #[test] fn test_create_flow_tql_expr() { let sql = r#" diff --git a/src/sql/Cargo.toml b/src/sql/Cargo.toml index 685d96336b1..de972bec30e 100644 --- a/src/sql/Cargo.toml +++ b/src/sql/Cargo.toml @@ -13,6 +13,7 @@ enterprise = [] [dependencies] api.workspace = true arrow-buffer.workspace = true +arrow-schema.workspace = true chrono.workspace = true common-base.workspace = true common-catalog.workspace = true diff --git a/src/sql/src/statements.rs b/src/sql/src/statements.rs index 2ea716678c7..8a001510045 100644 --- a/src/sql/src/statements.rs +++ b/src/sql/src/statements.rs @@ -36,7 +36,10 @@ pub mod truncate; use std::sync::Arc; use api::helper::ColumnDataTypeWrapper; -use api::v1::SemanticType; +use api::v1::{ColumnOptions, SemanticType}; +use arrow_schema::extension::{ + EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY, ExtensionType, +}; use common_sql::default_constraint::parse_column_default_constraint; use common_time::timezone::Timezone; use datatypes::extension::json::{Json2ExtensionType, JsonMetadata}; @@ -148,25 +151,26 @@ pub fn column_to_schema( column_schema.set_inverted_index(column.extensions.inverted_index_options.is_some()); - let is_json2_column = if let SqlDataType::Custom(object_name, _) = column.data_type() { - object_name - .0 - .first() - .map(|x| x.to_string_unquoted().eq_ignore_ascii_case(JSON2_TYPE_NAME)) - .unwrap_or_default() - } else { - false - }; - if is_json2_column { - let settings = column - .extensions - .build_json_settings()? - .unwrap_or_else(JsonSettings::new_v2); - let extension = Json2ExtensionType::new(Arc::new(JsonMetadata::new(settings))); + set_json2_extension(&mut column_schema, column)?; + + Ok(column_schema) +} + +/// Attaches extension metadata and settings to JSON2 columns. +fn set_json2_extension(column_schema: &mut ColumnSchema, column: &Column) -> Result<()> { + if !column_schema.data_type.is_json2() { + return Ok(()); + } + + let settings = column + .extensions + .build_json_settings()? + .unwrap_or_else(JsonSettings::new_v2); + if let Some(extension) = json2_extension(column.data_type(), settings) { column_schema.with_extension_type(&extension); } - Ok(column_schema) + Ok(()) } /// Convert `ColumnDef` in sqlparser to `ColumnDef` in gRPC proto. @@ -204,6 +208,22 @@ pub fn sql_column_def_to_grpc_column_def( SemanticType::Field }; + // TODO(fys): Extend the ALTER TABLE ADD COLUMN parser to support JSON2 + // type hints and pass the parsed JsonSettings through this conversion. + let options = json2_extension(&col.data_type, JsonSettings::new_v2()).map(|extension| { + let mut options = ColumnOptions::default(); + options.options.insert( + EXTENSION_TYPE_NAME_KEY.to_string(), + Json2ExtensionType::NAME.to_string(), + ); + if let Some(metadata) = extension.serialize_metadata() { + options + .options + .insert(EXTENSION_TYPE_METADATA_KEY.to_string(), metadata); + } + options + }); + Ok(api::v1::ColumnDef { name, data_type: datatype as i32, @@ -212,10 +232,26 @@ pub fn sql_column_def_to_grpc_column_def( semantic_type: semantic_type as _, comment: String::new(), datatype_extension: datatype_ext, - options: None, + options, }) } +fn json2_extension(data_type: &SqlDataType, settings: JsonSettings) -> Option { + let SqlDataType::Custom(name, _) = data_type else { + return None; + }; + if !name.0.first().is_some_and(|name| { + name.to_string_unquoted() + .eq_ignore_ascii_case(JSON2_TYPE_NAME) + }) { + return None; + } + + Some(Json2ExtensionType::new(Arc::new(JsonMetadata::new( + settings, + )))) +} + pub fn sql_data_type_to_concrete_data_type(data_type: &SqlDataType) -> Result { match data_type { SqlDataType::BigInt(_) | SqlDataType::Int64 => Ok(ConcreteDataType::int64_datatype()), diff --git a/tests/cases/standalone/common/types/json/json2.result b/tests/cases/standalone/common/types/json/json2.result index ce0c9748ce1..7114d9bafa1 100644 --- a/tests/cases/standalone/common/types/json/json2.result +++ b/tests/cases/standalone/common/types/json/json2.result @@ -513,3 +513,65 @@ drop table json2_finite_paths; Affected Rows: 0 +-- A newly added JSON2 column must not read data from a dropped column with the +-- same name in an older SST. Column identity is determined by column id, not +-- by the parquet root name. +create table json2_drop_add_same_name ( + ts timestamp time index, + b json2 +) with ( + 'append_mode' = 'true', + 'sst_format' = 'flat', +); + +Affected Rows: 0 + +insert into json2_drop_add_same_name +values (1, '{"a": "old"}'); + +Affected Rows: 1 + +admin flush_table('json2_drop_add_same_name'); + ++-----------------------------------------------+ +| ADMIN flush_table('json2_drop_add_same_name') | ++-----------------------------------------------+ +| 0 | ++-----------------------------------------------+ + +alter table json2_drop_add_same_name drop column b; + +Affected Rows: 0 + +alter table json2_drop_add_same_name add column b json2; + +Affected Rows: 0 + +insert into json2_drop_add_same_name +values (2, '{"a": "new"}'); + +Affected Rows: 1 + +admin flush_table('json2_drop_add_same_name'); + ++-----------------------------------------------+ +| ADMIN flush_table('json2_drop_add_same_name') | ++-----------------------------------------------+ +| 0 | ++-----------------------------------------------+ + +select ts, b.a +from json2_drop_add_same_name +order by ts; + ++-------------------------+--------------------------------------------------+ +| ts | json_get(json2_drop_add_same_name.b,Utf8("$.a")) | ++-------------------------+--------------------------------------------------+ +| 1970-01-01T00:00:00.001 | | +| 1970-01-01T00:00:00.002 | new | ++-------------------------+--------------------------------------------------+ + +drop table json2_drop_add_same_name; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/types/json/json2.sql b/tests/cases/standalone/common/types/json/json2.sql index 17248741c35..55642b7530c 100644 --- a/tests/cases/standalone/common/types/json/json2.sql +++ b/tests/cases/standalone/common/types/json/json2.sql @@ -157,3 +157,33 @@ from json2_finite_paths order by ts; drop table json2_finite_paths; + +-- A newly added JSON2 column must not read data from a dropped column with the +-- same name in an older SST. Column identity is determined by column id, not +-- by the parquet root name. +create table json2_drop_add_same_name ( + ts timestamp time index, + b json2 +) with ( + 'append_mode' = 'true', + 'sst_format' = 'flat', +); + +insert into json2_drop_add_same_name +values (1, '{"a": "old"}'); + +admin flush_table('json2_drop_add_same_name'); + +alter table json2_drop_add_same_name drop column b; +alter table json2_drop_add_same_name add column b json2; + +insert into json2_drop_add_same_name +values (2, '{"a": "new"}'); + +admin flush_table('json2_drop_add_same_name'); + +select ts, b.a +from json2_drop_add_same_name +order by ts; + +drop table json2_drop_add_same_name; diff --git a/tests/cases/standalone/common/types/json/json2_limit.result b/tests/cases/standalone/common/types/json/json2_limit.result index 555e76f1e5e..32bbfbaf0a7 100644 --- a/tests/cases/standalone/common/types/json/json2_limit.result +++ b/tests/cases/standalone/common/types/json/json2_limit.result @@ -95,6 +95,27 @@ drop table json2_alter_non_append; Affected Rows: 0 +create table json2_alter_add_settings ( + ts timestamp time index +) with ( + 'append_mode' = 'true' +); + +Affected Rows: 0 + +-- ALTER TABLE ADD COLUMN does not yet support JSON2 settings or type hints. +alter table json2_alter_add_settings add column j json2(max_auto_expanded_paths = 100); + +Error: 2000(InvalidSyntax), Invalid SQL syntax: sql parser error: Expected: type modifiers, found: = at Line: 1, Column: 81 + +alter table json2_alter_add_settings add column k json2(service string); + +Error: 2000(InvalidSyntax), SQL data type not supported yet: Custom(ObjectName([Identifier(Ident { value: "json2", quote_style: None, span: Span(Location(1,51)..Location(1,56)) })]), ["service", "string"]) + +drop table json2_alter_add_settings; + +Affected Rows: 0 + create table json2_set_append_mode_false ( ts timestamp time index, j json2 diff --git a/tests/cases/standalone/common/types/json/json2_limit.sql b/tests/cases/standalone/common/types/json/json2_limit.sql index 653fdcec12c..1bbadfd5adc 100644 --- a/tests/cases/standalone/common/types/json/json2_limit.sql +++ b/tests/cases/standalone/common/types/json/json2_limit.sql @@ -58,6 +58,19 @@ alter table json2_alter_non_append add column j json2; drop table json2_alter_non_append; +create table json2_alter_add_settings ( + ts timestamp time index +) with ( + 'append_mode' = 'true' +); + +-- ALTER TABLE ADD COLUMN does not yet support JSON2 settings or type hints. +alter table json2_alter_add_settings add column j json2(max_auto_expanded_paths = 100); + +alter table json2_alter_add_settings add column k json2(service string); + +drop table json2_alter_add_settings; + create table json2_set_append_mode_false ( ts timestamp time index, j json2