Skip to content

fix(sql): preserve rightmost duplicate column names in window aggregates (#47) #48

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

Merged
ankur-paan merged 1 commit into from
Oct 19, 2026
Merged

fix(sql): preserve rightmost duplicate column names in window aggregates (#47) #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
25 changes: 15 additions & 10 deletions crates/rekuiper-server/src/engine/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand All @@ -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 => {}
Expand All @@ -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;
Expand Down Expand Up @@ -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,
Expand Down
115 changes: 109 additions & 6 deletions crates/rekuiper-server/src/engine/runtime.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,9 @@ pub(crate) fn spawn_checkpoint_ticker(
#[inline]
pub(crate) fn check_record_error<R: RowLookup + ?Sized>(data: &R) -> Option<String> {
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())
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down
33 changes: 21 additions & 12 deletions crates/rekuiper-server/tests/perf_throughput.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Vec<u8>> = (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<String, Value> = 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");
Expand All @@ -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()
Expand All @@ -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();
Expand Down Expand Up @@ -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()
Expand All @@ -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.
Expand Down
25 changes: 12 additions & 13 deletions crates/rekuiper-sql/src/eval.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
}
}
Expand All @@ -330,7 +330,7 @@ impl Evaluator {
{
continue;
}
output.entry(k.clone()).or_insert_with(|| v.clone());
output.insert(k.clone(), v.clone());
}
}
}
Expand All @@ -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,
Expand All @@ -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() {
Expand All @@ -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());
}
}
}
Expand All @@ -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(
Expand All @@ -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);
}
}
}
Expand Down
Loading
Loading