introduce line protocol version

Author: kafka1991

ci/run_all_tests.py

@@ -51,8 +51,6 @@ def main():
             '--', '--nocapture', cwd='questdb-rs')
     run_cmd('cargo', 'test', '--features=almost-all-features',
             '--', '--nocapture', cwd='questdb-rs')
-    run_cmd('cargo', 'test', '--features=almost-all-features,protocol-version-1',
-        '--', '--nocapture', cwd='questdb-rs')
     run_cmd(str(test_line_sender_path))
     run_cmd(str(test_line_sender_path_CXX20))
     #run_cmd('python3', str(system_test_path), 'run', '--versions', qdb_v, '-v')

cpp_test/test_line_sender.cpp

@@ -694,13 +694,19 @@ TEST_CASE("os certs")
 
     {
         questdb::ingress::opts opts{
-            questdb::ingress::protocol::https, "localhost", server.port()};
+            questdb::ingress::protocol::https,
+            "localhost",
+            server.port(),
+            true};
         opts.tls_ca(questdb::ingress::ca::os_roots);
     }
 
     {
         questdb::ingress::opts opts{
-            questdb::ingress::protocol::https, "localhost", server.port()};
+            questdb::ingress::protocol::https,
+            "localhost",
+            server.port(),
+            true};
         opts.tls_ca(questdb::ingress::ca::webpki_and_os_roots);
     }
 }
@@ -729,9 +735,12 @@ TEST_CASE("Opts copy ctor, assignment and move testing.")
 
     {
         questdb::ingress::opts opts1{
-            questdb::ingress::protocol::https, "localhost", "9009"};
+            questdb::ingress::protocol::https, "localhost", "9009", true};
         questdb::ingress::opts opts2{
-            questdb::ingress::protocol::https, "altavista.digital.com", "9009"};
+            questdb::ingress::protocol::https,
+            "altavista.digital.com",
+            "9009",
+            true};
         opts1 = opts2;
     }
 }
@@ -901,15 +910,15 @@ TEST_CASE("Opts from conf")
 TEST_CASE("HTTP basics")
 {
     questdb::ingress::opts opts1{
-        questdb::ingress::protocol::http, "localhost", 1};
+        questdb::ingress::protocol::http, "localhost", 1, true};
     questdb::ingress::opts opts1conf = questdb::ingress::opts::from_conf(
         "http::addr=localhost:1;username=user;password=pass;request_timeout="
-        "5000;retry_timeout=5;");
+        "5000;retry_timeout=5;disable_line_protocol_validation=on;");
     questdb::ingress::opts opts2{
-        questdb::ingress::protocol::https, "localhost", "1"};
+        questdb::ingress::protocol::https, "localhost", "1", true};
     questdb::ingress::opts opts2conf = questdb::ingress::opts::from_conf(
         "http::addr=localhost:1;token=token;request_min_throughput=1000;retry_"
-        "timeout=0;");
+        "timeout=0;disable_line_protocol_validation=on;");
     opts1.username("user")
         .password("pass")
         .max_buf_size(1000000)

include/questdb/ingress/line_sender.h

@@ -95,6 +95,18 @@ typedef enum line_sender_protocol
     line_sender_protocol_https,
 } line_sender_protocol;
 
+/** The line protocol version used to write data to buffer. */
+typedef enum line_protocol_version
+{
+    /** Version 1 of InfluxDB Line Protocol.
+    Uses text format serialization for f64. */
+    line_protocol_version_1,
+
+    /** Version 2 of InfluxDB Line Protocol.
+    Uses binary format serialization for f64, and support array data type.*/
+    line_protocol_version_2,
+} line_protocol_version;
+
 /** Possible sources of the root certificates used to validate the server's
  * TLS certificate. */
 typedef enum line_sender_ca
@@ -296,6 +308,23 @@ line_sender_buffer* line_sender_buffer_new();
 LINESENDER_API
 line_sender_buffer* line_sender_buffer_with_max_name_len(size_t max_name_len);
 
