feat: Ability to set a marker and roll back a buffer. General Python testing improvements. (#17)

Author: amunra

Cargo.toml

@@ -42,3 +42,6 @@ ffi = ["insecure_skip_verify"]
 # Auto-generate the header. This is for dev-debugging-diffing only.
 # A hand-crafted header is easier on the eyes.
 gen_h = ["ffi", "cbindgen"]
+
+# Auto-generate a Cython .pxd file.
+gen_cython = ["ffi", "cbindgen"]

build.rs

@@ -191,6 +191,26 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         bindings.write_to_file("include/questdb/ilp/line_sender.h");
     }
 
+    #[cfg(feature = "gen_cython")]
+    {
+        let crate_dir = std::env::var("CARGO_MANIFEST_DIR")?;
+
+        let config = cbindgen::Config {
+            language: cbindgen::Language::Cython,
+            documentation: false,
+            cython: cbindgen::CythonConfig {
+                header: Some("questdb/ilp/line_sender.h".to_owned()),
+                cimports: std::collections::BTreeMap::new()},
+            ..Default::default()
+        };
+
+        let bindings = cbindgen::Builder::new()
+            .with_crate(crate_dir)
+            .with_config(config)
+            .generate()?;
+        bindings.write_to_file("cython/questdb/ilp/line_sender.pxd");
+    }
+
     json_tests::build()?;
 
     Ok(())

cpp_test/test_line_sender.cpp

@@ -607,4 +607,46 @@ TEST_CASE("Test timestamp column.")
 
     CHECK(server.recv() == 1);
     CHECK(server.msgs()[0] == exp);
+}
+
+TEST_CASE("Test Marker")
+{
+    questdb::ilp::line_sender_buffer buffer;
+    buffer.clear_marker();
+    buffer.clear_marker();
+
+    buffer.set_marker();
+    buffer.table("test");
+    CHECK(buffer.peek() == "test");
+    CHECK(buffer.size() == 4);
+
+    buffer.rewind_to_marker();
+    CHECK(buffer.peek() == "");
+    CHECK(buffer.size() == 0);
+
+    // Can't rewind, no marker set: Cleared by `rewind_to_marker`.
+    CHECK_THROWS_AS(buffer.rewind_to_marker(), questdb::ilp::line_sender_error);
+
+    buffer.table("a").symbol("b", "c");
+    CHECK_THROWS_AS(buffer.set_marker(), questdb::ilp::line_sender_error);
+    CHECK_THROWS_AS(buffer.rewind_to_marker(), questdb::ilp::line_sender_error);
+    CHECK(buffer.peek() == "a,b=c");
+
+    buffer.at_now();
+    CHECK(buffer.peek() == "a,b=c\n");
+
+    buffer.set_marker();
+    buffer.clear_marker();
+    buffer.clear_marker();
+    CHECK_THROWS_AS(buffer.rewind_to_marker(), questdb::ilp::line_sender_error);
+    buffer.set_marker();
+    buffer.table("d").symbol("e", "f");
+    CHECK(buffer.peek() == "a,b=c\nd,e=f");
+
+    buffer.rewind_to_marker();
+    CHECK(buffer.peek() == "a,b=c\n");
+
+    buffer.clear();
+    CHECK(buffer.peek() == "");
+    CHECK_THROWS_AS(buffer.rewind_to_marker(), questdb::ilp::line_sender_error);
 }

include/questdb/ilp/line_sender.h

@@ -75,7 +75,11 @@ typedef enum line_sender_error_code
 LINESENDER_API
 line_sender_error_code line_sender_error_get_code(const line_sender_error*);
 
-/** ASCII encoded error message. Never returns NULL. */
+/**
+ * UTF-8 encoded error message. Never returns NULL.
+ * The `len_out` argument is set to the number of bytes in the string.
+ * The string is NOT null-terminated.
+ */
 LINESENDER_API
 const char* line_sender_error_msg(const line_sender_error*, size_t* len_out);
 
@@ -86,7 +90,10 @@ void line_sender_error_free(line_sender_error*);
 
 /////////// Preparing strings and names
 
