fix: reject duplicate sorted_series column instead of silently skipping

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

Author: g-talbot

quickwit/quickwit-parquet-engine/src/sorted_series/mod.rs

@@ -102,13 +102,20 @@ pub fn compute_sorted_series_column(
 
 /// Append the sorted_series column to a [`RecordBatch`].
 ///
-/// If the column already exists, returns the batch unchanged.
+/// # Errors
+///
+/// Returns an error if the column already exists (indicates a bug in
+/// the caller), if the sort fields string cannot be parsed, or if
+/// storekey encoding fails.
 pub fn append_sorted_series_column(
     sort_fields_str: &str,
     batch: &RecordBatch,
 ) -> Result<RecordBatch> {
     if batch.schema().index_of(SORTED_SERIES_COLUMN).is_ok() {
-        return Ok(batch.clone());
+        anyhow::bail!(
+            "batch already contains a '{}' column — double computation is a bug",
+            SORTED_SERIES_COLUMN
+        );
     }
 
     let sorted_series = compute_sorted_series_column(sort_fields_str, batch)?;

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

@@ -395,14 +395,17 @@ fn test_append_sorted_series_column() {
 }
 
 #[test]
-fn test_append_sorted_series_idempotent() {
+fn test_append_sorted_series_rejects_duplicate() {
     let batch = build_test_batch(&[TestRow::new("cpu.usage", Some("api"), None, 100)]);
 
     let first = append_sorted_series_column(METRICS_SORT_FIELDS, &batch).unwrap();
-    let second = append_sorted_series_column(METRICS_SORT_FIELDS, &first).unwrap();
+    let err = append_sorted_series_column(METRICS_SORT_FIELDS, &first).unwrap_err();
 
-    // Second call should be a no-op.
-    assert_eq!(first.num_columns(), second.num_columns());
+    assert!(
+        err.to_string().contains("already contains"),
+        "expected duplicate-column error, got: {}",
+        err
+    );
 }
 
 // -----------------------------------------------------------------------