add Timeplus APIs (#18)

.clang-format

@@ -2,10 +2,11 @@ Language: Cpp
 BasedOnStyle: Google
 
 AccessModifierOffset: -4
-AlignConsecutiveAssignments: true
+AlignConsecutiveAssignments: false
 AllowShortFunctionsOnASingleLine: InlineOnly
 ColumnLimit: 140
 DerivePointerAlignment: false
 FixNamespaceComments: true
+IncludeBlocks: Preserve
 IndentWidth: 4
 PointerAlignment: Left

.github/workflows/timeplus_cpp_ci.yml

@@ -21,10 +21,10 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-20.04]
-        compiler: [clang-10, clang-18, gcc-7, gcc-8, gcc-9]
+        compiler: [clang-10, clang-18, gcc-8, gcc-9]
         ssl: [ssl_ON, ssl_OFF]
         dependencies: [dependencies_BUILT_IN]
-        timeplusd: [2.3.22]
+        timeplusd: [2.3.23]
 
         include:
         - compiler: clang-10
@@ -37,11 +37,6 @@ jobs:
           C_COMPILER: clang-18
           CXX_COMPILER: clang++-18
 
-        - compiler: gcc-7
-          COMPILER_INSTALL: gcc-7 g++-7
-          C_COMPILER: gcc-7
-          CXX_COMPILER: g++-7
-
         - compiler: gcc-8
           COMPILER_INSTALL: gcc-8 g++-8
           C_COMPILER: gcc-8
@@ -58,7 +53,7 @@ jobs:
         - dependencies: dependencies_SYSTEM
           compiler: compiler_SYSTEM
           os: ubuntu-22.04
-          timeplusd: 2.3.22
+          timeplusd: 2.3.23
           COMPILER_INSTALL: gcc g++
           C_COMPILER: gcc
           CXX_COMPILER: g++

CMakeLists.txt

@@ -16,16 +16,24 @@ OPTION (WITH_SYSTEM_ABSEIL "Use system ABSEIL" OFF)
 OPTION (WITH_SYSTEM_LZ4 "Use system LZ4" OFF)
 OPTION (WITH_SYSTEM_CITYHASH "Use system cityhash" OFF)
 OPTION (WITH_SYSTEM_ZSTD "Use system ZSTD" OFF)
-OPTION (DEBUG_DEPENDENCIES "Print debug info about dependencies duting build" ON)
+OPTION (DEBUG_DEPENDENCIES "Print debug info about dependencies during build" ON)
 OPTION (CHECK_VERSION "Check that version number corresponds to git tag, usefull in CI/CD to validate that new version published on GitHub has same version in sources" OFF)
 OPTION (BUILD_GTEST "Build Google Test" OFF)
 OPTION (ENABLE_GEOMETRIC_TEST "Enable geometric test" OFF)
+OPTION (ENABLE_TRACE_TIMEPLUS_CPP "Enable tracing log" OFF)
 
 PROJECT (TIMEPLUS-CLIENT
     VERSION "${TIMEPLUS_CPP_VERSION}"
     DESCRIPTION "Timeplus C++ client library"
 )
 
+set(CMAKE_EXPORT_COMPILE_COMMANDS 1)
+
+IF (ENABLE_TRACE_TIMEPLUS_CPP)
+    ADD_DEFINITIONS(-DTRACE_TIMEPLUS_CPP)
+ENDIF()
+
+
 USE_CXX17 ()
 USE_OPENSSL ()
 
@@ -127,6 +135,8 @@ IF (BUILD_TESTS)
     INCLUDE_DIRECTORIES (contrib/gtest/include contrib/gtest)
     SUBDIRS (
         tests/simple
+        tests/insert
+        tests/insert-async
     )
 ENDIF (BUILD_TESTS)
 

README.md

@@ -39,7 +39,7 @@ Optional dependencies:
 ```sh
 $ mkdir build .
 $ cd build
-$ cmake -D CMAKE_BUILD_TYPE=Release ..
+$ cmake -D CMAKE_BUILD_TYPE=Release -D BUILD_TESTS=ON -D BUILD_EXAMPLES=ON -D BUILD_GTEST=ON ..
 $ make
 ```
 

