rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-14 07:52:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add per-column COMPRESSION option for SST columns (#9426)

Browse Source 
* feat: add per-column COMPRESSION option for SST columns

CREATE TABLE accepts COMPRESSION WITH (type = 'zstd', level = N,
page_size = '<size>') on a column. The options are stored in the column
metadata, carried through gRPC column options, shown by SHOW CREATE
TABLE, and applied by the mito SST writer and bulk part encoder as
parquet column properties. Unset fields keep the SST defaults. Nested
types are rejected.

A page size below the dictionary page limit also lowers the column's
dictionary page limit: parquet sizes write batches by the dictionary
budget and keeps that size after falling back to plain encoding, which
would otherwise ignore the smaller data page limit.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: validate column compression options before persistence

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: reject compression overrides on metric logical tables

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
dennis zhuang
2026-10-13 08:03:49 +00:00
committed by GitHub
parent 2039f508fc
commit 2a70194930
29 changed files with 1340 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/src/error.rs
+10 -1
View File
@@ -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(),
}
}
src/api/src/v1/column_def.rs
+73 -7
View File
@@ -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<ColumnSchema> {
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> {
}
}
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<Column
.options
.insert(SKIPPING_INDEX_GRPC_KEY.to_string(), skipping_index.clone());
}
if let Some(compression) = column_schema.metadata().get(COMPRESSION_KEY) {
options
.options
.insert(COMPRESSION_GRPC_KEY.to_string(), compression.clone());
}
if let Some(extension_name) = column_schema.metadata().get(EXTENSION_TYPE_NAME_KEY) {
options
.options
@@ -258,8 +275,9 @@ pub fn as_skipping_index_type(skipping_index_type: PbSkippingIndexType) -> 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);
src/datatypes/src/error.rs
+9 -1
View File
@@ -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 { .. }
src/datatypes/src/schema.rs
+8 -6
View File
@@ -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;
src/datatypes/src/schema/column_schema.rs
+275
View File
@@ -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<Option<CompressionOptions>> {
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> {
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<E>(&self) -> Result<Option<E>>
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<i32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub page_size: Option<ReadableSize>,
}
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<String, String> {
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<i32> = 1..=22;
impl TryFrom<HashMap<String, String>> for CompressionOptions {
type Error = Error;
fn try_from(options: HashMap<String, String>) -> Result<Self> {
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::<i32>()
.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<HashMap<String, String>> 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::<HashMap<_, _>>(),
)
};
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::<Vec<_>>()
)
.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\"}";
src/metric-engine/src/engine/alter/validate.rs
+53 -1
View File
@@ -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")
);
}
}
src/metric-engine/src/engine/create.rs
+12 -1
View File
@@ -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 {
src/mito2/src/engine/create_test.rs
+68
View File
@@ -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;
src/mito2/src/error.rs
+10
View File
@@ -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, .. }
src/mito2/src/memtable/bulk/part.rs
+3 -1
View File
@@ -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 {
src/mito2/src/region/opener.rs
+2
View File
@@ -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, &region_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, &region_options)?;
region_options.validate_column_compression(&metadata)?;
let region_id = self.region_id;
let provider = self.provider::<S>(&region_options.wal_options)?;
src/mito2/src/region/options.rs
+29
View File
@@ -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
src/mito2/src/sst/parquet.rs
+175 -2
View File
@@ -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<usize> =
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<WriterPropertiesBuilder> {
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<String> = (0..256).map(|i| format!("word{i:03}")).collect();
let mut seed = 42u64;
let values: Vec<Vec<u8>> = (0..2048)
.map(|_| {
(0..250)
.map(|_| {
seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1);
words[(seed >> 33) as usize % words.len()].as_str()
})
.collect::<Vec<_>>()
.join(" ")
.into_bytes()
})
.collect();
async fn write(
values: &[Vec<u8>],
compression: Option<CompressionOptions>,
) -> (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::<UInt32Type>::new();
let mut pks = BinaryDictionaryBuilder::<UInt32Type>::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;
src/mito2/src/sst/parquet/writer.rs
+2 -1
View File
@@ -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(
src/mito2/src/worker/handle_alter.rs
+13
View File
@@ -197,6 +197,10 @@ impl<S: LogStore> RegionWorkerLoop<S> {
};
// 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<S: LogStore> RegionWorkerLoop<S> {
}
.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)
src/operator/src/expr_helper.rs
+9 -1
View File
@@ -375,6 +375,14 @@ pub fn expr_to_create(expr: &CreateTableExpr, quote_style: Option<char>) -> 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`)
)
src/query/src/sql/show_create_table.rs
+4
View File
@@ -169,6 +169,10 @@ fn create_column(column_schema: &ColumnSchema, quote_style: char) -> Result<Colu
extensions.inverted_index_options = Some(HashMap::new().into());
}
if let Some(opt) = column_schema.compression_options()? {
extensions.compression_options = Some(opt.to_options_map().into());
}
let mut data_type = concrete_data_type_to_sql_data_type(&column_schema.data_type)
.with_context(|_| ConvertSqlTypeSnafu {
datatype: column_schema.data_type.clone(),
src/sql/src/error.rs
+8
View File
@@ -294,6 +294,13 @@ pub enum Error {
location: Location,
},
#[snafu(display("Failed to set COMPRESSION option"))]
SetCompressionOption {
source: datatypes::error::Error,
#[snafu(implicit)]
location: Location,
},
#[snafu(display(
"Invalid partition number: {}, should be in range [2, 65536]",
partition_num
@@ -405,6 +412,7 @@ impl ErrorExt for Error {
PermissionDenied { .. } => StatusCode::PermissionDenied,
SetFulltextOption { .. } | SetSkippingIndexOption { .. } => StatusCode::Unexpected,
SetCompressionOption { source, .. } => source.status_code(),
}
}
src/sql/src/parsers/create_parser.rs
+128 -7
View File
@@ -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::<Result<Vec<_>>>()?;
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<Option<TableConstraint>> {
@@ -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::<Vec<_>>()
);
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";
src/sql/src/parsers/utils.rs
+15 -4
View File
@@ -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(
src/sql/src/statements.rs
+8 -2
View File
@@ -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)?;
src/sql/src/statements/create.rs
+19 -3
View File
@@ -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<OptionMap>,
/// SST compression options.
pub compression_options: Option<OptionMap>,
/// JSON2-specific column options.
pub json2_options: Option<Json2Options>,
}
@@ -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<Option<CompressionOptions>> {
let Some(options) = self.compression_options.as_ref() else {
return Ok(None);
};
let options: HashMap<String, String> = options.clone().into_map();
Ok(Some(options.try_into().context(SetCompressionOptionSnafu)?))
}
pub fn build_json_settings(&self) -> Result<Option<JsonSettings>> {
let Some(options) = &self.json2_options else {
return Ok(None);
src/store-api/src/metadata.rs
+4
View File
@@ -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(),
tests/cases/standalone/common/create/create_with_compression.result
+218
View File
@@ -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"<Null>"
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
tests/cases/standalone/common/create/create_with_compression.sql
+96
View File
@@ -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;
tests/compatibility/cases/column_compression/case.toml
+8
View File
@@ -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"
tests/compatibility/cases/column_compression/setup.sql
+3
View File
@@ -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');
tests/compatibility/cases/column_compression/verify.result
+63
View File
@@ -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 |
+-------------------------+----------------+
tests/compatibility/cases/column_compression/verify.sql
+15
View File
@@ -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;