fix: nightly rustfmt import ordering

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: g-talbot

quickwit/quickwit-parquet-engine/src/merge/tests.rs

@@ -323,7 +323,9 @@ fn test_merge_per_output_metadata_from_actual_rows() {
         num_outputs: 1,
         writer_config: ParquetWriterConfig::default(),
     };
-    let outputs = merge_sorted_parquet_files(&[input1.clone(), input2.clone()], &output_dir, &config).unwrap();
+    let outputs =
+        merge_sorted_parquet_files(&[input1.clone(), input2.clone()], &output_dir, &config)
+            .unwrap();
     assert_eq!(outputs.len(), 1);
     let output = &outputs[0];
     assert!(output.metric_names.contains("cpu"));

quickwit/quickwit-parquet-engine/src/merge/writer.rs

@@ -40,7 +40,9 @@ use crate::row_keys;
 use crate::sort_fields::parse_sort_fields;
 use crate::sorted_series::SORTED_SERIES_COLUMN;
 use crate::split::TAG_SERVICE;
-use crate::storage::split_writer::{extract_metric_names, extract_service_names, extract_time_range};
+use crate::storage::split_writer::{
+    extract_metric_names, extract_service_names, extract_time_range,
+};
 use crate::storage::{
     PARQUET_META_NUM_MERGE_OPS, PARQUET_META_ROW_KEYS, PARQUET_META_ROW_KEYS_JSON,
     PARQUET_META_SORT_FIELDS, PARQUET_META_WINDOW_DURATION, PARQUET_META_WINDOW_START,
@@ -119,8 +121,8 @@ pub fn write_merge_outputs(
         // Extract per-output logical metadata from the actual rows.
         let metric_names = extract_metric_names(&sorted_batch)
             .context("extracting metric names from merge output")?;
-        let time_range = extract_time_range(&sorted_batch)
-            .context("extracting time range from merge output")?;
+        let time_range =
+            extract_time_range(&sorted_batch).context("extracting time range from merge output")?;
         let service_names = extract_service_names(&sorted_batch)
             .context("extracting service names from merge output")?;
 

quickwit/quickwit-parquet-engine/src/storage/split_writer.rs

@@ -194,69 +194,86 @@ pub(crate) fn extract_time_range(batch: &RecordBatch) -> Result<TimeRange, Parqu
 }
 
 /// Extracts distinct metric names from a RecordBatch.
-pub(crate) fn extract_metric_names(batch: &RecordBatch) -> Result<HashSet<String>, ParquetWriteError> {
+pub(crate) fn extract_metric_names(
+    batch: &RecordBatch,
+) -> Result<HashSet<String>, ParquetWriteError> {
     let metric_idx = batch
         .schema()
         .index_of("metric_name")
         .map_err(|_| ParquetWriteError::SchemaValidation("missing metric_name column".into()))?;
-    let metric_col = batch.column(metric_idx);
-    let mut names = HashSet::new();
-
-    // The column is Dictionary(Int32, Utf8)
-    if let Some(dict_array) = metric_col
-        .as_any()
-        .downcast_ref::<arrow::array::DictionaryArray<arrow::datatypes::Int32Type>>()
-    {
-        let values = dict_array.values();
-        if let Some(string_values) = values.as_any().downcast_ref::<arrow::array::StringArray>() {
-            // Get all dictionary values that are actually used
-            for i in 0..dict_array.len() {
-                if !dict_array.is_null(i)
-                    && let Ok(key) = dict_array.keys().value(i).try_into()
-                {
-                    let key: usize = key;
-                    if key < string_values.len() && !string_values.is_null(key) {
-                        names.insert(string_values.value(key).to_string());
-                    }
-                }
-            }
-        }
-    }
-
-    Ok(names)
+    extract_distinct_strings(batch.column(metric_idx))
 }
 
 /// Extracts distinct service names from a RecordBatch.
-pub(crate) fn extract_service_names(batch: &RecordBatch) -> Result<HashSet<String>, ParquetWriteError> {
+pub(crate) fn extract_service_names(
+    batch: &RecordBatch,
+) -> Result<HashSet<String>, ParquetWriteError> {
     let service_idx = match batch.schema().index_of("service").ok() {
         Some(idx) => idx,
         None => return Ok(HashSet::new()),
     };
-    let service_col = batch.column(service_idx);
-    let mut names = HashSet::new();
+    extract_distinct_strings(batch.column(service_idx))
+}
 
-    // The column is Dictionary(Int32, Utf8)
-    if let Some(dict_array) = service_col
+/// Extracts distinct non-null string values from a column.
+///
+/// Handles both Dictionary(Int32, Utf8) encoding (common at ingest) and
+/// plain Utf8/LargeUtf8 (possible after optimize_output_batch in the merge
+/// path when cardinality is too high for dictionary encoding).
+fn extract_distinct_strings(
+    col: &dyn arrow::array::Array,
+) -> Result<HashSet<String>, ParquetWriteError> {
+    let mut values = HashSet::new();
+
+    // Try Dictionary(Int32, Utf8) first — the common case at ingest.
+    if let Some(dict_array) = col
         .as_any()
         .downcast_ref::<arrow::array::DictionaryArray<arrow::datatypes::Int32Type>>()
+        && let Some(string_values) = dict_array
+            .values()
+            .as_any()
+            .downcast_ref::<arrow::array::StringArray>()
     {
-        let values = dict_array.values();
-        if let Some(string_values) = values.as_any().downcast_ref::<arrow::array::StringArray>() {
-            // Get all dictionary values that are actually used
-            for i in 0..dict_array.len() {
-                if !dict_array.is_null(i)
-                    && let Ok(key) = dict_array.keys().value(i).try_into()
-                {
-                    let key: usize = key;
-                    if key < string_values.len() && !string_values.is_null(key) {
-                        names.insert(string_values.value(key).to_string());
-                    }
+        for i in 0..dict_array.len() {
+            if !dict_array.is_null(i)
+                && let Ok(key) = dict_array.keys().value(i).try_into()
+            {
+                let key: usize = key;
+                if key < string_values.len() && !string_values.is_null(key) {
+                    values.insert(string_values.value(key).to_string());
                 }
             }
         }
+        return Ok(values);
+    }
+
+    // Fall back to plain Utf8 (after optimize_output_batch strips dictionary
+    // encoding for high-cardinality columns).
+    if let Some(string_array) = col.as_any().downcast_ref::<arrow::array::StringArray>() {
+        for i in 0..string_array.len() {
+            if !string_array.is_null(i) {
+                values.insert(string_array.value(i).to_string());
+            }
+        }
+        return Ok(values);
+    }
+
+    // LargeUtf8 variant.
+    if let Some(string_array) = col
+        .as_any()
+        .downcast_ref::<arrow::array::LargeStringArray>()
+    {
+        for i in 0..string_array.len() {
+            if !string_array.is_null(i) {
+                values.insert(string_array.value(i).to_string());
+            }
+        }
+        return Ok(values);
     }
 
-    Ok(names)
+    // Unrecognized column type — return empty rather than error, since the
+    // column may legitimately be a type we don't extract strings from.
+    Ok(values)
 }
 
 #[cfg(test)]