examples/CMakeLists.txt

@@ -3,3 +3,6 @@ add_executable(timeplus-client
     main.cpp)
 
 target_link_libraries(timeplus-client PRIVATE timeplus-cpp-lib)
+
+add_executable(insert-examples insert_examples.cpp)
+target_link_libraries(insert-examples PRIVATE timeplus-cpp-lib)

examples/insert_examples.cpp

@@ -0,0 +1,58 @@
+#include <timeplus/timeplus.h>
+
+#include <iostream>
+
+using namespace timeplus;
+
+/// Stream to insert is created with DDL:
+/// `CREATE STREAM insert_examples(i uint64, v string)`
+const std::string TABLE_NAME = "insert_examples";
+
+int main() {
+    TimeplusConfig config;
+    config.client_options.endpoints.push_back({"localhost", 8463});
+    config.max_connections = 3;
+    config.max_retries = 10;
+    config.wait_time_before_retry_ms = 1000;
+    config.task_executors = 1;
+
+    Timeplus tp{std::move(config)};
+
+    auto block = std::make_shared<Block>();
+
+    auto col_i = std::make_shared<ColumnUInt64>();
+    col_i->Append(5);
+    col_i->Append(7);
+    block->AppendColumn("i", col_i);
+
+    auto col_v = std::make_shared<ColumnString>();
+    col_v->Append("five");
+    col_v->Append("seven");
+    block->AppendColumn("v", col_v);
+
+    /// Use synchronous insert API.
+    auto insert_result = tp.Insert(TABLE_NAME, block, /*idempotent_id=*/"block-1");
+    if (insert_result.ok()) {
+        std::cout << "Synchronous insert suceeded." << std::endl;
+    } else {
+        std::cout << "Synchronous insert failed: code=" << insert_result.err_code << " msg=" << insert_result.err_msg << std::endl;
+    }
+
+    /// Use asynchrounous insert API.
+    std::atomic<bool> done = false;
+    tp.InsertAsync(TABLE_NAME, block, /*idempotent_id=*/"block-2", [&done](const BaseResult& result) {
+        const auto& async_insert_result = static_cast<const InsertResult&>(result);
+        if (async_insert_result.ok()) {
+            std::cout << "Asynchronous insert suceeded." << std::endl;
+        } else {
+            std::cout << "Asynchronous insert failed: code=" << async_insert_result.err_code << " msg=" << async_insert_result.err_msg
+                      << std::endl;
+        }
+        done = true;
+    });
+
+    while (!done) {
+    }
+
+    return 0;
+}

tests/insert-async/CMakeLists.txt

@@ -0,0 +1,11 @@
+ADD_EXECUTABLE (insert-async-test
+    main.cpp
+)
+
+TARGET_LINK_LIBRARIES (insert-async-test
+    timeplus-cpp-lib
+)
+
+IF (MSVC)
+    TARGET_LINK_LIBRARIES (insert-async-test Crypt32)
+ENDIF()

tests/insert-async/main.cpp

