diff --git a/src/config/src/config.rs b/src/config/src/config.rs
index 1b952d621e..2fbc94d958 100644
--- a/src/config/src/config.rs
+++ b/src/config/src/config.rs
@@ -36,7 +36,10 @@ use serde::{Deserialize, Serialize};
 use sha256::digest;
 
 use crate::{
-    meta::{cluster, stream::QueryPartitionStrategy},
+    meta::{
+        cluster,
+        stream::{QueryPartitionStrategy, StreamType},
+    },
     utils::sysinfo,
 };
 
@@ -562,6 +565,14 @@ impl std::str::FromStr for FileFormat {
 }
 
 impl FileFormat {
+    pub fn for_ingester_stream(stream_type: StreamType, configured: Self) -> Self {
+        if stream_type == StreamType::Metrics {
+            Self::Parquet
+        } else {
+            configured
+        }
+    }
+
     pub fn extension(&self) -> &'static str {
         match self {
             Self::Parquet => FILE_EXT_PARQUET,
@@ -3780,6 +3791,22 @@ mod tests {
         assert_eq!(FileFormat::Vortex.extension(), ".vortex");
     }
 
+    #[test]
+    fn test_file_format_for_ingester_stream() {
+        assert_eq!(
+            FileFormat::for_ingester_stream(StreamType::Metrics, FileFormat::Vortex),
+            FileFormat::Parquet
+        );
+        assert_eq!(
+            FileFormat::for_ingester_stream(StreamType::Logs, FileFormat::Vortex),
+            FileFormat::Vortex
+        );
+        assert_eq!(
+            FileFormat::for_ingester_stream(StreamType::Traces, FileFormat::Parquet),
+            FileFormat::Parquet
+        );
+    }
+
     #[test]
     fn test_file_format_from_extension() {
         assert_eq!(
diff --git a/src/infra/src/cache/file_data/mod.rs b/src/infra/src/cache/file_data/mod.rs
index d1e1bb09cf..32be605448 100644
--- a/src/infra/src/cache/file_data/mod.rs
+++ b/src/infra/src/cache/file_data/mod.rs
@@ -52,6 +52,7 @@ enum CacheStrategy {
 enum FileType {
     Parquet,
     Ttv,
+    Vortex,
 }
 
 impl CacheStrategy {
@@ -220,6 +221,15 @@ async fn validate_file(bytes: &[u8], ftype: FileType) -> Result<(), anyhow::Erro
                 return Err(anyhow::anyhow!("payload size mismatch"));
             }
         }
+        FileType::Vortex => {
+            if bytes.len() < 12 {
+                return Err(anyhow::anyhow!("invalid vortex file"));
+            }
+            const VORTEX_MAGIC: &[u8; 4] = b"VTXF";
+            if &bytes[..4] != VORTEX_MAGIC || &bytes[bytes.len() - 4..] != VORTEX_MAGIC {
+                return Err(anyhow::anyhow!("vortex magic bytes mismatch"));
+            }
+        }
     }
     Ok(())
 }
@@ -289,16 +299,20 @@ async fn download_from_storage(
                 // so we check if the footer is valid. If it is, then the db entry is invalid
                 // and we reset it. If footer is invalid, the store has a corrupted file
                 // so we mark it as deleted, and return error.
+                // data files (parquet/vortex) are tracked in file_list, ttv index files are not
+                let is_data_file = file.ends_with(".parquet") || file.ends_with(".vortex");
                 let valid_parquet = file.ends_with(".parquet")
                     && validate_file(&data_bytes, FileType::Parquet).await.is_ok();
+                let valid_vortex = file.ends_with(".vortex")
+                    && validate_file(&data_bytes, FileType::Vortex).await.is_ok();
                 let valid_ttv = file.ends_with(".ttv")
                     && validate_file(&data_bytes, FileType::Ttv).await.is_ok();
-                if valid_parquet || valid_ttv {
+                if valid_parquet || valid_vortex || valid_ttv {
                     log::warn!(
                         "download file {file} found size mismatch, remote : {expected_blob_size}, db: {size}, correcting db as valid file",
                     );
-                    // only update for parquet files, not ttv files
-                    if file.ends_with(".parquet") {
+                    // only update for data files, not ttv files
+                    if is_data_file {
                         crate::file_list::update_compressed_size(file, data_len as i64).await?;
                         crate::file_list::LOCAL_CACHE
                             .update_compressed_size(file, data_len as i64)
@@ -309,8 +323,8 @@ async fn download_from_storage(
                     log::warn!(
                         "download file {file} found corrupt file, remote: {expected_blob_size}, db: {size}, deleting entry from file_list "
                     );
-                    // only update for parquet files, not ttv files
-                    if file.ends_with(".parquet") {
+                    // only update for data files, not ttv files
+                    if is_data_file {
                         crate::file_list::remove(file).await?;
                         crate::file_list::LOCAL_CACHE.remove(file).await?;
                     }
diff --git a/src/job/files/mod.rs b/src/job/files/mod.rs
index c9b3183604..1de7e2547c 100644
--- a/src/job/files/mod.rs
+++ b/src/job/files/mod.rs
@@ -14,6 +14,7 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 use config::{
+    FileFormat,
     cluster::{LOCAL_NODE, is_offline},
     ider,
     meta::stream::StreamType,
@@ -54,11 +55,12 @@ async fn clean_empty_dirs() -> Result<(), anyhow::Error> {
     Ok(())
 }
 
-pub fn generate_storage_file_name(
+pub fn generate_ingester_storage_file_key(
     org_id: &str,
     stream_type: StreamType,
     stream_name: &str,
     wal_file_name: &str,
+    file_format: FileFormat,
 ) -> String {
     // eg: 0/2023/08/21/08/8b8a5451bbe1c44b/ip=1234/7099303408192061440f3XQ2p.json
     let file_columns = wal_file_name.splitn(7, '/').collect::<Vec<&str>>();
@@ -76,6 +78,42 @@ pub fn generate_storage_file_name(
     } else {
         format!("{}/{}", &file_name[..file_name_pos], id)
     };
-    let file_format = config::get_config().common.file_format.extension();
-    format!("files/{stream_key}/{file_date}/{file_name}{file_format}")
+    format!(
+        "files/{stream_key}/{file_date}/{file_name}{}",
+        file_format.extension()
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use config::{FileFormat, meta::stream::StreamType};
+
+    use super::generate_ingester_storage_file_key;
+
+    #[test]
+    fn test_ingester_storage_file_key_uses_given_format_extension() {
+        let file_name = generate_ingester_storage_file_key(
+            "default",
+            StreamType::Metrics,
+            "quickstart",
+            "0/2026/07/02/12/hash/service_name=ingester/source.json",
+            FileFormat::Parquet,
+        );
+
+        assert!(
+            file_name.starts_with(
+                "files/default/metrics/quickstart/2026/07/02/12/service_name=ingester/"
+            )
+        );
+        assert!(file_name.ends_with(".parquet"));
+
+        let file_name = generate_ingester_storage_file_key(
+            "default",
+            StreamType::Logs,
+            "quickstart",
+            "0/2026/07/02/12/hash/service_name=ingester/source.json",
+            FileFormat::Vortex,
+        );
+        assert!(file_name.ends_with(".vortex"));
+    }
 }
diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs
index c5377e9719..dcd3c090a0 100644
--- a/src/job/files/parquet.rs
+++ b/src/job/files/parquet.rs
@@ -22,8 +22,6 @@ use std::{
 use arrow_schema::Schema;
 use bytes::Bytes;
 use chrono::{Duration, Utc};
-#[cfg(feature = "enterprise")]
-use config::utils::parquet::get_recordbatch_reader_from_bytes;
 use config::{
     FxIndexMap, cluster, get_config,
     meta::{
@@ -51,6 +49,19 @@ use tokio::{
     fs::remove_file,
     sync::{Mutex, RwLock},
 };
+#[cfg(feature = "enterprise")]
+use {
+    config::{FileFormat, cluster::LOCAL_NODE, utils::parquet::get_recordbatch_reader_from_bytes},
+    o2_enterprise::enterprise::{
+        common::config::get_config as get_enterprise_config,
+        service_streams::{
+            batch_processor::queue_services, cache::get_coverage_deficit, meta::ServiceMetadata,
+            processor::StreamProcessor, sampler::should_process_file,
+        },
+    },
+    std::collections::HashMap,
+    tokio::sync::mpsc,
+};
 
 use crate::{
     common::infra::wal,
@@ -800,8 +811,12 @@ async fn merge_files(
         }
     };
 
-    let (buf, mut new_file_meta) = match buf {
-        MergeParquetResult::Single(buf, meta) => (buf, meta),
+    let (buf, mut new_file_meta, file_format) = match buf {
+        MergeParquetResult::Single {
+            buf,
+            file_meta,
+            file_format,
+        } => (buf, file_meta, file_format),
         MergeParquetResult::Multiple { .. } => {
             // ingester should not support multiple files, it will be handled in compactor mode
             panic!("[INGESTER:JOB] merge_parquet_files error: multiple files");
@@ -813,8 +828,13 @@ async fn merge_files(
             "merge_parquet_files error: compressed_size is 0"
         ));
     }
-    let new_file_key =
-        super::generate_storage_file_name(&org_id, stream_type, &stream_name, &file_name);
+    let new_file_key = super::generate_ingester_storage_file_key(
+        &org_id,
+        stream_type,
+        &stream_name,
+        &file_name,
+        file_format,
+    );
     log::info!(
         "[INGESTER:JOB:{thread_id}] merged {} files into a new file: {new_file_key}, original_size: {}, compressed_size: {}, took: {} ms",
         retain_file_list.len(),
@@ -840,9 +860,7 @@ async fn merge_files(
     // This runs BEFORE indexing checks to ensure all stream types are discovered
     #[cfg(feature = "enterprise")]
     {
-        use config::cluster::LOCAL_NODE;
-        let service_streams_config =
-            &o2_enterprise::enterprise::common::config::get_config().service_streams;
+        let service_streams_config = &get_enterprise_config().service_streams;
 
         let valid_stream_type = stream_type == StreamType::Logs
             || stream_type == StreamType::Metrics
@@ -863,25 +881,20 @@ async fn merge_files(
             // active), tapering back to normal as coverage fills in.
             // Passing (0, 0) when disabled is the documented no-op sentinel for the sampler.
             let coverage_deficit = if service_streams_config.coverage_catchup_enabled {
-                o2_enterprise::enterprise::service_streams::cache::get_coverage_deficit(
-                    &org_id,
-                    stream_type,
-                )
-                .await
+                get_coverage_deficit(&org_id, stream_type).await
             } else {
                 (0, 0)
             };
 
             // Check if we should process this file (adaptive per-type sampling with fast lane)
-            let should_process =
-                o2_enterprise::enterprise::service_streams::sampler::should_process_file(
-                    &org_id,
-                    stream_type,
-                    &stream_name,
-                    &new_file_key,
-                    stream_count,
-                    coverage_deficit,
-                );
+            let should_process = should_process_file(
+                &org_id,
+                stream_type,
+                &stream_name,
+                &new_file_key,
+                stream_count,
+                coverage_deficit,
+            );
 
             if should_process {
                 let buf_clone = buf.clone();
@@ -890,20 +903,17 @@ async fn merge_files(
 
                 // Queue services for batched processing (non-blocking)
                 tokio::spawn(async move {
-                    if let Err(e) = queue_services_from_parquet(
+                    if let Err(e) = queue_services_from_data_file(
                         &org_id_clone,
                         stream_type,
                         &stream_name_clone,
-                        &buf_clone,
+                        file_format,
+                        buf_clone,
                     )
                     .await
                     {
                         log::error!(
-                            "[ServiceStreams] Failed to queue services for {}/{}/{}: {}",
-                            org_id_clone,
-                            stream_type,
-                            stream_name_clone,
-                            e
+                            "[ServiceStreams] Failed to queue services for {org_id_clone}/{stream_type}/{stream_name_clone}: {e}"
                         );
                     }
                 });
@@ -961,16 +971,13 @@ fn split_perfix(prefix: &str) -> (String, StreamType, String, String) {
 }
 
 #[cfg(feature = "enterprise")]
-async fn queue_services_from_parquet(
+async fn queue_services_from_data_file(
     org_id: &str,
     stream_type: StreamType,
     stream_name: &str,
-    parquet_data: &[u8],
+    file_format: FileFormat,
+    file_data: Bytes,
 ) -> Result<(), anyhow::Error> {
-    use std::collections::HashMap;
-
-    use tokio::sync::mpsc;
-
     let start = std::time::Instant::now();
 
     // Get semantic field groups upfront (before spawning tasks)
@@ -982,27 +989,21 @@ async fn queue_services_from_parquet(
     // Create bounded channel for backpressure - drops records if consumer can't keep up
     // ARROW-NATIVE: Channel now sends RecordBatch directly (no HashMap conversion!)
     let (tx, mut rx) = mpsc::channel::<arrow::record_batch::RecordBatch>(
-        o2_enterprise::enterprise::common::config::get_config()
-            .service_streams
-            .channel_capacity,
+        get_enterprise_config().service_streams.channel_capacity,
     );
 
-    // Clone data needed for producer task
-    let parquet_bytes = Bytes::copy_from_slice(parquet_data);
-
-    // Spawn producer task to read parquet and send Arrow batches through channel
+    // Spawn producer task to read the data file and send Arrow batches through channel.
     // ARROW-NATIVE: No HashMap conversion! Sends RecordBatch directly.
     let producer_handle = tokio::spawn(async move {
         let mut records_sent = 0u64;
         let mut records_dropped = 0u64;
         let mut batches_sent = 0u64;
 
-        let file_format = config::get_config().common.file_format;
-        let reader_result = get_recordbatch_reader_from_bytes(file_format, parquet_bytes).await;
+        let reader_result = get_recordbatch_reader_from_bytes(file_format, file_data).await;
         let (_schema, mut reader) = match reader_result {
             Ok(r) => r,
             Err(e) => {
-                log::error!("[ServiceStreams] Failed to read parquet: {e}");
+                log::error!("[ServiceStreams] Failed to read {file_format} file: {e}");
                 return (records_sent, records_dropped);
             }
         };
@@ -1051,16 +1052,9 @@ async fn queue_services_from_parquet(
     });
 
     // Consumer: Process batches as they arrive
-    let processor = o2_enterprise::enterprise::service_streams::processor::StreamProcessor::new(
-        org_id.to_string(),
-        semantic_groups,
-        identity_config,
-    );
+    let processor = StreamProcessor::new(org_id.to_string(), semantic_groups, identity_config);
 
-    let mut all_services: HashMap<
-        String,
-        o2_enterprise::enterprise::service_streams::meta::ServiceMetadata,
-    > = HashMap::new();
+    let mut all_services: HashMap<String, ServiceMetadata> = HashMap::new();
     let mut total_records_processed = 0u64;
 
     // Process Arrow batches from channel
@@ -1089,10 +1083,7 @@ async fn queue_services_from_parquet(
 
     if records_dropped > 0 {
         log::warn!(
-            "[ServiceStreams] Dropped {} records due to backpressure for {}/{}",
-            records_dropped,
-            org_id,
-            stream_name
+            "[ServiceStreams] Dropped {records_dropped} records due to backpressure for {org_id}/{stream_name}",
         );
         metrics::SERVICE_STREAMS_RECORDS_DROPPED
             .with_label_values(&[org_id, &stream_type.to_string()])
@@ -1115,22 +1106,16 @@ async fn queue_services_from_parquet(
         .observe(duration.as_secs_f64());
 
     log::debug!(
-        "[ServiceStreams] Processed {} records, discovered {} services in {:?} for {}/{} (sent: {}, dropped: {})",
-        total_records_processed,
-        service_count,
-        duration,
-        org_id,
-        stream_name,
-        records_sent,
-        records_dropped
+        "[ServiceStreams] Processed {total_records_processed} records, discovered {service_count} services in {duration:?} for {org_id}/{stream_name} (sent: {records_sent}, dropped: {records_dropped})",
     );
 
     // Queue services for batched processing
-    o2_enterprise::enterprise::service_streams::batch_processor::queue_services(
-        org_id.to_string(),
-        all_services,
-    )
-    .await;
+    queue_services(org_id.to_string(), all_services).await;
+
+    let duration = start.elapsed().as_millis();
+    log::info!(
+        "[ServiceStreams] queue_services_from_data_file completed in {duration} ms for {org_id}/{stream_name} (processed: {total_records_processed}, services: {service_count}, sent: {records_sent}, dropped: {records_dropped})",
+    );
 
     Ok(())
 }
diff --git a/src/job/mod.rs b/src/job/mod.rs
index 07070ddc2a..cda2be9b4b 100644
--- a/src/job/mod.rs
+++ b/src/job/mod.rs
@@ -751,7 +751,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
     tokio::task::spawn(alert_grouping::process_expired_batches());
     tokio::task::spawn(file_downloader::run());
     // Note: Service discovery extraction runs automatically during parquet file processing
-    // See src/job/files/parquet.rs:queue_services_from_parquet for implementation
+    // See src/job/files/parquet.rs:queue_services_from_data_file for implementation
     #[cfg(feature = "enterprise")]
     spawn_pausable_job!(
         "service_streams_batch_processor",
diff --git a/src/service/compact/deleted.rs b/src/service/compact/deleted.rs
index 69457aa3d8..92e0469e0f 100644
--- a/src/service/compact/deleted.rs
+++ b/src/service/compact/deleted.rs
@@ -86,11 +86,7 @@ pub async fn delete(org_id: &str, time_max: i64) -> Result<i64, anyhow::Error> {
             if file.flattened {
                 Some((
                     file.account.to_string(),
-                    format!(
-                        "files{}/{}",
-                        config::get_config().common.column_all,
-                        file.file.strip_prefix("files/").unwrap()
-                    ),
+                    super::flatten::generate_flatten_file_key(&file.file),
                 ))
             } else {
                 None
diff --git a/src/service/compact/flatten.rs b/src/service/compact/flatten.rs
index 26975e661b..48b750849c 100644
--- a/src/service/compact/flatten.rs
+++ b/src/service/compact/flatten.rs
@@ -149,6 +149,21 @@ pub async fn generate_by_stream(
     Ok(())
 }
 
+/// Generates the storage key of the flattened version of a data file.
+pub fn generate_flatten_file_key(file_key: &str) -> String {
+    let key = file_key.strip_prefix("files/").unwrap_or(file_key);
+    let key = match FileFormat::from_extension(key) {
+        Some(format) => key.strip_suffix(format.extension()).unwrap_or(key),
+        None => key,
+    };
+    format!(
+        "files{}/{}{}",
+        get_config().common.column_all,
+        key,
+        FileFormat::Parquet.extension()
+    )
+}
+
 pub async fn generate_file(file: &FileKey) -> Result<(), anyhow::Error> {
     let start = std::time::Instant::now();
     log::debug!("[FLATTEN_COMPACTOR] generate flatten file for {}", file.key);
@@ -172,11 +187,7 @@ pub async fn generate_file(file: &FileKey) -> Result<(), anyhow::Error> {
             file.key
         ));
     }
-    let new_file = format!(
-        "files{}/{}",
-        get_config().common.column_all,
-        file.key.strip_prefix("files/").unwrap()
-    );
+    let new_file = generate_flatten_file_key(&file.key);
     let org_id = columns[1];
     let stream_type = StreamType::from(columns[2]);
     let stream_name = columns[3];
@@ -304,6 +315,20 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_generate_flatten_file_key_always_uses_parquet_extension() {
+        let column_all = &get_config().common.column_all;
+        assert_eq!(
+            generate_flatten_file_key("files/default/logs/quickstart/2026/07/02/12/abc.parquet"),
+            format!("files{column_all}/default/logs/quickstart/2026/07/02/12/abc.parquet")
+        );
+        // flattened output is written as parquet even when the source is vortex
+        assert_eq!(
+            generate_flatten_file_key("files/default/logs/quickstart/2026/07/02/12/abc.vortex"),
+            format!("files{column_all}/default/logs/quickstart/2026/07/02/12/abc.parquet")
+        );
+    }
+
     // Helper function to create a record batch with the "_all" field
     fn create_test_batch_with_all_field(
         all_field_values: Vec<Option<&str>>,
diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs
index 87436d05d6..4fc48342d0 100644
--- a/src/service/compact/merge.rs
+++ b/src/service/compact/merge.rs
@@ -879,7 +879,11 @@ pub async fn merge_files(
 
     let mut new_files = Vec::new();
     match buf {
-        MergeParquetResult::Single(buf, mut new_file_meta) => {
+        MergeParquetResult::Single {
+            buf,
+            file_meta: mut new_file_meta,
+            file_format,
+        } => {
             if new_file_meta.compressed_size == 0 {
                 return Err(anyhow::anyhow!(
                     "merge_parquet_files error: compressed_size is 0"
@@ -887,8 +891,7 @@ pub async fn merge_files(
             }
 
             let id = ider::generate_file_name();
-            let file_format = get_config().common.file_format.extension();
-            let new_file_key = format!("{prefix}/{id}{file_format}");
+            let new_file_key = format!("{prefix}/{id}{}", file_format.extension());
             log::info!(
                 "[COMPACTOR:WORKER:{thread_id}] merged {} files into a new file: {new_file_key}, original_size: {}, compressed_size: {}, took: {} ms",
                 retain_file_list.len(),
@@ -930,7 +933,11 @@ pub async fn merge_files(
             }
             new_files.push(FileKey::new(0, account, new_file_key, new_file_meta, false));
         }
-        MergeParquetResult::Multiple { bufs, file_metas } => {
+        MergeParquetResult::Multiple {
+            bufs,
+            file_metas,
+            file_format,
+        } => {
             for (buf, file_meta) in bufs.into_iter().zip(file_metas) {
                 let mut new_file_meta = file_meta;
                 new_file_meta.compressed_size = buf.len() as i64;
@@ -941,8 +948,7 @@ pub async fn merge_files(
                 }
 
                 let id = ider::generate_file_name();
-                let file_format = get_config().common.file_format.extension();
-                let new_file_key = format!("{prefix}/{id}{file_format}");
+                let new_file_key = format!("{prefix}/{id}{}", file_format.extension());
 
                 // upload file to storage
                 let buf = Bytes::from(buf);
diff --git a/src/service/search/datafusion/merge/downsampling.rs b/src/service/search/datafusion/merge/downsampling.rs
index 0ce9cc3993..c651f34417 100644
--- a/src/service/search/datafusion/merge/downsampling.rs
+++ b/src/service/search/datafusion/merge/downsampling.rs
@@ -57,6 +57,7 @@ pub async fn merge_parquet_files_with_downsampling(
     bloom_filter_fields: &[String],
     rule: &DownsamplingRule,
     metadata: &FileMeta,
+    file_format: FileFormat,
 ) -> Result<MergeParquetResult> {
     let start = std::time::Instant::now();
     let cfg = get_config();
@@ -107,7 +108,7 @@ pub async fn merge_parquet_files_with_downsampling(
     });
 
     // Write batches to the appropriate format
-    let (bufs, file_metas) = match cfg.common.file_format {
+    let (bufs, file_metas) = match file_format {
         FileFormat::Parquet => {
             write_downsampled_parquet(rx, &schema, bloom_filter_fields, metadata, &cfg).await?
         }
@@ -133,7 +134,11 @@ pub async fn merge_parquet_files_with_downsampling(
         start.elapsed().as_millis()
     );
 
-    Ok(MergeParquetResult::Multiple { bufs, file_metas })
+    Ok(MergeParquetResult::Multiple {
+        bufs,
+        file_metas,
+        file_format,
+    })
 }
 
 async fn write_downsampled_parquet(
diff --git a/src/service/search/datafusion/merge/mod.rs b/src/service/search/datafusion/merge/mod.rs
index 53fb50aa39..0ee49c3d1c 100644
--- a/src/service/search/datafusion/merge/mod.rs
+++ b/src/service/search/datafusion/merge/mod.rs
@@ -42,11 +42,16 @@ use {
 };
 
 pub enum MergeParquetResult {
-    Single(Vec<u8>, FileMeta),
+    Single {
+        buf: Vec<u8>,
+        file_meta: FileMeta,
+        file_format: FileFormat,
+    },
     #[allow(unused)]
     Multiple {
         bufs: Vec<Vec<u8>>,
         file_metas: Vec<FileMeta>,
+        file_format: FileFormat,
     },
 }
 
@@ -62,6 +67,8 @@ pub async fn merge_parquet_files(
     let start = std::time::Instant::now();
     let cfg = get_config();
 
+    let file_format = merge_output_file_format(stream_type, is_ingester, cfg.common.file_format);
+
     #[cfg(feature = "enterprise")]
     if stream_type == StreamType::Metrics && !is_ingester {
         let rule = get_largest_downsampling_rule(stream_name, metadata.max_ts);
@@ -75,6 +82,7 @@ pub async fn merge_parquet_files(
                 bloom_filter_fields,
                 rule,
                 &metadata,
+                file_format,
             )
             .await;
         }
@@ -150,8 +158,7 @@ pub async fn merge_parquet_files(
         Ok(())
     });
 
-    // write batches to the appropriate format
-    let buf = match cfg.common.file_format {
+    let buf = match file_format {
         FileFormat::Parquet => {
             write_parquet(
                 &schema,
@@ -172,7 +179,23 @@ pub async fn merge_parquet_files(
     );
 
     metadata.compressed_size = buf.len() as i64;
-    Ok(MergeParquetResult::Single(buf, metadata))
+    Ok(MergeParquetResult::Single {
+        buf,
+        file_meta: metadata,
+        file_format,
+    })
+}
+
+fn merge_output_file_format(
+    stream_type: StreamType,
+    is_ingester: bool,
+    configured: FileFormat,
+) -> FileFormat {
+    if is_ingester {
+        FileFormat::for_ingester_stream(stream_type, configured)
+    } else {
+        configured
+    }
 }
 
 async fn write_parquet(
@@ -274,6 +297,26 @@ mod tests {
         ]))
     }
 
+    #[test]
+    fn test_merge_output_file_format_uses_parquet_for_ingester_metrics() {
+        assert_eq!(
+            merge_output_file_format(StreamType::Metrics, true, FileFormat::Vortex),
+            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),
+            FileFormat::Parquet
+        );
+    }
+
     #[tokio::test]
     async fn test_merge_parquet_files_error_handling() {
         // Test with empty tables vector