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..157f26f7 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)); @@ -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), @@ -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.