add Timeplus APIs #18

yuzifeng1984 · 2024-08-12T14:23:15Z

Idempotent insert
Thread-safe insert APIs
More better controlled retry on exception.

yuzifeng1984 · 2024-08-12T14:25:42Z

timeplus/blocking_queue.h

@@ -0,0 +1,243 @@
+#pragma once


cloned from proton

tests/insert-async/main.cpp

timeplus/timeplus.cpp

chenziliang · 2024-08-13T05:25:15Z

timeplus/timeplus.h

+
+namespace timeplus {
+
+struct TimeplusConfig {


TimeplusConfig can contain a ClientConfig directly ? would this simplify code ?

We overwrite some client config items and not want user to change it. So only expose a portion of the client parameters to config. The draw back is user could not change the unexposed parameters (such as ssl).

I am thinking of what is the best way to do this...

refer to https://github.com/timeplus-io/timeplus-cpp/pull/18/files#diff-cab949bec8607457ecea447bb6500ef4f75df665c6b193a77a12afce1a48417fR74

PS. segregate user from directly config client might make future extension easier, (eg. auto node host/port discovery)

chenziliang · 2024-08-13T05:27:27Z

timeplus/timeplus.h

+    std::string password;
+
+    /// Max number of connections maintained in pool.
+    unsigned int max_connections = 1;


let's stick to uint32_t etc new style

Is this per host_port max_connections ?

It is the total number of clients in the pool (so at-most max_connections will be created). Each client will round-robin pick the host/port when connect/re-connect, starting from the first host/port.

chenziliang · 2024-08-14T06:23:14Z

tests/insert-async/main.cpp

+    block.AppendColumn("i", col_i);
+    block.AppendColumn("s", col_s);
+
+    class Inserter {


Let's remove this class and use a plain loop in the main which could be way more direct usage

chenziliang · 2024-08-14T06:25:45Z

tests/insert-async/main.cpp

+        inserter.InsertBlock();
+    }
+
+    while (inserter.BlockInserted() != INSERT_BLOCKS) {


We can have a bit advanced usage by using atomic wait https://en.cppreference.com/w/cpp/atomic/atomic/wait instead of spin

currently sdk is based on C++17. I will add notification later.

chenziliang · 2024-08-14T06:27:26Z

tests/insert-async/main.cpp

+                    }
+
+                } else {
+                    std::cout << "[" << timestamp() << "]\t Failed to insert block: insert_id=" << id << std::endl;


Do we have a fail message / reason ?

added to print out result.err_msg

chenziliang · 2024-08-14T06:28:41Z

tests/insert-async/main.cpp

+        }
+
+        void InsertBlock() {
+            tp_.InsertAsync(table_name_, block_, [this](uint64_t id, const auto& result) {


If the id is passed via callback, what can users do with it ?

Removed the internal id and changed the insert APIs in timeplus.h
The InsertResult contains the original parameters for user convinience.

auto handle_insert_result = [&](size_t user_block_id, const InsertResult& result) {...} tp.InsertAsync("table", block, [user_block_id, &handle_insert_result](const BaseResult& result) { const auto& insert_result = static_cast<const InsertResult&>(result); handle_insert_result(block_id, insert_result); });

chenziliang · 2024-08-14T06:35:31Z

timeplus/client.cpp

-    Query query("INSERT INTO " + table_name + " ( " + fields_section.str() + " ) VALUES", query_id);
+    std::string settings;
+    if (!idempotent_id.empty()) {
+        settings = " SETTINGS idempotent_id='" + idempotent_id + "'";


a bit more perf

settings.reserve(128); settings.append(" SETTINGS idempotent_id='").append(idempotent_id).append("'");

chenziliang · 2024-08-14T06:36:18Z

timeplus/client.cpp

+        settings = " SETTINGS idempotent_id='" + idempotent_id + "'";
+    }
+
+    Query query("INSERT INTO " + table_name + " ( " + fields_section.str() + " )" + settings + " VALUES", query_id);


Similarly, let's try to minimize string memory allocation

This in general seems very verbose string concatenation and lower perf, it is not even using native wire protocol ?

Changed to use the query settings serialization.

chenziliang · 2024-08-14T06:41:14Z

timeplus/timeplus.h

+
+using InsertResult = BaseResult;
+
+using Callback = std::function<void(uint64_t id, const BaseResult&)>;


If we provide an id for the callback, what users can do with the id ? It seems too late.

changed the insert arguments and return. please check the update.

chenziliang · 2024-08-14T06:42:43Z

timeplus/timeplus.h

+
+namespace timeplus {
+
+struct TimeplusConfig {


nite : we can consider creating a separate timeplus_config.h file

chenziliang · 2024-08-14T06:45:23Z

timeplus/client_pool.h

+
+    std::optional<std::pair<ClientPtr, bool>> Acquire(int64_t timeout_ms);
+
+    void Release(ClientPtr& client, bool valid) {


let's take ClientPtr since it will be more convenient and perf-wise, it is still fine and caller can std::move(...) here

chenziliang · 2024-08-14T06:47:12Z

timeplus/client_pool.h

+
+    ~ClientPool() { assert(clients_.size() == pool_size_); }
+
+    std::optional<std::pair<ClientPtr, bool>> Acquire(int64_t timeout_ms);


Do we need return a bool ? Can acquire return a valid ClientPtr ?

I was thinking to distinct the cases between blocking queue is empty and the returned client cannot connect within timeout.

I changed the behavior to throw different exceptions when could not acquire the client.

timeplus/client_pool.h

chenziliang · 2024-08-14T06:52:12Z

timeplus/client_pool.cpp

+
+    auto& [client, valid] = maybe_client.value();
+    try {
+        /// Lazy init client


we can init the client greedy-ly

Client init will create connection to server which is relative heavy. May take even longer time when timeout. So put the init where user is using the it.

The challenge with lazy init is we don't have a good way to report back, oh, your hostname port configure is wrong, we can't connect. For me an initial connect to report this configuration issue (which happens often) is good

timeplus/timeplus.cpp

chenziliang · 2024-08-14T06:55:26Z

timeplus/timeplus.cpp

+        auto task_id = next_task_id_.fetch_add(1);
+        auto task = std::make_shared<InsertTask>();
+        task->task_id = task_id;
+        task->table_name = std::move(table_name);


task->table_name.swap(table_name) task->block.swap(block) task->idempotent_id.swap(idempotent_id);

Or provide a construct to InsertTask which takes these && parameters there

added a constructor

timeplus/timeplus.cpp

timeplus/client.h

timeplus/timeplus.cpp

chenziliang · 2024-08-14T17:11:45Z

timeplus/client.cpp

-        settings = " SETTINGS idempotent_id='" + idempotent_id + "'";
+    Query query("INSERT INTO " + table_name + " ( " + fields_section.str() + " ) VALUES", query_id);
+
+    if (!idempotent_id.empty() && versionNumber(server_info_.version_major, server_info_.version_minor, server_info_.version_patch) >=


Can we cache version number ? and in our case probably we don't need this version number guard since all of our production enterprise users have version number bigger than this ?

remove the server version check for idempotent insert

timeplus/timeplus.cpp

tests/insert-async/main.cpp

timeplus/client.cpp

chenziliang · 2024-08-14T17:36:18Z

timeplus/client.h

-    void Insert(const std::string& table_name, const Block& block);
-    void Insert(const std::string& table_name, const std::string& query_id, const Block& block);
+    /// Insertion will be idempotent when `idempotent_id` is not empty.
+    void Insert(const std::string& table_name, const Block& block, const std::string & idempotent_id = "");


it seems it is good to introduce BlockPtr which is a std::shared_ptr<const Block> in this way we don't need copy block during ingest and when retry

BTW, the current legacy interface / implementation of this SDK is not necessary optimal, we can revise when necessary

I change the new timeplus APIs to receive BlockPtr; while keep the legacy client interface since it does not need to own the block for usage and performance consideration. so user has more flexibility in using client::insert

Does legacy client copy the Block internally ?

just double checked and no copy as far as I saw. all sub-function access/keep the block as const reference.

chenziliang · 2024-08-15T04:34:02Z

timeplus/client.cpp

+    size_t query_text_size = INSERT_INTO.size() + table_name.size() + LEFT_PAREN.size() + fields_section_size + RIGHT_PAREN_VALUES.size();
+
+    std::string query_text;
+    query_text.reserve(query_text_size);


we can do query_text_size * 1.2 for example for safety

changed.

also refactor TimeplusConfig to allow user directly set client options.

yuzifeng1984 requested a review from chenziliang August 12, 2024 14:23

yuzifeng1984 commented Aug 12, 2024

View reviewed changes

timeplus/blocking_queue.h

@@ -0,0 +1,243 @@

#pragma once

Copy link

Contributor Author

yuzifeng1984 Aug 12, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cloned from proton

yuzifeng1984 force-pushed the feature/huatai-requirements branch from 388c528 to 7568fe7 Compare August 12, 2024 14:29

chenziliang reviewed Aug 13, 2024

View reviewed changes

tests/insert-async/main.cpp Show resolved Hide resolved

chenziliang reviewed Aug 13, 2024

View reviewed changes

timeplus/timeplus.cpp Show resolved Hide resolved

chenziliang reviewed Aug 13, 2024

View reviewed changes