From ee71df7ce73cec00a9dba13dbe4b2e910e16ebfc Mon Sep 17 00:00:00 2001 From: Huaijin Date: Fri, 7 Aug 2026 12:45:31 +0800 Subject: [PATCH] chore: support per-stream file formats (#13677) ## Summary - support a default ZO_FILE_FORMAT with logs, metrics, and traces overrides - resolve the configured format when producing merged files while preserving Parquet output for metrics on ingesters - reject invalid or duplicate overrides and preserve string serialization compatibility ## Examples - ZO_FILE_FORMAT=parquet - ZO_FILE_FORMAT=vortex - ZO_FILE_FORMAT=parquet,metrics=vortex - ZO_FILE_FORMAT=vortex,logs=parquet,traces=parquet ## Testing - cargo test -p config test_file_format --lib - cargo test -p config test_common_config_preserves_file_format_overrides --lib - cargo test -p search test_merge_output_file_format_uses_parquet_for_ingester_metrics --lib - cargo clippy -p config -p search --lib -- -D warnings --- src/config/src/config.rs | 183 ++++++++++++++++++++++++- src/search/src/datafusion/merge/mod.rs | 34 +++-- 2 files changed, 200 insertions(+), 17 deletions(-) diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 4a95c80ba6..0f66ec1911 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -606,6 +606,123 @@ impl FileFormat { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct FileFormatConfig { + default: FileFormat, + logs: Option, + metrics: Option, + traces: Option, +} + +impl FileFormatConfig { + pub const fn new(default: FileFormat) -> Self { + Self { + default, + logs: None, + metrics: None, + traces: None, + } + } + + pub fn for_stream(self, stream_type: StreamType) -> FileFormat { + match stream_type { + StreamType::Logs => self.logs, + StreamType::Metrics => self.metrics, + StreamType::Traces => self.traces, + _ => None, + } + .unwrap_or(self.default) + } +} + +impl Default for FileFormatConfig { + fn default() -> Self { + Self::new(FileFormat::default()) + } +} + +impl std::fmt::Display for FileFormatConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.default)?; + for (stream_type, file_format) in [ + (StreamType::Logs, self.logs), + (StreamType::Metrics, self.metrics), + (StreamType::Traces, self.traces), + ] { + if let Some(file_format) = file_format { + write!(f, ",{stream_type}={file_format}")?; + } + } + Ok(()) + } +} + +impl std::str::FromStr for FileFormatConfig { + type Err = anyhow::Error; + + fn from_str(value: &str) -> Result { + let mut parts = value.split(','); + let default = parts + .next() + .map(str::trim) + .filter(|part| !part.is_empty() && !part.contains('=')) + .ok_or_else(|| { + anyhow::anyhow!( + "Invalid file format config: a default format must be specified first" + ) + })? + .parse()?; + let mut config = Self::new(default); + + for part in parts { + let (stream_type, file_format) = part.trim().split_once('=').ok_or_else(|| { + anyhow::anyhow!( + "Invalid file format override '{part}': expected =" + ) + })?; + let file_format = file_format.trim().parse()?; + let target = match stream_type.trim().to_lowercase().as_str() { + "logs" => &mut config.logs, + "metrics" => &mut config.metrics, + "traces" => &mut config.traces, + stream_type => { + return Err(anyhow::anyhow!( + "Invalid stream type '{stream_type}' in file format config: expected logs, metrics, or traces" + )); + } + }; + if target.replace(file_format).is_some() { + return Err(anyhow::anyhow!( + "Duplicate file format override for stream type '{}'", + stream_type.trim() + )); + } + } + + Ok(config) + } +} + +impl Serialize for FileFormatConfig { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + serializer.collect_str(self) + } +} + +impl<'de> Deserialize<'de> for FileFormatConfig { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + String::deserialize(deserializer)? + .parse() + .map_err(serde::de::Error::custom) + } +} + #[derive(Serialize, EnvConfig, Default)] pub struct Config { pub auth: Auth, @@ -1228,9 +1345,9 @@ pub struct Common { name = "ZO_FILE_FORMAT", parse, default = "parquet", - help = "File format for data storage: parquet or vortex" + help = "Default file format for data storage with optional per-stream overrides, for example: parquet,metrics=vortex" )] - pub file_format: FileFormat, + pub file_format: FileFormatConfig, #[env_config( name = "ZO_VORTEX_USE_NATIVE_COMPRESSION", default = false, @@ -4173,6 +4290,59 @@ mod tests { ); } + #[test] + fn test_file_format_config_from_str() { + let config = " parquet , LOGS=vortex, metrics = vortex, traces=parquet " + .parse::() + .unwrap(); + + assert_eq!(config.for_stream(StreamType::Logs), FileFormat::Vortex); + assert_eq!(config.for_stream(StreamType::Metrics), FileFormat::Vortex); + assert_eq!(config.for_stream(StreamType::Traces), FileFormat::Parquet); + assert_eq!(config.for_stream(StreamType::Metadata), FileFormat::Parquet); + assert_eq!( + config.to_string(), + "parquet,logs=vortex,metrics=vortex,traces=parquet" + ); + + let config = "vortex".parse::().unwrap(); + assert_eq!(config.for_stream(StreamType::Logs), FileFormat::Vortex); + assert_eq!(config.for_stream(StreamType::Metrics), FileFormat::Vortex); + assert_eq!(config.for_stream(StreamType::Traces), FileFormat::Vortex); + } + + #[test] + fn test_file_format_config_rejects_invalid_values() { + for value in [ + "", + "metrics=vortex", + "parquet,logs", + "parquet,unknown=vortex", + "parquet,metrics=unknown", + "parquet,logs=vortex,LOGS=parquet", + "parquet,vortex", + ] { + assert!( + value.parse::().is_err(), + "'{value}' should be rejected" + ); + } + } + + #[test] + fn test_file_format_config_serde_as_string() { + let config = "parquet,metrics=vortex" + .parse::() + .unwrap(); + let serialized = serde_json::to_string(&config).unwrap(); + + assert_eq!(serialized, r#""parquet,metrics=vortex""#); + assert_eq!( + serde_json::from_str::(&serialized).unwrap(), + config + ); + } + #[test] fn test_file_format_from_extension() { assert_eq!( @@ -4193,13 +4363,16 @@ mod tests { } #[test] - fn test_common_config_preserves_vortex_file_format() { + fn test_common_config_preserves_file_format_overrides() { let mut cfg = Config::init().unwrap(); - cfg.common.file_format = FileFormat::Vortex; + cfg.common.file_format = "parquet,metrics=vortex".parse().unwrap(); check_common_config(&mut cfg).unwrap(); - assert_eq!(cfg.common.file_format, FileFormat::Vortex); + assert_eq!( + cfg.common.file_format, + "parquet,metrics=vortex".parse().unwrap() + ); } #[test] diff --git a/src/search/src/datafusion/merge/mod.rs b/src/search/src/datafusion/merge/mod.rs index 20679064f2..597d640f83 100644 --- a/src/search/src/datafusion/merge/mod.rs +++ b/src/search/src/datafusion/merge/mod.rs @@ -17,7 +17,7 @@ use std::sync::Arc; use arrow::array::RecordBatch; use config::{ - FileFormat, TIMESTAMP_COL_NAME, get_config, + FileFormat, FileFormatConfig, TIMESTAMP_COL_NAME, get_config, meta::stream::{FileMeta, StreamType}, utils::{ parquet::{VORTEX_FILE_META_KEY, encode_vortex_file_meta, new_parquet_writer}, @@ -204,8 +204,9 @@ pub async fn merge_parquet_files( fn merge_output_file_format( stream_type: StreamType, is_ingester: bool, - configured: FileFormat, + configured: FileFormatConfig, ) -> FileFormat { + let configured = configured.for_stream(stream_type); if is_ingester { FileFormat::for_ingester_stream(stream_type, configured) } else { @@ -341,22 +342,31 @@ mod tests { #[test] fn test_merge_output_file_format_uses_parquet_for_ingester_metrics() { + let configured = "parquet,metrics=vortex" + .parse::() + .unwrap(); assert_eq!( - merge_output_file_format(StreamType::Metrics, true, FileFormat::Vortex), + merge_output_file_format(StreamType::Metrics, true, configured), FileFormat::Parquet ); assert_eq!( - merge_output_file_format(StreamType::Logs, true, FileFormat::Vortex), - FileFormat::Vortex - ); - assert_eq!( - merge_output_file_format(StreamType::Metrics, false, FileFormat::Vortex), - FileFormat::Vortex - ); - assert_eq!( - merge_output_file_format(StreamType::Logs, false, FileFormat::Parquet), + merge_output_file_format(StreamType::Logs, true, configured), FileFormat::Parquet ); + assert_eq!( + merge_output_file_format(StreamType::Metrics, false, configured), + FileFormat::Vortex + ); + assert_eq!( + merge_output_file_format(StreamType::Traces, false, configured), + FileFormat::Parquet + ); + + let configured = FileFormatConfig::new(FileFormat::Vortex); + assert_eq!( + merge_output_file_format(StreamType::Logs, true, configured), + FileFormat::Vortex + ); } #[tokio::test]