fix(point): escape line protocol control characters

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## 0.4.0 [unreleased]
 
+### Bug Fixes
+
+1. [#59](https://github.com/InfluxCommunity/influxdb3-rust/pull/59): Escape
+   newline, carriage return, and tab characters as literal `\n`, `\r`, and
+   `\t` sequences in measurements, tags, field keys, and string field values
+   across Point and DataFrame writes.
+
 ## 0.3.0 [2026-08-27]
 
 > ⚠️ This release requires Rust 1.91 or later.

src/point.rs

@@ -355,45 +355,48 @@ fn escape_with(input: &str, needs_escape: fn(u8) -> bool) -> Cow<'_, str> {
     for ch in input.chars() {
         if ch.is_ascii() && needs_escape(ch as u8) {
             out.push('\\');
+            match ch {
+                '\n' => out.push('n'),
+                '\r' => out.push('r'),
+                '\t' => out.push('t'),
+                _ => out.push(ch),
+            }
+        } else {
+            out.push(ch);
         }
-        out.push(ch);
     }
     Cow::Owned(out)
 }
 
 fn measurement_needs_escape(b: u8) -> bool {
-    matches!(b, b',' | b' ')
+    matches!(b, b',' | b' ' | b'\n' | b'\r' | b'\t')
 }
 
 fn tag_needs_escape(b: u8) -> bool {
-    matches!(b, b',' | b'=' | b' ')
+    matches!(b, b',' | b'=' | b' ' | b'\n' | b'\r' | b'\t')
 }
 
-/// Escape a measurement name (commas and spaces). Shared with the DataFrame
-/// writer so both paths use the same rules.
+fn string_field_needs_escape(b: u8) -> bool {
+    matches!(b, b'\\' | b'"' | b'\n' | b'\r' | b'\t')
+}
+
+/// Escape a measurement name (commas, spaces, and control characters).
+/// Shared with the DataFrame writer so both paths use the same rules.
 pub(crate) fn escape_measurement(s: &str) -> Cow<'_, str> {
     escape_with(s, measurement_needs_escape)
 }
 
-/// Escape a tag key, tag value, or field key (commas, equals, spaces).
+/// Escape a tag key, tag value, or field key (commas, equals, spaces, and
+/// control characters).
 pub(crate) fn escape_tag(s: &str) -> Cow<'_, str> {
     escape_with(s, tag_needs_escape)
 }
 
-/// Escape the contents of a string field (backslash and double-quote). The
-/// caller is responsible for the surrounding quotes.
+/// Escape the contents of a string field (backslash, double-quote, newline,
+/// carriage return, and tab). The caller is responsible for the surrounding
+/// quotes.
 pub(crate) fn escape_string_field(s: &str) -> Cow<'_, str> {
