diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt index 7ff07a786..c39bcbc28 100755 --- a/cpp/CMakeLists.txt +++ b/cpp/CMakeLists.txt @@ -229,6 +229,9 @@ endif () option(BUILD_TOOLS "Build the tsfile command-line tools" ON) message("cmake using: BUILD_TOOLS=${BUILD_TOOLS}") +option(BUILD_BENCHMARK "Build the read-backend benchmark" OFF) +message("cmake using: BUILD_BENCHMARK=${BUILD_BENCHMARK}") + option(ENABLE_ANTLR4 "Enable ANTLR4 runtime" ON) message("cmake using: ENABLE_ANTLR4=${ENABLE_ANTLR4}") @@ -378,6 +381,15 @@ if (NOT "${_TSFILE_PROJECT_DEPENDENCIES}" STREQUAL "") endif () add_subdirectory(src) +if (BUILD_BENCHMARK) + find_package(Threads REQUIRED) + add_executable(read_backend_benchmark + bench_mark/bench_mark_src/read_backend_benchmark.cc) + target_include_directories(read_backend_benchmark PRIVATE + ${PROJECT_SRC_DIR}) + target_link_libraries(read_backend_benchmark PRIVATE + tsfile Threads::Threads) +endif () if (BUILD_TOOLS) add_subdirectory(tools) endif () diff --git a/cpp/README-zh.md b/cpp/README-zh.md index b0d76f506..c1eb1d71c 100644 --- a/cpp/README-zh.md +++ b/cpp/README-zh.md @@ -173,6 +173,24 @@ storage::set_write_thread_count(4); 默认情况下，当机器 CPU 核数大于 1 时自动启用并行写入，线程数设为硬件核数（上限 64）。 +### 本地文件读取后端 + +Reader 可以为本地文件选择内存映射 I/O 或传统的定位读取路径。配置会在 +reader 打开文件时确定，因此修改配置不会影响已经打开的 reader。 + +```cpp +#include "common/global.h" + +common::set_file_read_backend(common::FileReadBackend::AUTO); // 默认值 +common::set_file_read_backend(common::FileReadBackend::MMAP); // 必须使用 mmap +common::set_file_read_backend(common::FileReadBackend::PREAD); // 兼容旧读取路径 +``` + +C API 可通过 `tsfile_set_file_read_backend(TSFILE_READ_BACKEND_*)` 设置相同 +选项。`AUTO` 会优先映射受支持的普通文件，映射不可用时自动回退到 `pread`； +`MMAP` 不回退：输入不受支持时返回 `RET_NOT_SUPPORT`，映射失败时返回 +`RET_FILE_MAP_ERR`。映射 reader 打开期间不得修改或截断文件。 + --- ## 使用 TsFile diff --git a/cpp/README.md b/cpp/README.md index 6341287e4..816dcf23e 100644 --- a/cpp/README.md +++ b/cpp/README.md @@ -348,6 +348,27 @@ storage::set_write_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). +### Local File Read Backend + +Readers can use memory-mapped I/O or the traditional positioned-read path for +local files. The setting is captured when a reader opens a file, so changing it +does not affect readers that are already open. + +```cpp +#include "common/global.h" + +common::set_file_read_backend(common::FileReadBackend::AUTO); // default +common::set_file_read_backend(common::FileReadBackend::MMAP); // require mmap +common::set_file_read_backend(common::FileReadBackend::PREAD); // legacy path +``` + +The C API exposes the same setting through +`tsfile_set_file_read_backend(TSFILE_READ_BACKEND_*)`. `AUTO` prefers memory +mapping for supported regular files and falls back to `pread` if mapping is not +available. `MMAP` does not fall back: unsupported inputs return +`RET_NOT_SUPPORT`, while mapping failures return `RET_FILE_MAP_ERR`. Files must +not be modified or truncated while a mapped reader is open. + ## Use TsFile You can find examples on how to read and write data in `demo_read.cpp` and `demo_write.cpp` located under `./examples/cpp_examples`. There are also examples under `./examples/c_examples` on how to use a C-style API to read and write data in a C environment. The examples will be built automatically when you run the main build command. diff --git a/cpp/bench_mark/README.md b/cpp/bench_mark/README.md new file mode 100644 index 000000000..14950e14b --- /dev/null +++ b/cpp/bench_mark/README.md @@ -0,0 +1,56 @@ + + +# C++ Benchmarks + +`read_backend_benchmark` compares `PREAD` and `MMAP` without enforcing a +performance threshold. It reports sequential 64 KiB reads, deterministic +random 4 KiB reads, repeated parsing of real TsFile metadata, random bounded +`queryByRow` queries, and one concurrent random-read worker per input file. The +query-planning scan is performed before the timed random-query interval. +Files with no queryable rows still participate in the byte-read and metadata +workloads, but are skipped for the random-query workload. + +Build it through the main CMake project so it links the exact SDK under test: + +```bash +cmake -S cpp -B cpp/target/read-backend-benchmark \ + -DCMAKE_BUILD_TYPE=Release \ + -DBUILD_BENCHMARK=ON -DBUILD_TEST=OFF -DBUILD_TOOLS=OFF \ + -DTSFILE_BUILD_SHARED=OFF +cmake --build cpp/target/read-backend-benchmark \ + --target read_backend_benchmark --config Release +``` + +With a single-config generator such as Ninja, the executable is +`cpp/target/read-backend-benchmark/read_backend_benchmark` (plus `.exe` on +Windows). Run it with one or, preferably, several representative TsFiles: + +```bash +./read_backend_benchmark data-1.tsfile data-2.tsfile data-3.tsfile +./read_backend_benchmark --mmap-first data-1.tsfile data-2.tsfile data-3.tsfile +``` + +The second form reverses backend order to expose warm page-cache bias. For +meaningful results, repeat both forms and record filesystem, storage device, +file sizes, compiler flags, and whether the OS page cache was warm. The checksum +makes accidental short reads, metadata-load failures, or optimizer removal +visible; it is not a TsFile-content checksum. diff --git a/cpp/bench_mark/bench_mark_src/read_backend_benchmark.cc b/cpp/bench_mark/bench_mark_src/read_backend_benchmark.cc new file mode 100644 index 000000000..b2bb35312 --- /dev/null +++ b/cpp/bench_mark/bench_mark_src/read_backend_benchmark.cc @@ -0,0 +1,564 @@ +/* + * 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/global.h" +#include "file/read_file.h" +#include "file/tsfile_io_reader.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" + +namespace { + +const int32_t kSequentialBlockSize = 64 * 1024; +const int32_t kRandomBlockSize = 4 * 1024; +const size_t kRandomOperationsPerFile = 50000; +const size_t kMetadataLoadsPerFile = 100; +const size_t kRandomQueriesPerFile = 200; +const int kQueryRowLimit = 64; +const size_t kConcurrentOperationsPerFile = 50000; + +struct Result { + Result() + : bytes(0), + operations(0), + checksum(0), + seconds(0), + success(true), + skipped(false) {} + + uint64_t bytes; + uint64_t operations; + uint64_t checksum; + double seconds; + bool success; + bool skipped; +}; + +struct QueryPlan { + QueryPlan() : table_model(false), row_count(0) {} + + bool table_model; + std::string table_name; + std::vector columns_or_paths; + int row_count; +}; + +typedef std::vector> OpenFiles; + +uint64_t update_checksum(uint64_t checksum, const std::vector& buffer, + int32_t read_len) { + if (read_len == 0) { + return checksum; + } + const uint64_t first = + static_cast(static_cast(buffer.front())); + const uint64_t last = static_cast( + static_cast(buffer[static_cast(read_len - 1)])); + return (checksum * 1099511628211ULL) ^ (first << 8) ^ last ^ + static_cast(read_len); +} + +int consume_result(storage::TsFileReader& reader, storage::ResultSet* result, + uint64_t& row_count) { + row_count = 0; + if (result == nullptr) { + return common::E_INVALID_ARG; + } + bool has_next = false; + int ret = common::E_OK; + while ((ret = result->next(has_next)) == common::E_OK && has_next) { + ++row_count; + } + reader.destroy_query_data_set(result); + return ret; +} + +int build_query_plan(storage::TsFileReader& reader, QueryPlan& plan, + bool& found) { + found = false; + const std::vector> table_schemas = + reader.get_all_table_schemas(); + for (size_t i = 0; i < table_schemas.size(); ++i) { + if (table_schemas[i] == nullptr) { + continue; + } + const std::vector columns = + table_schemas[i]->get_measurement_names(); + if (columns.empty()) { + continue; + } + storage::ResultSet* result = nullptr; + const int ret = reader.queryByRow(table_schemas[i]->get_table_name(), + columns, 0, -1, result); + uint64_t row_count = 0; + const int consume_ret = ret == common::E_OK + ? consume_result(reader, result, row_count) + : ret; + if (ret != common::E_OK) { + if (result != nullptr) { + reader.destroy_query_data_set(result); + } + return ret; + } + if (consume_ret != common::E_OK) { + return consume_ret; + } + if (row_count > + static_cast(std::numeric_limits::max())) { + return common::E_NOT_SUPPORT; + } + if (row_count > 0) { + plan.table_model = true; + plan.table_name = table_schemas[i]->get_table_name(); + plan.columns_or_paths = columns; + plan.row_count = static_cast(row_count); + found = true; + return common::E_OK; + } + } + + const std::vector> devices = + reader.get_all_device_ids(); + for (size_t device_index = 0; device_index < devices.size(); + ++device_index) { + if (devices[device_index] == nullptr) { + continue; + } + std::vector schemas; + const int schema_ret = + reader.get_timeseries_schema(devices[device_index], schemas); + if (schema_ret != common::E_OK) { + return schema_ret; + } + if (schemas.empty()) { + continue; + } + std::vector paths; + const size_t path_count = std::min(schemas.size(), 4); + for (size_t schema_index = 0; schema_index < path_count; + ++schema_index) { + paths.push_back(devices[device_index]->get_device_name() + "." + + schemas[schema_index].measurement_name_); + } + storage::ResultSet* result = nullptr; + const int ret = reader.queryByRow(paths, 0, -1, result); + uint64_t row_count = 0; + const int consume_ret = ret == common::E_OK + ? consume_result(reader, result, row_count) + : ret; + if (ret != common::E_OK) { + if (result != nullptr) { + reader.destroy_query_data_set(result); + } + return ret; + } + if (consume_ret != common::E_OK) { + return consume_ret; + } + if (row_count > + static_cast(std::numeric_limits::max())) { + return common::E_NOT_SUPPORT; + } + if (row_count > 0) { + plan.table_model = false; + plan.columns_or_paths = paths; + plan.row_count = static_cast(row_count); + found = true; + return common::E_OK; + } + } + return common::E_OK; +} + +bool open_files(const std::vector& paths, OpenFiles& files) { + files.clear(); + for (size_t i = 0; i < paths.size(); ++i) { + std::unique_ptr file(new storage::ReadFile()); + const int ret = file->open(paths[i]); + if (ret != common::E_OK) { + std::cerr << "failed to open " << paths[i] << ": error " << ret + << std::endl; + return false; + } + files.push_back(std::move(file)); + } + return true; +} + +Result run_sequential(const OpenFiles& files) { + Result result; + std::vector buffer(static_cast(kSequentialBlockSize)); + const std::chrono::steady_clock::time_point start = + std::chrono::steady_clock::now(); + for (size_t file_index = 0; file_index < files.size(); ++file_index) { + storage::ReadFile& file = *files[file_index]; + for (int64_t offset = 0; offset < file.file_size(); + offset += kSequentialBlockSize) { + int32_t read_len = 0; + const int ret = file.read(offset, buffer.data(), + kSequentialBlockSize, read_len); + if (ret != common::E_OK) { + std::cerr << "sequential read failed: error " << ret + << std::endl; + result.success = false; + return result; + } + result.bytes += static_cast(read_len); + ++result.operations; + result.checksum = + update_checksum(result.checksum, buffer, read_len); + } + } + result.seconds = + std::chrono::duration(std::chrono::steady_clock::now() - start) + .count(); + return result; +} + +Result run_random(const OpenFiles& files, int32_t requested_block_size, + size_t operations_per_file) { + Result result; + std::vector buffer(static_cast(requested_block_size)); + std::mt19937_64 random(0x903ULL); + const std::chrono::steady_clock::time_point start = + std::chrono::steady_clock::now(); + for (size_t file_index = 0; file_index < files.size(); ++file_index) { + storage::ReadFile& file = *files[file_index]; + const uint64_t file_size = static_cast(file.file_size()); + const int32_t block_size = static_cast(std::min( + file_size, static_cast(requested_block_size))); + const uint64_t maximum_offset = file_size - block_size; + std::uniform_int_distribution offsets(0, maximum_offset); + for (size_t operation = 0; operation < operations_per_file; + ++operation) { + int32_t read_len = 0; + const uint64_t offset = offsets(random); + const int ret = file.read(static_cast(offset), + buffer.data(), block_size, read_len); + if (ret != common::E_OK || read_len != block_size) { + std::cerr << "random read failed: error " << ret << std::endl; + result.success = false; + return result; + } + result.bytes += static_cast(read_len); + ++result.operations; + result.checksum = + update_checksum(result.checksum, buffer, read_len); + } + } + result.seconds = + std::chrono::duration(std::chrono::steady_clock::now() - start) + .count(); + return result; +} + +Result run_metadata(const std::vector& paths, + const std::vector& file_sizes) { + Result result; + const std::chrono::steady_clock::time_point start = + std::chrono::steady_clock::now(); + for (size_t file_index = 0; file_index < paths.size(); ++file_index) { + for (size_t operation = 0; operation < kMetadataLoadsPerFile; + ++operation) { + storage::TsFileIOReader reader; + const int ret = reader.init(paths[file_index]); + if (ret != common::E_OK) { + std::cerr << "metadata reader open failed: error " << ret + << std::endl; + result.success = false; + return result; + } + storage::TsFileMeta* metadata = reader.get_tsfile_meta(); + if (metadata == nullptr || metadata->meta_offset_ <= 0 || + static_cast(metadata->meta_offset_) >= + file_sizes[file_index]) { + std::cerr << "metadata load produced an invalid offset" + << std::endl; + result.success = false; + return result; + } + const uint64_t metadata_bytes = + file_sizes[file_index] - + static_cast(metadata->meta_offset_); + result.bytes += metadata_bytes; + ++result.operations; + result.checksum = + (result.checksum * 1099511628211ULL) ^ metadata_bytes ^ + static_cast(metadata->table_schemas_.size()) ^ + static_cast(metadata->tsfile_properties_.size()); + } + } + result.seconds = + std::chrono::duration(std::chrono::steady_clock::now() - start) + .count(); + return result; +} + +Result run_random_queries(const std::vector& paths) { + Result result; + std::vector> readers; + std::vector plans; + readers.reserve(paths.size()); + plans.reserve(paths.size()); + for (size_t file_index = 0; file_index < paths.size(); ++file_index) { + std::unique_ptr reader( + new storage::TsFileReader()); + const int ret = reader->open(paths[file_index]); + if (ret != common::E_OK) { + std::cerr << "query reader open failed: error " << ret << std::endl; + result.success = false; + return result; + } + QueryPlan plan; + bool found = false; + const int plan_ret = build_query_plan(*reader, plan, found); + if (plan_ret != common::E_OK) { + std::cerr << "random-query planning failed for " + << paths[file_index] << ": error " << plan_ret + << std::endl; + result.success = false; + return result; + } + if (!found) { + std::cerr << "skipping random queries for " << paths[file_index] + << ": no queryable rows" << std::endl; + continue; + } + readers.push_back(std::move(reader)); + plans.push_back(plan); + } + if (readers.empty()) { + result.skipped = true; + return result; + } + + std::mt19937 random(0x903U); + const std::chrono::steady_clock::time_point start = + std::chrono::steady_clock::now(); + for (size_t file_index = 0; file_index < readers.size(); ++file_index) { + QueryPlan& plan = plans[file_index]; + std::uniform_int_distribution offsets(0, plan.row_count - 1); + for (size_t operation = 0; operation < kRandomQueriesPerFile; + ++operation) { + const int offset = offsets(random); + const int limit = std::min(kQueryRowLimit, plan.row_count - offset); + storage::ResultSet* query_result = nullptr; + int ret = common::E_OK; + if (plan.table_model) { + ret = readers[file_index]->queryByRow( + plan.table_name, plan.columns_or_paths, offset, limit, + query_result); + } else { + ret = readers[file_index]->queryByRow( + plan.columns_or_paths, offset, limit, query_result); + } + uint64_t rows = 0; + if (ret != common::E_OK || + consume_result(*readers[file_index], query_result, rows) != + common::E_OK || + rows != static_cast(limit)) { + if (query_result != nullptr && ret != common::E_OK) { + readers[file_index]->destroy_query_data_set(query_result); + } + std::cerr << "random query failed: error " << ret << std::endl; + result.success = false; + return result; + } + ++result.operations; + result.checksum = (result.checksum * 1099511628211ULL) ^ + static_cast(offset) ^ (rows << 32); + } + } + result.seconds = + std::chrono::duration(std::chrono::steady_clock::now() - start) + .count(); + return result; +} + +Result run_concurrent(const std::vector& paths) { + const std::chrono::steady_clock::time_point start = + std::chrono::steady_clock::now(); + std::vector per_file(paths.size()); + std::vector workers; + workers.reserve(paths.size()); + for (size_t file_index = 0; file_index < paths.size(); ++file_index) { + workers.push_back(std::thread([&, file_index]() { + storage::ReadFile file; + if (file.open(paths[file_index]) != common::E_OK) { + per_file[file_index].success = false; + return; + } + const uint64_t file_size = static_cast(file.file_size()); + const int32_t block_size = static_cast( + std::min(file_size, kRandomBlockSize)); + const uint64_t maximum_offset = file_size - block_size; + std::mt19937_64 random(0x903ULL + file_index); + std::uniform_int_distribution offsets(0, maximum_offset); + std::vector buffer(static_cast(block_size)); + Result& result = per_file[file_index]; + for (size_t operation = 0; operation < kConcurrentOperationsPerFile; + ++operation) { + int32_t read_len = 0; + const int ret = file.read(static_cast(offsets(random)), + buffer.data(), block_size, read_len); + if (ret != common::E_OK || read_len != block_size) { + result.success = false; + return; + } + result.bytes += static_cast(read_len); + ++result.operations; + result.checksum = + update_checksum(result.checksum, buffer, read_len); + } + })); + } + for (size_t i = 0; i < workers.size(); ++i) { + workers[i].join(); + } + + Result total; + for (size_t i = 0; i < per_file.size(); ++i) { + total.success = total.success && per_file[i].success; + total.bytes += per_file[i].bytes; + total.operations += per_file[i].operations; + total.checksum ^= per_file[i].checksum; + } + total.seconds = + std::chrono::duration(std::chrono::steady_clock::now() - start) + .count(); + return total; +} + +void print_result(const char* workload, const Result& result, + bool report_throughput = true) { + if (result.skipped) { + std::cout << std::left << std::setw(24) << workload + << "SKIPPED (no queryable rows)" << std::endl; + return; + } + if (!result.success) { + std::cout << std::left << std::setw(24) << workload << "FAILED" + << std::endl; + return; + } + const double mib = static_cast(result.bytes) / (1024.0 * 1024.0); + const double throughput = result.seconds > 0 ? mib / result.seconds : 0; + const double operations_per_second = + result.seconds > 0 ? result.operations / result.seconds : 0; + std::cout << std::left << std::setw(24) << workload << std::right + << std::fixed << std::setprecision(3) << std::setw(10) + << result.seconds << " s "; + if (report_throughput) { + std::cout << std::setw(12) << throughput << " MiB/s "; + } else { + std::cout << std::setw(21) << ""; + } + std::cout << std::setw(14) << operations_per_second + << " ops/s checksum=" << result.checksum << std::endl; +} + +const char* backend_name(common::FileReadBackend backend) { + return backend == common::FileReadBackend::MMAP ? "MMAP" : "PREAD"; +} + +bool run_backend(common::FileReadBackend backend, + const std::vector& paths) { + if (common::set_file_read_backend(backend) != common::E_OK) { + return false; + } + OpenFiles files; + if (!open_files(paths, files)) { + return false; + } + + std::cout << "\n" << backend_name(backend) << std::endl; + const Result sequential = run_sequential(files); + const Result random = + run_random(files, kRandomBlockSize, kRandomOperationsPerFile); + std::vector file_sizes; + file_sizes.reserve(files.size()); + for (size_t i = 0; i < files.size(); ++i) { + file_sizes.push_back(static_cast(files[i]->file_size())); + } + print_result("sequential-64KiB", sequential); + print_result("random-read-4KiB", random); + files.clear(); + const Result metadata = run_metadata(paths, file_sizes); + print_result("metadata-parse", metadata); + const Result random_queries = run_random_queries(paths); + print_result("random-query-by-row", random_queries, false); + const Result concurrent = run_concurrent(paths); + print_result("concurrent-multi-file", concurrent); + return sequential.success && random.success && metadata.success && + random_queries.success && concurrent.success; +} + +} // namespace + +int main(int argc, char** argv) { + const bool mmap_first = argc > 1 && std::string(argv[1]) == "--mmap-first"; + const int first_path = mmap_first ? 2 : 1; + if (argc <= first_path) { + std::cerr << "usage: read_backend_benchmark [--mmap-first] FILE.tsfile " + "[FILE.tsfile ...]" + << std::endl; + return 2; + } + + std::vector paths; + for (int i = first_path; i < argc; ++i) { + paths.push_back(argv[i]); + } + + const int init_ret = storage::libtsfile_init(); + if (init_ret != common::E_OK) { + std::cerr << "failed to initialize TsFile: error " << init_ret + << std::endl; + return 1; + } + + const common::FileReadBackend original = common::get_file_read_backend(); + bool pread_ok = false; + bool mmap_ok = false; + if (mmap_first) { + mmap_ok = run_backend(common::FileReadBackend::MMAP, paths); + pread_ok = run_backend(common::FileReadBackend::PREAD, paths); + } else { + pread_ok = run_backend(common::FileReadBackend::PREAD, paths); + mmap_ok = run_backend(common::FileReadBackend::MMAP, paths); + } + common::set_file_read_backend(original); + storage::libtsfile_destroy(); + std::cout << "\nRun again with the opposite order (toggle --mmap-first) " + "when measuring warm page-cache effects." + << std::endl; + return pread_ok && mmap_ok ? 0 : 1; +} diff --git a/cpp/src/common/config/config.h b/cpp/src/common/config/config.h index 5cf968688..03f2f02b8 100644 --- a/cpp/src/common/config/config.h +++ b/cpp/src/common/config/config.h @@ -25,6 +25,13 @@ namespace common { +/** Backend used by local ReadFile instances. */ +enum class FileReadBackend : uint8_t { + AUTO = 0, + MMAP = 1, + PREAD = 2, +}; + typedef struct ConfigValue { uint32_t tsblock_mem_inc_step_size_; // tsblock memory self-increment step size diff --git a/cpp/src/common/global.cc b/cpp/src/common/global.cc index 91b0e99c3..1b58d4e33 100644 --- a/cpp/src/common/global.cc +++ b/cpp/src/common/global.cc @@ -19,6 +19,8 @@ #include "global.h" +#include + #ifdef ENABLE_THREADS #include "common/thread_pool.h" #endif @@ -35,6 +37,13 @@ namespace common { +namespace { +// Kept outside ConfigValue for ABI compatibility. It is also intentionally not +// reset by init_common(), so callers may configure the first reader before +// libtsfile_init(). +std::atomic g_file_read_backend(FileReadBackend::AUTO); +} // namespace + ColumnSchema g_time_column_schema; ConfigValue g_config_value_; #ifdef ENABLE_THREADS @@ -185,6 +194,22 @@ int set_thread_count(int32_t count) { return E_OK; } +int set_file_read_backend(FileReadBackend backend) { + switch (backend) { + case FileReadBackend::AUTO: + case FileReadBackend::MMAP: + case FileReadBackend::PREAD: + g_file_read_backend.store(backend, std::memory_order_relaxed); + return E_OK; + default: + return E_INVALID_ARG; + } +} + +FileReadBackend get_file_read_backend() { + return g_file_read_backend.load(std::memory_order_relaxed); +} + bool is_timestamp_column_name(const char* time_col_name) { // both "time" and "timestamp" refer to timestamp column. int32_t len = strlen(time_col_name); @@ -264,4 +289,14 @@ void print_backtrace() { std::map g_all_inject_points; +#ifdef ENABLE_TEST +void enable_injection(const char* inject_point_name, int count) { + g_all_inject_points[inject_point_name] = InjectPoint{count}; +} + +void disable_injection(const char* inject_point_name) { + g_all_inject_points.erase(inject_point_name); +} +#endif + } // namespace common diff --git a/cpp/src/common/global.h b/cpp/src/common/global.h index 66a95c1a7..5ef7d6d40 100644 --- a/cpp/src/common/global.h +++ b/cpp/src/common/global.h @@ -199,6 +199,13 @@ FORCE_INLINE bool get_parallel_write_enabled() { return g_config_value_.parallel_write_enabled_; } +// Select the backend used by subsequently opened local files. Existing +// ReadFile instances retain the backend selected when they were opened. This +// setting deliberately lives outside exported ConfigValue so adding it does +// not change that public data structure's ABI. +extern int set_file_read_backend(FileReadBackend backend); +extern FileReadBackend get_file_read_backend(); + // Size of the single global worker pool. Rejects values outside [1, 64] with // E_INVALID_ARG, leaving the field untouched. If the pool already exists // (libtsfile_init has run) it is rebuilt at the new size immediately; the diff --git a/cpp/src/cwrapper/errno_define_c.h b/cpp/src/cwrapper/errno_define_c.h index 3ad1c1301..b07da986e 100644 --- a/cpp/src/cwrapper/errno_define_c.h +++ b/cpp/src/cwrapper/errno_define_c.h @@ -56,5 +56,6 @@ #define RET_INVALID_NODE_TYPE 52 #define RET_ENCODE_ERR 53 #define RET_DECODE_ERR 54 +#define RET_FILE_MAP_ERR 55 #endif /* CWRAPPER_ERRNO_DEFINE_H */ diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc index e54afecb9..54dc40af6 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.cc +++ b/cpp/src/cwrapper/tsfile_cwrapper.cc @@ -37,6 +37,7 @@ #include #include "common/device_id.h" +#include "common/global.h" #include "common/statistic.h" #include "common/tablet.h" #include "common/tsfile_common.h" @@ -92,6 +93,24 @@ int set_global_compression(uint8_t compression) { return common::set_global_compression(compression); } +ERRNO tsfile_set_file_read_backend(int32_t backend) { + switch (backend) { + case TSFILE_READ_BACKEND_AUTO: + return common::set_file_read_backend(common::FileReadBackend::AUTO); + case TSFILE_READ_BACKEND_MMAP: + return common::set_file_read_backend(common::FileReadBackend::MMAP); + case TSFILE_READ_BACKEND_PREAD: + return common::set_file_read_backend( + common::FileReadBackend::PREAD); + default: + return common::E_INVALID_ARG; + } +} + +TsFileReadBackend tsfile_get_file_read_backend() { + return static_cast(common::get_file_read_backend()); +} + WriteFile write_file_new(const char* pathname, ERRNO* err_code) { int ret; init_tsfile_config(); diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h index 0476f691d..1820740bc 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.h +++ b/cpp/src/cwrapper/tsfile_cwrapper.h @@ -77,6 +77,13 @@ typedef enum { TS_COMPRESSION_INVALID = 255 } CompressionType; +/** Local-file read backend selected for subsequently opened readers. */ +typedef enum { + TSFILE_READ_BACKEND_AUTO = 0, + TSFILE_READ_BACKEND_MMAP = 1, + TSFILE_READ_BACKEND_PREAD = 2 +} TsFileReadBackend; + typedef enum column_category { TAG = 0, FIELD = 1, @@ -333,6 +340,22 @@ typedef struct arrow_array { typedef int32_t ERRNO; typedef int64_t Timestamp; +/** + * @brief Select the backend used by subsequently opened local TsFile readers. + * + * AUTO prefers memory mapping and falls back to pread, MMAP requires memory + * mapping, and PREAD preserves the traditional positioned-read path. Existing + * readers are unaffected. + * + * @param backend One of TSFILE_READ_BACKEND_AUTO, TSFILE_READ_BACKEND_MMAP, + * or TSFILE_READ_BACKEND_PREAD. + * @return RET_OK on success, or RET_INVALID_ARG for any other value. + */ +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 Get the encoding type for global time column * diff --git a/cpp/src/file/read_file.cc b/cpp/src/file/read_file.cc index ce1f67197..8176fecb4 100644 --- a/cpp/src/file/read_file.cc +++ b/cpp/src/file/read_file.cc @@ -22,18 +22,24 @@ #include #include +#include +#include #include +#include #ifdef _WIN32 #include #include ssize_t pread(int fd, void* buf, size_t count, uint64_t offset); #else +#include #include #endif +#include "common/global.h" #include "common/logger/elog.h" #include "common/tsfile_common.h" +#include "utils/injection.h" #include "utils/util_define.h" // ssize_t and other platform-compat shims using namespace common; @@ -53,12 +59,33 @@ uint64_t generation_hash(uint64_t size, int64_t mtime_ns) { } } // namespace +ReadFile::ReadFile() + : file_path_(), + fd_(-1), + file_size_(-1), + mapped_data_(nullptr), + mapped_size_(0), + active_backend_(common::FileReadBackend::PREAD) +#ifdef _WIN32 + , + mapping_handle_(nullptr) +#else + , + file_device_(0), + file_inode_(0) +#endif +{ +} + int ReadFile::generation(uint64_t& size, uint64_t& fingerprint) const { - if (fd_ < 0) { + if (!is_opened()) { return E_FILE_READ_ERR; } int64_t mtime_ns = 0; #ifdef _WIN32 + if (fd_ < 0) { + return E_FILE_READ_ERR; + } intptr_t handle_value = _get_osfhandle(fd_); if (handle_value == -1) { return E_FILE_READ_ERR; @@ -70,9 +97,19 @@ int ReadFile::generation(uint64_t& size, uint64_t& fingerprint) const { } static const int64_t WINDOWS_TO_UNIX_100NS = 116444736000000000LL; mtime_ns = (info.LastWriteTime.QuadPart - WINDOWS_TO_UNIX_100NS) * 100; + LARGE_INTEGER current_size; + if (!GetFileSizeEx(reinterpret_cast(handle_value), ¤t_size) || + current_size.QuadPart < 0) { + return E_FILE_READ_ERR; + } + size = static_cast(current_size.QuadPart); #else struct stat info; - if (::fstat(fd_, &info) != 0) { + if (::fstat(fd_, &info) != 0 || info.st_size < 0) { + return E_FILE_READ_ERR; + } + if (static_cast(info.st_dev) != file_device_ || + static_cast(info.st_ino) != file_inode_) { return E_FILE_READ_ERR; } #ifdef __APPLE__ @@ -82,22 +119,29 @@ int ReadFile::generation(uint64_t& size, uint64_t& fingerprint) const { mtime_ns = static_cast(info.st_mtim.tv_sec) * 1000000000LL + info.st_mtim.tv_nsec; #endif + size = static_cast(info.st_size); #endif - size = static_cast(file_size_); fingerprint = generation_hash(size, mtime_ns); return E_OK; } void ReadFile::close() { + unmap_file(); if (fd_ >= 0) { ::close(fd_); fd_ = -1; } file_size_ = -1; + active_backend_ = common::FileReadBackend::PREAD; +#ifndef _WIN32 + file_device_ = 0; + file_inode_ = 0; +#endif } int ReadFile::open(const std::string& file_path) { int ret = E_OK; + close(); file_path_ = file_path; int flags = O_RDONLY; #ifdef _WIN32 @@ -111,7 +155,28 @@ int ReadFile::open(const std::string& file_path) { } if (RET_FAIL(get_file_size(file_size_))) { - } else if (RET_FAIL(check_file_magic())) { + } else if (file_size_ < MIN_FILE_SIZE) { + ret = E_TSFILE_CORRUPTED; + LOGE("tsfile " << file_path_.c_str() + << " is corrupted, file_size=" << file_size_); + } else { + const common::FileReadBackend configured_backend = + common::get_file_read_backend(); + active_backend_ = common::FileReadBackend::PREAD; + if (configured_backend != common::FileReadBackend::PREAD) { + const int map_ret = map_file(); + if (map_ret == E_OK) { + active_backend_ = common::FileReadBackend::MMAP; + } else if (configured_backend == common::FileReadBackend::MMAP) { + ret = map_ret; + } else { + LOGW("mmap unavailable for " << file_path_.c_str() + << "; falling back to pread"); + } + } + if (ret == E_OK) { + ret = check_file_magic(); + } } if (IS_FAIL(ret)) { close(); @@ -132,9 +197,107 @@ int ReadFile::get_file_size(int64_t& file_size) { return E_FILE_STAT_ERR; } file_size = static_cast(s.st_size); +#ifndef _WIN32 + file_device_ = static_cast(s.st_dev); + file_inode_ = static_cast(s.st_ino); +#endif + return E_OK; +} + +int ReadFile::map_file() { + DBUG_EXECUTE_IF("read_file_mmap_fail", return E_FILE_MAP_ERR;); + DBUG_EXECUTE_IF("read_file_mmap_unsupported", return E_NOT_SUPPORT;); + + if (fd_ < 0 || file_size_ <= 0) { + return E_FILE_MAP_ERR; + } + if (static_cast(file_size_) > + static_cast(std::numeric_limits::max())) { + return E_NOT_SUPPORT; + } + +#ifdef _WIN32 + struct __stat64 file_stat; + if (_fstat64(fd_, &file_stat) != 0) { + return E_FILE_STAT_ERR; + } + if ((file_stat.st_mode & _S_IFREG) == 0) { + return E_NOT_SUPPORT; + } + const intptr_t handle_value = _get_osfhandle(fd_); + if (handle_value == -1) { + return E_FILE_MAP_ERR; + } + HANDLE mapping = CreateFileMapping(reinterpret_cast(handle_value), + nullptr, PAGE_READONLY, 0, 0, nullptr); + if (mapping == nullptr) { + const DWORD error = GetLastError(); + if (error == ERROR_NOT_SUPPORTED || error == ERROR_INVALID_FUNCTION) { + return E_NOT_SUPPORT; + } + return E_FILE_MAP_ERR; + } + void* view = MapViewOfFile(mapping, FILE_MAP_READ, 0, 0, 0); + if (view == nullptr) { + const DWORD error = GetLastError(); + CloseHandle(mapping); + if (error == ERROR_NOT_SUPPORTED || error == ERROR_INVALID_FUNCTION) { + return E_NOT_SUPPORT; + } + return E_FILE_MAP_ERR; + } + mapping_handle_ = mapping; + mapped_data_ = static_cast(view); +#else + struct stat file_stat; + if (::fstat(fd_, &file_stat) != 0) { + return E_FILE_STAT_ERR; + } + if (!S_ISREG(file_stat.st_mode)) { + return E_NOT_SUPPORT; + } + void* view = ::mmap(nullptr, static_cast(file_size_), PROT_READ, + MAP_PRIVATE, fd_, 0); + if (view == MAP_FAILED) { + if (errno == ENODEV || errno == ENOSYS +#ifdef EOPNOTSUPP + || errno == EOPNOTSUPP +#endif + ) { + return E_NOT_SUPPORT; + } + return E_FILE_MAP_ERR; + } + // C++ cannot safely represent a usable object at the null pointer value, + // even though POSIX theoretically permits mmap() to return address zero. + if (view == nullptr) { + ::munmap(view, static_cast(file_size_)); + return E_FILE_MAP_ERR; + } + mapped_data_ = static_cast(view); +#endif + mapped_size_ = static_cast(file_size_); return E_OK; } +void ReadFile::unmap_file() { + if (mapped_data_ != nullptr) { +#ifdef _WIN32 + UnmapViewOfFile(mapped_data_); +#else + ::munmap(const_cast(mapped_data_), mapped_size_); +#endif + mapped_data_ = nullptr; + } +#ifdef _WIN32 + if (mapping_handle_ != nullptr) { + CloseHandle(static_cast(mapping_handle_)); + mapping_handle_ = nullptr; + } +#endif + mapped_size_ = 0; +} + int ReadFile::check_file_magic() { int ret = E_OK; if (file_size_ < MIN_FILE_SIZE) { @@ -170,17 +333,49 @@ int ReadFile::check_file_magic() { int ReadFile::read(int64_t offset, char* buf, int32_t buf_size, int32_t& read_len) { - int ret = E_OK; read_len = 0; - while (read_len < buf_size) { + if (offset < 0 || buf_size < 0 || (buf == nullptr && buf_size > 0)) { + return E_INVALID_ARG; + } + if (!is_opened()) { + return E_FILE_READ_ERR; + } + if (buf_size == 0 || offset >= file_size_) { + return E_OK; + } + + const int64_t available = file_size_ - offset; + const int32_t target_size = available < static_cast(buf_size) + ? static_cast(available) + : buf_size; + + if (active_backend_ == common::FileReadBackend::MMAP) { + std::memcpy(buf, mapped_data_ + static_cast(offset), + static_cast(target_size)); + read_len = target_size; + return E_OK; + } + + if (fd_ < 0) { + return E_FILE_READ_ERR; + } + int ret = E_OK; + while (read_len < target_size) { #ifdef _WIN32 - ssize_t pread_size = ::pread(fd_, buf + read_len, buf_size - read_len, - static_cast(offset + read_len)); + ssize_t pread_size = + ::pread(fd_, buf + read_len, target_size - read_len, + static_cast(offset + read_len)); #else - ssize_t pread_size = ::pread(fd_, buf + read_len, buf_size - read_len, - static_cast(offset + read_len)); + ssize_t pread_size = + ::pread(fd_, buf + read_len, target_size - read_len, + static_cast(offset + read_len)); #endif if (pread_size < 0) { +#ifndef _WIN32 + if (errno == EINTR) { + continue; + } +#endif ret = E_FILE_READ_ERR; ////log_err("tsfile reader error, file_path=%s, errno=%d", /// file_path_.c_str(), errno); diff --git a/cpp/src/file/read_file.h b/cpp/src/file/read_file.h index fbfe50a8b..fceb9d8a7 100644 --- a/cpp/src/file/read_file.h +++ b/cpp/src/file/read_file.h @@ -22,8 +22,10 @@ #include +#include #include +#include "common/config/config.h" #include "utils/errno_define.h" #include "utils/util_define.h" @@ -31,14 +33,22 @@ namespace storage { class ReadFile { public: - ReadFile() : file_path_(), fd_(-1), file_size_(-1) {} + ReadFile(); ~ReadFile() { destroy(); } + ReadFile(const ReadFile&) = delete; + ReadFile& operator=(const ReadFile&) = delete; + void destroy() { close(); } int open(const std::string& file_path); - FORCE_INLINE bool is_opened() const { return fd_ > 0; } + FORCE_INLINE bool is_opened() const { + return fd_ >= 0 || mapped_data_ != nullptr; + } FORCE_INLINE int64_t file_size() const { return file_size_; } FORCE_INLINE const std::string& file_path() const { return file_path_; } + FORCE_INLINE common::FileReadBackend active_backend() const { + return active_backend_; + } /** Return size and the Dataset Index v1 FNV fingerprint of size+mtime_ns. */ @@ -50,11 +60,15 @@ class ReadFile { */ int read(int64_t offset, char* buf, int32_t buf_size, int32_t& ret_read_len); + // open()/close() must not race with read() or generation(). Concurrent + // reads are supported after the file has been opened. void close(); private: int get_file_size(int64_t& file_size); int check_file_magic(); + int map_file(); + void unmap_file(); private: // 2 magic strings + file_version @@ -64,6 +78,15 @@ class ReadFile { std::string file_path_; int fd_; int64_t file_size_; + const char* mapped_data_; + size_t mapped_size_; + common::FileReadBackend active_backend_; +#ifdef _WIN32 + void* mapping_handle_; +#else + uint64_t file_device_; + uint64_t file_inode_; +#endif }; } // end namespace storage diff --git a/cpp/src/utils/errno_define.h b/cpp/src/utils/errno_define.h index ca2d1397a..96956495f 100644 --- a/cpp/src/utils/errno_define.h +++ b/cpp/src/utils/errno_define.h @@ -57,6 +57,7 @@ const int E_UNSUPPORTED_ORDER = 51; const int E_INVALID_NODE_TYPE = 52; const int E_ENCODE_ERR = 53; const int E_DECODE_ERR = 54; +const int E_FILE_MAP_ERR = 55; } // end namespace common diff --git a/cpp/src/utils/injection.h b/cpp/src/utils/injection.h index 0ad79dd6a..664ea3c35 100644 --- a/cpp/src/utils/injection.h +++ b/cpp/src/utils/injection.h @@ -23,6 +23,8 @@ #include #include +#include "utils/util_define.h" + namespace common { // define struct @@ -59,6 +61,11 @@ struct InjectPoint { // the map save all inject points extern std::map g_all_inject_points; +#ifdef ENABLE_TEST +TSFILE_API void enable_injection(const char* inject_point_name, int count); +TSFILE_API void disable_injection(const char* inject_point_name); +#endif + } // end namespace common #endif // COMMON_INJECTION_H diff --git a/cpp/test/cwrapper/cwrapper_test.cc b/cpp/test/cwrapper/cwrapper_test.cc index d8908cc31..a75bdf215 100644 --- a/cpp/test/cwrapper/cwrapper_test.cc +++ b/cpp/test/cwrapper/cwrapper_test.cc @@ -52,6 +52,20 @@ class CWrapperTest : public testing::Test { } }; +TEST_F(CWrapperTest, FileReadBackendConfigurationRoundTripsAndValidates) { + const TsFileReadBackend original = tsfile_get_file_read_backend(); + + EXPECT_EQ(tsfile_set_file_read_backend(TSFILE_READ_BACKEND_MMAP), RET_OK); + EXPECT_EQ(tsfile_get_file_read_backend(), TSFILE_READ_BACKEND_MMAP); + EXPECT_EQ(tsfile_set_file_read_backend(TSFILE_READ_BACKEND_PREAD), RET_OK); + EXPECT_EQ(tsfile_get_file_read_backend(), TSFILE_READ_BACKEND_PREAD); + EXPECT_EQ(tsfile_set_file_read_backend(99), RET_INVALID_ARG); + EXPECT_EQ(tsfile_set_file_read_backend(256), RET_INVALID_ARG); + EXPECT_EQ(tsfile_get_file_read_backend(), TSFILE_READ_BACKEND_PREAD); + + EXPECT_EQ(tsfile_set_file_read_backend(original), RET_OK); +} + 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); diff --git a/cpp/test/file/read_file_test.cc b/cpp/test/file/read_file_test.cc new file mode 100644 index 000000000..3d7420e8a --- /dev/null +++ b/cpp/test/file/read_file_test.cc @@ -0,0 +1,227 @@ +/* + * 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 "file/read_file.h" + +#include + +#include +#include +#include +#include +#include + +#ifdef _WIN32 +#include +#else +#include +#endif + +#include "common/global.h" +#include "utils/injection.h" + +namespace { + +std::string process_temp_path(const char* stem) { + std::ostringstream path; + path << ::testing::TempDir() << stem << "_"; +#ifdef _WIN32 + path << _getpid(); +#else + path << getpid(); +#endif + path << ".tsfile"; + return path.str(); +} + +class BackendGuard { + public: + BackendGuard() : original_(common::get_file_read_backend()) {} + ~BackendGuard() { common::set_file_read_backend(original_); } + + private: + common::FileReadBackend original_; +}; + +class InjectionGuard { + public: + explicit InjectionGuard(const char* point) : point_(point) { + common::enable_injection(point_, 0); + } + ~InjectionGuard() { common::disable_injection(point_); } + + private: + const char* point_; +}; + +class ReadFileBackendTest : public ::testing::Test { + protected: + void SetUp() override { + content_ = "TsFile"; + content_.push_back('\x03'); + content_ += "read-backend-payload"; + content_ += "TsFile"; + write_file(file_name_, content_); + } + + void TearDown() override { + std::remove(file_name_.c_str()); + std::remove(empty_file_name_.c_str()); + } + + static void write_file(const std::string& path, + const std::string& content) { + std::ofstream output( + path.c_str(), std::ios::out | std::ios::binary | std::ios::trunc); + ASSERT_TRUE(output.is_open()); + output.write(content.data(), + static_cast(content.size())); + ASSERT_TRUE(output.good()); + } + + BackendGuard backend_guard_; + const std::string file_name_ = process_temp_path("read_file_backend_test"); + const std::string empty_file_name_ = + process_temp_path("read_file_backend_empty"); + std::string content_; +}; + +TEST_F(ReadFileBackendTest, PreadPreservesPositionedReadBehavior) { + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::PREAD), + common::E_OK); + storage::ReadFile file; + ASSERT_EQ(file.open(file_name_), common::E_OK); + EXPECT_TRUE(file.is_opened()); + EXPECT_EQ(file.active_backend(), common::FileReadBackend::PREAD); + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::MMAP), + common::E_OK); + EXPECT_EQ(file.active_backend(), common::FileReadBackend::PREAD); + + std::vector buffer(content_.size() + 8, '\0'); + int32_t read_len = -1; + ASSERT_EQ(file.read(0, buffer.data(), static_cast(buffer.size()), + read_len), + common::E_OK); + EXPECT_EQ(read_len, static_cast(content_.size())); + EXPECT_EQ(std::string(buffer.data(), static_cast(read_len)), + content_); + + EXPECT_EQ(file.read(-1, buffer.data(), 1, read_len), common::E_INVALID_ARG); + EXPECT_EQ(file.read(0, nullptr, 1, read_len), common::E_INVALID_ARG); + EXPECT_EQ(file.read(0, nullptr, 0, read_len), common::E_OK); + EXPECT_EQ(read_len, 0); +} + +TEST_F(ReadFileBackendTest, MmapReadsBoundedRangesAndReleasesResources) { + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::MMAP), + common::E_OK); + storage::ReadFile file; + ASSERT_EQ(file.open(file_name_), common::E_OK); + EXPECT_TRUE(file.is_opened()); + EXPECT_EQ(file.active_backend(), common::FileReadBackend::MMAP); + + char tail[16] = {}; + int32_t read_len = -1; + const int64_t offset = static_cast(content_.size()) - 3; + ASSERT_EQ(file.read(offset, tail, sizeof(tail), read_len), common::E_OK); + EXPECT_EQ(read_len, 3); + EXPECT_EQ(std::string(tail, static_cast(read_len)), + content_.substr(content_.size() - 3)); + + uint64_t size = 0; + uint64_t fingerprint = 0; + EXPECT_EQ(file.generation(size, fingerprint), common::E_OK); + EXPECT_EQ(size, content_.size()); + EXPECT_NE(fingerprint, 0u); + + file.close(); + EXPECT_FALSE(file.is_opened()); + EXPECT_EQ(std::remove(file_name_.c_str()), 0); +} + +TEST_F(ReadFileBackendTest, AutoPrefersMmapAndCloseIsIdempotent) { + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::AUTO), + common::E_OK); + storage::ReadFile file; + ASSERT_EQ(file.open(file_name_), common::E_OK); + EXPECT_EQ(file.active_backend(), common::FileReadBackend::MMAP); + + file.close(); + file.close(); + EXPECT_FALSE(file.is_opened()); + + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::PREAD), + common::E_OK); + ASSERT_EQ(file.open(file_name_), common::E_OK); + EXPECT_EQ(file.active_backend(), common::FileReadBackend::PREAD); +} + +TEST_F(ReadFileBackendTest, AutoFallsBackButRequiredMmapReportsFailure) { + InjectionGuard mmap_failure("read_file_mmap_fail"); + + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::AUTO), + common::E_OK); + storage::ReadFile automatic; + ASSERT_EQ(automatic.open(file_name_), common::E_OK); + EXPECT_EQ(automatic.active_backend(), common::FileReadBackend::PREAD); + automatic.close(); + + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::MMAP), + common::E_OK); + storage::ReadFile required; + EXPECT_EQ(required.open(file_name_), common::E_FILE_MAP_ERR); + EXPECT_FALSE(required.is_opened()); +} + +TEST_F(ReadFileBackendTest, AutoFallsBackButRequiredMmapReportsUnsupported) { + InjectionGuard mmap_unsupported("read_file_mmap_unsupported"); + + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::AUTO), + common::E_OK); + storage::ReadFile automatic; + ASSERT_EQ(automatic.open(file_name_), common::E_OK); + EXPECT_EQ(automatic.active_backend(), common::FileReadBackend::PREAD); + automatic.close(); + + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::MMAP), + common::E_OK); + storage::ReadFile required; + EXPECT_EQ(required.open(file_name_), common::E_NOT_SUPPORT); + EXPECT_FALSE(required.is_opened()); +} + +TEST_F(ReadFileBackendTest, EmptyFileIsRejectedBeforeMapping) { + write_file(empty_file_name_, ""); + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::MMAP), + common::E_OK); + storage::ReadFile file; + EXPECT_EQ(file.open(empty_file_name_), common::E_TSFILE_CORRUPTED); + EXPECT_FALSE(file.is_opened()); +} + +TEST_F(ReadFileBackendTest, InvalidConfigurationDoesNotChangeBackend) { + ASSERT_EQ(common::set_file_read_backend(common::FileReadBackend::PREAD), + common::E_OK); + EXPECT_EQ( + common::set_file_read_backend(static_cast(99)), + common::E_INVALID_ARG); + EXPECT_EQ(common::get_file_read_backend(), common::FileReadBackend::PREAD); +} + +} // namespace diff --git a/cpp/test/tools/output_format_test.cc b/cpp/test/tools/output_format_test.cc index 926772166..c9710eab3 100644 --- a/cpp/test/tools/output_format_test.cc +++ b/cpp/test/tools/output_format_test.cc @@ -45,6 +45,8 @@ TEST(ErrorCodeMessageTest, KnownCodesMapToReadablePhrases) { "data is out of order"); EXPECT_STREQ(tsfile_cli::error_code_message(common::E_DECODE_ERR), "failed to decode data"); + EXPECT_STREQ(tsfile_cli::error_code_message(common::E_FILE_MAP_ERR), + "failed to memory-map file"); } TEST(ErrorCodeMessageTest, UnknownCodeFallsBackToInternalError) { diff --git a/cpp/tools/format/output_format.cc b/cpp/tools/format/output_format.cc index 3fecfb69b..bce10b7bd 100644 --- a/cpp/tools/format/output_format.cc +++ b/cpp/tools/format/output_format.cc @@ -66,6 +66,8 @@ const char* error_code_message(int code) { return "failed to encode data"; case common::E_DECODE_ERR: return "failed to decode data"; + case common::E_FILE_MAP_ERR: + return "failed to memory-map file"; default: return "internal error"; } diff --git a/python/README-zh.md b/python/README-zh.md index dd3a59ea4..301cd3b24 100644 --- a/python/README-zh.md +++ b/python/README-zh.md @@ -83,3 +83,23 @@ with TsFileReader("example.tsfile") as reader: ``` Property value 不携带数据类型；保存数字或结构体时应使用明确、可跨语言的字节编码。 + +## 本地文件读取后端 + +Python reader 会在打开文件时继承进程级读取后端配置。`AUTO` 是默认值， +`MMAP` 要求必须使用内存映射，`PREAD` 则恢复传统的定位读取行为。 + +```python +from tsfile import FileReadBackend, TsFileReader, set_file_read_backend + +set_file_read_backend(FileReadBackend.MMAP) +with TsFileReader("example.tsfile") as reader: + ... + +# 配置字典入口具有相同效果。 +from tsfile import set_tsfile_config +set_tsfile_config({"file_read_backend_": FileReadBackend.AUTO}) +``` + +该配置只影响之后打开的 reader。通过内存映射后端打开文件期间，请勿修改或 +截断该文件。 diff --git a/python/README.md b/python/README.md index 8e2716a2c..cdfa22d95 100644 --- a/python/README.md +++ b/python/README.md @@ -77,3 +77,24 @@ with TsFileReader("example.tsfile") as reader: Values do not carry a data type; use an explicit portable encoding when storing numbers or structures. + +## Local File Read Backend + +Python readers inherit the process-wide backend setting when they open a file. +`AUTO` is the default, `MMAP` requires memory mapping, and `PREAD` restores the +traditional positioned-read behavior. + +```python +from tsfile import FileReadBackend, TsFileReader, set_file_read_backend + +set_file_read_backend(FileReadBackend.MMAP) +with TsFileReader("example.tsfile") as reader: + ... + +# The configuration-dictionary API is equivalent. +from tsfile import set_tsfile_config +set_tsfile_config({"file_read_backend_": FileReadBackend.AUTO}) +``` + +The setting only affects readers opened afterward. Do not modify or truncate a +file while it is open through the memory-mapped backend. diff --git a/python/tests/test_exceptions.py b/python/tests/test_exceptions.py index dbc12e792..87e2807c2 100644 --- a/python/tests/test_exceptions.py +++ b/python/tests/test_exceptions.py @@ -22,6 +22,7 @@ from tsfile.exceptions import ( AlreadyExistsError, ErrorCode, + FileMapError, FileOpenError, InvalidArgumentError, InvalidPathError, @@ -63,6 +64,10 @@ def test_get_exception_preserves_known_and_unknown_codes(): assert isinstance(invalid_path, InvalidPathError) assert invalid_path.code == ErrorCode.INVALID_PATH + map_error = get_exception(ErrorCode.FILE_MAP_ERROR) + assert isinstance(map_error, FileMapError) + assert map_error.code == ErrorCode.FILE_MAP_ERROR + def test_invalid_tree_path_propagates_native_error(tmp_path): path = tmp_path / "invalid-path.tsfile" diff --git a/python/tests/test_write_and_read.py b/python/tests/test_write_and_read.py index fda145efd..da5288478 100644 --- a/python/tests/test_write_and_read.py +++ b/python/tests/test_write_and_read.py @@ -600,6 +600,75 @@ def test_tsfile_config(): os.remove("test1.tsfile") +def test_file_read_backend_configuration_and_query_equivalence(): + from tsfile import ( + FileReadBackend, + get_file_read_backend, + get_tsfile_config, + set_file_read_backend, + set_tsfile_config, + ) + + file_name = "read_backend_python_test.tsfile" + original_backend = get_file_read_backend() + table = TableSchema( + "backend_table", + [ + ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG), + ColumnSchema("value", TSDataType.INT64, ColumnCategory.FIELD), + ], + ) + + def read_rows(backend): + set_file_read_backend(backend) + assert get_file_read_backend() == backend + rows = [] + with TsFileReader(file_name) as reader: + with reader.query_table( + "backend_table", ["device", "value"], 0, 9 + ) as result: + while result.next(): + rows.append( + ( + result.get_value_by_name(TIME_COLUMN), + result.get_value_by_name("device"), + result.get_value_by_name("value"), + ) + ) + return rows + + try: + if os.path.exists(file_name): + os.remove(file_name) + with TsFileTableWriter(file_name, table) as writer: + tablet = Tablet( + ["device", "value"], + [TSDataType.STRING, TSDataType.INT64], + 10, + ) + for i in range(10): + tablet.add_timestamp(i, i) + tablet.add_value_by_name("device", i, f"device{i % 2}") + tablet.add_value_by_name("value", i, i * 10) + writer.write_table(tablet) + + pread_rows = read_rows(FileReadBackend.PREAD) + mmap_rows = read_rows(FileReadBackend.MMAP) + assert mmap_rows == pread_rows + assert len(mmap_rows) == 10 + + set_tsfile_config({"file_read_backend_": FileReadBackend.AUTO}) + assert get_tsfile_config()["file_read_backend_"] == FileReadBackend.AUTO + with pytest.raises(TypeError): + set_file_read_backend(1) + with pytest.raises(TypeError): + set_tsfile_config({"file_read_backend_": "mmap"}) + finally: + set_file_read_backend(original_backend) + if os.path.exists(file_name): + os.remove(file_name) + + def test_tsfile_to_df(): table = TableSchema( "test_table", diff --git a/python/tsfile/__init__.py b/python/tsfile/__init__.py index a1c37fce1..c0bf8fb85 100644 --- a/python/tsfile/__init__.py +++ b/python/tsfile/__init__.py @@ -87,7 +87,12 @@ def _preload_dll(path): tag_not_between, ) from .tsfile_writer import TsFileWriterPy as TsFileWriter -from .tsfile_py_cpp import get_tsfile_config, set_tsfile_config +from .tsfile_py_cpp import ( + get_file_read_backend, + get_tsfile_config, + set_file_read_backend, + set_tsfile_config, +) from .tsfile_table_writer import TsFileTableWriter from .utils import to_dataframe, dataframe_to_tsfile from .dataset import TsFileDataFrame, Timeseries, AlignedTimeseries, SeriesPath diff --git a/python/tsfile/constants.py b/python/tsfile/constants.py index 659988c5b..ba515e5fb 100644 --- a/python/tsfile/constants.py +++ b/python/tsfile/constants.py @@ -213,6 +213,15 @@ class Compressor(IntEnum): LZMA2 = 9 +@unique +class FileReadBackend(IntEnum): + """Backend used by TsFile readers opened after configuration.""" + + AUTO = 0 + MMAP = 1 + PREAD = 2 + + @unique class ColumnCategory(IntEnum): TAG = 0 diff --git a/python/tsfile/exceptions.py b/python/tsfile/exceptions.py index c0a963862..b8aac622c 100644 --- a/python/tsfile/exceptions.py +++ b/python/tsfile/exceptions.py @@ -57,6 +57,7 @@ class ErrorCode(IntEnum): INVALID_NODE_TYPE = 52 ENCODE_ERROR = 53 DECODE_ERROR = 54 + FILE_MAP_ERROR = 55 class LibraryError(Exception): @@ -237,6 +238,11 @@ class DecodeError(LibraryError): _default_code = 54 +class FileMapError(LibraryError): + _default_message = "Failed to memory-map file" + _default_code = 55 + + ERROR_MAPPING = { 1: OOMError, 2: NotExistsError, @@ -271,6 +277,7 @@ class DecodeError(LibraryError): 52: InvalidNodeTypeError, 53: EncodeError, 54: DecodeError, + 55: FileMapError, } diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index ed750c79b..5fbb0311d 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -87,6 +87,11 @@ cdef extern from "cwrapper/tsfile_cwrapper.h": TS_COMPRESSION_LZMA2 = 9, TS_COMPRESSION_INVALID = 255 + ctypedef enum TsFileReadBackend: + TSFILE_READ_BACKEND_AUTO = 0 + TSFILE_READ_BACKEND_MMAP = 1 + TSFILE_READ_BACKEND_PREAD = 2 + ctypedef enum ColumnCategory: TAG = 0, FIELD = 1, @@ -223,6 +228,9 @@ cdef extern from "cwrapper/tsfile_cwrapper.h": # Function Declarations + ErrorCode tsfile_set_file_read_backend(int32_t backend) + TsFileReadBackend tsfile_get_file_read_backend() + ctypedef void * TagFilterHandle # reader：new and close diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd index adfe30939..098caa854 100644 --- a/python/tsfile/tsfile_py_cpp.pxd +++ b/python/tsfile/tsfile_py_cpp.pxd @@ -87,3 +87,5 @@ cdef public api object reader_get_timeseries_metadata_c(TsFileReader reader, object device_ids) cpdef public api object get_tsfile_config() cpdef public api void set_tsfile_config(dict new_config) +cpdef public api object get_file_read_backend() +cpdef public api void set_file_read_backend(object backend) diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index 54a48a8c0..79073bd11 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -45,6 +45,7 @@ from tsfile.schema import StringTimeseriesStatistic as StringTimeseriesStatistic from tsfile.schema import TextTimeseriesStatistic as TextTimeseriesStatisticPy from tsfile.schema import TimeseriesStatistic as TimeseriesStatisticPy from tsfile.schema import TimeseriesMetadata as TimeseriesMetadataPy +from tsfile.constants import FileReadBackend as FileReadBackendPy # check exception and set py exception object cdef inline void check_error(int errcode, const char * context=NULL) except*: @@ -866,9 +867,22 @@ cpdef object get_tsfile_config(): "double_encoding_type_": TSEncodingPy(int(g_config_value_.double_encoding_type_)), "string_encoding_type_": TSEncodingPy(int(g_config_value_.string_encoding_type_)), "default_compression_type_": CompressorPy(int(g_config_value_.default_compression_type_)), + "file_read_backend_": FileReadBackendPy(int(tsfile_get_file_read_backend())), } +cpdef object get_file_read_backend(): + """Return the backend configured for subsequently opened readers.""" + return FileReadBackendPy(int(tsfile_get_file_read_backend())) + +cpdef void set_file_read_backend(object backend): + """Select the backend used by subsequently opened readers.""" + if not isinstance(backend, FileReadBackendPy): + raise TypeError(f"Unsupported FileReadBackend: {backend}") + check_error(tsfile_set_file_read_backend( int(backend.value))) + cpdef void set_tsfile_config(dict new_config): + if "file_read_backend_" in new_config: + set_file_read_backend(new_config["file_read_backend_"]) if "tsblock_mem_inc_step_size_" in new_config: _check_uint32(new_config["tsblock_mem_inc_step_size_"]) g_config_value_.tsblock_max_memory_ = new_config["tsblock_mem_inc_step_size_"]