chore(cubestore): Upgrade DF: Make test_queue_item_sort pass consistently

Author: srh

rust/cubestore/Cargo.lock

@@ -43,7 +43,7 @@ async-trait = "0.1.36"
 cubestore = { path = "../cubestore" }
 flate2 = "1.0.22"
 futures = "0.3.5"
-itertools = "0.9.0"
+itertools = "0.14.0"
 lazy_static = "1.4.0"
 log = "0.4.11"
 pretty_assertions = "0.7.1"

rust/cubestore/cubestore-sql-tests/Cargo.toml

@@ -56,7 +56,7 @@ lazy_static = "1.4.0"
 mockall = "0.8.1"
 async-std = "0.99"
 async-stream = "0.3.6"
-itertools = "0.11.0"
+itertools = "0.14.0"
 bigdecimal = { version = "0.2.0", features = ["serde"] }
 # Right now, it's not possible to use the 0.33 release because it has bugs
 # At the same time, 0.34-rc has a problem with large files uploading because it doesn't control number of parallels put(s)

rust/cubestore/cubestore/Cargo.toml

@@ -442,11 +442,22 @@ mod tests {
     #[test]
     fn test_queue_item_sort() -> Result<(), CubeError> {
         let priority0_1 = QueueItem::new("1".to_string(), QueueItemStatus::Active, 0, None);
-        let priority0_2 = QueueItem::new("2".to_string(), QueueItemStatus::Active, 0, None);
-        let priority0_3 = QueueItem::new("3".to_string(), QueueItemStatus::Active, 0, None);
-        let priority10_4 = QueueItem::new("4".to_string(), QueueItemStatus::Active, 10, None);
-        let priority0_5 = QueueItem::new("5".to_string(), QueueItemStatus::Active, 0, None);
-        let priority_n5_6 = QueueItem::new("6".to_string(), QueueItemStatus::Active, -5, None);
+        let mut priority0_2 = QueueItem::new("2".to_string(), QueueItemStatus::Active, 0, None);
+        let mut priority0_3 = QueueItem::new("3".to_string(), QueueItemStatus::Active, 0, None);
+        let mut priority10_4 = QueueItem::new("4".to_string(), QueueItemStatus::Active, 10, None);
+        let mut priority0_5 = QueueItem::new("5".to_string(), QueueItemStatus::Active, 0, None);
+        let mut priority_n5_6 = QueueItem::new("6".to_string(), QueueItemStatus::Active, -5, None);
+
+        // Force timestamps to be distinct (on systems that are too fast or have low clock resolution)
+        for (i, item) in (1..).zip([
+            &mut priority0_2,
+            &mut priority0_3,
+            &mut priority10_4,
+            &mut priority0_5,
+            &mut priority_n5_6,
+        ]) {
+            item.created = priority0_1.created + Duration::milliseconds(i);
+        }
 
         assert_eq!(
             vec![
@@ -491,7 +502,7 @@ mod tests {
                 "3".to_string(),
                 "5".to_string(),
                 "6".to_string()
-            ]
+            ],
         );
 
         Ok(())

rust/cubestore/cubestore/src/cachestore/queue_item.rs

@@ -464,7 +464,7 @@ impl SchedulerImpl {
 
         for (table_id, handles) in &without_failed
             .into_iter()
-            .group_by(|(h, _)| h.get_row().table_id())
+            .chunk_by(|(h, _)| h.get_row().table_id())
         {
             let mut seq_pointer_by_location = None;
             let mut ids = Vec::new();