@@ -0,0 +1,122 @@
+#include <timeplus/blocking_queue.h>
+#include <timeplus/timeplus.h>
+
+#include <atomic>
+#include <chrono>
+#include <iomanip>
+#include <iostream>
+#include <thread>
+
+using namespace timeplus;
+
+const size_t INSERT_BLOCKS = 100'000;
+const size_t BLOCKS_PER_BATCH = 1000;
+
+const std::vector<std::pair<std::string, uint16_t>> HOST_PORTS = {
+    /// Single instance
+    {"localhost", 8463},
+    /// Cluster nodes
+    {"localhost", 18463},
+    {"localhost", 28463},
+    {"localhost", 38463},
+};
+
+void prepareTable() {
+    ClientOptions options;
+    for (const auto& [host, port] : HOST_PORTS) {
+        options.endpoints.push_back({host, port});
+    }
+
+    Client client{options};
+    client.Execute("DROP STREAM IF EXISTS insert_async_test;");
+    client.Execute("CREATE STREAM insert_async_test (i uint64, s string);");
+}
+
+auto timestamp() {
+    auto now = std::chrono::system_clock::now();
+    std::time_t now_time = std::chrono::system_clock::to_time_t(now);
+    std::tm* local_time = std::localtime(&now_time);
+    return std::put_time(local_time, "%Y-%m-%d %H:%M:%S");
+}
+
+int main() {
+    prepareTable();
+
+    TimeplusConfig config;
+    for (const auto& [host, port] : HOST_PORTS) {
+        config.client_options.endpoints.push_back({host, port});
+    }
+    config.max_connections = 4;
+    config.max_retries = 5;
+    config.task_executors = 4;
+    config.task_queue_capacity = BLOCKS_PER_BATCH;  /// use large input queue to avoid deadlock on retry failure
+
+    Timeplus tp{std::move(config)};
+
+    auto block = std::make_shared<Block>();
+    auto col_i = std::make_shared<ColumnUInt64>(std::vector<uint64_t>{5, 7, 4, 8});
+    auto col_s = std::make_shared<ColumnString>(
+        std::vector<std::string>{"Before my bed, the moon is bright,", "I think that it is frost on the ground.",
+                                 "I raise my head to gaze at the bright moon,", "And lower it to think of my hometown."});
+    block->AppendColumn("i", col_i);
+    block->AppendColumn("s", col_s);
+
+    /// Queue to store failed inserts which need to be resent.
+    BlockingQueue<std::pair<size_t, BlockPtr>> insert_failure(BLOCKS_PER_BATCH);
+    std::atomic<size_t> insert_success_count{0};
+
+    auto handle_insert_result = [&insert_failure, &insert_success_count](size_t block_id, const InsertResult& result) {
+        if (result.ok()) {
+            insert_success_count.fetch_add(1);
+        } else {
+            std::cout << "[" << timestamp() << "]\t Failed to insert block: insert_id=" << block_id << " err=" << result.err_msg
+                      << std::endl;
+            insert_failure.emplace(block_id, result.block);
+        }
+    };
+
+    auto async_insert_block = [&tp, &handle_insert_result](size_t block_id, BlockPtr block) {
+        tp.InsertAsync(/*table_name=*/"insert_async_test", block, [block_id, &handle_insert_result](const BaseResult& result) {
+            const auto& insert_result = static_cast<const InsertResult&>(result);
+            handle_insert_result(block_id, insert_result);
+        });
+    };
+
+    auto start_time = std::chrono::high_resolution_clock::now();
+    auto last_time = start_time;
+    for (size_t batch = 0; batch < INSERT_BLOCKS / BLOCKS_PER_BATCH; ++batch) {
+        insert_success_count = 0;
+        /// Insert blocks asynchronously.
+        for (size_t i = 0; i < BLOCKS_PER_BATCH; ++i) {
+            async_insert_block(i, block);
+        }
+
+        /// Wait for all blocks of the batch are inserted.
+        while (insert_success_count.load() != BLOCKS_PER_BATCH) {
+            if (!insert_failure.empty()) {
+                /// Re-insert the failed blocks
+                auto blocks = insert_failure.drain();
+                for (auto &[i, b] : blocks) {
+                    async_insert_block(i, b);
+                }
+            }
+
+            std::this_thread::yield();
+        }
+
+        /// Print insert statistics of the batch.
+        auto current_time = std::chrono::high_resolution_clock::now();
+        std::chrono::duration<double> elapsed = current_time - last_time;
+        last_time = current_time;
+        std::cout << "[" << timestamp() << "]\t" << (batch + 1) * BLOCKS_PER_BATCH << " blocks inserted\telapsed = " << elapsed.count()
+                  << " sec\teps = " << static_cast<double>(BLOCKS_PER_BATCH * block->GetRowCount()) / elapsed.count() << std::endl;
+    }
+
+    /// Print summary.
+    auto current_time = std::chrono::high_resolution_clock::now();
+    std::chrono::duration<double> elapsed = current_time - start_time;
+    std::cout << "\nInsert Done. Total Events = " << INSERT_BLOCKS * block->GetRowCount() << " Total Time = " << elapsed.count() << " sec"
+              << std::endl;
+
+    return 0;
+}

