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

@@ -606,6 +606,123 @@ impl FileFormat {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct FileFormatConfig {
+    default: FileFormat,
+    logs: Option<FileFormat>,
+    metrics: Option<FileFormat>,
+    traces: Option<FileFormat>,
+}
+
+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<Self, Self::Err> {
+        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 <stream_type>=<file_format>"
+                )
+            })?;
+            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<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.collect_str(self)
+    }
+}
+
+impl<'de> Deserialize<'de> for FileFormatConfig {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    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::<FileFormatConfig>()
+            .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::<FileFormatConfig>().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::<FileFormatConfig>().is_err(),
+                "'{value}' should be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn test_file_format_config_serde_as_string() {
+        let config = "parquet,metrics=vortex"
+            .parse::<FileFormatConfig>()
+            .unwrap();
+        let serialized = serde_json::to_string(&config).unwrap();
+
+        assert_eq!(serialized, r#""parquet,metrics=vortex""#);
+        assert_eq!(
+            serde_json::from_str::<FileFormatConfig>(&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]

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::<FileFormatConfig>()
+            .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]