[Feature][C++][Python] Support seekable random-access TsFile sources

.gitignore

@@ -33,6 +33,7 @@ python/tsfile/*so*
 python/tsfile/*dll*
 python/tsfile/*dylib*
 python/tsfile/*.h
+!python/tsfile/python_random_access_read_file.h
 python/tsfile/*.cpp
 python/tsfile/**/*.cpp
 python/tsfile/**/*so*

cpp/README-zh.md

@@ -175,6 +175,9 @@ storage::set_write_thread_count(4);
 
 ### 本地文件读取后端
 
+`LocalRandomAccessReadFile` 是 `RandomAccessReadFile` 的本地文件实现，
+对应的公开头文件为 `file/local_random_access_read_file.h`。
+
 Reader 可以为本地文件选择内存映射 I/O 或传统的定位读取路径。配置会在
 reader 打开文件时确定，因此修改配置不会影响已经打开的 reader。
 

cpp/README.md

@@ -350,6 +350,9 @@ By default, parallel write is enabled when the machine has more than one CPU cor
 
 ### Local File Read Backend
 
+`LocalRandomAccessReadFile` implements `RandomAccessReadFile` for local files. Its
+public header is `file/local_random_access_read_file.h`.
+
 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/bench_mark/bench_mark_src/read_backend_benchmark.cc

@@ -30,7 +30,7 @@
 #include <vector>
 
 #include "common/global.h"
-#include "file/read_file.h"
+#include "file/local_random_access_read_file.h"
 #include "file/tsfile_io_reader.h"
 #include "reader/result_set.h"
 #include "reader/tsfile_reader.h"
@@ -71,7 +71,8 @@ struct QueryPlan {
     int row_count;
 };
 
