feat(aggregation, stats): add missing value handling for collapse ops and improve group by error handling

- add exclude_missing and missing_val config to Collapse calculator
- skip empty field values when exclude_missing is true
- replace empty values with configured missing_val in formatted output
- add tests for new missing value handling features
- fix KeyExtractor strict mode flag for stats command
- add proper error propagation for out-of-bounds group by fields
- remove redundant group by sorting and simplify iteration
- clean up obsolete group by test cases

Author: wang-q

src/cmd_tva/stats.rs

@@ -467,7 +467,7 @@ pub fn execute(matches: &ArgMatches) -> anyhow::Result<()> {
             if idxs.is_empty() {
                 None
             } else {
-                Some(KeyExtractor::new(Some(idxs), false, false)) // strict=false for stats
+                Some(KeyExtractor::new(Some(idxs), false, true)) // strict=true for stats
             }
         } else {
             None
@@ -546,10 +546,17 @@ pub fn execute(matches: &ArgMatches) -> anyhow::Result<()> {
                             .as_mut()
                             .unwrap()
                             .extract_from_row(row, opt_delimiter);
-                        let key = match key_res {
-                            Ok(k) => k.into_owned(),
-                            Err(_) => KeyBuffer::new(),
-                        };
+                        let key = key_res
+                            .map_err(|idx| {
+                                std::io::Error::new(
+                                    std::io::ErrorKind::InvalidData,
+                                    format!(
+                                        "Not enough fields in line for group-by key. Required field index: {}",
+                                        idx
+                                    ),
+                                )
+                            })?
+                            .into_owned();
 
                         let agg = groups
                             .entry(key)
@@ -568,11 +575,7 @@ pub fn execute(matches: &ArgMatches) -> anyhow::Result<()> {
 
     if let Some(proc) = &processor {
         if use_grouping {
-            let mut keys: Vec<_> = groups.keys().collect();
-            keys.sort();
-
-            for key in keys {
-                let agg = &groups[key];
+            for (key, agg) in &groups {
                 print!("{}", String::from_utf8_lossy(key));
 
                 let values = proc.format_results(agg);

src/libs/aggregation/ops/text.rs

@@ -38,15 +38,29 @@ pub struct Collapse {
     pub field_idx: usize,
     pub string_values_slot: usize,
     pub delimiter: String,
+    pub exclude_missing: bool,
+    pub missing_val: Option<String>,
 }
 
 impl Calculator for Collapse {
     fn update(&self, agg: &mut Aggregator, row: &dyn Row) {
-        agg.string_values[self.string_values_slot].push(get_str(row, self.field_idx));
+        let val = get_str(row, self.field_idx);
+        if self.exclude_missing && val.is_empty() {
+            return;
+        }
+        agg.string_values[self.string_values_slot].push(val);
     }
 
     fn format(&self, agg: &Aggregator) -> String {
-        agg.string_values[self.string_values_slot].join(&self.delimiter)
+        let vals = &agg.string_values[self.string_values_slot];
+        if let Some(ref replacement) = self.missing_val {
+            vals.iter()
+                .map(|v| if v.is_empty() { replacement.as_str() } else { v.as_str() })
+                .collect::<Vec<&str>>()
+                .join(&self.delimiter)
+        } else {
+            vals.join(&self.delimiter)
+        }
     }
 }
 
@@ -138,14 +152,52 @@ mod tests {
             field_idx: 0,
             string_values_slot: 0,
             delimiter: ",".to_string(),
+            exclude_missing: false,
+            missing_val: None,
+        };
+
+        calc.update(&mut agg, &StrSliceRow { fields: &["A"] });
+        calc.update(&mut agg, &StrSliceRow { fields: &["B"] });
+
+        assert_eq!(calc.format(&agg), "A,B");
+    }
+
+    #[test]
+    fn test_collapse_exclude_missing() {
+        let mut agg = new_agg();
+        let calc = Collapse {
+            field_idx: 0,
+            string_values_slot: 0,
+            delimiter: ",".to_string(),
+            exclude_missing: true,
+            missing_val: None,
         };
 
         calc.update(&mut agg, &StrSliceRow { fields: &["A"] });
+        calc.update(&mut agg, &StrSliceRow { fields: &[""] });
         calc.update(&mut agg, &StrSliceRow { fields: &["B"] });
 
         assert_eq!(calc.format(&agg), "A,B");
     }
 
+    #[test]
+    fn test_collapse_replace_missing() {
+        let mut agg = new_agg();
+        let calc = Collapse {
+            field_idx: 0,
+            string_values_slot: 0,
+            delimiter: ",".to_string(),
+            exclude_missing: false,
+            missing_val: Some("NA".to_string()),
+        };
+
+        calc.update(&mut agg, &StrSliceRow { fields: &["A"] });
+        calc.update(&mut agg, &StrSliceRow { fields: &[""] });
+        calc.update(&mut agg, &StrSliceRow { fields: &["B"] });
+
+        assert_eq!(calc.format(&agg), "A,NA,B");
+    }
+
     #[test]
     fn test_rand() {
         let mut agg = new_agg();

src/libs/aggregation/processor.rs

@@ -334,6 +334,8 @@ impl StatsProcessor {
                             field_idx: idx,
                             string_values_slot: slot,
                             delimiter: config.delimiter.to_string(),
+                            exclude_missing: config.exclude_missing,
+                            missing_val: config.missing_val.clone(),
                         }));
                     }
                 }