-/** Non-owning validated UTF-8 encoded string. */
+/**
+ * Non-owning validated UTF-8 encoded string.
+ * The string need not be null-terminated.
+ */
 typedef struct line_sender_utf8
 {
     // Don't initialize fields directly.
@@ -100,7 +107,7 @@ typedef struct line_sender_utf8
  *
  * @param[out] str The object to be initialized.
  * @param[in] len Length in bytes of the buffer.
- * @param[in] buf UTF-8 encoded buffer.
+ * @param[in] buf UTF-8 encoded buffer. Need not be null-terminated.
  * @param[out] err_out Set on error.
  * @return true on success, false on error.
  */
@@ -124,7 +131,10 @@ line_sender_utf8 line_sender_utf8_assert(size_t len, const char* buf);
 #define QDB_UTF8_LITERAL(literal)                                              \
     line_sender_utf8_assert(sizeof(literal) - 1, (literal))
 
-/** Non-owning validated table, symbol or column name. UTF-8 encoded. */
+/**
+ * Non-owning validated table, symbol or column name. UTF-8 encoded.
+ * Need not be null-terminated.
+ */
 typedef struct line_sender_table_name
 {
     // Don't initialize fields directly.
@@ -139,7 +149,7 @@ typedef struct line_sender_table_name
  *
  * @param[out] name The object to be initialized.
  * @param[in] len Length in bytes of the buffer.
- * @param[in] buf UTF-8 encoded buffer.
+ * @param[in] buf UTF-8 encoded buffer. Need not be null-terminated.
  * @param[out] err_out Set on error.
  * @return true on success, false on error.
  */
@@ -166,7 +176,10 @@ line_sender_table_name line_sender_table_name_assert(
 #define QDB_TABLE_NAME_LITERAL(literal)                                        \
     line_sender_table_name_assert(sizeof(literal) - 1, (literal))
 
-/** Non-owning validated table, symbol or column name. UTF-8 encoded. */
+/**
+ * Non-owning validated table, symbol or column name. UTF-8 encoded.
+ * Need not be null-terminated.
+ */
 typedef struct line_sender_column_name
 {
     // Don't initialize fields directly.
@@ -181,7 +194,7 @@ typedef struct line_sender_column_name
  *
  * @param[out] name The object to be initialized.
  * @param[in] len Length in bytes of the buffer.
- * @param[in] buf UTF-8 encoded buffer.
+ * @param[in] buf UTF-8 encoded buffer. Need not be null-terminated.
  * @param[out] err_out Set on error.
  * @return true on success, false on error.
  */
@@ -236,6 +249,7 @@ line_sender_buffer* line_sender_buffer_clone(const line_sender_buffer* buffer);
 /**
  * Pre-allocate to ensure the buffer has enough capacity for at least the
  * specified additional byte count. This may be rounded up.
+ * This does not allocate if such additional capacity is already satisfied.
  * See: `capacity`.
  */
 LINESENDER_API
@@ -247,6 +261,32 @@ void line_sender_buffer_reserve(
 LINESENDER_API
 size_t line_sender_buffer_capacity(const line_sender_buffer* buffer);
 
+/**
+ * Mark a rewind point.
+ * This allows undoing accumulated changes to the buffer for one or more
+ * rows by calling `rewind_to_marker`.
+ * Any previous marker will be discarded.
+ * Once the marker is no longer needed, call `clear_marker`.
+ */
+LINESENDER_API
+bool line_sender_buffer_set_marker(
+    line_sender_buffer* buffer,
+    line_sender_error** err_out);
+
+/**
+ * Undo all changes since the last `set_marker` call.
+ * As a side-effect, this also clears the marker.
+ */
+LINESENDER_API
+bool line_sender_buffer_rewind_to_marker(
+    line_sender_buffer* buffer,
+    line_sender_error** err_out);
+
+/** Discard the marker. */
+LINESENDER_API
+void line_sender_buffer_clear_marker(
+    line_sender_buffer* buffer);
+
 /**
  * Remove all accumulated data and prepare the buffer for new lines.
  * This does not affect the buffer's capacity.

include/questdb/ilp/line_sender.hpp

@@ -317,6 +317,13 @@ namespace questdb::ilp
             return *this;
         }
 
+        /**
+         * Pre-allocate to ensure the buffer has enough capacity for at least
+         * the specified additional byte count. This may be rounded up.
+         * This does not allocate if such additional capacity is already
+         * satisfied.
+         * See: `capacity`.
+         */
         void reserve(size_t additional)
         {
             may_init();
@@ -357,6 +364,28 @@ namespace questdb::ilp
             }
         }
 
+        void set_marker()
+        {
+            may_init();
+            line_sender_error::wrapped_call(
+                ::line_sender_buffer_set_marker, _impl);
+        }
+
+        void rewind_to_marker()
+        {
+            may_init();
+            line_sender_error::wrapped_call(
+                ::line_sender_buffer_rewind_to_marker, _impl);
+        }
+
+        void clear_marker() noexcept
+        {
+            if (_impl)
+            {
+                ::line_sender_buffer_clear_marker(_impl);
+            }
+        }
+
         void clear() noexcept
         {
             if (_impl)

src/ffi.rs

@@ -134,15 +134,17 @@ pub unsafe extern "C" fn line_sender_error_get_code(
     (&*error).0.code().into()
 }
 
-/// ASCII encoded error message. Never returns NULL.
+/// UTF-8 encoded error message. Never returns NULL.
+/// The `len_out` argument is set to the number of bytes in the string.
+/// The string is NOT null-terminated.
 #[no_mangle]
 pub unsafe extern "C" fn line_sender_error_msg(
     error: *const line_sender_error,
     len_out: *mut size_t) -> *const c_char
 {
     let msg: &str = &(&*error).0.msg;
     *len_out = msg.len();
-    msg.as_ptr() as *mut i8
+    msg.as_ptr() as *mut c_char
 }
 
 /// Clean up the error.
@@ -154,6 +156,7 @@ pub unsafe extern "C" fn line_sender_error_free(error: *mut line_sender_error) {
 }
 
 /// Non-owning validated UTF-8 encoded string.
+/// The string need not be null-terminated.
 #[repr(C)]
 #[derive(Debug, Copy, Clone)]
 pub struct line_sender_utf8 {
@@ -259,7 +262,7 @@ unsafe fn unwrap_utf8(
 ///
 /// @param[out] str The object to be initialized.
 /// @param[in] len Length in bytes of the buffer.
-/// @param[in] buf UTF-8 encoded buffer.
+/// @param[in] buf UTF-8 encoded buffer. Need not be null-terminated.
 /// @param[out] err_out Set on error.
 /// @return true on success, false on error.
 #[no_mangle]
@@ -298,6 +301,7 @@ pub unsafe extern "C" fn line_sender_utf8_assert(
 }
 
 /// Non-owning validated table name. UTF-8 encoded.
+/// Need not be null-terminated.
 #[repr(C)]
 #[derive(Debug, Copy, Clone)]
 pub struct line_sender_table_name
@@ -317,6 +321,7 @@ impl line_sender_table_name {
 }
 
 /// Non-owning validated symbol or column name. UTF-8 encoded.
+/// Need not be null-terminated.
 #[repr(C)]
 #[derive(Debug, Copy, Clone)]
 pub struct line_sender_column_name
@@ -340,7 +345,7 @@ impl line_sender_column_name {
 ///
 /// @param[out] name The object to be initialized.
 /// @param[in] len Length in bytes of the buffer.
-/// @param[in] buf UTF-8 encoded buffer.
+/// @param[in] buf UTF-8 encoded buffer. Need not be null-terminated.
 /// @param[out] err_out Set on error.
 /// @return true on success, false on error.
 #[no_mangle]
@@ -387,7 +392,7 @@ pub unsafe extern "C" fn line_sender_table_name_assert(
 ///
 /// @param[out] name The object to be initialized.
 /// @param[in] len Length in bytes of the buffer.
-/// @param[in] buf UTF-8 encoded buffer.
+/// @param[in] buf UTF-8 encoded buffer. Need not be null-terminated.
 /// @param[out] err_out Set on error.
 /// @return true on success, false on error.
 #[no_mangle]
@@ -634,6 +639,42 @@ pub unsafe extern "C" fn line_sender_buffer_capacity(
     unwrap_buffer(buffer).capacity()
 }
 
+/// Mark a rewind point.
+/// This allows undoing accumulated changes to the buffer for one or more
+/// rows by calling `rewind_to_marker`.
+/// Any previous marker will be discarded.
+/// Once the marker is no longer needed, call `clear_marker`.
+#[no_mangle]
+pub unsafe extern "C" fn line_sender_buffer_set_marker(
+    buffer: *mut line_sender_buffer,
+    err_out: *mut *mut line_sender_error) -> bool
+{
+    let buffer = unwrap_buffer_mut(buffer);
+    bubble_err_to_c!(err_out, buffer.set_marker());
+    true
+}
+
+/// Undo all changes since the last `set_marker` call.
+/// As a side-effect, this also clears the marker.
+#[no_mangle]
+pub unsafe extern "C" fn line_sender_buffer_rewind_to_marker(
+    buffer: *mut line_sender_buffer,
+    err_out: *mut *mut line_sender_error) -> bool
+{
+    let buffer = unwrap_buffer_mut(buffer);
+    bubble_err_to_c!(err_out, buffer.rewind_to_marker());
+    true
+}
+
+/// Discard the marker.
+#[no_mangle]
+pub unsafe extern "C" fn line_sender_buffer_clear_marker(
+    buffer: *mut line_sender_buffer)
+{
+    let buffer = unwrap_buffer_mut(buffer);
+    buffer.clear_marker();
+}
+
 /// Remove all accumulated data and prepare the buffer for new lines.
 /// This does not affect the buffer's capacity.
 #[no_mangle]