diff --git a/src/api/src/error.rs b/src/api/src/error.rs index 0b35bbc934f..f6694c68679 100644 --- a/src/api/src/error.rs +++ b/src/api/src/error.rs @@ -60,6 +60,14 @@ pub enum Error { source: datatypes::error::Error, }, + #[snafu(display("Invalid compression options for column {column}"))] + InvalidColumnCompression { + column: String, + #[snafu(implicit)] + location: Location, + source: datatypes::error::Error, + }, + #[snafu(display("Failed to serialize JSON"))] SerializeJson { #[snafu(source)] @@ -93,7 +101,8 @@ impl ErrorExt for Error { StatusCode::Unexpected } Error::ConvertColumnDefaultConstraint { source, .. } - | Error::InvalidColumnDefaultConstraint { source, .. } => source.status_code(), + | Error::InvalidColumnDefaultConstraint { source, .. } + | Error::InvalidColumnCompression { source, .. } => source.status_code(), } } diff --git a/src/api/src/v1/column_def.rs b/src/api/src/v1/column_def.rs index b02b06feafd..af62ed716e5 100644 --- a/src/api/src/v1/column_def.rs +++ b/src/api/src/v1/column_def.rs @@ -16,9 +16,9 @@ use std::collections::HashMap; use arrow_schema::extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY}; use datatypes::schema::{ - COMMENT_KEY, ColumnDefaultConstraint, ColumnSchema, FULLTEXT_KEY, FulltextAnalyzer, - FulltextBackend, FulltextOptions, INVERTED_INDEX_KEY, Metadata, SKIPPING_INDEX_KEY, - SkippingIndexOptions, SkippingIndexType, + COMMENT_KEY, COMPRESSION_KEY, ColumnDefaultConstraint, ColumnSchema, FULLTEXT_KEY, + FulltextAnalyzer, FulltextBackend, FulltextOptions, INVERTED_INDEX_KEY, Metadata, + SKIPPING_INDEX_KEY, SkippingIndexOptions, SkippingIndexType, }; use greptime_proto::v1::{ Analyzer, FulltextBackend as PbFulltextBackend, SkippingIndexType as PbSkippingIndexType, @@ -35,11 +35,14 @@ const FULLTEXT_GRPC_KEY: &str = "fulltext"; const INVERTED_INDEX_GRPC_KEY: &str = "inverted_index"; /// Key used to store skip index options in gRPC column options. const SKIPPING_INDEX_GRPC_KEY: &str = "skipping_index"; +/// Key used to store SST compression options in gRPC column options. +const COMPRESSION_GRPC_KEY: &str = "compression"; -const COLUMN_OPTION_MAPPINGS: [(&str, &str); 5] = [ +const COLUMN_OPTION_MAPPINGS: [(&str, &str); 6] = [ (FULLTEXT_GRPC_KEY, FULLTEXT_KEY), (INVERTED_INDEX_GRPC_KEY, INVERTED_INDEX_KEY), (SKIPPING_INDEX_GRPC_KEY, SKIPPING_INDEX_KEY), + (COMPRESSION_GRPC_KEY, COMPRESSION_KEY), (EXTENSION_TYPE_NAME_KEY, EXTENSION_TYPE_NAME_KEY), (EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_METADATA_KEY), ]; @@ -77,6 +80,9 @@ pub fn try_as_column_schema(column_def: &ColumnDef) -> Result { if let Some(skipping_index) = options.options.get(SKIPPING_INDEX_GRPC_KEY) { metadata.insert(SKIPPING_INDEX_KEY.to_string(), skipping_index.to_owned()); } + if let Some(compression) = options.options.get(COMPRESSION_GRPC_KEY) { + metadata.insert(COMPRESSION_KEY.to_string(), compression.to_owned()); + } if let Some(extension_name) = options.options.get(EXTENSION_TYPE_NAME_KEY) { metadata.insert(EXTENSION_TYPE_NAME_KEY.to_string(), extension_name.clone()); } @@ -88,8 +94,14 @@ pub fn try_as_column_schema(column_def: &ColumnDef) -> Result { } } - ColumnSchema::new(&column_def.name, data_type.into(), column_def.is_nullable) - .with_metadata(metadata) + let schema = ColumnSchema::new(&column_def.name, data_type.into(), column_def.is_nullable) + .with_metadata(metadata); + schema + .compression_options() + .context(error::InvalidColumnCompressionSnafu { + column: &column_def.name, + })?; + schema .with_time_index(column_def.semantic_type() == SemanticType::Timestamp) .with_default_constraint(constraint) .context(error::InvalidColumnDefaultConstraintSnafu { @@ -172,6 +184,11 @@ pub fn options_from_column_schema(column_schema: &ColumnSchema) -> Option Skipp #[cfg(test)] mod tests { + use common_base::readable_size::ReadableSize; use datatypes::data_type::ConcreteDataType; - use datatypes::schema::{FulltextAnalyzer, FulltextBackend}; + use datatypes::schema::{CompressionOptions, FulltextAnalyzer, FulltextBackend}; use super::*; use crate::v1::ColumnDataType; @@ -307,6 +325,54 @@ mod tests { assert!(schema.is_inverted_indexed()); } + #[test] + fn test_compression_options_round_trip() { + let options = CompressionOptions { + level: Some(3), + page_size: Some(ReadableSize::mb(8)), + ..Default::default() + }; + let schema = ColumnSchema::new("test", ConcreteDataType::string_datatype(), true) + .with_compression_options(options) + .unwrap(); + + let column_def = try_as_column_def(&schema, false).unwrap(); + let schema = try_as_column_schema(&column_def).unwrap(); + assert_eq!(Some(options), schema.compression_options().unwrap()); + + let metadata = collect_column_options(column_def.options.as_ref()); + assert!(metadata.contains_key(COMPRESSION_KEY)); + } + + #[test] + fn test_reject_invalid_compression_metadata() { + use common_error::ext::ErrorExt; + use common_error::status_code::StatusCode; + + for (data_type, json) in [ + (ConcreteDataType::string_datatype(), r#"{"level":23}"#), + (ConcreteDataType::string_datatype(), r#"{"level":0}"#), + (ConcreteDataType::string_datatype(), r#"{"page_size":"0B"}"#), + (ConcreteDataType::string_datatype(), r#"{"type":"lz4"}"#), + (ConcreteDataType::string_datatype(), "not json"), + ( + ConcreteDataType::list_datatype(std::sync::Arc::new( + ConcreteDataType::string_datatype(), + )), + r#"{"level":3}"#, + ), + ] { + let schema = + ColumnSchema::new("test", data_type, true).with_metadata(HashMap::from([( + COMPRESSION_KEY.to_string(), + json.to_string(), + )])); + let column_def = try_as_column_def(&schema, false).unwrap(); + let err = try_as_column_schema(&column_def).unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code(), "{json}"); + } + } + #[test] fn test_options_from_column_schema() { let schema = ColumnSchema::new("test", ConcreteDataType::string_datatype(), true); diff --git a/src/datatypes/src/error.rs b/src/datatypes/src/error.rs index 32780270874..745268f6e35 100644 --- a/src/datatypes/src/error.rs +++ b/src/datatypes/src/error.rs @@ -285,6 +285,13 @@ pub enum Error { location: Location, }, + #[snafu(display("Invalid compression option: {}", msg))] + InvalidCompressionOption { + msg: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Inconsistent struct field count {field_len} and item count {item_len}"))] InconsistentStructFieldsAndItems { field_len: usize, @@ -360,7 +367,8 @@ impl ErrorExt for Error { | InvalidJsonb { .. } | InvalidVector { .. } | InvalidFulltextOption { .. } - | InvalidSkippingIndexOption { .. } => StatusCode::InvalidArguments, + | InvalidSkippingIndexOption { .. } + | InvalidCompressionOption { .. } => StatusCode::InvalidArguments, ValueExceedsPrecision { .. } | CastType { .. } diff --git a/src/datatypes/src/schema.rs b/src/datatypes/src/schema.rs index 0f1a832ebd2..66867f6141d 100644 --- a/src/datatypes/src/schema.rs +++ b/src/datatypes/src/schema.rs @@ -28,12 +28,14 @@ use snafu::{ResultExt, ensure}; use crate::error::{self, DuplicateColumnSnafu, Error, ProjectArrowSchemaSnafu, Result}; use crate::prelude::ConcreteDataType; pub use crate::schema::column_schema::{ - COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE, COLUMN_FULLTEXT_OPT_KEY_ANALYZER, - COLUMN_FULLTEXT_OPT_KEY_BACKEND, COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, - COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE, COLUMN_FULLTEXT_OPT_KEY_GRANULARITY, - COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE, COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY, - COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, COMMENT_KEY, ColumnExtType, ColumnSchema, FULLTEXT_KEY, - FulltextAnalyzer, FulltextBackend, FulltextOptions, INVERTED_INDEX_KEY, Metadata, + COLUMN_COMPRESSION_OPT_KEY_LEVEL, COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE, + COLUMN_COMPRESSION_OPT_KEY_TYPE, COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE, + COLUMN_FULLTEXT_OPT_KEY_ANALYZER, COLUMN_FULLTEXT_OPT_KEY_BACKEND, + COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE, + COLUMN_FULLTEXT_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE, + COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, COMMENT_KEY, + COMPRESSION_KEY, ColumnExtType, ColumnSchema, CompressionOptions, CompressionType, + FULLTEXT_KEY, FulltextAnalyzer, FulltextBackend, FulltextOptions, INVERTED_INDEX_KEY, Metadata, SKIPPING_INDEX_KEY, SkippingIndexOptions, SkippingIndexType, TIME_INDEX_KEY, }; pub use crate::schema::constraint::ColumnDefaultConstraint; diff --git a/src/datatypes/src/schema/column_schema.rs b/src/datatypes/src/schema/column_schema.rs index b3ce9cafedb..2d48ca72ad5 100644 --- a/src/datatypes/src/schema/column_schema.rs +++ b/src/datatypes/src/schema/column_schema.rs @@ -20,6 +20,7 @@ use arrow::datatypes::Field; use arrow_schema::extension::{ EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY, ExtensionType, }; +use common_base::readable_size::ReadableSize; use serde::{Deserialize, Serialize}; use snafu::{ResultExt, ensure}; use sqlparser_derive::{Visit, VisitMut}; @@ -48,6 +49,8 @@ pub const FULLTEXT_KEY: &str = "greptime:fulltext"; pub const INVERTED_INDEX_KEY: &str = "greptime:inverted_index"; /// Key used to store skip options in arrow field's metadata. pub const SKIPPING_INDEX_KEY: &str = "greptime:skipping_index"; +/// Key used to store SST compression options in arrow field's metadata. +pub const COMPRESSION_KEY: &str = "greptime:compression"; /// Keys used in fulltext options pub const COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE: &str = "enable"; @@ -62,6 +65,11 @@ pub const COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY: &str = "granularity"; pub const COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE: &str = "false_positive_rate"; pub const COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE: &str = "type"; +/// Keys used in compression options +pub const COLUMN_COMPRESSION_OPT_KEY_TYPE: &str = "type"; +pub const COLUMN_COMPRESSION_OPT_KEY_LEVEL: &str = "level"; +pub const COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE: &str = "page_size"; + pub const DEFAULT_GRANULARITY: u32 = 10240; pub const DEFAULT_FALSE_POSITIVE_RATE: f64 = 0.01; @@ -427,6 +435,48 @@ impl ColumnSchema { Ok(()) } + /// Retrieves the SST compression options for the column. + pub fn compression_options(&self) -> Result> { + match self.metadata.get(COMPRESSION_KEY) { + None => Ok(None), + Some(json) => { + let options = serde_json::from_str(json).map_err(|e| { + error::InvalidCompressionOptionSnafu { + msg: format!("invalid compression metadata for column {}: {e}", self.name), + } + .build() + })?; + self.validate_compression_options(&options)?; + Ok(Some(options)) + } + } + } + + /// Sets the SST compression options. Nested types are rejected: their values are + /// stored in child columns that the options would not reach. + pub fn with_compression_options(mut self, options: CompressionOptions) -> Result { + self.validate_compression_options(&options)?; + self.metadata.insert( + COMPRESSION_KEY.to_string(), + serde_json::to_string(&options).context(error::SerializeSnafu)?, + ); + Ok(self) + } + + fn validate_compression_options(&self, options: &CompressionOptions) -> Result<()> { + options.validate()?; + ensure!( + !self.data_type.as_arrow_type().is_nested(), + error::InvalidCompressionOptionSnafu { + msg: format!( + "compression options are not supported on column {} of type {}", + self.name, self.data_type + ), + } + ); + Ok(()) + } + pub fn extension_type(&self) -> Result> where E: ExtensionType, @@ -952,6 +1002,141 @@ impl fmt::Display for SkippingIndexType { } } +/// Compression algorithms for SST columns. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CompressionType { + #[default] + Zstd, +} + +impl fmt::Display for CompressionType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + CompressionType::Zstd => write!(f, "zstd"), + } + } +} + +/// Parquet compression settings for a column in SSTs. Unset fields keep the SST defaults. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct CompressionOptions { + #[serde(rename = "type", default)] + pub compression_type: CompressionType, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub level: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub page_size: Option, +} + +impl CompressionOptions { + fn validate(&self) -> Result<()> { + if let Some(level) = self.level { + ensure!( + ZSTD_LEVELS.contains(&level), + error::InvalidCompressionOptionSnafu { + msg: format!("invalid level: {level}, expected: an integer in 1..=22"), + } + ); + } + if let Some(page_size) = self.page_size { + ensure!( + page_size.as_bytes() > 0, + error::InvalidCompressionOptionSnafu { + msg: "page_size must be positive", + } + ); + } + Ok(()) + } + + /// Returns the options as `COMPRESSION WITH (...)` key-value pairs. + pub fn to_options_map(&self) -> HashMap { + let mut map = HashMap::from([( + COLUMN_COMPRESSION_OPT_KEY_TYPE.to_string(), + self.compression_type.to_string(), + )]); + if let Some(level) = self.level { + map.insert( + COLUMN_COMPRESSION_OPT_KEY_LEVEL.to_string(), + level.to_string(), + ); + } + if let Some(page_size) = self.page_size { + // `Display` rounds to one decimal; the serde form is exact and parses back. + let page_size = match serde_json::to_value(page_size) { + Ok(serde_json::Value::String(size)) => size, + _ => page_size.as_bytes().to_string(), + }; + map.insert(COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE.to_string(), page_size); + } + map + } +} + +/// Valid zstd levels, as accepted by the parquet writer. +const ZSTD_LEVELS: std::ops::RangeInclusive = 1..=22; + +impl TryFrom> for CompressionOptions { + type Error = Error; + + fn try_from(options: HashMap) -> Result { + let compression_type = match options.get(COLUMN_COMPRESSION_OPT_KEY_TYPE) { + Some(value) if value.eq_ignore_ascii_case("zstd") => CompressionType::Zstd, + Some(value) => { + return error::InvalidCompressionOptionSnafu { + msg: format!("invalid type: {value}, expected: 'zstd'"), + } + .fail(); + } + None => CompressionType::default(), + }; + + let level = options + .get(COLUMN_COMPRESSION_OPT_KEY_LEVEL) + .map(|value| { + value + .parse::() + .ok() + .filter(|level| ZSTD_LEVELS.contains(level)) + .ok_or_else(|| { + error::InvalidCompressionOptionSnafu { + msg: format!( + "invalid level: {value}, expected: an integer in {}..={}", + ZSTD_LEVELS.start(), + ZSTD_LEVELS.end() + ), + } + .build() + }) + }) + .transpose()?; + + let page_size = options + .get(COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE) + .map(|value| { + ReadableSize::from_str(value) + .ok() + .filter(|size| size.as_bytes() > 0) + .ok_or_else(|| { + error::InvalidCompressionOptionSnafu { + msg: format!( + "invalid page_size: {value}, expected: a positive size such as '8MiB'" + ), + } + .build() + }) + }) + .transpose()?; + + Ok(CompressionOptions { + compression_type, + level, + page_size, + }) + } +} + impl TryFrom> for SkippingIndexOptions { type Error = Error; @@ -1323,6 +1508,96 @@ mod tests { assert!(column_schema.metadata.is_empty()); } + #[test] + fn test_compression_options_from_map() { + let parse = |pairs: &[(&str, &str)]| { + CompressionOptions::try_from( + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect::>(), + ) + }; + + let options = parse(&[("type", "ZSTD"), ("level", "3"), ("page_size", "8MiB")]).unwrap(); + assert_eq!( + CompressionOptions { + compression_type: CompressionType::Zstd, + level: Some(3), + page_size: Some(ReadableSize::mb(8)), + }, + options + ); + // Stored in column metadata; this format must stay readable. + assert_eq!( + r#"{"type":"zstd","level":3,"page_size":"8MiB"}"#, + serde_json::to_string(&options).unwrap() + ); + assert_eq!( + options, + parse( + &options + .to_options_map() + .iter() + .map(|(k, v)| (k.as_str(), v.as_str())) + .collect::>() + ) + .unwrap() + ); + + let odd_size = parse(&[("page_size", "1.25MiB")]).unwrap(); + assert_eq!( + odd_size, + parse(&[("page_size", odd_size.to_options_map()["page_size"].as_str())]).unwrap() + ); + + let page_only = parse(&[("page_size", "1MiB")]).unwrap(); + assert_eq!(None, page_only.level); + assert_eq!( + r#"{"type":"zstd","page_size":"1MiB"}"#, + serde_json::to_string(&page_only).unwrap() + ); + + for invalid in [ + &[("type", "lz4")][..], + &[("level", "0")], + &[("level", "23")], + &[("level", "high")], + &[("page_size", "0")], + &[("page_size", "large")], + ] { + assert!(parse(invalid).is_err(), "{invalid:?}"); + } + } + + #[test] + fn test_compression_options_reject_nested_types() { + let options = CompressionOptions { + level: Some(3), + ..Default::default() + }; + let string_column = ColumnSchema::new("s", ConcreteDataType::string_datatype(), true) + .with_compression_options(options) + .unwrap(); + assert_eq!(Some(options), string_column.compression_options().unwrap()); + + let struct_type = + ConcreteDataType::struct_datatype(StructType::new(Arc::new(vec![StructField::new( + "a".to_string(), + ConcreteDataType::string_datatype(), + true, + )]))); + let list_type = + ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::string_datatype())); + for data_type in [struct_type, list_type] { + assert!( + ColumnSchema::new("n", data_type, true) + .with_compression_options(options) + .is_err() + ); + } + } + #[test] fn test_skipping_index_options_deserialization() { let original_options = "{\"granularity\":1024,\"false-positive-rate-in-10000\":10,\"index-type\":\"BloomFilter\"}"; diff --git a/src/metric-engine/src/engine/alter/validate.rs b/src/metric-engine/src/engine/alter/validate.rs index 2e2d91eccfc..cd7d8f2f172 100644 --- a/src/metric-engine/src/engine/alter/validate.rs +++ b/src/metric-engine/src/engine/alter/validate.rs @@ -12,11 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +use datatypes::schema::COMPRESSION_KEY; use snafu::ensure; use store_api::region_request::{AlterKind, RegionAlterRequest}; use store_api::storage::RegionId; -use crate::error::{Result, UnsupportedAlterKindSnafu}; +use crate::error::{Result, UnexpectedRequestSnafu, UnsupportedAlterKindSnafu}; /// Validate the alter region requests. pub fn validate_alter_region_requests(requests: &[(RegionId, RegionAlterRequest)]) -> Result<()> { @@ -27,7 +28,58 @@ pub fn validate_alter_region_requests(requests: &[(RegionId, RegionAlterRequest) kind: request.kind.as_ref() } ); + if let AlterKind::AddColumns { columns } = &request.kind { + for column in columns { + ensure!( + !column.column_metadata.column_schema.metadata().contains_key(COMPRESSION_KEY), + UnexpectedRequestSnafu { + reason: "COMPRESSION is not supported on logical tables; configure it on the physical table".to_string(), + } + ); + } + } } Ok(()) } + +#[cfg(test)] +mod tests { + use api::v1::SemanticType; + use common_error::ext::ErrorExt; + use common_error::status_code::StatusCode; + use datatypes::prelude::ConcreteDataType; + use datatypes::schema::{ColumnSchema, CompressionOptions}; + use store_api::metadata::ColumnMetadata; + use store_api::region_request::AddColumn; + + use super::*; + + #[test] + fn test_reject_logical_column_compression() { + let column_schema = ColumnSchema::new("host", ConcreteDataType::string_datatype(), true) + .with_compression_options(CompressionOptions { + level: Some(3), + ..Default::default() + }) + .unwrap(); + let request = RegionAlterRequest { + kind: AlterKind::AddColumns { + columns: vec![AddColumn { + column_metadata: ColumnMetadata { + column_schema, + semantic_type: SemanticType::Tag, + column_id: 3, + }, + location: None, + }], + }, + }; + let err = validate_alter_region_requests(&[(RegionId::new(1, 1), request)]).unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + assert!( + err.to_string() + .contains("configure it on the physical table") + ); + } +} diff --git a/src/metric-engine/src/engine/create.rs b/src/metric-engine/src/engine/create.rs index 952643cd7a4..ec35c0653c4 100644 --- a/src/metric-engine/src/engine/create.rs +++ b/src/metric-engine/src/engine/create.rs @@ -21,7 +21,7 @@ use common_query::native_histogram::is_native_histogram_value_type; use common_telemetry::info; use common_time::{FOREVER, Timestamp}; use datatypes::data_type::ConcreteDataType; -use datatypes::schema::{ColumnSchema, SkippingIndexOptions}; +use datatypes::schema::{COMPRESSION_KEY, ColumnSchema, SkippingIndexOptions}; use datatypes::value::Value; use mito2::engine::MITO_ENGINE_NAME; use snafu::{OptionExt, ResultExt, ensure}; @@ -391,6 +391,17 @@ impl MetricEngineInner { ConflictRegionOptionSnafu {} ); + if request.options.contains_key(LOGICAL_TABLE_METADATA_KEY) { + for column in &request.column_metadatas { + ensure!( + !column.column_schema.metadata().contains_key(COMPRESSION_KEY), + UnexpectedRequestSnafu { + reason: "COMPRESSION is not supported on logical tables; configure it on the physical table".to_string(), + } + ); + } + } + // check if field columns are either a normal metric value or a native histogram. let mut field_cols = Vec::new(); for col in &request.column_metadatas { diff --git a/src/mito2/src/engine/create_test.rs b/src/mito2/src/engine/create_test.rs index 2c04e2e24f1..00c3686c764 100644 --- a/src/mito2/src/engine/create_test.rs +++ b/src/mito2/src/engine/create_test.rs @@ -30,6 +30,74 @@ use crate::test_util::{ CreateRequestBuilder, TestEnv, build_rows, flush_region, put_rows, reopen_region, rows_schema, }; +#[tokio::test] +async fn test_engine_column_compression_validation() { + use datatypes::schema::COMPRESSION_KEY; + use store_api::region_request::{AlterKind, RegionAlterRequest, SetRegionOption}; + + let mut env = TestEnv::new().await; + let engine = env.create_engine(MitoConfig::default()).await; + let region_id = RegionId::new(1, 1); + for (column_index, compression, options, accepted) in [ + (1, r#"{"level":23}"#, vec![], false), + ( + 0, + r#"{"level":3}"#, + vec![ + ("sst_format", "primary_key"), + ("memtable.type", "time_series"), + ], + false, + ), + ( + 0, + r#"{"level":3}"#, + vec![("primary_key_encoding", "sparse")], + false, + ), + (0, r#"{"level":3}"#, vec![], true), + ] { + let mut request = CreateRequestBuilder::new().build(); + request.options.extend( + options + .into_iter() + .map(|(k, v)| (k.to_string(), v.to_string())), + ); + let column = &mut request.column_metadatas[column_index].column_schema; + *column = column + .clone() + .with_metadata(std::collections::HashMap::from([( + COMPRESSION_KEY.to_string(), + compression.to_string(), + )])); + let result = engine + .handle_request(region_id, RegionRequest::Create(request)) + .await; + if accepted { + result.unwrap(); + } else { + assert_eq!( + StatusCode::InvalidArguments, + result.unwrap_err().status_code() + ); + assert!(engine.get_region(region_id).is_none()); + } + } + + let err = engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::SetRegionOptions { + options: vec![SetRegionOption::Format("primary_key".to_string())], + }, + }), + ) + .await + .unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); +} + #[tokio::test] async fn test_engine_create_new_region() { test_engine_create_new_region_with_format(false).await; diff --git a/src/mito2/src/error.rs b/src/mito2/src/error.rs index 5af4b7cec09..54bf1ba6b6e 100644 --- a/src/mito2/src/error.rs +++ b/src/mito2/src/error.rs @@ -1094,6 +1094,14 @@ pub enum Error { column_name: String, }, + #[snafu(display("Failed to retrieve compression options of column {column_name}"))] + CompressionOptions { + #[snafu(implicit)] + location: Location, + source: datatypes::error::Error, + column_name: String, + }, + #[snafu(display("Failed to create fulltext index creator"))] CreateFulltextCreator { source: index::fulltext_index::error::Error, @@ -1586,6 +1594,7 @@ impl ErrorExt for Error { RemoteCompaction { .. } => StatusCode::Unexpected, IndexOptions { source, .. } => source.status_code(), + CompressionOptions { source, .. } => source.status_code(), CreateFulltextCreator { source, .. } => source.status_code(), CastVector { source, .. } => source.status_code(), FulltextPushText { source, .. } @@ -1691,6 +1700,7 @@ impl ErrorExt for Error { | ConvertVector { source, .. } | ConvertValue { source, .. } | IndexOptions { source, .. } + | CompressionOptions { source, .. } | CastVector { source, .. } => source.retry_hint(), BuildIndexApplier { source, .. } diff --git a/src/mito2/src/memtable/bulk/part.rs b/src/mito2/src/memtable/bulk/part.rs index fecfceafeee..27fc7983513 100644 --- a/src/mito2/src/memtable/bulk/part.rs +++ b/src/mito2/src/memtable/bulk/part.rs @@ -73,7 +73,8 @@ use crate::sst::index::IndexOutput; use crate::sst::parquet::flat_format::primary_key_column_index; use crate::sst::parquet::format::{PrimaryKeyArray, PrimaryKeyArrayBuilder}; use crate::sst::parquet::{ - COLUMN_INDEX_TRUNCATE_LENGTH, PARQUET_METADATA_KEY, SstInfo, apply_float_field_encoding, + COLUMN_INDEX_TRUNCATE_LENGTH, PARQUET_METADATA_KEY, SstInfo, apply_column_compression, + apply_float_field_encoding, }; const INIT_DICT_VALUE_CAPACITY: usize = 8; @@ -1347,6 +1348,7 @@ impl BulkPartEncoder { .set_column_index_truncate_length(COLUMN_INDEX_TRUNCATE_LENGTH) .set_statistics_truncate_length(None); props = apply_float_field_encoding(props, &metadata, float_field_encoding); + props = apply_column_compression(props, &metadata)?; let writer_props = Some(props.build()); Ok(Self { diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs index 71c1192cbf2..7a93799ae5f 100644 --- a/src/mito2/src/region/opener.rs +++ b/src/mito2/src/region/opener.rs @@ -384,6 +384,7 @@ impl RegionOpener { FormatType::PrimaryKey }; ensure_json2_not_use_time_series_memtable(&metadata, &options)?; + options.validate_column_compression(&metadata)?; // 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); @@ -539,6 +540,7 @@ impl RegionOpener { // Updates the region options with the manifest. sanitize_region_options(&manifest, &mut region_options); ensure_json2_not_use_time_series_memtable(&metadata, ®ion_options)?; + region_options.validate_column_compression(&metadata)?; let region_id = self.region_id; let provider = self.provider::(®ion_options.wal_options)?; diff --git a/src/mito2/src/region/options.rs b/src/mito2/src/region/options.rs index 89b413a5d8f..7e050fb4a76 100644 --- a/src/mito2/src/region/options.rs +++ b/src/mito2/src/region/options.rs @@ -19,16 +19,19 @@ use std::collections::HashMap; use std::time::Duration; +use api::v1::SemanticType; use common_base::readable_size::ReadableSize; use common_telemetry::info; use common_time::TimeToLive; use common_wal::options::{WAL_OPTIONS_KEY, WalOptions}; +use datatypes::schema::COMPRESSION_KEY; use serde::de::Error as _; use serde::{Deserialize, Deserializer, Serialize}; use serde_json::Value; use serde_with::{DisplayFromStr, NoneAsEmptyString, serde_as, with_prefix}; use snafu::{ResultExt, ensure}; use store_api::codec::PrimaryKeyEncoding; +use store_api::metadata::RegionMetadata; use store_api::metric_engine_consts::{ MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING, PRIMARY_KEY_ENCODING, }; @@ -194,6 +197,32 @@ impl RegionOptions { Ok(()) } + /// Rejects tag compression when the SST layout packs tags into `__primary_key`. + /// The caller must resolve the SST format before validation. + pub(crate) fn validate_column_compression(&self, metadata: &RegionMetadata) -> Result<()> { + if self.sst_format == Some(FormatType::Flat) + && metadata.primary_key_encoding == PrimaryKeyEncoding::Dense + { + return Ok(()); + } + for column in &metadata.column_metadatas { + ensure!( + column.semantic_type != SemanticType::Tag + || !column + .column_schema + .metadata() + .contains_key(COMPRESSION_KEY), + InvalidRegionOptionsSnafu { + reason: format!( + "COMPRESSION on tag column {} requires flat SST format and dense primary key encoding", + column.column_schema.name + ), + } + ); + } + Ok(()) + } + /// Returns the configured row group size, falling back to [DEFAULT_ROW_GROUP_SIZE]. pub fn row_group_size(&self) -> usize { self.max_row_group_row_count diff --git a/src/mito2/src/sst/parquet.rs b/src/mito2/src/sst/parquet.rs index 9404d4484fc..e6968447dce 100644 --- a/src/mito2/src/sst/parquet.rs +++ b/src/mito2/src/sst/parquet.rs @@ -20,13 +20,16 @@ use std::sync::Arc; use api::v1::SemanticType; use common_base::readable_size::ReadableSize; use datatypes::json::JsonSettings; +use parquet::basic::{Compression, ZstdLevel}; use parquet::file::metadata::ParquetMetaData; -use parquet::file::properties::WriterPropertiesBuilder; +use parquet::file::properties::{DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT, WriterPropertiesBuilder}; use parquet::schema::types::ColumnPath; +use snafu::ResultExt; use store_api::metadata::RegionMetadataRef; use store_api::mito_engine_options::FloatFieldEncoding; use store_api::storage::{ColumnId, FileId}; +use crate::error::{CompressionOptionsSnafu, Result, WriteParquetSnafu}; use crate::sst::DEFAULT_WRITE_BUFFER_SIZE; use crate::sst::file::FileTimeRange; use crate::sst::index::IndexOutput; @@ -89,6 +92,41 @@ pub const DEFAULT_ROW_GROUP_SIZE: usize = 100 * 1024; pub(crate) const COLUMN_INDEX_TRUNCATE_LENGTH: Option = parquet::file::properties::DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH; +/// Applies the per-column options declared with `COMPRESSION WITH (...)`. +pub(crate) fn apply_column_compression( + mut builder: WriterPropertiesBuilder, + metadata: &RegionMetadataRef, +) -> Result { + for column in &metadata.column_metadatas { + let Some(options) = + column + .column_schema + .compression_options() + .context(CompressionOptionsSnafu { + column_name: &column.column_schema.name, + })? + else { + continue; + }; + let path = ColumnPath::new(vec![column.column_schema.name.clone()]); + if let Some(level) = options.level { + let level = ZstdLevel::try_new(level).context(WriteParquetSnafu)?; + builder = builder.set_column_compression(path.clone(), Compression::ZSTD(level)); + } + if let Some(page_size) = options.page_size { + let page_size = page_size.as_bytes() as usize; + builder = builder.set_column_data_page_size_limit(path.clone(), page_size); + // While dictionary encoding, parquet sizes write batches by the dictionary page + // budget and keeps that size for the rest of the batch after falling back to plain + // encoding, so a data page limit below the dictionary limit would be ignored. + if page_size < DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT { + builder = builder.set_column_dictionary_page_size_limit(path, page_size); + } + } + } + Ok(builder) +} + /// Applies the configured encoding to direct floating-point field columns. pub(crate) fn apply_float_field_encoding( mut builder: WriterPropertiesBuilder, @@ -167,6 +205,7 @@ mod tests { use api::v1::{OpType, SemanticType}; use bytes::Bytes; + use common_base::readable_size::ReadableSize; use common_function::function::FunctionRef; use common_function::function_factory::ScalarFunctionFactory; use common_function::scalars::matches::MatchesFunction; @@ -184,7 +223,9 @@ mod tests { use datatypes::arrow::datatypes::{DataType, Field, Schema, TimeUnit, UInt32Type}; use datatypes::arrow::util::pretty::pretty_format_batches; use datatypes::prelude::ConcreteDataType; - use datatypes::schema::{FulltextAnalyzer, FulltextBackend, FulltextOptions}; + use datatypes::schema::{ + CompressionOptions, FulltextAnalyzer, FulltextBackend, FulltextOptions, + }; use object_store::ObjectStore; use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; use parquet::arrow::{ArrowWriter, AsyncArrowWriter}; @@ -820,6 +861,138 @@ mod tests { .await; } + #[tokio::test] + async fn test_column_compression_options() { + // Text-like values: pseudo-random words from a small vocabulary, so zstd levels + // compress differently and the dictionary falls back to plain encoding. + let words: Vec = (0..256).map(|i| format!("word{i:03}")).collect(); + let mut seed = 42u64; + let values: Vec> = (0..2048) + .map(|_| { + (0..250) + .map(|_| { + seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1); + words[(seed >> 33) as usize % words.len()].as_str() + }) + .collect::>() + .join(" ") + .into_bytes() + }) + .collect(); + + async fn write( + values: &[Vec], + compression: Option, + ) -> (usize, usize) { + let mut metadata = (*build_test_binary_test_region_metadata()).clone(); + if let Some(options) = compression { + let column = &mut metadata.column_metadatas[1]; + assert_eq!("field_0", column.column_schema.name); + column.column_schema = column + .column_schema + .clone() + .with_compression_options(options) + .unwrap(); + } + let metadata = Arc::new(metadata); + let flat_schema = to_flat_sst_arrow_schema(&metadata, &FlatSchemaOptions::default()); + let num_rows = values.len(); + let pk = new_primary_key(&["a"]); + let mut tags = StringDictionaryBuilder::::new(); + let mut pks = BinaryDictionaryBuilder::::new(); + for _ in 0..num_rows { + tags.append_value("a"); + pks.append(&pk).unwrap(); + } + let batch = RecordBatch::try_new( + flat_schema, + vec![ + Arc::new(tags.finish()), + Arc::new(datatypes::arrow::array::BinaryArray::from_iter_values( + values.iter().map(|v| v.as_slice()), + )), + Arc::new(TimestampMillisecondArray::from_iter_values( + 0..num_rows as i64, + )), + Arc::new(pks.finish()), + Arc::new(UInt64Array::from_value(1000, num_rows)), + Arc::new(UInt8Array::from_value(OpType::Put as u8, num_rows)), + ], + ) + .unwrap(); + + let mut env = TestEnv::new().await; + let object_store = env.init_object_store_manager(); + let handle = sst_file_handle(0, 1000); + let mut metrics = Metrics::new(WriteType::Flush); + let mut writer = ParquetWriter::new_with_object_store( + object_store.clone(), + metadata, + IndexConfig::default(), + NoopIndexBuilder, + FixedPathProvider { + region_file_id: handle.file_id(), + }, + &mut metrics, + ) + .await; + writer + .write_all_flat_as_primary_key( + new_flat_source_from_record_batches(vec![batch]), + None, + &WriteOptions::default(), + ) + .await + .unwrap(); + + let bytes = object_store + .read(&handle.file_path(FILE_DIR, PathType::Bare)) + .await + .unwrap() + .to_bytes(); + let options = parquet::arrow::arrow_reader::ArrowReaderOptions::new() + .with_page_index_policy(PageIndexPolicy::Required); + let builder = + ParquetRecordBatchReaderBuilder::try_new_with_options(bytes, options).unwrap(); + let schema = builder.metadata().file_metadata().schema_descr(); + let field = (0..schema.num_columns()) + .find(|i| schema.column(*i).name() == "field_0") + .unwrap(); + let pages = builder.metadata().offset_index().unwrap()[0][field] + .page_locations() + .len(); + let size = builder + .metadata() + .row_group(0) + .column(field) + .compressed_size() as usize; + (size, pages) + } + + let (default_size, default_pages) = write(&values, None).await; + let (level_size, level_pages) = write( + &values, + Some(CompressionOptions { + level: Some(19), + ..Default::default() + }), + ) + .await; + let (_, pages) = write( + &values, + Some(CompressionOptions { + page_size: Some(ReadableSize::kb(64)), + ..Default::default() + }), + ) + .await; + + assert!(default_pages > 1, "{default_pages}"); + assert_eq!(default_pages, level_pages); + assert!(level_size < default_size, "{level_size} >= {default_size}"); + assert!(pages > default_pages, "{pages} <= {default_pages}"); + } + #[tokio::test] async fn test_read_large_binary() { let mut env = TestEnv::new().await; diff --git a/src/mito2/src/sst/parquet/writer.rs b/src/mito2/src/sst/parquet/writer.rs index 932654dfdd8..9d2defcceb3 100644 --- a/src/mito2/src/sst/parquet/writer.rs +++ b/src/mito2/src/sst/parquet/writer.rs @@ -63,7 +63,7 @@ use crate::sst::parquet::flat_format::{ use crate::sst::parquet::format::{PrimaryKeyArray, PrimaryKeyWriteFormat}; use crate::sst::parquet::{ COLUMN_INDEX_TRUNCATE_LENGTH, PARQUET_METADATA_KEY, SstInfo, WriteOptions, - apply_float_field_encoding, + apply_column_compression, apply_float_field_encoding, }; use crate::sst::{ DEFAULT_WRITE_CONCURRENCY, FlatSchemaOptions, SeriesEstimator, maybe_wrap_schema, @@ -563,6 +563,7 @@ where &self.metadata, opts.float_field_encoding, ); + let props_builder = apply_column_compression(props_builder, &self.metadata)?; let writer_props = props_builder.build(); let sst_file_path = self.path_provider.build_sst_file_path(RegionFileId::new( diff --git a/src/mito2/src/worker/handle_alter.rs b/src/mito2/src/worker/handle_alter.rs index 66cb2dd95bb..2bb1fadc7dc 100644 --- a/src/mito2/src/worker/handle_alter.rs +++ b/src/mito2/src/worker/handle_alter.rs @@ -197,6 +197,10 @@ impl RegionWorkerLoop { }; // Persist the metadata to region's manifest. let options = new_options.as_ref().unwrap_or(&version.options); + if let Err(e) = options.validate_column_compression(&new_meta) { + sender.send(Err(e)); + return; + } let change = RegionChange { metadata: new_meta, sst_format: options.sst_format.unwrap_or_default(), @@ -356,6 +360,15 @@ impl RegionWorkerLoop { } .build() })?; + candidate + .validate_column_compression(&version.metadata) + .map_err(|e| { + store_api::metadata::InvalidRegionRequestSnafu { + region_id: region.region_id, + err: e.to_string(), + } + .build() + })?; if all_options_altered { region.version_control.alter_options(candidate); Ok(None) diff --git a/src/operator/src/expr_helper.rs b/src/operator/src/expr_helper.rs index d69277a1c90..935ce149414 100644 --- a/src/operator/src/expr_helper.rs +++ b/src/operator/src/expr_helper.rs @@ -375,6 +375,14 @@ pub fn expr_to_create(expr: &CreateTableExpr, quote_style: Option) -> Resu extensions.inverted_index_options = Some(HashMap::new().into()); } + if let Some(opt) = column_schema + .compression_options() + .map_err(BoxedError::new) + .context(ExternalSnafu)? + { + extensions.compression_options = Some(opt.to_options_map().into()); + } + let sql_column = SqlColumn { column_def: ColumnDef { name: Ident::with_quote(quote_style, &column_def.name), @@ -2117,7 +2125,7 @@ ALTER TABLE metrics PARTITION ON COLUMNS (device_id, area) ( `protocol` STRING NULL, `status_code` INT NULL INVERTED INDEX, `response_size` BIGINT NULL, - `message` STRING NULL, + `message` STRING NULL COMPRESSION WITH(level = '3', page_size = '64KiB', type = 'zstd'), TIME INDEX (`timestamp`), PRIMARY KEY (`username`, `status_code`) ) diff --git a/src/query/src/sql/show_create_table.rs b/src/query/src/sql/show_create_table.rs index c56e83d385a..b9b902a0468 100644 --- a/src/query/src/sql/show_create_table.rs +++ b/src/query/src/sql/show_create_table.rs @@ -169,6 +169,10 @@ fn create_column(column_schema: &ColumnSchema, quote_style: char) -> Result StatusCode::PermissionDenied, SetFulltextOption { .. } | SetSkippingIndexOption { .. } => StatusCode::Unexpected, + SetCompressionOption { source, .. } => source.status_code(), } } diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs index c87d0274382..20eba9f2087 100644 --- a/src/sql/src/parsers/create_parser.rs +++ b/src/sql/src/parsers/create_parser.rs @@ -47,8 +47,8 @@ use crate::error::{ use crate::parser::{FLOW, ParserContext}; use crate::parsers::tql_parser; use crate::parsers::utils::{ - self, parse_with_options, validate_column_fulltext_create_option, - validate_column_skipping_index_create_option, + self, parse_with_options, validate_column_compression_create_option, + validate_column_fulltext_create_option, validate_column_skipping_index_create_option, }; use crate::statements::create::{ Column, ColumnExtensions, CreateDatabase, CreateExternalTable, CreateFlow, CreateTable, @@ -891,6 +891,7 @@ impl<'a> ParserContext<'a> { /// This function will handle: /// - Vector type /// - Indexes + /// - SST compression options fn parse_column_extensions( parser: &mut Parser<'_>, column_name: &Ident, @@ -923,7 +924,7 @@ impl<'a> ParserContext<'a> { } // parse index options in column definition - let mut is_index_declared = false; + let mut is_extension_declared = false; // skipping index if let Token::Word(word) = parser.peek_token().token @@ -965,7 +966,7 @@ impl<'a> ParserContext<'a> { let options = OptionMap::new(options); column_extensions.skipping_index_options = Some(options); - is_index_declared |= true; + is_extension_declared |= true; } // fulltext index @@ -1016,7 +1017,7 @@ impl<'a> ParserContext<'a> { let options = OptionMap::new(options); column_extensions.fulltext_index_options = Some(options); - is_index_declared |= true; + is_extension_declared |= true; } // inverted index @@ -1058,10 +1059,54 @@ impl<'a> ParserContext<'a> { ); column_extensions.inverted_index_options = Some(OptionMap::default()); - is_index_declared |= true; + is_extension_declared |= true; } - Ok(is_index_declared) + // SST compression + if parser.parse_keyword(Keyword::COMPRESSION) { + ensure!( + column_extensions.compression_options.is_none(), + InvalidColumnOptionSnafu { + name: column_name.to_string(), + msg: "duplicated COMPRESSION option", + } + ); + + let options = parser + .parse_options(Keyword::WITH) + .context(error::SyntaxSnafu)? + .into_iter() + .map(parse_option_string) + .collect::>>()?; + ensure!( + !options.is_empty(), + InvalidColumnOptionSnafu { + name: column_name.to_string(), + msg: "expect WITH (...) after COMPRESSION keyword", + } + ); + for (key, value) in options.iter() { + ensure!( + value.as_string().is_some(), + InvalidColumnOptionSnafu { + name: column_name.to_string(), + msg: format!("COMPRESSION option {key} requires a scalar value"), + } + ); + ensure!( + validate_column_compression_create_option(key), + InvalidColumnOptionSnafu { + name: column_name.to_string(), + msg: format!("invalid COMPRESSION option: {key}"), + } + ); + } + + column_extensions.compression_options = Some(OptionMap::new(options)); + is_extension_declared |= true; + } + + Ok(is_extension_declared) } fn parse_optional_table_constraint(&mut self) -> Result> { @@ -3194,6 +3239,82 @@ CREATE TABLE log ( } } + #[test] + fn test_parse_create_table_compression_options() { + let parse = |sql: &str| { + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + }; + + for value in ["[3]", "NULL", "['zstd']"] { + for key in ["type", "level", "page_size"] { + let sql = format!( + "CREATE TABLE t (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH ({key}={value}))" + ); + let err = parse(&sql).unwrap_err(); + assert!( + err.to_string().contains("requires a scalar value"), + "{sql}: {err}" + ); + } + } + + let sql = r" +CREATE TABLE spans ( + ts TIMESTAMP TIME INDEX, + prompt STRING COMPRESSION WITH (type='zstd', level='3', page_size='8MiB') SKIPPING INDEX, + reply STRING NULL COMPRESSION WITH (level='9'), +)"; + let result = parse(sql).unwrap(); + let Statement::CreateTable(c) = &result[0] else { + panic!("should be create_table statement"); + }; + let prompt = &c.columns[1]; + assert_eq!( + vec!["level=3", "page_size=8MiB", "type=zstd"], + prompt + .extensions + .compression_options + .as_ref() + .unwrap() + .kv_pairs() + .into_iter() + .map(|kv| kv.replace(['\'', ' '], "")) + .sorted() + .collect::>() + ); + assert!(prompt.extensions.skipping_index_options.is_some()); + // Display must produce SQL that parses back to the same column. + let reparsed = parse(&format!( + "CREATE TABLE t (ts TIMESTAMP TIME INDEX, {prompt})" + )) + .unwrap(); + let Statement::CreateTable(r) = &reparsed[0] else { + panic!("should be create_table statement"); + }; + assert_eq!(prompt.extensions, r.columns[1].extensions); + + for (sql, err) in [ + ( + "CREATE TABLE t (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION)", + "expect WITH (...) after COMPRESSION keyword", + ), + ( + "CREATE TABLE t (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (codec='zstd'))", + "invalid COMPRESSION option: codec", + ), + ( + "CREATE TABLE t (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (level='3') COMPRESSION WITH (level='5'))", + "duplicated COMPRESSION option", + ), + ] { + let result = parse(sql); + assert!( + result.as_ref().is_err_and(|e| e.to_string().contains(err)), + "{sql}: {result:?}" + ); + } + } + #[test] fn test_parse_create_view_with_columns() { let sql = "CREATE VIEW test () AS SELECT * FROM NUMBERS"; diff --git a/src/sql/src/parsers/utils.rs b/src/sql/src/parsers/utils.rs index afba437045c..65744708232 100644 --- a/src/sql/src/parsers/utils.rs +++ b/src/sql/src/parsers/utils.rs @@ -28,10 +28,12 @@ use datafusion_expr::{AggregateUDF, Expr, HigherOrderUDF, ScalarUDF, TableSource use datafusion_sql::planner::{ContextProvider, SqlToRel}; use datatypes::arrow::datatypes::DataType; use datatypes::schema::{ - COLUMN_FULLTEXT_OPT_KEY_ANALYZER, COLUMN_FULLTEXT_OPT_KEY_BACKEND, - COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE, - COLUMN_FULLTEXT_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE, - COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, + COLUMN_COMPRESSION_OPT_KEY_LEVEL, COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE, + COLUMN_COMPRESSION_OPT_KEY_TYPE, COLUMN_FULLTEXT_OPT_KEY_ANALYZER, + COLUMN_FULLTEXT_OPT_KEY_BACKEND, COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, + COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE, COLUMN_FULLTEXT_OPT_KEY_GRANULARITY, + COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE, COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY, + COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, }; use snafu::{ResultExt, ensure}; use sqlparser::dialect::Dialect; @@ -375,6 +377,15 @@ pub fn validate_column_skipping_index_create_option(key: &str) -> bool { .contains(&key) } +pub fn validate_column_compression_create_option(key: &str) -> bool { + [ + COLUMN_COMPRESSION_OPT_KEY_TYPE, + COLUMN_COMPRESSION_OPT_KEY_LEVEL, + COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE, + ] + .contains(&key) +} + /// Convert an [`IntervalMonthDayNano`] to a [`Duration`]. #[cfg(feature = "enterprise")] pub fn convert_month_day_nano_to_duration( diff --git a/src/sql/src/statements.rs b/src/sql/src/statements.rs index c2c97c34567..34a4c483cae 100644 --- a/src/sql/src/statements.rs +++ b/src/sql/src/statements.rs @@ -58,8 +58,8 @@ use crate::ast::{ }; use crate::error::{ self, ConvertToGrpcDataTypeSnafu, ConvertValueSnafu, Result, - SerializeColumnDefaultConstraintSnafu, SetFulltextOptionSnafu, SetSkippingIndexOptionSnafu, - SqlCommonSnafu, + SerializeColumnDefaultConstraintSnafu, SetCompressionOptionSnafu, SetFulltextOptionSnafu, + SetSkippingIndexOptionSnafu, SqlCommonSnafu, }; use crate::statements::create::Column; pub use crate::statements::option_map::OptionMap; @@ -149,6 +149,12 @@ pub fn column_to_schema( .context(SetSkippingIndexOptionSnafu)?; } + if let Some(options) = column.extensions.build_compression_options()? { + column_schema = column_schema + .with_compression_options(options) + .context(SetCompressionOptionSnafu)?; + } + column_schema.set_inverted_index(column.extensions.inverted_index_options.is_some()); set_json2_extension(&mut column_schema, column)?; diff --git a/src/sql/src/statements/create.rs b/src/sql/src/statements/create.rs index 919c1cd31f9..13fb4cb0a3a 100644 --- a/src/sql/src/statements/create.rs +++ b/src/sql/src/statements/create.rs @@ -18,7 +18,7 @@ use std::fmt::{Display, Formatter, Write}; use common_catalog::consts::FILE_ENGINE; use datatypes::json::{JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS, JsonSettings}; use datatypes::prelude::ConcreteDataType; -use datatypes::schema::{FulltextOptions, SkippingIndexOptions}; +use datatypes::schema::{CompressionOptions, FulltextOptions, SkippingIndexOptions}; use itertools::Itertools; use serde::Serialize; use snafu::ResultExt; @@ -27,8 +27,8 @@ use sqlparser_derive::{Visit, VisitMut}; use crate::ast::{ColumnDef, Ident, ObjectName}; use crate::error::{ - InvalidFlowQuerySnafu, InvalidSqlSnafu, Result, SetFulltextOptionSnafu, - SetSkippingIndexOptionSnafu, + InvalidFlowQuerySnafu, InvalidSqlSnafu, Result, SetCompressionOptionSnafu, + SetFulltextOptionSnafu, SetSkippingIndexOptionSnafu, }; use crate::statements::query::Query as GtQuery; use crate::statements::statement::Statement; @@ -160,6 +160,8 @@ pub struct ColumnExtensions { /// /// Inverted index doesn't have options at present. There won't be any options in that map. pub inverted_index_options: Option, + /// SST compression options. + pub compression_options: Option, /// JSON2-specific column options. pub json2_options: Option, } @@ -294,6 +296,11 @@ impl Column { } } + if let Some(compression_options) = &self.extensions.compression_options { + let options = compression_options.kv_pairs(); + write!(f, " COMPRESSION WITH({})", format_list_comma!(options))?; + } + Ok(()) } } @@ -319,6 +326,15 @@ impl ColumnExtensions { )) } + pub fn build_compression_options(&self) -> Result> { + let Some(options) = self.compression_options.as_ref() else { + return Ok(None); + }; + + let options: HashMap = options.clone().into_map(); + Ok(Some(options.try_into().context(SetCompressionOptionSnafu)?)) + } + pub fn build_json_settings(&self) -> Result> { let Some(options) = &self.json2_options else { return Ok(None); diff --git a/src/store-api/src/metadata.rs b/src/store-api/src/metadata.rs index 6983bafb504..55ff240ee4a 100644 --- a/src/store-api/src/metadata.rs +++ b/src/store-api/src/metadata.rs @@ -549,6 +549,10 @@ impl RegionMetadata { /// Checks whether it is a valid column. fn validate_column_metadata(column_metadata: &ColumnMetadata) -> Result<()> { + column_metadata + .column_schema + .compression_options() + .context(InvalidSchemaSnafu)?; if column_metadata.semantic_type == SemanticType::Timestamp { ensure!( column_metadata.column_schema.data_type.is_timestamp(), diff --git a/tests/cases/standalone/common/create/create_with_compression.result b/tests/cases/standalone/common/create/create_with_compression.result new file mode 100644 index 00000000000..8b0e25a7c0f --- /dev/null +++ b/tests/cases/standalone/common/create/create_with_compression.result @@ -0,0 +1,218 @@ +CREATE TABLE spans ( + ts TIMESTAMP TIME INDEX, + trace_id STRING SKIPPING INDEX, + prompt STRING COMPRESSION WITH (type = 'zstd', level = 3, page_size = '8MiB'), + reply STRING COMPRESSION WITH (level = 9), + tokens BIGINT COMPRESSION WITH (page_size = '64KiB'), +) WITH (append_mode = 'true'); + +Affected Rows: 0 + +SHOW CREATE TABLE spans; + ++-------+--------------------------------------------------------------------------------------------------------------------+ +| Table | Create Table | ++-------+--------------------------------------------------------------------------------------------------------------------+ +| spans | CREATE TABLE IF NOT EXISTS "spans" ( | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | "trace_id" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'), | +| | "prompt" STRING NULL COMPRESSION WITH(level = '3', page_size = '8MiB', type = 'zstd'), | +| | "reply" STRING NULL COMPRESSION WITH(level = '9', type = 'zstd'), | +| | "tokens" BIGINT NULL COMPRESSION WITH(page_size = '64KiB', type = 'zstd'), | +| | TIME INDEX ("ts") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | append_mode = 'true' | +| | ) | ++-------+--------------------------------------------------------------------------------------------------------------------+ + +INSERT INTO spans VALUES + (1, 't1', 'fix the flaky test', 'running it', 120), + (2, 't1', 'fix the flaky test\nrunning it', 'patched', 340), + (3, 't2', 'another prompt', NULL, NULL); + +Affected Rows: 3 + +ADMIN flush_table('spans'); + ++----------------------------+ +| ADMIN flush_table('spans') | ++----------------------------+ +| 0 | ++----------------------------+ + +SELECT ts, trace_id, prompt, reply, tokens FROM spans ORDER BY ts; + ++-------------------------+----------+--------------------------------+------------+--------+ +| ts | trace_id | prompt | reply | tokens | ++-------------------------+----------+--------------------------------+------------+--------+ +| 1970-01-01T00:00:00.001 | t1 | fix the flaky test | running it | 120 | +| 1970-01-01T00:00:00.002 | t1 | fix the flaky test\nrunning it | patched | 340 | +| 1970-01-01T00:00:00.003 | t2 | another prompt | | | ++-------------------------+----------+--------------------------------+------------+--------+ + +DROP TABLE spans; + +Affected Rows: 0 + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (level = 0)); + +Error: 1004(InvalidArguments), Invalid compression option: invalid level: 0, expected: an integer in 1..=22 + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (type = 'lz4')); + +Error: 1004(InvalidArguments), Invalid compression option: invalid type: lz4, expected: 'zstd' + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (page_size = 'large')); + +Error: 1004(InvalidArguments), Invalid compression option: invalid page_size: large, expected: a positive size such as '8MiB' + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (codec = 'zstd')); + +Error: 1004(InvalidArguments), Invalid column option, column name: s, error: invalid COMPRESSION option: codec + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, j JSON2 COMPRESSION WITH (level = 3)); + +Error: 1004(InvalidArguments), Invalid compression option: compression options are not supported on column j of type Json2"" + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (level = [3])); + +Error: 1004(InvalidArguments), Invalid column option, column name: s, error: COMPRESSION option level requires a scalar value + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (page_size = NULL)); + +Error: 1004(InvalidArguments), Invalid column option, column name: s, error: COMPRESSION option page_size requires a scalar value + +CREATE TABLE bad ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3) +) WITH ('sst_format' = 'primary_key', 'memtable.type' = 'time_series'); + +Error: 1004(InvalidArguments), Invalid region options, COMPRESSION on tag column host requires flat SST format and dense primary key encoding + +CREATE TABLE bad ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3) +) WITH ('primary_key_encoding' = 'sparse'); + +Error: 1004(InvalidArguments), Invalid region options, COMPRESSION on tag column host requires flat SST format and dense primary key encoding + +CREATE TABLE compressed_tags ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3, page_size = '64KiB'), + val DOUBLE +) WITH ('sst_format' = 'flat', 'primary_key_encoding' = 'dense'); + +Affected Rows: 0 + +SHOW CREATE TABLE compressed_tags; + ++-----------------+-----------------------------------------------------------------------------------------+ +| Table | Create Table | ++-----------------+-----------------------------------------------------------------------------------------+ +| compressed_tags | CREATE TABLE IF NOT EXISTS "compressed_tags" ( | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | "host" STRING NULL COMPRESSION WITH(level = '3', page_size = '64KiB', type = 'zstd'), | +| | "val" DOUBLE NULL, | +| | TIME INDEX ("ts"), | +| | PRIMARY KEY ("host") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | primary_key_encoding = 'dense', | +| | sst_format = 'flat' | +| | ) | ++-----------------+-----------------------------------------------------------------------------------------+ + +INSERT INTO compressed_tags VALUES (1, 'host-a', 1), (2, 'host-b', 2); + +Affected Rows: 2 + +ADMIN flush_table('compressed_tags'); + ++--------------------------------------+ +| ADMIN flush_table('compressed_tags') | ++--------------------------------------+ +| 0 | ++--------------------------------------+ + +SELECT * FROM compressed_tags ORDER BY ts; + ++-------------------------+--------+-----+ +| ts | host | val | ++-------------------------+--------+-----+ +| 1970-01-01T00:00:00.001 | host-a | 1.0 | +| 1970-01-01T00:00:00.002 | host-b | 2.0 | ++-------------------------+--------+-----+ + +DROP TABLE compressed_tags; + +Affected Rows: 0 + +CREATE TABLE compression_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE COMPRESSION WITH (level = 3) +) ENGINE = metric WITH (physical_metric_table = ''); + +Affected Rows: 0 + +CREATE TABLE compression_conflict ( + ts TIMESTAMP TIME INDEX, + val DOUBLE COMPRESSION WITH (level = 9) +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +Error: 1004(InvalidArguments), Unexpected request: COMPRESSION is not supported on logical tables; configure it on the physical table + +CREATE TABLE compression_same ( + ts TIMESTAMP TIME INDEX, + val DOUBLE COMPRESSION WITH (level = 3) +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +Error: 1004(InvalidArguments), Unexpected request: COMPRESSION is not supported on logical tables; configure it on the physical table + +CREATE TABLE compression_new_tag ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3), + val DOUBLE +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +Error: 1004(InvalidArguments), Unexpected request: COMPRESSION is not supported on logical tables; configure it on the physical table + +CREATE TABLE compression_logical ( + ts TIMESTAMP TIME INDEX, + val DOUBLE +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +Affected Rows: 0 + +INSERT INTO compression_logical VALUES (1, 42); + +Affected Rows: 1 + +ADMIN flush_table('compression_phy'); + ++--------------------------------------+ +| ADMIN flush_table('compression_phy') | ++--------------------------------------+ +| 0 | ++--------------------------------------+ + +SELECT * FROM compression_logical; + ++-------------------------+------+ +| ts | val | ++-------------------------+------+ +| 1970-01-01T00:00:00.001 | 42.0 | ++-------------------------+------+ + +DROP TABLE compression_logical; + +Affected Rows: 0 + +DROP TABLE compression_phy; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/create/create_with_compression.sql b/tests/cases/standalone/common/create/create_with_compression.sql new file mode 100644 index 00000000000..166c92be952 --- /dev/null +++ b/tests/cases/standalone/common/create/create_with_compression.sql @@ -0,0 +1,96 @@ +CREATE TABLE spans ( + ts TIMESTAMP TIME INDEX, + trace_id STRING SKIPPING INDEX, + prompt STRING COMPRESSION WITH (type = 'zstd', level = 3, page_size = '8MiB'), + reply STRING COMPRESSION WITH (level = 9), + tokens BIGINT COMPRESSION WITH (page_size = '64KiB'), +) WITH (append_mode = 'true'); + +SHOW CREATE TABLE spans; + +INSERT INTO spans VALUES + (1, 't1', 'fix the flaky test', 'running it', 120), + (2, 't1', 'fix the flaky test\nrunning it', 'patched', 340), + (3, 't2', 'another prompt', NULL, NULL); + +ADMIN flush_table('spans'); + +SELECT ts, trace_id, prompt, reply, tokens FROM spans ORDER BY ts; + +DROP TABLE spans; + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (level = 0)); + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (type = 'lz4')); + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (page_size = 'large')); + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (codec = 'zstd')); + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, j JSON2 COMPRESSION WITH (level = 3)); + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (level = [3])); + +CREATE TABLE bad (ts TIMESTAMP TIME INDEX, s STRING COMPRESSION WITH (page_size = NULL)); + +CREATE TABLE bad ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3) +) WITH ('sst_format' = 'primary_key', 'memtable.type' = 'time_series'); + +CREATE TABLE bad ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3) +) WITH ('primary_key_encoding' = 'sparse'); + +CREATE TABLE compressed_tags ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3, page_size = '64KiB'), + val DOUBLE +) WITH ('sst_format' = 'flat', 'primary_key_encoding' = 'dense'); + +SHOW CREATE TABLE compressed_tags; + +INSERT INTO compressed_tags VALUES (1, 'host-a', 1), (2, 'host-b', 2); + +ADMIN flush_table('compressed_tags'); + +SELECT * FROM compressed_tags ORDER BY ts; + +DROP TABLE compressed_tags; + +CREATE TABLE compression_phy ( + ts TIMESTAMP TIME INDEX, + val DOUBLE COMPRESSION WITH (level = 3) +) ENGINE = metric WITH (physical_metric_table = ''); + +CREATE TABLE compression_conflict ( + ts TIMESTAMP TIME INDEX, + val DOUBLE COMPRESSION WITH (level = 9) +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +CREATE TABLE compression_same ( + ts TIMESTAMP TIME INDEX, + val DOUBLE COMPRESSION WITH (level = 3) +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +CREATE TABLE compression_new_tag ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY COMPRESSION WITH (level = 3), + val DOUBLE +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +CREATE TABLE compression_logical ( + ts TIMESTAMP TIME INDEX, + val DOUBLE +) ENGINE = metric WITH (on_physical_table = 'compression_phy'); + +INSERT INTO compression_logical VALUES (1, 42); + +ADMIN flush_table('compression_phy'); + +SELECT * FROM compression_logical; + +DROP TABLE compression_logical; + +DROP TABLE compression_phy; diff --git a/tests/compatibility/cases/column_compression/case.toml b/tests/compatibility/cases/column_compression/case.toml new file mode 100644 index 00000000000..6369410dc08 --- /dev/null +++ b/tests/compatibility/cases/column_compression/case.toml @@ -0,0 +1,8 @@ +name = "column_compression" +reason = "Read legacy column metadata after upgrade, then persist compression options and read compressed SSTs across restart." +introduced_by = "PR #9426" +topologies = ["distributed", "standalone"] +from_range = ["*"] +to_range = [">=v1.3.0-beta.1"] +features = ["table", "compression", "write", "read"] +owner = "storage" diff --git a/tests/compatibility/cases/column_compression/setup.sql b/tests/compatibility/cases/column_compression/setup.sql new file mode 100644 index 00000000000..ec5aa23bd9a --- /dev/null +++ b/tests/compatibility/cases/column_compression/setup.sql @@ -0,0 +1,3 @@ +CREATE TABLE legacy_payload (ts TIMESTAMP TIME INDEX, payload STRING); +INSERT INTO legacy_payload VALUES (1, 'before upgrade'); +ADMIN FLUSH_TABLE('legacy_payload'); diff --git a/tests/compatibility/cases/column_compression/verify.result b/tests/compatibility/cases/column_compression/verify.result new file mode 100644 index 00000000000..d4d9809582f --- /dev/null +++ b/tests/compatibility/cases/column_compression/verify.result @@ -0,0 +1,63 @@ +SELECT * FROM legacy_payload ORDER BY ts; + ++-------------------------+----------------+ +| ts | payload | ++-------------------------+----------------+ +| 1970-01-01T00:00:00.001 | before upgrade | ++-------------------------+----------------+ + +CREATE TABLE compressed_payload ( + ts TIMESTAMP TIME INDEX, + payload STRING COMPRESSION WITH (level = 3, page_size = '64KiB') +); + +Affected Rows: 0 + +INSERT INTO compressed_payload SELECT * FROM legacy_payload; + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('compressed_payload'); + ++-----------------------------------------+ +| ADMIN FLUSH_TABLE('compressed_payload') | ++-----------------------------------------+ +| 0 | ++-----------------------------------------+ + +-- SQLNESS ARG restart=true +SHOW CREATE TABLE compressed_payload; + ++--------------------+--------------------------------------------------------------------------------------------+ +| Table | Create Table | ++--------------------+--------------------------------------------------------------------------------------------+ +| compressed_payload | CREATE TABLE IF NOT EXISTS "compressed_payload" ( | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | "payload" STRING NULL COMPRESSION WITH(level = '3', page_size = '64KiB', type = 'zstd'), | +| | TIME INDEX ("ts") | +| | ) | +| | | +| | ENGINE=mito | +| | | ++--------------------+--------------------------------------------------------------------------------------------+ + +INSERT INTO compressed_payload VALUES (2, 'after restart'); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('compressed_payload'); + ++-----------------------------------------+ +| ADMIN FLUSH_TABLE('compressed_payload') | ++-----------------------------------------+ +| 0 | ++-----------------------------------------+ + +SELECT * FROM compressed_payload ORDER BY ts; + ++-------------------------+----------------+ +| ts | payload | ++-------------------------+----------------+ +| 1970-01-01T00:00:00.001 | before upgrade | +| 1970-01-01T00:00:00.002 | after restart | ++-------------------------+----------------+ diff --git a/tests/compatibility/cases/column_compression/verify.sql b/tests/compatibility/cases/column_compression/verify.sql new file mode 100644 index 00000000000..4ddf8082317 --- /dev/null +++ b/tests/compatibility/cases/column_compression/verify.sql @@ -0,0 +1,15 @@ +SELECT * FROM legacy_payload ORDER BY ts; + +CREATE TABLE compressed_payload ( + ts TIMESTAMP TIME INDEX, + payload STRING COMPRESSION WITH (level = 3, page_size = '64KiB') +); +INSERT INTO compressed_payload SELECT * FROM legacy_payload; +ADMIN FLUSH_TABLE('compressed_payload'); + +-- SQLNESS ARG restart=true +SHOW CREATE TABLE compressed_payload; + +INSERT INTO compressed_payload VALUES (2, 'after restart'); +ADMIN FLUSH_TABLE('compressed_payload'); +SELECT * FROM compressed_payload ORDER BY ts;