rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-25 05:12:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mito2): require flat SST output for JSON2 (#9893)

Browse Source 
* fix(mito2): require flat SST output for JSON2

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(mito2): avoid contradictory JSON2 recovery warnings

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

---------

Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
2026-10-23 04:15:46 +00:00
committed by GitHub
parent 6a34078706
commit c4043efc27
7 changed files with 826 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/engine/alter_test.rs
+190
View File
@@ -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,10 +34,13 @@ 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::{
ColumnDefaultConstraint, 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};
@@ -140,6 +144,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::<usize>()
);
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,
src/mito2/src/engine/create_test.rs
+403 -39
View File
@@ -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,
};
@@ -104,36 +123,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(),
)))
@@ -141,8 +213,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,
@@ -150,32 +422,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(&region_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::<Error>().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(())
}
src/mito2/src/memtable.rs
+23 -8
View File
@@ -437,8 +437,8 @@ pub(crate) struct MemtableBuilderProvider {
compact_dispatcher: Arc<CompactDispatcher>,
}
/// 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(())
}
}
src/mito2/src/region/opener.rs
+52 -30
View File
@@ -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,21 +354,12 @@ impl RegionOpener {
return Ok(region);
}
Ok(None) => {
None => {
debug!(
"No data under directory {}, region_id: {}",
region_dir, self.region_id
);
}
// The region exists but the WAL does not hold its history:
// creating it again would replace its manifest.
Err(e @ error::Error::CheckWalHistory { .. }) => return Err(e),
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();
@@ -386,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)?;
options.validate_column_compression(&metadata)?;
// The region's WAL history starts in this log store, so reopening
// the region on it passes the history check without any entry. It is
@@ -549,7 +540,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, &region_options)?;
ensure_json2_uses_flat_format(&metadata, &region_options)?;
region_options.validate_column_compression(&metadata)?;
let region_id = self.region_id;
@@ -804,23 +795,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
@@ -1656,6 +1653,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<u8> {
let key_value_meta = KeyValue::new(
PARQUET_METADATA_KEY.to_string(),
src/mito2/src/worker/handle_alter.rs
+6 -1
View File
@@ -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,12 +196,16 @@ impl<S: LogStore> RegionWorkerLoop<S> {
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;
}
if let Err(e) = options.validate_column_compression(&new_meta) {
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(),
tests/cases/standalone/common/types/json/json2_limit.result
+108
View File
@@ -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
tests/cases/standalone/common/types/json/json2_limit.sql
+44
View File
@@ -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;