feat(cpp/python): expose the parallel read/write switches

cpp/README.md

@@ -340,14 +340,26 @@ cmake .. -DENABLE_THREADS=OFF  # disable — all thread code is stripped at comp
 ```cpp
 #include "common/global.h"
 
-// Enable or disable parallel write at runtime (auto-disabled on single-core machines)
+// Enable or disable the parallel write / read paths at runtime.  The flags are
+// consulted on every write and when a query opens its iterators.
 storage::set_parallel_write_enabled(true);
+storage::set_parallel_read_enabled(true);
 
-// Set the number of worker threads (must be called before creating TsFileWriter)
-storage::set_write_thread_count(4);
+// Resize the global worker pool (1..64).  Takes effect immediately when the
+// library has been initialized; do not call it while any reader, writer or
+// result set is in use.
+storage::set_thread_count(4);
 ```
 
-By default, parallel write is enabled when the machine has more than one CPU core, and the thread count is set to the number of hardware cores (capped at 64).
+The C API exposes the same switches as
+`tsfile_set_parallel_read_enabled()`, `tsfile_set_parallel_write_enabled()` and
+`tsfile_set_thread_count()` (plus the matching getters); the Python binding
+has `set_parallel_read_enabled()`, `set_parallel_write_enabled()` and
+`set_thread_count()`.
+
+Parallel read and write are enabled by default with a 6-thread pool.  With
+`ENABLE_THREADS=OFF` the setters only record the values and the parallel paths
+are compiled out.
 
 ### Local File Read Backend
 

cpp/src/common/config/config.h