+/**
+ * Sets the Line Protocol version for line_sender_buffer.
+ *
+ * The buffer defaults is line_protocol_version_2 which uses
+ * binary format f64 serialization and support array data type. Call this to
+ * switch to version 1 (text format f64) when connecting to servers that don't
+ * support line_protocol_version_2(under 8.3.2).
+ *
+ * Must be called before adding any data to the buffer. Protocol version cannot
+ * be changed after the buffer contains data.
+ */
+LINESENDER_API
+line_sender_buffer* line_sender_buffer_set_line_protocol_version(
+    line_sender_buffer* buffer,
+    line_protocol_version version,
+    line_sender_error** err_out);
+
 /** Release the `line_sender_buffer` object. */
 LINESENDER_API
 void line_sender_buffer_free(line_sender_buffer* buffer);
@@ -884,6 +913,25 @@ line_sender* line_sender_from_conf(
 LINESENDER_API
 line_sender* line_sender_from_env(line_sender_error** err_out);
 
+/**
+ * Returns the client's recommended default line protocol version.
+ * Will be used to [`line_sender_buffer_set_line_protocol_version`]
+ *
+ * The version selection follows these rules:
+ * 1. TCP/TCPS Protocol: Always returns [`LineProtocolVersion::V2`]
+ * 2. HTTP/HTTPS Protocol:
+ *   - If line protocol auto-detection is disabled
+ *    [`line_sender_opts_disable_line_protocol_validation`], returns
+ *    [`LineProtocolVersion::V2`]
+ *   - If line protocol auto-detection is enabled:
+ *     - Uses the server's default version if supported by the client
+ *     - Otherwise uses the highest mutually supported version from the
+ *       intersection of client and server compatible versions.
+ */
+LINESENDER_API
+line_protocol_version line_sender_default_line_protocol_version(
+    line_sender* sender);
+
 /**
  * Tell whether the sender is no longer usable and must be closed.
  * This happens when there was an earlier failure.

include/questdb/ingress/line_sender.hpp

@@ -401,17 +401,30 @@ class line_sender_buffer
     {
     }
 
+    line_sender_buffer(
+        size_t init_buf_size,
+        size_t max_name_len,
+        line_protocol_version version) noexcept
+        : _impl{nullptr}
+        , _init_buf_size{init_buf_size}
+        , _max_name_len{max_name_len}
+        , _line_protocol_version{version}
+    {
+    }
+
     line_sender_buffer(const line_sender_buffer& other) noexcept
         : _impl{::line_sender_buffer_clone(other._impl)}
         , _init_buf_size{other._init_buf_size}
         , _max_name_len{other._max_name_len}
+        , _line_protocol_version{other._line_protocol_version}
     {
     }
 
     line_sender_buffer(line_sender_buffer&& other) noexcept
         : _impl{other._impl}
         , _init_buf_size{other._init_buf_size}
         , _max_name_len{other._max_name_len}
+        , _line_protocol_version{other._line_protocol_version}
     {
         other._impl = nullptr;
     }
@@ -427,6 +440,7 @@ class line_sender_buffer
                 _impl = nullptr;
             _init_buf_size = other._init_buf_size;
             _max_name_len = other._max_name_len;
+            _line_protocol_version = other._line_protocol_version;
         }
         return *this;
     }
@@ -439,11 +453,32 @@ class line_sender_buffer
             _impl = other._impl;
             _init_buf_size = other._init_buf_size;
             _max_name_len = other._max_name_len;
+            _line_protocol_version = other._line_protocol_version;
             other._impl = nullptr;
         }
         return *this;
     }
 
+    /**
+     * Sets the Line Protocol version for line_sender_buffer.
+     *
+     * The buffer defaults is line_protocol_version_2 which uses
+     * binary format f64 serialization and support array data type. Call this to
+     * switch to version 1 (text format f64) when connecting to servers that
+     * don't support line_protocol_version_2(under 8.3.2).
+     *
+     * Must be called before adding any data to the buffer. Protocol version
+     * cannot be changed after the buffer contains data.
+     */
+    line_sender_buffer& set_line_protocol_version(line_protocol_version v)
+    {
+        may_init();
+        line_sender_error::wrapped_call(
+            ::line_sender_buffer_set_line_protocol_version, _impl, v);
+        _line_protocol_version = v;
+        return *this;
+    }
+
     /**
      * Pre-allocate to ensure the buffer has enough capacity for at least
      * the specified additional byte count. This may be rounded up.
@@ -803,12 +838,17 @@ class line_sender_buffer
         {
             _impl = ::line_sender_buffer_with_max_name_len(_max_name_len);
             ::line_sender_buffer_reserve(_impl, _init_buf_size);
+            line_sender_error::wrapped_call(
+                line_sender_buffer_set_line_protocol_version,
+                _impl,
+                _line_protocol_version);
         }
     }
 
     ::line_sender_buffer* _impl;
     size_t _init_buf_size;
     size_t _max_name_len;
+    line_protocol_version _line_protocol_version{::line_protocol_version_2};
 
     friend class line_sender;
 };
@@ -868,13 +908,24 @@ class opts
      * @param[in] protocol The protocol to use.
      * @param[in] host The QuestDB database host.
      * @param[in] port The QuestDB tcp or http port.
+     * @param[in] disable_line_protocol_validation disable line protocol version
+     * validation.
      */
-    opts(protocol protocol, utf8_view host, uint16_t port) noexcept
+    opts(
+        protocol protocol,
+        utf8_view host,
+        uint16_t port,
+        bool disable_line_protocol_validation = false) noexcept
         : _impl{::line_sender_opts_new(
               static_cast<::line_sender_protocol>(protocol), host._impl, port)}
     {
         line_sender_error::wrapped_call(
             ::line_sender_opts_user_agent, _impl, _user_agent::name());
+        if (disable_line_protocol_validation)
+        {
+            line_sender_error::wrapped_call(
+                ::line_sender_opts_disable_line_protocol_validation, _impl);
+        }
     }
 
     /**
@@ -883,15 +934,25 @@ class opts
      * @param[in] protocol The protocol to use.
      * @param[in] host The QuestDB database host.
      * @param[in] port The QuestDB tcp or http port as service name.
+     * @param[in] disable_line_protocol_validation disable line protocol version
      */
-    opts(protocol protocol, utf8_view host, utf8_view port) noexcept
+    opts(
+        protocol protocol,
+        utf8_view host,
+        utf8_view port,
+        bool disable_line_protocol_validation = false) noexcept
         : _impl{::line_sender_opts_new_service(
               static_cast<::line_sender_protocol>(protocol),
               host._impl,
               port._impl)}
     {
         line_sender_error::wrapped_call(
             ::line_sender_opts_user_agent, _impl, _user_agent::name());
+        if (disable_line_protocol_validation)
+        {
+            line_sender_error::wrapped_call(
+                ::line_sender_opts_disable_line_protocol_validation, _impl);
+        }
     }
 
     opts(const opts& other) noexcept
@@ -1194,13 +1255,23 @@ class line_sender
         return {opts::from_env()};
     }
 
-    line_sender(protocol protocol, utf8_view host, uint16_t port)
-        : line_sender{opts{protocol, host, port}}
+    line_sender(
+        protocol protocol,
+        utf8_view host,
+        uint16_t port,
+        bool disable_line_protocol_validation = false)
+        : line_sender{
+              opts{protocol, host, port, disable_line_protocol_validation}}
     {
     }
 
-    line_sender(protocol protocol, utf8_view host, utf8_view port)
-        : line_sender{opts{protocol, host, port}}
+    line_sender(
+        protocol protocol,
+        utf8_view host,
+        utf8_view port,
+        bool disable_line_protocol_validation = false)
+        : line_sender{
+              opts{protocol, host, port, disable_line_protocol_validation}}
     {
     }