Skip to content

fix(sql): preserve SELECT column order and deterministic JSON field output (#45) #46

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
ankur-paan merged 2 commits into from
Oct 19, 2026
Merged

fix(sql): preserve SELECT column order and deterministic JSON field output (#45) #46

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
784efec
fix(sql): preserve SELECT column order and deterministic JSON field o…
ankur-paan Oct 19, 2026
8024867
test(sql): update test_object_json_parity to match insertion order fo…
ankur-paan Oct 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 2 additions & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
1 change: 1 addition & 0 deletions crates/rekuiper-connectors/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand Down
13 changes: 9 additions & 4 deletions crates/rekuiper-connectors/src/file.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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};
Expand Down Expand Up @@ -478,12 +479,16 @@ impl FileSource {
}
// JSON / lines: objects emit directly, arrays fan out per element.
match serde_json::from_str::<Value>(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::<IndexMap<_, _>>(),
)]),
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::<IndexMap<_, _>>(),
));
} else {
tracing::warn!("Skipping non-object element in file source line");
}
Expand All @@ -509,8 +514,8 @@ pub fn parse_delimited_line(
line: &str,
delimiter: char,
headers: &[String],
) -> HashMap<String, Value> {
let mut map = HashMap::with_capacity(headers.len());
) -> IndexMap<String, Value> {
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));
Expand Down
9 changes: 7 additions & 2 deletions crates/rekuiper-connectors/src/http.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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};
Expand Down Expand Up @@ -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::<IndexMap<_, _>>(),
)]),
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::<IndexMap<_, _>>(),
));
} else {
tracing::warn!(
"Skipping non-object element in HTTP pull response from {}",
Expand Down
9 changes: 7 additions & 2 deletions crates/rekuiper-connectors/src/kafka.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -300,12 +301,16 @@ fn decode_kafka_payload(payload: &[u8]) -> Result<Vec<StreamRecord>> {
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::<IndexMap<_, _>>(),
)]),
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::<IndexMap<_, _>>(),
));
} else {
tracing::warn!("Skipping non-object element in Kafka message");
}
Expand Down
37 changes: 21 additions & 16 deletions crates/rekuiper-connectors/src/mqtt.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -667,13 +668,15 @@ pub fn decode_mqtt_payload(payload: &[u8]) -> Result<StreamRecord> {
// 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::<HashMap<String, Value>>(payload) {
match serde_json::from_slice::<IndexMap<String, Value>>(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::<IndexMap<_, _>>(),
)),
_ => Err(anyhow::anyhow!("MQTT payload must be a JSON object")),
}
}
Expand Down Expand Up @@ -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::<IndexMap<_, _>>(),
)),
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::<IndexMap<_, _>>(),
)),
_ => {
out.truncate(first);
bail!(
Expand All @@ -732,21 +735,21 @@ pub fn decode_payload_into_with_mask(
}
}
} else {
match serde_json::from_slice::<HashMap<String, Value>>(payload) {
match serde_json::from_slice::<IndexMap<String, Value>>(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::<IndexMap<_, _>>(),
)),
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::<IndexMap<_, _>>(),
)),
_ => {
out.truncate(first);
bail!(
Expand Down Expand Up @@ -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::<IndexMap<_, _>>(),
)),
_ => bail!("payload is not a valid '{}' protobuf message", message.name),
},
PayloadFormat::EdgeX => EdgeXCodec::decode(payload, meta.clone(), out)?,
Expand Down
9 changes: 7 additions & 2 deletions crates/rekuiper-connectors/src/redis.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -251,12 +252,16 @@ fn decode_redis_payload(payload: &[u8]) -> Result<Vec<StreamRecord>> {
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::<IndexMap<_, _>>(),
)]),
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::<IndexMap<_, _>>(),
));
} else {
tracing::warn!("Skipping non-object element in Redis message");
}
Expand Down
9 changes: 7 additions & 2 deletions crates/rekuiper-connectors/src/sql.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -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::<IndexMap<_, _>>(),
));
}
advance_index(&mut self.index, &out);
return Ok(out);
Expand All @@ -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::<IndexMap<_, _>>(),
));
}
advance_index(&mut self.index, &out);
return Ok(out);
Expand Down
9 changes: 7 additions & 2 deletions crates/rekuiper-connectors/src/websocket.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -71,12 +72,16 @@ impl WebSocketConfig {
pub fn decode_websocket_message(text: &str) -> Result<Vec<StreamRecord>> {
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::<IndexMap<_, _>>(),
)]),
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::<IndexMap<_, _>>(),
));
} else {
tracing::warn!("Skipping non-object element in WebSocket message");
}
Expand Down
5 changes: 3 additions & 2 deletions crates/rekuiper-connectors/src/zmq.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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};
Expand Down Expand Up @@ -176,7 +177,7 @@ pub fn decode_zmq_payload(payload: &[u8], topic: Option<&str>) -> Result<Vec<Str
let mut records = Vec::new();
match value {
Value::Object(map) => {
let mut rec = StreamRecord::new(map.into_iter().collect());
let mut rec = StreamRecord::new(map.into_iter().collect::<IndexMap<_, _>>());
if let Some(t) = topic {
let mut meta = std::collections::HashMap::new();
meta.insert("topic".to_string(), Value::String(t.to_string()));
Expand All @@ -190,7 +191,7 @@ pub fn decode_zmq_payload(payload: &[u8], topic: Option<&str>) -> Result<Vec<Str
Value::Array(items) => {
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::<IndexMap<_, _>>());
if let Some(t) = topic {
let mut meta = std::collections::HashMap::new();
meta.insert("topic".to_string(), Value::String(t.to_string()));
Expand Down
1 change: 1 addition & 0 deletions crates/rekuiper-core/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand Down
Loading
Loading