introduce ndarr support

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,12 @@ 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')
+    #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,6 +54,17 @@ 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)
@@ -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

@@ -190,43 +190,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