From 784efece688925efe0339897f88c73b9a311e5f3 Mon Sep 17 00:00:00 2001 From: ankur-paan Date: Mon, 19 Oct 2026 07:09:57 +0100 Subject: [PATCH 1/2] fix(sql): preserve SELECT column order and deterministic JSON field output (#45) Ensure deterministic field order in SQL projections and JSON outputs. Replace random HashMap iteration in SharedMap and SQL evaluation with IndexMap and the RowLookup trait. Maintain pure engine throughput at 2.55M EPS and streaming pipeline throughput at 1.07M EPS. Update documentation, SVG diagrams, and MCP tool references according to ASD-STE100. --- Cargo.lock | 7 + Cargo.toml | 3 +- crates/rekuiper-connectors/Cargo.toml | 1 + crates/rekuiper-connectors/src/file.rs | 13 +- crates/rekuiper-connectors/src/http.rs | 9 +- crates/rekuiper-connectors/src/kafka.rs | 9 +- crates/rekuiper-connectors/src/mqtt.rs | 37 ++- crates/rekuiper-connectors/src/redis.rs | 9 +- crates/rekuiper-connectors/src/sql.rs | 9 +- crates/rekuiper-connectors/src/websocket.rs | 9 +- crates/rekuiper-connectors/src/zmq.rs | 5 +- crates/rekuiper-core/Cargo.toml | 1 + crates/rekuiper-core/src/model.rs | 172 +++++++++-- .../rekuiper-core/tests/test_udf_plugins.rs | 2 +- crates/rekuiper-mcp/README.md | 4 +- crates/rekuiper-server/Cargo.toml | 1 + crates/rekuiper-server/src/engine/joins.rs | 70 +++-- crates/rekuiper-server/src/engine/mod.rs | 6 +- crates/rekuiper-server/src/engine/runtime.rs | 23 +- crates/rekuiper-server/src/engine/sinks.rs | 95 +++--- crates/rekuiper-server/src/engine/windows.rs | 12 +- .../rekuiper-server/src/handlers/ruletest.rs | 15 +- .../rekuiper-server/src/handlers/streams.rs | 9 +- crates/rekuiper-server/src/state.rs | 6 +- crates/rekuiper-server/tests/fvt_compat.rs | 2 +- crates/rekuiper-server/tests/iiot_mqtt.rs | 2 +- .../tests/test_issue22_output_shape.rs | 6 +- .../tests/test_issue24_windows.rs | 123 ++++++-- .../tests/test_issue25_joins_server.rs | 6 +- .../test_issue34_window_having_had_changed.rs | 29 +- crates/rekuiper-sql/Cargo.toml | 1 + crates/rekuiper-sql/src/eval.rs | 275 +++++++++++++----- crates/rekuiper-sql/src/eval/aggregate.rs | 90 +++--- crates/rekuiper-sql/src/eval/window.rs | 11 +- crates/rekuiper-sql/tests/sql_test.rs | 4 +- .../tests/test_issue17_analytic_filter.rs | 10 +- .../rekuiper-sql/tests/test_issue22_parity.rs | 2 +- .../rekuiper-sql/tests/test_issue27_parity.rs | 7 +- .../rekuiper-sql/tests/test_issue45_order.rs | 68 +++++ .../rekuiper-sql/tests/test_sql_functions.rs | 14 +- crates/rekuiper-sql/tests/test_sql_parity.rs | 4 +- .../tests/validate_catalog_functions.rs | 2 +- crates/rekuiper-sql/tests/window_eval_test.rs | 4 +- docs/en_US/mcp/overview.md | 2 +- .../diagrams/deterministic_field_order.svg | 54 ++++ docs/en_US/sqls/query_language_elements.md | 23 ++ 46 files changed, 907 insertions(+), 359 deletions(-) create mode 100644 crates/rekuiper-sql/tests/test_issue45_order.rs create mode 100644 docs/en_US/public/diagrams/deterministic_field_order.svg diff --git a/Cargo.lock b/Cargo.lock index 5d82999e..1b0479c5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2164,6 +2164,8 @@ checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" dependencies = [ "equivalent", "hashbrown 0.17.1", + "serde", + "serde_core", ] [[package]] @@ -3396,6 +3398,7 @@ dependencies = [ "bytes", "chrono", "futures", + "indexmap", "lapin", "parking_lot", "parquet", @@ -3425,6 +3428,7 @@ dependencies = [ "anyhow", "async-trait", "chrono", + "indexmap", "parking_lot", "regex", "rekuiper-conf", @@ -3466,6 +3470,7 @@ dependencies = [ "boa_engine", "chrono", "futures", + "indexmap", "parking_lot", "rekuiper-conf", "rekuiper-connectors", @@ -3497,6 +3502,7 @@ dependencies = [ "chrono-tz", "crc32fast", "flate2", + "indexmap", "md-5", "parking_lot", "rand 0.8.8", @@ -3933,6 +3939,7 @@ version = "1.0.151" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" dependencies = [ + "indexmap", "itoa", "memchr", "serde", diff --git a/Cargo.toml b/Cargo.toml index afb588f2..649e74f5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,7 +28,8 @@ categories = ["embedded", "network-programming", "science"] tokio = { version = "1.38", features = ["full"] } axum = { version = "0.7", features = ["macros"] } serde = { version = "1.0", features = ["derive"] } -serde_json = "1.0" +serde_json = { version = "1.0", features = ["preserve_order"] } +indexmap = { version = "2.6", features = ["serde"] } serde_yaml = { package = "serde_norway", version = "0.9" } serde_norway = "0.9" toml = "0.8" diff --git a/crates/rekuiper-connectors/Cargo.toml b/crates/rekuiper-connectors/Cargo.toml index c310ad39..63caab57 100644 --- a/crates/rekuiper-connectors/Cargo.toml +++ b/crates/rekuiper-connectors/Cargo.toml @@ -14,6 +14,7 @@ categories.workspace = true tokio = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +indexmap = { workspace = true } anyhow = { workspace = true } tracing = { workspace = true } async-trait = { workspace = true } diff --git a/crates/rekuiper-connectors/src/file.rs b/crates/rekuiper-connectors/src/file.rs index f3aaa5f6..270ebf37 100644 --- a/crates/rekuiper-connectors/src/file.rs +++ b/crates/rekuiper-connectors/src/file.rs @@ -1,6 +1,7 @@ use crate::{parquet_io, Sink, META_KEY}; use anyhow::{bail, Context, Result}; use async_trait::async_trait; +use indexmap::IndexMap; use rekuiper_core::model::StreamRecord; use rekuiper_core::StreamSender; use serde::{Deserialize, Serialize}; @@ -478,12 +479,16 @@ impl FileSource { } // JSON / lines: objects emit directly, arrays fan out per element. match serde_json::from_str::(line) { - Ok(Value::Object(map)) => Some(vec![StreamRecord::new(map.into_iter().collect())]), + Ok(Value::Object(map)) => Some(vec![StreamRecord::new( + map.into_iter().collect::>(), + )]), Ok(Value::Array(items)) => { let mut records = Vec::new(); for item in items { if let Value::Object(map) = item { - records.push(StreamRecord::new(map.into_iter().collect())); + records.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } else { tracing::warn!("Skipping non-object element in file source line"); } @@ -509,8 +514,8 @@ pub fn parse_delimited_line( line: &str, delimiter: char, headers: &[String], -) -> HashMap { - let mut map = HashMap::with_capacity(headers.len()); +) -> IndexMap { + let mut map = IndexMap::with_capacity(headers.len()); for (i, header) in headers.iter().enumerate() { let token = line.split(delimiter).nth(i).unwrap_or("").trim(); map.insert(header.clone(), parse_delimited_token(token)); diff --git a/crates/rekuiper-connectors/src/http.rs b/crates/rekuiper-connectors/src/http.rs index e93deaf8..e7f4ce5c 100644 --- a/crates/rekuiper-connectors/src/http.rs +++ b/crates/rekuiper-connectors/src/http.rs @@ -1,6 +1,7 @@ use crate::{Sink, META_KEY}; use anyhow::{bail, Context, Result}; use async_trait::async_trait; +use indexmap::IndexMap; use rekuiper_core::model::StreamRecord; use rekuiper_core::StreamSender; use serde::{Deserialize, Serialize}; @@ -149,12 +150,16 @@ impl HttpPullSource { .await .with_context(|| format!("HTTP pull response from {} is not JSON", self.config.url))?; match payload { - Value::Object(map) => Ok(vec![StreamRecord::new(map.into_iter().collect())]), + Value::Object(map) => Ok(vec![StreamRecord::new( + map.into_iter().collect::>(), + )]), Value::Array(items) => { let mut records = Vec::new(); for item in items { if let Value::Object(map) = item { - records.push(StreamRecord::new(map.into_iter().collect())); + records.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } else { tracing::warn!( "Skipping non-object element in HTTP pull response from {}", diff --git a/crates/rekuiper-connectors/src/kafka.rs b/crates/rekuiper-connectors/src/kafka.rs index cdbc0609..5e8d1c64 100644 --- a/crates/rekuiper-connectors/src/kafka.rs +++ b/crates/rekuiper-connectors/src/kafka.rs @@ -1,5 +1,6 @@ use crate::META_KEY; use anyhow::{Context, Result}; +use indexmap::IndexMap; use parking_lot::RwLock; use rekuiper_core::model::StreamRecord; use rekuiper_core::StreamSender; @@ -300,12 +301,16 @@ fn decode_kafka_payload(payload: &[u8]) -> Result> { let value: Value = serde_json::from_slice(payload).context("Kafka payload is not valid JSON")?; match value { - Value::Object(map) => Ok(vec![StreamRecord::new(map.into_iter().collect())]), + Value::Object(map) => Ok(vec![StreamRecord::new( + map.into_iter().collect::>(), + )]), Value::Array(items) => { let mut records = Vec::new(); for item in items { if let Value::Object(map) = item { - records.push(StreamRecord::new(map.into_iter().collect())); + records.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } else { tracing::warn!("Skipping non-object element in Kafka message"); } diff --git a/crates/rekuiper-connectors/src/mqtt.rs b/crates/rekuiper-connectors/src/mqtt.rs index 73588459..9affcbff 100644 --- a/crates/rekuiper-connectors/src/mqtt.rs +++ b/crates/rekuiper-connectors/src/mqtt.rs @@ -5,6 +5,7 @@ use crate::{ }; use anyhow::{bail, Context, Result}; use async_trait::async_trait; +use indexmap::IndexMap; use rekuiper_core::model::StreamRecord; use rekuiper_core::{RuleCounters, StreamSender}; use rumqttc::{AsyncClient, Event, MqttOptions, Packet, QoS}; @@ -667,13 +668,15 @@ pub fn decode_mqtt_payload(payload: &[u8]) -> Result { // Fast path: deserialize straight into the row map (no intermediate DOM, // no object clone). Only a failing payload pays for the diagnostic parse // that distinguishes invalid JSON from a non-object value. - match serde_json::from_slice::>(payload) { + match serde_json::from_slice::>(payload) { Ok(data) => Ok(StreamRecord::new(data)), Err(_) => { let value: Value = serde_json::from_slice(payload).context("MQTT payload is not valid JSON")?; match value { - Value::Object(map) => Ok(StreamRecord::new(map.into_iter().collect())), + Value::Object(map) => Ok(StreamRecord::new( + map.into_iter().collect::>(), + )), _ => Err(anyhow::anyhow!("MQTT payload must be a JSON object")), } } @@ -709,15 +712,15 @@ pub fn decode_payload_into_with_mask( let value: Value = serde_json::from_slice(payload) .context("MQTT payload is not valid JSON")?; match value { - Value::Object(map) => { - out.push(StreamRecord::new(map.into_iter().collect())) - } + Value::Object(map) => out.push(StreamRecord::new( + map.into_iter().collect::>(), + )), Value::Array(items) => { for item in items { match item { - Value::Object(map) => { - out.push(StreamRecord::new(map.into_iter().collect())) - } + Value::Object(map) => out.push(StreamRecord::new( + map.into_iter().collect::>(), + )), _ => { out.truncate(first); bail!( @@ -732,21 +735,21 @@ pub fn decode_payload_into_with_mask( } } } else { - match serde_json::from_slice::>(payload) { + match serde_json::from_slice::>(payload) { Ok(data) => out.push(StreamRecord::new(data)), Err(_) => { let value: Value = serde_json::from_slice(payload) .context("MQTT payload is not valid JSON")?; match value { - Value::Object(map) => { - out.push(StreamRecord::new(map.into_iter().collect())) - } + Value::Object(map) => out.push(StreamRecord::new( + map.into_iter().collect::>(), + )), Value::Array(items) => { for item in items { match item { - Value::Object(map) => { - out.push(StreamRecord::new(map.into_iter().collect())) - } + Value::Object(map) => out.push(StreamRecord::new( + map.into_iter().collect::>(), + )), _ => { out.truncate(first); bail!( @@ -780,7 +783,9 @@ pub fn decode_payload_into_with_mask( out.push(StreamRecord::new(codec.decode_row(line))); } PayloadFormat::Protobuf(message) => match ProtobufCodec::decode(payload, message) { - Value::Object(map) => out.push(StreamRecord::new(map.into_iter().collect())), + Value::Object(map) => out.push(StreamRecord::new( + map.into_iter().collect::>(), + )), _ => bail!("payload is not a valid '{}' protobuf message", message.name), }, PayloadFormat::EdgeX => EdgeXCodec::decode(payload, meta.clone(), out)?, diff --git a/crates/rekuiper-connectors/src/redis.rs b/crates/rekuiper-connectors/src/redis.rs index 63052054..d0d7a0ff 100644 --- a/crates/rekuiper-connectors/src/redis.rs +++ b/crates/rekuiper-connectors/src/redis.rs @@ -1,5 +1,6 @@ use crate::META_KEY; use anyhow::{Context, Result}; +use indexmap::IndexMap; use parking_lot::RwLock; use rekuiper_core::model::StreamRecord; use rekuiper_core::StreamSender; @@ -251,12 +252,16 @@ fn decode_redis_payload(payload: &[u8]) -> Result> { let value: Value = serde_json::from_slice(payload).context("Redis payload is not valid JSON")?; match value { - Value::Object(map) => Ok(vec![StreamRecord::new(map.into_iter().collect())]), + Value::Object(map) => Ok(vec![StreamRecord::new( + map.into_iter().collect::>(), + )]), Value::Array(items) => { let mut records = Vec::new(); for item in items { if let Value::Object(map) = item { - records.push(StreamRecord::new(map.into_iter().collect())); + records.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } else { tracing::warn!("Skipping non-object element in Redis message"); } diff --git a/crates/rekuiper-connectors/src/sql.rs b/crates/rekuiper-connectors/src/sql.rs index cbccfc05..d562b91b 100644 --- a/crates/rekuiper-connectors/src/sql.rs +++ b/crates/rekuiper-connectors/src/sql.rs @@ -1,5 +1,6 @@ use crate::META_KEY; use anyhow::{bail, Result}; +use indexmap::IndexMap; use parking_lot::RwLock; use rekuiper_core::model::StreamRecord; use rekuiper_core::StreamSender; @@ -968,7 +969,9 @@ impl SqlSource { let name = col.name(); map.insert(name.to_string(), sqlite_column_value(r, name)); } - out.push(StreamRecord::new(map.into_iter().collect())); + out.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } advance_index(&mut self.index, &out); return Ok(out); @@ -990,7 +993,9 @@ impl SqlSource { let name = col.name(); map.insert(name.to_string(), pg_column_value(r, name)); } - out.push(StreamRecord::new(map.into_iter().collect())); + out.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } advance_index(&mut self.index, &out); return Ok(out); diff --git a/crates/rekuiper-connectors/src/websocket.rs b/crates/rekuiper-connectors/src/websocket.rs index 39c02699..8e431029 100644 --- a/crates/rekuiper-connectors/src/websocket.rs +++ b/crates/rekuiper-connectors/src/websocket.rs @@ -1,5 +1,6 @@ use crate::META_KEY; use anyhow::{Context, Result}; +use indexmap::IndexMap; use parking_lot::Mutex; use rekuiper_core::model::StreamRecord; use rekuiper_core::StreamSender; @@ -71,12 +72,16 @@ impl WebSocketConfig { pub fn decode_websocket_message(text: &str) -> Result> { let value: Value = serde_json::from_str(text).context("WebSocket payload is not valid JSON")?; match value { - Value::Object(map) => Ok(vec![StreamRecord::new(map.into_iter().collect())]), + Value::Object(map) => Ok(vec![StreamRecord::new( + map.into_iter().collect::>(), + )]), Value::Array(items) => { let mut records = Vec::new(); for item in items { if let Value::Object(map) = item { - records.push(StreamRecord::new(map.into_iter().collect())); + records.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } else { tracing::warn!("Skipping non-object element in WebSocket message"); } diff --git a/crates/rekuiper-connectors/src/zmq.rs b/crates/rekuiper-connectors/src/zmq.rs index a613198b..5f236ec1 100644 --- a/crates/rekuiper-connectors/src/zmq.rs +++ b/crates/rekuiper-connectors/src/zmq.rs @@ -1,5 +1,6 @@ use anyhow::{Context, Result}; use bytes::Bytes; +use indexmap::IndexMap; use rekuiper_core::model::StreamRecord; use rekuiper_core::StreamSender; use serde::{Deserialize, Serialize}; @@ -176,7 +177,7 @@ pub fn decode_zmq_payload(payload: &[u8], topic: Option<&str>) -> Result { - let mut rec = StreamRecord::new(map.into_iter().collect()); + let mut rec = StreamRecord::new(map.into_iter().collect::>()); if let Some(t) = topic { let mut meta = std::collections::HashMap::new(); meta.insert("topic".to_string(), Value::String(t.to_string())); @@ -190,7 +191,7 @@ pub fn decode_zmq_payload(payload: &[u8], topic: Option<&str>) -> Result { for item in items { if let Value::Object(map) = item { - let mut rec = StreamRecord::new(map.into_iter().collect()); + let mut rec = StreamRecord::new(map.into_iter().collect::>()); if let Some(t) = topic { let mut meta = std::collections::HashMap::new(); meta.insert("topic".to_string(), Value::String(t.to_string())); diff --git a/crates/rekuiper-core/Cargo.toml b/crates/rekuiper-core/Cargo.toml index 8072e68f..87b60ff7 100644 --- a/crates/rekuiper-core/Cargo.toml +++ b/crates/rekuiper-core/Cargo.toml @@ -13,6 +13,7 @@ categories.workspace = true [dependencies] serde = { workspace = true } serde_json = { workspace = true } +indexmap = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } anyhow = { workspace = true } diff --git a/crates/rekuiper-core/src/model.rs b/crates/rekuiper-core/src/model.rs index 7e3d4ff1..e60d7248 100644 --- a/crates/rekuiper-core/src/model.rs +++ b/crates/rekuiper-core/src/model.rs @@ -1,29 +1,41 @@ +use indexmap::IndexMap; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use serde_json::Value; use std::collections::HashMap; use std::ops::{Deref, DerefMut}; use std::sync::Arc; -/// Copy-on-write map wrapper backed by `Arc>`. +/// Copy-on-write map wrapper backed by `Arc>`. +/// Preserves the SELECT projection field order or input record field order. /// Cloning a `StreamRecord` across N subscribers in `StreamBus` is an O(1) /// atomic refcount increment; mutation via `DerefMut` triggers `Arc::make_mut`. #[derive(Debug, Clone, Default)] -pub struct SharedMap(pub Arc>); +pub struct SharedMap(pub Arc>); impl SharedMap { #[inline] - pub fn new(map: HashMap) -> Self { + pub fn new(map: IndexMap) -> Self { Self(Arc::new(map)) } #[inline] - pub fn into_inner(self) -> HashMap { + pub fn from_hashmap(map: HashMap) -> Self { + Self(Arc::new(map.into_iter().collect())) + } + + #[inline] + pub fn into_inner(self) -> IndexMap { Arc::try_unwrap(self.0).unwrap_or_else(|arc| (*arc).clone()) } + + #[inline] + pub fn to_hashmap(&self) -> HashMap { + self.0.iter().map(|(k, v)| (k.clone(), v.clone())).collect() + } } impl Deref for SharedMap { - type Target = HashMap; + type Target = IndexMap; #[inline] fn deref(&self) -> &Self::Target { &self.0 @@ -43,13 +55,28 @@ impl PartialEq for SharedMap { } } +impl PartialEq> for SharedMap { + fn eq(&self, other: &IndexMap) -> bool { + *self.0 == *other + } +} + impl PartialEq> for SharedMap { fn eq(&self, other: &HashMap) -> bool { - *self.0 == *other + if self.0.len() != other.len() { + return false; + } + self.0.iter().all(|(k, v)| other.get(k) == Some(v)) } } impl PartialEq for HashMap { + fn eq(&self, other: &SharedMap) -> bool { + other == self + } +} + +impl PartialEq for IndexMap { fn eq(&self, other: &SharedMap) -> bool { *self == *other.0 } @@ -57,7 +84,7 @@ impl PartialEq for HashMap { impl<'a> IntoIterator for &'a SharedMap { type Item = (&'a String, &'a Value); - type IntoIter = std::collections::hash_map::Iter<'a, String, Value>; + type IntoIter = indexmap::map::Iter<'a, String, Value>; #[inline] fn into_iter(self) -> Self::IntoIter { @@ -67,7 +94,7 @@ impl<'a> IntoIterator for &'a SharedMap { impl IntoIterator for SharedMap { type Item = (String, Value); - type IntoIter = std::collections::hash_map::IntoIter; + type IntoIter = indexmap::map::IntoIter; #[inline] fn into_iter(self) -> Self::IntoIter { @@ -82,20 +109,34 @@ impl FromIterator<(String, Value)> for SharedMap { } } +impl From> for SharedMap { + #[inline] + fn from(map: IndexMap) -> Self { + Self::new(map) + } +} + impl From> for SharedMap { #[inline] fn from(map: HashMap) -> Self { - Self::new(map) + Self::from_hashmap(map) } } -impl From for HashMap { +impl From for IndexMap { #[inline] fn from(shared: SharedMap) -> Self { shared.into_inner() } } +impl From for HashMap { + #[inline] + fn from(shared: SharedMap) -> Self { + shared.into_inner().into_iter().collect() + } +} + impl Serialize for SharedMap { fn serialize(&self, serializer: S) -> Result { self.0.serialize(serializer) @@ -104,7 +145,7 @@ impl Serialize for SharedMap { impl<'de> Deserialize<'de> for SharedMap { fn deserialize>(deserializer: D) -> Result { - HashMap::::deserialize(deserializer).map(Self::new) + IndexMap::::deserialize(deserializer).map(Self::new) } } @@ -115,10 +156,17 @@ pub struct StreamRecord { } impl StreamRecord { - pub fn new(data: HashMap) -> Self { + pub fn new(data: impl Into) -> Self { Self { timestamp: chrono::Utc::now().timestamp_millis(), - data: SharedMap::new(data), + data: data.into(), + } + } + + pub fn from_hashmap(data: HashMap) -> Self { + Self { + timestamp: chrono::Utc::now().timestamp_millis(), + data: SharedMap::from_hashmap(data), } } @@ -130,10 +178,10 @@ impl StreamRecord { } #[inline] - pub fn with_timestamp(data: HashMap, timestamp: i64) -> Self { + pub fn with_timestamp(data: impl Into, timestamp: i64) -> Self { Self { timestamp, - data: SharedMap::new(data), + data: data.into(), } } @@ -813,9 +861,89 @@ pub fn parse_default_value(raw: &str, field_type: &str) -> Value { coerce_value_to_type(Value::String(s.to_string()), field_type) } +pub trait SchemaMap { + fn contains_key(&self, key: &str) -> bool; + fn get(&self, key: &str) -> Option<&Value>; + fn remove(&mut self, key: &str) -> Option; + fn insert(&mut self, key: String, val: Value) -> Option; + fn find_key_case_insensitive(&self, name: &str) -> Option; +} + +impl SchemaMap for HashMap { + #[inline] + fn contains_key(&self, key: &str) -> bool { + self.contains_key(key) + } + #[inline] + fn get(&self, key: &str) -> Option<&Value> { + self.get(key) + } + #[inline] + fn remove(&mut self, key: &str) -> Option { + self.remove(key) + } + #[inline] + fn insert(&mut self, key: String, val: Value) -> Option { + self.insert(key, val) + } + #[inline] + fn find_key_case_insensitive(&self, name: &str) -> Option { + self.keys().find(|k| k.eq_ignore_ascii_case(name)).cloned() + } +} + +impl SchemaMap for IndexMap { + #[inline] + fn contains_key(&self, key: &str) -> bool { + self.contains_key(key) + } + #[inline] + fn get(&self, key: &str) -> Option<&Value> { + self.get(key) + } + #[inline] + fn remove(&mut self, key: &str) -> Option { + self.shift_remove(key) + } + #[inline] + fn insert(&mut self, key: String, val: Value) -> Option { + self.insert(key, val) + } + #[inline] + fn find_key_case_insensitive(&self, name: &str) -> Option { + self.keys().find(|k| k.eq_ignore_ascii_case(name)).cloned() + } +} + +impl SchemaMap for SharedMap { + #[inline] + fn contains_key(&self, key: &str) -> bool { + self.0.contains_key(key) + } + #[inline] + fn get(&self, key: &str) -> Option<&Value> { + self.0.get(key) + } + #[inline] + fn remove(&mut self, key: &str) -> Option { + Arc::make_mut(&mut self.0).shift_remove(key) + } + #[inline] + fn insert(&mut self, key: String, val: Value) -> Option { + Arc::make_mut(&mut self.0).insert(key, val) + } + #[inline] + fn find_key_case_insensitive(&self, name: &str) -> Option { + self.0 + .keys() + .find(|k| k.eq_ignore_ascii_case(name)) + .cloned() + } +} + /// Coerce an incoming record map according to declared stream fields. /// When `fields` is empty (schemaless stream), this returns immediately (zero-cost no-op). -pub fn enforce_stream_schema(data: &mut HashMap, fields: &[StreamField]) { +pub fn enforce_stream_schema(data: &mut M, fields: &[StreamField]) { if fields.is_empty() { return; } @@ -824,9 +952,7 @@ pub fn enforce_stream_schema(data: &mut HashMap, fields: &[Stream let existing_key = if data.contains_key(&field.name) { Some(field.name.clone()) } else { - data.keys() - .find(|k| k.eq_ignore_ascii_case(&field.name)) - .cloned() + data.find_key_case_insensitive(&field.name) }; if let Some(key) = existing_key { @@ -1009,8 +1135,8 @@ impl std::error::Error for SchemaMismatchError {} /// Strict schema validator: validates and coerces declared fields, returning an error /// if any non-null present value cannot be coerced into the declared type. -pub fn enforce_stream_schema_checked( - data: &mut HashMap, +pub fn enforce_stream_schema_checked( + data: &mut M, fields: &[StreamField], ) -> Result<(), SchemaMismatchError> { if fields.is_empty() { @@ -1020,9 +1146,7 @@ pub fn enforce_stream_schema_checked( let existing_key = if data.contains_key(&field.name) { Some(field.name.clone()) } else { - data.keys() - .find(|k| k.eq_ignore_ascii_case(&field.name)) - .cloned() + data.find_key_case_insensitive(&field.name) }; if let Some(key) = existing_key { diff --git a/crates/rekuiper-core/tests/test_udf_plugins.rs b/crates/rekuiper-core/tests/test_udf_plugins.rs index d4b316a4..3c2007ac 100644 --- a/crates/rekuiper-core/tests/test_udf_plugins.rs +++ b/crates/rekuiper-core/tests/test_udf_plugins.rs @@ -52,7 +52,7 @@ fn eval_project(sql: &str, record: &HashMap, alias: &str) -> Valu let stmt = parser.parse_select().expect("Should parse"); rekuiper_sql::Evaluator::eval_select(&stmt, record) .expect("Should project") - .remove(alias) + .shift_remove(alias) .unwrap() } diff --git a/crates/rekuiper-mcp/README.md b/crates/rekuiper-mcp/README.md index f57ad68f..450f66cf 100644 --- a/crates/rekuiper-mcp/README.md +++ b/crates/rekuiper-mcp/README.md @@ -10,7 +10,7 @@ An enterprise-grade, native Rust **Model Context Protocol (MCP)** server providi - **Zero-Network Offline SQL Intelligence & Simulation**: - `validate_sql`: Offline AST parsing, static analysis, column resolution, and DDL validation directly through the embedded `rekuiper-sql` parser without network overhead. - - `test_sql_expression`: Instant in-memory execution of streaming SQL transformations against mock event payloads using `Evaluator::eval_select`. + - `test_sql_expression`: Instant in-memory execution of streaming SQL transformations against mock event payloads with deterministic SELECT column order preservation using `Evaluator::eval_select`. - `explain_sql`: Deep AST query deconstruction into sources, projections, joins, where filters, group by keys, and time/count window clauses. - **Full-Fidelity Stream & Table DDL & Event Ingestion**: @@ -76,7 +76,7 @@ An enterprise-grade, native Rust **Model Context Protocol (MCP)** server providi | Tool | Description | | :--- | :--- | | `validate_sql` | Offline AST syntax parser and static validator for SQL, window functions, CEP pattern matchers (`sequence_match`, `sequence_count`), `ASOF JOIN` syntax, and DDL statements (including `DEFAULT` column values and `BUFFER_FULL_POLICY`). | -| `test_sql_expression` | In-memory streaming query evaluator against mock JSON telemetry payloads, supporting vector search (`cosine_similarity`), CEP sequence patterns (`sequence_match`), array/object transforms (`json_map`, `object_erase`), and stateful analytics (`acc_distinct_collect`, `lead`). | +| `test_sql_expression` | In-memory streaming query evaluator against mock JSON telemetry payloads with exact SELECT field order preservation, supporting vector search (`cosine_similarity`), CEP sequence patterns (`sequence_match`), array/object transforms (`json_map`, `object_erase`), and stateful analytics (`acc_distinct_collect`, `lead`). | | `explain_sql` | Deconstructs streaming SQL into projections, sources, joins, and window clauses. | | `explain_pipeline_profile` | Static performance profiler analyzing whether a query qualifies for zero-allocation WHERE fast-paths, Rete-Lite StreamBus indexing, IncrementalWindow, or bounded ASOF JOIN state. | diff --git a/crates/rekuiper-server/Cargo.toml b/crates/rekuiper-server/Cargo.toml index 3c467f2a..1823c19f 100644 --- a/crates/rekuiper-server/Cargo.toml +++ b/crates/rekuiper-server/Cargo.toml @@ -20,6 +20,7 @@ tower = { workspace = true, features = ["util"] } tower-http = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +indexmap = { workspace = true } serde_yaml = { workspace = true } anyhow = { workspace = true } tracing = { workspace = true } diff --git a/crates/rekuiper-server/src/engine/joins.rs b/crates/rekuiper-server/src/engine/joins.rs index 8cbb532b..2f0bfd33 100644 --- a/crates/rekuiper-server/src/engine/joins.rs +++ b/crates/rekuiper-server/src/engine/joins.rs @@ -1,7 +1,9 @@ use crate::engine::{resolve_modbus_table_config, resolve_opcua_table_config, resolve_redis_addr}; +use indexmap::IndexMap; use parking_lot::RwLock; use rekuiper_connectors::SqlConnectorConfig; use rekuiper_core::TableManager; +use rekuiper_sql::eval::RowLookup; use rekuiper_sql::{Evaluator, Expr, JoinClause, JoinType, RuleState, SelectStmt}; use serde_json::Value; use std::collections::HashMap; @@ -19,11 +21,11 @@ use std::sync::Arc; /// Derive the point-lookup key for an external table join from an equality /// `ON` condition between the stream side and the table side (either order). /// Returns `(table_column, key_value)`; `None` when no usable key exists. -pub(crate) fn join_key_parts( +pub(crate) fn join_key_parts( join: &JoinClause, from: &str, from_alias: Option<&str>, - combined: &HashMap, + combined: &R, ) -> Option<(String, String)> { fn side(expr: &Expr, from: &str, from_alias: Option<&str>, target: &JoinClause) -> u8 { match expr { @@ -83,11 +85,11 @@ pub(crate) fn join_key_parts( } /// Derive the point-lookup key value for a Redis table join. -pub(crate) fn extract_lookup_key( +pub(crate) fn extract_lookup_key( join: &JoinClause, from: &str, from_alias: Option<&str>, - combined: &HashMap, + combined: &R, ) -> Option { join_key_parts(join, from, from_alias, combined).map(|(_, value)| value) } @@ -173,13 +175,13 @@ pub(crate) fn resolve_sql_lookup( /// Fetch candidate rows for one join clause: a Redis `GET` point lookup for /// `TYPE="redis"` tables, a SQL point lookup for `TYPE="sql"` tables, /// otherwise the locally stored table rows. -pub(crate) async fn lookup_candidates( +pub(crate) async fn lookup_candidates( table_manager: &TableManager, source_configs: &Arc>>, from: &str, from_alias: Option<&str>, join: &JoinClause, - combined: &HashMap, + combined: &R, ) -> Vec> { let table_type = table_manager .get_table(&join.target) @@ -258,8 +260,8 @@ pub(crate) async fn apply_lookup_joins( table_manager: &TableManager, source_configs: &Arc>>, select_stmt: &SelectStmt, - record: &HashMap, -) -> Option> { + record: &IndexMap, +) -> Option> { if select_stmt.joins.is_empty() { return Some(record.clone()); } @@ -318,13 +320,19 @@ pub(crate) async fn apply_lookup_joins( /// Insert a row under its stream/table name plus alias so qualified /// references (`A.id`, `a.id`) resolve through nested objects. -pub(crate) fn insert_namespaced( - map: &mut HashMap, +pub(crate) fn insert_namespaced<'a, I>( + map: &mut IndexMap, name: &str, alias: Option<&str>, - row: &HashMap, -) { - let obj = Value::Object(row.iter().map(|(k, v)| (k.clone(), v.clone())).collect()); + row: I, +) where + I: IntoIterator, +{ + let obj = Value::Object( + row.into_iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + ); map.insert(name.to_string(), obj.clone()); if let Some(a) = alias { if a != name { @@ -352,7 +360,7 @@ pub(crate) const MAX_ASOF_KEYS: usize = 10_000; #[derive(Debug, Default)] pub(crate) struct AsofJoinState { /// Maps `(target_stream_name, join_key_value)` -> `(right_timestamp_ms, right_row)`. - latest_by_key: HashMap<(String, String), (i64, HashMap)>, + latest_by_key: HashMap<(String, String), (i64, IndexMap)>, } #[allow(dead_code)] @@ -368,7 +376,7 @@ impl AsofJoinState { select_stmt: &SelectStmt, target_stream: &str, right_ts: i64, - right_row: &HashMap, + right_row: &IndexMap, ) { for join in &select_stmt.joins { if join.target != target_stream @@ -420,8 +428,8 @@ impl AsofJoinState { &self, select_stmt: &SelectStmt, left_ts: i64, - left_row: &HashMap, - ) -> Option> { + left_row: &IndexMap, + ) -> Option> { let mut combined = left_row.clone(); insert_namespaced( &mut combined, @@ -470,7 +478,7 @@ impl AsofJoinState { join: &JoinClause, from: &str, from_alias: Option<&str>, - right_probe: &HashMap, + right_probe: &IndexMap, ) -> Option { let (col, _) = join_key_parts(join, from, from_alias, right_probe).or_else(|| { // Fallback: evaluate the right-side column directly from `right_probe` @@ -526,16 +534,16 @@ pub(crate) async fn eval_window_join_batch( select_stmt: &SelectStmt, batch: &[TaggedRow], rule_state: Option<&RuleState>, -) -> Vec> { +) -> Vec> { if batch.is_empty() { return Vec::new(); } - let from_rows: Vec<&HashMap> = batch + let from_rows: Vec<&IndexMap> = batch .iter() .filter(|r| r.source == select_stmt.from) .map(|r| &*r.data) .collect(); - let mut combined_rows: Vec> = Vec::new(); + let mut combined_rows: Vec> = Vec::new(); for row in from_rows { let mut combined = row.clone(); insert_namespaced( @@ -548,7 +556,7 @@ pub(crate) async fn eval_window_join_batch( } for join in &select_stmt.joins { let is_table = table_manager.get_table(&join.target).is_some(); - let mut next: Vec> = Vec::new(); + let mut next: Vec> = Vec::new(); if is_table { for left in &combined_rows { let mut matched: Option> = None; @@ -608,7 +616,7 @@ pub(crate) async fn eval_window_join_batch( } } } else { - let right_rows: Vec<&HashMap> = batch + let right_rows: Vec<&IndexMap> = batch .iter() .filter(|r| r.source == join.target) .map(|r| &*r.data) @@ -623,7 +631,7 @@ pub(crate) async fn eval_window_join_batch( .and_then(|v| v.as_i64().or_else(|| v.as_f64().map(|f| f as i64))) .unwrap_or(i64::MAX); - let mut best: Option<(i64, HashMap)> = None; + let mut best: Option<(i64, IndexMap)> = None; for right in &right_rows { let right_ts = right .get("timestamp") @@ -637,7 +645,7 @@ pub(crate) async fn eval_window_join_batch( for (k, v) in right.iter() { probe.entry(k.clone()).or_insert(v.clone()); } - insert_namespaced(&mut probe, &join.target, join.alias.as_deref(), right); + insert_namespaced(&mut probe, &join.target, join.alias.as_deref(), *right); let cond_ok = match &join.on { Some(cond) => Evaluator::eval_bool(cond, &probe), None => true, @@ -660,7 +668,7 @@ pub(crate) async fn eval_window_join_batch( for (k, v) in right.iter() { probe.entry(k.clone()).or_insert(v.clone()); } - insert_namespaced(&mut probe, &join.target, join.alias.as_deref(), right); + insert_namespaced(&mut probe, &join.target, join.alias.as_deref(), *right); let cond_ok = match &join.on { Some(cond) => Evaluator::eval_bool(cond, &probe), // No ON: cross product of the window. @@ -700,7 +708,7 @@ pub(crate) async fn eval_window_join_batch( &mut preserved, &join.target, join.alias.as_deref(), - right, + *right, ); next.push(preserved); } @@ -731,13 +739,13 @@ mod tests { .parse_select() .unwrap(); - let mut right1 = HashMap::new(); + let mut right1 = IndexMap::new(); right1.insert("id".to_string(), Value::Number(1.into())); right1.insert("temp".to_string(), Value::Number(20.into())); state.ingest_right_event(&stmt, "b", 100, &right1); // Probe before right1 timestamp (left_ts = 50) -> None (ASOF requires right_ts <= left_ts) - let mut left0 = HashMap::new(); + let mut left0 = IndexMap::new(); left0.insert("id".to_string(), Value::Number(1.into())); left0.insert("vib".to_string(), Value::Number(100.into())); assert!(state.probe_left_event(&stmt, 50, &left0).is_none()); @@ -747,7 +755,7 @@ mod tests { assert_eq!(matched1.get("temp"), Some(&Value::Number(20.into()))); // Ingest next right event at ts 200 - let mut right2 = HashMap::new(); + let mut right2 = IndexMap::new(); right2.insert("id".to_string(), Value::Number(1.into())); right2.insert("temp".to_string(), Value::Number(25.into())); state.ingest_right_event(&stmt, "b", 200, &right2); @@ -764,7 +772,7 @@ mod tests { .parse_select() .unwrap(); - let mut left0 = HashMap::new(); + let mut left0 = IndexMap::new(); left0.insert("id".to_string(), Value::Number(99.into())); left0.insert("vib".to_string(), Value::Number(100.into())); // Left ASOF preserves left row even if no prior right row exists diff --git a/crates/rekuiper-server/src/engine/mod.rs b/crates/rekuiper-server/src/engine/mod.rs index d0c294c0..c6759e5b 100644 --- a/crates/rekuiper-server/src/engine/mod.rs +++ b/crates/rekuiper-server/src/engine/mod.rs @@ -604,7 +604,7 @@ pub(crate) fn bootstrap_table_source(state: &AppState, table_name: &str) { tokio::select! { record = bus_rx.recv() => { match record { - Some(rec) => table_mgr_bus.insert_table_row(&t_name_bus, rec.data.into_inner()), + Some(rec) => table_mgr_bus.insert_table_row(&t_name_bus, rec.data.to_hashmap()), None => break, } } @@ -616,7 +616,7 @@ pub(crate) fn bootstrap_table_source(state: &AppState, table_name: &str) { } } => { match record { - Some(rec) => table_mgr_bus.insert_table_row(&t_name_bus, rec.data.into_inner()), + Some(rec) => table_mgr_bus.insert_table_row(&t_name_bus, rec.data.to_hashmap()), None => break, } } @@ -649,7 +649,7 @@ pub(crate) fn bootstrap_table_source(state: &AppState, table_name: &str) { tokio::select! { record = mqtt_rx.recv() => { match record { - Some(rec) => table_mgr_mqtt.insert_table_row(&t_name_mqtt, rec.data.into_inner()), + Some(rec) => table_mgr_mqtt.insert_table_row(&t_name_mqtt, rec.data.to_hashmap()), None => break, } } diff --git a/crates/rekuiper-server/src/engine/runtime.rs b/crates/rekuiper-server/src/engine/runtime.rs index 639b9591..714f421c 100644 --- a/crates/rekuiper-server/src/engine/runtime.rs +++ b/crates/rekuiper-server/src/engine/runtime.rs @@ -1,8 +1,10 @@ use crate::engine::joins::apply_lookup_joins; use crate::engine::sinks::enqueue_sink_record; +use indexmap::IndexMap; use parking_lot::RwLock; use rekuiper_core::model::{RuleStatus, StreamRecord}; use rekuiper_core::{RuleCounters, RuleManager, StreamReceiver, TableManager}; +use rekuiper_sql::eval::RowLookup; use rekuiper_sql::{Evaluator, RuleState, SelectStmt}; use serde_json::Value; use std::collections::HashMap; @@ -64,11 +66,8 @@ pub(crate) fn spawn_checkpoint_ticker( /// Extracts an upstream error message when the record carries an `error` or /// `__error` / `__raw_error__` field (emitted by source decoders on malformed payloads). #[inline] -pub(crate) fn check_record_error(data: &HashMap) -> Option { - if data.is_empty() { - return None; - } - for (k, v) in data.iter() { +pub(crate) fn check_record_error(data: &R) -> Option { + for (k, v) in data.iter_row() { if k == "error" || k == "__error" || k == "__raw_error__" { return Some( v.as_str() @@ -88,7 +87,7 @@ pub(crate) async fn handle_runtime_error( sink: &tokio::sync::mpsc::Sender, err_msg: String, stage: Option, - record_data: Option<&HashMap>, + record_data: Option<&IndexMap>, dlq_router: Option<&crate::engine::dlq::DlqRouter>, ) { if let Some(dlq) = dlq_router { @@ -106,7 +105,7 @@ pub(crate) async fn handle_runtime_error( counters.record_exception(&err_msg); } if send_error { - let mut err_data = HashMap::new(); + let mut err_data = IndexMap::new(); err_data.insert("error".to_string(), Value::String(err_msg.clone())); err_data.insert("rule_id".to_string(), Value::String(rule_id.to_string())); err_data.insert("__raw_error__".to_string(), Value::String(err_msg)); @@ -137,7 +136,7 @@ pub(crate) async fn handle_error_record( sink, err_msg, Some(rekuiper_core::model::DlqStage::CodecDecode), - Some(&record.data), + Some(&*record.data), dlq_router, ) .await; @@ -176,8 +175,8 @@ pub(crate) fn parse_timestamp_val(v: &Value) -> Option { None } -pub(crate) fn extract_event_timestamp( - data: &HashMap, +pub(crate) fn extract_event_timestamp( + data: &R, configured_field: Option<&str>, ) -> i64 { if let Some(field) = configured_field { @@ -405,7 +404,7 @@ pub(crate) async fn run_stateless_rule( &sink, err_msg, Some(rekuiper_core::model::DlqStage::CodecDecode), - Some(&record.data), + Some(&*record.data), Some(&dlq_router), ) .await; @@ -470,7 +469,7 @@ pub(crate) async fn run_stateless_rule( &sink, err, Some(rekuiper_core::model::DlqStage::SqlEval), - Some(&record.data), + Some(&*record.data), Some(&dlq_router), ) .await; diff --git a/crates/rekuiper-server/src/engine/sinks.rs b/crates/rekuiper-server/src/engine/sinks.rs index 522cdde7..9bec4122 100644 --- a/crates/rekuiper-server/src/engine/sinks.rs +++ b/crates/rekuiper-server/src/engine/sinks.rs @@ -8,6 +8,7 @@ use rekuiper_connectors::{ }; use rekuiper_core::model::StreamRecord; use rekuiper_core::{RuleCounters, RuleManager, StreamBus}; +use rekuiper_sql::eval::RowLookup; use serde_json::Value; use std::collections::HashMap; use std::sync::Arc; @@ -141,7 +142,7 @@ pub(crate) fn clean_sink_value(v: &Value, send_nil_field: bool) -> Value { .collect(), ), Value::Object(obj) => { - let mut map = std::collections::BTreeMap::new(); + let mut map = indexmap::IndexMap::new(); for (k, val) in obj { if k == rekuiper_sql::eval::META_KEY || k.starts_with("__") { continue; @@ -157,12 +158,12 @@ pub(crate) fn clean_sink_value(v: &Value, send_nil_field: bool) -> Value { } } -pub(crate) fn format_record_for_sink( - data: &HashMap, +pub(crate) fn format_record_for_sink( + data: &R, opts: &CommonSinkOpts, ) -> Value { if data.contains_key("__raw_error__") { - let mut map = std::collections::BTreeMap::new(); + let mut map = indexmap::IndexMap::new(); if let Some(err) = data.get("error") { map.insert("error".to_string(), err.clone()); } @@ -174,7 +175,7 @@ pub(crate) fn format_record_for_sink( if let Some(ref df) = opts.data_field { if let Some(val) = data.get(df) { if let Value::Object(obj) = val { - let mut map = std::collections::BTreeMap::new(); + let mut map = indexmap::IndexMap::new(); if let Some(ref fields) = opts.fields { for f in fields { let v = obj.get(f).unwrap_or(&Value::Null); @@ -200,7 +201,7 @@ pub(crate) fn format_record_for_sink( } return Value::Null; } - let mut map = std::collections::BTreeMap::new(); + let mut map = indexmap::IndexMap::new(); if let Some(ref fields) = opts.fields { for f in fields { let val = data.get(f).unwrap_or(&Value::Null); @@ -212,7 +213,7 @@ pub(crate) fn format_record_for_sink( map.insert(f.clone(), clean_sink_value(val, opts.send_nil_field)); } } else { - for (k, v) in data { + for (k, v) in data.iter() { if k == rekuiper_sql::eval::META_KEY || k.starts_with("__") { continue; } @@ -232,11 +233,7 @@ pub(crate) fn format_record_for_sink( pub(crate) fn format_record_for_sink_val(val: &Value, opts: &CommonSinkOpts) -> Value { match val { - Value::Object(map) => { - let hm: HashMap = - map.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); - format_record_for_sink(&hm, opts) - } + Value::Object(map) => format_record_for_sink(map, opts), other => other.clone(), } } @@ -993,9 +990,9 @@ impl FileBatchWriter { self.pending += 1; } - fn push_delimited( + fn push_delimited( &mut self, - data: &HashMap, + data: &R, delimiter: &str, header: bool, opts: &CommonSinkOpts, @@ -1004,7 +1001,8 @@ impl FileBatchWriter { f.clone() } else { let mut k: Vec = data - .keys() + .iter_row() + .map(|(k, _)| k) .filter(|k| *k != rekuiper_sql::eval::META_KEY && !k.starts_with("__")) .cloned() .collect(); @@ -1084,7 +1082,9 @@ pub(crate) fn action_template(opts: &Value) -> Option<&str> { opts.get("dataTemplate").and_then(|v| v.as_str()) } -pub(crate) fn record_template_map(data: &HashMap) -> serde_json::Map { +pub(crate) fn record_template_map( + data: &R, +) -> serde_json::Map { data.iter() .filter(|(k, _)| *k != rekuiper_sql::eval::META_KEY && !k.starts_with("__")) .map(|(k, v)| (k.clone(), v.clone())) @@ -1105,9 +1105,9 @@ pub(crate) fn render_sink_template_row(template: &str, row: &Value) -> String { } } -pub(crate) fn render_sink_template( +pub(crate) fn render_sink_template( template: &str, - record: &HashMap, + record: &R, send_single: bool, ) -> String { let map = record_template_map(record); @@ -1261,7 +1261,7 @@ async fn execute_rest_request( headers: &HashMap, content_type: Option<&str>, body_str: Option<&str>, - map: &HashMap, + map: &impl rekuiper_sql::RowLookup, enable_private_net: bool, ) -> Result<(), SendError> { if !enable_private_net { @@ -1439,7 +1439,7 @@ pub(crate) async fn send_action( let is_delimited = format.as_deref() == Some("delimited") || opts.format.as_deref() == Some("delimited"); - let resolve_url = |data_map: &HashMap| { + let resolve_url = |data_map: &indexmap::IndexMap| { if base_url.contains("{{") { apply_data_template(base_url, &record_template_map(data_map)) } else { @@ -1484,9 +1484,9 @@ pub(crate) async fn send_action( } if opts.send_single { for row in arr { - let row_map: HashMap = match row { + let row_map: indexmap::IndexMap = match row { Value::Object(m) => m.clone().into_iter().collect(), - _ => HashMap::new(), + _ => indexmap::IndexMap::new(), }; let final_url = resolve_url(&row_map); let body_str = if let Some(tpl) = template { @@ -1617,7 +1617,7 @@ pub(crate) async fn send_action( let is_delimited = format.as_deref() == Some("delimited") || opts.format.as_deref() == Some("delimited"); - let resolve_topic = |data_map: &HashMap| -> String { + let resolve_topic = |data_map: &indexmap::IndexMap| -> String { match destination { Some(d) => d.to_string(), None => { @@ -1643,9 +1643,9 @@ pub(crate) async fn send_action( .map_err(|e| SendError::Retry(format!("mqtt action failed: {}", e)))?; } else if opts.send_single { for row in arr { - let row_map: HashMap = match row { + let row_map: indexmap::IndexMap = match row { Value::Object(m) => m.clone().into_iter().collect(), - _ => HashMap::new(), + _ => indexmap::IndexMap::new(), }; let topic = resolve_topic(&row_map); let payload = if let Some(tpl) = template { @@ -1995,8 +1995,13 @@ pub(crate) async fn send_action( }; let payload = match template { Some(tpl) => render_sink_template(tpl, &output.data, true).into_bytes(), - None => EdgeXCodec::encode(&output.data, device_name, profile_name, source_name) - .map_err(|e| SendError::Permanent(format!("edgex payload encode: {}", e)))?, + None => EdgeXCodec::encode( + &output.data.to_hashmap(), + device_name, + profile_name, + source_name, + ) + .map_err(|e| SendError::Permanent(format!("edgex payload encode: {}", e)))?, }; sink.send_raw_to(destination.unwrap_or(&config.topic), payload) .await @@ -2029,7 +2034,7 @@ pub(crate) async fn send_action( lines.join("\n") } else { rekuiper_connectors::record_to_line_protocol( - &output.data, + &output.data.to_hashmap(), &config.measurement, config.tags.as_ref(), config.fields.as_deref(), @@ -2111,7 +2116,7 @@ pub(crate) async fn send_action( lines.join("\n") } else { rekuiper_connectors::record_to_line_protocol( - &output.data, + &output.data.to_hashmap(), &config.measurement, config.tags.as_ref(), config.fields.as_deref(), @@ -2183,9 +2188,11 @@ pub(crate) async fn send_action( SendError::Permanent(format!("tdengine batch sql generation error: {}", e)) })? } else { - config.build_insert_sql(&output.data).map_err(|e| { - SendError::Permanent(format!("tdengine sql generation error: {}", e)) - })? + config + .build_insert_sql(&output.data.to_hashmap()) + .map_err(|e| { + SendError::Permanent(format!("tdengine sql generation error: {}", e)) + })? }; if sql_statement.trim().is_empty() { @@ -2265,9 +2272,11 @@ pub(crate) async fn send_action( SendError::Permanent(format!("iotdb payload generation error: {}", e)) })? } else { - vec![config.build_insert_payload(&output.data).map_err(|e| { - SendError::Permanent(format!("iotdb payload generation error: {}", e)) - })?] + vec![config + .build_insert_payload(&output.data.to_hashmap()) + .map_err(|e| { + SendError::Permanent(format!("iotdb payload generation error: {}", e)) + })?] }; for payload in payloads { @@ -2347,7 +2356,7 @@ pub(crate) async fn send_action( } } } else { - match config.save_image(&output.data).await { + match config.save_image(&output.data.to_hashmap()).await { Ok(_) => {} Err(rekuiper_connectors::ImageSinkError::Validation(e)) => { return Err(SendError::Permanent(e)); @@ -2369,7 +2378,7 @@ pub(crate) async fn send_action( } let sink = rt.zmq.as_ref().unwrap(); - let resolve_topic = |data_map: &HashMap| { + let resolve_topic = |data_map: &indexmap::IndexMap| { let default_top = config.topic.as_deref().unwrap_or(""); let t = destination.unwrap_or(default_top); if t.contains("{{") { @@ -2382,9 +2391,9 @@ pub(crate) async fn send_action( if let Some(Value::Array(arr)) = output.data.get(BATCH_ROWS_KEY) { if opts.send_single { for row in arr { - let row_map: HashMap = match row { + let row_map: indexmap::IndexMap = match row { Value::Object(m) => m.clone().into_iter().collect(), - _ => HashMap::new(), + _ => indexmap::IndexMap::new(), }; let topic = resolve_topic(&row_map); let payload = if let Some(tpl) = template { @@ -2460,7 +2469,7 @@ pub(crate) async fn send_action( } else { let map = match format_record_for_sink(&output.data, opts) { Value::Object(m) => m.into_iter().collect(), - _ => output.data.clone(), + _ => output.data.to_hashmap(), }; sink.write_record(&map).await.map_err(|(is_perm, err_str)| { if is_perm { @@ -2495,7 +2504,7 @@ pub(crate) async fn send_action( } else { let map = match format_record_for_sink(&output.data, opts) { Value::Object(m) => m.into_iter().collect(), - _ => output.data.clone(), + _ => output.data.to_hashmap(), }; sink.write_record(&map).await.map_err(|(is_perm, err_str)| { if is_perm { @@ -2530,7 +2539,7 @@ pub(crate) async fn send_action( } else { let map = match format_record_for_sink(&output.data, opts) { Value::Object(m) => m.into_iter().collect(), - _ => output.data.clone(), + _ => output.data.to_hashmap(), }; sink.write_record(&map).await.map_err(|(is_perm, err_str)| { if is_perm { @@ -3039,7 +3048,7 @@ async fn dispatch_rule_actions_legacy( .into_bytes() } None => match EdgeXCodec::encode( - &output.data, + &output.data.to_hashmap(), device_name, profile_name, source_name, diff --git a/crates/rekuiper-server/src/engine/windows.rs b/crates/rekuiper-server/src/engine/windows.rs index dc0b4a76..d22de0e2 100644 --- a/crates/rekuiper-server/src/engine/windows.rs +++ b/crates/rekuiper-server/src/engine/windows.rs @@ -3,9 +3,11 @@ use crate::engine::runtime::{ extract_event_timestamp, handle_error_record, is_rule_running, EventTimeConfig, }; use crate::engine::sinks::{enqueue_sink_record, BATCH_ROWS_KEY}; +use indexmap::IndexMap; use parking_lot::RwLock; use rekuiper_core::model::{RuleStatus, StreamRecord}; use rekuiper_core::{RuleCounters, StreamReceiver, TableManager}; +use rekuiper_sql::eval::RowLookup; use rekuiper_sql::{Evaluator, Expr, IncrementalWindow, RuleState, SelectStmt, TimeUnit}; use serde_json::Value; use std::collections::{HashMap, HashSet}; @@ -36,7 +38,7 @@ pub(crate) async fn emit_window_batch( } let eval_start = std::time::Instant::now(); let outputs = if select_stmt.joins.is_empty() { - let rows: Vec> = + let rows: Vec> = batch.into_iter().map(|r| r.data.into_inner()).collect(); if prefiltered { Evaluator::eval_window_filtered_stateful(select_stmt, rows, rule_state) @@ -63,7 +65,7 @@ pub(crate) async fn emit_window_batch( pub(crate) async fn emit_window_outputs( counters: &RuleCounters, sink: &tokio::sync::mpsc::Sender, - outputs: Vec>, + outputs: Vec>, ) { if outputs.is_empty() { return; @@ -76,7 +78,7 @@ pub(crate) async fn emit_window_outputs( ) .await; } else { - let mut map = HashMap::new(); + let mut map = IndexMap::new(); let arr: Vec = outputs .into_iter() .map(|m| serde_json::to_value(m).unwrap_or(Value::Null)) @@ -90,9 +92,9 @@ pub(crate) async fn emit_window_outputs( /// window (as eKuiper's predicate push-down does), so rejected rows are /// never buffered. With joins it may reference joined columns and runs at /// trigger time instead. FILTER (WHERE ...) is always evaluated before the window. -pub(crate) fn window_ingest_passes( +pub(crate) fn window_ingest_passes( select_stmt: &SelectStmt, - data: &HashMap, + data: &R, ) -> bool { let where_pass = !select_stmt.joins.is_empty() || Evaluator::passes_where(select_stmt, data); if !where_pass { diff --git a/crates/rekuiper-server/src/handlers/ruletest.rs b/crates/rekuiper-server/src/handlers/ruletest.rs index 48b9edda..c8846f1d 100644 --- a/crates/rekuiper-server/src/handlers/ruletest.rs +++ b/crates/rekuiper-server/src/handlers/ruletest.rs @@ -11,6 +11,7 @@ use axum::{ routing::get, Json, }; +use indexmap::IndexMap; use parking_lot::RwLock; use rekuiper_connectors::{parse_interval_ms, SimulatorConfig}; use rekuiper_core::model::StreamRecord; @@ -169,9 +170,9 @@ pub async fn start_ruletest(State(state): State, Path(name): Path) { - let mut map = std::collections::BTreeMap::new(); - for (k, v) in row { +pub fn emit_ruletest_line(session: &RuletestSession, row: &R) { + let mut map = indexmap::IndexMap::new(); + for (k, v) in row.iter() { if k == rekuiper_sql::eval::META_KEY || k.starts_with("__") { continue; } @@ -342,13 +343,17 @@ pub async fn http_data_push_handler( } else { match serde_json::from_slice::(&body_bytes) { Ok(Value::Object(map)) => { - vec![StreamRecord::new(map.into_iter().collect())] + vec![StreamRecord::new( + map.into_iter().collect::>(), + )] } Ok(Value::Array(items)) => { let mut recs = Vec::with_capacity(items.len()); for item in items { if let Value::Object(map) = item { - recs.push(StreamRecord::new(map.into_iter().collect())); + recs.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } } recs diff --git a/crates/rekuiper-server/src/handlers/streams.rs b/crates/rekuiper-server/src/handlers/streams.rs index 7769dbd2..2a4c714b 100644 --- a/crates/rekuiper-server/src/handlers/streams.rs +++ b/crates/rekuiper-server/src/handlers/streams.rs @@ -7,6 +7,7 @@ use axum::{ response::{IntoResponse, Response}, Json, }; +use indexmap::IndexMap; use rekuiper_core::model::{StreamField, StreamRecord}; use rekuiper_core::{StreamDefinition, TableDefinition, MAX_HTTP_BATCH_RECORDS}; use rekuiper_sql::{Parser, StreamColumn}; @@ -517,11 +518,15 @@ pub async fn push_stream_data( if let Ok(Value::Array(items)) = serde_json::from_slice::(&body_bytes) { for item in items { if let Value::Object(map) = item { - records.push(StreamRecord::new(map.into_iter().collect())); + records.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } } } else if let Ok(Value::Object(map)) = serde_json::from_slice::(&body_bytes) { - records.push(StreamRecord::new(map.into_iter().collect())); + records.push(StreamRecord::new( + map.into_iter().collect::>(), + )); } else if let Ok(text) = std::str::from_utf8(&body_bytes) { for line in text.lines() { let line = line.trim(); diff --git a/crates/rekuiper-server/src/state.rs b/crates/rekuiper-server/src/state.rs index 66bbab46..febe6044 100644 --- a/crates/rekuiper-server/src/state.rs +++ b/crates/rekuiper-server/src/state.rs @@ -131,11 +131,11 @@ impl TraceManager { } } -pub fn maybe_trace_record( +pub fn maybe_trace_record( trace_mgr: &TraceManager, rule_id: &str, - input_data: &HashMap, - output_data: Option<&HashMap>, + input_data: &M1, + output_data: Option<&M2>, ) { if !trace_mgr.is_tracing(rule_id) { return; diff --git a/crates/rekuiper-server/tests/fvt_compat.rs b/crates/rekuiper-server/tests/fvt_compat.rs index ab03a4c3..7eab2bd8 100644 --- a/crates/rekuiper-server/tests/fvt_compat.rs +++ b/crates/rekuiper-server/tests/fvt_compat.rs @@ -2979,7 +2979,7 @@ async fn test_edgex_stream_e2e_data_pipeline() { // 5. Test EdgeX encoding on output record: let encoded = EdgeXCodec::encode( - &received.data, + &received.data.to_hashmap(), "kuiper_alert", "alert_profile", "rule_edgex_filter", diff --git a/crates/rekuiper-server/tests/iiot_mqtt.rs b/crates/rekuiper-server/tests/iiot_mqtt.rs index 2401f097..0d88ed20 100644 --- a/crates/rekuiper-server/tests/iiot_mqtt.rs +++ b/crates/rekuiper-server/tests/iiot_mqtt.rs @@ -330,7 +330,7 @@ async fn collect( let deadline = tokio::time::Instant::now() + within; while got.len() < want { match tokio::time::timeout_at(deadline, rx.recv()).await { - Ok(Some(record)) => got.push(record.data.into_inner()), + Ok(Some(record)) => got.push(record.data.to_hashmap()), _ => break, } } diff --git a/crates/rekuiper-server/tests/test_issue22_output_shape.rs b/crates/rekuiper-server/tests/test_issue22_output_shape.rs index f9c8bd1f..bc72e29d 100644 --- a/crates/rekuiper-server/tests/test_issue22_output_shape.rs +++ b/crates/rekuiper-server/tests/test_issue22_output_shape.rs @@ -336,9 +336,9 @@ async fn test_stable_sorted_json_key_order_in_rest_sink() { } let raw = payload.expect("REST sink did not deliver payload"); - // Keys must be alphabetical: a, then m, then z + // Keys must preserve SELECT order: z, then a, then m (Issue #45) assert_eq!( - raw, r#"{"a":200,"m":300,"z":100}"#, - "JSON key order must be stable and sorted alphabetically" + raw, r#"{"z":100,"a":200,"m":300}"#, + "JSON key order must preserve SELECT field order" ); } diff --git a/crates/rekuiper-server/tests/test_issue24_windows.rs b/crates/rekuiper-server/tests/test_issue24_windows.rs index 72147e21..df48f33a 100644 --- a/crates/rekuiper-server/tests/test_issue24_windows.rs +++ b/crates/rekuiper-server/tests/test_issue24_windows.rs @@ -1,3 +1,4 @@ +use indexmap::IndexMap; use rekuiper_conf::KuiperConfig; use rekuiper_core::manager::{RuleManager, StreamManager, TableManager}; use rekuiper_core::{StreamBus, StreamRecord}; @@ -156,7 +157,11 @@ async fn test_stateless_single_row_aggregate() { let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(1))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(1))] + .into_iter() + .collect::>(), + ), ); let rec = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -199,7 +204,11 @@ async fn test_tumbling_window_zero_duration_emits_nothing() { for i in 0..5 { let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(i))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(i))] + .into_iter() + .collect::>(), + ), ); } @@ -251,7 +260,7 @@ async fn test_window_start_and_end_event_time() { StreamRecord::new( [("ts".to_string(), json!(500)), ("v".to_string(), json!(1))] .into_iter() - .collect(), + .collect::>(), ), ); let _ = state.stream_bus.publish( @@ -259,7 +268,7 @@ async fn test_window_start_and_end_event_time() { StreamRecord::new( [("ts".to_string(), json!(1500)), ("v".to_string(), json!(2))] .into_iter() - .collect(), + .collect::>(), ), ); @@ -269,7 +278,7 @@ async fn test_window_start_and_end_event_time() { StreamRecord::new( [("ts".to_string(), json!(2500)), ("v".to_string(), json!(3))] .into_iter() - .collect(), + .collect::>(), ), ); @@ -318,7 +327,11 @@ async fn test_window_filter_clause() { for v in vals { let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(v))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(v))] + .into_iter() + .collect::>(), + ), ); } @@ -369,14 +382,22 @@ async fn test_hopping_window_event_time() { for ts in [1000, 2000, 3000, 4000, 4500] { let _ = state.stream_bus.publish( "s", - StreamRecord::new([("ts".to_string(), json!(ts))].into_iter().collect()), + StreamRecord::new( + [("ts".to_string(), json!(ts))] + .into_iter() + .collect::>(), + ), ); } // Advance watermark past 5000 by sending ts = 5500. This triggers 1st hop [0, 5000) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("ts".to_string(), json!(5500))].into_iter().collect()), + StreamRecord::new( + [("ts".to_string(), json!(5500))] + .into_iter() + .collect::>(), + ), ); let rec1 = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -391,14 +412,22 @@ async fn test_hopping_window_event_time() { for ts in [6000, 7000, 8000, 9000] { let _ = state.stream_bus.publish( "s", - StreamRecord::new([("ts".to_string(), json!(ts))].into_iter().collect()), + StreamRecord::new( + [("ts".to_string(), json!(ts))] + .into_iter() + .collect::>(), + ), ); } // Advance watermark past 10000 by sending ts = 10500. 2nd hop covers [0, 10000) = 5 + 5 = 10 events let _ = state.stream_bus.publish( "s", - StreamRecord::new([("ts".to_string(), json!(10500))].into_iter().collect()), + StreamRecord::new( + [("ts".to_string(), json!(10500))] + .into_iter() + .collect::>(), + ), ); let rec2 = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -449,14 +478,22 @@ async fn test_session_window_event_time_gap() { for i in 1..=30 { let _ = state.stream_bus.publish( "s", - StreamRecord::new([("ts".to_string(), json!(i * 1000))].into_iter().collect()), + StreamRecord::new( + [("ts".to_string(), json!(i * 1000))] + .into_iter() + .collect::>(), + ), ); } // Send 30s gap: next event at ts = 60000 (> 5000 timeout, triggers emission of 1st session) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("ts".to_string(), json!(60000))].into_iter().collect()), + StreamRecord::new( + [("ts".to_string(), json!(60000))] + .into_iter() + .collect::>(), + ), ); let rec1 = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -475,7 +512,7 @@ async fn test_session_window_event_time_gap() { StreamRecord::new( [("ts".to_string(), json!(60000 + i * 1000))] .into_iter() - .collect(), + .collect::>(), ), ); } @@ -483,7 +520,11 @@ async fn test_session_window_event_time_gap() { // Trigger second session via gap > 5s let _ = state.stream_bus.publish( "s", - StreamRecord::new([("ts".to_string(), json!(80000))].into_iter().collect()), + StreamRecord::new( + [("ts".to_string(), json!(80000))] + .into_iter() + .collect::>(), + ), ); let rec2 = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -529,7 +570,11 @@ async fn test_sliding_window_over_when() { // Send temp = 20 (does NOT trigger trigger condition) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("temp".to_string(), json!(20))].into_iter().collect()), + StreamRecord::new( + [("temp".to_string(), json!(20))] + .into_iter() + .collect::>(), + ), ); let res = tokio::time::timeout(Duration::from_millis(200), sink_rx.recv()).await; @@ -538,7 +583,11 @@ async fn test_sliding_window_over_when() { // Send temp = 35 (triggers condition! window contains 20 and 35) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("temp".to_string(), json!(35))].into_iter().collect()), + StreamRecord::new( + [("temp".to_string(), json!(35))] + .into_iter() + .collect::>(), + ), ); let rec = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -582,25 +631,41 @@ async fn test_state_window_two_conditions() { // a = 0: not on begin let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(0))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(0))] + .into_iter() + .collect::>(), + ), ); // a = 1: starts window (1st) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(1))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(1))] + .into_iter() + .collect::>(), + ), ); // a = 3: buffered (2nd) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(3))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(3))] + .into_iter() + .collect::>(), + ), ); // a = 5: emits window (3rd) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(5))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(5))] + .into_iter() + .collect::>(), + ), ); let rec = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -644,19 +709,31 @@ async fn test_state_window_single_condition() { // Send a = 10 (starts 1st window) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(10))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(10))] + .into_iter() + .collect::>(), + ), ); // Send a = 10 (buffered in 1st window) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(10))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(10))] + .into_iter() + .collect::>(), + ), ); // Send a = 20 (had_changed(a) triggers emission of 1st window!) let _ = state.stream_bus.publish( "s", - StreamRecord::new([("a".to_string(), json!(20))].into_iter().collect()), + StreamRecord::new( + [("a".to_string(), json!(20))] + .into_iter() + .collect::>(), + ), ); let rec = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) diff --git a/crates/rekuiper-server/tests/test_issue25_joins_server.rs b/crates/rekuiper-server/tests/test_issue25_joins_server.rs index 87afdc1f..e1640535 100644 --- a/crates/rekuiper-server/tests/test_issue25_joins_server.rs +++ b/crates/rekuiper-server/tests/test_issue25_joins_server.rs @@ -171,7 +171,7 @@ async fn test_defect1_stream_table_lookup_join() { ("label".to_string(), json!("one")), ] .into_iter() - .collect(), + .collect::>(), ), ); let _ = state.stream_bus.publish( @@ -182,7 +182,7 @@ async fn test_defect1_stream_table_lookup_join() { ("label".to_string(), json!("two")), ] .into_iter() - .collect(), + .collect::>(), ), ); @@ -248,7 +248,7 @@ async fn test_defect1_stream_table_lookup_join() { ("label".to_string(), json!("three")), ] .into_iter() - .collect(), + .collect::>(), ), ); tokio::time::sleep(std::time::Duration::from_millis(50)).await; diff --git a/crates/rekuiper-server/tests/test_issue34_window_having_had_changed.rs b/crates/rekuiper-server/tests/test_issue34_window_having_had_changed.rs index 5a7da9ae..ce4ee2fe 100644 --- a/crates/rekuiper-server/tests/test_issue34_window_having_had_changed.rs +++ b/crates/rekuiper-server/tests/test_issue34_window_having_had_changed.rs @@ -1,3 +1,4 @@ +use indexmap::IndexMap; use rekuiper_conf::KuiperConfig; use rekuiper_core::manager::{RuleManager, StreamManager, TableManager}; use rekuiper_core::{StreamBus, StreamRecord}; @@ -66,11 +67,11 @@ async fn test_count_window_having_had_changed() { // Window 1: val = 10, 10 -> aggregate latest(val) is 10. First window evaluation: had_changed is true. let _ = state.stream_bus.publish( "s", - StreamRecord::new([("val".to_string(), json!(10))].into_iter().collect()), + StreamRecord::new(IndexMap::from([("val".to_string(), json!(10))])), ); let _ = state.stream_bus.publish( "s", - StreamRecord::new([("val".to_string(), json!(10))].into_iter().collect()), + StreamRecord::new(IndexMap::from([("val".to_string(), json!(10))])), ); let rec1 = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -83,11 +84,11 @@ async fn test_count_window_having_had_changed() { // had_changed should return false, so HAVING filters this window output. let _ = state.stream_bus.publish( "s", - StreamRecord::new([("val".to_string(), json!(10))].into_iter().collect()), + StreamRecord::new(IndexMap::from([("val".to_string(), json!(10))])), ); let _ = state.stream_bus.publish( "s", - StreamRecord::new([("val".to_string(), json!(10))].into_iter().collect()), + StreamRecord::new(IndexMap::from([("val".to_string(), json!(10))])), ); let res_empty = tokio::time::timeout(Duration::from_millis(200), sink_rx.recv()).await; @@ -100,11 +101,11 @@ async fn test_count_window_having_had_changed() { // had_changed should return true, so Window 3 emits. let _ = state.stream_bus.publish( "s", - StreamRecord::new([("val".to_string(), json!(20))].into_iter().collect()), + StreamRecord::new(IndexMap::from([("val".to_string(), json!(20))])), ); let _ = state.stream_bus.publish( "s", - StreamRecord::new([("val".to_string(), json!(20))].into_iter().collect()), + StreamRecord::new(IndexMap::from([("val".to_string(), json!(20))])), ); let rec3 = tokio::time::timeout(Duration::from_secs(2), sink_rx.recv()) @@ -170,16 +171,12 @@ async fn test_hopping_window_having_had_changed_issue34_repro() { for (ts, s, q, f) in events { let _ = state.stream_bus.publish( "demo", - StreamRecord::new( - [ - ("ts".to_string(), json!(ts)), - ("self".to_string(), json!(s)), - ("quality".to_string(), json!(q)), - ("freshness".to_string(), json!(f)), - ] - .into_iter() - .collect(), - ), + StreamRecord::new(IndexMap::from([ + ("ts".to_string(), json!(ts)), + ("self".to_string(), json!(s)), + ("quality".to_string(), json!(q)), + ("freshness".to_string(), json!(f)), + ])), ); } diff --git a/crates/rekuiper-sql/Cargo.toml b/crates/rekuiper-sql/Cargo.toml index ac2fe278..e3fc4ba0 100644 --- a/crates/rekuiper-sql/Cargo.toml +++ b/crates/rekuiper-sql/Cargo.toml @@ -13,6 +13,7 @@ categories.workspace = true [dependencies] serde = { workspace = true } serde_json = { workspace = true } +indexmap = { workspace = true } anyhow = { workspace = true } thiserror = { workspace = true } regex = { workspace = true } diff --git a/crates/rekuiper-sql/src/eval.rs b/crates/rekuiper-sql/src/eval.rs index 62b75986..91b2bb01 100644 --- a/crates/rekuiper-sql/src/eval.rs +++ b/crates/rekuiper-sql/src/eval.rs @@ -1,9 +1,122 @@ use crate::ast::{BinaryOperator, Expr, SelectStmt, SetOp, UnaryOperator}; +use indexmap::IndexMap; use parking_lot::RwLock; use serde_json::Value; use std::collections::HashMap; use std::sync::Arc; +/// Trait abstracting read-only row key/value lookups and iteration. +/// Monomorphized over row storage types (`IndexMap`, `HashMap`, `serde_json::Map`, `SharedMap`) +/// to ensure zero heap allocations and full inlining on the evaluation hot path. +pub trait RowLookup { + type Iter<'a>: Iterator + where + Self: 'a; + + fn get(&self, key: &str) -> Option<&Value>; + fn iter(&self) -> Self::Iter<'_>; + + #[inline(always)] + fn iter_row(&self) -> Self::Iter<'_> { + self.iter() + } + + #[inline(always)] + fn contains_key(&self, key: &str) -> bool { + self.get(key).is_some() + } +} + +impl RowLookup for HashMap { + type Iter<'a> = std::collections::hash_map::Iter<'a, String, Value>; + + #[inline(always)] + fn get(&self, key: &str) -> Option<&Value> { + HashMap::get(self, key) + } + + #[inline(always)] + fn iter(&self) -> Self::Iter<'_> { + self.iter() + } +} + +impl RowLookup for IndexMap { + type Iter<'a> = indexmap::map::Iter<'a, String, Value>; + + #[inline(always)] + fn get(&self, key: &str) -> Option<&Value> { + IndexMap::get(self, key) + } + + #[inline(always)] + fn iter(&self) -> Self::Iter<'_> { + self.iter() + } +} + +impl RowLookup for serde_json::Map { + type Iter<'a> = serde_json::map::Iter<'a>; + + #[inline(always)] + fn get(&self, key: &str) -> Option<&Value> { + serde_json::Map::get(self, key) + } + + #[inline(always)] + fn iter(&self) -> Self::Iter<'_> { + self.iter() + } +} + +impl RowLookup for rekuiper_core::model::SharedMap { + type Iter<'a> = indexmap::map::Iter<'a, String, Value>; + + #[inline(always)] + fn get(&self, key: &str) -> Option<&Value> { + self.0.get(key) + } + + #[inline(always)] + fn iter(&self) -> Self::Iter<'_> { + self.0.iter() + } +} + +impl RowLookup for &T { + type Iter<'a> + = T::Iter<'a> + where + Self: 'a; + + #[inline(always)] + fn get(&self, key: &str) -> Option<&Value> { + (**self).get(key) + } + + #[inline(always)] + fn iter(&self) -> Self::Iter<'_> { + (**self).iter() + } +} + +impl RowLookup for Arc { + type Iter<'a> + = T::Iter<'a> + where + Self: 'a; + + #[inline(always)] + fn get(&self, key: &str) -> Option<&Value> { + (**self).get(key) + } + + #[inline(always)] + fn iter(&self) -> Self::Iter<'_> { + (**self).iter() + } +} + pub mod aggregate; pub mod analytic; pub mod array; @@ -98,9 +211,9 @@ impl Evaluator { /// the right branch winning conflicts. `Union` vs `UnionAll` multiplicity /// only manifests in multi-row APIs; see `eval_select_stateful_multi`. fn merge_union_rows( - left: Option>, - right: Option>, - ) -> Option> { + left: Option>, + right: Option>, + ) -> Option> { match (left, right) { (Some(mut l), Some(r)) => { for (k, v) in r { @@ -131,10 +244,10 @@ impl Evaluator { } } - pub fn eval_select( + pub fn eval_select( stmt: &SelectStmt, - record: &HashMap, - ) -> Option> { + record: &R, + ) -> Option> { if let Some((_, rhs)) = &stmt.set_op { // Each branch filters/projects the same record independently. let left = Self::eval_select(&Self::without_set_op(stmt), record); @@ -153,10 +266,10 @@ impl Evaluator { .flatten() } - pub fn eval_select_fallible( + pub fn eval_select_fallible( stmt: &SelectStmt, - record: &HashMap, - ) -> Result>, String> { + record: &R, + ) -> Result>, String> { let state = RuleState::default(); Self::eval_select_stateful_fallible(stmt, record, &state) } @@ -168,19 +281,19 @@ impl Evaluator { /// and applies `having` (evaluated against the aggregated row, with /// aggregate calls resolved over the batch). Returns `None` when `having` /// filters the row out. - pub fn eval_aggregate( + pub fn eval_aggregate( stmt: &SelectStmt, - records: &[HashMap], - ) -> Option> { + records: &[R], + ) -> Option> { Self::eval_aggregate_stateful(stmt, records, None, "") } - pub fn eval_aggregate_stateful( + pub fn eval_aggregate_stateful( stmt: &SelectStmt, - records: &[HashMap], + records: &[R], state: Option<&RuleState>, partition_key: &str, - ) -> Option> { + ) -> Option> { if let Some((_, rhs)) = &stmt.set_op { // Each branch aggregates the same batch independently; the two // single-row outputs merge (right wins on conflict). @@ -193,15 +306,15 @@ impl Evaluator { let right = Self::eval_aggregate_stateful(rhs, records, state, partition_key); return Self::merge_union_rows(left, right); } - let first: Option<&HashMap> = records.first(); - let mut output = HashMap::new(); + let first: Option<&R> = records.first(); + let mut output = IndexMap::new(); let field_names = Self::select_field_names(stmt); for (idx, field) in stmt.fields.iter().enumerate() { match field { Expr::Wildcard => { if let Some(rec) = first { - for (k, v) in rec { + for (k, v) in rec.iter_row() { if k != META_KEY && !k.starts_with("__") { output.entry(k.clone()).or_insert_with(|| v.clone()); } @@ -210,7 +323,7 @@ impl Evaluator { } Expr::WildcardModified { except, replace } => { if let Some(rec) = first { - for (k, v) in rec { + for (k, v) in rec.iter_row() { if k != META_KEY && !k.starts_with("__") { if except.iter().any(|e| e == k) && !replace.iter().any(|(_, c)| c == k) @@ -253,11 +366,11 @@ impl Evaluator { } } } else if let Expr::Identifier(p_name) = parent.as_ref() { - let has_other_namespace = rec.iter().any(|(k, v)| { + let has_other_namespace = rec.iter_row().any(|(k, v)| { k != p_name && !k.starts_with("__") && matches!(v, Value::Object(_)) }); if !has_other_namespace { - for (k, v) in rec { + for (k, v) in rec.iter_row() { if k != META_KEY && !k.starts_with("__") { output.entry(k.clone()).or_insert_with(|| v.clone()); } @@ -340,10 +453,10 @@ impl Evaluator { ) } - fn eval_agg_expr_stateful( + fn eval_agg_expr_stateful( expr: &Expr, - records: &[HashMap], - output: &HashMap, + records: &[R], + output: &IndexMap, state: Option<&RuleState>, partition_key: &str, ) -> Value { @@ -583,11 +696,11 @@ impl Evaluator { /// filter to decide emission. This mirrors eKuiper analytic semantics, /// where e.g. `acc_map_agg` accumulates rows that a later `WHERE` filters /// out of the output. - pub fn eval_select_stateful( + pub fn eval_select_stateful( stmt: &SelectStmt, - record: &HashMap, + record: &R, state: &RuleState, - ) -> Option> { + ) -> Option> { Self::eval_select_stateful_fallible(stmt, record, state) .ok() .flatten() @@ -718,11 +831,11 @@ impl Evaluator { /// Advance analytic functions before WHERE, then project only passing rows. /// The per-row cache prevents SELECT and WHERE from advancing the same call twice. - pub fn eval_select_filtered_stateful_fallible( + pub fn eval_select_filtered_stateful_fallible( stmt: &SelectStmt, - record: &HashMap, + record: &R, state: &RuleState, - ) -> Result>, String> { + ) -> Result>, String> { let field_names = Self::select_field_names(stmt); let has_analytic = Self::stmt_has_analytic(stmt); Self::eval_select_stateless_fast(stmt, &field_names, has_analytic, record, state) @@ -731,13 +844,13 @@ impl Evaluator { /// Reuses pre-computed `field_names` and `has_analytic` to avoid traversing /// the AST or re-computing field alias names on every row. #[inline] - pub fn eval_select_stateless_fast( + pub fn eval_select_stateless_fast( stmt: &SelectStmt, field_names: &[String], has_analytic: bool, - record: &HashMap, + record: &R, state: &RuleState, - ) -> Result>, String> { + ) -> Result>, String> { let Some(condition) = &stmt.where_clause else { return Self::eval_select_stateful_with_names(stmt, field_names, record, state); }; @@ -894,9 +1007,9 @@ impl Evaluator { } #[inline] - fn try_eval_numeric_fast(expr: &Expr, record: &HashMap) -> Option { + fn try_eval_numeric_fast(expr: &Expr, record: &R) -> Option { #[inline] - fn eval_f64(e: &Expr, rec: &HashMap, all_i64: &mut bool) -> Option { + fn eval_f64(e: &Expr, rec: &R, all_i64: &mut bool) -> Option { match e { Expr::Literal(v) => { if !v.is_i64() { @@ -957,41 +1070,41 @@ impl Evaluator { serde_json::Number::from_f64(res).map(Value::Number) } - pub fn eval_select_stateful_fallible( + pub fn eval_select_stateful_fallible( stmt: &SelectStmt, - record: &HashMap, + record: &R, state: &RuleState, - ) -> Result>, String> { + ) -> Result>, String> { let field_names = Self::select_field_names(stmt); Self::eval_select_stateful_with_names(stmt, &field_names, record, state) } #[inline] - pub fn eval_select_stateful_with_names( + pub fn eval_select_stateful_with_names( stmt: &SelectStmt, field_names: &[String], - record: &HashMap, + record: &R, state: &RuleState, - ) -> Result>, String> { + ) -> Result>, String> { if let Some((_, rhs)) = &stmt.set_op { let left = Self::eval_select_stateful_fallible(&Self::without_set_op(stmt), record, state)?; let right = Self::eval_select_stateful_fallible(rhs, record, state)?; return Ok(Self::merge_union_rows(left, right)); } - let mut output = HashMap::with_capacity(stmt.fields.len()); + let mut output = IndexMap::with_capacity(stmt.fields.len()); for (idx, field) in stmt.fields.iter().enumerate() { let alias = stmt.field_aliases.get(idx).and_then(|a| a.as_deref()); match field { Expr::Wildcard => { - for (k, v) in record { + for (k, v) in record.iter() { if k != META_KEY && !k.starts_with("__") { output.insert(k.clone(), v.clone()); } } } Expr::WildcardModified { except, replace } => { - for (k, v) in record { + for (k, v) in record.iter() { if k != META_KEY && !k.starts_with("__") { if except.iter().any(|e| e == k) && !replace.iter().any(|(_, c)| c == k) { @@ -1057,7 +1170,7 @@ impl Evaluator { k != p_name && !k.starts_with("__") && matches!(v, Value::Object(_)) }); if !has_other_namespace { - for (k, v) in record { + for (k, v) in record.iter() { if k != META_KEY && !k.starts_with("__") { output.insert(k.clone(), v.clone()); } @@ -1119,7 +1232,7 @@ impl Evaluator { }; for (arg_idx, expr_arg) in args[2..].iter().enumerate() { if matches!(expr_arg, Expr::Wildcard) { - for (k, v) in record { + for (k, v) in record.iter() { if k == META_KEY { continue; } @@ -1198,11 +1311,11 @@ impl Evaluator { /// field alias (or generated column name). A non-array argument passes /// the base row through unchanged; without any `unnest` field the result /// is a single-element vector. - pub fn eval_select_stateful_multi( + pub fn eval_select_stateful_multi( stmt: &SelectStmt, - record: &HashMap, + record: &R, state: &RuleState, - ) -> Vec> { + ) -> Vec> { let mut rows = Self::eval_own_rows_multi(stmt, record, state); if let Some((op, rhs)) = &stmt.set_op { let right = Self::eval_select_stateful_multi(rhs, record, state); @@ -1221,11 +1334,11 @@ impl Evaluator { /// Single-statement portion of [`Self::eval_select_stateful_multi`]: /// projection (+ unnest expansion) without following `set_op`. - fn eval_own_rows_multi( + fn eval_own_rows_multi( stmt: &SelectStmt, - record: &HashMap, + record: &R, state: &RuleState, - ) -> Vec> { + ) -> Vec> { // NOTE: set_op-free base on purpose — the caller combines branches. let single = Self::without_set_op(stmt); let Some(base) = Self::eval_select_stateful(&single, record, state) else { @@ -1251,7 +1364,7 @@ impl Evaluator { let mut rows = Vec::with_capacity(items.len()); for item in items { let mut row = base.clone(); - row.remove(&placeholder); + row.shift_remove(&placeholder); match item { Value::Object(map) => { for (k, v) in map { @@ -1269,17 +1382,17 @@ impl Evaluator { /// Evaluate an expression with analytic (`acc_*`) calls resolved against /// `state`. All other expressions behave exactly like [`Self::eval_val`]. - fn eval_stateful_expr( + fn eval_stateful_expr( expr: &Expr, - record: &HashMap, + record: &R, state: &RuleState, ) -> Value { Self::eval_stateful_expr_fallible(expr, record, state).unwrap_or(Value::Null) } - pub fn eval_stateful_expr_fallible( + pub fn eval_stateful_expr_fallible( expr: &Expr, - record: &HashMap, + record: &R, state: &RuleState, ) -> Result { if !state.row_values.is_empty() && Self::is_analytic_expr(expr) { @@ -1428,9 +1541,9 @@ impl Evaluator { } } - fn eval_stateful_call_fallible( + fn eval_stateful_call_fallible( expr: &Expr, - record: &HashMap, + record: &R, state: &RuleState, partition_key: Option<&str>, skip_update: bool, @@ -1610,10 +1723,12 @@ impl Evaluator { name.to_ascii_lowercase().as_str(), "count" | "sum" | "avg" | "min" | "max" ) { + let row_map: HashMap = + record.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); return Ok(Self::eval_aggregate_call( name, args, - std::slice::from_ref(record), + std::slice::from_ref(&row_map), )); } let vals: Vec = args @@ -1895,17 +2010,17 @@ impl Evaluator { schema } - pub fn eval_bool_fallible( + pub fn eval_bool_fallible( expr: &Expr, - record: &HashMap, + record: &R, ) -> Result { Self::eval_bool_stateful_fallible(expr, record, &RuleState::new()) } #[inline] - pub fn eval_bool_stateful_fallible( + pub fn eval_bool_stateful_fallible( expr: &Expr, - record: &HashMap, + record: &R, state: &RuleState, ) -> Result { // Fast path: direct borrowed comparison for `col literal` or `literal col` @@ -1991,26 +2106,26 @@ impl Evaluator { } } - pub fn eval_bool(expr: &Expr, record: &HashMap) -> bool { + pub fn eval_bool(expr: &Expr, record: &R) -> bool { Self::eval_bool_fallible(expr, record).unwrap_or(false) } - pub fn eval_bool_stateful( + pub fn eval_bool_stateful( expr: &Expr, - record: &HashMap, + record: &R, state: &RuleState, ) -> bool { Self::eval_bool_stateful_fallible(expr, record, state).unwrap_or(false) } - pub fn eval_val_fallible( + pub fn eval_val_fallible( expr: &Expr, - record: &HashMap, + record: &R, ) -> Result { Self::eval_stateful_expr_fallible(expr, record, &RuleState::new()) } - pub fn eval_val(expr: &Expr, record: &HashMap) -> Value { + pub fn eval_val(expr: &Expr, record: &R) -> Value { Self::eval_val_fallible(expr, record).unwrap_or(Value::Null) } @@ -2872,10 +2987,10 @@ impl Evaluator { /// Resolve a contextual system function against the current record. /// Returns `None` for non-contextual names so callers fall through to /// normal argument evaluation. - fn eval_context_call( + fn eval_context_call( name: &str, args: &[Expr], - record: &HashMap, + record: &R, ) -> Option { match name.to_ascii_lowercase().as_str() { "meta" => Some(Self::resolve_meta( @@ -2905,14 +3020,14 @@ impl Evaluator { /// Look a metadata key up in the first present meta object, falling back /// to a top-level record field. No argument returns the whole object. - fn resolve_meta( + fn resolve_meta( arg: Option<&Expr>, - record: &HashMap, + record: &R, object_keys: &[&str], ) -> Value { let meta_obj = object_keys .iter() - .find_map(|k| record.get(*k)) + .find_map(|k| record.get(k)) .and_then(|v| v.as_object()); let Some(arg) = arg else { return meta_obj @@ -2949,7 +3064,7 @@ impl Evaluator { record.get(&key).cloned().unwrap_or(Value::Null) } - fn resolve_event_time(record: &HashMap) -> Value { + fn resolve_event_time(record: &R) -> Value { for key in ["timestamp", "event_time", "__timestamp__"] { if let Some(v) = record.get(key) { if !v.is_null() { @@ -2960,23 +3075,23 @@ impl Evaluator { serde_json::json!(chrono::Utc::now().timestamp_millis()) } - fn resolve_rule_id(record: &HashMap) -> Value { + fn resolve_rule_id(record: &R) -> Value { match record.get("__rule_id__") { Some(Value::String(s)) => Value::String(s.clone()), _ => Value::String(String::new()), } } - fn resolve_rule_start(record: &HashMap) -> Value { + fn resolve_rule_start(record: &R) -> Value { match record.get("__rule_start__") { Some(Value::Number(n)) => Value::Number(n.clone()), _ => Value::from(chrono::Utc::now().timestamp_millis()), } } - fn resolve_window_bound(record: &HashMap, keys: &[&str]) -> Value { + fn resolve_window_bound(record: &R, keys: &[&str]) -> Value { keys.iter() - .find_map(|k| record.get(*k)) + .find_map(|k| record.get(k)) .cloned() .unwrap_or(Value::Null) } diff --git a/crates/rekuiper-sql/src/eval/aggregate.rs b/crates/rekuiper-sql/src/eval/aggregate.rs index e8556162..2e010d96 100644 --- a/crates/rekuiper-sql/src/eval/aggregate.rs +++ b/crates/rekuiper-sql/src/eval/aggregate.rs @@ -1,23 +1,22 @@ -use super::{ArithOp, Evaluator, RuleState, META_KEY}; +use super::{ArithOp, Evaluator, RowLookup, RuleState, META_KEY}; use crate::ast::Expr; use serde_json::Value; -use std::collections::HashMap; const MAX_ACC_ITEMS: usize = 10_000; impl Evaluator { - pub(crate) fn eval_aggregate_call( + pub(crate) fn eval_aggregate_call( name: &str, args: &[Expr], - records: &[HashMap], + records: &[R], ) -> Value { Self::eval_aggregate_call_stateful(name, args, records, None, "") } - pub(crate) fn eval_aggregate_call_stateful( + pub(crate) fn eval_aggregate_call_stateful( name: &str, args: &[Expr], - records: &[HashMap], + records: &[R], state: Option<&RuleState>, _partition_key: &str, ) -> Value { @@ -338,8 +337,8 @@ impl Evaluator { Value::Array(list) } - pub(crate) fn agg_last_agg_hit_count_stateful( - records: &[HashMap], + pub(crate) fn agg_last_agg_hit_count_stateful( + records: &[R], state: Option<&RuleState>, ) -> Value { if let Some(s) = state { @@ -353,8 +352,8 @@ impl Evaluator { } } - pub(crate) fn agg_last_agg_hit_time_stateful( - records: &[HashMap], + pub(crate) fn agg_last_agg_hit_time_stateful( + records: &[R], state: Option<&RuleState>, ) -> Value { if let Some(s) = state { @@ -373,13 +372,13 @@ impl Evaluator { } } - pub(crate) fn agg_last_agg_hit_count(_records: &[HashMap]) -> Value { + pub(crate) fn agg_last_agg_hit_count(_records: &[R]) -> Value { static COUNT: std::sync::atomic::AtomicI64 = std::sync::atomic::AtomicI64::new(0); let cur = COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); Value::from(cur) } - pub(crate) fn agg_last_agg_hit_time(records: &[HashMap]) -> Value { + pub(crate) fn agg_last_agg_hit_time(records: &[R]) -> Value { static LAST_TIME: std::sync::atomic::AtomicI64 = std::sync::atomic::AtomicI64::new(0); let prev = LAST_TIME.load(std::sync::atomic::Ordering::Relaxed); let window_end = records @@ -391,7 +390,7 @@ impl Evaluator { Value::from(prev) } - pub(crate) fn agg_count(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_count(args: &[Expr], records: &[R]) -> Value { if args.is_empty() { return Value::Null; } @@ -410,7 +409,7 @@ impl Evaluator { } /// Collect numeric arg values across the batch, skipping nulls/non-numerics. - pub(crate) fn agg_numeric_values(arg: &Expr, records: &[HashMap]) -> Vec { + pub(crate) fn agg_numeric_values(arg: &Expr, records: &[R]) -> Vec { let mut out = Vec::new(); for rec in records { let v = Self::eval_val(arg, rec); @@ -424,7 +423,7 @@ impl Evaluator { out } - pub(crate) fn agg_sum(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_sum(args: &[Expr], records: &[R]) -> Value { if args.len() != 1 { return Value::Null; } @@ -454,7 +453,7 @@ impl Evaluator { serde_json::json!(sum) } - pub(crate) fn agg_avg(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_avg(args: &[Expr], records: &[R]) -> Value { if args.len() != 1 { return Value::Null; } @@ -469,7 +468,7 @@ impl Evaluator { serde_json::json!(sum / (vals.len() as f64)) } - pub(crate) fn agg_min(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_min(args: &[Expr], records: &[R]) -> Value { if args.len() != 1 { return Value::Null; } @@ -504,7 +503,7 @@ impl Evaluator { best.clone() } - pub(crate) fn agg_max(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_max(args: &[Expr], records: &[R]) -> Value { if args.len() != 1 { return Value::Null; } @@ -541,7 +540,7 @@ impl Evaluator { /// `collect(col)`: all non-null values of `col` across the window batch, /// in row order. - pub(crate) fn agg_collect(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_collect(args: &[Expr], records: &[R]) -> Value { if args.len() != 1 { return Value::Null; } @@ -551,7 +550,7 @@ impl Evaluator { .iter() .map(|rec| { let mut map = serde_json::Map::new(); - for (k, v) in rec { + for (k, v) in rec.iter_row() { if k != META_KEY && !k.starts_with("__") { map.insert(k.clone(), v.clone()); } @@ -572,7 +571,7 @@ impl Evaluator { /// `latest(col, [default])`: the most recent (last) non-null value of `col` in the /// batch, or `default` (or `Null`) when there is none. - pub(crate) fn agg_latest(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_latest(args: &[Expr], records: &[R]) -> Value { if args.is_empty() || args.len() > 2 { return Value::Null; } @@ -600,7 +599,7 @@ impl Evaluator { /// (default 1) counts forward from the first row (0-based); out-of-range /// offsets yield `default` (default `Null`). `ignoreNull` (default `true`) /// skips null records while counting forward. - pub(crate) fn agg_lead(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_lead(args: &[Expr], records: &[R]) -> Value { if args.is_empty() || args.len() > 4 { return Value::Null; } @@ -658,7 +657,7 @@ impl Evaluator { /// Sorted numeric column values across the window batch, skipping /// nulls and non-numerics. Shared by the statistical aggregates. - pub(crate) fn agg_sorted_numbers(arg: &Expr, records: &[HashMap]) -> Vec { + pub(crate) fn agg_sorted_numbers(arg: &Expr, records: &[R]) -> Vec { let mut vals: Vec = Self::agg_numeric_values(arg, records) .iter() .filter_map(|v| v.as_f64()) @@ -667,7 +666,7 @@ impl Evaluator { vals } - pub(crate) fn agg_median(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_median(args: &[Expr], records: &[R]) -> Value { if args.len() != 1 { return Value::Null; } @@ -699,10 +698,7 @@ impl Evaluator { /// Single-pass Welford's online algorithm for sum-of-squared-deviations; /// returns `(ssd, count)` with zero heap allocations and zero sorting. - pub(crate) fn agg_ssd( - args: &[Expr], - records: &[HashMap], - ) -> Option<(f64, usize)> { + pub(crate) fn agg_ssd(args: &[Expr], records: &[R]) -> Option<(f64, usize)> { if args.len() != 1 || matches!(args[0], Expr::Wildcard) { return None; } @@ -727,7 +723,7 @@ impl Evaluator { } } - pub(crate) fn agg_stddev(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_stddev(args: &[Expr], records: &[R]) -> Value { match Self::agg_ssd(args, records) { None => Value::Null, Some((_, 0)) => Value::Null, @@ -735,7 +731,7 @@ impl Evaluator { } } - pub(crate) fn agg_stddevs(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_stddevs(args: &[Expr], records: &[R]) -> Value { match Self::agg_ssd(args, records) { // Sample statistics need at least 2 points (0 dof otherwise). None => Value::Null, @@ -744,7 +740,7 @@ impl Evaluator { } } - pub(crate) fn agg_var(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_var(args: &[Expr], records: &[R]) -> Value { match Self::agg_ssd(args, records) { None => Value::Null, Some((_, 0)) => Value::Null, @@ -752,7 +748,7 @@ impl Evaluator { } } - pub(crate) fn agg_vars(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_vars(args: &[Expr], records: &[R]) -> Value { match Self::agg_ssd(args, records) { None => Value::Null, Some((_, n)) if n < 2 => Value::Null, @@ -763,7 +759,7 @@ impl Evaluator { /// Evaluate the percentile fraction `p` against the first batch record /// (constants in practice); `None` when missing, non-numeric or outside /// `[0, 1]`. - fn agg_percentile_p(args: &[Expr], records: &[HashMap]) -> Option { + fn agg_percentile_p(args: &[Expr], records: &[R]) -> Option { if args.len() != 2 { return None; } @@ -778,7 +774,7 @@ impl Evaluator { Some(p) } - pub(crate) fn agg_percentile(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_percentile(args: &[Expr], records: &[R]) -> Value { let Some(p) = Self::agg_percentile_p(args, records) else { return Value::Null; }; @@ -799,7 +795,7 @@ impl Evaluator { serde_json::json!(vals[i] + frac * (vals[i + 1] - vals[i])) } - pub(crate) fn agg_percentile_disc(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_percentile_disc(args: &[Expr], records: &[R]) -> Value { let Some(p) = Self::agg_percentile_p(args, records) else { return Value::Null; }; @@ -826,7 +822,7 @@ impl Evaluator { raw[index].clone() } - pub(crate) fn agg_last_value(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_last_value(args: &[Expr], records: &[R]) -> Value { if args.is_empty() || args.len() > 2 { return Value::Null; } @@ -843,15 +839,25 @@ impl Evaluator { return records .last() .map(|rec| { - Value::Object(rec.iter().map(|(k, v)| (k.clone(), v.clone())).collect()) + Value::Object( + rec.iter_row() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + ) }) .unwrap_or(Value::Null); } return records .iter() .rev() - .find(|rec| rec.values().any(|v| !v.is_null())) - .map(|rec| Value::Object(rec.iter().map(|(k, v)| (k.clone(), v.clone())).collect())) + .find(|rec| rec.iter_row().any(|(_, v)| !v.is_null())) + .map(|rec| { + Value::Object( + rec.iter_row() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + ) + }) .unwrap_or(Value::Null); } if !ignore_nulls { @@ -868,14 +874,14 @@ impl Evaluator { .unwrap_or(Value::Null) } - pub(crate) fn agg_merge_agg(args: &[Expr], records: &[HashMap]) -> Value { + pub(crate) fn agg_merge_agg(args: &[Expr], records: &[R]) -> Value { if args.len() != 1 { return Value::Null; } let mut merged = serde_json::Map::new(); if matches!(args[0], Expr::Wildcard) { for rec in records { - for (k, v) in rec { + for (k, v) in rec.iter_row() { merged.insert(k.clone(), v.clone()); } } @@ -891,7 +897,7 @@ impl Evaluator { Value::Object(merged) } - pub(crate) fn agg_row_number(args: &[Expr], _records: &[HashMap]) -> Value { + pub(crate) fn agg_row_number(args: &[Expr], _records: &[R]) -> Value { if !args.is_empty() { return Value::Null; } diff --git a/crates/rekuiper-sql/src/eval/window.rs b/crates/rekuiper-sql/src/eval/window.rs index 64349cbd..f8414fba 100644 --- a/crates/rekuiper-sql/src/eval/window.rs +++ b/crates/rekuiper-sql/src/eval/window.rs @@ -10,8 +10,9 @@ //! (group columns plus `count/sum/avg/min/max`) without retaining rows: memory //! is O(groups) instead of O(rows in the window). -use super::{Evaluator, RuleState}; +use super::{Evaluator, RowLookup, RuleState}; use crate::ast::{Expr, SelectStmt, SortOrder}; +use indexmap::IndexMap; use serde_json::Value; use std::borrow::Cow; use std::collections::{HashMap, VecDeque}; @@ -19,7 +20,7 @@ use std::fmt::Write as _; static NULL: Value = Value::Null; -type Row = HashMap; +type Row = IndexMap; impl Evaluator { /// True when `expr` calls an aggregate function anywhere inside it. @@ -94,7 +95,7 @@ impl Evaluator { } /// Input-row `WHERE` test (rows pass when there is no `WHERE`). - pub fn passes_where(stmt: &SelectStmt, row: &Row) -> bool { + pub fn passes_where(stmt: &SelectStmt, row: &R) -> bool { match &stmt.where_clause { Some(cond) => Self::eval_bool(cond, row), None => true, @@ -742,7 +743,7 @@ impl IncrementalWindow { let groups = std::mem::take(&mut self.groups); let mut out = Vec::with_capacity(groups.len()); for group in groups { - let mut row: Row = HashMap::with_capacity(self.group_names.len() + self.plans.len()); + let mut row: Row = IndexMap::with_capacity(self.group_names.len() + self.plans.len()); for (name, value) in self.group_names.iter().zip(group.group_values) { row.insert(name.clone(), value); } @@ -1140,7 +1141,7 @@ impl SlidingIncrementalWindow { if !has_active && !group.accs.is_empty() { continue; } - let mut row: Row = HashMap::with_capacity(self.group_names.len() + self.plans.len()); + let mut row: Row = IndexMap::with_capacity(self.group_names.len() + self.plans.len()); for (name, value) in self.group_names.iter().zip(&group.group_values) { row.insert(name.clone(), value.clone()); } diff --git a/crates/rekuiper-sql/tests/sql_test.rs b/crates/rekuiper-sql/tests/sql_test.rs index 8797aee9..94e7d5e7 100644 --- a/crates/rekuiper-sql/tests/sql_test.rs +++ b/crates/rekuiper-sql/tests/sql_test.rs @@ -1165,7 +1165,7 @@ fn test_lag_stateful() { record.insert("temp".to_string(), t); Evaluator::eval_select_stateful(&stmt, &record, &state) .expect("row always projects") - .remove("prev") + .shift_remove("prev") .unwrap() }; @@ -1183,7 +1183,7 @@ fn test_lag_stateful() { record.insert("temp".to_string(), t); Evaluator::eval_select_stateful(&stmt, &record, &state) .expect("row always projects") - .remove("prev2") + .shift_remove("prev2") .unwrap() }; assert_eq!(row(json!(10)), json!(-1)); diff --git a/crates/rekuiper-sql/tests/test_issue17_analytic_filter.rs b/crates/rekuiper-sql/tests/test_issue17_analytic_filter.rs index 0af23149..030e4d6c 100644 --- a/crates/rekuiper-sql/tests/test_issue17_analytic_filter.rs +++ b/crates/rekuiper-sql/tests/test_issue17_analytic_filter.rs @@ -1,13 +1,17 @@ +use indexmap::IndexMap; use rekuiper_sql::{ eval::{Evaluator, RuleState}, parser::Parser, }; use serde_json::{json, Value}; -use std::collections::HashMap; -fn run(sql: &str, row: Value, state: &RuleState) -> Result>, String> { +fn run( + sql: &str, + row: Value, + state: &RuleState, +) -> Result>, String> { let stmt = Parser::new(sql).parse_select().unwrap(); - let row = serde_json::from_value(row).unwrap(); + let row: IndexMap = serde_json::from_value(row).unwrap(); Evaluator::eval_select_filtered_stateful_fallible(&stmt, &row, state) } diff --git a/crates/rekuiper-sql/tests/test_issue22_parity.rs b/crates/rekuiper-sql/tests/test_issue22_parity.rs index bd34b136..7e54db14 100644 --- a/crates/rekuiper-sql/tests/test_issue22_parity.rs +++ b/crates/rekuiper-sql/tests/test_issue22_parity.rs @@ -9,7 +9,7 @@ fn json_to_record(v: &Value) -> HashMap { } } -fn run_sql(sql: &str, record: &serde_json::Value) -> HashMap { +fn run_sql(sql: &str, record: &serde_json::Value) -> indexmap::IndexMap { let mut parser = Parser::new(sql); let stmt = parser.parse_select().expect("failed to parse SQL"); let rec = json_to_record(record); diff --git a/crates/rekuiper-sql/tests/test_issue27_parity.rs b/crates/rekuiper-sql/tests/test_issue27_parity.rs index c5a40859..b193bcdb 100644 --- a/crates/rekuiper-sql/tests/test_issue27_parity.rs +++ b/crates/rekuiper-sql/tests/test_issue27_parity.rs @@ -9,14 +9,17 @@ fn json_to_record(v: &Value) -> HashMap { } } -fn run_sql(sql: &str, record: &serde_json::Value) -> HashMap { +fn run_sql(sql: &str, record: &serde_json::Value) -> indexmap::IndexMap { let mut parser = Parser::new(sql); let stmt = parser.parse_select().expect("failed to parse SQL"); let rec = json_to_record(record); Evaluator::eval_select(&stmt, &rec).expect("failed to eval SQL") } -fn run_sql_records(sql: &str, records: &[serde_json::Value]) -> Vec> { +fn run_sql_records( + sql: &str, + records: &[serde_json::Value], +) -> Vec> { let mut parser = Parser::new(sql); let stmt = parser.parse_select().expect("failed to parse SQL"); let state = RuleState::default(); diff --git a/crates/rekuiper-sql/tests/test_issue45_order.rs b/crates/rekuiper-sql/tests/test_issue45_order.rs new file mode 100644 index 00000000..2ff3c8e5 --- /dev/null +++ b/crates/rekuiper-sql/tests/test_issue45_order.rs @@ -0,0 +1,68 @@ +use indexmap::IndexMap; +use rekuiper_sql::{Evaluator, Parser}; +use serde_json::json; + +#[test] +fn test_select_field_order_preserved() { + let sql = "SELECT z, a, m FROM demo"; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("failed to parse SQL"); + + let mut record = IndexMap::new(); + record.insert("a".to_string(), json!(1)); + record.insert("m".to_string(), json!(2)); + record.insert("z".to_string(), json!(3)); + + let out = Evaluator::eval_select(&stmt, &record).expect("eval failed"); + let keys: Vec<&String> = out.keys().collect(); + assert_eq!( + keys, + vec!["z", "a", "m"], + "Output keys must follow SELECT order" + ); + + let json_str = serde_json::to_string(&out).expect("serialize"); + assert_eq!(json_str, r#"{"z":3,"a":1,"m":2}"#); +} + +#[test] +fn test_select_wildcard_preserves_input_order() { + let sql = "SELECT * FROM demo"; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("failed to parse SQL"); + + let mut record = IndexMap::new(); + record.insert("field_c".to_string(), json!(30)); + record.insert("field_a".to_string(), json!(10)); + record.insert("field_b".to_string(), json!(20)); + + let out = Evaluator::eval_select(&stmt, &record).expect("eval failed"); + let keys: Vec<&String> = out.keys().collect(); + assert_eq!( + keys, + vec!["field_c", "field_a", "field_b"], + "SELECT * must preserve input key order" + ); + + let json_str = serde_json::to_string(&out).expect("serialize"); + assert_eq!(json_str, r#"{"field_c":30,"field_a":10,"field_b":20}"#); +} + +#[test] +fn test_select_computed_and_alias_order() { + let sql = "SELECT x, a + 10 AS y, b FROM demo"; + let mut parser = Parser::new(sql); + let stmt = parser.parse_select().expect("failed to parse SQL"); + + let mut record = IndexMap::new(); + record.insert("a".to_string(), json!(5)); + record.insert("b".to_string(), json!("hello")); + record.insert("x".to_string(), json!(true)); + + let out = Evaluator::eval_select(&stmt, &record).expect("eval failed"); + let keys: Vec<&String> = out.keys().collect(); + assert_eq!(keys, vec!["x", "y", "b"]); + + let json_str = serde_json::to_string(&out).expect("serialize"); + assert_eq!(json_str, r#"{"x":true,"y":15,"b":"hello"}"#); +} diff --git a/crates/rekuiper-sql/tests/test_sql_functions.rs b/crates/rekuiper-sql/tests/test_sql_functions.rs index e02b9c8c..0bb2dc9d 100644 --- a/crates/rekuiper-sql/tests/test_sql_functions.rs +++ b/crates/rekuiper-sql/tests/test_sql_functions.rs @@ -21,7 +21,7 @@ fn eval_one(sql: &str, record: &Record) -> Value { let stmt = parser.parse_select().expect("Should parse"); assert_eq!(stmt.fields.len(), 1); match Evaluator::eval_select(&stmt, record) { - Some(mut row) => row.remove("v").unwrap_or(Value::Null), + Some(mut row) => row.shift_remove("v").unwrap_or(Value::Null), None => Value::Null, } } @@ -33,7 +33,7 @@ fn eval_agg_one(sql: &str, records: &[Record]) -> Value { assert_eq!(stmt.fields.len(), 1); Evaluator::eval_aggregate(&stmt, records) .expect("Should aggregate") - .remove("v") + .shift_remove("v") .unwrap() } @@ -756,7 +756,7 @@ fn test_stateful_had_changed_and_changed_col() { let row = |t: Value| { Evaluator::eval_select_stateful(&stmt, &rec(&[("temp", t)]), &state) .expect("row always projects") - .remove("ch") + .shift_remove("ch") .unwrap() }; assert_eq!(row(json!(10)), json!(true)); @@ -769,7 +769,7 @@ fn test_stateful_had_changed_and_changed_col() { let row = |t: Value| { Evaluator::eval_select_stateful(&stmt, &rec(&[("temp", t)]), &state) .expect("row always projects") - .remove("c") + .shift_remove("c") .unwrap() }; assert_eq!(row(json!(10)), json!(10)); @@ -2596,7 +2596,7 @@ fn test_acc_distinct_collect_and_alias() { r.insert("val".to_string(), v); Evaluator::eval_select_stateful(&stmt, &r, &state) .expect("eval") - .remove("v") + .shift_remove("v") .unwrap() }; @@ -2617,7 +2617,7 @@ fn test_acc_distinct_collect_and_alias() { r.insert("val".to_string(), v); Evaluator::eval_select_stateful(&stmt_alias, &r, &state_alias) .expect("eval") - .remove("v") + .shift_remove("v") .unwrap() }; @@ -2859,7 +2859,7 @@ fn test_issue26_lag_with_ignore_null() { let stmt = parser.parse_select().unwrap(); Evaluator::eval_select_stateful(&stmt, &r, state) .unwrap() - .remove("v") + .shift_remove("v") .unwrap() }; diff --git a/crates/rekuiper-sql/tests/test_sql_parity.rs b/crates/rekuiper-sql/tests/test_sql_parity.rs index 191b925a..b33d2c43 100644 --- a/crates/rekuiper-sql/tests/test_sql_parity.rs +++ b/crates/rekuiper-sql/tests/test_sql_parity.rs @@ -126,7 +126,7 @@ fn test_complex_case_expression() { let grade = |score: i64| { Evaluator::eval_select(&stmt, &rec(&[("score", json!(score))])) .expect("row projects") - .remove("grade") + .shift_remove("grade") .unwrap() }; assert_eq!(grade(95), json!("EXCELLENT")); @@ -141,7 +141,7 @@ fn test_complex_case_expression() { let level = |s: &str| { Evaluator::eval_select(&stmt, &rec(&[("level", json!(s))])) .expect("row projects") - .remove("n") + .shift_remove("n") .unwrap() }; assert_eq!(level("high"), json!(3)); diff --git a/crates/rekuiper-sql/tests/validate_catalog_functions.rs b/crates/rekuiper-sql/tests/validate_catalog_functions.rs index b6b57d3e..4ad3ffac 100644 --- a/crates/rekuiper-sql/tests/validate_catalog_functions.rs +++ b/crates/rekuiper-sql/tests/validate_catalog_functions.rs @@ -11,7 +11,7 @@ fn eval_expr(sql_expr: &str, record: &Record) -> Value { .parse_select() .unwrap_or_else(|e| panic!("Failed to parse {}: {}", sql, e)); match Evaluator::eval_select(&stmt, record) { - Some(mut row) => row.remove("v").unwrap_or(Value::Null), + Some(mut row) => row.shift_remove("v").unwrap_or(Value::Null), None => Value::Null, } } diff --git a/crates/rekuiper-sql/tests/window_eval_test.rs b/crates/rekuiper-sql/tests/window_eval_test.rs index 6a45ce74..82c7727c 100644 --- a/crates/rekuiper-sql/tests/window_eval_test.rs +++ b/crates/rekuiper-sql/tests/window_eval_test.rs @@ -2,13 +2,13 @@ //! the row-free incremental evaluator, SESSIONWINDOW parsing and source //! metadata visibility. +use indexmap::IndexMap; use rekuiper_sql::{ Evaluator, IncrementalWindow, Parser, SlidingIncrementalWindow, TimeUnit, WindowDef, }; use serde_json::{json, Value}; -use std::collections::HashMap; -type Row = HashMap; +type Row = IndexMap; fn row(v: Value) -> Row { v.as_object().unwrap().clone().into_iter().collect() diff --git a/docs/en_US/mcp/overview.md b/docs/en_US/mcp/overview.md index aa70f8bc..49651fd0 100644 --- a/docs/en_US/mcp/overview.md +++ b/docs/en_US/mcp/overview.md @@ -86,7 +86,7 @@ These tools use the embedded `rekuiper-sql` parser and evaluation engine in memo | Tool | Purpose | | :--- | :--- | | `validate_sql` | Parses SQL abstract syntax trees, checks keywords, and validates streaming DDL syntax (including `BUFFER_FULL_POLICY`) offline. | -| `test_sql_expression` | Executes queries against mock JSON event payloads in memory, supporting vector search (`cosine_similarity`), array operations (`array_positions`), and stateful analytics (`acc_distinct_collect`, `lead`). | +| `test_sql_expression` | Executes queries against mock JSON event payloads in memory with exact SELECT field order preservation, supporting vector search (`cosine_similarity`), array operations (`array_positions`), and stateful analytics (`acc_distinct_collect`, `lead`). | | `explain_sql` | Deconstructs a SQL query into sources, projections, joins, window clauses, and filters. | | `explain_pipeline_profile` | Analyzes SQL queries offline. Reports execution tiers, memory boundaries, and execution bottlenecks. | diff --git a/docs/en_US/public/diagrams/deterministic_field_order.svg b/docs/en_US/public/diagrams/deterministic_field_order.svg new file mode 100644 index 00000000..f426aca6 --- /dev/null +++ b/docs/en_US/public/diagrams/deterministic_field_order.svg @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + Input Stream Record + {"id": 1, "temp": 25.0} + Stream Ingest Order + + + + + + + + SQL Evaluation + SELECT temp, id + Declared Column Order + + + + + + + + IndexMap Order State + [temp: 0, id: 1] + Deterministic Preservation + + + + + + + + Sink Output Payload + {"temp": 25.0, "id": 1} + Predictable Field Sequence + + + + Deterministic JSON Field Order: SELECT column sequence is preserved from evaluation through serialization. + diff --git a/docs/en_US/sqls/query_language_elements.md b/docs/en_US/sqls/query_language_elements.md index 8a4dc4a8..0287ef40 100644 --- a/docs/en_US/sqls/query_language_elements.md +++ b/docs/en_US/sqls/query_language_elements.md @@ -73,6 +73,29 @@ SELECT - `expression`: Constants, function invocations, column references, or operators. +### Field Order in Output Rows + +rekuiper keeps the exact order of columns specified in the `SELECT` clause. + +![Deterministic Field Order](/diagrams/deterministic_field_order.svg) + +- When you select explicit columns, output JSON fields match the sequence in the query: + + ```sql + SELECT device_id, temperature, humidity, status FROM sensor_stream; + -- Output JSON: {"device_id": "d1", "temperature": 22.5, "humidity": 65, "status": "ok"} + ``` + +- When you use `SELECT *`, output fields match the sequence defined in the input stream schema. +- When you use computed expressions and aliases, output fields appear in their declared sequence: + + ```sql + SELECT temp * 1.8 + 32 AS temp_f, id FROM perf_stream; + -- Output JSON: {"temp_f": 77.0, "id": "dev_0"} + ``` + +Downstream systems receive predictable JSON objects without random field reordering. + ## FROM Specifies the input source. Every `SELECT` statement requires a `FROM` clause. From 80248672811fd50ed0a7349aa8264d499ea08fa5 Mon Sep 17 00:00:00 2001 From: ankur-paan Date: Mon, 19 Oct 2026 09:16:37 +0100 Subject: [PATCH 2/2] test(sql): update test_object_json_parity to match insertion order for to_json Update test_object_json_parity to verify that to_json preserves the input sequence of keys rather than sorting alphabetically. --- crates/rekuiper-sql/tests/test_sql_functions.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/rekuiper-sql/tests/test_sql_functions.rs b/crates/rekuiper-sql/tests/test_sql_functions.rs index 0bb2dc9d..157f26f7 100644 --- a/crates/rekuiper-sql/tests/test_sql_functions.rs +++ b/crates/rekuiper-sql/tests/test_sql_functions.rs @@ -1705,7 +1705,7 @@ fn test_object_json_parity() { "SELECT to_json({\"msg\": \"hello\", \"code\": 200}) AS v FROM demo", &empty ), - json!("{\"code\":200,\"msg\":\"hello\"}") + json!("{\"msg\":\"hello\",\"code\":200}") ); assert_eq!( eval_one("SELECT tojson(7) AS v FROM demo", &empty),