diff --git a/CHANGELOG.md b/CHANGELOG.md index 01737ca..6b15af7 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/point.rs b/src/point.rs index c6423df..c50678d 100644 --- a/src/point.rs +++ b/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, s: &str) { diff --git a/src/write_dataframe.rs b/src/write_dataframe.rs index f78aedc..7f2f70d 100644 --- a/src/write_dataframe.rs +++ b/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::, 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"); + } } diff --git a/tests/point_tests.rs b/tests/point_tests.rs index 4c2bd73..13e264a 100644 --- a/tests/point_tests.rs +++ b/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""#); +} diff --git a/tests/write_tests.rs b/tests/write_tests.rs index 688a8b2..c2e4009 100644 --- a/tests/write_tests.rs +++ b/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.