@@ -53,8 +53,12 @@ typedef struct ConfigValue {
     TSEncoding double_encoding_type_;
     TSEncoding string_encoding_type_;
     CompressionType default_compression_type_;
-    bool parallel_read_enabled_;
-    bool parallel_write_enabled_;
+    // Unlike the fields above, these two keep their in-class default instead
+    // of being assigned by init_config_value(): a caller may configure the
+    // parallel paths before the first reader/writer triggers lazy library
+    // initialization, and the init-time assignment would silently reset them.
+    bool parallel_read_enabled_ = true;
+    bool parallel_write_enabled_ = true;
     // Size of the single global worker pool (common::g_thread_pool_) shared by
     // the parallel write and parallel read paths.  The pool is (re)created from
     // this value in init_common().  Like sync_on_close_/encrypt_flag_ it keeps

cpp/src/common/global.cc

@@ -20,6 +20,8 @@
 #include "global.h"
 
 #include <atomic>
+#include <exception>
+#include <memory>
 
 #ifdef ENABLE_THREADS
 #include "common/thread_pool.h"
@@ -86,10 +88,9 @@ void init_config_value() {
 #else
     g_config_value_.default_compression_type_ = UNCOMPRESSED;
 #endif
-    g_config_value_.parallel_read_enabled_ = true;
-    g_config_value_.parallel_write_enabled_ = true;
-    // thread_count_ keeps its in-class default (see config.h) so a
-    // set_thread_count() before libtsfile_init() is not reset here.
+    // parallel_read_enabled_ / parallel_write_enabled_ / thread_count_ keep
+    // their in-class defaults (see config.h) so setters called before
+    // libtsfile_init() are not reset here.
 }
 
 extern TSEncoding get_value_encoder(TSDataType data_type) {
@@ -178,19 +179,35 @@ int init_common() {
 
 int set_thread_count(int32_t count) {
     if (count < 1 || count > 64) return E_INVALID_ARG;
-    g_config_value_.thread_count_ = count;
 #ifdef ENABLE_THREADS
     // If the global pool already exists (libtsfile_init has run) rebuild it at
     // the new size so the change takes effect immediately instead of only at
     // the next libtsfile_init().  This joins all current workers and recreates
-    // them, so the caller must ensure no read/write is concurrently using the
-    // pool — intended for setup / benchmark reconfiguration, not mid-operation
-    // resizing.
+    // them, so the caller must ensure no reader, writer or result set is using
+    // the pool — intended for setup / benchmark reconfiguration, not
+    // mid-operation resizing.
     if (g_thread_pool_ != nullptr) {
-        delete g_thread_pool_;
-        g_thread_pool_ = new ThreadPool(static_cast<size_t>(count));
+        if (count == g_config_value_.thread_count_) {
+            return E_OK;  // already this size; do not join and respawn workers
+        }
+        // Build the replacement before touching the old pool so a failure
+        // (thread creation or allocation) leaves the old one intact, and never
+        // propagate an exception across the C API boundary.
+        std::unique_ptr<ThreadPool> new_pool;
+        try {
+            new_pool.reset(new ThreadPool(static_cast<size_t>(count)));
+        } catch (...) {
+            // Nothing may escape into the C API; keep the old pool intact.
+            return E_OOM;
+        }
+        ThreadPool* old_pool = g_thread_pool_;
+        g_thread_pool_ = new_pool.release();
+        g_config_value_.thread_count_ = count;
+        delete old_pool;
+        return E_OK;
     }
 #endif
+    g_config_value_.thread_count_ = count;
     return E_OK;
 }
 

cpp/src/cwrapper/tsfile_cwrapper.cc

@@ -169,6 +169,32 @@ TsFileReadBackend tsfile_get_file_read_backend() {
     return static_cast<TsFileReadBackend>(common::get_file_read_backend());
 }
 
+ERRNO tsfile_set_parallel_read_enabled(int32_t enabled) {
+    common::set_parallel_read_enabled(enabled != 0);
+    return common::E_OK;
+}
+
+int tsfile_get_parallel_read_enabled() {
+    return common::get_parallel_read_enabled() ? 1 : 0;
+}
+
+ERRNO tsfile_set_parallel_write_enabled(int32_t enabled) {
+    common::set_parallel_write_enabled(enabled != 0);
+    return common::E_OK;
+}
+
+int tsfile_get_parallel_write_enabled() {
+    return common::get_parallel_write_enabled() ? 1 : 0;
+}
+
+ERRNO tsfile_set_thread_count(int32_t count) {
+    return common::set_thread_count(count);
+}
+
+int32_t tsfile_get_thread_count() {
+    return common::g_config_value_.thread_count_;
+}
+
 WriteFile write_file_new(const char* pathname, ERRNO* err_code) {
     if (err_code == nullptr) {
         return nullptr;

cpp/src/cwrapper/tsfile_cwrapper.h

@@ -357,6 +357,54 @@ ERRNO tsfile_set_file_read_backend(int32_t backend);
 /** @return The backend configured for subsequently opened readers. */
 TsFileReadBackend tsfile_get_file_read_backend(void);
 
+/**
+ * @brief Enable/disable the parallel read path.
+ *
+ * The switch is consulted when a query opens its series iterators, so it
+ * applies to every subsequent read, not only to readers opened afterwards.
+ * The flags and the pool are read without synchronization: set them before
+ * starting queries/writes, not while another thread is reading or writing.
+ *
+ * @param enabled Non-zero to enable, zero to disable.
+ * @return RET_OK on success.
+ */
+ERRNO tsfile_set_parallel_read_enabled(int32_t enabled);
+
+/** @return Whether the parallel read path is enabled. */
+int tsfile_get_parallel_read_enabled(void);
+
+/**
+ * @brief Enable/disable the parallel write path.
+ *
+ * Consulted on every write, not only by writers opened afterwards.  Same
+ * concurrency rule as tsfile_set_parallel_read_enabled().
+ *
+ * @param enabled Non-zero to enable, zero to disable.
+ * @return RET_OK on success.
+ */
+ERRNO tsfile_set_parallel_write_enabled(int32_t enabled);
+
+/** @return Whether the parallel write path is enabled. */
+int tsfile_get_parallel_write_enabled(void);
+
+/**
+ * @brief Set the size of the global worker thread pool.
+ *
+ * Takes effect immediately when the library has been initialized (an existing
+ * pool is joined and recreated).  Readers and writers borrow the pool, so this
+ * must not be called while any reader, writer or result set is in use — it is
+ * meant for setup / benchmark reconfiguration.  With ENABLE_THREADS off this
+ * only records the value.
+ *
+ * @param count Number of worker threads, in [1, 64].
+ * @return RET_OK on success, RET_INVALID_ARG for an out-of-range count, or
+ * RET_OOM if the new pool could not be created (the old one is kept).
+ */
+ERRNO tsfile_set_thread_count(int32_t count);
+
+/** @return The configured worker thread count (1..64). */
+int32_t tsfile_get_thread_count(void);
+
 /**
  * @brief Get the encoding type for global time column
  *

cpp/test/cwrapper/cwrapper_test.cc

@@ -25,11 +25,17 @@
 #include <utils/db_utils.h>
 
 #include <cstring>
+#include <limits>
+#include <string>
+#include <vector>
 
+#include "common/global.h"
 #include "common/row_record.h"
 #include "cwrapper/tsfile_cwrapper.h"
+#include "file/write_file.h"
 #include "reader/result_set.h"
 #include "reader/tsfile_reader.h"
+#include "writer/tsfile_table_writer.h"
 #include "writer/tsfile_writer.h"
 
 namespace storage {
@@ -66,6 +72,180 @@ TEST_F(CWrapperTest, FileReadBackendConfigurationRoundTripsAndValidates) {
     EXPECT_EQ(tsfile_set_file_read_backend(original), RET_OK);
 }
 
+TEST_F(CWrapperTest, ParallelTogglesAndThreadCountValidate) {
+    const bool read_enabled = common::get_parallel_read_enabled();
+    const bool write_enabled = common::get_parallel_write_enabled();
+    const int32_t thread_count = common::g_config_value_.thread_count_;
+
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(0), RET_OK);
+    EXPECT_FALSE(common::get_parallel_read_enabled());
+    EXPECT_EQ(0, tsfile_get_parallel_read_enabled());
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(1), RET_OK);
+    EXPECT_TRUE(common::get_parallel_read_enabled());
+    EXPECT_EQ(1, tsfile_get_parallel_read_enabled());
+    // Any non-zero value enables the path, matching the bool parameter.
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(42), RET_OK);
+    EXPECT_TRUE(common::get_parallel_read_enabled());
+
+    EXPECT_EQ(tsfile_set_parallel_write_enabled(0), RET_OK);
+    EXPECT_FALSE(common::get_parallel_write_enabled());
+    EXPECT_EQ(0, tsfile_get_parallel_write_enabled());
+    EXPECT_EQ(tsfile_set_parallel_write_enabled(1), RET_OK);
+    EXPECT_TRUE(common::get_parallel_write_enabled());
+    EXPECT_EQ(1, tsfile_get_parallel_write_enabled());
+
+    // Out-of-range counts are rejected and leave the configured size alone.
+    for (int32_t invalid : {0, -1, 65, 1000}) {
+        EXPECT_EQ(tsfile_set_thread_count(invalid), RET_INVALID_ARG) << invalid;
+        EXPECT_EQ(common::g_config_value_.thread_count_, thread_count)
+            << invalid;
+        EXPECT_EQ(tsfile_get_thread_count(), thread_count) << invalid;
+    }
+    for (int32_t valid : {1, 32, 64}) {
+        EXPECT_EQ(tsfile_set_thread_count(valid), RET_OK) << valid;
+        EXPECT_EQ(common::g_config_value_.thread_count_, valid) << valid;
+        EXPECT_EQ(tsfile_get_thread_count(), valid) << valid;
+    }
+
+    EXPECT_EQ(tsfile_set_thread_count(thread_count), RET_OK);
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(read_enabled), RET_OK);
+    EXPECT_EQ(tsfile_set_parallel_write_enabled(write_enabled), RET_OK);
+}
+
+// The switches are read by the library's lazy initialization.  Setting them
+// before the first reader/writer used to be undone: init_config_value()
+// assigned both flags true, silently re-enabling the parallel paths.
+TEST_F(CWrapperTest, ParallelTogglesSurviveLazyInitialization) {
+    storage::libtsfile_destroy();
+    const std::string path = "cwrapper_toggles_before_init.tsfile";
+    remove(path.c_str());
+
+    ASSERT_EQ(tsfile_set_parallel_read_enabled(0), RET_OK);
+    ASSERT_EQ(tsfile_set_parallel_write_enabled(0), RET_OK);
+
+    ERRNO code = RET_OK;
+    WriteFile file = write_file_new(path.c_str(), &code);
+    ASSERT_EQ(RET_OK, code);
+    ASSERT_NE(nullptr, file);
+
+    EXPECT_EQ(0, tsfile_get_parallel_read_enabled());
+    EXPECT_EQ(0, tsfile_get_parallel_write_enabled());
+
+    free_write_file(&file);
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(1), RET_OK);
+    EXPECT_EQ(tsfile_set_parallel_write_enabled(1), RET_OK);
+    remove(path.c_str());
+}
+
+// Flipping the parallel read switch must not change what a query returns; it
+// only selects whether the aligned multi-column read uses the worker pool.
+TEST_F(CWrapperTest, ParallelReadToggleReturnsIdenticalRows) {
+    storage::libtsfile_init();
+    const std::string path = "cwrapper_parallel_read_toggle.tsfile";
+    remove(path.c_str());
+    constexpr int kDevices = 2;
+    constexpr int kRowsPerDevice = 500;
+    constexpr int kFields = 5;
+    {
+        storage::WriteFile file;
+        int flags = O_WRONLY | O_CREAT | O_TRUNC;
+#ifdef _WIN32
+        flags |= O_BINARY;
+#endif
+        ASSERT_EQ(file.create(path, flags, 0666), common::E_OK);
+        std::vector<common::ColumnSchema> columns;
+        columns.emplace_back("device", common::STRING, common::UNCOMPRESSED,
+                             common::PLAIN, common::ColumnCategory::TAG);
+        for (int c = 0; c < kFields; ++c) {
+            columns.emplace_back("s" + std::to_string(c), common::INT64,
+                                 common::UNCOMPRESSED, common::PLAIN,
+                                 common::ColumnCategory::FIELD);
+        }
+        storage::TableSchema schema("t", columns);
+        storage::TsFileTableWriter writer(&file, &schema);
+        storage::Tablet tablet(
+            "t", schema.get_measurement_names(), schema.get_data_types(),
+            schema.get_column_categories(), kDevices * kRowsPerDevice);
+        int index = 0;
+        for (int device = 0; device < kDevices; ++device) {
+            const std::string tag = "d" + std::to_string(device);
+            for (int row = 0; row < kRowsPerDevice; ++row, ++index) {
+                ASSERT_EQ(tablet.add_timestamp(index, row), common::E_OK);
+                ASSERT_EQ(tablet.add_value(index, "device", tag.c_str()),
+                          common::E_OK);
+                for (int c = 0; c < kFields; ++c) {
+                    const std::string name = "s" + std::to_string(c);
+                    ASSERT_EQ(
+                        tablet.add_value(index, name, int64_t(row) * 10 + c),
+                        common::E_OK);
+                }
+            }
+        }
+        ASSERT_EQ(writer.get_error(), common::E_OK);
+        ASSERT_EQ(writer.write_table(tablet), common::E_OK);
+        ASSERT_EQ(writer.flush(), common::E_OK);
+        ASSERT_EQ(writer.close(), common::E_OK);
+    }
+
+    std::vector<std::string> names{"device", "s0", "s1", "s2", "s3", "s4"};
+    std::vector<char*> name_ptrs;
+    for (auto& name : names) name_ptrs.push_back(&name[0]);
+    auto read_all = [&]() {
+        ERRNO code = RET_OK;
+        TsFileReader reader = tsfile_reader_new(path.c_str(), &code);
+        EXPECT_EQ(RET_OK, code);
+        if (reader == nullptr) {
+            return std::vector<std::string>();
+        }
+        ResultSet result_set =
+            tsfile_query_table(reader, "t", name_ptrs.data(),
+                               static_cast<uint32_t>(name_ptrs.size()), 0,
+                               std::numeric_limits<int64_t>::max(), &code);
+        EXPECT_EQ(RET_OK, code);
+        std::vector<std::string> rows;
+        if (result_set == nullptr) {
+            tsfile_reader_close(reader);
+            return rows;
+        }
+        while (tsfile_result_set_next(result_set, &code) && code == RET_OK) {
+            char* tag = tsfile_result_set_get_value_by_name_string(result_set,
+                                                                   "device");
+            std::string line =
+                tag == nullptr ? std::string() : std::string(tag);
+            free(tag);
+            line += ":" +
+                    std::to_string(tsfile_result_set_get_value_by_name_int64_t(
+                        result_set, "time"));
+            for (int c = 0; c < kFields; ++c) {
+                const std::string name = "s" + std::to_string(c);
+                line += ":" + std::to_string(
+                                  tsfile_result_set_get_value_by_name_int64_t(
+                                      result_set, name.c_str()));
+            }
+            rows.push_back(line);
+        }
+        EXPECT_EQ(RET_OK, code);
+        tsfile_reader_close(reader);
+        free_tsfile_result_set(&result_set);
+        return rows;
+    };
+
+    const bool saved = common::get_parallel_read_enabled();
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(0), RET_OK);
+    const std::vector<std::string> serial_rows = read_all();
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(1), RET_OK);
+    const std::vector<std::string> parallel_rows = read_all();
+    EXPECT_EQ(tsfile_set_parallel_read_enabled(saved), RET_OK);
+
+    ASSERT_EQ(static_cast<size_t>(kDevices * kRowsPerDevice),
+              serial_rows.size());
+    EXPECT_EQ(serial_rows, parallel_rows);
+    EXPECT_EQ("d0:0:0:1:2:3:4", serial_rows.front());
+    EXPECT_EQ("d1:499:4990:4991:4992:4993:4994", serial_rows.back());
+    remove(path.c_str());
+    storage::libtsfile_destroy();
+}
+
 TEST_F(CWrapperTest, CodecAndCompressionConfigIncludesJavaIds) {
     uint8_t old_int32_encoding = get_datatype_encoding(TS_DATATYPE_INT32);
     uint8_t old_int64_encoding = get_datatype_encoding(TS_DATATYPE_INT64);