Skip to content

feat(cubestore): Support QUEUE ADD_AND_RETRIEVE command #11590

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ovr wants to merge 7 commits into
base: master
Choose a base branch
from
Open

feat(cubestore): Support QUEUE ADD_AND_RETRIEVE command #11590

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6fecf52
feat(cubestore): QUEUE ADD_AND_RETRIEVE command
ovr Aug 18, 2026
3d7bb9f
chore(cubestore): drop comments which repeat the implementation
ovr Aug 18, 2026
7a54bee
docs(query-orchestrator): queue diagrams + fast track
ovr Aug 19, 2026
2f53dab
feat(cubestore): bound the priority inversion of ADD_AND_RETRIEVE
ovr Aug 19, 2026
b62a000
test(cubestore): QUEUE ADD_AND_RETRIEVE backlog guard over SQL
ovr Aug 19, 2026
fe6380e
chore(cubestore): stop restating the claim rule in doc comments
ovr Aug 19, 2026
ab917ac
Update packages/cubejs-query-orchestrator/DEVELOPMENT.md
ovr Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
352 changes: 265 additions & 87 deletions packages/cubejs-query-orchestrator/DEVELOPMENT.md
View file
Open in desktop

Large diffs are not rendered by default.

208 changes: 208 additions & 0 deletions rust/cubestore/cubestore-sql-tests/src/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,6 +316,11 @@ pub fn sql_tests(prefix: &str) -> Vec<(&'static str, TestFn)> {
),
t("queue_latest_result_v1", queue_latest_result_v1),
t("queue_retrieve_extended", queue_retrieve_extended),
t("queue_add_and_retrieve", queue_add_and_retrieve),
t(
"queue_add_and_retrieve_backlog",
queue_add_and_retrieve_backlog,
),
t("queue_ack_then_result_v1", queue_ack_then_result_v1),
t("queue_ack_then_result_v2", queue_ack_then_result_v2),
t(
Expand Down Expand Up @@ -456,6 +461,8 @@ lazy_static::lazy_static! {
"prefilter_chunks_shared_scan",
"planning_topk_hash_aggregate",
"topk_hash_aggregate_trim",
"queue_add_and_retrieve",
"queue_add_and_retrieve_backlog",
].into_iter().map(ToOwned::to_owned).collect();
}

Expand Down Expand Up @@ -11881,6 +11888,41 @@ fn assert_queue_add_columns(response: &Arc<DataFrame>) {
);
}

fn queue_add_and_retrieve_row(
id: &str,
added: bool,
pending: i64,
active: Option<&str>,
payload: Option<&str>,
) -> Row {
let to_value =
|v: Option<&str>| v.map_or(TableValue::Null, |v| TableValue::String(v.to_string()));

Row::new(vec![
TableValue::String(id.to_string()),
TableValue::Boolean(added),
TableValue::Int(pending),
to_value(active),
to_value(payload),
// extra is always empty for a freshly added item
TableValue::Null,
])
}

fn assert_queue_add_and_retrieve_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),
Column::new("active".to_string(), ColumnType::String, 3),
Column::new("payload".to_string(), ColumnType::String, 4),
Column::new("extra".to_string(), ColumnType::String, 5),
]
);
}

fn assert_queue_add_and_get_id(response: &Arc<DataFrame>) -> Result<String, CubeError> {
assert_queue_add_columns(response);

Expand Down Expand Up @@ -11998,6 +12040,172 @@ async fn queue_retrieve_extended(service: Box<dyn SqlClient>) -> Result<(), Cube
Ok(())
}

async fn queue_add_and_retrieve(service: Box<dyn SqlClient>) -> Result<(), CubeError> {
{
let add_response = service
.exec_query(r#"QUEUE ADD_AND_RETRIEVE PRIORITY 1 "STANDALONE#queue:1" "payload1" 1"#)
.await?;
assert_queue_add_and_retrieve_columns(&add_response);
assert_eq!(
add_response.get_rows(),
&vec![queue_add_and_retrieve_row(
"1",
true,
0,
Some("1"),
Some("payload1")
)]
);
}

{
let add_response = service
.exec_query(r#"QUEUE ADD_AND_RETRIEVE "STANDALONE#queue:2" "payload2" 1"#)
.await?;
assert_eq!(
add_response.get_rows(),
&vec![queue_add_and_retrieve_row("2", true, 1, Some("1"), None)]
);
}

{
// The stored payload is returned, not the payload of this call
let add_response = service
.exec_query(r#"QUEUE ADD_AND_RETRIEVE "STANDALONE#queue:2" "payload2-dup" 2"#)
.await?;
assert_eq!(
add_response.get_rows(),
&vec![queue_add_and_retrieve_row(
"2",
false,
0,
Some("1,2"),
Some("payload2")
)]
);
}

{
let add_response = service
.exec_query(r#"QUEUE ADD_AND_RETRIEVE "STANDALONE#queue:1" "payload1" 5"#)
.await?;
assert_eq!(
add_response.get_rows(),
&vec![queue_add_and_retrieve_row("1", false, 0, Some("1,2"), None)]
);
}

{
let add_response = service
.exec_query(r#"QUEUE ADD "STANDALONE#queue:3" "payload3""#)
.await?;
assert_queue_add_columns(&add_response);
}

{
let pending_response = service
.exec_query(r#"QUEUE PENDING "STANDALONE#queue""#)
.await?;
assert_eq!(
pending_response.get_rows(),
&vec![Row::new(vec![
TableValue::String("3".to_string()),
TableValue::String("3".to_string()),
TableValue::String("pending".to_string()),
TableValue::Null,
]),]
);

let active_response = service
.exec_query(r#"QUEUE ACTIVE "STANDALONE#queue""#)
.await?;
assert_eq!(active_response.get_rows().len(), 2);
}

{
// A claimed item can be acknowledged without an explicit QUEUE RETRIEVE
let ack_response = service
.exec_query(r#"QUEUE ACK "STANDALONE#queue:1" "result1""#)
.await?;
assert_eq!(
ack_response.get_rows(),
&vec![Row::new(vec![TableValue::Boolean(true)])]
);

let result_response = service
.exec_query(r#"QUEUE RESULT "STANDALONE#queue:1""#)
.await?;
assert_queue_result_columns(&result_response);
assert_eq!(
result_response.get_rows(),
&vec![queue_result_row("result1", "1", None)]
);
}

Ok(())
}

async fn queue_add_and_retrieve_backlog(service: Box<dyn SqlClient>) -> Result<(), CubeError> {
for id in 1..=2 {
service
.exec_query(&format!(
r#"QUEUE ADD "STANDALONE#queue:{}" "payload{}""#,
id, id
))
.await?;
}

{
// Every concurrency slot is free, only the backlog declines the claim
let add_response = service
.exec_query(r#"QUEUE ADD_AND_RETRIEVE "STANDALONE#queue:3" "payload3" 4"#)
.await?;
assert_queue_add_and_retrieve_columns(&add_response);
assert_eq!(
add_response.get_rows(),
&vec![queue_add_and_retrieve_row("3", true, 3, None, None)]
);
}

{
let add_response = service
.exec_query(r#"QUEUE ADD_AND_RETRIEVE "STANDALONE#queue:4" "payload4" 7"#)
.await?;
assert_eq!(
add_response.get_rows(),
&vec![queue_add_and_retrieve_row(
"4",
true,
3,
Some("4"),
Some("payload4")
)]
);
}

{
let pending_response = service
.exec_query(r#"QUEUE PENDING "STANDALONE#queue""#)
.await?;
assert_eq!(pending_response.get_rows().len(), 3);

let active_response = service
.exec_query(r#"QUEUE ACTIVE "STANDALONE#queue""#)
.await?;
assert_eq!(
active_response.get_rows(),
&vec![Row::new(vec![
TableValue::String("4".to_string()),
TableValue::String("4".to_string()),
TableValue::String("active".to_string()),
TableValue::Null,
]),]
Comment on lines +12197 to +12202

@claude claude Bot Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit, test strength: the numeric path suffixes make key == id for every item, so this assertion can't tell the key and id columns apart. QUEUE ACTIVE returns (key, id, status, extra) — swap those two columns in the SQL layer and this test still passes, and so does queue_add_and_retrieve_row("4", true, 3, Some("4"), ...) above (id and active are also both "4").

The pre-existing queue_retrieve_extended test avoids this by using a non-numeric key (tests.rs:11544-11546: key "queue_key_3", id "3"). Naming the paths STANDALONE#queue:key1 … :key4 here costs nothing and makes the columns distinguishable — worth it since ADD_AND_RETRIEVE is the command that newly reports active alongside id in one row.

Also minor: assert_queue_add_and_retrieve_columns is called on the first response but not on the second (tests.rs:12176). Since the second is the claimed case — the one whose column set the driver will actually read — it's the more useful of the two to pin.

);
}

Ok(())
}

async fn queue_ack_then_result_v1(service: Box<dyn SqlClient>) -> Result<(), CubeError> {
let add_response = service
.exec_query(r#"QUEUE ADD PRIORITY 1 "STANDALONE#queue:5555" "payload1";"#)
Expand Down
Loading
Loading