feat: protocol version 2 and n-dimensional array ingestion support

Author: kafka1991

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 4.0.4
+current_version = 5.0.0-rc1
 commit = False
 tag = False
 

.gitignore

@@ -9,4 +9,5 @@ cmake-build-release
 .cache
 questdb-rs/Cargo.lock
 include/questdb/ingress/line_sender.gen.h
-cython/questdb/ingress/line_sender.pxd
+cython/questdb/ingress/line_sender.pxd
+profile.out

CMakeLists.txt

@@ -1,5 +1,6 @@
 cmake_minimum_required(VERSION 3.15.0)
-project(c-questdb-client VERSION 4.0.4)
+project(c-questdb-client VERSION 5.0.0)
+set(PROJECT_PRE_RELEASE "rc1")
 
 set(CPACK_PROJECT_NAME ${PROJECT_NAME})
 set(CPACK_PROJECT_VERSION ${PROJECT_VERSION})
@@ -98,6 +99,14 @@ if (QUESTDB_TESTS_AND_EXAMPLES)
         line_sender_c_example
         examples/concat.c
         examples/line_sender_c_example.c)
+    compile_example(
+        line_sender_c_example_array_byte_strides
+        examples/concat.c
+        examples/line_sender_c_example_array_byte_strides.c)
+    compile_example(
+        line_sender_c_example_array_elem_strides
+        examples/concat.c
+        examples/line_sender_c_example_array_elem_strides.c)
     compile_example(
         line_sender_c_example_auth
         examples/concat.c
@@ -123,6 +132,12 @@ if (QUESTDB_TESTS_AND_EXAMPLES)
     compile_example(
         line_sender_cpp_example
         examples/line_sender_cpp_example.cpp)
+    compile_example(
+        line_sender_cpp_example_array_byte_strides
+        examples/line_sender_cpp_example_array_byte_strides.cpp)
+    compile_example(
+        line_sender_cpp_example_array_elem_strides
+        examples/line_sender_cpp_example_array_elem_strides.cpp)
     compile_example(
         line_sender_cpp_example_auth
         examples/line_sender_cpp_example_auth.cpp)

README.md

@@ -3,9 +3,13 @@
 
 This library makes it easy to insert data into [QuestDB](https://questdb.io/).
 
-This client library implements the [InfluxDB Line Protocol](
+This client library implements the QuestDB's variant of the [InfluxDB Line Protocol](
 https://questdb.io/docs/reference/api/ilp/overview/) (ILP) over HTTP and TCP.
 
+When connecting to QuestDB over HTTP, the library will auto-detect the server's
+latest supported version and use it. Version 1 is compatible with
+the [InfluxDB Database](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/).
+
 * Implementation is in Rust, with no additional
   [run-time or link-time dependencies](doc/BUILD.md#pre-requisites-and-dependencies)
   on the C++ standard library or other libraries.
@@ -38,6 +42,21 @@ For an overview and code examples, see the
 To understand the protocol in more depth, consult the
 [protocol reference docs](https://questdb.io/docs/reference/api/ilp/overview/).
 
+## Protocol Versions
+
+The library supports the following ILP protocol versions.
+
+These protocol versions are supported over both HTTP and TCP.
+
+If you use HTTP, the library will automatically detect the server's
+latest supported protocol version and use it. If you use TCP, you can specify the
+`protocol_version=N` parameter when constructing the `Sender` object.
+
+| Version | Description                                             | Server Comatibility   |
+| ------- | ------------------------------------------------------- | --------------------- |
+| **1**   | Over HTTP it's compatible InfluxDB Line Protocol (ILP)  | All QuestDB versions  |
+| **2**   | 64-bit floats sent as binary, adds n-dimentional arrays | 8.4.0+ (2023-10-30)   |
+
 ## Getting Started
 
 To get started, read the language-specific guides.

ci/compile.yaml

@@ -4,6 +4,10 @@ steps:
       rustup default $(toolchain)
     condition: ne(variables['toolchain'], '')
     displayName: "Update and set Rust toolchain"
+  - script: |
+      python -m pip install --upgrade pip
+      pip install numpy
+    displayName: 'Install Python Dependencies'
   - script: cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -DQUESTDB_TESTS_AND_EXAMPLES=ON
     env:
       JAVA_HOME: $(JAVA_HOME_11_X64)

ci/run_all_tests.py

@@ -41,7 +41,7 @@ def main():
         build_cxx20_dir.glob(f'**/test_line_sender{exe_suffix}'))) 
 
     system_test_path = pathlib.Path('system_test') / 'test.py'
-    qdb_v = '8.2.3'  # The version of QuestDB we'll test against.
+    #qdb_v = '8.2.3'  # The version of QuestDB we'll test against.
 
     run_cmd('cargo', 'test',
             '--', '--nocapture', cwd='questdb-rs')
@@ -51,9 +51,11 @@ def main():
             '--', '--nocapture', cwd='questdb-rs')
     run_cmd('cargo', 'test', '--features=almost-all-features',
             '--', '--nocapture', cwd='questdb-rs')
+    run_cmd('cargo', 'test', cwd='questdb-rs-ffi')
     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')
+    #run_cmd('python3', str(system_test_path), 'run', '--versions', qdb_v, '-v')
+    run_cmd('python3', str(system_test_path), 'run', '--repo', './questdb_nd_arr', '-v')
 
 
 if __name__ == '__main__':

ci/run_tests_pipeline.yaml

@@ -54,14 +54,25 @@ stages:
               cd questdb-rs
               cargo build --examples --features almost-all-features
             displayName: "Build Rust examples"
+          ############################# temp for test begin #####################
+          - script: |
+              git clone -b nd_arr --depth 1 https://github.com/questdb/questdb.git ./questdb_nd_arr
+            displayName: git clone questdb
+          - task: Maven@3
+            displayName: "Compile QuestDB"
+            inputs:
+              mavenPOMFile: 'questdb_nd_arr/pom.xml'
+              jdkVersionOption: '1.11'
+              options: "-DskipTests -Pbuild-web-console"
+          ############################# temp for test end ##################### 
           - script: python3 ci/run_all_tests.py
             env:
               JAVA_HOME: $(JAVA_HOME_11_X64)
             displayName: "Tests"
-          - task: PublishBuildArtifacts@1
-            inputs:
-              pathToPublish: ./build
-            displayName: "Publish build directory"
+          # - task: PublishBuildArtifacts@1
+          #   inputs:
+          #     pathToPublish: ./build
+          #   displayName: "Publish build directory"
       - job: FormatAndLinting
         displayName: "cargo fmt and clippy"
         pool:
@@ -115,7 +126,7 @@ stages:
             submodules: false
           - template: compile.yaml
           - script: |
-              git clone --depth 1 https://github.com/questdb/questdb.git
+              git clone -b nd_arr --depth 1 https://github.com/questdb/questdb.git
             displayName: git clone questdb
           - task: Maven@3
             displayName: "Compile QuestDB"

cpp_test/mock_server.cpp

@@ -25,6 +25,7 @@
 #include "mock_server.hpp"
 
 #include <string.h>
+#include <string>
 
 #if defined(PLATFORM_UNIX)
 #    include <fcntl.h>
@@ -190,43 +191,81 @@ bool mock_server::wait_for_data(std::optional<double> wait_timeout_sec)
     return !!count;
 }
 
+int32_t bytes_to_int32_le(const std::byte* bytes)
+{
+    return static_cast<int32_t>(
+        (bytes[0] << 0) | (bytes[1] << 8) | (bytes[2] << 16) |
+        (bytes[3] << 24));
+}
+
 size_t mock_server::recv(double wait_timeout_sec)
 {
     if (!wait_for_data(wait_timeout_sec))
         return 0;
 
-    char chunk[1024];
+    std::byte chunk[1024];
     size_t chunk_len{sizeof(chunk)};
-    std::vector<char> accum;
+    std::vector<std::byte> accum;
     for (;;)
     {
         wait_for_data();
-        sock_ssize_t count =
-            ::recv(_conn_fd, &chunk[0], static_cast<sock_len_t>(chunk_len), 0);
+        sock_ssize_t count = ::recv(
+            _conn_fd,
+            reinterpret_cast<char*>(&chunk[0]),
+            static_cast<sock_len_t>(chunk_len),
+            0);
         if (count == -1)
             throw std::runtime_error{"Bad `recv()`."};
         const size_t u_count = static_cast<size_t>(count);
         accum.insert(accum.end(), chunk, chunk + u_count);
         if (accum.size() < 2)
             continue;
-        if ((accum[accum.size() - 1] == '\n') &&
-            (accum[accum.size() - 2] != '\\'))
+        if ((accum[accum.size() - 1] == std::byte('\n')) &&
+            (accum[accum.size() - 2] != std::byte('\\')))
             break;
     }
 
     size_t received_count{0};
-    const char* head{&accum[0]};
-    for (size_t index = 1; index < accum.size(); ++index)
+    const std::byte* head{&accum[0]};
+    size_t index{1};
+    while (index < accum.size())
     {
-        const char& last = accum[index];
-        const char& prev = accum[index - 1];
-        if ((last == '\n') && (prev != '\\'))
+        const std::byte& last = accum[index];
+        const std::byte& prev = accum[index - 1];
+        if (last == std::byte('=') && prev == std::byte('='))
+        {
+            index++;
+            std::byte& binary_type = accum[index];
+            if (binary_type == std::byte(16)) // DOUBLE_BINARY_FORMAT_TYPE
+                index += sizeof(double) + 1;
+            else if (binary_type == std::byte(14)) // ARRAY_BINARY_FORMAT_TYPE
+            {
+                index++;
+                const std::byte& array_elem_type = accum[index];
+                if (array_elem_type == std::byte(10))
+                {
+                    index++;
+                    const size_t dims = size_t(accum[index]);
+                    index++;
+                    size_t data_size{sizeof(double)};
+                    for (size_t i = 0; i < dims; i++)
+                    {
+                        data_size *= bytes_to_int32_le(&accum[index]);
+                        index += sizeof(int32_t);
+                    }
+                    index += data_size;
+                }
+            }
+            continue;
+        }
+        else if ((last == std::byte('\n')) && (prev != std::byte('\\')))
         {
-            const char* tail{&last + 1};
-            _msgs.emplace_back(head, tail - head);
+            const std::byte* tail{&last + 1};
+            _msgs.emplace_back(head, tail);
             head = tail;
             ++received_count;
         }
+        index++;
     }
     return received_count;
 }

cpp_test/mock_server.hpp

@@ -24,13 +24,17 @@
 
 #pragma once
 
+#include <cassert>
 #include <vector>
-#include <string>
 #include <cstdint>
 #include <optional>
 #include <stdexcept>
-
 #include "build_env.h"
+#if __cplusplus < 202002L
+#    include "questdb/ingress/line_sender.hpp"
+#else
+#    include <span>
+#endif
 
 #if defined(PLATFORM_UNIX)
 typedef int socketfd_t;
@@ -60,9 +64,14 @@ class mock_server
 
     size_t recv(double wait_timeout_sec = 0.1);
 
-    const std::vector<std::string>& msgs() const
+#if __cplusplus >= 202002L
+    using buffer_view = std::span<const std::byte>;
+#endif
+
+    buffer_view msgs(size_t index) const
     {
-        return _msgs;
+        assert(index < _msgs.size());
+        return {_msgs[index].data(), _msgs[index].size()};
     }
 
     void close();
@@ -75,7 +84,7 @@ class mock_server
     socketfd_t _listen_fd;
     socketfd_t _conn_fd;
     uint16_t _port;
-    std::vector<std::string> _msgs;
+    std::vector<std::vector<std::byte>> _msgs;
 };
 
 } // namespace questdb::ingress::test