From b369d35d8b8a841bac9fae108508be8c24e354b8 Mon Sep 17 00:00:00 2001 From: LFC <990479+MichaelScofield@users.noreply.github.com> Date: Wed, 28 Oct 2026 02:43:29 +0000 Subject: [PATCH] feat: enable JSON2 columns on non-append-only tables (#9985) feat(mito2): support JSON2 in non-append-only tables Signed-off-by: luofucong --- src/mito2/src/read/compat.rs | 177 ++++- src/mito2/src/read/flat_projection.rs | 159 ++--- src/mito2/src/read/last_row.rs | 67 +- src/mito2/src/read/scan_util.rs | 215 +++++- src/mito2/src/read/seq_scan.rs | 6 +- src/mito2/src/read/series_reader.rs | 7 +- src/mito2/src/read/unordered_scan.rs | 4 +- src/mito2/src/sst/parquet/file_range.rs | 73 +- src/mito2/src/sst/parquet/json_align.rs | 75 +- src/mito2/src/sst/parquet/reader.rs | 8 + src/mito2/src/test_util/sst_util.rs | 57 ++ src/operator/src/statement/ddl.rs | 75 +- src/query/src/datafusion.rs | 61 +- .../common/types/json/json2_limit.result | 53 +- .../common/types/json/json2_limit.sql | 33 +- .../common/types/json/json2_non_append.result | 675 ++++++++++++++++++ .../common/types/json/json2_non_append.sql | 201 ++++++ 17 files changed, 1655 insertions(+), 291 deletions(-) create mode 100644 tests/cases/standalone/common/types/json/json2_non_append.result create mode 100644 tests/cases/standalone/common/types/json/json2_non_append.sql diff --git a/src/mito2/src/read/compat.rs b/src/mito2/src/read/compat.rs index acc05f0e9d2..6f9a4a36e50 100644 --- a/src/mito2/src/read/compat.rs +++ b/src/mito2/src/read/compat.rs @@ -42,9 +42,9 @@ use crate::error::{ NewRecordBatchSnafu, Result, UnexpectedSnafu, UnsupportedOperationSnafu, }; use crate::read::flat_projection::{FlatProjectionMapper, flat_projected_columns}; +use crate::sst::internal_fields; use crate::sst::parquet::flat_format::{FlatReadFormat, primary_key_column_index}; use crate::sst::parquet::format::{INTERNAL_COLUMN_NUM, PrimaryKeyArray}; -use crate::sst::{internal_fields, tag_maybe_to_dictionary_field, with_field_id}; /// Returns true if the columns in the `projection_mapper` and `read_format` have same data types /// and primary key encodings. @@ -116,8 +116,12 @@ impl FlatCompatBatch { return FlatCompatBatch::try_new_compact_sparse(mapper, actual); } - let (index_or_defaults, fields) = - Self::compute_index_and_fields(&actual_schema, expect_schema, mapper.metadata())?; + let (index_or_defaults, fields) = Self::compute_index_and_fields( + &actual_schema, + expect_schema, + mapper.metadata(), + mapper.input_arrow_schema(compaction).fields(), + )?; let compat_pk = FlatCompatPrimaryKey::new(mapper.metadata(), actual)?; @@ -132,6 +136,7 @@ impl FlatCompatBatch { actual_schema: &[(ColumnId, ConcreteDataType)], expect_schema: &[(ColumnId, ConcreteDataType)], expect_metadata: &RegionMetadata, + target_fields: &[FieldRef], ) -> Result<(Vec, Vec)> { // Maps column id to the index and data type in the actual schema. let actual_schema_index: HashMap<_, _> = actual_schema @@ -142,29 +147,18 @@ impl FlatCompatBatch { let mut index_or_defaults = Vec::with_capacity(expect_schema.len()); let mut fields = Vec::with_capacity(expect_schema.len()); - for (column_id, expect_data_type) in expect_schema { + for (field_index, (column_id, expect_data_type)) in expect_schema.iter().enumerate() { // Safety: expect_schema comes from the same mapper. let column_index = expect_metadata.column_index_by_id(*column_id).unwrap(); let expect_column = &expect_metadata.column_metadatas[column_index]; - let column_field = &expect_metadata.schema.arrow_schema().fields()[column_index]; - // For tag columns, we need to create a dictionary field. - if expect_column.semantic_type == SemanticType::Tag { - let field = tag_maybe_to_dictionary_field( - &expect_column.column_schema.data_type, - column_field, - ); - fields.push(Arc::new(with_field_id( - (*field).clone(), - expect_column.column_id, - ))); - } else { - let field = with_field_id( - Arc::unwrap_or_clone(column_field.clone()), - expect_column.column_id, - ) - .with_data_type(expect_data_type.as_arrow_type()); - fields.push(Arc::new(field)) - }; + // Query target fields follow mapper.batch_schema(); compaction reads all columns in + // PK -> fields -> timestamp order (fields -> timestamp for sparse encoding), before + // trailing internal columns. Keep this order consistent with expect_schema. + // The mapper already applies string-tag dictionary types and JSON2 compaction layouts, + // which can differ from the logical types in expect_schema. + let column_field = &target_fields[field_index]; + debug_assert_eq!(column_field.name(), &expect_column.column_schema.name); + fields.push(column_field.clone()); if let Some((index, actual_data_type)) = actual_schema_index.get(column_id) { let mut cast_type = None; @@ -194,14 +188,13 @@ impl FlatCompatBatch { // Create a default vector with 1 element for that column. 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 { + // A logical JSON2 placeholder cannot describe a compaction layout. + Helper::try_into_vector(new_null_array(column_field.data_type(), 1)).context( + CreateDefaultSnafu { region_id: expect_metadata.region_id, column: &expect_column.column_schema.name, - })? + }, + )? } else { expect_column .column_schema @@ -256,8 +249,12 @@ impl FlatCompatBatch { .map(|col| (col.column_id, col.column_schema.data_type.clone())) .collect(); - let (index_or_defaults, fields) = - Self::compute_index_and_fields(&actual_schema, &expect_schema, mapper.metadata())?; + let (index_or_defaults, fields) = Self::compute_index_and_fields( + &actual_schema, + &expect_schema, + mapper.metadata(), + mapper.input_arrow_schema(true).fields(), + )?; let compat_pk = FlatCompatPrimaryKey::default(); @@ -663,10 +660,15 @@ mod tests { }; use datatypes::arrow::datatypes::UInt32Type; use datatypes::arrow::record_batch::RecordBatch; + use datatypes::extension::json::JsonMetadata; + use datatypes::json::JsonSettings; use datatypes::prelude::ConcreteDataType; - use datatypes::schema::ColumnSchema; + use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema}; use datatypes::types::json_type::JsonNativeType; + use datatypes::types::parse_string_to_jsonb; use datatypes::value::ValueRef; + use datatypes::vectors::json::array::JsonArray; + use datatypes::vectors::json::json2_physical_data_type; use mito_codec::row_converter::{ DensePrimaryKeyCodec, PrimaryKeyCodecExt, SparsePrimaryKeyCodec, }; @@ -678,7 +680,9 @@ mod tests { use crate::read::flat_projection::FlatProjectionMapper; use crate::read::read_columns::ReadColumns; use crate::sst::parquet::flat_format::FlatReadFormat; + use crate::sst::parquet::{Json2RewriteTargets, Json2TargetLayout}; use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema}; + use crate::test_util; /// Creates a new [RegionMetadata]. fn new_metadata( @@ -1484,4 +1488,113 @@ mod tests { assert_eq!(expected_batch, result); } + #[test] + fn test_compaction_json2_uses_target_fields() + -> std::result::Result<(), Box> { + let (metadata, batch) = test_util::sst_util::new_json2_record_batch()?; + let logical_settings = JsonSettings::default(); + let layout = JsonSettings::try_new(vec![], Some(0))?; + let targets: Json2RewriteTargets = Arc::new(BTreeMap::from([( + 0, + Json2TargetLayout { + extension_metadata: serde_json::to_string(&JsonMetadata::new( + logical_settings.clone(), + ))?, + target_layout: layout.clone(), + }, + )])); + // Model the existing Rewrite output; the compatibility layer only selects or fills arrays. + let json = JsonArray::from(batch.column(0)).rewrite_to_v2( + batch.schema_ref().field(0), + &logical_settings, + &layout, + )?; + let mut fields = batch.schema_ref().fields().to_vec(); + fields[0] = Arc::new( + fields[0] + .as_ref() + .clone() + .with_data_type(json.data_type().clone()), + ); + let mut columns = batch.columns().to_vec(); + columns[0] = json; + let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?; + for encoding in [PrimaryKeyEncoding::Dense, PrimaryKeyEncoding::Sparse] { + for missing_json in [false, true] { + let source = if missing_json { + new_metadata( + &[( + 1, + SemanticType::Timestamp, + ConcreteDataType::timestamp_millisecond_datatype(), + )], + &[], + ) + } else { + metadata.as_ref().clone() + }; + let mut builder = RegionMetadataBuilder::from_existing(source); + builder.primary_key_encoding(encoding); + let actual = Arc::new(builder.build()?); + let mut builder = RegionMetadataBuilder::from_existing(metadata.as_ref().clone()); + builder.primary_key_encoding(encoding); + builder.push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "added", + ConcreteDataType::int64_datatype(), + true, + ) + .with_default_constraint(Some( + ColumnDefaultConstraint::Value(Value::Int64(42)), + ))?, + semantic_type: SemanticType::Field, + column_id: 2, + }); + let expected = Arc::new(builder.build()?); + let read_columns = ReadColumns::new([0, 1, 2]); + let mapper = FlatProjectionMapper::new_with_json2_rewrite_targets( + &expected, + vec![0, 1, 2], + read_columns.clone(), + &targets, + )?; + let read_format = FlatReadFormat::new(actual, read_columns, None, "test", true)?; + let compat = FlatCompatBatch::try_new(&mapper, &read_format, true)? + .ok_or("missing columns require compatibility")?; + let input = if missing_json { + batch.project(&[1, 2, 3, 4])? + } else { + batch.clone() + }; + let output = compat.compat(input)?; + assert_eq!( + output.column(0).data_type(), + &json2_physical_data_type(&layout) + ); + assert_eq!( + output.column(0).null_count(), + if missing_json { 2 } else { 0 } + ); + let defaults = output + .column(1) + .as_any() + .downcast_ref::() + .ok_or("expected int64 defaults")?; + assert_eq!(defaults.values().as_ref(), &[42, 42]); + if !missing_json { + let jsonb = JsonArray::from(output.column(0)).project_to_v2( + output.schema_ref().field(0), + &datatypes::arrow::datatypes::DataType::Binary, + )?; + let jsonb = jsonb + .as_any() + .downcast_ref::() + .ok_or("expected JSONB")?; + assert_eq!(jsonb.value(0), parse_string_to_jsonb(r#"{"a":1}"#)?); + assert_eq!(jsonb.value(1), parse_string_to_jsonb(r#"{"a":2}"#)?); + } + } + } + Ok(()) + } } diff --git a/src/mito2/src/read/flat_projection.rs b/src/mito2/src/read/flat_projection.rs index 1734ca364af..69df6256762 100644 --- a/src/mito2/src/read/flat_projection.rs +++ b/src/mito2/src/read/flat_projection.rs @@ -17,10 +17,7 @@ use std::sync::Arc; use api::v1::SemanticType; -use common_error::ext::BoxedError; -use common_recordbatch::error::{ - ArrowComputeSnafu, DataTypesSnafu, ExternalSnafu, NewDfRecordBatchSnafu, -}; +use common_recordbatch::error::{ArrowComputeSnafu, NewDfRecordBatchSnafu}; use common_recordbatch::{DfRecordBatch, RecordBatch}; use datatypes::arrow::array::{Array, AsArray}; use datatypes::arrow::datatypes::{DataType as ArrowDataType, Field}; @@ -28,15 +25,13 @@ use datatypes::extension::json::is_json2_extension_type; use datatypes::prelude::{ConcreteDataType, DataType}; use datatypes::schema::{Schema, SchemaRef}; use datatypes::value::Value; -use datatypes::vectors::Helper; -use datatypes::vectors::json::array::JsonArray; use datatypes::vectors::json::json2_physical_data_type; use snafu::{OptionExt, ResultExt}; use store_api::metadata::{RegionMetadata, RegionMetadataRef}; use store_api::storage::ColumnId; use crate::cache::CacheStrategy; -use crate::error::{InvalidRequestSnafu, RecordBatchSnafu, Result}; +use crate::error::{InvalidRequestSnafu, NewRecordBatchSnafu, Result}; use crate::read::projection::{read_column_ids_from_projection, repeated_vector_with_cache}; use crate::read::read_columns::{JsonTargetTypes, ReadColumns}; use crate::sst::parquet::Json2RewriteTargets; @@ -150,7 +145,7 @@ impl FlatProjectionMapper { flat_projected_columns(metadata, &format_projection, read_cols.json_target_types()); let input_arrow_schema = - compute_input_arrow_schema(metadata, &batch_schema, &read_cols, json2_rewrite_targets); + compute_input_arrow_schema(metadata, &batch_schema, json2_rewrite_targets); // If projection is empty, we don't output any column. let output_schema = if is_empty_projection { @@ -307,6 +302,7 @@ impl FlatProjectionMapper { /// Converts a flat format [RecordBatch] to a normal [RecordBatch]. /// /// The batch must match the `projection` using to build the mapper. + /// JSON2 columns must already match query types before merge and deduplication. pub(crate) fn convert( &self, batch: &datatypes::arrow::record_batch::RecordBatch, @@ -365,13 +361,6 @@ impl FlatProjectionMapper { } } - let field = &self.output_schema.arrow_schema().fields()[output_idx]; - if is_json2_extension_type(field) && array.data_type() != field.data_type() { - array = JsonArray::from(&array) - .project_to_v2(batch.schema_ref().field(*index), field.data_type()) - .context(DataTypesSnafu)?; - } - arrays.push(array); } @@ -383,30 +372,6 @@ impl FlatProjectionMapper { df_record_batch, )) } - - /// Projects columns from the input batch and converts them into vectors. - pub(crate) fn project_vectors( - &self, - batch: &datatypes::arrow::record_batch::RecordBatch, - ) -> common_recordbatch::error::Result> { - let mut columns = Vec::with_capacity(self.output_schema.num_columns()); - for index in &self.batch_indices { - let mut array = batch.column(*index).clone(); - // Casts dictionary values to the target type. - if let datatypes::arrow::datatypes::DataType::Dictionary(_key_type, value_type) = - array.data_type() - { - let casted = datatypes::arrow::compute::cast(&array, value_type) - .context(ArrowComputeSnafu)?; - array = casted; - } - let vector = Helper::try_into_vector(array) - .map_err(BoxedError::new) - .context(ExternalSnafu)?; - columns.push(vector); - } - Ok(columns) - } } fn single_value_string_dictionary<'a>( @@ -481,7 +446,6 @@ pub(crate) fn flat_projected_columns( pub(crate) fn compute_input_arrow_schema( metadata: &RegionMetadata, batch_schema: &[(ColumnId, ConcreteDataType)], - read_cols: &ReadColumns, json2_rewrite_targets: &Json2RewriteTargets, ) -> datatypes::arrow::datatypes::SchemaRef { let mut new_fields = Vec::with_capacity(batch_schema.len() + 3); @@ -489,11 +453,6 @@ pub(crate) fn compute_input_arrow_schema( let data_type = json2_rewrite_targets .get(column_id) .map(|x| json2_physical_data_type(&x.target_layout)) - .or_else(|| { - read_cols - .json_target_type(*column_id) - .map(|x| x.as_arrow_type()) - }) .unwrap_or_else(|| data_type.as_arrow_type()); let column_metadata = metadata.column_by_id(*column_id).unwrap(); @@ -521,8 +480,7 @@ pub(crate) fn compute_input_arrow_schema( /// Helper to project compaction batches into flat format columns /// (fields + time index + __primary_key + __sequence + __op_type). pub(crate) struct CompactionProjectionMapper { - mapper: FlatProjectionMapper, - assembler: DfBatchAssembler, + projection: Vec, } impl CompactionProjectionMapper { @@ -544,77 +502,74 @@ impl CompactionProjectionMapper { let read_col_ids = metadata.column_metadatas.iter().map(|col| col.column_id); let read_cols = ReadColumns::new(read_col_ids); let mapper = FlatProjectionMapper::new_with_read_columns(metadata, projection, read_cols)?; - let assembler = DfBatchAssembler::new(mapper.output_schema()); - - Ok(Self { mapper, assembler }) + Ok(Self { + projection: mapper.batch_indices, + }) } /// Projects columns and appends internal columns for compaction output. /// /// The input batch is expected to be in flat format with internal columns appended. pub(crate) fn project(&self, batch: DfRecordBatch) -> Result { - let columns = self - .mapper - .project_vectors(&batch) - .context(RecordBatchSnafu)?; - self.assembler - .build_df_record_batch_with_internal(&batch, columns) - .context(RecordBatchSnafu) - } -} - -/// Builds [DfRecordBatch] with internal columns appended. -pub(crate) struct DfBatchAssembler { - output_arrow_schema_with_internal: datatypes::arrow::datatypes::SchemaRef, -} - -impl DfBatchAssembler { - /// Precomputes the output schema with internal columns. - pub(crate) fn new(output_schema: SchemaRef) -> Self { - let fields = output_schema - .arrow_schema() - .fields() - .into_iter() - .chain(internal_fields().iter()) - .cloned() - .collect::>(); - let output_arrow_schema_with_internal = - Arc::new(datatypes::arrow::datatypes::Schema::new(fields)); - Self { - output_arrow_schema_with_internal, - } - } - - /// Builds a [DfRecordBatch] from projected vectors plus internal columns. - /// - /// Assumes the input batch already contains internal columns as the last three fields - /// ("__primary_key", "__sequence", "__op_type"). - pub(crate) fn build_df_record_batch_with_internal( - &self, - batch: &datatypes::arrow::record_batch::RecordBatch, - mut columns: Vec, - ) -> common_recordbatch::error::Result { - let num_columns = batch.columns().len(); - // The last 3 columns are the internal columns. - let internal_indices = [num_columns - 3, num_columns - 2, num_columns - 1]; - for index in internal_indices.iter() { - let array = batch.column(*index).clone(); - let vector = Helper::try_into_vector(array) - .map_err(BoxedError::new) - .context(ExternalSnafu)?; - columns.push(vector); - } - RecordBatch::to_df_record_batch(self.output_arrow_schema_with_internal.clone(), columns) + let mut projection = self.projection.clone(); + projection.extend(batch.num_columns() - 3..batch.num_columns()); + // Projection must preserve the Arrow layouts already produced by compaction Rewrite. + batch.project(&projection).context(NewRecordBatchSnafu) } } #[cfg(test)] mod tests { use datatypes::schema::ColumnSchema; + use datatypes::vectors::Helper; use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder}; use store_api::storage::RegionId; use super::*; + use datatypes::types::json_type::JsonNativeType; + + #[test] + fn test_json2_root_hint_uses_output_type_before_merge() + -> std::result::Result<(), Box> { + let mut builder = RegionMetadataBuilder::new(RegionId::new(1024, 0)); + builder + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "j", + ConcreteDataType::json2(JsonNativeType::object()), + true, + ), + semantic_type: SemanticType::Field, + column_id: 0, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 1, + }); + let metadata = Arc::new(builder.build()?); + let read_cols = ReadColumns::new([0, 1]) + .with_json_target_types([(0, JsonNativeType::String)].into_iter().collect()); + let mapper = FlatProjectionMapper::new_with_read_columns(&metadata, vec![0, 1], read_cols)?; + let input = mapper.input_arrow_schema(false); + assert_eq!( + &ArrowDataType::Binary, + input.field_with_name("j")?.data_type() + ); + assert_eq!( + mapper + .output_schema() + .arrow_schema() + .field_with_name("j")? + .data_type(), + input.field_with_name("j")?.data_type(), + ); + Ok(()) + } fn metadata_with_legacy_json() -> RegionMetadataRef { let mut builder = RegionMetadataBuilder::new(RegionId::new(1024, 0)); diff --git a/src/mito2/src/read/last_row.rs b/src/mito2/src/read/last_row.rs index b7da3e1f70c..6af6a08717a 100644 --- a/src/mito2/src/read/last_row.rs +++ b/src/mito2/src/read/last_row.rs @@ -16,7 +16,8 @@ use std::sync::Arc; -use datatypes::arrow::array::{Array, BinaryArray, UInt32Array}; +use api::v1::OpType; +use datatypes::arrow::array::{Array, BinaryArray, UInt8Array, UInt32Array}; use datatypes::arrow::compute::{concat_batches, take_record_batch}; use datatypes::arrow::record_batch::RecordBatch; use futures::{Stream, TryStreamExt}; @@ -31,7 +32,9 @@ use crate::error::{ComputeArrowSnafu, Result}; use crate::read::read_columns::JsonTargetTypes; use crate::read::{BoxedRecordBatchStream, timestamp_array_to_i64_slice}; use crate::sst::parquet::DEFAULT_READ_BATCH_SIZE; -use crate::sst::parquet::flat_format::{primary_key_column_index, time_index_column_index}; +use crate::sst::parquet::flat_format::{ + op_type_column_index, primary_key_column_index, time_index_column_index, +}; use crate::sst::parquet::format::{PrimaryKeyArray, primary_key_offsets}; use crate::sst::parquet::read_columns::ParquetReadColumns; use crate::sst::parquet::reader::FlatRowGroupReader; @@ -287,6 +290,31 @@ impl LastKeyState { } impl FlatLastTimestampSelector { + /// Keeps the latest timestamp candidates in one sorted PUT batch, including ties. + pub(crate) fn select_batch(batch: RecordBatch) -> Result { + if batch.num_rows() <= 1 { + return Ok(batch); + } + let Some(op_types) = batch + .column(op_type_column_index(batch.num_columns())) + .as_any() + .downcast_ref::() + else { + return Ok(batch); + }; + if op_types.values().iter().any(|op| *op != OpType::Put as u8) { + return Ok(batch); + } + let mut selector = Self::default(); + let mut pending = BatchBuffer::new(); + selector.on_next(batch.clone(), &mut pending)?; + selector.finish(&mut pending)?; + if pending.num_rows == batch.num_rows() { + return Ok(batch); + } + pending.concat() + } + /// Processes the next batch and appends completed-key results into `output_buffer`. pub(crate) fn on_next( &mut self, @@ -471,6 +499,41 @@ mod tests { use super::*; + #[test] + fn test_select_batch_preserves_timestamp_ties_and_deletes() + -> std::result::Result<(), Box> { + let input = new_flat_batch( + &[b"a", b"a", b"a", b"b", b"b", b"c"], + &[1, 2, 2, 1, 2, 1], + &[1, 2, 3, 4, 5, 6], + ); + let expected = take_record_batch(&input, &UInt32Array::from(vec![1, 2, 4, 5]))?; + assert_eq!( + expected, + FlatLastTimestampSelector::select_batch(input.clone())? + ); + let empty = input.slice(0, 0); + assert_eq!( + empty, + FlatLastTimestampSelector::select_batch(empty.clone())? + ); + let mut columns = input.columns().to_vec(); + columns[op_type_column_index(input.num_columns())] = Arc::new(UInt8Array::from(vec![ + OpType::Delete as u8, + OpType::Put as u8, + OpType::Put as u8, + OpType::Put as u8, + OpType::Put as u8, + OpType::Put as u8, + ])); + let with_delete = RecordBatch::try_new(input.schema(), columns)?; + assert_eq!( + with_delete, + FlatLastTimestampSelector::select_batch(with_delete.clone())? + ); + Ok(()) + } + /// Helper to build a flat format RecordBatch for testing. fn new_flat_batch(primary_keys: &[&[u8]], timestamps: &[i64], fields: &[i64]) -> RecordBatch { let num_rows = timestamps.len(); diff --git a/src/mito2/src/read/scan_util.rs b/src/mito2/src/read/scan_util.rs index a67e2449baa..5f82743f92d 100644 --- a/src/mito2/src/read/scan_util.rs +++ b/src/mito2/src/read/scan_util.rs @@ -26,7 +26,9 @@ use common_telemetry::tracing; use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder, Time}; use datatypes::arrow::record_batch::RecordBatch; use datatypes::timestamp::timestamp_array_to_primitive; -use futures::Stream; +#[cfg(feature = "enterprise")] +use futures::TryStreamExt; +use futures::{Stream, StreamExt}; use prometheus::IntGauge; use smallvec::SmallVec; use snafu::ResultExt; @@ -40,6 +42,7 @@ use crate::metrics::{ }; use crate::read::dedup::{DedupMetrics, DedupMetricsReport}; use crate::read::flat_merge::{MergeMetrics, MergeMetricsReport}; +use crate::read::last_row::FlatLastTimestampSelector; use crate::read::pruner::PartitionPruner; use crate::read::range::{RangeMeta, RowGroupIndex}; use crate::read::scan_region::StreamContext; @@ -50,6 +53,7 @@ use crate::sst::index::fulltext_index::applier::FulltextIndexApplyMetrics; use crate::sst::index::inverted_index::applier::InvertedIndexApplyMetrics; use crate::sst::parquet::file_range::{FileRange, PreFilterMode}; use crate::sst::parquet::flat_format::{sequence_column_index, time_index_column_index}; +use crate::sst::parquet::json_align; use crate::sst::parquet::reader::{MetadataCacheMetrics, ReaderFilterMetrics, ReaderMetrics}; use crate::sst::parquet::row_group::ParquetFetchMetrics; use crate::sst::parquet::{DEFAULT_READ_BATCH_SIZE, DEFAULT_ROW_GROUP_SIZE}; @@ -1150,6 +1154,15 @@ pub(crate) struct SeriesDistributorMetrics { pub(crate) divider_cost: Duration, } +/// Aligns memtable and extension batches; SST readers align before schema compatibility. +pub(crate) fn align_query_source_batch( + context: &StreamContext, + batch: RecordBatch, +) -> Result { + let query_schema = context.input.mapper.input_arrow_schema(false); + json_align::align_query_batch(batch, &query_schema) +} + /// Scans memtable ranges at `index` using flat format that returns RecordBatch. #[tracing::instrument( skip_all, @@ -1187,6 +1200,39 @@ pub(crate) fn scan_flat_mem_ranges( } } +/// Aligns memtable JSON2 columns to query types before merge/dedup. +/// +/// For append-only queries with `TimeSeriesRowSelector::LastRow { after_merge: false }`, +/// first keeps all rows tied at each series' maximum timestamp within each sorted +/// PUT-only batch, avoiding JSON2 conversion for discarded rows. Final cross-source +/// LastRow selection still runs. Batches containing deletes are not reduced. +/// This optimization uses the query's series row selector, not its merge mode; +/// non-append tables retain all input rows for deduplication and last-non-null backfill. +pub(crate) fn prepare_flat_mem_stream( + context: Arc, + stream: impl Stream> + Send + 'static, +) -> BoxedRecordBatchStream { + if context.input.read_cols.json_target_types().is_empty() { + return Box::pin(stream); + } + + let select_last = context.input.append_mode + && matches!( + context.input.series_row_selector, + Some(TimeSeriesRowSelector::LastRow { after_merge: false }) + ); + Box::pin(stream.map(move |batch| { + let batch = batch?; + // Select raw candidates before decoding JSON2; final cross-source selection remains. + let batch = if select_last { + FlatLastTimestampSelector::select_batch(batch)? + } else { + batch + }; + align_query_source_batch(&context, batch) + })) +} + /// Files with row count greater than this threshold can contribute to the estimation. const SPLIT_ROW_THRESHOLD: u64 = DEFAULT_ROW_GROUP_SIZE as u64; /// Number of series threshold for splitting batches. @@ -1562,12 +1608,10 @@ pub fn build_flat_file_range_scan_stream( let mapper = range.compaction_projection_mapper(); while let Some(record_batch) = reader.next_batch().await? { let record_batch = if let Some(mapper) = mapper { - let batch = mapper.project(record_batch)?; - batch + mapper.project(record_batch)? } else { record_batch }; - let Some(record_batch) = filter_flat_batch_by_sequence( record_batch, stream_ctx.input.sequence_range, @@ -1622,10 +1666,15 @@ pub(crate) async fn scan_flat_extension_range( let range = context.input.extension_range(index.index); let reader = range.flat_reader(context.as_ref(), options); let stream = reader - .read(context, partition_metrics, index) + .read(context.clone(), partition_metrics, index) .await .context(crate::error::ScanExternalRangeSnafu)?; - Ok(stream) + if context.input.read_cols.json_target_types().is_empty() { + return Ok(stream); + } + Ok(Box::pin(stream.and_then(move |batch| { + std::future::ready(align_query_source_batch(&context, batch)) + }))) } pub(crate) async fn maybe_scan_flat_other_ranges( @@ -1736,21 +1785,27 @@ pub(crate) fn split_record_batch(record_batch: RecordBatch, batches: &mut VecDeq #[cfg(test)] mod tests { + use arrow_select::concat::concat_batches; + use common_time::Timestamp; + use datatypes::arrow::array::{BinaryArray, TimestampMillisecondArray, UInt64Array}; + use datatypes::types::json_type::JsonNativeType; + use futures::TryStreamExt; + use smallvec::{SmallVec, smallvec}; use std::sync::Arc; use std::time::Instant; - - use common_time::Timestamp; - use smallvec::{SmallVec, smallvec}; use store_api::storage::RegionId; use super::*; use crate::cache::CacheStrategy; + use crate::memtable::bulk::context::BulkIterContext; + use crate::memtable::bulk::part_reader::BulkPartBatchIter; use crate::memtable::{ BoxedBatchIterator, BoxedRecordBatchIterator, IterBuilder, MemtableRange, MemtableRangeContext, MemtableStats, }; use crate::read::flat_projection::FlatProjectionMapper; use crate::read::range::{MemRangeBuilder, SourceIndex}; + use crate::read::read_columns::ReadColumns; use crate::read::scan_region::ScanInput; use crate::sst::file::{FileHandle, FileMeta}; use crate::sst::file_purger::NoopFilePurger; @@ -1853,6 +1908,148 @@ mod tests { } } + #[tokio::test] + async fn test_memtable_last_row_selects_before_json2_alignment() + -> std::result::Result<(), Box> { + struct BatchIter(RecordBatch, Arc, Option); + impl IterBuilder for BatchIter { + fn build(&self, _metrics: Option) -> Result { + Ok(Box::new(std::iter::empty())) + } + fn is_record_batch(&self) -> bool { + true + } + fn build_record_batch( + &self, + _time_range: Option, + metrics: Option, + ) -> Result { + Ok(Box::new(BulkPartBatchIter::from_single( + self.0.clone(), + self.1.clone(), + self.2, + 1, + metrics, + ))) + } + } + let (metadata, batch) = crate::test_util::sst_util::new_json2_record_batch()?; + let batch = concat_batches(&batch.schema(), &[batch.clone(), batch.slice(0, 1)])?; + let mut columns = batch.columns().to_vec(); + columns[1] = Arc::new(TimestampMillisecondArray::from(vec![1, 2, 3])); + let seq_index = sequence_column_index(batch.num_columns()); + columns[seq_index] = Arc::new(UInt64Array::from(vec![1, 2, 3])); + let batch = RecordBatch::try_new(batch.schema(), columns)?; + let reader_context = Arc::new(BulkIterContext::new( + metadata.clone(), + None, + None, + false, + DEFAULT_READ_BATCH_SIZE, + )?); + let time_range = (Timestamp::new_millisecond(1), Timestamp::new_millisecond(3)); + let env = SchedulerEnv::new().await; + for (append, selector, sequence, count) in [ + ( + true, + Some(TimeSeriesRowSelector::LastRow { after_merge: false }), + None, + 1, + ), + ( + true, + Some(TimeSeriesRowSelector::LastRow { after_merge: true }), + None, + 3, + ), + ( + false, + Some(TimeSeriesRowSelector::LastRow { after_merge: false }), + None, + 3, + ), + (true, None, None, 3), + ( + true, + Some(TimeSeriesRowSelector::LastRow { after_merge: false }), + Some(SequenceRange::GtLtEq { min: 0, max: 2 }), + 1, + ), + ] { + let range_context = Arc::new(MemtableRangeContext::new( + 0, + Box::new(BatchIter(batch.clone(), reader_context.clone(), sequence)), + Default::default(), + )); + let stats = MemtableStats { + time_range: Some(time_range), + num_rows: 3, + num_ranges: 1, + series_count: 1, + ..Default::default() + }; + let range = MemtableRange::new(range_context, stats.clone()); + let read_cols = ReadColumns::new([0, 1]) + .with_json_target_types([(0, JsonNativeType::Variant)].into_iter().collect()); + let mapper = + FlatProjectionMapper::new_with_read_columns(&metadata, vec![0, 1], read_cols)?; + let input = ScanInput::builder(env.access_layer.clone(), mapper) + .with_append_mode(append) + .with_series_row_selector(selector) + .with_sequence_range(sequence) + .with_memtables(vec![MemRangeBuilder::new(range, stats)]) + .build(); + let context = Arc::new(StreamContext { + input, + ranges: vec![], + query_start: Instant::now(), + }); + let metrics = PartitionMetrics::new( + metadata.region_id, + 0, + "test_json2", + context.query_start, + false, + &ExecutionPlanMetricsSet::new(), + ); + let mut raw_stream = Box::pin(scan_flat_mem_ranges( + context.clone(), + metrics, + RowGroupIndex { + index: 0, + row_group_index: -1, + }, + time_range, + )); + let raw = raw_stream.try_next().await?.ok_or("expected a raw batch")?; + let expected_raw = if sequence.is_some() { + batch.slice(0, 2) + } else { + batch.clone() + }; + assert_eq!(expected_raw, raw); + assert!(raw_stream.try_next().await?.is_none()); + let mut stream = prepare_flat_mem_stream(context, futures::stream::iter([Ok(raw)])); + let output = stream.try_next().await?.ok_or("expected a JSON2 batch")?; + assert_eq!(count, output.num_rows()); + let values = output + .column(0) + .as_any() + .downcast_ref::() + .ok_or("expected aligned JSONB")?; + assert_eq!( + datatypes::types::parse_string_to_jsonb(if sequence.is_some() { + r#"{"a":2}"# + } else { + r#"{"a":1}"# + })?, + values.value(count - 1) + ); + assert!(stream.try_next().await?.is_none()); + } + Ok(()) + } + #[tokio::test] async fn test_should_split_flat_batches_for_merge_uses_splittable_file_rows_per_series() { let num_rows = SPLIT_ROW_THRESHOLD * 2; diff --git a/src/mito2/src/read/seq_scan.rs b/src/mito2/src/read/seq_scan.rs index 4b09533e29a..b57622e07e7 100644 --- a/src/mito2/src/read/seq_scan.rs +++ b/src/mito2/src/read/seq_scan.rs @@ -46,8 +46,8 @@ use crate::read::range_cache::{CachedRangeReader, build_range_cache_key}; use crate::read::scan_region::{ScanInput, StreamContext}; use crate::read::scan_util::{ BATCH_SIZE_THRESHOLD, MIN_COMPACTION_SPLIT_ROWS, PartitionMetrics, PartitionMetricsList, - SplitRecordBatchStream, compute_parallel_channel_size, scan_flat_file_ranges, - scan_flat_mem_ranges, should_split_flat_batches_for_merge, + SplitRecordBatchStream, compute_parallel_channel_size, prepare_flat_mem_stream, + scan_flat_file_ranges, scan_flat_mem_ranges, should_split_flat_batches_for_merge, }; use crate::read::stream::{ConvertBatchStream, ScanBatch, ScanBatchStream}; use crate::read::{BoxedRecordBatchStream, ScannerMetrics, scan_util}; @@ -725,7 +725,7 @@ pub(crate) async fn build_flat_sources( *index, range_meta.time_range, ); - ordered_sources[position] = Some(Box::pin(stream) as _); + ordered_sources[position] = Some(prepare_flat_mem_stream(stream_ctx.clone(), stream)); } else if stream_ctx.is_file_range_index(*index) { // Common manifest-level fast-skip shared by SeqScan and UnorderedScan. // Compaction should keep reading its selected input ranges completely. diff --git a/src/mito2/src/read/series_reader.rs b/src/mito2/src/read/series_reader.rs index c030b5b3e7e..021928f2c01 100644 --- a/src/mito2/src/read/series_reader.rs +++ b/src/mito2/src/read/series_reader.rs @@ -36,7 +36,7 @@ use crate::read::scan_region::StreamContext; use crate::read::scan_util::{ PartitionMetrics, SplitRecordBatchStream, compute_average_batch_size, compute_parallel_channel_size, filter_flat_batch_by_sequence, new_filter_metrics, - scan_flat_mem_ranges, should_split_flat_batches_for_merge, + prepare_flat_mem_stream, scan_flat_mem_ranges, should_split_flat_batches_for_merge, }; use crate::read::seq_scan::SeqScan; use crate::read::series_candidate::validate_metric_metadata; @@ -411,12 +411,13 @@ async fn build_uncached_series_partition_range( for index in range_meta.row_group_indices.iter().copied() { if stream_ctx.is_mem_range_index(index) { - let stream = Box::pin(scan_flat_mem_ranges( + let stream = scan_flat_mem_ranges( stream_ctx.clone(), part_metrics.clone(), index, range_meta.time_range, - )); + ); + let stream = prepare_flat_mem_stream(stream_ctx.clone(), stream); sources.push(filter_flat_stream_by_series( stream, codec.clone(), diff --git a/src/mito2/src/read/unordered_scan.rs b/src/mito2/src/read/unordered_scan.rs index 9362898023f..05387be0ce1 100644 --- a/src/mito2/src/read/unordered_scan.rs +++ b/src/mito2/src/read/unordered_scan.rs @@ -37,7 +37,8 @@ use crate::error::{PartitionOutOfRangeSnafu, Result}; use crate::read::pruner::{PartitionPruner, Pruner}; use crate::read::scan_region::{ScanInput, StreamContext}; use crate::read::scan_util::{ - PartitionMetrics, PartitionMetricsList, scan_flat_file_ranges, scan_flat_mem_ranges, + PartitionMetrics, PartitionMetricsList, prepare_flat_mem_stream, scan_flat_file_ranges, + scan_flat_mem_ranges, }; use crate::read::stream::{ConvertBatchStream, ScanBatch, ScanBatchStream}; use crate::read::{ScannerMetrics, scan_util}; @@ -134,6 +135,7 @@ impl UnorderedScan { *index, range_meta.time_range, ); + let stream = prepare_flat_mem_stream(stream_ctx.clone(), stream); for await record_batch in stream { yield record_batch?; } diff --git a/src/mito2/src/sst/parquet/file_range.rs b/src/mito2/src/sst/parquet/file_range.rs index 1c15afd386d..f79569d6d4e 100644 --- a/src/mito2/src/sst/parquet/file_range.rs +++ b/src/mito2/src/sst/parquet/file_range.rs @@ -722,10 +722,20 @@ impl RangeBase { tag_decode_state: &mut TagDecodeState, ) -> Result> { let metadata = self.read_format.metadata(); - // A pruned column rejects the batch without decoding any tag payloads. - if self - .filters - .iter() + // Honor skip_fields/skip_tags before inspecting MaybeFilter::Pruned: a predicate + // marked Pruned is still skipped if it must not run before merge/dedup. + // For example, after adding nullable JSON2 column j to a last_non_null table, + // an old SST may contain (key=a, ts=1, f=10) without j, while a newer row has + // (key=a, ts=1, f=NULL, j={"a":1}). WHERE j IS NOT NULL is Pruned for the old + // SST, but discarding that row here would lose f=10 needed for backfill. + // Keep it for merge/dedup and evaluate the field predicate on the merged row. + // This also applies to added non-JSON2 fields, not only JSON2 columns. + let filters = self.filters.iter().filter(|ctx| { + !(skip_fields && ctx.semantic_type() == SemanticType::Field + || skip_tags && ctx.semantic_type() == SemanticType::Tag) + }); + if filters + .clone() .any(|ctx| matches!(ctx.filter(), MaybeFilter::Pruned)) { return Ok(None); @@ -757,7 +767,7 @@ impl RangeBase { let mut mask = BooleanBuffer::new_set(input.num_rows()); // Run filter one by one and combine them result - for filter_ctx in &self.filters { + for filter_ctx in filters { let filter = match filter_ctx.filter() { MaybeFilter::Filter(f) => f, // Column matches. @@ -766,14 +776,6 @@ impl RangeBase { MaybeFilter::Pruned => return Ok(None), }; - // Skip field filters if skip_fields is true - if skip_fields && filter_ctx.semantic_type() == SemanticType::Field { - continue; - } - if skip_tags && filter_ctx.semantic_type() == SemanticType::Tag { - continue; - } - // Get the column directly by its projected index. // If the column is missing and it's not a tag/time column, this filter is skipped. // Assumes the projection indices align with the input batch schema. @@ -998,6 +1000,14 @@ impl RangeBase { mod tests { use std::sync::Arc; + use super::*; + use crate::read::read_columns::ReadColumns; + use crate::sst::parquet::flat_format::FlatReadFormat; + use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema}; + use crate::test_util::sst_util::{ + new_record_batch_with_custom_sequence, sst_region_metadata, + sst_region_metadata_with_encoding, + }; use datafusion_expr::{col, lit}; use datatypes::arrow::array::{ BinaryDictionaryBuilder, TimestampMillisecondArray, UInt8Array, UInt64Array, @@ -1009,15 +1019,7 @@ mod tests { use mito_codec::row_converter::SparsePrimaryKeyCodec; use parquet::arrow::arrow_reader::RowSelector; use partition::expr::col as partition_col; - - use super::*; - use crate::read::read_columns::ReadColumns; - use crate::sst::parquet::flat_format::FlatReadFormat; - use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema}; - use crate::test_util::sst_util::{ - new_record_batch_with_custom_sequence, sst_region_metadata, - sst_region_metadata_with_encoding, - }; + use store_api::metadata::RegionMetadataBuilder; fn new_test_range_base(filters: Vec) -> RangeBase { let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); @@ -1333,4 +1335,31 @@ mod tests { ]); assert_eq!(actual, expected); } + #[test] + fn test_pruned_missing_field_respects_skip_fields() + -> std::result::Result<(), Box> { + let metadata = Arc::new(sst_region_metadata()); + let mut builder = RegionMetadataBuilder::from_existing(metadata.as_ref().clone()); + builder.push_column_metadata(store_api::metadata::ColumnMetadata { + column_schema: ColumnSchema::new("added", ConcreteDataType::float64_datatype(), true), + semantic_type: SemanticType::Field, + column_id: 100, + }); + let expected = builder.build()?; + let filter = + SimpleFilterContext::new_opt(&metadata, Some(&expected), &col("added").is_not_null()) + .ok_or("expected a missing-column filter")?; + assert!(matches!(filter.filter(), MaybeFilter::Pruned)); + let base = new_test_range_base(vec![filter]); + let batch = new_record_batch_with_custom_sequence(&["b", "x"], 0, 4, 1); + assert!( + base.compute_filter_mask_flat(&batch, false, false, &mut TagDecodeState::new())? + .is_none() + ); + let mask = base + .compute_filter_mask_flat(&batch, true, false, &mut TagDecodeState::new())? + .ok_or("skipped field filter must preserve backfill rows")?; + assert_eq!(batch.num_rows(), mask.count_set_bits()); + Ok(()) + } } diff --git a/src/mito2/src/sst/parquet/json_align.rs b/src/mito2/src/sst/parquet/json_align.rs index 93ecffa072f..c03c7f1d62b 100644 --- a/src/mito2/src/sst/parquet/json_align.rs +++ b/src/mito2/src/sst/parquet/json_align.rs @@ -12,8 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +//! Shared JSON2 conversion for Parquet streams and in-memory query batches. +//! Query types must agree before merge/dedup; compaction Rewrite retains physical layouts. + use std::collections::HashMap; use std::pin::Pin; +use std::sync::Arc; use std::task::{Context, Poll}; use datafusion_common::cast_column; @@ -169,6 +173,45 @@ where } } +/// Aligns existing JSON2 columns to the Arrow types of matching fields in +/// `query_schema`, matching columns by name. +/// Preserves column order and skips columns absent from the batch. +pub(crate) fn align_query_batch( + batch: RecordBatch, + query_schema: &SchemaRef, +) -> Result { + let mut aligned = None; + for target in query_schema + .fields() + .iter() + .filter(|field| is_json2_extension_type(field)) + { + let Ok(index) = batch.schema_ref().index_of(target.name()) else { + continue; + }; + let source = batch.schema_ref().field(index); + if source.data_type() != target.data_type() { + let array = align_array(batch.column(index), source, target)?; + let (fields, columns) = aligned.get_or_insert_with(|| { + ( + batch.schema_ref().fields().to_vec(), + batch.columns().to_vec(), + ) + }); + fields[index] = Arc::new(source.clone().with_data_type(target.data_type().clone())); + columns[index] = array; + } + } + let Some((fields, columns)) = aligned else { + return Ok(batch); + }; + let schema = Arc::new(Schema::new_with_metadata( + fields, + batch.schema().metadata().clone(), + )); + RecordBatch::try_new(schema, columns).context(NewRecordBatchSnafu) +} + fn align_projected_batch( rb: RecordBatch, output_schema: &SchemaRef, @@ -327,6 +370,8 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; + use super::*; + use crate::test_util; use datatypes::arrow::array::{ Array, ArrayRef, BinaryArray, Int64Array, StringArray, StringViewArray, StructArray, }; @@ -337,8 +382,6 @@ mod tests { use datatypes::types::parse_string_to_jsonb; use futures::{StreamExt, stream}; - use super::*; - #[test] fn test_aligner_resolves_json2_rewrite_settings() -> std::result::Result<(), Box> { @@ -955,4 +998,32 @@ mod tests { fn string_array(values: impl IntoIterator) -> ArrayRef { Arc::new(StringArray::from_iter_values(values)) } + #[test] + fn test_query_alignment_preserves_source_columns_until_compatibility() + -> std::result::Result<(), Box> { + let (_, batch) = test_util::sst_util::new_json2_record_batch()?; + let values = int_array([10, 20]); + let mut fields = vec![Arc::new(Field::new("value", DataType::Int64, true))]; + fields.extend(batch.schema_ref().fields().iter().cloned()); + let mut columns = vec![values.clone()]; + columns.extend_from_slice(batch.columns()); + let input = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?; + let target = schema([ + Field::new("new_column", DataType::Utf8, true), + Field::new("new_json", DataType::Binary, true) + .with_extension_type(Json2ExtensionType::default()), + Field::new("j", DataType::Binary, true) + .with_extension_type(Json2ExtensionType::default()), + Field::new("value", DataType::Int64, true), + ]); + let aligned = align_query_batch(input, &target)?; + assert_eq!(batch.num_columns() + 1, aligned.num_columns()); + assert_eq!("value", aligned.schema_ref().field(0).name()); + assert!(Arc::ptr_eq(&values, aligned.column(0))); + assert_eq!("j", aligned.schema_ref().field(1).name()); + assert_eq!(&DataType::Binary, aligned.column(1).data_type()); + let repeated = align_query_batch(aligned.clone(), &target)?; + assert!(Arc::ptr_eq(aligned.column(1), repeated.column(1))); + Ok(()) + } } diff --git a/src/mito2/src/sst/parquet/reader.rs b/src/mito2/src/sst/parquet/reader.rs index 56488669f5a..1c98444946c 100644 --- a/src/mito2/src/sst/parquet/reader.rs +++ b/src/mito2/src/sst/parquet/reader.rs @@ -655,6 +655,14 @@ impl ParquetReaderBuilder { let json2_rewrite_targets = self .json2_rewrite_targets .iter() + // Global compaction targets may include JSON2 columns added after this SST was written. + // Rewrite only projected columns present in this SST; schema compatibility fills + // missing columns with NULL later, without looking them up in this SST's metadata. + .filter(|(column_id, _)| { + read_format + .parquet_projected_index_by_id(**column_id) + .is_some() + }) .map(|(column_id, layout)| { let column = region_meta .column_by_id(*column_id) diff --git a/src/mito2/src/test_util/sst_util.rs b/src/mito2/src/test_util/sst_util.rs index fd6321b0895..84933af2946 100644 --- a/src/mito2/src/test_util/sst_util.rs +++ b/src/mito2/src/test_util/sst_util.rs @@ -24,10 +24,16 @@ use datatypes::arrow::array::{ ArrayRef, BinaryDictionaryBuilder, RecordBatch, StringDictionaryBuilder, TimestampMillisecondArray, UInt8Array, UInt64Array, }; +use datatypes::arrow::array::{Int64Array, StructArray}; use datatypes::arrow::datatypes::UInt32Type; +use datatypes::arrow::datatypes::{DataType, Field}; +use datatypes::extension::json::Json2ExtensionType; +use datatypes::json::JsonSettings; use datatypes::prelude::ConcreteDataType; use datatypes::schema::{ColumnSchema, SkippingIndexOptions}; +use datatypes::types::json_type::JsonNativeType; use datatypes::value::ValueRef; +use datatypes::vectors::json::array::JsonArray; use mito_codec::row_converter::{DensePrimaryKeyCodec, PrimaryKeyCodecExt, SortField}; use object_store::layers::mock::{self, MockLayer, MockLayerBuilder, oio}; use store_api::metadata::{ @@ -44,6 +50,57 @@ use crate::sst::file::{FileHandle, FileMeta}; use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema}; use crate::test_util::{new_batch_builder, new_noop_file_purger}; +/// Builds a current-layout JSON2 flat batch with two timestamps in one series. +pub(crate) fn new_json2_record_batch() +-> Result<(RegionMetadataRef, RecordBatch), Box> { + let legacy: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("a", DataType::Int64, true)), + Arc::new(Int64Array::from_iter_values([1, 2])) as ArrayRef, + )])); + let field = Field::new("j", legacy.data_type().clone(), true) + .with_extension_type(Json2ExtensionType::default()); + let settings = JsonSettings::default(); + let physical = JsonArray::from(&legacy).rewrite_to_v2(&field, &settings, &settings)?; + let mut column_schema = + ColumnSchema::new("j", ConcreteDataType::json2(JsonNativeType::object()), true); + column_schema.with_extension_type(&Json2ExtensionType::default()); + let mut metadata = RegionMetadataBuilder::new(RegionId::new(1024, 0)); + metadata + .push_column_metadata(ColumnMetadata { + column_schema, + semantic_type: SemanticType::Field, + column_id: 0, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 1, + }); + let metadata = Arc::new(metadata.build()?); + let mut options = FlatSchemaOptions::default(); + options + .concretized_json_types + .insert("j".to_string(), physical.data_type().clone()); + let mut keys = BinaryDictionaryBuilder::::new(); + keys.append(b"series")?; + keys.append(b"series")?; + let input = RecordBatch::try_new( + to_flat_sst_arrow_schema(&metadata, &options), + vec![ + physical, + Arc::new(TimestampMillisecondArray::from(vec![1, 2])), + Arc::new(keys.finish()), + Arc::new(UInt64Array::from_value(1, 2)), + Arc::new(UInt8Array::from_value(OpType::Put as u8, 2)), + ], + )?; + Ok((metadata, input)) +} + /// Test region id. const REGION_ID: RegionId = RegionId::new(0, 0); diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index c19be914a73..40a83613a56 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -100,7 +100,6 @@ use sqlparser::ast::{ use store_api::metric_engine_consts::{ LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, }; -use store_api::mito_engine_options::APPEND_MODE_KEY; use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan}; use table::TableRef; use table::dist_table::DistTable; @@ -2632,7 +2631,7 @@ pub fn verify_alter( .build() .context(error::BuildTableMetaSnafu { table_name })?; - validate_json2_columns_append_mode(&new_meta.schema, &new_meta.options)?; + validate_json2_primary_keys(&new_meta.schema, &new_meta.primary_key_indices)?; Ok(true) } @@ -2682,7 +2681,7 @@ pub fn create_table_info( let mut table_options = TableOptions::try_from_iter(&create_table.table_options) .context(UnrecognizedTableOptionSnafu)?; - validate_json2_columns_append_mode(&schema, &table_options)?; + validate_json2_primary_keys(&schema, &primary_key_indices)?; validate_and_normalize_annotations(&mut table_options, &schema, &partition_key_indices)?; @@ -2721,26 +2720,16 @@ pub fn create_table_info( Ok(table_info) } -fn validate_json2_columns_append_mode(schema: &Schema, table_options: &TableOptions) -> Result<()> { - let append_mode = table_options - .extra_options - .get(APPEND_MODE_KEY) - .is_some_and(|value| value == "true"); - - for column in schema.column_schemas() { - if column.data_type.is_json2() { - ensure!( - append_mode, - InvalidSqlSnafu { - err_msg: format!( - "JSON2 column `{}` requires {}='true'", - column.name, APPEND_MODE_KEY - ), - } - ); - } +fn validate_json2_primary_keys(schema: &Schema, primary_key_indices: &[usize]) -> Result<()> { + for &index in primary_key_indices { + let column = &schema.column_schemas()[index]; + ensure!( + !column.data_type.is_json2(), + InvalidSqlSnafu { + err_msg: format!("JSON2 column `{}` cannot be a primary key", column.name), + } + ); } - Ok(()) } @@ -4147,26 +4136,36 @@ WITH ('repartition.column.hint' = ' host ')", } #[test] - fn test_json2_requires_append_mode() { - let cases = [ - "CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX);", - "CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX) WITH (append_mode='false');", - ]; - - for sql in cases { - let expr = create_expr_from_sql(sql); - let err = create_table_info(&expr, vec![]).unwrap_err(); - assert!( - err.to_string() - .contains("JSON2 column `payload` requires append_mode='true'"), - "{err}" - ); + fn test_json2_supports_non_append_mode() -> std::result::Result<(), Box> + { + for options in [ + "", + "WITH (append_mode='false')", + "WITH (append_mode='true')", + "WITH (merge_mode='last_non_null')", + ] { + let expr = create_expr_from_sql(&format!( + "CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX) {options};" + )); + create_table_info(&expr, vec![])?; } + Ok(()) + } + #[test] + fn test_json2_cannot_be_primary_key() -> std::result::Result<(), Box> { let expr = create_expr_from_sql( - "CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX) WITH (append_mode='true');", + "CREATE TABLE monitor (payload JSON2, ts TIMESTAMP TIME INDEX, PRIMARY KEY(payload));", ); - create_table_info(&expr, vec![]).unwrap(); + let err = create_table_info(&expr, vec![]) + .err() + .ok_or("expected JSON2 primary key rejection")?; + assert!( + err.to_string() + .contains("JSON2 column `payload` cannot be a primary key"), + "{err}" + ); + Ok(()) } #[test] diff --git a/src/query/src/datafusion.rs b/src/query/src/datafusion.rs index 4b23f81eb16..2d9ae7a7baf 100644 --- a/src/query/src/datafusion.rs +++ b/src/query/src/datafusion.rs @@ -233,14 +233,47 @@ impl DatafusionQueryEngine { } } WriteOp::Delete => { + // Read-only system tables may have no time index; reject DELETE before key lookup. + ensure!( + !is_readonly_table(&table_name.schema, &table_name.table), + TableReadOnlySnafu { + table: table_name.table.to_string(), + } + ); + let table_info = table.table_info(); + let table_schema = table.schema(); + let ts_column = table_schema.timestamp_column().with_context(|| { + MissingTimestampColumnSnafu { + table_name: table_info.name.clone(), + } + })?; + let key_indices = stream + .schema() + .column_schemas() + .iter() + .enumerate() + .filter(|(_, column)| { + ts_column.name == column.name + || table_info + .meta + .row_key_column_names() + .any(|name| name == &column.name) + }) + .map(|(index, _)| index) + .collect::>(); while let Some(batch) = stream.next().await { - let batch = batch.context(CreateRecordBatchSnafu)?; + // The input plan already evaluated WHERE; storage deletes need only keys/time. + // Project before vector conversion because JSON2's logical Struct is a placeholder. + let batch = batch + .context(CreateRecordBatchSnafu)? + .try_project(&key_indices) + .context(CreateRecordBatchSnafu)?; let column_vectors = batch .column_vectors(&table_name.to_string(), table.schema()) .map_err(BoxedError::new) .context(QueryExecutionSnafu)?; affected_rows += self - .delete(&table_name, &table, column_vectors, query_ctx.clone()) + .delete(&table_name, column_vectors, query_ctx.clone()) .await?; } } @@ -256,36 +289,12 @@ impl DatafusionQueryEngine { async fn delete( &self, table_name: &ResolvedTableReference, - table: &TableRef, column_vectors: HashMap, query_ctx: QueryContextRef, ) -> Result { let catalog_name = table_name.catalog.to_string(); let schema_name = table_name.schema.to_string(); let table_name = table_name.table.to_string(); - let table_schema = table.schema(); - - ensure!( - !is_readonly_table(&schema_name, &table_name), - TableReadOnlySnafu { table: table_name } - ); - - let ts_column = table_schema - .timestamp_column() - .map(|x| &x.name) - .with_context(|| MissingTimestampColumnSnafu { - table_name: table_name.clone(), - })?; - - let table_info = table.table_info(); - let rowkey_columns = table_info - .meta - .row_key_column_names() - .collect::>(); - let column_vectors = column_vectors - .into_iter() - .filter(|x| &x.0 == ts_column || rowkey_columns.contains(&&x.0)) - .collect::>(); let request = DeleteRequest { catalog_name, diff --git a/tests/cases/standalone/common/types/json/json2_limit.result b/tests/cases/standalone/common/types/json/json2_limit.result index 6c362a5cc62..a3e237bf753 100644 --- a/tests/cases/standalone/common/types/json/json2_limit.result +++ b/tests/cases/standalone/common/types/json/json2_limit.result @@ -65,36 +65,6 @@ drop table json2_whole_and_path_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 @@ -104,14 +74,35 @@ create table json2_set_append_mode_false ( Affected Rows: 0 +-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED alter table json2_set_append_mode_false set 'append_mode' = 'false'; -Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` requires append_mode='true' +Error: 1004(InvalidArguments), Invalid region request, region_id: REDACTED, err: Only allow changing append_mode from false to true drop table json2_set_append_mode_false; Affected Rows: 0 +create table json2_primary_key ( + ts timestamp time index, + j json2, + primary key (j) +); + +Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` cannot be a primary key + +create table json2_add_primary_key (ts timestamp time index); + +Affected Rows: 0 + +alter table json2_add_primary_key add column j json2 primary key; + +Error: 1004(InvalidArguments), Invalid SQL, error: JSON2 column `j` cannot be a primary key + +drop table json2_add_primary_key; + +Affected Rows: 0 + -- JSON2 requires flat SST output even when append mode is enabled. create table json2_primary_key_format ( ts timestamp time index, diff --git a/tests/cases/standalone/common/types/json/json2_limit.sql b/tests/cases/standalone/common/types/json/json2_limit.sql index 7b428abb5e0..6a7de6d113b 100644 --- a/tests/cases/standalone/common/types/json/json2_limit.sql +++ b/tests/cases/standalone/common/types/json/json2_limit.sql @@ -38,26 +38,6 @@ order by json_get(j, 'a.b'); drop table json2_whole_and_path_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 @@ -65,10 +45,23 @@ create table json2_set_append_mode_false ( 'append_mode' = 'true' ); +-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED alter table json2_set_append_mode_false set 'append_mode' = 'false'; drop table json2_set_append_mode_false; +create table json2_primary_key ( + ts timestamp time index, + j json2, + primary key (j) +); + +create table json2_add_primary_key (ts timestamp time index); + +alter table json2_add_primary_key add column j json2 primary key; + +drop table json2_add_primary_key; + -- JSON2 requires flat SST output even when append mode is enabled. create table json2_primary_key_format ( ts timestamp time index, diff --git a/tests/cases/standalone/common/types/json/json2_non_append.result b/tests/cases/standalone/common/types/json/json2_non_append.result new file mode 100644 index 00000000000..eed66518477 --- /dev/null +++ b/tests/cases/standalone/common/types/json/json2_non_append.result @@ -0,0 +1,675 @@ +create table json2_without_append_mode ( + ts timestamp time index, + j json2 +); + +Affected Rows: 0 + +drop table json2_without_append_mode; + +Affected Rows: 0 + +create table json2_append_mode_false ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'false' +); + +Affected Rows: 0 + +drop table json2_append_mode_false; + +Affected Rows: 0 + +create table json2_alter_non_append ( + ts timestamp time index +); + +Affected Rows: 0 + +insert into json2_alter_non_append values (1); + +Affected Rows: 1 + +alter table json2_alter_non_append add column j json2; + +Affected Rows: 0 + +insert into json2_alter_non_append values (1, '{"a":1}'); + +Affected Rows: 1 + +select ts, j.a from json2_alter_non_append; + ++-------------------------+------------------------------------------------+ +| ts | json_get(json2_alter_non_append.j,Utf8("$.a")) | ++-------------------------+------------------------------------------------+ +| 1970-01-01T00:00:00.001 | 1 | ++-------------------------+------------------------------------------------+ + +drop table json2_alter_non_append; + +Affected Rows: 0 + +-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls. +CREATE TABLE json2_last_row ( + host STRING, + ts TIMESTAMP TIME INDEX, + j JSON2, + k JSON2(max_auto_expanded_paths = 0), + f DOUBLE, + PRIMARY KEY (host) +) WITH ( + 'append_mode' = 'false', + 'merge_mode' = 'last_row' +); + +Affected Rows: 0 + +INSERT INTO json2_last_row VALUES + ('one', 1, '{"a":1}', '{"a":1}', 10), + ('one', 2, '{"a":2}', '{"a":2}', 10), + ('one', 3, '{"a":3}', '{"a":3}', 10), + ('one', 4, '{"a":4}', '{"a":4}', 10), + ('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10), + ('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10), + ('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10), + ('two', 1, '{"a":99}', '{"a":99}', 99); + +Affected Rows: 8 + +ADMIN FLUSH_TABLE('json2_last_row'); + ++-------------------------------------+ +| ADMIN FLUSH_TABLE('json2_last_row') | ++-------------------------------------+ +| 0 | ++-------------------------------------+ + +-- New objects replace old objects; only SQL NULL can fall back in last_non_null. +INSERT INTO json2_last_row VALUES + ('one', 1, '{"b":2}', '{"b":2}', NULL), + ('one', 2, '{}', '{}', NULL), + ('one', 3, '{"a":null}', '{"a":null}', NULL), + ('one', 4, NULL, NULL, NULL), + ('one', 5, '{"a":42}', '{"a":42}', NULL), + ('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL), + ('one', 7, NULL, NULL, NULL); + +Affected Rows: 7 + +-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback. +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | | +| one | 1970-01-01T00:00:00.004 | | | | | | | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | | +| one | 1970-01-01T00:00:00.007 | | | | | | | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | | | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +SELECT host, ts FROM json2_last_row WHERE j.a = '1' ORDER BY host, ts; + ++------+----+ +| host | ts | ++------+----+ ++------+----+ + +ADMIN FLUSH_TABLE('json2_last_row'); + ++-------------------------------------+ +| ADMIN FLUSH_TABLE('json2_last_row') | ++-------------------------------------+ +| 0 | ++-------------------------------------+ + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | | +| one | 1970-01-01T00:00:00.004 | | | | | | | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | | +| one | 1970-01-01T00:00:00.007 | | | | | | | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | | | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +ADMIN COMPACT_TABLE('json2_last_row'); + ++---------------------------------------+ +| ADMIN COMPACT_TABLE('json2_last_row') | ++---------------------------------------+ +| 0 | ++---------------------------------------+ + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | | +| one | 1970-01-01T00:00:00.004 | | | | | | | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | | +| one | 1970-01-01T00:00:00.007 | | | | | | | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | | | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +-- SQLNESS ARG restart=true +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | | +| one | 1970-01-01T00:00:00.004 | | | | | | | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | | +| one | 1970-01-01T00:00:00.007 | | | | | | | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | | | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +-- A delete must prevent last_non_null from filling fields from older versions. +DELETE FROM json2_last_row WHERE host = 'one' AND ts = 1; + +Affected Rows: 1 + +INSERT INTO json2_last_row VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL); + +Affected Rows: 1 + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | | +| one | 1970-01-01T00:00:00.004 | | | | | | | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | | +| one | 1970-01-01T00:00:00.007 | | | | | | | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +ADMIN FLUSH_TABLE('json2_last_row'); + ++-------------------------------------+ +| ADMIN FLUSH_TABLE('json2_last_row') | ++-------------------------------------+ +| 0 | ++-------------------------------------+ + +ADMIN COMPACT_TABLE('json2_last_row'); + ++---------------------------------------+ +| ADMIN COMPACT_TABLE('json2_last_row') | ++---------------------------------------+ +| 0 | ++---------------------------------------+ + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | | +| one | 1970-01-01T00:00:00.004 | | | | | | | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | | +| one | 1970-01-01T00:00:00.007 | | | | | | | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +DELETE FROM json2_last_row WHERE j.a = '99'; + +Affected Rows: 1 + +SELECT COUNT(*) FROM json2_last_row WHERE host = 'two'; + ++----------+ +| count(*) | ++----------+ +| 0 | ++----------+ + +DROP TABLE json2_last_row; + +Affected Rows: 0 + +-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls. +CREATE TABLE json2_last_non_null ( + host STRING, + ts TIMESTAMP TIME INDEX, + j JSON2, + k JSON2(max_auto_expanded_paths = 0), + f DOUBLE, + PRIMARY KEY (host) +) WITH ( + 'append_mode' = 'false', + 'merge_mode' = 'last_non_null' +); + +Affected Rows: 0 + +INSERT INTO json2_last_non_null VALUES + ('one', 1, '{"a":1}', '{"a":1}', 10), + ('one', 2, '{"a":2}', '{"a":2}', 10), + ('one', 3, '{"a":3}', '{"a":3}', 10), + ('one', 4, '{"a":4}', '{"a":4}', 10), + ('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10), + ('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10), + ('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10), + ('two', 1, '{"a":99}', '{"a":99}', 99); + +Affected Rows: 8 + +ADMIN FLUSH_TABLE('json2_last_non_null'); + ++------------------------------------------+ +| ADMIN FLUSH_TABLE('json2_last_non_null') | ++------------------------------------------+ +| 0 | ++------------------------------------------+ + +-- New objects replace old objects; only SQL NULL can fall back in last_non_null. +INSERT INTO json2_last_non_null VALUES + ('one', 1, '{"b":2}', '{"b":2}', NULL), + ('one', 2, '{}', '{}', NULL), + ('one', 3, '{"a":null}', '{"a":null}', NULL), + ('one', 4, NULL, NULL, NULL), + ('one', 5, '{"a":42}', '{"a":42}', NULL), + ('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL), + ('one', 7, NULL, NULL, NULL); + +Affected Rows: 7 + +-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback. +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 | +| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 | +| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | 8 | 8 | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +SELECT host, ts FROM json2_last_non_null WHERE j.a = '1' ORDER BY host, ts; + ++------+----+ +| host | ts | ++------+----+ ++------+----+ + +ADMIN FLUSH_TABLE('json2_last_non_null'); + ++------------------------------------------+ +| ADMIN FLUSH_TABLE('json2_last_non_null') | ++------------------------------------------+ +| 0 | ++------------------------------------------+ + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 | +| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 | +| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | 8 | 8 | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +ADMIN COMPACT_TABLE('json2_last_non_null'); + ++--------------------------------------------+ +| ADMIN COMPACT_TABLE('json2_last_non_null') | ++--------------------------------------------+ +| 0 | ++--------------------------------------------+ + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 | +| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 | +| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | 8 | 8 | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +-- SQLNESS ARG restart=true +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"b":2} | | | {"b":2} | | 10.0 | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 | +| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 | +| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+----+-----+ +| host | ts | ax | kax | ++------+-------------------------+----+-----+ +| one | 1970-01-01T00:00:00.001 | | | +| one | 1970-01-01T00:00:00.002 | | | +| one | 1970-01-01T00:00:00.003 | | | +| one | 1970-01-01T00:00:00.004 | | | +| one | 1970-01-01T00:00:00.005 | | | +| one | 1970-01-01T00:00:00.006 | 7 | 7 | +| one | 1970-01-01T00:00:00.007 | 8 | 8 | +| two | 1970-01-01T00:00:00.001 | | | ++------+-------------------------+----+-----+ + +-- A delete must prevent last_non_null from filling fields from older versions. +DELETE FROM json2_last_non_null WHERE host = 'one' AND ts = 1; + +Affected Rows: 1 + +INSERT INTO json2_last_non_null VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL); + +Affected Rows: 1 + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 | +| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 | +| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +ADMIN FLUSH_TABLE('json2_last_non_null'); + ++------------------------------------------+ +| ADMIN FLUSH_TABLE('json2_last_non_null') | ++------------------------------------------+ +| 0 | ++------------------------------------------+ + +ADMIN COMPACT_TABLE('json2_last_non_null'); + ++--------------------------------------------+ +| ADMIN COMPACT_TABLE('json2_last_non_null') | ++--------------------------------------------+ +| 0 | ++--------------------------------------------+ + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| host | ts | j | a | ax | k | ka | f | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"c":3} | | | {"c":3} | | | +| one | 1970-01-01T00:00:00.002 | {} | | | {} | | 10.0 | +| one | 1970-01-01T00:00:00.003 | {"a":null} | | | {"a":null} | | 10.0 | +| one | 1970-01-01T00:00:00.004 | {"a":4} | 4 | | {"a":4} | 4 | 10.0 | +| one | 1970-01-01T00:00:00.005 | {"a":42} | 42 | | {"a":42} | 42 | 10.0 | +| one | 1970-01-01T00:00:00.006 | {"a":{"x":7}} | {"x":7} | 7 | {"a":{"x":7}} | {"x":7} | 10.0 | +| one | 1970-01-01T00:00:00.007 | {"a":{"x":8}} | {"x":8} | 8 | {"a":{"x":8}} | {"x":8} | 10.0 | +| two | 1970-01-01T00:00:00.001 | {"a":99} | 99 | | {"a":99} | 99 | 99.0 | ++------+-------------------------+---------------+---------+----+---------------+---------+------+ + +DELETE FROM json2_last_non_null WHERE j.a = '99'; + +Affected Rows: 1 + +SELECT COUNT(*) FROM json2_last_non_null WHERE host = 'two'; + ++----------+ +| count(*) | ++----------+ +| 0 | ++----------+ + +DROP TABLE json2_last_non_null; + +Affected Rows: 0 + +-- A predicate on an added JSON2 field must retain older values of other fields. +CREATE TABLE json2_alter_backfill ( + host STRING, + ts TIMESTAMP TIME INDEX, + f DOUBLE, + PRIMARY KEY (host) +) WITH ('merge_mode' = 'last_non_null'); + +Affected Rows: 0 + +INSERT INTO json2_alter_backfill VALUES ('one', 1, 10); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('json2_alter_backfill'); + ++-------------------------------------------+ +| ADMIN FLUSH_TABLE('json2_alter_backfill') | ++-------------------------------------------+ +| 0 | ++-------------------------------------------+ + +ALTER TABLE json2_alter_backfill ADD COLUMN j JSON2; + +Affected Rows: 0 + +INSERT INTO json2_alter_backfill VALUES ('one', 1, NULL, '{"a":1}'); + +Affected Rows: 1 + +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL; + ++------+-------------------------+---------+------+ +| host | ts | j | f | ++------+-------------------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 | ++------+-------------------------+---------+------+ + +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j.a = '1'; + ++------+-------------------------+---------+------+ +| host | ts | j | f | ++------+-------------------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 | ++------+-------------------------+---------+------+ + +ADMIN FLUSH_TABLE('json2_alter_backfill'); + ++-------------------------------------------+ +| ADMIN FLUSH_TABLE('json2_alter_backfill') | ++-------------------------------------------+ +| 0 | ++-------------------------------------------+ + +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL; + ++------+-------------------------+---------+------+ +| host | ts | j | f | ++------+-------------------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 | ++------+-------------------------+---------+------+ + +ADMIN COMPACT_TABLE('json2_alter_backfill'); + ++---------------------------------------------+ +| ADMIN COMPACT_TABLE('json2_alter_backfill') | ++---------------------------------------------+ +| 0 | ++---------------------------------------------+ + +-- SQLNESS ARG restart=true +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL; + ++------+-------------------------+---------+------+ +| host | ts | j | f | ++------+-------------------------+---------+------+ +| one | 1970-01-01T00:00:00.001 | {"a":1} | 10.0 | ++------+-------------------------+---------+------+ + +DROP TABLE json2_alter_backfill; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/types/json/json2_non_append.sql b/tests/cases/standalone/common/types/json/json2_non_append.sql new file mode 100644 index 00000000000..7d5082ffbea --- /dev/null +++ b/tests/cases/standalone/common/types/json/json2_non_append.sql @@ -0,0 +1,201 @@ +create table json2_without_append_mode ( + ts timestamp time index, + j json2 +); + +drop table json2_without_append_mode; + +create table json2_append_mode_false ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'false' +); + +drop table json2_append_mode_false; + +create table json2_alter_non_append ( + ts timestamp time index +); + +insert into json2_alter_non_append values (1); + +alter table json2_alter_non_append add column j json2; + +insert into json2_alter_non_append values (1, '{"a":1}'); + +select ts, j.a from json2_alter_non_append; + +drop table json2_alter_non_append; + +-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls. +CREATE TABLE json2_last_row ( + host STRING, + ts TIMESTAMP TIME INDEX, + j JSON2, + k JSON2(max_auto_expanded_paths = 0), + f DOUBLE, + PRIMARY KEY (host) +) WITH ( + 'append_mode' = 'false', + 'merge_mode' = 'last_row' +); + +INSERT INTO json2_last_row VALUES + ('one', 1, '{"a":1}', '{"a":1}', 10), + ('one', 2, '{"a":2}', '{"a":2}', 10), + ('one', 3, '{"a":3}', '{"a":3}', 10), + ('one', 4, '{"a":4}', '{"a":4}', 10), + ('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10), + ('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10), + ('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10), + ('two', 1, '{"a":99}', '{"a":99}', 99); + +ADMIN FLUSH_TABLE('json2_last_row'); + +-- New objects replace old objects; only SQL NULL can fall back in last_non_null. +INSERT INTO json2_last_row VALUES + ('one', 1, '{"b":2}', '{"b":2}', NULL), + ('one', 2, '{}', '{}', NULL), + ('one', 3, '{"a":null}', '{"a":null}', NULL), + ('one', 4, NULL, NULL, NULL), + ('one', 5, '{"a":42}', '{"a":42}', NULL), + ('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL), + ('one', 7, NULL, NULL, NULL); + +-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback. +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; +SELECT host, ts FROM json2_last_row WHERE j.a = '1' ORDER BY host, ts; + +ADMIN FLUSH_TABLE('json2_last_row'); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; + +ADMIN COMPACT_TABLE('json2_last_row'); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; + +-- SQLNESS ARG restart=true +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_row ORDER BY host, ts; + +-- A delete must prevent last_non_null from filling fields from older versions. +DELETE FROM json2_last_row WHERE host = 'one' AND ts = 1; +INSERT INTO json2_last_row VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + +ADMIN FLUSH_TABLE('json2_last_row'); +ADMIN COMPACT_TABLE('json2_last_row'); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_row ORDER BY host, ts; + +DELETE FROM json2_last_row WHERE j.a = '99'; +SELECT COUNT(*) FROM json2_last_row WHERE host = 'two'; + +DROP TABLE json2_last_row; + +-- JSON2 is merged as a whole field, including empty objects and explicit JSON nulls. +CREATE TABLE json2_last_non_null ( + host STRING, + ts TIMESTAMP TIME INDEX, + j JSON2, + k JSON2(max_auto_expanded_paths = 0), + f DOUBLE, + PRIMARY KEY (host) +) WITH ( + 'append_mode' = 'false', + 'merge_mode' = 'last_non_null' +); + +INSERT INTO json2_last_non_null VALUES + ('one', 1, '{"a":1}', '{"a":1}', 10), + ('one', 2, '{"a":2}', '{"a":2}', 10), + ('one', 3, '{"a":3}', '{"a":3}', 10), + ('one', 4, '{"a":4}', '{"a":4}', 10), + ('one', 5, '{"a":{"x":5}}', '{"a":{"x":5}}', 10), + ('one', 6, '{"a":{"x":6}}', '{"a":{"x":6}}', 10), + ('one', 7, '{"a":{"x":8}}', '{"a":{"x":8}}', 10), + ('two', 1, '{"a":99}', '{"a":99}', 99); + +ADMIN FLUSH_TABLE('json2_last_non_null'); + +-- New objects replace old objects; only SQL NULL can fall back in last_non_null. +INSERT INTO json2_last_non_null VALUES + ('one', 1, '{"b":2}', '{"b":2}', NULL), + ('one', 2, '{}', '{}', NULL), + ('one', 3, '{"a":null}', '{"a":null}', NULL), + ('one', 4, NULL, NULL, NULL), + ('one', 5, '{"a":42}', '{"a":42}', NULL), + ('one', 6, '{"a":{"x":7}}', '{"a":{"x":7}}', NULL), + ('one', 7, NULL, NULL, NULL); + +-- Path-only reads must keep the new nested value and distinguish SQL NULL fallback. +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; +SELECT host, ts FROM json2_last_non_null WHERE j.a = '1' ORDER BY host, ts; + +ADMIN FLUSH_TABLE('json2_last_non_null'); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; + +ADMIN COMPACT_TABLE('json2_last_non_null'); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; + +-- SQLNESS ARG restart=true +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; +SELECT host, ts, j.a.x AS ax, k.a.x AS kax FROM json2_last_non_null ORDER BY host, ts; + +-- A delete must prevent last_non_null from filling fields from older versions. +DELETE FROM json2_last_non_null WHERE host = 'one' AND ts = 1; +INSERT INTO json2_last_non_null VALUES ('one', 1, '{"c":3}', '{"c":3}', NULL); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + +ADMIN FLUSH_TABLE('json2_last_non_null'); +ADMIN COMPACT_TABLE('json2_last_non_null'); + +SELECT host, ts, j, j.a AS a, j.a.x AS ax, k, k.a AS ka, f +FROM json2_last_non_null ORDER BY host, ts; + +DELETE FROM json2_last_non_null WHERE j.a = '99'; +SELECT COUNT(*) FROM json2_last_non_null WHERE host = 'two'; + +DROP TABLE json2_last_non_null; + +-- A predicate on an added JSON2 field must retain older values of other fields. +CREATE TABLE json2_alter_backfill ( + host STRING, + ts TIMESTAMP TIME INDEX, + f DOUBLE, + PRIMARY KEY (host) +) WITH ('merge_mode' = 'last_non_null'); +INSERT INTO json2_alter_backfill VALUES ('one', 1, 10); +ADMIN FLUSH_TABLE('json2_alter_backfill'); +ALTER TABLE json2_alter_backfill ADD COLUMN j JSON2; +INSERT INTO json2_alter_backfill VALUES ('one', 1, NULL, '{"a":1}'); +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL; +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j.a = '1'; +ADMIN FLUSH_TABLE('json2_alter_backfill'); +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL; +ADMIN COMPACT_TABLE('json2_alter_backfill'); +-- SQLNESS ARG restart=true +SELECT host, ts, j, f FROM json2_alter_backfill WHERE j IS NOT NULL; +DROP TABLE json2_alter_backfill;