[to dev/1.3] C++ client: backport thread-safe SessionPool (#17800) (#17817)

* C++ client: add thread-safe SessionPool, enable RPC compression, and harden buffers (#17800)

* Wire RPC compression flag through Session to its connections

The enableRPCCompression option set via Session::open(bool) or the session
builder was never propagated to SessionConnection, whose flag was hardcoded
to false, so the compact Thrift protocol never took effect. Thread the flag
from the builder/open() into both the data SessionConnection and the
node-discovery NodesSupplier client so compression actually applies.

* Use snprintf for Tablet bounds-check error messages

Tablet::addValue and the OBJECT-value overload formatted out-of-range
diagnostics with sprintf into a fixed 100-byte stack buffer, risking an
overflow. Switch to snprintf bounded by sizeof(buffer) and cast the size_t
arguments to long to match the %ld format.

* Append big-endian bytes in MyStringBuffer instead of overwriting

On big-endian hosts MyStringBuffer::putOrderedByte used str.assign, which
replaced the whole buffer with each numeric write and corrupted previously
serialized content. Use str.append so bytes accumulate, matching the
little-endian path.

* Add thread-safe SessionPool to the C++ client

Introduce SessionPool and SessionPoolBuilder so multiple threads can share a
bounded set of connections without external locking. A single Session is not
safe to use concurrently, so the pool lends each Session to one borrower at a
time via an RAII PooledSession handle and reclaims it on scope exit. Sessions
are created outside the lock to avoid blocking other borrowers during the
handshake, and getSession() blocks up to a configurable timeout when the pool
is exhausted.

Query results are returned as a PooledSessionDataSet that keeps the Session
leased until the result set is fully read, since SessionDataSet lazily fetches
further blocks over the same connection. Connections that raise
IoTDBConnectionException are evicted rather than recycled. Add integration
tests covering basic borrow/insert/query, concurrent writers, and
exhaustion-timeout behavior.

* Reject zero maxSize in SessionPool instead of clamping to 1

Address review feedback: maxSize is size_t, so a non-positive check reduces
to == 0 (and "<= 0" would be a tautological-comparison warning under -Wall).
Rather than silently clamping an invalid 0 to 1, fail fast by throwing
IoTDBException so the misuse surfaces at construction time.

* Tolerate missing timeseries in SessionPool test cleanup

The pre-test cleanup deleted root.test.pool.* timeseries unconditionally,
which threw 508 (does not exist) on a fresh database and failed the new
[sessionPool] cases. Ignore that error since the cleanup is best-effort.

* Revert "Wire RPC compression flag through Session to its connections"

This reverts commit 2f35cc5.

Honoring the compression flag makes the client negotiate the compact Thrift
protocol, which the binary-only IoTDB server used by the C++ integration
tests cannot speak, breaking the pre-existing ts_session_open_with_compression
smoke test (it had only passed because the flag was a no-op). Compression
needs a compact-protocol-enabled test server, so it will be reintroduced in a
dedicated PR with the matching server-side test support. SessionPool keeps its
compression option for forward compatibility; it is currently a no-op, as the
rest of the client has always been.

* fix format

* Discard SessionPool session if pool closed during construction

Address review feedback: acquire() releases the lock while building a new
connection, so a concurrent close() could set closed_ after the slot was
reserved, and the freshly opened session would still be handed out from a
closed pool. Re-check closed_ under the lock after construction; if the pool
was closed meanwhile, release the slot, tear the session down outside the
lock, and throw instead of returning it.

* C++ client: adapt SessionPool to dev/1.3 Session/builder APIs

The cherry-picked SessionPool targeted master's expanded AbstractSessionBuilder
(DEFAULT_* constants plus nodeUrls/connectTimeoutMs/useSSL/trustCertFilePath
fields), none of which exist on dev/1.3, so the C++ client failed to compile.

Adapt only the newly added SessionPool.h/.cpp (no existing dev/1.3 interface or
implementation is changed):
- Replace AbstractSessionBuilder::DEFAULT_* with dev/1.3's literal defaults.
- Keep connectTimeoutMs (still applied via Session::open()).
- Drop the multi-node (nodeUrls) constructor and SSL options, since dev/1.3's
  Session(AbstractSessionBuilder*) wires neither and would silently ignore them.

Author: SpriCoder

iotdb-client/client-cpp/src/main/Common.cpp

@@ -384,7 +384,7 @@ const char* MyStringBuffer::getOrderedByte(size_t len) {
 
 void MyStringBuffer::putOrderedByte(char* buf, int len) {
     if (isBigEndian) {
-        str.assign(buf, len);
+        str.append(buf, len);
     }
     else {
         for (int i = len - 1; i > -1; i--) {

iotdb-client/client-cpp/src/main/Session.h

@@ -31,6 +31,7 @@
 #include <thread>
 #include <stdexcept>
 #include <cstdlib>
+#include <cstdio>
 #include <future>
 #include <boost/date_time/gregorian/gregorian.hpp>
 #include <thrift/protocol/TBinaryProtocol.h>
@@ -272,15 +273,17 @@ class Tablet {
     void addValue(size_t schemaId, size_t rowIndex, const T& value) {
         if (schemaId >= schemas.size()) {
             char tmpStr[100];
-            sprintf(tmpStr, "Tablet::addValue(), schemaId >= schemas.size(). schemaId=%ld, schemas.size()=%ld.",
-                    schemaId, schemas.size());
+            snprintf(tmpStr, sizeof(tmpStr),
+                     "Tablet::addValue(), schemaId >= schemas.size(). schemaId=%ld, schemas.size()=%ld.",
+                     (long)schemaId, (long)schemas.size());
             throw std::out_of_range(tmpStr);
         }
 
         if (rowIndex >= rowSize) {
             char tmpStr[100];
-            sprintf(tmpStr, "Tablet::addValue(), rowIndex >= rowSize. rowIndex=%ld, rowSize.size()=%ld.", rowIndex,
-                    rowSize);
+            snprintf(tmpStr, sizeof(tmpStr),
+                     "Tablet::addValue(), rowIndex >= rowSize. rowIndex=%ld, rowSize.size()=%ld.",
+                     (long)rowIndex, (long)rowSize);
             throw std::out_of_range(tmpStr);
         }
 

iotdb-client/client-cpp/src/main/SessionPool.cpp

@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "SessionPool.h"
+
+void PooledSession::reset() {
+  if (session_ && pool_ != nullptr) {
+    pool_->putBack(session_, broken_);
+  }
+  pool_ = nullptr;
+  session_ = nullptr;
+  broken_ = false;
+}
+
+SessionPool::SessionPool(std::string host, int rpcPort, std::string username, std::string password,
+                         size_t maxSize)
+    : host_(std::move(host)), rpcPort_(rpcPort), username_(std::move(username)),
+      password_(std::move(password)), maxSize_(maxSize) {
+  if (maxSize_ == 0) {
+    throw IoTDBException("SessionPool maxSize must be greater than 0.");
+  }
+}
+
+SessionPool::~SessionPool() {
+  try {
+    close();
+  } catch (const std::exception& e) {
+    log_debug(std::string("SessionPool::~SessionPool(), ") + e.what());
+  }
+}
+
+SessionPool& SessionPool::setFetchSize(int fetchSize) {
+  fetchSize_ = fetchSize;
+  return *this;
+}
+
+SessionPool& SessionPool::setZoneId(std::string zoneId) {
+  zoneId_ = std::move(zoneId);
+  return *this;
+}
+
+SessionPool& SessionPool::setSqlDialect(std::string sqlDialect) {
+  sqlDialect_ = std::move(sqlDialect);
+  return *this;
+}
+
+SessionPool& SessionPool::setDatabase(std::string database) {
+  database_ = std::move(database);
+  return *this;
+}
+
+SessionPool& SessionPool::setEnableRedirection(bool enable) {
+  enableRedirection_ = enable;
+  return *this;
+}
+
+SessionPool& SessionPool::setEnableAutoFetch(bool enable) {
+  enableAutoFetch_ = enable;
+  return *this;
+}
+
+SessionPool& SessionPool::setEnableRPCCompression(bool enable) {
+  enableRPCCompression_ = enable;
+  return *this;
+}
+
+SessionPool& SessionPool::setConnectTimeoutMs(int connectTimeoutMs) {
+  connectTimeoutMs_ = connectTimeoutMs;
+  return *this;
+}
+
+SessionPool& SessionPool::setWaitToGetSessionTimeoutMs(int64_t timeoutMs) {
+  waitTimeoutMs_ = timeoutMs;
+  return *this;
+}
+
+std::shared_ptr<Session> SessionPool::constructNewSession() {
+  AbstractSessionBuilder builder;
+  builder.host = host_;
+  builder.rpcPort = rpcPort_;
+  builder.username = username_;
+  builder.password = password_;
+  builder.zoneId = zoneId_;
+  builder.fetchSize = fetchSize_;
+  builder.sqlDialect = sqlDialect_;
+  builder.database = database_;
+  builder.enableAutoFetch = enableAutoFetch_;
+  builder.enableRedirections = enableRedirection_;
+  builder.enableRPCCompression = enableRPCCompression_;
+
+  auto session = std::make_shared<Session>(&builder);
+  session->open(enableRPCCompression_, connectTimeoutMs_);
+  return session;
+}
+
+std::shared_ptr<Session> SessionPool::acquire(int64_t timeoutMs) {
+  const int64_t effectiveTimeout = timeoutMs > 0 ? timeoutMs : waitTimeoutMs_;
+  std::unique_lock<std::mutex> lock(mutex_);
+  const auto deadline =
+      std::chrono::steady_clock::now() + std::chrono::milliseconds(effectiveTimeout);
+
+  while (true) {
+    if (closed_) {
+      throw IoTDBException("SessionPool is closed.");
+    }
+    if (!idleQueue_.empty()) {
+      auto session = idleQueue_.front();
+      idleQueue_.pop_front();
+      return session;
+    }
+    if (size_ < maxSize_) {
+      // Reserve a slot, then build the connection outside the lock so other
+      // borrowers are not blocked by network/handshake latency.
+      ++size_;
+      lock.unlock();
+      std::shared_ptr<Session> session;
+      try {
+        session = constructNewSession();
+      } catch (...) {
+        lock.lock();
+        --size_;
+        cv_.notify_one();
+        throw;
+      }
+      lock.lock();
+      if (closed_) {
+        // The pool was closed while this session was being built; do not hand it
+        // out. Release its slot and let it be torn down outside the lock.
+        --size_;
+        lock.unlock();
+        throw IoTDBException("SessionPool is closed.");
+      }
+      return session;
+    }
+
+    // Pool exhausted: wait for a Session to be returned.
+    if (effectiveTimeout <= 0) {
+      cv_.wait(lock);
+    } else {
+      if (cv_.wait_until(lock, deadline) == std::cv_status::timeout && idleQueue_.empty() &&
+          size_ >= maxSize_ && !closed_) {
+        throw IoTDBException(
+            "Wait to get session timeout in SessionPool, maxSize=" + std::to_string(maxSize_) +
+            ", waitTimeoutMs=" + std::to_string(effectiveTimeout) + ".");
+      }
+    }
+  }
+}
+
+void SessionPool::putBack(const std::shared_ptr<Session>& session, bool broken) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  if (broken || closed_) {
+    // Drop the Session and free its slot so a healthy replacement can be created
+    // on demand. The caller (PooledSession::reset) still holds the last reference
+    // and tears the connection down after we return, i.e. outside this lock.
+    --size_;
+  } else {
+    idleQueue_.push_back(session);
+  }
+  cv_.notify_one();
+}
+
+PooledSession SessionPool::getSession() {
+  return getSession(waitTimeoutMs_);
+}
+
+PooledSession SessionPool::getSession(int64_t timeoutMs) {
+  return PooledSession(this, acquire(timeoutMs));
+}
+
+void SessionPool::insertTablet(Tablet& tablet, bool sorted) {
+  execute([&](Session& s) { s.insertTablet(tablet, sorted); });
+}
+
+void SessionPool::insertAlignedTablet(Tablet& tablet, bool sorted) {
+  execute([&](Session& s) { s.insertAlignedTablet(tablet, sorted); });
+}
+
+void SessionPool::insertTablets(std::unordered_map<std::string, Tablet*>& tablets, bool sorted) {
+  execute([&](Session& s) { s.insertTablets(tablets, sorted); });
+}
+
+void SessionPool::insertRecord(const std::string& deviceId, int64_t time,
+                               const std::vector<std::string>& measurements,
+                               const std::vector<std::string>& values) {
+  execute([&](Session& s) { s.insertRecord(deviceId, time, measurements, values); });
+}
+
+void SessionPool::insertRecords(const std::vector<std::string>& deviceIds,
+                                const std::vector<int64_t>& times,
+                                const std::vector<std::vector<std::string>>& measurementsList,
+                                const std::vector<std::vector<std::string>>& valuesList) {
+  execute([&](Session& s) { s.insertRecords(deviceIds, times, measurementsList, valuesList); });
+}
+
+void SessionPool::executeNonQueryStatement(const std::string& sql) {
+  execute([&](Session& s) { s.executeNonQueryStatement(sql); });
+}
+
+PooledSessionDataSet SessionPool::executeQueryStatement(const std::string& sql) {
+  PooledSession lease = getSession();
+  try {
+    auto dataSet = lease->executeQueryStatement(sql);
+    return PooledSessionDataSet(std::move(lease), std::move(dataSet));
+  } catch (const IoTDBConnectionException&) {
+    lease.markBroken();
+    throw;
+  }
+}
+
+PooledSessionDataSet SessionPool::executeQueryStatement(const std::string& sql,
+                                                        int64_t timeoutInMs) {
+  PooledSession lease = getSession();
+  try {
+    auto dataSet = lease->executeQueryStatement(sql, timeoutInMs);
+    return PooledSessionDataSet(std::move(lease), std::move(dataSet));
+  } catch (const IoTDBConnectionException&) {
+    lease.markBroken();
+    throw;
+  }
+}
+
+void SessionPool::close() {
+  std::deque<std::shared_ptr<Session>> toClose;
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (closed_) {
+      return;
+    }
+    closed_ = true;
+    toClose.swap(idleQueue_);
+    size_ -= toClose.size();
+  }
+  cv_.notify_all();
+  // Sessions destructed here (outside the lock) close their connections.
+  toClose.clear();
+}
+
+size_t SessionPool::activeCount() {
+  std::lock_guard<std::mutex> lock(mutex_);
+  return size_ - idleQueue_.size();
+}