tests/insert/CMakeLists.txt

@@ -0,0 +1,11 @@
+ADD_EXECUTABLE (insert-test
+    main.cpp
+)
+
+TARGET_LINK_LIBRARIES (insert-test
+    timeplus-cpp-lib
+)
+
+IF (MSVC)
+    TARGET_LINK_LIBRARIES (insert-test Crypt32)
+ENDIF()

tests/insert/main.cpp

@@ -0,0 +1,85 @@
+#include <timeplus/timeplus.h>
+
+#include <chrono>
+#include <iomanip>
+#include <iostream>
+
+using namespace timeplus;
+
+const size_t INSERT_BLOCKS = 100'000;
+
+const std::vector<std::pair<std::string, uint16_t>> HOST_PORTS = {
+    /// Single instance
+    {"localhost", 8463},
+    /// Cluster nodes
+    {"localhost", 18463},
+    {"localhost", 28463},
+    {"localhost", 38463},
+};
+
+void prepareTable() {
+    ClientOptions options;
+    for (const auto& [host, port] : HOST_PORTS) {
+        options.endpoints.push_back({host, port});
+    }
+
+    Client client{options};
+    client.Execute("DROP STREAM IF EXISTS insert_test;");
+    client.Execute("CREATE STREAM insert_test (i uint64, s string);");
+}
+
+auto timestamp() {
+    auto now = std::chrono::system_clock::now();
+    std::time_t now_time = std::chrono::system_clock::to_time_t(now);
+    std::tm* local_time = std::localtime(&now_time);
+    return std::put_time(local_time, "%Y-%m-%d %H:%M:%S");
+}
+
+int main() {
+    prepareTable();
+
+    TimeplusConfig config;
+    for (const auto& [host, port] : HOST_PORTS) {
+        config.client_options.endpoints.push_back({host, port});
+    }
+    config.max_connections = 1;
+    config.max_retries = 5;
+    config.task_executors = 0;
+    Timeplus tp{std::move(config)};
+
+    auto block = std::make_shared<Block>();
+    auto col_i = std::make_shared<ColumnUInt64>(std::vector<uint64_t>{5, 7, 4, 8});
+    auto col_s = std::make_shared<ColumnString>(
+        std::vector<std::string>{"Before my bed, the moon is bright,", "I think that it is frost on the ground.",
+                                 "I raise my head to gaze at the bright moon,", "And lower it to think of my hometown."});
+    block->AppendColumn("i", col_i);
+    block->AppendColumn("s", col_s);
+
+    auto start_time = std::chrono::high_resolution_clock::now();
+    auto last_time = start_time;
+    for (size_t i = 1; i <= INSERT_BLOCKS; ++i) {
+        while (true) {
+            try {
+                tp.Insert("insert_test", block);
+                break;
+            } catch (const std::exception& ex) {
+                std::cout << timestamp() << "\t Failed to insert block " << i << " : " << ex.what() << std::endl;
+            }
+        }
+
+        if (i % 1000 == 0) {
+            auto current_time = std::chrono::high_resolution_clock::now();
+            std::chrono::duration<double> elapsed = current_time - last_time;
+            last_time = current_time;
+            std::cout << "[" << timestamp() << "]\t" << i << " blocks inserted\telapsed = " << elapsed.count()
+                      << " sec\teps = " << 1000.0 * block->GetRowCount() / elapsed.count() << std::endl;
+        }
+    }
+
+    auto current_time = std::chrono::high_resolution_clock::now();
+    std::chrono::duration<double> elapsed = current_time - start_time;
+    std::cout << "\nInsert Done. Total Events = " << INSERT_BLOCKS * block->GetRowCount() << " Total Time = " << elapsed.count() << " sec"
+              << std::endl;
+
+    return 0;
+}