chore: modify the underlying data structure of the pipeline value map…

… type from hashmap to btremap to keep key order

src/pipeline/src/etl.rs

@@ -214,16 +214,19 @@ where
         match val {
             Value::Map(map) => {
                 let mut search_from = 0;
+                // because of the key in the json map is ordered
                 for (payload_key, payload_value) in map.values.into_iter() {
                     if search_from >= self.required_keys.len() {
                         break;
                     }
 
+                    // because of map key is ordered, required_keys is ordered too
                     if let Some(pos) = self.required_keys[search_from..]
                         .iter()
                         .position(|k| k == &payload_key)
                     {
                         result[search_from + pos] = payload_value;
+                        // next search from is always after the current key
                         search_from += pos;
                     }
                 }

src/pipeline/src/etl/processor/cmcd.rs

@@ -402,7 +402,8 @@ impl Processor for CmcdProcessor {
 
 #[cfg(test)]
 mod tests {
-    use ahash::HashMap;
+    use std::collections::BTreeMap;
+
     use urlencoding::decode;
 
     use super::{CmcdProcessorBuilder, CMCD_KEYS};
@@ -563,14 +564,14 @@ mod tests {
             let values = vec
                 .into_iter()
                 .map(|(k, v)| (k.to_string(), v))
-                .collect::<HashMap<String, Value>>();
+                .collect::<BTreeMap<String, Value>>();
             let expected = Map { values };
 
             let actual = processor.parse(0, &decoded).unwrap();
             let actual = actual
                 .into_iter()
                 .map(|(index, value)| (intermediate_keys[index].clone(), value))
-                .collect::<HashMap<String, Value>>();
+                .collect::<BTreeMap<String, Value>>();
             let actual = Map { values: actual };
             assert_eq!(actual, expected);
         }

src/pipeline/src/etl/processor/regex.rs

@@ -383,6 +383,8 @@ impl Processor for RegexProcessor {
 }
 #[cfg(test)]
 mod tests {
+    use std::collections::BTreeMap;
+
     use ahash::{HashMap, HashMapExt};
     use itertools::Itertools;
 
@@ -475,14 +477,14 @@ ignore_missing: false"#;
             .map(|k| k.to_string())
             .collect_vec();
             let processor = builder.build(&intermediate_keys).unwrap();
-            let mut result = HashMap::new();
+            let mut result = BTreeMap::new();
             for (index, pattern) in processor.patterns.iter().enumerate() {
                 let r = processor
                     .process(&breadcrumbs_str, pattern, (0, index))
                     .unwrap()
                     .into_iter()
                     .map(|(k, v)| (intermediate_keys[k].clone(), v))
-                    .collect::<HashMap<_, _>>();
+                    .collect::<BTreeMap<_, _>>();
                 result.extend(r);
             }
             let map = Map { values: result };

src/pipeline/src/etl/value.rs

@@ -16,7 +16,8 @@ pub mod array;
 pub mod map;
 pub mod time;
 
-use ahash::{HashMap, HashMapExt};
+use std::collections::BTreeMap;
+
 pub use array::Array;
 use jsonb::{Number as JsonbNumber, Object as JsonbObject, Value as JsonbValue};
 pub use map::Map;
@@ -289,7 +290,7 @@ impl TryFrom<serde_json::Value> for Value {
                 Ok(Value::Array(Array { values }))
             }
             serde_json::Value::Object(v) => {
-                let mut values = HashMap::with_capacity(v.len());
+                let mut values = BTreeMap::new();
                 for (k, v) in v {
                     values.insert(k, Value::try_from(v)?);
                 }
@@ -320,7 +321,7 @@ impl TryFrom<&yaml_rust::Yaml> for Value {
                 Ok(Value::Array(Array { values }))
             }
             yaml_rust::Yaml::Hash(v) => {
-                let mut values = HashMap::new();
+                let mut values = BTreeMap::new();
                 for (k, v) in v {
                     let key = k
                         .as_str()

src/pipeline/src/etl/value/map.rs

@@ -12,19 +12,21 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use ahash::{HashMap, HashMapExt};
+use std::collections::BTreeMap;
+
+use ahash::HashMap;
 
 use crate::etl::value::Value;
 
 #[derive(Debug, Clone, PartialEq)]
 pub struct Map {
-    pub values: HashMap<String, Value>,
+    pub values: BTreeMap<String, Value>,
 }
 
 impl Default for Map {
     fn default() -> Self {
         Self {
-            values: HashMap::with_capacity(30),
+            values: BTreeMap::default(),
         }
     }
 }
@@ -47,12 +49,16 @@ impl Map {
 
 impl From<HashMap<String, Value>> for Map {
     fn from(values: HashMap<String, Value>) -> Self {
-        Map { values }
+        let mut map = Map::default();
+        for (k, v) in values.into_iter() {
+            map.insert(k, v);
+        }
+        map
     }
 }
 
 impl std::ops::Deref for Map {
-    type Target = HashMap<String, Value>;
+    type Target = BTreeMap<String, Value>;
 
     fn deref(&self) -> &Self::Target {
         &self.values

src/servers/src/otlp/logs.rs

@@ -14,7 +14,6 @@
 
 use std::collections::{BTreeMap, HashMap as StdHashMap};
 
-use ahash::{HashMap, HashMapExt};
 use api::v1::column_data_type_extension::TypeExt;
 use api::v1::value::ValueData;
 use api::v1::{
@@ -144,7 +143,7 @@ fn log_to_pipeline_value(
     let log_attrs = PipelineValue::Map(Map {
         values: key_value_to_map(log.attributes),
     });
-    let mut map = HashMap::new();
+    let mut map = BTreeMap::new();
     map.insert(
         "Timestamp".to_string(),
         PipelineValue::Uint64(log.time_unix_nano),
@@ -485,8 +484,8 @@ fn any_value_to_pipeline_value(value: any_value::Value) -> PipelineValue {
 }
 
 // convert otlp keyValue vec to map
-fn key_value_to_map(key_values: Vec<KeyValue>) -> HashMap<String, PipelineValue> {
-    let mut map = HashMap::new();
+fn key_value_to_map(key_values: Vec<KeyValue>) -> BTreeMap<String, PipelineValue> {
+    let mut map = BTreeMap::new();
     for kv in key_values {
         let value = match kv.value {
             Some(value) => match value.value {