fix(connectors): simd_json OwnedValue serialization bug in Elasticsearch sink

[seokjin0414]

Bug Description

In core/connectors/sinks/elasticsearch_sink/src/lib.rs:175-176, OwnedValue::to_string() (Display trait) is used to convert simd_json::OwnedValue to a string before parsing it as serde_json::Value. However, simd-json's Display implementation for Object and Array types uses Rust's Debug formatter ({:?}), which produces invalid JSON.

Root Cause

simd-json source (simd-json/src/value/owned.rs):

impl fmt::Display for Value {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            Self::Static(s) => write!(f, "{s}"),
            Self::String(s) => write!(f, "{s}"),
            Self::Array(a) => write!(f, "{a:?}"),   // Debug format!
            Self::Object(o) => write!(f, "{o:?}"),   // Debug format!
        }
    }
}

Output Comparison

Method	Output	Valid JSON
doc.to_string() (Display)	{"id": Static(U64(1)), "name": String("user_0")}	No
simd_json::to_string(&doc)	{"id":1,"name":"user_0"}	Yes

Impact

• serde_json::from_str() silently fails → unwrap_or_else returns empty {}
• Elasticsearch indexes empty documents → data loss
• No error logs (silent failure)

Buggy Code

// lib.rs:175-176
let doc_json: serde_json::Value =
    serde_json::from_str(&doc.to_string()).unwrap_or_else(|_| serde_json::json!({}));

Fix

let doc_str = simd_json::to_string(&doc)
    .unwrap_or_else(|_| "{}".to_string());
let doc_json: serde_json::Value =
    serde_json::from_str(&doc_str).unwrap_or_else(|_| serde_json::json!({}));

Reproduction

use simd_json::json;

fn main() {
    let doc = json!({"id": 1u64, "name": "test_user"});

    println!("Display:          {}", doc.to_string());
    // {"id": Static(U64(1)), "name": String("test_user")}

    println!("simd_json::to_string: {}", simd_json::to_string(&doc).unwrap());
    // {"id":1,"name":"test_user"}

    let result: serde_json::Value = serde_json::from_str(&doc.to_string())
        .unwrap_or_else(|_| serde_json::json!({}));
    println!("Parsed result: {}", result);
    // {}
}

Affected File

• core/connectors/sinks/elasticsearch_sink/src/lib.rs (line 175-176, bulk_index_documents method)

[seokjin0414]

Discovered while working on #2592.

[spetz]

Thank you for reporting this, I actually keep on improving the connectors runtime and also adding new tests, so will look into this soon.

[seokjin0414]

Resolved in #2685 - the simd_json::OwnedValue serialization issue was fixed by introducing owned_value_to_serde_json() recursive conversion function instead of using doc.to_string().