-typedef std::vector<std::unique_ptr<storage::ReadFile>> OpenFiles;
+typedef std::vector<std::unique_ptr<storage::LocalRandomAccessReadFile>>
+    OpenFiles;
 
 uint64_t update_checksum(uint64_t checksum, const std::vector<char>& buffer,
                          int32_t read_len) {
@@ -201,7 +202,8 @@ int build_query_plan(storage::TsFileReader& reader, QueryPlan& plan,
 bool open_files(const std::vector<std::string>& paths, OpenFiles& files) {
     files.clear();
     for (size_t i = 0; i < paths.size(); ++i) {
-        std::unique_ptr<storage::ReadFile> file(new storage::ReadFile());
+        std::unique_ptr<storage::LocalRandomAccessReadFile> file(
+            new storage::LocalRandomAccessReadFile());
         const int ret = file->open(paths[i]);
         if (ret != common::E_OK) {
             std::cerr << "failed to open " << paths[i] << ": error " << ret
@@ -219,7 +221,7 @@ Result run_sequential(const OpenFiles& files) {
     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];
+        storage::LocalRandomAccessReadFile& file = *files[file_index];
         for (int64_t offset = 0; offset < file.file_size();
              offset += kSequentialBlockSize) {
             int32_t read_len = 0;
@@ -251,7 +253,7 @@ Result run_random(const OpenFiles& files, int32_t requested_block_size,
     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];
+        storage::LocalRandomAccessReadFile& file = *files[file_index];
         const uint64_t file_size = static_cast<uint64_t>(file.file_size());
         const int32_t block_size = static_cast<int32_t>(std::min<uint64_t>(
             file_size, static_cast<uint64_t>(requested_block_size)));
@@ -411,7 +413,7 @@ Result run_concurrent(const std::vector<std::string>& paths) {
     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;
+            storage::LocalRandomAccessReadFile file;
             if (file.open(paths[file_index]) != common::E_OK) {
                 per_file[file_index].success = false;
                 return;

cpp/src/common/config/config.h

@@ -25,7 +25,7 @@
 
 namespace common {
 
-/** Backend used by local ReadFile instances. */
+/** Backend used by local LocalRandomAccessReadFile instances. */
 enum class FileReadBackend : uint8_t {
     AUTO = 0,
     MMAP = 1,

cpp/src/common/global.h

@@ -200,9 +200,9 @@ FORCE_INLINE bool get_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.
+// LocalRandomAccessReadFile instances retain the backend selected when 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();
 

cpp/src/cwrapper/tsfile_cwrapper.cc

@@ -874,8 +874,30 @@ TableSchema tsfile_reader_get_table_schema(TsFileReader reader,
 
 TableSchema* tsfile_reader_get_all_table_schemas(TsFileReader reader,
                                                  uint32_t* size) {
+    ERRNO error_code = common::E_OK;
+    return tsfile_reader_get_all_table_schemas_with_error(reader, size,
+                                                          &error_code);
+}
+
+TableSchema* tsfile_reader_get_all_table_schemas_with_error(TsFileReader reader,
+                                                            uint32_t* size,
+                                                            ERRNO* error_code) {
+    if (size != nullptr) {
+        *size = 0;
+    }
+    if (error_code == nullptr) {
+        return nullptr;
+    }
+    *error_code = common::E_INVALID_ARG;
+    if (reader == nullptr || size == nullptr) {
+        return nullptr;
+    }
     auto* r = static_cast<storage::TsFileReader*>(reader);
-    auto table_schemas = r->get_all_table_schemas();
+    std::vector<std::shared_ptr<storage::TableSchema>> table_schemas;
+    *error_code = r->get_all_table_schemas(table_schemas);
+    if (*error_code != common::E_OK || table_schemas.empty()) {
+        return nullptr;
+    }
     size_t table_num = table_schemas.size();
     TableSchema* ret =
         static_cast<TableSchema*>(malloc(sizeof(TableSchema) * table_num));
@@ -1594,7 +1616,11 @@ ERRNO tsfile_reader_get_all_devices(TsFileReader reader, DeviceID** out_devices,
     *out_devices = nullptr;
     *out_length = 0;
     auto* r = static_cast<storage::TsFileReader*>(reader);
-    const auto ids = r->get_all_devices();
+    std::vector<std::shared_ptr<storage::IDeviceID>> ids;
+    const int ret = r->get_all_devices(ids);
+    if (ret != common::E_OK) {
+        return ret;
+    }
     if (ids.empty()) {
         return common::E_OK;
     }

cpp/src/cwrapper/tsfile_cwrapper.h

@@ -980,10 +980,23 @@ TableSchema tsfile_reader_get_table_schema(TsFileReader reader,
  *
  * @return TableSchema, contains table and column info.
  * @note Caller should call free_table_schema and free to free the ptr.
+ * @note Use tsfile_reader_get_all_table_schemas_with_error to distinguish
+ *       metadata read failures from an empty schema list.
  */
 TableSchema* tsfile_reader_get_all_table_schemas(TsFileReader reader,
                                                  uint32_t* size);
 
+/**
+ * @brief Gets all table schemas and reports metadata read failures.
+ * @return Schema array, or NULL when there are no tables or on error. Check
+ * error_code to distinguish these cases; size is zero on error.
+ * @note Caller must free each schema with free_table_schema, then free the
+ * array.
+ */
+TableSchema* tsfile_reader_get_all_table_schemas_with_error(TsFileReader reader,
+                                                            uint32_t* size,
+                                                            ERRNO* error_code);
+
 /**
  * @brief Gets all timeseries schema in the tsfile.
  *

cpp/src/file/read_file.cc → cpp/src/file/local_random_access_read_file.cc

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-#include "file/read_file.h"
+#include "file/local_random_access_read_file.h"
 
 #include <fcntl.h>
 #include <sys/stat.h>
@@ -59,7 +59,7 @@ uint64_t generation_hash(uint64_t size, int64_t mtime_ns) {
 }
 }  // namespace
 
-ReadFile::ReadFile()
+LocalRandomAccessReadFile::LocalRandomAccessReadFile()
     : file_path_(),
       fd_(-1),
       file_size_(-1),
@@ -79,7 +79,8 @@ ReadFile::ReadFile()
 {
 }
 
-int ReadFile::generation(uint64_t& size, uint64_t& fingerprint) const {
+int LocalRandomAccessReadFile::generation(uint64_t& size,
+                                          uint64_t& fingerprint) const {
     if (!is_opened()) {
         return E_FILE_READ_ERR;
     }
@@ -93,8 +94,8 @@ int ReadFile::generation(uint64_t& size, uint64_t& fingerprint) const {
     return get_file_generation_from_descriptor(size, fingerprint);
 }
 
-int ReadFile::get_file_generation_from_descriptor(uint64_t& size,
-                                                  uint64_t& fingerprint) const {
+int LocalRandomAccessReadFile::get_file_generation_from_descriptor(
+    uint64_t& size, uint64_t& fingerprint) const {
     int64_t mtime_ns = 0;
 #ifdef _WIN32
     if (fd_ < 0) {
@@ -139,7 +140,7 @@ int ReadFile::get_file_generation_from_descriptor(uint64_t& size,
     return E_OK;
 }
 
-void ReadFile::close() {
+void LocalRandomAccessReadFile::close() {
     unmap_file();
     if (fd_ >= 0) {
         ::close(fd_);
@@ -155,7 +156,7 @@ void ReadFile::close() {
 #endif
 }
 
-int ReadFile::open(const std::string& file_path) {
+int LocalRandomAccessReadFile::open(const std::string& file_path) {
     int ret = E_OK;
     close();
     file_path_ = file_path;
@@ -238,7 +239,7 @@ int ReadFile::open(const std::string& file_path) {
     return ret;
 }
 
-int ReadFile::get_file_size(int64_t& file_size) {
+int LocalRandomAccessReadFile::get_file_size(int64_t& file_size) {
 #ifdef _WIN32
     struct __stat64 s;
     if (_fstat64(fd_, &s) < 0) {
@@ -258,7 +259,7 @@ int ReadFile::get_file_size(int64_t& file_size) {
     return E_OK;
 }
 
-int ReadFile::map_file() {
+int LocalRandomAccessReadFile::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;);
 
@@ -354,7 +355,7 @@ int ReadFile::map_file() {
     return E_OK;
 }
 
-void ReadFile::unmap_file() {
+void LocalRandomAccessReadFile::unmap_file() {
     if (mapped_data_ != nullptr) {
 #ifdef _WIN32
         UnmapViewOfFile(mapped_data_);
@@ -372,63 +373,12 @@ void ReadFile::unmap_file() {
     mapped_size_ = 0;
 }
 
-int ReadFile::check_file_magic() {
-    int ret = E_OK;
-    if (file_size_ < MIN_FILE_SIZE) {
-        ret = E_TSFILE_CORRUPTED;
-        LOGE("tsfile" << file_path_.c_str()
-                      << "is corrupted, file_size=" << file_size_);
-    } else {
-        char buf[MAGIC_STRING_TSFILE_LEN];
-        int32_t read_len = 0;
-        // file header magic
-        memset(buf, 0, MAGIC_STRING_TSFILE_LEN);
-        if (RET_FAIL(read(0, buf, MAGIC_STRING_TSFILE_LEN, read_len))) {
-        } else if (read_len != MAGIC_STRING_TSFILE_LEN) {
-            ret = E_TSFILE_CORRUPTED;
-        } else if (memcmp(buf, MAGIC_STRING_TSFILE, MAGIC_STRING_TSFILE_LEN) !=
-                   0) {
-            ret = E_TSFILE_CORRUPTED;
-        }
-        if (IS_FAIL(ret)) {
-            return ret;
-        }
-
-        char version = 0;
-        if (RET_FAIL(read(MAGIC_STRING_TSFILE_LEN, &version, 1, read_len))) {
-        } else if (read_len != 1) {
-            ret = E_TSFILE_CORRUPTED;
-        } else {
-            file_version_ = static_cast<unsigned char>(version);
-            // Version 3 remains readable for backward compatibility; version
-            // 4 is the current writer format.  Other values are not safely
-            // interpretable and must be reported as an input failure before
-            // metadata parsing begins.
-            if (file_version_ != 3 &&
-                file_version_ != static_cast<unsigned char>(VERSION_NUM_BYTE)) {
-                ret = E_UNSUPPORTED_VERSION;
-            }
-        }
-        if (IS_FAIL(ret)) {
-            return ret;
-        }
-
-        // file footer magic
-        memset(buf, 0, MAGIC_STRING_TSFILE_LEN);
-        if (RET_FAIL(read(file_size_ - MAGIC_STRING_TSFILE_LEN, buf,
-                          MAGIC_STRING_TSFILE_LEN, read_len))) {
-        } else if (read_len != MAGIC_STRING_TSFILE_LEN) {
-            ret = E_TSFILE_CORRUPTED;
-        } else if (memcmp(buf, MAGIC_STRING_TSFILE, MAGIC_STRING_TSFILE_LEN) !=
-                   0) {
-            ret = E_TSFILE_CORRUPTED;
-        }
-    }
-    return ret;
+int LocalRandomAccessReadFile::check_file_magic() {
+    return validate_tsfile(*this, &file_version_);
 }
 
-int ReadFile::read(int64_t offset, char* buf, int32_t buf_size,
-                   int32_t& read_len) {
+int LocalRandomAccessReadFile::read(int64_t offset, char* buf, int32_t buf_size,
+                                    int32_t& read_len) {
     read_len = 0;
     if (offset < 0 || buf_size < 0 || (buf == nullptr && buf_size > 0)) {
         return E_INVALID_ARG;