feat(cubestore): Queue - expose processingId in queue_add (#6540)

Author: ovr

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -8074,30 +8074,92 @@ async fn limit_pushdown_unique_key(service: Box<dyn SqlClient>) {
 }
 
 async fn queue_full_workflow(service: Box<dyn SqlClient>) {
-    service
+    let add_response = service
         .exec_query(r#"QUEUE ADD PRIORITY 1 "STANDALONE#queue:1" "payload1";"#)
         .await
         .unwrap();
+    assert_queue_add_columns(&add_response);
+    assert_eq!(
+        add_response.get_rows(),
+        &vec![Row::new(vec![
+            TableValue::String("1".to_string()),
+            TableValue::Boolean(true),
+            TableValue::Int(1)
+        ])]
+    );
 
-    service
+    let add_response = service
         .exec_query(r#"QUEUE ADD PRIORITY 10 "STANDALONE#queue:2" "payload2";"#)
         .await
         .unwrap();
+    assert_queue_add_columns(&add_response);
+    assert_eq!(
+        add_response.get_rows(),
+        &vec![Row::new(vec![
+            TableValue::String("2".to_string()),
+            TableValue::Boolean(true),
+            TableValue::Int(2)
+        ])]
+    );
 
-    service
+    let add_response = service
         .exec_query(r#"QUEUE ADD PRIORITY 100 "STANDALONE#queue:3" "payload3";"#)
         .await
         .unwrap();
+    assert_queue_add_columns(&add_response);
+    assert_eq!(
+        add_response.get_rows(),
+        &vec![Row::new(vec![
+            TableValue::String("3".to_string()),
+            TableValue::Boolean(true),
+            TableValue::Int(3)
+        ])]
+    );
 
-    service
+    let add_response = service
         .exec_query(r#"QUEUE ADD PRIORITY 50 "STANDALONE#queue:4" "payload4";"#)
         .await
         .unwrap();
+    assert_queue_add_columns(&add_response);
+    assert_eq!(
+        add_response.get_rows(),
+        &vec![Row::new(vec![
+            TableValue::String("4".to_string()),
+            TableValue::Boolean(true),
+            TableValue::Int(4)
+        ])]
+    );
 
-    service
+    let add_response = service
         .exec_query(r#"QUEUE ADD PRIORITY -1 "STANDALONE#queue:5" "payload5";"#)
         .await
         .unwrap();
+    assert_queue_add_columns(&add_response);
+    assert_eq!(
+        add_response.get_rows(),
+        &vec![Row::new(vec![
+            TableValue::String("5".to_string()),
+            TableValue::Boolean(true),
+            TableValue::Int(5)
+        ])]
+    );
+
+    // deduplication check
+    {
+        let add_response = service
+            .exec_query(r#"QUEUE ADD PRIORITY 1 "STANDALONE#queue:1" "payload1";"#)
+            .await
+            .unwrap();
+        assert_queue_add_columns(&add_response);
+        assert_eq!(
+            add_response.get_rows(),
+            &vec![Row::new(vec![
+                TableValue::String("1".to_string()),
+                TableValue::Boolean(false),
+                TableValue::Int(5)
+            ])]
+        );
+    }
 
     {
         let pending_response = service
@@ -8273,6 +8335,17 @@ async fn queue_full_workflow(service: Box<dyn SqlClient>) {
     }
 }
 
+fn assert_queue_add_columns(response: &Arc<DataFrame>) {
+    assert_eq!(
+        response.get_columns(),
+        &vec![
+            Column::new("id".to_string(), ColumnType::String, 0),
+            Column::new("added".to_string(), ColumnType::Boolean, 1),
+            Column::new("pending".to_string(), ColumnType::Int, 2),
+        ]
+    );
+}
+
 fn assert_queue_retrieve_columns(response: &Arc<DataFrame>) {
     assert_eq!(
         response.get_columns(),

rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs

@@ -332,6 +332,7 @@ impl RocksCacheStore {
 
 #[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq, Hash)]
 pub struct QueueAddResponse {
+    pub id: u64,
     pub added: bool,
     pub pending: u64,
 }
@@ -572,15 +573,16 @@ impl CacheStore for RocksCacheStore {
                 let index_key = QueueItemIndexKey::ByPath(item.get_path());
                 let id_row_opt = queue_schema
                     .get_single_opt_row_by_index(&index_key, &QueueItemRocksIndex::ByPath)?;
-                let added = if id_row_opt.is_none() {
-                    queue_schema.insert(item, batch_pipe)?;
 
-                    true
+                let (id, added) = if let Some(row) = id_row_opt {
+                    (row.id, false)
                 } else {
-                    false
+                    let row = queue_schema.insert(item, batch_pipe)?;
+                    (row.id, true)
                 };
 
                 Ok(QueueAddResponse {
+                    id,
                     added,
                     pending: if added { pending + 1 } else { pending },
                 })

rust/cubestore/cubestore/src/sql/cachestore.rs

@@ -173,10 +173,12 @@ impl CacheStoreSqlService {
                 (
                     Arc::new(DataFrame::new(
                         vec![
-                            Column::new("added".to_string(), ColumnType::Boolean, 0),
-                            Column::new("pending".to_string(), ColumnType::Int, 1),
+                            Column::new("id".to_string(), ColumnType::String, 0),
+                            Column::new("added".to_string(), ColumnType::Boolean, 1),
+                            Column::new("pending".to_string(), ColumnType::Int, 2),
                         ],
                         vec![Row::new(vec![
+                            TableValue::String(response.id.to_string()),
                             TableValue::Boolean(response.added),
                             TableValue::Int(response.pending as i64),
                         ])],