-    if !s.bytes().any(|b| b == b'\\' || b == b'"') {
-        return Cow::Borrowed(s);
-    }
-    let mut out = String::with_capacity(s.len() + 8);
-    for ch in s.chars() {
-        if ch == '\\' || ch == '"' {
-            out.push('\\');
-        }
-        out.push(ch);
-    }
-    Cow::Owned(out)
+    escape_with(s, string_field_needs_escape)
 }
 
 fn write_escaped_measurement(buf: &mut Vec<u8>, s: &str) {

src/write_dataframe.rs

@@ -217,6 +217,8 @@ fn row_access_err(e: polars::error::PolarsError) -> Error {
 /// * Null field values omit that field for the row.
 /// * Rows where **all** fields are null are dropped entirely.
 /// * A null timestamp is omitted, so the server assigns the time.
+/// * Newline, carriage return, and tab characters in structured values are
+///   escaped to the literal sequences `\n`, `\r`, and `\t`.
 pub fn dataframe_to_line_protocol(
     df: &DataFrame,
     measurement: &str,
@@ -234,21 +236,28 @@ pub fn dataframe_to_line_protocol(
 
     // Resolve columns and escape their names once, before the row loop.
     // Missing tag columns are silently skipped (unchanged behaviour).
-    let mut tag_cols: Vec<(Cow<'_, str>, TagReader<'_>)> = tags
-        .iter()
-        .filter_map(|&t| df.column(t).ok().map(|c| (escape_tag(t), tag_reader(c))))
-        .collect();
+    let mut tag_cols: Vec<(Cow<'_, str>, TagReader<'_>)> = Vec::new();
+    for &t in tags {
+        if let Ok(c) = df.column(t) {
+            tag_cols.push((escape_tag(t), tag_reader(c)));
+        }
+    }
 
     // All columns that are not tag columns and not the timestamp column,
     // in frame order.
-    let mut field_cols: Vec<(Cow<'_, str>, FieldReader<'_>)> = (0..df.width())
-        .filter_map(|i| df.select_at_idx(i))
-        .filter(|c| {
+    let mut field_cols: Vec<(Cow<'_, str>, FieldReader<'_>)> = Vec::new();
+    for i in 0..df.width() {
+        let Some(c) = df.select_at_idx(i) else {
+            continue;
+        };
+        let is_field = {
             let name = c.name().as_str();
             !tag_set.contains(name) && Some(name) != timestamp_column
-        })
-        .map(|c| (escape_tag(c.name().as_str()), field_reader(c)))
-        .collect();
+        };
+        if is_field {
+            field_cols.push((escape_tag(c.name().as_str()), field_reader(c)));
+        }
+    }
 
     let mut ts_reader = timestamp_column
         .and_then(|t| df.column(t).ok())
@@ -678,4 +687,66 @@ mod tests {
             .unwrap();
         assert_eq!(lp, "m,host=a v=1.5 10\nm,host=b v=2.5 20");
     }
+
+    #[test]
+    fn dataframe_escapes_control_characters() {
+        let cases = [
+            (
+                "measurement",
+                df!["v" => [1_i64]].unwrap(),
+                "me\nas",
+                &[][..],
+                r#"me\nas v=1i"#,
+            ),
+            (
+                "tag key",
+                df!["tag\rkey" => ["value"], "v" => [1_i64]].unwrap(),
+                "m",
+                &["tag\rkey"][..],
+                r#"m,tag\rkey=value v=1i"#,
+            ),
+            (
+                "tag value",
+                df!["tag" => ["value\n"], "v" => [1_i64]].unwrap(),
+                "m",
+                &["tag"][..],
+                r#"m,tag=value\n v=1i"#,
+            ),
+            (
+                "field key",
+                df!["field\tkey" => [1_i64]].unwrap(),
+                "m",
+                &[][..],
+                r#"m field\tkey=1i"#,
+            ),
+            (
+                "string field value",
+                df!["v" => ["value\n"]].unwrap(),
+                "m",
+                &[][..],
+                r#"m v="value\n""#,
+            ),
+        ];
+
+        for (position, df, measurement, tags, expected) in cases {
+            let actual =
+                dataframe_to_line_protocol(&df, measurement, tags, None, Precision::Nanosecond)
+                    .unwrap();
+            assert_eq!(actual, expected, "{position} should be escaped");
+        }
+    }
+
+    #[test]
+    fn dataframe_drops_all_null_rows_with_control_characters_in_tags() {
+        let df = df![
+            "host" => ["invalid\ntag", "safe"],
+            "v" => [None::<i64>, Some(1_i64)],
+        ]
+        .unwrap();
+
+        let lp =
+            dataframe_to_line_protocol(&df, "m", &["host"], None, Precision::Nanosecond).unwrap();
+
+        assert_eq!(lp, "m,host=safe v=1i");
+    }
 }

tests/point_tests.rs

@@ -80,3 +80,50 @@ fn last_write_wins() {
     assert_eq!(lp.matches("v=").count(), 1);
     assert!(lp.contains("v=2i"));
 }
+
+#[test]
+fn line_protocol_escapes_control_characters() {
+    let cases = [
+        (
+            "measurement",
+            Point::new("me\nasurement").field("v", 1_i64),
+            r#"me\nasurement v=1i"#,
+        ),
+        (
+            "tag key",
+            Point::new("m").tag("tag\rkey", "value").field("v", 1_i64),
+            r#"m,tag\rkey=value v=1i"#,
+        ),
+        (
+            "tag value",
+            Point::new("m").tag("key", "value\t").field("v", 1_i64),
+            r#"m,key=value\t v=1i"#,
+        ),
+        (
+            "field key",
+            Point::new("m").field("field\nkey", 1_i64),
+            r#"m field\nkey=1i"#,
+        ),
+        (
+            "string field value",
+            Point::new("m").field("field", "value\r"),
+            r#"m field="value\r""#,
+        ),
+    ];
+
+    for (position, point, expected) in cases {
+        let actual = point.to_line_protocol(Precision::Nanosecond).unwrap();
+        assert_eq!(actual, expected, "{position} should be escaped");
+    }
+}
+
+#[test]
+fn line_protocol_preserves_literal_backslash_sequences() {
+    let lp = Point::new("m")
+        .tag("key", r#"literal\n"#)
+        .field("field", r#"literal\r\t"#)
+        .to_line_protocol(Precision::Nanosecond)
+        .unwrap();
+
+    assert_eq!(lp, r#"m,key=literal\n field="literal\\r\\t""#);
+}

tests/write_tests.rs

@@ -153,6 +153,52 @@ async fn default_tags_and_order_reach_the_wire() {
     m.assert_async().await;
 }
 
+#[tokio::test]
+async fn default_tags_escape_line_breaks_and_tabs() {
+    let mut server = Server::new_async().await;
+    let client = make_client(&server).await;
+
+    for (key, value, expected) in [
+        ("env\nkey", "prod", r#"m,env\nkey=prod v=1i"#),
+        ("env", "prod\rvalue\t", r#"m,env=prod\rvalue\t v=1i"#),
+    ] {
+        let mock = server
+            .mock("POST", "/api/v3/write_lp")
+            .match_query(Matcher::Any)
+            .match_body(expected)
+            .with_status(204)
+            .create_async()
+            .await;
+
+        client
+            .write(vec![Point::new("m").field("v", 1_i64)])
+            .default_tag(key, value)
+            .await
+            .unwrap();
+        mock.assert_async().await;
+    }
+}
+
+#[tokio::test]
+async fn point_tag_override_does_not_serialize_default_value() {
+    let mut server = Server::new_async().await;
+    let m = server
+        .mock("POST", "/api/v3/write_lp")
+        .match_query(Matcher::Any)
+        .match_body("m,env=safe v=1i")
+        .with_status(204)
+        .create_async()
+        .await;
+
+    let client = make_client(&server).await;
+    client
+        .write(vec![Point::new("m").tag("env", "safe").field("v", 1_i64)])
+        .default_tag("env", "invalid\nvalue")
+        .await
+        .unwrap();
+    m.assert_async().await;
+}
+
 #[tokio::test]
 async fn non_retryable_error_surfaces_once() {
     // A 404 is deterministic, so it surfaces immediately without retrying.