This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new fe5612964 perf(connectors): stop cloning JSON leaves when the caller
already owns them (#4399)
fe5612964 is described below
commit fe56129640f6fae2afd56f8bcca9f9edce256eaa
Author: Matthew Patton <[email protected]>
AuthorDate: Tue Oct 13 05:21:27 2026 -0400
perf(connectors): stop cloning JSON leaves when the caller already owns
them (#4399)
---
.claude/skills/connector-sdk/SKILL.md | 4 +-
core/connectors/sdk/src/convert.rs | 128 ++++++++++++++++--
core/connectors/sdk/src/encoders/avro.rs | 4 +-
core/connectors/sdk/src/lib.rs | 2 +-
core/connectors/sinks/delta_sink/src/sink.rs | 42 ++++--
core/connectors/sinks/http_sink/src/lib.rs | 58 +-------
core/connectors/sinks/meilisearch_sink/src/lib.rs | 28 +---
core/connectors/sinks/s3_sink/src/formatter.rs | 95 +++++++++----
core/connectors/sinks/s3_sink/src/sink.rs | 158 +++++++++++++++++++++-
core/connectors/sinks/surrealdb_sink/src/lib.rs | 15 +-
10 files changed, 389 insertions(+), 145 deletions(-)
diff --git a/.claude/skills/connector-sdk/SKILL.md
b/.claude/skills/connector-sdk/SKILL.md
index 440e9a1ec..36f7b3247 100644
--- a/.claude/skills/connector-sdk/SKILL.md
+++ b/.claude/skills/connector-sdk/SKILL.md
@@ -45,7 +45,7 @@ sdk/src/
├── sink.rs SinkContainer + sink_connector! macro (FFI plumbing).
├── source.rs SourceContainer + source_connector! macro (FFI
plumbing).
├── api.rs ConnectorStatus, ConnectorStats (feature = "api").
-├── convert.rs owned_value_to_serde_json (simd_json ⇄ serde_json
bridge).
+├── convert.rs owned_value_into_serde_json /
owned_value_to_serde_json (simd_json ⇄ serde_json bridge).
├── log.rs CallbackLayer for tracing across FFI.
├── retry.rs retry_async + RetryPolicy, CircuitBreaker,
HttpRetryMiddleware.
├── decoders/ One per schema: json, raw, text, proto, flatbuffer,
avro.
@@ -60,7 +60,7 @@ sdk/src/
2. **`Send + Sync`** on every public trait (FFI runs across thread pools).
3. **`#[repr(C)]`** on every type that crosses FFI (`Schema`, `TopicMetadata`,
`MessagesMetadata`, `RawMessage`, `ProducedMessages`, `ConsumedMessage`,
`DecodedMessage`, etc. in `lib.rs`). Adding fields requires bumping the SDK
version - existing plugins built against the old layout will misalign on
`postcard::from_bytes`.
4. **postcard** for FFI message serialization (handled by
`SinkContainer::consume`, `SourceContainer`'s `handle_messages`). **MessagePack
(`rmp_serde`)** for `ConnectorState`. **JSON** (`serde_json`) only for
human-editable config that crosses FFI. Don't mix.
-5. **`simd_json::OwnedValue`** for JSON payloads, not `serde_json::Value`. Use
`convert::owned_value_to_serde_json` as a bridge when interop is required.
+5. **`simd_json::OwnedValue`** for JSON payloads, not `serde_json::Value`. Use
`convert::owned_value_into_serde_json` as a bridge when you own the value (the
common case - it moves strings instead of copying them); fall back to the
borrowing `convert::owned_value_to_serde_json` only when you still need the
`OwnedValue` afterward.
6. **`BTreeMap`** for headers - deterministic ordering. Never `HashMap` on the
wire.
7. **No breaking changes** to
`Sink`/`Source`/`StreamDecoder`/`StreamEncoder`/`Transform` trait signatures
without coordinating with all in-tree plugins in the same PR.
8. **A `Schema` variant that changes meaning is as breaking as a new
variant.** The discriminant crosses FFI unchanged, so a plugin built against
the old SDK keeps decoding it the old way and misreads every batch with no
error anywhere. Same process as a trait change: bump the SDK minor version,
update every in-tree plugin in the same PR, and state in the PR description
that plugins built against the previous SDK must be rebuilt.
diff --git a/core/connectors/sdk/src/convert.rs
b/core/connectors/sdk/src/convert.rs
index 993e052b4..09896fb0c 100644
--- a/core/connectors/sdk/src/convert.rs
+++ b/core/connectors/sdk/src/convert.rs
@@ -23,20 +23,12 @@
use crate::Payload;
/// Convert `simd_json::OwnedValue` to `serde_json::Value` via direct
structural mapping.
+/// NaN/Infinity f64 values are mapped to `null`.
///
-/// NaN/Infinity f64 values are mapped to `null` since JSON has no
representation
-/// for these IEEE 754 special values.
+/// Prefer [`owned_value_into_serde_json`] when you own the value.
pub fn owned_value_to_serde_json(value: &simd_json::OwnedValue) ->
serde_json::Value {
match value {
- simd_json::OwnedValue::Static(s) => match s {
- simd_json::StaticNode::Null => serde_json::Value::Null,
- simd_json::StaticNode::Bool(b) => serde_json::Value::Bool(*b),
- simd_json::StaticNode::I64(n) =>
serde_json::Value::Number((*n).into()),
- simd_json::StaticNode::U64(n) =>
serde_json::Value::Number((*n).into()),
- simd_json::StaticNode::F64(n) => serde_json::Number::from_f64(*n)
- .map(serde_json::Value::Number)
- .unwrap_or(serde_json::Value::Null),
- },
+ simd_json::OwnedValue::Static(s) => static_node_to_serde_json(*s),
simd_json::OwnedValue::String(s) =>
serde_json::Value::String(s.to_string()),
simd_json::OwnedValue::Array(arr) => {
serde_json::Value::Array(arr.iter().map(owned_value_to_serde_json).collect())
@@ -51,6 +43,40 @@ pub fn owned_value_to_serde_json(value:
&simd_json::OwnedValue) -> serde_json::V
}
}
+/// Convert an owned `simd_json::OwnedValue` to `serde_json::Value` via direct
structural mapping.
+/// NaN/Infinity f64 values are mapped to `null`.
+///
+/// Strings and object keys move into the result instead of being copied.
+pub fn owned_value_into_serde_json(value: simd_json::OwnedValue) ->
serde_json::Value {
+ match value {
+ simd_json::OwnedValue::Static(s) => static_node_to_serde_json(s),
+ simd_json::OwnedValue::String(s) => serde_json::Value::String(s),
+ simd_json::OwnedValue::Array(arr) => {
+
serde_json::Value::Array(arr.into_iter().map(owned_value_into_serde_json).collect())
+ }
+ simd_json::OwnedValue::Object(obj) => {
+ let map: serde_json::Map<String, serde_json::Value> = obj
+ .into_iter()
+ .map(|(k, v)| (k, owned_value_into_serde_json(v)))
+ .collect();
+ serde_json::Value::Object(map)
+ }
+ }
+}
+
+/// Converts a `StaticNode` leaf. Unlike `simd_json::OwnedValue`'s own
`TryInto<serde_json::Value>`, NaN/Infinity map to `null` here instead of
erroring.
+fn static_node_to_serde_json(node: simd_json::StaticNode) -> serde_json::Value
{
+ match node {
+ simd_json::StaticNode::Null => serde_json::Value::Null,
+ simd_json::StaticNode::Bool(b) => serde_json::Value::Bool(b),
+ simd_json::StaticNode::I64(n) => serde_json::Value::Number(n.into()),
+ simd_json::StaticNode::U64(n) => serde_json::Value::Number(n.into()),
+ simd_json::StaticNode::F64(n) => serde_json::Number::from_f64(n)
+ .map(serde_json::Value::Number)
+ .unwrap_or(serde_json::Value::Null),
+ }
+}
+
/// Renames each key of a `Payload::Json` object via `rename`; other payloads
pass through. Collisions keep the later entry while the map is Vec-backed,
unspecified once hash-backed.
pub(crate) fn apply_field_mappings(
payload: Payload,
@@ -191,6 +217,38 @@ mod tests {
);
}
+ #[test]
+ fn test_into_matches_to_for_every_variant() {
+ let mut populated_obj = simd_json::owned::Object::new();
+ populated_obj.insert("k".into(), OwnedValue::String("v".into()));
+ let cases = [
+ OwnedValue::Static(StaticNode::Null),
+ OwnedValue::Static(StaticNode::Bool(true)),
+ OwnedValue::Static(StaticNode::Bool(false)),
+ OwnedValue::Static(StaticNode::I64(-42)),
+ OwnedValue::Static(StaticNode::U64(42)),
+ OwnedValue::Static(StaticNode::F64(3.54)),
+ OwnedValue::Static(StaticNode::F64(f64::NAN)),
+ OwnedValue::Static(StaticNode::F64(f64::INFINITY)),
+ OwnedValue::Static(StaticNode::F64(f64::NEG_INFINITY)),
+ OwnedValue::String("hello".into()),
+ OwnedValue::Array(Box::default()),
+ OwnedValue::Object(Box::new(simd_json::owned::Object::new())),
+ OwnedValue::Array(Box::new(vec![
+ OwnedValue::String("x".into()),
+ OwnedValue::Static(StaticNode::I64(1)),
+ ])),
+ OwnedValue::Object(Box::new(populated_obj)),
+ ];
+ for case in cases {
+ assert_eq!(
+ owned_value_into_serde_json(case.clone()),
+ owned_value_to_serde_json(&case),
+ "owned_value_into_serde_json must match
owned_value_to_serde_json for {case:?}"
+ );
+ }
+ }
+
#[test]
fn apply_field_mappings_should_rename_mapped_key_and_keep_unmapped_key() {
let mappings = HashMap::from([("old_field".to_string(),
"new_field".to_string())]);
@@ -215,6 +273,54 @@ mod tests {
}
}
+ #[test]
+ fn test_into_moves_string_leaf_and_object_key_without_reallocating() {
+ let leaf = String::from("leaf-value-not-interned");
+ let leaf_ptr = leaf.as_ptr();
+ let key = String::from("object-key-not-interned");
+ let key_ptr = key.as_ptr();
+
+ let mut obj = simd_json::owned::Object::new();
+ obj.insert(key, OwnedValue::String(leaf));
+ let input = OwnedValue::Object(Box::new(obj));
+
+ let serde_json::Value::Object(map) =
owned_value_into_serde_json(input) else {
+ panic!("expected an object");
+ };
+ let (result_key, result_value) = map.into_iter().next().expect("one
entry");
+ let serde_json::Value::String(result_leaf) = result_value else {
+ panic!("expected a string value");
+ };
+
+ assert_eq!(
+ result_key.as_ptr(),
+ key_ptr,
+ "object key must move into the result, not reallocate"
+ );
+ assert_eq!(
+ result_leaf.as_ptr(),
+ leaf_ptr,
+ "string leaf must move into the result, not reallocate"
+ );
+ }
+
+ #[test]
+ fn test_into_nested_object_converts_recursively() {
+ let mut inner = simd_json::owned::Object::new();
+ inner.insert("b".into(), OwnedValue::String("v".into()));
+ inner.insert(
+ "arr".into(),
+
OwnedValue::Array(Box::new(vec![OwnedValue::Static(StaticNode::I64(-1))])),
+ );
+ let mut outer = simd_json::owned::Object::new();
+ outer.insert("a".into(), OwnedValue::Object(Box::new(inner)));
+ let input = OwnedValue::Object(Box::new(outer));
+ assert_eq!(
+ owned_value_into_serde_json(input),
+ serde_json::json!({"a": {"b": "v", "arr": [-1]}})
+ );
+ }
+
#[test]
fn
apply_field_mappings_should_swap_two_fields_without_losing_either_value() {
let mappings = HashMap::from([
diff --git a/core/connectors/sdk/src/encoders/avro.rs
b/core/connectors/sdk/src/encoders/avro.rs
index 588207fc9..67bfdd00e 100644
--- a/core/connectors/sdk/src/encoders/avro.rs
+++ b/core/connectors/sdk/src/encoders/avro.rs
@@ -17,7 +17,7 @@
use crate::{
Error, Payload, Schema, StreamEncoder,
- convert::{apply_field_mappings, owned_value_to_serde_json},
+ convert::{apply_field_mappings, owned_value_into_serde_json},
};
use apache_avro::Schema as AvroSchema;
use apache_avro::writer::datum::GenericDatumWriter;
@@ -135,7 +135,7 @@ impl AvroStreamEncoder {
Error::InvalidConfigValue("Avro schema is required for
encoding".to_string())
})?;
- let serde_value = owned_value_to_serde_json(&json_value);
+ let serde_value = owned_value_into_serde_json(json_value);
let avro_value = Self::serde_json_to_avro_value(serde_value, schema)?;
GenericDatumWriter::builder(schema)
diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs
index cde3797db..a5aba1c1d 100644
--- a/core/connectors/sdk/src/lib.rs
+++ b/core/connectors/sdk/src/lib.rs
@@ -46,7 +46,7 @@ pub mod sink;
pub mod source;
pub mod transforms;
-pub use convert::owned_value_to_serde_json;
+pub use convert::{owned_value_into_serde_json, owned_value_to_serde_json};
pub use log::LogCallback;
pub use transforms::Transform;
diff --git a/core/connectors/sinks/delta_sink/src/sink.rs
b/core/connectors/sinks/delta_sink/src/sink.rs
index 36ffbe358..9afff9913 100644
--- a/core/connectors/sinks/delta_sink/src/sink.rs
+++ b/core/connectors/sinks/delta_sink/src/sink.rs
@@ -22,7 +22,8 @@ use crate::storage::build_storage_options;
use async_trait::async_trait;
use deltalake::writer::{DeltaWriter, JsonWriter};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata,
owned_value_to_serde_json,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
+ owned_value_into_serde_json,
};
use tracing::{debug, error, info};
@@ -131,7 +132,7 @@ impl Sink for DeltaSink {
messages_metadata.current_offset,
);
- let mut json_values = json_values(&messages)?;
+ let mut json_values = json_values(messages)?;
if json_values.is_empty() {
debug!("No JSON values to write");
@@ -199,17 +200,18 @@ impl Sink for DeltaSink {
/// payload that has none. Proto text holding JSON is the descriptor-less
/// `proto_convert` fallback and is written as the document it holds; proto
/// text that is not JSON fails the batch like any other non-JSON payload.
-fn json_values(messages: &[ConsumedMessage]) -> Result<Vec<serde_json::Value>,
Error> {
+fn json_values(messages: Vec<ConsumedMessage>) ->
Result<Vec<serde_json::Value>, Error> {
let mut json_values = Vec::with_capacity(messages.len());
for message in messages {
- let Some(document) = message.payload.json_document() else {
- error!(
- "Unsupported payload type: {}. Delta sink only supports JSON
payloads.",
- message.payload
- );
- return Err(Error::InvalidPayloadType);
- };
- json_values.push(owned_value_to_serde_json(document.as_ref()));
+ match message.payload.into_json_document() {
+ Payload::Json(document) =>
json_values.push(owned_value_into_serde_json(document)),
+ payload => {
+ error!(
+ "Unsupported payload type: {payload}. Delta sink only
supports JSON payloads."
+ );
+ return Err(Error::InvalidPayloadType);
+ }
+ }
}
Ok(json_values)
}
@@ -238,7 +240,7 @@ mod tests {
message(Payload::Proto(r#"{"id": 2}"#.to_owned())),
];
- let values = json_values(&messages).expect("proto text holding JSON is
a document");
+ let values = json_values(messages).expect("proto text holding JSON is
a document");
assert_eq!(
values,
@@ -251,7 +253,21 @@ mod tests {
let messages = vec![message(Payload::Proto("id: 2".to_owned()))];
assert_eq!(
- json_values(&messages).expect_err("proto text that is not JSON has
no document"),
+ json_values(messages).expect_err("proto text that is not JSON has
no document"),
+ Error::InvalidPayloadType
+ );
+ }
+
+ #[test]
+ fn
json_values_rejects_a_batch_with_one_unsupported_payload_among_valid_ones() {
+ let messages = vec![
+ message(Payload::Json(simd_json::json!({"id": 1}))),
+ message(Payload::Raw(b"not json".to_vec())),
+ message(Payload::Json(simd_json::json!({"id": 3}))),
+ ];
+
+ assert_eq!(
+ json_values(messages).expect_err("one unsupported payload must
fail the whole batch"),
Error::InvalidPayloadType
);
}
diff --git a/core/connectors/sinks/http_sink/src/lib.rs
b/core/connectors/sinks/http_sink/src/lib.rs
index 7a1c226db..371daae7f 100644
--- a/core/connectors/sinks/http_sink/src/lib.rs
+++ b/core/connectors/sinks/http_sink/src/lib.rs
@@ -22,7 +22,7 @@ use bytes::Bytes;
use humantime::Duration as HumanDuration;
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
- convert::owned_value_to_serde_json, sink_connector,
+ convert::owned_value_into_serde_json, sink_connector,
};
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
use reqwest_retry::{
@@ -380,11 +380,7 @@ impl HttpSink {
fn payload_to_json(&self, payload: Payload) -> Result<serde_json::Value,
Error> {
let payload = payload.into_json_document();
match payload {
- Payload::Json(value) => {
- // Direct structural conversion (not serialization roundtrip).
- // Follows the Elasticsearch sink pattern. NaN/Infinity f64 →
null.
- Ok(owned_value_to_serde_json(&value))
- }
+ Payload::Json(value) => Ok(owned_value_into_serde_json(value)),
Payload::Text(text) | Payload::Proto(text) =>
Ok(serde_json::Value::String(text)),
Payload::Raw(bytes) | Payload::FlatBuffer(bytes) => {
let encoded = EncodedPayload {
@@ -1706,56 +1702,6 @@ mod tests {
);
}
- #[test]
- fn given_null_value_should_convert_to_null() {
- let v = simd_json::OwnedValue::Static(simd_json::StaticNode::Null);
- assert_eq!(owned_value_to_serde_json(&v), serde_json::Value::Null);
- }
-
- #[test]
- fn given_bool_value_should_convert_correctly() {
- let v =
simd_json::OwnedValue::Static(simd_json::StaticNode::Bool(true));
- assert_eq!(owned_value_to_serde_json(&v),
serde_json::Value::Bool(true));
- }
-
- #[test]
- fn given_integer_values_should_convert_correctly() {
- let i64_val =
simd_json::OwnedValue::Static(simd_json::StaticNode::I64(-42));
- assert_eq!(owned_value_to_serde_json(&i64_val),
serde_json::json!(-42));
-
- let u64_val =
simd_json::OwnedValue::Static(simd_json::StaticNode::U64(42));
- assert_eq!(owned_value_to_serde_json(&u64_val), serde_json::json!(42));
- }
-
- #[test]
- fn given_f64_value_should_convert_correctly() {
- let v =
simd_json::OwnedValue::Static(simd_json::StaticNode::F64(3.54));
- let result = owned_value_to_serde_json(&v);
- assert_eq!(result.as_f64().unwrap(), 3.54);
- }
-
- #[test]
- fn given_nan_f64_should_convert_to_null() {
- let v =
simd_json::OwnedValue::Static(simd_json::StaticNode::F64(f64::NAN));
- assert_eq!(owned_value_to_serde_json(&v), serde_json::Value::Null);
- }
-
- #[test]
- fn given_infinity_f64_should_convert_to_null() {
- let v =
simd_json::OwnedValue::Static(simd_json::StaticNode::F64(f64::INFINITY));
- assert_eq!(owned_value_to_serde_json(&v), serde_json::Value::Null);
- }
-
- #[test]
- fn given_nested_object_should_convert_recursively() {
- let v = simd_json_from_str(r#"{"nested":{"key":"val"},"arr":[1,2]}"#);
-
- let result = owned_value_to_serde_json(&v);
- assert_eq!(result["nested"]["key"], "val");
- assert_eq!(result["arr"][0], 1);
- assert_eq!(result["arr"][1], 2);
- }
-
#[test]
fn given_minimal_toml_config_should_deserialize() {
let toml_str = r#"url = "https://example.com""#;
diff --git a/core/connectors/sinks/meilisearch_sink/src/lib.rs
b/core/connectors/sinks/meilisearch_sink/src/lib.rs
index 3f56cea57..b6fb1c7e1 100644
--- a/core/connectors/sinks/meilisearch_sink/src/lib.rs
+++ b/core/connectors/sinks/meilisearch_sink/src/lib.rs
@@ -20,6 +20,7 @@ use base64::{Engine as _, engine::general_purpose};
use iggy_common::IggyTimestamp;
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
+ convert::owned_value_into_serde_json,
retry::{parse_duration, retry_backoff},
sink_connector,
};
@@ -936,33 +937,6 @@ fn upsert_metadata_field(object: &mut Map<String, Value>,
field: &str, value: Va
}
}
-fn owned_value_into_serde_json(value: simd_json::OwnedValue) -> Value {
- match value {
- simd_json::OwnedValue::Static(node) => match node {
- simd_json::StaticNode::Null => Value::Null,
- simd_json::StaticNode::Bool(value) => Value::Bool(value),
- simd_json::StaticNode::I64(value) => Value::Number(value.into()),
- simd_json::StaticNode::U64(value) => Value::Number(value.into()),
- simd_json::StaticNode::F64(value) =>
serde_json::Number::from_f64(value)
- .map(Value::Number)
- .unwrap_or(Value::Null),
- },
- simd_json::OwnedValue::String(value) => Value::String(value),
- simd_json::OwnedValue::Array(values) => Value::Array(
- values
- .into_iter()
- .map(owned_value_into_serde_json)
- .collect(),
- ),
- simd_json::OwnedValue::Object(values) => Value::Object(
- values
- .into_iter()
- .map(|(key, value)| (key, owned_value_into_serde_json(value)))
- .collect(),
- ),
- }
-}
-
fn sanitize_url_for_log(raw: &str) -> String {
let normalized = normalize_host(raw).unwrap_or_else(|_|
raw.trim().to_string());
let Ok(mut url) = Url::parse(&normalized) else {
diff --git a/core/connectors/sinks/s3_sink/src/formatter.rs
b/core/connectors/sinks/s3_sink/src/formatter.rs
index a70ecab46..5ca3cbaf2 100644
--- a/core/connectors/sinks/s3_sink/src/formatter.rs
+++ b/core/connectors/sinks/s3_sink/src/formatter.rs
@@ -19,7 +19,7 @@ use crate::OutputFormat;
use crate::buffer::FileBuffer;
use chrono::{DateTime, Utc};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata,
owned_value_to_serde_json,
+ ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata,
owned_value_into_serde_json,
};
use serde::Serialize;
use serde_json::Value;
@@ -42,7 +42,7 @@ struct JsonMessage<'a> {
}
pub(crate) fn format_message(
- message: &ConsumedMessage,
+ message: ConsumedMessage,
topic_metadata: &TopicMetadata,
messages_metadata: &MessagesMetadata,
include_metadata: bool,
@@ -62,7 +62,7 @@ pub(crate) fn format_message(
}
fn format_json_message(
- message: &ConsumedMessage,
+ message: ConsumedMessage,
topic_metadata: &TopicMetadata,
messages_metadata: &MessagesMetadata,
include_metadata: bool,
@@ -100,7 +100,7 @@ fn format_json_message(
} else {
None
},
- payload: payload_to_json_value(&message.payload),
+ payload: payload_to_json_value(message.payload),
};
serde_json::to_vec(&msg).map_err(|e| {
@@ -111,8 +111,8 @@ fn format_json_message(
})
}
-fn format_raw_message(message: &ConsumedMessage) -> Result<Vec<u8>, Error> {
- message.payload.try_to_bytes().map_err(|e| {
+fn format_raw_message(message: ConsumedMessage) -> Result<Vec<u8>, Error> {
+ message.payload.try_into_vec().map_err(|e| {
Error::CannotStoreData(format!(
"Failed to extract raw bytes at offset {}: {e}",
message.offset
@@ -165,23 +165,17 @@ fn serialize_headers(
Value::Object(obj)
}
-fn payload_to_json_value(payload: &Payload) -> Value {
- match payload {
- Payload::Json(value) => owned_value_to_serde_json(value),
- Payload::Text(text) => Value::String(text.clone()),
- Payload::Raw(bytes) => match serde_json::from_slice(bytes) {
+// Proto text holding JSON is written as the document it holds, like a native
`json` batch.
+fn payload_to_json_value(payload: Payload) -> Value {
+ match payload.into_json_document() {
+ Payload::Json(value) => owned_value_into_serde_json(value),
+ Payload::Text(text) | Payload::Proto(text) => Value::String(text),
+ Payload::Raw(bytes) => match serde_json::from_slice(&bytes) {
Ok(v) => v,
- Err(_) => Value::String(base64_encode(bytes)),
+ Err(_) => Value::String(base64_encode(&bytes)),
},
- // Proto text holding JSON is the descriptor-less `proto_convert`
- // fallback and is written as the document it holds, the way the same
- // bytes were written when the batch was tagged `json`.
- Payload::Proto(text) => match payload.json_document() {
- Some(document) => owned_value_to_serde_json(document.as_ref()),
- None => Value::String(text.clone()),
- },
- Payload::FlatBuffer(bytes) => Value::String(base64_encode(bytes)),
- Payload::Avro(bytes) => Value::String(base64_encode(bytes)),
+ Payload::FlatBuffer(bytes) => Value::String(base64_encode(&bytes)),
+ Payload::Avro(bytes) => Value::String(base64_encode(&bytes)),
}
}
@@ -276,7 +270,7 @@ mod tests {
let meta = make_messages_metadata();
let bytes =
- format_message(&msg, &topic, &meta, true, false,
OutputFormat::JsonLines).unwrap();
+ format_message(msg, &topic, &meta, true, false,
OutputFormat::JsonLines).unwrap();
let value: Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(value["offset"], 42);
@@ -295,7 +289,7 @@ mod tests {
let meta = make_messages_metadata();
let bytes =
- format_message(&msg, &topic, &meta, false, false,
OutputFormat::JsonLines).unwrap();
+ format_message(msg, &topic, &meta, false, false,
OutputFormat::JsonLines).unwrap();
let value: Value = serde_json::from_slice(&bytes).unwrap();
assert!(value.get("offset").is_none());
@@ -318,7 +312,7 @@ mod tests {
let meta = make_messages_metadata();
let bytes =
- format_message(&msg, &topic, &meta, false, true,
OutputFormat::JsonLines).unwrap();
+ format_message(msg, &topic, &meta, false, true,
OutputFormat::JsonLines).unwrap();
let value: Value = serde_json::from_slice(&bytes).unwrap();
assert!(value["headers"].is_object());
@@ -330,21 +324,68 @@ mod tests {
let payload = Payload::Proto(r#"{"id":1,"name":"row-1"}"#.to_string());
assert_eq!(
- payload_to_json_value(&payload),
+ payload_to_json_value(payload),
serde_json::json!({"id": 1, "name": "row-1"})
);
}
+ #[test]
+ fn text_payload_holding_json_looking_content_stays_a_string() {
+ let payload = Payload::Text(r#"{"id":1}"#.to_string());
+
+ assert_eq!(
+ payload_to_json_value(payload),
+ Value::String(r#"{"id":1}"#.to_string())
+ );
+ }
+
#[test]
fn proto_text_that_is_not_json_is_written_as_a_string() {
let payload = Payload::Proto("name: \"row-1\"".to_string());
assert_eq!(
- payload_to_json_value(&payload),
+ payload_to_json_value(payload),
Value::String("name: \"row-1\"".to_string())
);
}
+ #[test]
+ fn raw_payload_holding_json_bytes_is_parsed() {
+ let payload = Payload::Raw(br#"{"id":1}"#.to_vec());
+
+ assert_eq!(payload_to_json_value(payload), serde_json::json!({"id":
1}));
+ }
+
+ #[test]
+ fn raw_payload_that_is_not_json_is_base64_encoded() {
+ let payload = Payload::Raw(vec![0xff, 0x00, 0x10]);
+
+ assert_eq!(
+ payload_to_json_value(payload),
+ Value::String("/wAQ".to_string())
+ );
+ }
+
+ #[test]
+ fn flatbuffer_payload_is_base64_encoded() {
+ let payload = Payload::FlatBuffer(vec![1, 2, 3]);
+
+ assert_eq!(
+ payload_to_json_value(payload),
+ Value::String("AQID".to_string())
+ );
+ }
+
+ #[test]
+ fn avro_payload_is_base64_encoded() {
+ let payload = Payload::Avro(vec![4, 5, 6]);
+
+ assert_eq!(
+ payload_to_json_value(payload),
+ Value::String("BAUG".to_string())
+ );
+ }
+
#[test]
fn raw_format() {
let payload = Payload::Text("hello world".to_string());
@@ -352,7 +393,7 @@ mod tests {
let topic = make_topic_metadata();
let meta = make_messages_metadata();
- let bytes = format_message(&msg, &topic, &meta, true, false,
OutputFormat::Raw).unwrap();
+ let bytes = format_message(msg, &topic, &meta, true, false,
OutputFormat::Raw).unwrap();
assert_eq!(bytes, b"hello world");
}
diff --git a/core/connectors/sinks/s3_sink/src/sink.rs
b/core/connectors/sinks/s3_sink/src/sink.rs
index fcf155f60..09fcaf399 100644
--- a/core/connectors/sinks/s3_sink/src/sink.rs
+++ b/core/connectors/sinks/s3_sink/src/sink.rs
@@ -100,7 +100,7 @@ impl Sink for S3Sink {
&buffer_arc,
topic_metadata,
&messages_metadata,
- &messages,
+ messages,
&mut processed,
)
.await;
@@ -250,11 +250,12 @@ impl S3Sink {
buffer_arc: &Arc<tokio::sync::Mutex<FileBuffer>>,
topic_metadata: &TopicMetadata,
messages_metadata: &MessagesMetadata,
- messages: &[ConsumedMessage],
+ messages: Vec<ConsumedMessage>,
processed: &mut u64,
) -> Result<(), Error> {
for message in messages {
let resolved = self.resolved();
+ let (offset, timestamp) = (message.offset, message.timestamp);
let formatted = formatter::format_message(
message,
topic_metadata,
@@ -266,7 +267,7 @@ impl S3Sink {
let flush_payload = {
let mut buffer = buffer_arc.lock().await;
- buffer.append(&formatted, message.offset, message.timestamp);
+ buffer.append(&formatted, offset, timestamp);
if buffer.should_rotate(
self.config.file_rotation,
@@ -619,6 +620,157 @@ mod tests {
});
}
+ // FileBuffer only ever records the first message's timestamp, so that's
all this can observe.
+ #[test]
+ fn given_multiple_messages_should_write_in_offset_order_with_batch_key() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test
runtime");
+ runtime.block_on(async {
+ let server = MockServer::start().await;
+ mock_write_probe(&server, "allowed/events/.iggy-sink-probe").await;
+ Mock::given(method("PUT"))
+ .respond_with(ResponseTemplate::new(200))
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ prefix: Some("/allowed/events/".to_string()),
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ path_template: "{topic}/{timestamp}".to_string(),
+ output_format: "raw".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ sink.open().await.expect("Sink should initialize");
+
+ let message = |offset, timestamp, data: &[u8]| ConsumedMessage {
+ id: u128::from(offset) + 1,
+ offset,
+ checksum: 0,
+ timestamp,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Raw(data.to_vec()),
+ };
+ sink.consume(
+ &TopicMetadata {
+ stream: "events".to_string(),
+ topic: "messages".to_string(),
+ },
+ MessagesMetadata {
+ partition_id: 0,
+ current_offset: 10,
+ schema: Schema::Raw,
+ },
+ vec![
+ message(10, 1_000_000, b"a"),
+ message(11, 2_000_000, b"b"),
+ message(12, 3_000_000, b"c"),
+ ],
+ )
+ .await
+ .expect("Batch should upload");
+ sink.close().await.expect("Sink should close");
+
+ let requests = server.received_requests().await.expect("record
requests");
+ assert_eq!(
+ requests.len(),
+ 3,
+ "Probe POST + DELETE, then one data PUT: {requests:?}"
+ );
+ assert_eq!(requests[2].method, "PUT");
+ assert_eq!(
+ requests[2].body, b"abc",
+ "Each message's own bytes must land in its own offset's
position"
+ );
+ assert_eq!(
+ requests[2].url.path(),
+
"/test-bucket/allowed/events/messages/1000/00000-00000000000000000010-00000000000000000012.bin",
+ "Key must reflect the batch's offset range and the first
message's own timestamp, not a shared or stale one"
+ );
+ });
+ }
+
+ // Rotation exposes the middle offset and last timestamp a single flush
hides.
+ #[test]
+ fn
given_rotation_mid_batch_should_use_each_segments_own_offset_and_timestamp() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test
runtime");
+ runtime.block_on(async {
+ let server = MockServer::start().await;
+ mock_write_probe(&server, "allowed/events/.iggy-sink-probe").await;
+ Mock::given(method("PUT"))
+ .respond_with(ResponseTemplate::new(200))
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ prefix: Some("/allowed/events/".to_string()),
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ path_template: "{topic}/{timestamp}".to_string(),
+ output_format: "raw".to_string(),
+ file_rotation: FileRotation::Messages,
+ max_messages_per_file: Some(2),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ sink.open().await.expect("Sink should initialize");
+
+ let message = |offset, timestamp, data: &[u8]| ConsumedMessage {
+ id: u128::from(offset) + 1,
+ offset,
+ checksum: 0,
+ timestamp,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Raw(data.to_vec()),
+ };
+ sink.consume(
+ &TopicMetadata {
+ stream: "events".to_string(),
+ topic: "messages".to_string(),
+ },
+ MessagesMetadata {
+ partition_id: 0,
+ current_offset: 10,
+ schema: Schema::Raw,
+ },
+ vec![
+ message(10, 1_000_000, b"a"),
+ message(11, 2_000_000, b"b"),
+ message(12, 3_000_000, b"c"),
+ ],
+ )
+ .await
+ .expect("Batch should upload");
+ sink.close().await.expect("Sink should close");
+
+ let requests = server.received_requests().await.expect("record
requests");
+ assert_eq!(
+ requests.len(),
+ 4,
+ "Probe POST + DELETE, then one PUT per rotated segment:
{requests:?}"
+ );
+ assert_eq!(requests[2].method, "PUT");
+ assert_eq!(
+ requests[2].body, b"ab",
+ "First segment holds messages 1-2"
+ );
+ assert_eq!(
+ requests[2].url.path(),
+
"/test-bucket/allowed/events/messages/1000/00000-00000000000000000010-00000000000000000011.bin",
+ "First segment's key must reflect message 2's own offset as
last_offset"
+ );
+ assert_eq!(requests[3].method, "PUT");
+ assert_eq!(requests[3].body, b"c", "Second segment holds message 3
alone");
+ assert_eq!(
+ requests[3].url.path(),
+
"/test-bucket/allowed/events/messages/3000/00000-00000000000000000012-00000000000000000012.bin",
+ "Second segment's key must reflect message 3's own timestamp,
not message 1's"
+ );
+ });
+ }
+
#[test]
fn
given_a_probe_prefix_when_opening_should_use_the_configured_write_scope() {
let runtime = tokio::runtime::Runtime::new().expect("Start test
runtime");
diff --git a/core/connectors/sinks/surrealdb_sink/src/lib.rs
b/core/connectors/sinks/surrealdb_sink/src/lib.rs
index 994d4373b..12755e68a 100644
--- a/core/connectors/sinks/surrealdb_sink/src/lib.rs
+++ b/core/connectors/sinks/surrealdb_sink/src/lib.rs
@@ -19,7 +19,7 @@ use async_trait::async_trait;
use base64::Engine;
use base64::engine::general_purpose;
use bytes::Bytes;
-use iggy_connector_sdk::convert::owned_value_to_serde_json;
+use iggy_connector_sdk::convert::owned_value_into_serde_json;
use iggy_connector_sdk::retry::{parse_duration, retry_backoff};
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
sink_connector,
@@ -896,7 +896,7 @@ fn build_auto_payload_document(payload: Payload) ->
Result<PayloadDocument, Erro
let payload = payload.into_json_document();
match payload {
Payload::Json(value) => Ok(PayloadDocument {
- value: owned_value_to_serde_json(&value),
+ value: owned_value_into_serde_json(value),
encoding: ENCODING_JSON,
}),
Payload::Text(text) | Payload::Proto(text) => Ok(PayloadDocument {
@@ -912,7 +912,7 @@ fn build_auto_payload_document(payload: Payload) ->
Result<PayloadDocument, Erro
fn build_json_payload_document(payload: Payload) -> Result<PayloadDocument,
Error> {
match payload {
Payload::Json(value) => Ok(PayloadDocument {
- value: owned_value_to_serde_json(&value),
+ value: owned_value_into_serde_json(value),
encoding: ENCODING_JSON,
}),
_ => {
@@ -1655,6 +1655,15 @@ mod tests {
assert_eq!(document.value, json!({"count": 3}));
}
+ #[test]
+ fn given_json_payload_format_should_convert_a_json_payload_directly() {
+ let payload = json_payload(json!({"count": 3}));
+ let document = build_json_payload_document(payload).expect("Failed to
build payload");
+
+ assert_eq!(document.encoding, ENCODING_JSON);
+ assert_eq!(document.value, json!({"count": 3}));
+ }
+
#[test]
fn given_json_payload_format_when_invalid_should_fail() {
let payload = Payload::Raw(b"not-json".to_vec());
