From 4267cb9889f1ae8490e6d260ffc0e1543811e13f Mon Sep 17 00:00:00 2001 From: ColinLee Date: Tue, 10 Nov 2026 10:16:43 +0800 Subject: [PATCH] perf(python): reuse prepared series for repeated single-device table queries Repeated batch queries whose tag filter selects exactly one device rebuilt the device and timeseries-index metadata on every call. query_table now resolves the requested field columns once through a bounded LRU and runs them as prepared queries: one column through query_prepared, several aligned columns through query_prepared_multi. Queries that select a TAG column, use row-by-row mode, or whose filter matches zero or several devices keep the previous path. - Add TsFileReader::prepare_table_series(), which resolves (table, column, tag filter) to a PreparedLocator through the existing device index and parses it once. - Let the prepared query paths take a TsBlock row capacity (block_rows) instead of always allocating PREPARED_BATCH_ROWS, so cached queries keep the caller's batch_size block boundaries. The existing C entry points keep their signatures; tsfile_reader_query_prepared_ex and tsfile_reader_query_prepared_multi_ex carry the new parameter. - Expose prepare_table_series() in Python and cache single-device field queries behind TSFILE_QUERY_CACHE_SIZE (default 1024, 0 disables). Only "not cacheable" error codes fall back to the regular path; I/O, short-read, corruption and OOM errors still reach the caller. close() releases the cached handles. Measured locally (323 devices x 20,000 points, 2048-point windows, 2,000 queries, warm cache): one column 229 us -> 54 us median; two columns 287 us -> 102 us median. Full Python suite: 620 passed. Closes #1081. Relates to #1080. --- cpp/src/cwrapper/tsfile_cwrapper.cc | 58 +++- cpp/src/cwrapper/tsfile_cwrapper.h | 29 ++ .../block/prepared_series_tsblock_reader.cc | 10 +- .../block/prepared_series_tsblock_reader.h | 7 +- cpp/src/reader/tsfile_executor.cc | 8 +- cpp/src/reader/tsfile_executor.h | 4 +- cpp/src/reader/tsfile_reader.cc | 120 +++++++- cpp/src/reader/tsfile_reader.h | 22 +- python/README-zh.md | 23 ++ python/README.md | 27 ++ python/tests/test_query_table_cache.py | 256 ++++++++++++++++++ python/tsfile/tag_filter.py | 38 +++ python/tsfile/tsfile_cpp.pxd | 12 + python/tsfile/tsfile_py_cpp.pxd | 6 + python/tsfile/tsfile_py_cpp.pyx | 29 ++ python/tsfile/tsfile_reader.pyx | 203 +++++++++++++- 16 files changed, 822 insertions(+), 30 deletions(-) create mode 100644 python/tests/test_query_table_cache.py diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc index be97ef484..0dd2d1d39 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.cc +++ b/cpp/src/cwrapper/tsfile_cwrapper.cc @@ -604,6 +604,40 @@ PreparedSeriesHandle tsfile_reader_prepare_series( return handle; } +PreparedSeriesHandle tsfile_reader_prepare_table_series( + TsFileReader reader, const char* table_name, const char* column_name, + TagFilterHandle tag_filter, ERRNO* err_code) { + if (err_code == nullptr) { + return nullptr; + } + *err_code = common::E_INVALID_ARG; + if (reader == nullptr || table_name == nullptr || column_name == nullptr) { + return nullptr; + } + try { + std::shared_ptr prepared; + *err_code = + static_cast(reader)->prepare_table_series( + table_name, column_name, + static_cast(tag_filter), prepared); + if (*err_code != common::E_OK) { + return nullptr; + } + auto* handle = new (std::nothrow) + std::shared_ptr(std::move(prepared)); + if (handle == nullptr) { + *err_code = common::E_OOM; + } + return handle; + } catch (const std::bad_alloc&) { + *err_code = common::E_OOM; + return nullptr; + } catch (...) { + *err_code = common::E_FILE_READ_ERR; + return nullptr; + } +} + PreparedSeriesHandle tsfile_reader_prepare_series_with_time_owner( TsFileReader reader, const TsFilePreparedLocator* locator, PreparedSeriesHandle aligned_time_owner, bool trust_index, @@ -656,6 +690,16 @@ ResultSet tsfile_reader_query_prepared(TsFileReader reader, PreparedSeriesHandle prepared, Timestamp start_time, Timestamp end_time, int offset, int limit, ERRNO* err_code) { + return tsfile_reader_query_prepared_ex( + reader, prepared, start_time, end_time, offset, limit, 0, err_code); +} + +ResultSet tsfile_reader_query_prepared_ex(TsFileReader reader, + PreparedSeriesHandle prepared, + Timestamp start_time, + Timestamp end_time, int offset, + int limit, uint32_t block_rows, + ERRNO* err_code) { if (err_code == nullptr) { return nullptr; } @@ -667,7 +711,7 @@ ResultSet tsfile_reader_query_prepared(TsFileReader reader, static_cast*>(prepared); storage::ResultSet* result = nullptr; *err_code = static_cast(reader)->query_prepared( - *handle, start_time, end_time, offset, limit, result); + *handle, start_time, end_time, offset, limit, result, block_rows); return result; } @@ -675,6 +719,15 @@ ResultSet tsfile_reader_query_prepared_multi( TsFileReader reader, const PreparedSeriesHandle* prepared, uint32_t prepared_count, Timestamp start_time, Timestamp end_time, int offset, int limit, ERRNO* err_code) { + return tsfile_reader_query_prepared_multi_ex( + reader, prepared, prepared_count, start_time, end_time, offset, limit, + 0, err_code); +} + +ResultSet tsfile_reader_query_prepared_multi_ex( + TsFileReader reader, const PreparedSeriesHandle* prepared, + uint32_t prepared_count, Timestamp start_time, Timestamp end_time, + int offset, int limit, uint32_t block_rows, ERRNO* err_code) { if (err_code == nullptr) { return nullptr; } @@ -697,7 +750,8 @@ ResultSet tsfile_reader_query_prepared_multi( storage::ResultSet* result = nullptr; *err_code = static_cast(reader)->query_prepared_multi( - native_prepared, start_time, end_time, offset, limit, result); + native_prepared, start_time, end_time, offset, limit, result, + block_rows); return result; } diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h index 6d9985c01..5c1fe50cb 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.h +++ b/cpp/src/cwrapper/tsfile_cwrapper.h @@ -747,6 +747,15 @@ PreparedSeriesHandle tsfile_reader_prepare_series_with_time_owner( */ void tsfile_prepared_series_free(PreparedSeriesHandle prepared); +/** + * Resolve one table column selected by a tag filter into a reusable series. + * The filter must match exactly one device; otherwise E_NOT_SUPPORT is + * returned and the caller should fall back to the regular table query. + */ +PreparedSeriesHandle tsfile_reader_prepare_table_series( + TsFileReader reader, const char* table_name, const char* column_name, + TagFilterHandle tag_filter, ERRNO* err_code); + /** * Query a prepared series without traversing the TsFile footer index. * An inverted time range succeeds with an empty result set. @@ -756,6 +765,17 @@ ResultSet tsfile_reader_query_prepared(TsFileReader reader, Timestamp start_time, Timestamp end_time, int offset, int limit, ERRNO* err_code); +/** + * Same as tsfile_reader_query_prepared, with an explicit TsBlock row + * capacity. block_rows == 0 keeps the default capacity. + */ +ResultSet tsfile_reader_query_prepared_ex(TsFileReader reader, + PreparedSeriesHandle prepared, + Timestamp start_time, + Timestamp end_time, int offset, + int limit, uint32_t block_rows, + ERRNO* err_code); + /** * Query multiple aligned prepared value columns sharing one time axis. * An inverted time range succeeds with an empty result set. @@ -765,6 +785,15 @@ ResultSet tsfile_reader_query_prepared_multi( uint32_t prepared_count, Timestamp start_time, Timestamp end_time, int offset, int limit, ERRNO* err_code); +/** + * Same as tsfile_reader_query_prepared_multi, with an explicit TsBlock row + * capacity. block_rows == 0 keeps the default capacity. + */ +ResultSet tsfile_reader_query_prepared_multi_ex( + TsFileReader reader, const PreparedSeriesHandle* prepared, + uint32_t prepared_count, Timestamp start_time, Timestamp end_time, + int offset, int limit, uint32_t block_rows, ERRNO* err_code); + /** * @brief Queries time series data from a specific table within time range. * diff --git a/cpp/src/reader/block/prepared_series_tsblock_reader.cc b/cpp/src/reader/block/prepared_series_tsblock_reader.cc index 61e5123a9..4ab1e6093 100644 --- a/cpp/src/reader/block/prepared_series_tsblock_reader.cc +++ b/cpp/src/reader/block/prepared_series_tsblock_reader.cc @@ -50,8 +50,9 @@ common::TSDataType prepared_value_data_type(ITimeseriesIndex* index) { int PreparedSeriesTsBlockReader::init( TsFileIOReader* io_reader, const std::shared_ptr& prepared, - Filter* owned_time_filter, int offset, int limit) { + Filter* owned_time_filter, int offset, int limit, uint32_t block_rows) { owned_time_filter_ = owned_time_filter; + block_rows_ = block_rows == 0 ? PREPARED_BATCH_ROWS : block_rows; if (io_reader == nullptr || prepared == nullptr || prepared->index() == nullptr || offset < 0 || closed_) { return common::E_INVALID_ARG; @@ -92,8 +93,9 @@ int PreparedSeriesTsBlockReader::init( int PreparedSeriesTsBlockReader::init_multi( TsFileIOReader* io_reader, const std::vector>& prepared, - Filter* owned_time_filter, int offset, int limit) { + Filter* owned_time_filter, int offset, int limit, uint32_t block_rows) { owned_time_filter_ = owned_time_filter; + block_rows_ = block_rows == 0 ? PREPARED_BATCH_ROWS : block_rows; if (io_reader == nullptr || prepared.empty() || offset < 0 || closed_) { return common::E_INVALID_ARG; } @@ -149,9 +151,9 @@ int PreparedSeriesTsBlockReader::has_next(bool& has_next) { while (true) { const uint32_t desired_rows = remaining_limit_ > 0 - ? std::min(PREPARED_BATCH_ROWS, + ? std::min(block_rows_, static_cast(remaining_limit_)) - : PREPARED_BATCH_ROWS; + : block_rows_; if (block_ != nullptr) { if (block_->get_max_row_count() != desired_rows) { ssi_->revert_tsblock(); diff --git a/cpp/src/reader/block/prepared_series_tsblock_reader.h b/cpp/src/reader/block/prepared_series_tsblock_reader.h index be416d47f..3e1930f00 100644 --- a/cpp/src/reader/block/prepared_series_tsblock_reader.h +++ b/cpp/src/reader/block/prepared_series_tsblock_reader.h @@ -41,10 +41,12 @@ class PreparedSeriesTsBlockReader final : public TsBlockReader { int init(TsFileIOReader* io_reader, const std::shared_ptr& prepared, - Filter* owned_time_filter, int offset, int limit); + Filter* owned_time_filter, int offset, int limit, + uint32_t block_rows = 0); int init_multi(TsFileIOReader* io_reader, const std::vector>& prepared, - Filter* owned_time_filter, int offset, int limit); + Filter* owned_time_filter, int offset, int limit, + uint32_t block_rows = 0); int has_next(bool& has_next) override; int next(common::TsBlock*& ret_block) override; @@ -64,6 +66,7 @@ class PreparedSeriesTsBlockReader final : public TsBlockReader { common::TsBlock* block_ = nullptr; common::TSDataType value_data_type_ = common::INVALID_DATATYPE; std::vector value_data_types_; + uint32_t block_rows_ = 0; int remaining_limit_ = -1; bool block_ready_ = false; bool exhausted_ = false; diff --git a/cpp/src/reader/tsfile_executor.cc b/cpp/src/reader/tsfile_executor.cc index 3b831dffb..c2aa8d8cd 100644 --- a/cpp/src/reader/tsfile_executor.cc +++ b/cpp/src/reader/tsfile_executor.cc @@ -125,7 +125,7 @@ int TsFileExecutor::prepare_series( int TsFileExecutor::execute_prepared( const std::shared_ptr& prepared, int64_t start_time, int64_t end_time, int offset, int limit, const std::string& column_name, - ResultSet*& ret_qds) { + ResultSet*& ret_qds, uint32_t block_rows) { ASSERT(is_inited_); ret_qds = nullptr; if (prepared == nullptr || offset < 0) { @@ -135,7 +135,7 @@ int TsFileExecutor::execute_prepared( new PreparedSeriesTsBlockReader()); int ret = tsblock_reader->init(&io_reader_, prepared, new TimeBetween(start_time, end_time, false), - offset, limit); + offset, limit, block_rows); if (ret != E_OK) { return ret; } @@ -150,7 +150,7 @@ int TsFileExecutor::execute_prepared( int TsFileExecutor::execute_prepared_multi( const std::vector>& prepared, int64_t start_time, int64_t end_time, int offset, int limit, - ResultSet*& ret_qds) { + ResultSet*& ret_qds, uint32_t block_rows) { ASSERT(is_inited_); ret_qds = nullptr; if (prepared.empty() || offset < 0) { @@ -161,7 +161,7 @@ int TsFileExecutor::execute_prepared_multi( new PreparedSeriesTsBlockReader()); int ret = tsblock_reader->init_multi( &io_reader_, prepared, new TimeBetween(start_time, end_time, false), - offset, limit); + offset, limit, block_rows); if (ret != E_OK) { return ret; } diff --git a/cpp/src/reader/tsfile_executor.h b/cpp/src/reader/tsfile_executor.h index cd862b836..57927d5dc 100644 --- a/cpp/src/reader/tsfile_executor.h +++ b/cpp/src/reader/tsfile_executor.h @@ -48,11 +48,11 @@ class TsFileExecutor // : public QueryExecutor int execute_prepared(const std::shared_ptr& prepared, int64_t start_time, int64_t end_time, int offset, int limit, const std::string& column_name, - ResultSet*& ret_qds); + ResultSet*& ret_qds, uint32_t block_rows = 0); int execute_prepared_multi( const std::vector>& prepared, int64_t start_time, int64_t end_time, int offset, int limit, - ResultSet*& ret_qds); + ResultSet*& ret_qds, uint32_t block_rows = 0); void destroy_query_data_set(ResultSet* qds); int get_tsfile_meta(TsFileMeta*& tsfile_meta) { return io_reader_.get_tsfile_meta(tsfile_meta); diff --git a/cpp/src/reader/tsfile_reader.cc b/cpp/src/reader/tsfile_reader.cc index 03755e4c7..eed139ec7 100644 --- a/cpp/src/reader/tsfile_reader.cc +++ b/cpp/src/reader/tsfile_reader.cc @@ -26,6 +26,7 @@ #include "common/allocator/byte_stream.h" #include "common/schema.h" #include "common/tsfile_common.h" +#include "device_meta_iterator.h" #include "file/local_random_access_read_file.h" #include "filter/time_operator.h" #include "tsfile_executor.h" @@ -295,23 +296,134 @@ int TsFileReader::prepare_series( aligned_time_owner, prepared); } +int TsFileReader::prepare_table_series( + const std::string& table_name, const std::string& column_name, + Filter* tag_filter, std::shared_ptr& prepared) { + int ret = E_OK; + prepared.reset(); + if (read_file_ == nullptr || tsfile_executor_ == nullptr) { + return E_INVALID_ARG; + } + TsFileIOReader* io_reader = tsfile_executor_->get_tsfile_io_reader(); + if (io_reader == nullptr) { + return E_INVALID_ARG; + } + TsFileMeta* file_metadata = nullptr; + if (RET_FAIL(io_reader->get_tsfile_meta(file_metadata))) { + return ret; + } + const std::string table = to_lower(table_name); + std::shared_ptr table_schema; + if (RET_FAIL(file_metadata->get_table_schema(table, table_schema))) { + return ret; + } + MetaIndexNode* table_root = nullptr; + file_metadata->get_table_metaindex_node(table, table_root); + if (table_root == nullptr) { + return E_DEVICE_NOT_EXIST; + } + + common::PageArena pa; + pa.init(512, common::MOD_TSFILE_READER); + DeviceMetaIterator device_iterator(io_reader, table_root, tag_filter); + bool has_next = false; + if (RET_FAIL(device_iterator.has_next(has_next))) { + return ret; + } + if (!has_next) { + return E_DEVICE_NOT_EXIST; + } + std::pair, MetaIndexNode*> device_meta; + if (RET_FAIL(device_iterator.next(device_meta))) { + return ret; + } + if (RET_FAIL(device_iterator.has_next(has_next))) { + return ret; + } + if (has_next) { + // Several devices match: a single prepared series would not reproduce + // the result of the regular query, so let the caller fall back. + return E_NOT_SUPPORT; + } + if (device_meta.first == nullptr) { + return E_DEVICE_NOT_EXIST; + } + + std::unordered_set measurement_names; + measurement_names.insert(to_lower(column_name)); + std::vector timeseries_indexs(1, nullptr); + if (RET_FAIL(io_reader->get_timeseries_indexes( + device_meta.first, measurement_names, timeseries_indexs, pa))) { + return ret; + } + ITimeseriesIndex* index = timeseries_indexs[0]; + if (index == nullptr) { + return E_MEASUREMENT_NOT_EXIST; + } + + PreparedLocator locator; + if (auto* aligned = dynamic_cast(index)) { + if (aligned->value_ts_idx_ == nullptr) { + return E_TYPE_NOT_MATCH; + } + locator.layout = 1; + const auto value_offset = aligned->value_ts_idx_->get_metadata_offset(); + const auto value_length = aligned->value_ts_idx_->get_metadata_length(); + if (value_offset < 0 || value_length == 0) { + return E_NOT_SUPPORT; + } + locator.value_metadata_offset = static_cast(value_offset); + locator.value_metadata_length = value_length; + if (aligned->time_ts_idx_ != nullptr) { + const auto time_offset = + aligned->time_ts_idx_->get_metadata_offset(); + const auto time_length = + aligned->time_ts_idx_->get_metadata_length(); + if (time_offset >= 0 && time_length > 0) { + locator.time_metadata_offset = + static_cast(time_offset); + locator.time_metadata_length = time_length; + } + } + } else if (auto* value = dynamic_cast(index)) { + locator.layout = 0; + const auto value_offset = value->get_metadata_offset(); + const auto value_length = value->get_metadata_length(); + if (value_offset < 0 || value_length == 0) { + return E_NOT_SUPPORT; + } + locator.value_metadata_offset = static_cast(value_offset); + locator.value_metadata_length = value_length; + } else { + return E_TYPE_NOT_MATCH; + } + + FileGeneration generation; + // file_fingerprint 0 means "not checked"; the size is validated by + // TsFileIOReader::prepare_series(). + generation.file_size = static_cast(read_file_->file_size()); + return io_reader->prepare_series(generation, locator, prepared); +} + int TsFileReader::query_prepared( const std::shared_ptr& prepared, int64_t start_time, - int64_t end_time, int offset, int limit, ResultSet*& result_set) { + int64_t end_time, int offset, int limit, ResultSet*& result_set, + uint32_t block_rows) { if (prepared == nullptr || prepared->index() == nullptr) { return E_INVALID_ARG; } return tsfile_executor_->execute_prepared( prepared, start_time, end_time, offset, limit, - prepared->index()->get_measurement_name().to_std_string(), result_set); + prepared->index()->get_measurement_name().to_std_string(), result_set, + block_rows); } int TsFileReader::query_prepared_multi( const std::vector>& prepared, int64_t start_time, int64_t end_time, int offset, int limit, - ResultSet*& result_set) { + ResultSet*& result_set, uint32_t block_rows) { return tsfile_executor_->execute_prepared_multi( - prepared, start_time, end_time, offset, limit, result_set); + prepared, start_time, end_time, offset, limit, result_set, block_rows); } int TsFileReader::queryByRow(const std::string& table_name, diff --git a/cpp/src/reader/tsfile_reader.h b/cpp/src/reader/tsfile_reader.h index 38a27a760..6f2c7d340 100644 --- a/cpp/src/reader/tsfile_reader.h +++ b/cpp/src/reader/tsfile_reader.h @@ -153,13 +153,31 @@ class TsFileReader { const FileGeneration& generation, const PreparedLocator& locator, const std::shared_ptr& aligned_time_owner, std::shared_ptr& prepared); + /** + * @brief Resolve one table column selected by @p tag_filter into a reusable + * prepared series. + * + * The device lookup and timeseries-index parse happen once; repeated + * `query_prepared()` calls then skip both. The filter must match exactly + * one device, otherwise E_NOT_SUPPORT is returned so that callers can fall + * back to the regular table query path. + * + * @param [in] table_name the table name + * @param [in] column_name the measurement (field) column name + * @param [in] tag_filter the device tag filter; nullptr means no filter + * @param [out] prepared the prepared series + */ + int prepare_table_series(const std::string& table_name, + const std::string& column_name, Filter* tag_filter, + std::shared_ptr& prepared); int query_prepared(const std::shared_ptr& prepared, int64_t start_time, int64_t end_time, int offset, - int limit, ResultSet*& result_set); + int limit, ResultSet*& result_set, + uint32_t block_rows = 0); int query_prepared_multi( const std::vector>& prepared, int64_t start_time, int64_t end_time, int offset, int limit, - ResultSet*& result_set); + ResultSet*& result_set, uint32_t block_rows = 0); /** * @brief Query table-model data by row with offset/limit pushdown. diff --git a/python/README-zh.md b/python/README-zh.md index 59d2eb0f1..0236e3370 100644 --- a/python/README-zh.md +++ b/python/README-zh.md @@ -183,3 +183,26 @@ with fsspec.open("s3://bucket/example.tsfile", "rb") as source: 文件对象必须提供 `seek()`、`tell()` 和带明确长度的二进制 `read(size)`。 该对象仍由调用方管理：`TsFileReader` 会在使用期间保持其存活、恢复其游标位置， 但不会关闭它。本地 `FileReadBackend` 配置不适用于文件对象。 + +## 重复点查 + +批量查询在 tag 过滤条件只命中一个设备时，会在 reader 上缓存已解析的字段列。 +对同一 (表, tag 过滤条件, 列组合) 的重复查询不再重新解析设备与 timeseries 索引； +单列走 `query_prepared`，多列走 `query_prepared_multi`。缓存是按 +`TSFILE_QUERY_CACHE_SIZE` 限制的 LRU（默认 `1024`，`0` 关闭），查询结果不变。 +同时选择 TAG 列、行模式查询、以及过滤命中 0 个或多个设备的情况保持原路径。 + +```python +reader = TsFileReader("data.tsfile") +for start, end in windows: + with reader.query_table( + "sensors", ["value"], start, end, + tag_filter=tag_eq("device", "dev_a"), batch_size=4096, + ) as result: + ... +``` + +需要显式控制时，`reader.prepare_table_series(table, column, tag_filter)` +只解析一列并返回句柄，可配合 `reader.query_prepared(...)` 使用； +`reader.query_prepared_multi([...])` 用于多列对齐读取。过滤条件必须只命中 +一个设备，否则自动缓存会退回普通查询路径。 diff --git a/python/README.md b/python/README.md index c99d11f9d..18018eb0e 100644 --- a/python/README.md +++ b/python/README.md @@ -197,3 +197,30 @@ The object must provide `seek()`, `tell()`, and explicitly sized binary `read(size)` operations. The caller owns the object: `TsFileReader` keeps it alive while in use, preserves its cursor position, and does not close it. The local `FileReadBackend` setting does not apply to file objects. + +## Repeated point queries + +A batch query whose tag filter selects exactly one device caches the resolved +field columns on the reader. Repeated queries for the same +(table, tag filter, columns) skip the device and timeseries-index lookup; one +column is served by `query_prepared`, several aligned columns by +`query_prepared_multi`. The cache is an LRU bounded by +`TSFILE_QUERY_CACHE_SIZE` (default `1024`; `0` disables it), and results are +unchanged. Queries that also select a TAG column, row-by-row queries, and +filters that select zero or several devices keep the regular path. + +```python +reader = TsFileReader("data.tsfile") +for start, end in windows: + with reader.query_table( + "sensors", ["value"], start, end, + tag_filter=tag_eq("device", "dev_a"), batch_size=4096, + ) as result: + ... +``` + +For explicit control, `reader.prepare_table_series(table, column, tag_filter)` +resolves one column once and returns a handle usable with +`reader.query_prepared(...)`; `reader.query_prepared_multi([...])` serves +several aligned columns together. The filter must select exactly one device; +otherwise the automatic cache is bypassed. diff --git a/python/tests/test_query_table_cache.py b/python/tests/test_query_table_cache.py new file mode 100644 index 000000000..dca9a2e54 --- /dev/null +++ b/python/tests/test_query_table_cache.py @@ -0,0 +1,256 @@ +# 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. +# + +"""Repeated table queries reuse prepared series (TSFILE_QUERY_CACHE_SIZE).""" + +import os + +import pytest + +from tsfile import ( + ColumnCategory, + ColumnSchema, + TableSchema, + Tablet, + TSDataType, + TsFileReader, + TsFileTableWriter, + tag_eq, + tag_gteq, +) + +TSFILE_PATH = "test_query_table_cache.tsfile" +TABLE = "sensors" +SCHEMA = TableSchema( + TABLE, + [ + ColumnSchema("region", TSDataType.STRING, ColumnCategory.TAG), + ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG), + ColumnSchema("value", TSDataType.DOUBLE, ColumnCategory.FIELD), + ColumnSchema("other", TSDataType.DOUBLE, ColumnCategory.FIELD), + ], +) + +# device -> (region, timestamps, values); None leaves the cell unset (null). +DEVICES = { + "dev_a": ("north", list(range(100)), [float(i) for i in range(100)]), + "dev_b": ("north", list(range(50, 150)), [float(i) for i in range(50, 150)]), + "dev_c": ("south", list(range(0, 200, 2)), [float(i) for i in range(0, 200, 2)]), + "dev_null": ( + "south", + list(range(10)), + [None if i % 3 else float(i) for i in range(10)], + ), +} + + +@pytest.fixture(scope="module", autouse=True) +def create_tsfile(): + if os.path.exists(TSFILE_PATH): + os.remove(TSFILE_PATH) + with TsFileTableWriter(TSFILE_PATH, SCHEMA) as writer: + for device, (region, timestamps, values) in DEVICES.items(): + tablet = Tablet( + ["region", "device", "value", "other"], + [ + TSDataType.STRING, + TSDataType.STRING, + TSDataType.DOUBLE, + TSDataType.DOUBLE, + ], + len(timestamps), + ) + for i, (timestamp, value) in enumerate(zip(timestamps, values)): + tablet.add_timestamp(i, timestamp) + tablet.add_value_by_name("region", i, region) + tablet.add_value_by_name("device", i, device) + if value is not None: + tablet.add_value_by_name("value", i, value) + tablet.add_value_by_name("other", i, value * 2.0) + writer.write_table(tablet) + yield + if os.path.exists(TSFILE_PATH): + os.remove(TSFILE_PATH) + + +def collect(reader, tag, start=None, end=None, columns=("value",), batch_size=8): + kwargs = {} + if start is not None: + kwargs["start_time"] = start + if end is not None: + kwargs["end_time"] = end + with reader.query_table( + TABLE, + list(columns), + tag_filter=tag_eq("device", tag), + batch_size=batch_size, + **kwargs, + ) as result: + batches = [] + while True: + batch = result.read_arrow_batch() + if batch is None: + break + batches.append(batch.to_pydict()) + return batches + + +def fresh_reader(cache_size=None): + if cache_size is None: + os.environ.pop("TSFILE_QUERY_CACHE_SIZE", None) + else: + os.environ["TSFILE_QUERY_CACHE_SIZE"] = str(cache_size) + return TsFileReader(TSFILE_PATH) + + +@pytest.fixture(autouse=True) +def restore_env(): + previous = os.environ.get("TSFILE_QUERY_CACHE_SIZE") + yield + if previous is None: + os.environ.pop("TSFILE_QUERY_CACHE_SIZE", None) + else: + os.environ["TSFILE_QUERY_CACHE_SIZE"] = previous + + +@pytest.mark.parametrize("device", list(DEVICES)) +def test_cached_results_match_uncached(device): + with fresh_reader() as cached, fresh_reader(0) as uncached: + for _ in range(3): + assert collect(cached, device) == collect(uncached, device) + + +@pytest.mark.parametrize( + "start,end", + [(None, None), (0, 49), (50, 149), (10, 10), (150, 149)], +) +def test_time_ranges_match_uncached(start, end): + with fresh_reader() as cached, fresh_reader(0) as uncached: + assert collect(cached, "dev_c", start, end) == collect( + uncached, "dev_c", start, end + ) + + +def test_repeated_queries_reuse_one_prepared_series(): + with fresh_reader() as reader: + for _ in range(5): + collect(reader, "dev_a") + assert reader.prepared_query_cache_size() == 1 + collect(reader, "dev_b") + assert reader.prepared_query_cache_size() == 2 + reader.clear_prepared_query_cache() + assert reader.prepared_query_cache_size() == 0 + assert collect(reader, "dev_a") + + +def test_multi_device_filter_does_not_use_single_series_cache(): + with fresh_reader() as cached, fresh_reader(0) as uncached: + # region == south matches two devices: the cache must fall back to the + # regular query path and still return both devices. + def region_rows(reader): + with reader.query_table( + TABLE, ["value"], tag_filter=tag_eq("region", "south"), batch_size=8 + ) as result: + rows = [] + while True: + batch = result.read_arrow_batch() + if batch is None: + break + rows.append(batch.to_pydict()) + return rows + + expected = region_rows(uncached) + for _ in range(3): + assert region_rows(cached) == expected + # The negative entry keeps the reader from retrying every call. + assert cached.prepared_query_cache_size() == 1 + + +def test_cache_eviction_keeps_results(): + with fresh_reader(1) as cached, fresh_reader(0) as uncached: + for device in ("dev_a", "dev_b", "dev_c", "dev_a"): + assert collect(cached, device) == collect(uncached, device) + assert cached.prepared_query_cache_size() <= 1 + + +def test_row_mode_unchanged(): + with fresh_reader() as cached, fresh_reader(0) as uncached: + # Row-by-row mode never uses the cache. + def rows(reader): + with reader.query_table( + TABLE, + ["value"], + start_time=0, + end_time=99, + tag_filter=tag_eq("device", "dev_a"), + batch_size=0, + ) as result: + out = [] + while result.next(): + out.append(result.get_value_by_name("value")) + return out + + assert rows(cached) == rows(uncached) + assert cached.prepared_query_cache_size() == 0 + + +def test_two_columns_are_cached_and_match_uncached(): + with fresh_reader() as cached, fresh_reader(0) as uncached: + for columns in (("value", "other"), ("other", "value")): + expected = collect(uncached, "dev_a", columns=columns) + for _ in range(3): + assert collect(cached, "dev_a", columns=columns) == expected + # One cache entry per distinct column combination. + assert cached.prepared_query_cache_size() == 2 + + +def test_cached_multi_column_keeps_requested_order(): + with fresh_reader() as reader: + batches = collect(reader, "dev_a", columns=("other", "value")) + assert list(batches[0]) == ["time", "other", "value"] + + +def test_tag_column_selection_falls_back(): + with fresh_reader() as cached, fresh_reader(0) as uncached: + expected = collect(uncached, "dev_a", columns=("value", "region")) + for _ in range(3): + assert collect(cached, "dev_a", columns=("value", "region")) == expected + + +def test_compound_tag_filter_is_cacheable(): + with fresh_reader() as cached, fresh_reader(0) as uncached: + + def query(reader): + with reader.query_table( + TABLE, + ["value"], + tag_filter=tag_eq("region", "north") & tag_gteq("device", "dev_b"), + batch_size=8, + ) as result: + batches = [] + while True: + batch = result.read_arrow_batch() + if batch is None: + break + batches.append(batch.to_pydict()) + return batches + + expected = query(uncached) + for _ in range(3): + assert query(cached) == expected + assert cached.prepared_query_cache_size() == 1 diff --git a/python/tsfile/tag_filter.py b/python/tsfile/tag_filter.py index ac6e46c5c..3e6b9c5ff 100644 --- a/python/tsfile/tag_filter.py +++ b/python/tsfile/tag_filter.py @@ -121,6 +121,44 @@ def __repr__(self): return f"NOT({self.filter})" +def cache_key(tag_filter): + """Return a hashable structural key for a tag filter. + + ``None`` means the filter is absent or contains an unknown node, in which + case callers must not cache a resolved device for it. + """ + if tag_filter is None: + return None + if isinstance(tag_filter, ComparisonTagFilter): + return ("cmp", tag_filter.column_name, tag_filter.op, tag_filter.value) + if isinstance(tag_filter, BetweenTagFilter): + return ( + "between", + tag_filter.column_name, + tag_filter.lower, + tag_filter.upper, + tag_filter.is_not, + ) + if isinstance(tag_filter, AndTagFilter): + left = cache_key(tag_filter.left) + right = cache_key(tag_filter.right) + if left is None or right is None: + return None + return ("and", left, right) + if isinstance(tag_filter, OrTagFilter): + left = cache_key(tag_filter.left) + right = cache_key(tag_filter.right) + if left is None or right is None: + return None + return ("or", left, right) + if isinstance(tag_filter, NotTagFilter): + inner = cache_key(tag_filter.filter) + if inner is None: + return None + return ("not", inner) + return None + + # Convenience factory functions def tag_eq(column_name: str, value: str) -> TagFilter: """Create a tag equality filter: column == value.""" diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index d5f2539cf..67c26e4fb 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -312,15 +312,27 @@ cdef extern from "cwrapper/tsfile_cwrapper.h": TsFileReader reader, const TsFilePreparedLocator * locator, PreparedSeriesHandle aligned_time_owner, bint trust_index, ErrorCode * err_code) nogil + PreparedSeriesHandle tsfile_reader_prepare_table_series( + TsFileReader reader, const char * table_name, const char * column_name, + TagFilterHandle tag_filter, ErrorCode * err_code) nogil void tsfile_prepared_series_free(PreparedSeriesHandle prepared) ResultSet tsfile_reader_query_prepared( TsFileReader reader, PreparedSeriesHandle prepared, int64_t start_time, int64_t end_time, int offset, int limit, ErrorCode * err_code) nogil + ResultSet tsfile_reader_query_prepared_ex( + TsFileReader reader, PreparedSeriesHandle prepared, + int64_t start_time, int64_t end_time, int offset, int limit, + uint32_t block_rows, ErrorCode * err_code) nogil ResultSet tsfile_reader_query_prepared_multi( TsFileReader reader, const PreparedSeriesHandle * prepared, uint32_t prepared_count, int64_t start_time, int64_t end_time, int offset, int limit, ErrorCode * err_code) nogil + ResultSet tsfile_reader_query_prepared_multi_ex( + TsFileReader reader, const PreparedSeriesHandle * prepared, + uint32_t prepared_count, int64_t start_time, int64_t end_time, + int offset, int limit, uint32_t block_rows, + ErrorCode * err_code) nogil ResultSet tsfile_query_table_on_tree(TsFileReader reader, char** columns, uint32_t column_num, diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd index b7106a03e..442da823f 100644 --- a/python/tsfile/tsfile_py_cpp.pxd +++ b/python/tsfile/tsfile_py_cpp.pxd @@ -51,9 +51,15 @@ cdef public api PreparedSeriesHandle tsfile_reader_prepare_series_c( cdef public api PreparedSeriesHandle tsfile_reader_prepare_series_with_time_owner_c( TsFileReader reader, object locator, PreparedSeriesHandle aligned_time_owner, bint trust_index) except NULL +cdef public api PreparedSeriesHandle tsfile_reader_prepare_table_series_c( + TsFileReader reader, object table_name, object column_name, + TagFilterHandle tag_filter) except NULL cdef public api ResultSet tsfile_reader_query_prepared_c( TsFileReader reader, PreparedSeriesHandle prepared, int64_t start_time, int64_t end_time, int offset, int limit) +cdef public api ResultSet tsfile_reader_query_prepared_ex_c( + TsFileReader reader, PreparedSeriesHandle prepared, int64_t start_time, + int64_t end_time, int offset, int limit, uint32_t block_rows) cdef public api ErrorCode tsfile_writer_register_device_py_cpp(TsFileWriter writer, DeviceSchema *schema) cdef public api ErrorCode tsfile_writer_register_timeseries_py_cpp(TsFileWriter writer, object device_name, TimeseriesSchema *schema) diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index b6a7795b6..2d4903723 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -881,6 +881,22 @@ cdef PreparedSeriesHandle tsfile_reader_prepare_series_with_time_owner_c( check_error(code, b"Failed to prepare aligned Dataset Index locator") return prepared +cdef PreparedSeriesHandle tsfile_reader_prepare_table_series_c( + TsFileReader reader, object table_name, object column_name, + TagFilterHandle tag_filter) except NULL: + cdef ErrorCode code = 0 + cdef bytes table_bytes = table_name.encode('utf-8') + cdef bytes column_bytes = column_name.encode('utf-8') + cdef const char * c_table = table_bytes + cdef const char * c_column = column_bytes + cdef PreparedSeriesHandle prepared + with nogil: + prepared = tsfile_reader_prepare_table_series( + reader, c_table, c_column, tag_filter, &code + ) + check_error(code, b"Failed to prepare table series") + return prepared + cdef ResultSet tsfile_reader_query_prepared_c( TsFileReader reader, PreparedSeriesHandle prepared, int64_t start_time, int64_t end_time, int offset, int limit): @@ -892,6 +908,19 @@ cdef ResultSet tsfile_reader_query_prepared_c( check_error(code, b"Failed to query prepared series") return result +cdef ResultSet tsfile_reader_query_prepared_ex_c( + TsFileReader reader, PreparedSeriesHandle prepared, + int64_t start_time, int64_t end_time, int offset, int limit, + uint32_t block_rows): + cdef ErrorCode code = 0 + cdef ResultSet result + with nogil: + result = tsfile_reader_query_prepared_ex( + reader, prepared, start_time, end_time, offset, limit, + block_rows, &code) + check_error(code, b"Failed to query prepared series") + return result + cpdef object get_tsfile_config(): return { "tsblock_mem_inc_step_size_": g_config_value_.tsblock_mem_inc_step_size_, diff --git a/python/tsfile/tsfile_reader.pyx b/python/tsfile/tsfile_reader.pyx index c4fa4fd7a..806c3178d 100644 --- a/python/tsfile/tsfile_reader.pyx +++ b/python/tsfile/tsfile_reader.pyx @@ -20,6 +20,7 @@ import os import weakref +from collections import OrderedDict from typing import List, Optional, Dict import pandas as pd @@ -33,7 +34,7 @@ from libc.stdint cimport INT64_MIN, INT64_MAX, uint32_t, uintptr_t from tsfile.schema import TSDataType as TSDataTypePy from tsfile.schema import DeviceID, DeviceTimeseriesMetadataGroup from tsfile.exceptions import TsFileCorruptedError -from tsfile.tag_filter import ComparisonTagFilter, BetweenTagFilter, AndTagFilter, OrTagFilter, NotTagFilter +from tsfile.tag_filter import ComparisonTagFilter, BetweenTagFilter, AndTagFilter, OrTagFilter, NotTagFilter, cache_key from .date_utils import parse_int_to_date from .tsfile_cpp cimport * from .tsfile_py_cpp cimport * @@ -42,6 +43,23 @@ cdef extern from "python_random_access_read_file.h": TsFileReader create_tsfile_reader_from_python_file( PyObject* source, int32_t* error_code) except? NULL + +def _query_cache_size(): + """Per-reader prepared-query cache size, from TSFILE_QUERY_CACHE_SIZE.""" + raw = os.environ.get("TSFILE_QUERY_CACHE_SIZE", "1024") + try: + value = int(raw) + except (TypeError, ValueError): + return 1024 + return value if value > 0 else 0 + + +# Errors that mean "this query cannot be served by a prepared series" (invalid +# argument, type mismatch, not supported, missing table/device/measurement). +# I/O, short-read, corruption and OOM errors are not listed: they must reach +# the caller instead of silently falling back to a second read. +_CACHE_FALLBACK_CODES = frozenset((4, 27, 40, 44, 45, 49, 50)) + cdef class ResultSetPy: """ Get data from a query result. When reader run a query, a query handler will return. @@ -370,6 +388,11 @@ cdef class TsFileReaderPy: cdef TsFileReader reader cdef object activate_result_set_list + # LRU of prepared single-column table queries, keyed by + # (table, tag-filter key, column). Values are PreparedSeriesPy or None for + # filters that select several devices (negative cache). + cdef object _prepared_query_cache + cdef int _prepared_query_cache_size def __init__(self, source): """ @@ -377,6 +400,8 @@ cdef class TsFileReaderPy: """ self.reader = NULL self.activate_result_set_list = weakref.WeakSet() + self._prepared_query_cache = OrderedDict() + self._prepared_query_cache_size = _query_cache_size() self.init_reader(source) cdef init_reader(self, source): @@ -393,12 +418,42 @@ cdef class TsFileReaderPy: tag_filter = None, batch_size : int = 0) -> ResultSetPy: """ Execute a time range query on specified table and columns. + + Batch queries whose tag filter selects exactly one device reuse cached + prepared series, so repeated point queries skip the device and + timeseries-index lookup. Any number of field columns is supported; a + query that also selects a TAG column, or whose filter selects several + devices, keeps the regular path. The cache size is set by + ``TSFILE_QUERY_CACHE_SIZE`` (default 1024, 0 disables it). + :param tag_filter: Optional TagFilter to filter by TAG column values. :param batch_size: <= 0 for row-by-row mode; > 0 for batch (TsBlock) mode. :return: query result handler. """ cdef ResultSet result cdef TagFilterHandle c_tag_filter = NULL + cdef ResultSetPy pyresult + if (self._prepared_query_cache_size > 0 and batch_size > 0 + and len(column_names) > 0 and tag_filter is not None): + key = self._prepared_query_key(table_name, column_names, tag_filter) + if key is not None: + prepared = self._prepared_query_for(key, tag_filter) + if prepared is not None: + try: + pyresult = self._run_prepared( + prepared, start_time, end_time, batch_size) + except Exception as exc: + if (getattr(exc, "code", None) + not in _CACHE_FALLBACK_CODES): + raise + # The prepared form cannot serve this query (for + # example a non-aligned multi-column read); remember + # that and use the regular path. + self._prepared_query_cache[key] = None + self._close_prepared_entry(prepared) + else: + pyresult.metadata.set_table_name(table_name) + return pyresult pyresult = ResultSetPy(self) if tag_filter is not None: c_tag_filter = self._build_c_tag_filter(table_name.lower(), tag_filter) @@ -417,6 +472,104 @@ cdef class TsFileReaderPy: self.activate_result_set_list.add(pyresult) return pyresult + def prepare_table_series(self, table_name : str, column_name : str, + tag_filter = None) -> PreparedSeriesPy: + """ + Resolve a table column selected by a tag filter once, for repeated queries. + + The filter must match exactly one device; otherwise an exception is + raised and the regular ``query_table`` path should be used instead. + Subsequent reads go through :meth:`query_prepared`. + + :param table_name: the table name. + :param column_name: the measurement (field) column name. + :param tag_filter: Optional TagFilter selecting the device. + :return: reusable prepared series handle. + """ + cdef TagFilterHandle c_tag_filter = NULL + table = table_name.lower() + column = column_name.lower() + if tag_filter is not None: + c_tag_filter = self._build_c_tag_filter(table, tag_filter) + try: + prepared = tsfile_reader_prepare_table_series_c( + self.reader, table, column, c_tag_filter) + finally: + if c_tag_filter != NULL: + tsfile_tag_filter_free(c_tag_filter) + py_prepared = PreparedSeriesPy() + py_prepared.init_c(prepared) + return py_prepared + + def prepared_query_cache_size(self): + """Return the number of prepared table queries currently cached.""" + return len(self._prepared_query_cache) + + def clear_prepared_query_cache(self): + """Drop every cached prepared table query.""" + for prepared in self._prepared_query_cache.values(): + self._close_prepared_entry(prepared) + self._prepared_query_cache.clear() + + cdef object _prepared_query_key(self, str table_name, object column_names, + object tag_filter): + cdef object filter_key = cache_key(tag_filter) + if filter_key is None: + return None + return ( + table_name.lower(), filter_key, + tuple(column.lower() for column in column_names)) + + cdef object _run_prepared(self, object prepared, int64_t start_time, + int64_t end_time, int block_rows): + if isinstance(prepared, tuple): + return self.query_prepared_multi( + list(prepared), start_time, end_time, 0, -1, block_rows) + return self.query_prepared( + prepared, start_time, end_time, 0, -1, block_rows) + + cdef object _prepared_query_for(self, object key, object tag_filter): + """Return cached prepared series (one or a tuple), or None to fall back.""" + if key in self._prepared_query_cache: + prepared = self._prepared_query_cache[key] + self._prepared_query_cache.move_to_end(key) + return prepared + prepared = self._prepare_series_group(key[0], tuple(key[2]), tag_filter) + # A None value is a negative cache entry: the filter selects no device + # or several devices, or a requested column is not a stored field, so + # the regular query path stays in charge. + self._prepared_query_cache[key] = prepared + while len(self._prepared_query_cache) > self._prepared_query_cache_size: + _, evicted = self._prepared_query_cache.popitem(last=False) + self._close_prepared_entry(evicted) + return prepared + + def _prepare_series_group(self, table_name, column_names, tag_filter): + """Prepare every requested column of one single-device query.""" + prepared = [] + try: + for column in column_names: + prepared.append( + self.prepare_table_series(table_name, column, tag_filter)) + except Exception as exc: + for entry in prepared: + entry.close() + if getattr(exc, "code", None) not in _CACHE_FALLBACK_CODES: + raise + return None + if len(prepared) == 1: + return prepared[0] + return tuple(prepared) + + cdef void _close_prepared_entry(self, object entry): + if entry is None: + return + if isinstance(entry, tuple): + for item in entry: + item.close() + else: + entry.close() + cdef TagFilterHandle _build_c_tag_filter(self, str table_name, object tag_filter): """Recursively build C TagFilterHandle from Python TagFilter tree.""" cdef ErrorCode code = 0 @@ -575,11 +728,24 @@ cdef class TsFileReaderPy: def query_prepared(self, PreparedSeriesPy prepared, start_time : int = INT64_MIN, end_time : int = INT64_MAX, - offset : int = 0, limit : int = -1) -> ResultSetPy: + offset : int = 0, limit : int = -1, + block_rows : int = 0) -> ResultSetPy: + """Query a prepared series. + + ``block_rows`` bounds each returned TsBlock; 0 keeps the native + default capacity. It does not limit the number of rows returned. + """ if prepared.prepared == NULL: raise RuntimeError("PreparedSeries is closed") - cdef ResultSet result = tsfile_reader_query_prepared_c( - self.reader, prepared.prepared, start_time, end_time, offset, limit) + cdef ResultSet result + if block_rows > 0: + result = tsfile_reader_query_prepared_ex_c( + self.reader, prepared.prepared, start_time, end_time, offset, + limit, block_rows) + else: + result = tsfile_reader_query_prepared_c( + self.reader, prepared.prepared, start_time, end_time, offset, + limit) pyresult = ResultSetPy(self, True) pyresult.init_c(result, "prepared") self.activate_result_set_list.add(pyresult) @@ -588,8 +754,13 @@ cdef class TsFileReaderPy: def query_prepared_multi(self, prepared_list, start_time : int = INT64_MIN, end_time : int = INT64_MAX, - offset : int = 0, limit : int = -1) -> ResultSetPy: - """Query aligned prepared value columns through one shared time reader.""" + offset : int = 0, limit : int = -1, + block_rows : int = 0) -> ResultSetPy: + """Query aligned prepared value columns through one shared time reader. + + ``block_rows`` bounds each returned TsBlock; 0 keeps the native + default capacity. It does not limit the number of rows returned. + """ cdef Py_ssize_t count = len(prepared_list) cdef Py_ssize_t i cdef PreparedSeriesPy prepared @@ -600,6 +771,7 @@ cdef class TsFileReaderPy: cdef int64_t c_end_time = end_time cdef int c_offset = offset cdef int c_limit = limit + cdef uint32_t c_block_rows = block_rows if block_rows > 0 else 0 if count <= 0: raise ValueError("prepared_list must not be empty") handles = malloc( @@ -612,10 +784,17 @@ cdef class TsFileReaderPy: if prepared.prepared == NULL: raise RuntimeError("PreparedSeries is closed") handles[i] = prepared.prepared - with nogil: - result = tsfile_reader_query_prepared_multi( - self.reader, handles, count, - c_start_time, c_end_time, c_offset, c_limit, &code) + if block_rows > 0: + with nogil: + result = tsfile_reader_query_prepared_multi_ex( + self.reader, handles, count, + c_start_time, c_end_time, c_offset, c_limit, + c_block_rows, &code) + else: + with nogil: + result = tsfile_reader_query_prepared_multi( + self.reader, handles, count, + c_start_time, c_end_time, c_offset, c_limit, &code) check_error(code, b"Failed to query aligned prepared series") finally: free(handles) @@ -718,6 +897,10 @@ cdef class TsFileReaderPy: """ if self.reader == NULL: return + if self._prepared_query_cache is not None: + for prepared in self._prepared_query_cache.values(): + self._close_prepared_entry(prepared) + self._prepared_query_cache.clear() # result_set_bak to avoid runtime error. result_set_bak = list(self.activate_result_set_list) for result_set in result_set_bak: