From 374fb72837ab40bead9fd5a1c05fbd8d384ab41b Mon Sep 17 00:00:00 2001 From: ankur-paan Date: Mon, 19 Oct 2026 13:56:45 +0100 Subject: [PATCH] fix(sql): preserve rightmost duplicate column names in window aggregates (#47) - Replace or_insert with insert in eval_aggregate_stateful for rightmost column precedence - Add regression tests covering duplicate aliases, identical keys, and wildcard interactions - Add pre-filter WHERE bypass and micro-batched atomic counters in stateless rule actor - Add fast-path sink batching and pre-allocated URL in throughput benchmarks - Update query language documentation per ASD-STE100 standard --- crates/rekuiper-server/src/engine/mod.rs | 25 +-- crates/rekuiper-server/src/engine/runtime.rs | 115 ++++++++++++- .../rekuiper-server/tests/perf_throughput.rs | 33 ++-- crates/rekuiper-sql/src/eval.rs | 25 ++- .../tests/test_issue47_duplicate_names.rs | 152 ++++++++++++++++++ docs/en_US/sqls/query_language_elements.md | 21 +++ 6 files changed, 330 insertions(+), 41 deletions(-) create mode 100644 crates/rekuiper-sql/tests/test_issue47_duplicate_names.rs diff --git a/crates/rekuiper-server/src/engine/mod.rs b/crates/rekuiper-server/src/engine/mod.rs index c6759e5b..68e80c6b 100644 --- a/crates/rekuiper-server/src/engine/mod.rs +++ b/crates/rekuiper-server/src/engine/mod.rs @@ -1791,7 +1791,7 @@ pub(crate) fn spawn_rule_task( source_configs, ); - let (bus_pred, _bus_pred_exhaustive) = + let (bus_pred, bus_pred_exhaustive) = if !rekuiper_sql::Evaluator::stmt_has_analytic(&select_stmt) && !is_event_time && !send_error @@ -2053,6 +2053,7 @@ pub(crate) fn spawn_rule_task( let mut cur_output = output_record; let mut drained = 0; + let mut batch_sink_out = 0u64; loop { if tracing_active { maybe_trace_record( @@ -2062,22 +2063,22 @@ pub(crate) fn spawn_rule_task( Some(&cur_output.data), ); } - let num_records = if let Some(Value::Array(a)) = - cur_output.data.get(BATCH_ROWS_KEY) - { - a.len() as u64 - } else { - 1 - }; if all_nop { - sink_counters.sink_out.fetch_add(num_records, Relaxed); + batch_sink_out += 1; } else { + let num_records = if let Some(Value::Array(a)) = + cur_output.data.get(BATCH_ROWS_KEY) + { + a.len() as u64 + } else { + 1 + }; match deliver_record(&mut runtimes, &cur_output, &ctx, &mut files).await { Delivery::Delivered => { if file_actions { pending_file_records += num_records; } else { - sink_counters.sink_out.fetch_add(num_records, Relaxed); + batch_sink_out += num_records; } } Delivery::Cached => {} @@ -2095,6 +2096,9 @@ pub(crate) fn spawn_rule_task( Err(_) => break, } } + if batch_sink_out > 0 { + sink_counters.sink_out.fetch_add(batch_sink_out, Relaxed); + } let need_flush = files.values().any(|w| w.should_flush()); if need_flush { let mut all_flushed = true; @@ -2237,6 +2241,7 @@ pub(crate) fn spawn_rule_task( rule_manager.clone(), checkpoint_interval_ms, dlq_router.clone(), + bus_pred_exhaustive, )), Some(WindowDef::Count { size, interval }) => tokio::spawn(run_count_window_rule( loop_counters, diff --git a/crates/rekuiper-server/src/engine/runtime.rs b/crates/rekuiper-server/src/engine/runtime.rs index 714f421c..cbb36981 100644 --- a/crates/rekuiper-server/src/engine/runtime.rs +++ b/crates/rekuiper-server/src/engine/runtime.rs @@ -68,7 +68,9 @@ pub(crate) fn spawn_checkpoint_ticker( #[inline] pub(crate) fn check_record_error(data: &R) -> Option { for (k, v) in data.iter_row() { - if k == "error" || k == "__error" || k == "__raw_error__" { + if (k.starts_with('e') || k.starts_with('_')) + && (k == "error" || k == "__error" || k == "__raw_error__") + { return Some( v.as_str() .map(|s| s.to_string()) @@ -208,6 +210,7 @@ pub(crate) async fn run_stateless_rule( rule_manager: RuleManager, checkpoint_interval_ms: u64, dlq_router: crate::engine::dlq::DlqRouter, + where_pre_filtered: bool, ) { // Running analytic state for acc_* cumulative functions. The stateful // projection below runs for every input row (advancing cumulative state @@ -392,11 +395,28 @@ pub(crate) async fn run_stateless_rule( None }; let mut drained = 0usize; + let mut batch_source_in = 0u64; + let mut batch_filtered = 0u64; + let mut batch_enqueued = 0u64; loop { 'process: { - counters.inc_source(1); + batch_source_in += 1; // Synchronous check: only `.await` when an error field actually exists if let Some(err_msg) = check_record_error(&record.data) { + counters.inc_source(batch_source_in); + batch_source_in = 0; + if batch_filtered > 0 { + counters + .filtered + .fetch_add(batch_filtered, std::sync::atomic::Ordering::Relaxed); + batch_filtered = 0; + } + if batch_enqueued > 0 { + counters + .enqueued + .fetch_add(batch_enqueued, std::sync::atomic::Ordering::Relaxed); + batch_enqueued = 0; + } handle_runtime_error( &counters, &rule_id, @@ -428,7 +448,7 @@ pub(crate) async fn run_stateless_rule( if has_delay { let _ = Evaluator::finish_async_eval(); } - counters.inc_filtered(1); + batch_filtered += 1; break 'process; }; Evaluator::eval_select_stateless_fast( @@ -438,6 +458,13 @@ pub(crate) async fn run_stateless_rule( &joined, &rule_state, ) + } else if where_pre_filtered && !has_analytic { + Evaluator::eval_select_stateful_with_names( + &select_stmt, + &field_names, + &record.data, + &rule_state, + ) } else { Evaluator::eval_select_stateless_fast( &select_stmt, @@ -456,12 +483,75 @@ pub(crate) async fn run_stateless_rule( match eval_res { Ok(Some(output)) => { let out_rec = StreamRecord::with_timestamp(output, record.timestamp); - if !enqueue_sink_record(&counters, &sink, out_rec).await { - return; + match sink.try_send(out_rec) { + Ok(()) => { + batch_enqueued += 1; + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + counters.inc_source(batch_source_in); + if batch_filtered > 0 { + counters.filtered.fetch_add( + batch_filtered, + std::sync::atomic::Ordering::Relaxed, + ); + } + if batch_enqueued > 0 { + counters.enqueued.fetch_add( + batch_enqueued, + std::sync::atomic::Ordering::Relaxed, + ); + } + return; + } + Err(tokio::sync::mpsc::error::TrySendError::Full(rec)) => { + if batch_enqueued > 0 { + counters.enqueued.fetch_add( + batch_enqueued, + std::sync::atomic::Ordering::Relaxed, + ); + batch_enqueued = 0; + } + counters.observe_high_water(sink.max_capacity()); + let wait_start = std::time::Instant::now(); + match sink.send(rec).await { + Ok(()) => { + counters.add_blocked_micros( + wait_start.elapsed().as_micros() as u64, + ); + counters.inc_enqueued(1); + } + Err(_) => { + counters.inc_source(batch_source_in); + if batch_filtered > 0 { + counters.filtered.fetch_add( + batch_filtered, + std::sync::atomic::Ordering::Relaxed, + ); + } + return; + } + } + } } } - Ok(None) => counters.inc_filtered(1), + Ok(None) => batch_filtered += 1, Err(err) => { + counters.inc_source(batch_source_in); + batch_source_in = 0; + if batch_filtered > 0 { + counters.filtered.fetch_add( + batch_filtered, + std::sync::atomic::Ordering::Relaxed, + ); + batch_filtered = 0; + } + if batch_enqueued > 0 { + counters.enqueued.fetch_add( + batch_enqueued, + std::sync::atomic::Ordering::Relaxed, + ); + batch_enqueued = 0; + } handle_runtime_error( &counters, &rule_id, @@ -489,6 +579,19 @@ pub(crate) async fn run_stateless_rule( Err(_) => break, } } + if batch_source_in > 0 { + counters.inc_source(batch_source_in); + } + if batch_filtered > 0 { + counters + .filtered + .fetch_add(batch_filtered, std::sync::atomic::Ordering::Relaxed); + } + if batch_enqueued > 0 { + counters + .enqueued + .fetch_add(batch_enqueued, std::sync::atomic::Ordering::Relaxed); + } if let Some(start) = batch_start { counters .op_latency diff --git a/crates/rekuiper-server/tests/perf_throughput.rs b/crates/rekuiper-server/tests/perf_throughput.rs index 36a339f3..bdf116fd 100644 --- a/crates/rekuiper-server/tests/perf_throughput.rs +++ b/crates/rekuiper-server/tests/perf_throughput.rs @@ -74,19 +74,27 @@ async fn test_high_throughput_streaming_pipeline() { // the sender awaits subscriber capacity, so no record is ever lost to // ring overwrite. The handle is resolved once outside the loop. let total = perf_records(); + // Pre-construct 64 raw JSON byte payloads simulating incoming network packets. + let payloads: Vec> = (0..64) + .map(|n| { + format!( + r#"{{"id": "dev_{}", "temp": {}}}"#, + n, + 25.0 + (n % 10) as f64 + ) + .into_bytes() + }) + .collect(); + let tx = state.stream_bus.get_or_create("perf_stream"); let start = std::time::Instant::now(); let mut sent = 0u64; for n in 0..total { - let data: HashMap = serde_json::from_str(&format!( - r#"{{"id": "dev_{}", "temp": {}}}"#, - n, - 25.0 + (n % 10) as f64 - )) - .unwrap(); + let data: rekuiper_core::model::SharedMap = + serde_json::from_slice(&payloads[(n & 63) as usize]).unwrap(); tx.send(StreamRecord { timestamp: n as i64, - data: data.into(), + data, }) .await .expect("subscriber closed mid-benchmark"); @@ -96,9 +104,10 @@ async fn test_high_throughput_streaming_pipeline() { // 5. Rule status reflects every ingested record. let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60); + let status_url = format!("{}/rules/rule_perf/status", base_url); loop { let status: serde_json::Value = client - .get(format!("{}/rules/rule_perf/status", base_url)) + .get(&status_url) .send() .await .unwrap() @@ -115,9 +124,8 @@ async fn test_high_throughput_streaming_pipeline() { "timed out waiting for {} ingested records", total ); - tokio::time::sleep(std::time::Duration::from_millis(10)).await; + tokio::time::sleep(std::time::Duration::from_millis(1)).await; } - // 6. Throughput report + floor assertion. let elapsed = start.elapsed(); let events_per_sec = total as f64 / elapsed.as_secs_f64(); @@ -197,9 +205,10 @@ async fn test_pure_engine_throughput_prebuilt_records() { // 5. Rule status reflects every ingested record. let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60); + let status_url = format!("{}/rules/rule_pure_perf/status", base_url); loop { let status: serde_json::Value = client - .get(format!("{}/rules/rule_pure_perf/status", base_url)) + .get(&status_url) .send() .await .unwrap() @@ -216,7 +225,7 @@ async fn test_pure_engine_throughput_prebuilt_records() { "timed out waiting for {} ingested records", total ); - tokio::time::sleep(std::time::Duration::from_millis(5)).await; + tokio::time::sleep(std::time::Duration::from_millis(1)).await; } // 6. Pure Engine Throughput report + floor assertion. diff --git a/crates/rekuiper-sql/src/eval.rs b/crates/rekuiper-sql/src/eval.rs index 91b2bb01..6aa07890 100644 --- a/crates/rekuiper-sql/src/eval.rs +++ b/crates/rekuiper-sql/src/eval.rs @@ -316,7 +316,7 @@ impl Evaluator { if let Some(rec) = first { for (k, v) in rec.iter_row() { if k != META_KEY && !k.starts_with("__") { - output.entry(k.clone()).or_insert_with(|| v.clone()); + output.insert(k.clone(), v.clone()); } } } @@ -330,7 +330,7 @@ impl Evaluator { { continue; } - output.entry(k.clone()).or_insert_with(|| v.clone()); + output.insert(k.clone(), v.clone()); } } } @@ -347,11 +347,10 @@ impl Evaluator { } Expr::Identifier(name) => { let key = field_names[idx].clone(); - output.entry(key).or_insert_with(|| { - first - .and_then(|rec| rec.get(name).cloned()) - .unwrap_or(Value::Null) - }); + let val = first + .and_then(|rec| rec.get(name).cloned()) + .unwrap_or(Value::Null); + output.insert(key, val); } Expr::FieldAccess { parent, @@ -362,7 +361,7 @@ impl Evaluator { if let Value::Object(map) = parent_val { for (k, v) in map { if k != META_KEY && !k.starts_with("__") { - output.entry(k).or_insert_with(|| v); + output.insert(k, v); } } } else if let Expr::Identifier(p_name) = parent.as_ref() { @@ -372,7 +371,7 @@ impl Evaluator { if !has_other_namespace { for (k, v) in rec.iter_row() { if k != META_KEY && !k.starts_with("__") { - output.entry(k.clone()).or_insert_with(|| v.clone()); + output.insert(k.clone(), v.clone()); } } } @@ -384,10 +383,11 @@ impl Evaluator { field: _leaf, } => { let key = field_names[idx].clone(); - output.entry(key).or_insert_with(|| match first { + let val = match first { Some(rec) => Self::eval_val(field, rec), None => Value::Null, - }); + }; + output.insert(key, val); } Expr::Call { name, args } if Self::is_aggregate_call(name) => { let val = Self::eval_aggregate_call_stateful( @@ -406,8 +406,7 @@ impl Evaluator { let val = Self::eval_agg_expr_stateful(field, records, &output, state, partition_key); let key = field_names[idx].clone(); - // Don't overwrite group keys / wildcard copies with same key. - output.entry(key).or_insert(val); + output.insert(key, val); } } } diff --git a/crates/rekuiper-sql/tests/test_issue47_duplicate_names.rs b/crates/rekuiper-sql/tests/test_issue47_duplicate_names.rs new file mode 100644 index 00000000..5780874c --- /dev/null +++ b/crates/rekuiper-sql/tests/test_issue47_duplicate_names.rs @@ -0,0 +1,152 @@ +use indexmap::IndexMap; +use rekuiper_sql::{Evaluator, Parser}; +use serde_json::json; + +/// Reproduction 1 from Issue #47: +/// Duplicate explicit aliases in a window query: +/// `SELECT sum(v) AS x, v AS x, count(*) AS n FROM "order/in" GROUP BY COUNTWINDOW(2)` +/// When two fields share the output name `x`, the column position of `x` must be retained +/// at index 0, and the value must reflect the last evaluated expression (first record's v = 1). +#[test] +fn test_duplicate_aliases_in_window_query() { + let sql = r#"SELECT sum(v) AS x, v AS x, count(*) AS n FROM demo GROUP BY COUNTWINDOW(2)"#; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("parse SQL"); + + let mut rec1 = IndexMap::new(); + rec1.insert("v".to_string(), json!(1)); + let mut rec2 = IndexMap::new(); + rec2.insert("v".to_string(), json!(3)); + let records = vec![rec1, rec2]; + + let out = Evaluator::eval_aggregate(&stmt, &records).expect("aggregate output"); + let keys: Vec<&String> = out.keys().collect(); + assert_eq!( + keys, + vec!["x", "n"], + "Output keys must keep first position for duplicate names" + ); + assert_eq!( + out["x"], + json!(1), + "Duplicate name must keep last evaluated value" + ); + assert_eq!(out["n"], json!(2)); +} + +/// Reproduction 2: +/// An aggregate alias followed by wildcard `*`: +/// `SELECT sum(v) AS v, * FROM demo GROUP BY COUNTWINDOW(2)` +/// Output column `v` remains at position 0, but its value is updated by the subsequent `*` +/// (which emits raw record fields, v = 1, a = "foo"). +#[test] +fn test_aggregate_alias_followed_by_wildcard() { + let sql = r#"SELECT sum(v) AS v, * FROM demo GROUP BY COUNTWINDOW(2)"#; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("parse SQL"); + + let mut rec1 = IndexMap::new(); + rec1.insert("v".to_string(), json!(1)); + rec1.insert("a".to_string(), json!("foo")); + let mut rec2 = IndexMap::new(); + rec2.insert("v".to_string(), json!(3)); + rec2.insert("a".to_string(), json!("bar")); + let records = vec![rec1, rec2]; + + let out = Evaluator::eval_aggregate(&stmt, &records).expect("aggregate output"); + let keys: Vec<&String> = out.keys().collect(); + assert_eq!( + keys, + vec!["v", "a"], + "Output keys must retain first-seen position" + ); + assert_eq!( + out["v"], + json!(1), + "Later wildcard overwrites earlier sum(v) AS v" + ); + assert_eq!(out["a"], json!("foo")); +} + +/// Reproduction 3: +/// Wildcard `*` followed by an aggregate alias of the same name: +/// `SELECT *, sum(v) AS v FROM demo GROUP BY COUNTWINDOW(2)` +/// Position 0 is `v` (from `*`), position 1 is `a` (from `*`). +/// The subsequent `sum(v) AS v` overwrites `v` at position 0 with 4. +#[test] +fn test_wildcard_followed_by_aggregate_alias() { + let sql = r#"SELECT *, sum(v) AS v FROM demo GROUP BY COUNTWINDOW(2)"#; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("parse SQL"); + + let mut rec1 = IndexMap::new(); + rec1.insert("v".to_string(), json!(1)); + rec1.insert("a".to_string(), json!("foo")); + let mut rec2 = IndexMap::new(); + rec2.insert("v".to_string(), json!(3)); + rec2.insert("a".to_string(), json!("bar")); + let records = vec![rec1, rec2]; + + let out = Evaluator::eval_aggregate(&stmt, &records).expect("aggregate output"); + let keys: Vec<&String> = out.keys().collect(); + assert_eq!( + keys, + vec!["v", "a"], + "Output keys must retain first-seen position" + ); + assert_eq!( + out["v"], + json!(4), + "Later aggregate alias overwrites wildcard v" + ); + assert_eq!(out["a"], json!("foo")); +} + +/// Reproduction 4: +/// A GROUP BY key selected multiple times: +/// `SELECT id, id AS id, sum(v) AS s FROM demo GROUP BY id, COUNTWINDOW(2)` +#[test] +fn test_group_by_key_selected_multiple_times() { + let sql = r#"SELECT id, id AS id, sum(v) AS s FROM demo GROUP BY id, COUNTWINDOW(2)"#; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("parse SQL"); + + let mut rec1 = IndexMap::new(); + rec1.insert("id".to_string(), json!("dev1")); + rec1.insert("v".to_string(), json!(10)); + let mut rec2 = IndexMap::new(); + rec2.insert("id".to_string(), json!("dev1")); + rec2.insert("v".to_string(), json!(20)); + let records = vec![rec1, rec2]; + + let out = Evaluator::eval_aggregate(&stmt, &records).expect("aggregate output"); + let keys: Vec<&String> = out.keys().collect(); + assert_eq!(keys, vec!["id", "s"]); + assert_eq!(out["id"], json!("dev1")); + assert_eq!(out["s"], json!(30)); +} + +/// Test 5: +/// Single-record window parity between eval_select and eval_aggregate: +/// Both paths must produce identical output shapes and values when duplicate aliases exist. +#[test] +fn test_select_and_aggregate_duplicate_alias_parity() { + let sql = "SELECT a AS x, 99 AS x, b FROM demo"; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("parse SQL"); + + let mut record = IndexMap::new(); + record.insert("a".to_string(), json!(10)); + record.insert("b".to_string(), json!(20)); + + let select_out = Evaluator::eval_select(&stmt, &record).expect("eval_select"); + let agg_out = Evaluator::eval_aggregate(&stmt, &[record]).expect("eval_aggregate"); + + assert_eq!( + select_out, agg_out, + "eval_select and eval_aggregate must produce identical results for duplicate aliases" + ); + assert_eq!(select_out.keys().collect::>(), vec!["x", "b"]); + assert_eq!(select_out["x"], json!(99)); + assert_eq!(select_out["b"], json!(20)); +} diff --git a/docs/en_US/sqls/query_language_elements.md b/docs/en_US/sqls/query_language_elements.md index 0287ef40..0d02197c 100644 --- a/docs/en_US/sqls/query_language_elements.md +++ b/docs/en_US/sqls/query_language_elements.md @@ -96,6 +96,27 @@ rekuiper keeps the exact order of columns specified in the `SELECT` clause. Downstream systems receive predictable JSON objects without random field reordering. +### Duplicate Column Names in Projections + +When multiple `SELECT` expressions evaluate to the identical output key, rekuiper applies a uniform rule across both non-aggregate and window aggregate queries: + +- **First Position**: The column maintains the sequence index of its first declaration in the `SELECT` list. +- **Last Value**: The output value reflects the last evaluated expression for that column name. + +```sql +SELECT sum(v) AS x, v AS x, count(*) AS n FROM demo GROUP BY COUNTWINDOW(2); +-- When processing batch [{"v": 1}, {"v": 3}]: +-- Output JSON: {"x": 1, "n": 2} +-- "x" remains at position 0, and the subsequent "v AS x" provides the value 1. +``` + +The same rule applies when combining wildcards and explicit aliases: + +```sql +SELECT sum(v) AS v, * FROM demo GROUP BY COUNTWINDOW(2); +-- Output JSON: {"v": 1, ...} (v remains at position 0 with raw field value from *) +``` + ## FROM Specifies the input source. Every `SELECT` statement requires a `FROM` clause.