From 2f7d0bb674d62231ea247126ff402df04b29ed09 Mon Sep 17 00:00:00 2001 From: shuiyisong <113876041+shuiyisong@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:54:04 +0800 Subject: [PATCH] fix(mito2): require flat SST output for JSON2 (#9893) (#9911) * fix(mito2): require flat SST output for JSON2 * fix(mito2): avoid contradictory JSON2 recovery warnings --------- Signed-off-by: shuiyisong --- src/mito2/src/engine/alter_test.rs | 190 ++++++++ src/mito2/src/engine/create_test.rs | 442 ++++++++++++++++-- src/mito2/src/memtable.rs | 31 +- src/mito2/src/region/opener.rs | 79 ++-- src/mito2/src/worker/handle_alter.rs | 7 +- .../common/types/json/json2_limit.result | 108 +++++ .../common/types/json/json2_limit.sql | 44 ++ 7 files changed, 826 insertions(+), 75 deletions(-) diff --git a/src/mito2/src/engine/alter_test.rs b/src/mito2/src/engine/alter_test.rs index cf039b52278..9ff56d0ddab 100644 --- a/src/mito2/src/engine/alter_test.rs +++ b/src/mito2/src/engine/alter_test.rs @@ -17,6 +17,7 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; +use api::helper::encode_json_value; use api::v1::bulk_wal_entry::Body; use api::v1::helper::{row, tag_column_schema}; use api::v1::region::{StrictWindow, compact_request}; @@ -33,8 +34,11 @@ use datafusion_common::ScalarValue; use datafusion_expr::{col, lit}; use datatypes::arrow::array::{ArrayRef, Float64Array, StringArray, TimestampMillisecondArray}; use datatypes::arrow::datatypes::{DataType, Field, Schema, TimeUnit}; +use datatypes::extension::json::Json2ExtensionType; +use datatypes::json::value::JsonValue; use datatypes::prelude::ConcreteDataType; use datatypes::schema::{ColumnSchema, FulltextAnalyzer, FulltextBackend, FulltextOptions}; +use datatypes::types::json_type::{JsonNativeType, JsonObjectType}; use store_api::codec::PrimaryKeyEncoding; use store_api::logstore::provider::Provider; use store_api::metadata::{ColumnMetadata, RegionMetadata}; @@ -137,6 +141,192 @@ fn add_nullable_field1() -> RegionAlterRequest { } } +#[rstest::rstest] +#[case::implicit_time_series(false, None, false)] +#[case::explicit_time_series(false, Some("time_series"), false)] +#[case::explicit_bulk(false, Some("bulk"), true)] +#[case::default_flat(true, None, true)] +#[case::flat_overrides_time_series(true, Some("time_series"), true)] +#[tokio::test] +async fn test_engine_add_json2_format_validation( + #[case] default_flat_format: bool, + #[case] memtable: Option<&str>, + #[case] accepted: bool, + #[values(false, true)] write_before_alter: bool, +) { + let mut env = TestEnv::with_prefix("add-json2-memtable-validation").await; + let engine = env + .create_engine(MitoConfig { + default_flat_format, + ..Default::default() + }) + .await; + let region_id = RegionId::new(1, 1); + let mut request = CreateRequestBuilder::new().build(); + if let Some(memtable) = memtable { + request + .options + .insert("memtable.type".into(), memtable.into()); + } + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await + .unwrap(); + if write_before_alter { + put_rows( + &engine, + region_id, + Rows { + schema: rows_schema(&request), + rows: build_rows(0, 1), + }, + ) + .await; + } + let region = engine.get_region(region_id).unwrap(); + let old_metadata = region.metadata(); + let mut column_schema = ColumnSchema::new( + "json", + ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())), + true, + ); + column_schema.with_extension_type(&Json2ExtensionType::default()); + let result = engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::AddColumns { + columns: vec![AddColumn { + column_metadata: ColumnMetadata { + column_schema, + semantic_type: SemanticType::Field, + column_id: 3, + }, + location: None, + }], + }, + }), + ) + .await; + if accepted { + result.unwrap(); + assert!(region.metadata().column_by_name("json").is_some()); + } else { + let err = result.unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + assert!( + err.output_msg() + .contains("JSON2 columns require sst_format=flat") + ); + assert_eq!(old_metadata, region.metadata()); + } + let expected_metadata = region.metadata(); + assert_eq!( + expected_metadata, + region.manifest_ctx.manifest().await.metadata + ); + reopen_region(&engine, region_id, request.table_dir, true, request.options).await; + assert_eq!( + expected_metadata, + engine.get_region(region_id).unwrap().metadata() + ); +} + +#[rstest::rstest] +#[tokio::test] +async fn test_engine_rejects_json2_primary_key_format_change( + #[values("bulk", "time_series")] memtable: &str, + #[values(false, true)] write_before_alter: bool, +) { + let mut env = TestEnv::with_prefix("alter-json2-format-validation").await; + let engine = env.create_engine(MitoConfig::default()).await; + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .field_datatype(ConcreteDataType::json2(JsonNativeType::Object( + JsonObjectType::new(), + ))) + .insert_option("memtable.type", memtable) + .build(); + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await + .unwrap(); + let json_rows = |n: i64| Rows { + schema: rows_schema(&request), + rows: vec![row(vec![ + ValueData::StringValue("tag".into()), + ValueData::JsonValue(encode_json_value(JsonValue::from([("a", n)]))), + ValueData::TimestampMillisecondValue(n * 1000), + ])], + }; + if write_before_alter { + put_rows(&engine, region_id, json_rows(1)).await; + } + let region = engine.get_region(region_id).unwrap(); + let manifest_version = region.manifest_ctx.manifest_version().await; + let err = engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::SetRegionOptions { + options: vec![SetRegionOption::Format("primary_key".into())], + }, + }), + ) + .await + .unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + assert!( + err.output_msg() + .contains("JSON2 columns require sst_format=flat") + ); + assert_eq!(Some(FormatType::Flat), region.version().options.sst_format); + assert_eq!( + FormatType::Flat, + region.manifest_ctx.manifest().await.sst_format + ); + if !write_before_alter { + assert_eq!( + manifest_version, + region.manifest_ctx.manifest_version().await + ); + } + put_rows(&engine, region_id, json_rows(2)).await; + let batches = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + usize::from(write_before_alter) + 1, + batches.iter().map(|b| b.num_rows()).sum::() + ); + let expected = batches.pretty_print().unwrap(); + flush_region(&engine, region_id, None).await; + reopen_region(&engine, region_id, request.table_dir, true, request.options).await; + assert_eq!( + Some(FormatType::Flat), + engine + .get_region(region_id) + .unwrap() + .version() + .options + .sst_format + ); + let batches = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(expected, batches.pretty_print().unwrap()); +} + fn build_row_with_added_field( metadata: &[ColumnMetadata], tag_0: &str, diff --git a/src/mito2/src/engine/create_test.rs b/src/mito2/src/engine/create_test.rs index 2c04e2e24f1..5734f64f8bb 100644 --- a/src/mito2/src/engine/create_test.rs +++ b/src/mito2/src/engine/create_test.rs @@ -12,20 +12,39 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::assert_matches; +use std::sync::Arc; use std::time::Duration; -use api::v1::Rows; +use api::helper::encode_json_value; +use api::v1::helper::row; +use api::v1::region::{StrictWindow, compact_request}; +use api::v1::value::ValueData; +use api::v1::{Rows, WriteHint}; use common_error::ext::{ErrorExt, WhateverResult}; use common_error::status_code::StatusCode; use common_recordbatch::RecordBatches; +use datatypes::json::value::JsonValue; use datatypes::prelude::ConcreteDataType; use datatypes::types::json_type::{JsonNativeType, JsonObjectType}; -use store_api::region_engine::RegionEngine; -use store_api::region_request::{PathType, RegionCloseRequest, RegionOpenRequest, RegionRequest}; +use store_api::codec::PrimaryKeyEncoding; +use store_api::region_engine::{RegionEngine, RegionManifestInfo, RegionRole}; +use store_api::region_request::{ + RegionCloseRequest, RegionCompactRequest, RegionOpenRequest, RegionPutRequest, RegionRequest, +}; +use store_api::storage::consts::PRIMARY_KEY_COLUMN_NAME; use store_api::storage::{RegionId, ScanRequest}; use crate::config::MitoConfig; +use crate::engine::MitoEngine; +use crate::error::Error; +use crate::manifest::action::{RegionChange, RegionMetaAction, RegionMetaActionList}; +use crate::manifest::storage::{delta_file, manifest_dir}; use crate::region::options::MemtableOptions; +use crate::sst::FormatType; +use crate::sst::location::region_dir_from_table_dir; +use crate::test_util::batch_util::sort_batches_and_print; +use crate::test_util::sst_util::{new_sparse_primary_key, sst_region_metadata_with_encoding}; use crate::test_util::{ CreateRequestBuilder, TestEnv, build_rows, flush_region, put_rows, reopen_region, rows_schema, }; @@ -36,36 +55,89 @@ async fn test_engine_create_new_region() { test_engine_create_new_region_with_format(true).await; } +#[rstest::rstest] +#[case::implicit_time_series(false, None, None, false)] +#[case::explicit_time_series(false, Some("time_series"), None, false)] +#[case::explicit_bulk(false, Some("bulk"), None, true)] +#[case::flat_overrides_time_series(false, Some("time_series"), Some("flat"), true)] +#[case::default_flat(true, None, None, true)] +#[case::default_flat_overrides_time_series(true, Some("time_series"), None, true)] +#[case::primary_key_overrides_default_flat(true, None, Some("primary_key"), false)] #[tokio::test] -async fn test_engine_rejects_json2_with_time_series_memtable_on_create_and_open() --> WhateverResult<()> { - let mut env = TestEnv::with_prefix("json2-rejects-time-series-memtable").await; +async fn test_engine_create_json2_format_validation( + #[case] default_flat_format: bool, + #[case] memtable: Option<&str>, + #[case] sst_format: Option<&str>, + #[case] accepted: bool, +) -> WhateverResult<()> { + let mut env = TestEnv::with_prefix("create-json2-memtable-validation").await; let engine = env .create_engine(MitoConfig { - default_flat_format: false, + default_flat_format, ..Default::default() }) .await; let region_id = RegionId::new(1, 1); - let request = CreateRequestBuilder::new() + let mut request = CreateRequestBuilder::new() .field_datatype(ConcreteDataType::json2(JsonNativeType::Object( JsonObjectType::new(), ))) .insert_option("append_mode", "true") - .insert_option("memtable.type", "time_series") .build(); + if let Some(memtable) = memtable { + request + .options + .insert("memtable.type".into(), memtable.into()); + } + if let Some(sst_format) = sst_format { + request + .options + .insert("sst_format".into(), sst_format.into()); + } - let err = engine - .handle_request(region_id, RegionRequest::Create(request)) - .await - .unwrap_err(); - assert_eq!(StatusCode::InvalidArguments, err.status_code()); - assert!( - err.to_string() - .contains("JSON2 columns only support BulkMemtable") - ); + let result = engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await; + if accepted { + result?; + } else { + let err = result.unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + assert!( + err.output_msg() + .contains("JSON2 columns require sst_format=flat") + ); + assert!(engine.get_region(region_id).is_none()); + request + .options + .insert("memtable.type".into(), "bulk".into()); + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await?; + } + reopen_region(&engine, region_id, request.table_dir, true, request.options).await; + Ok(()) +} - let request = CreateRequestBuilder::new() +#[rstest::rstest] +#[case::implicit_primary_key(false, false)] +#[case::default_flat(true, false)] +#[case::explicit_primary_key(false, true)] +#[tokio::test] +async fn test_engine_recover_json2_region( + #[case] default_flat_format: bool, + #[case] explicit_primary_key: bool, + #[values(false, true)] use_create: bool, + #[values(FormatType::PrimaryKey, FormatType::Flat)] persisted_format: FormatType, +) -> WhateverResult<()> { + let mut env = TestEnv::with_prefix("recover-legacy-json2").await; + let config = MitoConfig { + default_flat_format, + ..Default::default() + }; + let engine = env.create_engine(config.clone()).await; + let region_id = RegionId::new(1, 1); + let mut request = CreateRequestBuilder::new() .field_datatype(ConcreteDataType::json2(JsonNativeType::Object( JsonObjectType::new(), ))) @@ -73,8 +145,208 @@ async fn test_engine_rejects_json2_with_time_series_memtable_on_create_and_open( .insert_option("memtable.type", "bulk") .build(); engine - .handle_request(region_id, RegionRequest::Create(request)) + .handle_request(region_id, RegionRequest::Create(request.clone())) .await?; + let schema = rows_schema(&request); + let json_row = |n: i64| { + row(vec![ + ValueData::StringValue("tag".into()), + ValueData::JsonValue(encode_json_value(JsonValue::from([("a", n)]))), + ValueData::TimestampMillisecondValue(n * 1000), + ]) + }; + for n in 1..=2 { + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: vec![json_row(n)], + }, + ) + .await; + flush_region(&engine, region_id, None).await; + } + // Leave one row in the WAL to exercise replay into the recovered Bulk memtable. + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: vec![json_row(3)], + }, + ) + .await; + let expected = scan_region_data(&engine, region_id).await; + + request.options.remove("memtable.type"); + if explicit_primary_key { + request + .options + .insert("sst_format".into(), "primary_key".into()); + request + .options + .insert("memtable.type".into(), "time_series".into()); + } + let open_request = RegionOpenRequest { + engine: request.engine.clone(), + table_dir: request.table_dir.clone(), + path_type: request.path_type, + options: request.options.clone(), + skip_wal_replay: false, + checkpoint: None, + requirements: Default::default(), + }; + let follower = env.create_follower_engine(config).await; + follower + .handle_request( + region_id, + RegionRequest::Open(RegionOpenRequest { + skip_wal_replay: true, + ..open_request.clone() + }), + ) + .await?; + + let region = engine.get_region(region_id).unwrap(); + // Reproduce the persisted state from older binaries, bypassing the new ALTER validation. + region + .manifest_ctx + .manifest_manager + .write() + .await + .update( + RegionMetaActionList::new(vec![RegionMetaAction::Change(RegionChange { + metadata: region.metadata(), + sst_format: persisted_format, + append_mode: Some(true), + })]), + false, + ) + .await?; + let manifest = region.manifest_ctx.manifest().await; + assert_eq!(2, manifest.files.len()); + assert_eq!(persisted_format, manifest.sst_format); + follower + .sync_region( + region_id, + RegionManifestInfo::mito(manifest.manifest_version, 0, 0).into(), + ) + .await?; + let follower_region = follower.get_region(region_id).unwrap(); + assert_eq!( + Some(FormatType::Flat), + follower_region.version().options.sst_format + ); + assert_eq!( + manifest.files, + follower_region.manifest_ctx.manifest().await.files + ); + follower.stop().await?; + + engine + .handle_request( + region_id, + RegionRequest::Close(RegionCloseRequest::default()), + ) + .await?; + let recovery = if use_create { + RegionRequest::Create(request.clone()) + } else { + RegionRequest::Open(open_request) + }; + engine.handle_request(region_id, recovery).await?; + let region = engine.get_region(region_id).unwrap(); + let recovered_manifest = region.manifest_ctx.manifest().await; + assert_eq!( + manifest.manifest_version, + recovered_manifest.manifest_version + ); + assert_eq!(manifest.metadata, recovered_manifest.metadata); + assert_eq!(manifest.files, recovered_manifest.files); + assert_eq!(persisted_format, recovered_manifest.sst_format); + assert_eq!(Some(FormatType::Flat), region.version().options.sst_format); + assert_eq!(expected, scan_region_data(&engine, region_id).await); + + engine + .set_region_role(region_id, RegionRole::Leader) + .unwrap(); + put_rows( + &engine, + region_id, + Rows { + schema, + rows: vec![json_row(4)], + }, + ) + .await; + let expected = scan_region_data(&engine, region_id).await; + flush_region(&engine, region_id, None).await; + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest { + options: compact_request::Options::StrictWindow(StrictWindow { + window_seconds: 3600, + }), + ..Default::default() + }), + ) + .await?; + assert_eq!(1, region.manifest_ctx.manifest().await.files.len()); + assert_eq!(expected, scan_region_data(&engine, region_id).await); + reopen_region( + &engine, + region_id, + request.table_dir, + false, + request.options, + ) + .await; + assert_eq!(expected, scan_region_data(&engine, region_id).await); + Ok(()) +} + +async fn scan_region_data(engine: &MitoEngine, region_id: RegionId) -> String { + let stream = engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + sort_batches_and_print(&batches, &["ts"]) +} + +#[tokio::test] +async fn test_engine_create_preserves_manifest_on_open_error() -> WhateverResult<()> { + let mut env = TestEnv::with_prefix("create-open-error").await; + let engine = env + .create_engine(MitoConfig { + compress_manifest: false, + ..Default::default() + }) + .await; + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new().build(); + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await?; + put_rows( + &engine, + region_id, + Rows { + schema: rows_schema(&request), + rows: build_rows(0, 3), + }, + ) + .await; + flush_region(&engine, region_id, None).await; + let expected = scan_region_data(&engine, region_id).await; + let manifest = engine + .get_region(region_id) + .unwrap() + .manifest_ctx + .manifest() + .await; engine .handle_request( region_id, @@ -82,32 +354,124 @@ async fn test_engine_rejects_json2_with_time_series_memtable_on_create_and_open( ) .await?; - let options = [ - ("append_mode".to_string(), "true".to_string()), - ("memtable.type".to_string(), "time_series".to_string()), - ] - .into_iter() - .collect(); + let store = env.get_object_store().unwrap(); + let dir = manifest_dir(®ion_dir_from_table_dir( + &request.table_dir, + region_id, + request.path_type, + )); + let path = format!("{dir}{}", delta_file(0)); + let original = store.read(&path).await.unwrap().to_bytes(); + store.write(&path, "invalid manifest").await.unwrap(); + let mut manifest_bytes = Vec::new(); + for entry in store.list(&dir).await.unwrap() { + if entry.metadata().is_file() { + manifest_bytes.push(( + entry.path().to_string(), + store.read(entry.path()).await.unwrap().to_bytes(), + )); + } + } let err = engine - .handle_request( - region_id, - RegionRequest::Open(RegionOpenRequest { - engine: String::new(), - table_dir: "test".to_string(), - path_type: PathType::Bare, - options, - skip_wal_replay: false, - checkpoint: None, - requirements: Default::default(), - }), - ) + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await + .unwrap_err(); + assert_matches!( + err.as_any().downcast_ref::().unwrap(), + Error::SerdeJson { .. } + ); + assert!(!engine.is_region_exists(region_id)); + for (path, bytes) in manifest_bytes { + assert_eq!(bytes, store.read(&path).await.unwrap().to_bytes()); + } + + store.write(&path, original).await.unwrap(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await?; + let recovered = engine + .get_region(region_id) + .unwrap() + .manifest_ctx + .manifest() + .await; + assert_eq!(manifest.manifest_version, recovered.manifest_version); + assert_eq!(manifest.files, recovered.files); + assert_eq!(expected, scan_region_data(&engine, region_id).await); + Ok(()) +} + +#[tokio::test] +async fn test_engine_create_json2_sparse_requires_flat() -> WhateverResult<()> { + let mut env = TestEnv::with_prefix("create-json2-sparse").await; + let engine = env.create_engine(MitoConfig::default()).await; + let region_id = RegionId::new(1, 1); + let mut request = CreateRequestBuilder::new() + .field_datatype(ConcreteDataType::json2(JsonNativeType::Object( + JsonObjectType::new(), + ))) + .insert_option("memtable.type", "time_series") + .insert_option("primary_key_encoding", "sparse") + .insert_option("sst_format", "primary_key") + .build(); + let json_column = request.column_metadatas[1].column_schema.clone(); + let metadata = Arc::new(sst_region_metadata_with_encoding( + PrimaryKeyEncoding::Sparse, + )); + request.column_metadatas = metadata.column_metadatas.clone(); + request.primary_key = metadata.primary_key.clone(); + request.column_metadatas[4].column_schema = json_column; + let err = engine + .handle_request(region_id, RegionRequest::Create(request.clone())) .await .unwrap_err(); assert_eq!(StatusCode::InvalidArguments, err.status_code()); assert!( err.output_msg() - .contains("JSON2 columns only support BulkMemtable") + .contains("JSON2 columns require sst_format=flat") ); + assert!(!engine.is_region_exists(region_id)); + + request.options.insert("sst_format".into(), "flat".into()); + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await?; + let schema = rows_schema(&request); + let mut pk_schema = schema[0].clone(); + pk_schema.column_name = PRIMARY_KEY_COLUMN_NAME.into(); + pk_schema.datatype = api::v1::ColumnDataType::Binary.into(); + let rows = Rows { + schema: vec![pk_schema, schema[5].clone(), schema[4].clone()], + rows: vec![row(vec![ + ValueData::BinaryValue(new_sparse_primary_key(&["a", "x"], &metadata, 10, 0)), + ValueData::TimestampMillisecondValue(1000), + ValueData::JsonValue(encode_json_value(JsonValue::from([("a", 1i64)]))), + ])], + }; + engine + .handle_request( + region_id, + RegionRequest::Put(RegionPutRequest { + rows, + hint: Some(WriteHint { + primary_key_encoding: api::v1::PrimaryKeyEncoding::Sparse.into(), + }), + skip_wal: false, + partition_expr_version: None, + }), + ) + .await?; + let expected = scan_region_data(&engine, region_id).await; + flush_region(&engine, region_id, None).await; + reopen_region( + &engine, + region_id, + request.table_dir, + false, + request.options, + ) + .await; + assert_eq!(expected, scan_region_data(&engine, region_id).await); Ok(()) } diff --git a/src/mito2/src/memtable.rs b/src/mito2/src/memtable.rs index afb546340b1..691ba1c5623 100644 --- a/src/mito2/src/memtable.rs +++ b/src/mito2/src/memtable.rs @@ -437,8 +437,8 @@ pub(crate) struct MemtableBuilderProvider { compact_dispatcher: Arc, } -/// Ensures JSON2 columns are not used with [`TimeSeriesMemtable`]. -pub(crate) fn ensure_json2_not_use_time_series_memtable( +/// Ensures JSON2 columns use flat SST output after resolving the SST format. +pub(crate) fn ensure_json2_uses_flat_format( metadata: &RegionMetadata, options: &RegionOptions, ) -> Result<()> { @@ -447,10 +447,11 @@ pub(crate) fn ensure_json2_not_use_time_series_memtable( .iter() .any(|x| x.column_schema.data_type.is_json2()) { + // Bulk memtables alone do not ensure the SST writer supports JSON2. ensure!( - !matches!(&options.memtable, Some(MemtableOptions::TimeSeries)), + options.sst_format == Some(FormatType::Flat), InvalidRegionOptionsSnafu { - reason: "JSON2 columns only support BulkMemtable", + reason: "JSON2 columns require sst_format=flat", } ); } @@ -994,7 +995,7 @@ mod tests { } #[test] - fn test_json2_requires_bulk_memtable() -> WhateverResult<()> { + fn test_json2_requires_flat_format() -> WhateverResult<()> { let mut metadata = sst_region_metadata(); metadata.column_metadatas[2].column_schema.data_type = ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())); @@ -1005,14 +1006,28 @@ mod tests { ..Default::default() }; - let err = ensure_json2_not_use_time_series_memtable(&metadata, &options).unwrap_err(); + let err = ensure_json2_uses_flat_format(&metadata, &options).unwrap_err(); assert!( err.to_string() - .contains("JSON2 columns only support BulkMemtable") + .contains("JSON2 columns require sst_format=flat") ); + options.memtable = None; + assert!(ensure_json2_uses_flat_format(&metadata, &options).is_err()); + options.memtable = Some(MemtableOptions::Bulk(BulkMemtableConfig::default())); - ensure_json2_not_use_time_series_memtable(&metadata, &options)?; + assert!(ensure_json2_uses_flat_format(&metadata, &options).is_err()); + + options.memtable = Some(MemtableOptions::TimeSeries); + options.sst_format = Some(FormatType::Flat); + ensure_json2_uses_flat_format(&metadata, &options)?; + + options.sst_format = Some(FormatType::PrimaryKey); + options.primary_key_encoding = Some(PrimaryKeyEncoding::Sparse); + assert!(ensure_json2_uses_flat_format(&metadata, &options).is_err()); + + options.sst_format = Some(FormatType::Flat); + ensure_json2_uses_flat_format(&metadata, &options)?; Ok(()) } } diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs index 71c1192cbf2..e5b09ef1810 100644 --- a/src/mito2/src/region/opener.rs +++ b/src/mito2/src/region/opener.rs @@ -59,7 +59,7 @@ use crate::manifest::action::RegionManifest; use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions}; use crate::memtable::bulk::part::BulkPart; use crate::memtable::time_partition::TimePartitions; -use crate::memtable::{MemtableBuilderProvider, ensure_json2_not_use_time_series_memtable}; +use crate::memtable::{MemtableBuilderProvider, ensure_json2_uses_flat_format}; use crate::metrics::{CACHE_FILL_DOWNLOADED_FILES, CACHE_FILL_PENDING_FILES}; use crate::read::series_candidate::is_sparse_metric_metadata; use crate::region::options::RegionOptions; @@ -337,9 +337,9 @@ impl RegionOpener { let region_id = self.region_id; let region_dir = self.region_dir(); let metadata = self.build_metadata()?; - // Tries to open the region. - match self.maybe_open(config, wal).await { - Ok(Some(region)) => { + // Only an absent region may be created; an open error must not replace its manifest. + match self.maybe_open(config, wal).await? { + Some(region) => { let recovered = region.metadata(); // Checks the schema of the region. let expect = &metadata; @@ -354,18 +354,12 @@ impl RegionOpener { return Ok(region); } - Ok(None) => { + None => { debug!( "No data under directory {}, region_id: {}", region_dir, self.region_id ); } - Err(e) => { - warn!(e; - "Failed to open region {} before creating it, region_dir: {}", - self.region_id, region_dir - ); - } } // Safety: must be set before calling this method. let mut options = self.options.take().unwrap(); @@ -383,7 +377,7 @@ impl RegionOpener { options.sst_format = Some(FormatType::PrimaryKey); FormatType::PrimaryKey }; - ensure_json2_not_use_time_series_memtable(&metadata, &options)?; + ensure_json2_uses_flat_format(&metadata, &options)?; // Create a manifest manager for this region and writes regions to the manifest file. let mut region_manifest_options = RegionManifestOptions::new(config, ®ion_dir, &object_store); @@ -538,7 +532,7 @@ impl RegionOpener { let metadata = maybe_upgrade_json2_layout(metadata)?; // Updates the region options with the manifest. sanitize_region_options(&manifest, &mut region_options); - ensure_json2_not_use_time_series_memtable(&metadata, ®ion_options)?; + ensure_json2_uses_flat_format(&metadata, ®ion_options)?; let region_id = self.region_id; let provider = self.provider::(®ion_options.wal_options)?; @@ -791,23 +785,29 @@ pub(crate) fn version_builder_from_manifest( /// Updates region options by persistent options. pub(crate) fn sanitize_region_options(manifest: &RegionManifest, options: &mut RegionOptions) { - // The caller-supplied options win when they specify an explicit `sst_format`, - // so re-parsed options (e.g. bulk memtable forcing flat format) take effect on - // the running region instead of being clobbered by the persisted manifest value. - // Fall back to the manifest only when the caller did not provide a value. - match options.sst_format { - Some(format) if format != manifest.sst_format => { - common_telemetry::warn!( + // Caller options take precedence over the manifest, except JSON2 always requires flat output. + let format = options.sst_format.unwrap_or(manifest.sst_format); + // Older versions allowed JSON2 with primary-key output; recover without rewriting its history. + if format != FormatType::Flat + && manifest + .metadata + .column_metadatas + .iter() + .any(|column| column.column_schema.data_type.is_json2()) + { + warn!( + "Overriding SST format from {:?} to flat for existing JSON2 region {}", + format, manifest.metadata.region_id, + ); + options.sst_format = Some(FormatType::Flat); + } else { + if format != manifest.sst_format { + warn!( "Overriding SST format from {:?} (manifest) to {:?} (options) for region {}", - manifest.sst_format, - format, - manifest.metadata.region_id, + manifest.sst_format, format, manifest.metadata.region_id, ); } - Some(_) => {} - None => { - options.sst_format = Some(manifest.sst_format); - } + options.sst_format = Some(format); } if let Some(manifest_append_mode) = manifest.append_mode && options.append_mode != manifest_append_mode @@ -1624,6 +1624,31 @@ mod tests { assert_eq!(options.sst_format, Some(FormatType::Flat)); } + #[rstest::rstest] + fn test_sanitize_region_options_recovers_json2( + #[values(FormatType::PrimaryKey, FormatType::Flat)] persisted: FormatType, + #[values(None, Some(FormatType::PrimaryKey), Some(FormatType::Flat))] requested: Option< + FormatType, + >, + ) { + let mut manifest = build_test_manifest(persisted); + let mut metadata = manifest.metadata.as_ref().clone(); + metadata.column_metadatas[2].column_schema.data_type = + ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())); + manifest.metadata = Arc::new( + RegionMetadataBuilder::from_existing(metadata) + .build() + .unwrap(), + ); + let mut options = RegionOptions { + sst_format: requested, + ..Default::default() + }; + sanitize_region_options(&manifest, &mut options); + assert_eq!(Some(FormatType::Flat), options.sst_format); + assert_eq!(persisted, manifest.sst_format); + } + fn sst_parquet_bytes(batch: &RecordBatch) -> Vec { let key_value_meta = KeyValue::new( PARQUET_METADATA_KEY.to_string(), diff --git a/src/mito2/src/worker/handle_alter.rs b/src/mito2/src/worker/handle_alter.rs index 66cb2dd95bb..c413934b716 100644 --- a/src/mito2/src/worker/handle_alter.rs +++ b/src/mito2/src/worker/handle_alter.rs @@ -36,6 +36,7 @@ use store_api::storage::RegionId; use crate::error::{InvalidMetadataSnafu, InvalidRegionRequestSnafu, Result}; use crate::flush::FlushReason; use crate::manifest::action::RegionChange; +use crate::memtable::ensure_json2_uses_flat_format; use crate::region::MitoRegionRef; use crate::region::options::CompactionOptions::Twcs; use crate::region::options::{RegionOptions, TwcsOptions}; @@ -195,8 +196,12 @@ impl RegionWorkerLoop { return; } }; - // Persist the metadata to region's manifest. let options = new_options.as_ref().unwrap_or(&version.options); + if let Err(e) = ensure_json2_uses_flat_format(&new_meta, options) { + sender.send(Err(e)); + return; + } + // Persist the metadata to region's manifest. let change = RegionChange { metadata: new_meta, sst_format: options.sst_format.unwrap_or_default(), diff --git a/tests/cases/standalone/common/types/json/json2_limit.result b/tests/cases/standalone/common/types/json/json2_limit.result index 555e76f1e5e..6c362a5cc62 100644 --- a/tests/cases/standalone/common/types/json/json2_limit.result +++ b/tests/cases/standalone/common/types/json/json2_limit.result @@ -112,3 +112,111 @@ drop table json2_set_append_mode_false; Affected Rows: 0 +-- JSON2 requires flat SST output even when append mode is enabled. +create table json2_primary_key_format ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'true', + 'sst_format' = 'primary_key' +); + +Error: 1004(InvalidArguments), Invalid region options, JSON2 columns require sst_format=flat + +create table json2_alter_primary_key_format ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'true', + 'memtable.type' = 'bulk' +); + +Affected Rows: 0 + +insert into json2_alter_primary_key_format values (1, '{"a": 1}'); + +Affected Rows: 1 + +alter table json2_alter_primary_key_format set 'sst_format' = 'primary_key'; + +Error: 1004(InvalidArguments), Invalid region options, JSON2 columns require sst_format=flat + +show create table json2_alter_primary_key_format; + ++--------------------------------+---------------------------------------------------------------+ +| Table | Create Table | ++--------------------------------+---------------------------------------------------------------+ +| json2_alter_primary_key_format | CREATE TABLE IF NOT EXISTS "json2_alter_primary_key_format" ( | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | "j" JSON2( | +| | max_auto_expanded_paths = 100 | +| | ) NULL, | +| | TIME INDEX ("ts") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | append_mode = 'true', | +| | 'memtable.type' = 'bulk' | +| | ) | ++--------------------------------+---------------------------------------------------------------+ + +insert into json2_alter_primary_key_format values (2, '{"a": 2}'); + +Affected Rows: 1 + +admin flush_table('json2_alter_primary_key_format'); + ++-----------------------------------------------------+ +| ADMIN flush_table('json2_alter_primary_key_format') | ++-----------------------------------------------------+ +| 0 | ++-----------------------------------------------------+ + +select ts, json_get(j, 'a') from json2_alter_primary_key_format order by ts; + ++-------------------------+------------------------------------------------------+ +| ts | json_get(json2_alter_primary_key_format.j,Utf8("a")) | ++-------------------------+------------------------------------------------------+ +| 1970-01-01T00:00:00.001 | 1 | +| 1970-01-01T00:00:00.002 | 2 | ++-------------------------+------------------------------------------------------+ + +drop table json2_alter_primary_key_format; + +Affected Rows: 0 + +create table json2_add_to_primary_key_format ( + ts timestamp time index +) with ( + 'append_mode' = 'true', + 'sst_format' = 'primary_key' +); + +Affected Rows: 0 + +alter table json2_add_to_primary_key_format add column j json2; + +Error: 1004(InvalidArguments), Invalid region options, JSON2 columns require sst_format=flat + +show create table json2_add_to_primary_key_format; + ++---------------------------------+----------------------------------------------------------------+ +| Table | Create Table | ++---------------------------------+----------------------------------------------------------------+ +| json2_add_to_primary_key_format | CREATE TABLE IF NOT EXISTS "json2_add_to_primary_key_format" ( | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | TIME INDEX ("ts") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | append_mode = 'true', | +| | sst_format = 'primary_key' | +| | ) | ++---------------------------------+----------------------------------------------------------------+ + +drop table json2_add_to_primary_key_format; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/types/json/json2_limit.sql b/tests/cases/standalone/common/types/json/json2_limit.sql index 653fdcec12c..7b428abb5e0 100644 --- a/tests/cases/standalone/common/types/json/json2_limit.sql +++ b/tests/cases/standalone/common/types/json/json2_limit.sql @@ -68,3 +68,47 @@ create table json2_set_append_mode_false ( alter table json2_set_append_mode_false set 'append_mode' = 'false'; drop table json2_set_append_mode_false; + +-- JSON2 requires flat SST output even when append mode is enabled. +create table json2_primary_key_format ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'true', + 'sst_format' = 'primary_key' +); + +create table json2_alter_primary_key_format ( + ts timestamp time index, + j json2 +) with ( + 'append_mode' = 'true', + 'memtable.type' = 'bulk' +); + +insert into json2_alter_primary_key_format values (1, '{"a": 1}'); + +alter table json2_alter_primary_key_format set 'sst_format' = 'primary_key'; + +show create table json2_alter_primary_key_format; + +insert into json2_alter_primary_key_format values (2, '{"a": 2}'); + +admin flush_table('json2_alter_primary_key_format'); + +select ts, json_get(j, 'a') from json2_alter_primary_key_format order by ts; + +drop table json2_alter_primary_key_format; + +create table json2_add_to_primary_key_format ( + ts timestamp time index +) with ( + 'append_mode' = 'true', + 'sst_format' = 'primary_key' +); + +alter table json2_add_to_primary_key_format add column j json2; + +show create table json2_add_to_primary_key_format; + +drop table json2_add_to_primary_key_format;