Skip to content

perf(python): reuse prepared series for repeated single-device table queries #1082

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ColinLeeo wants to merge 1 commit into
base: develop
Choose a base branch
from
+822 −30
Open

perf(python): reuse prepared series for repeated single-device table queries #1082

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 56 additions & 2 deletions cpp/src/cwrapper/tsfile_cwrapper.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<storage::PreparedSeries> prepared;
*err_code =
static_cast<storage::TsFileReader*>(reader)->prepare_table_series(
table_name, column_name,
static_cast<storage::Filter*>(tag_filter), prepared);
if (*err_code != common::E_OK) {
return nullptr;
}
auto* handle = new (std::nothrow)
std::shared_ptr<storage::PreparedSeries>(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,
Expand Down Expand Up @@ -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;
}
Expand All @@ -667,14 +711,23 @@ ResultSet tsfile_reader_query_prepared(TsFileReader reader,
static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared);
storage::ResultSet* result = nullptr;
*err_code = static_cast<storage::TsFileReader*>(reader)->query_prepared(
*handle, start_time, end_time, offset, limit, result);
*handle, start_time, end_time, offset, limit, result, block_rows);
return result;
}

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;
}
Expand All @@ -697,7 +750,8 @@ ResultSet tsfile_reader_query_prepared_multi(
storage::ResultSet* result = nullptr;
*err_code =
static_cast<storage::TsFileReader*>(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;
}

Expand Down
29 changes: 29 additions & 0 deletions cpp/src/cwrapper/tsfile_cwrapper.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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.
Expand All @@ -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.
*
Expand Down
10 changes: 6 additions & 4 deletions cpp/src/reader/block/prepared_series_tsblock_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,8 +50,9 @@ common::TSDataType prepared_value_data_type(ITimeseriesIndex* index) {

int PreparedSeriesTsBlockReader::init(
TsFileIOReader* io_reader, const std::shared_ptr<PreparedSeries>& 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;
Expand Down Expand Up @@ -92,8 +93,9 @@ int PreparedSeriesTsBlockReader::init(
int PreparedSeriesTsBlockReader::init_multi(
TsFileIOReader* io_reader,
const std::vector<std::shared_ptr<PreparedSeries>>& 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;
}
Expand Down Expand Up @@ -149,9 +151,9 @@ int PreparedSeriesTsBlockReader::has_next(bool& has_next) {
while (true) {
const uint32_t desired_rows =
remaining_limit_ > 0
? std::min<uint32_t>(PREPARED_BATCH_ROWS,
? std::min<uint32_t>(block_rows_,
static_cast<uint32_t>(remaining_limit_))
: PREPARED_BATCH_ROWS;
: block_rows_;
if (block_ != nullptr) {
if (block_->get_max_row_count() != desired_rows) {
ssi_->revert_tsblock();
Expand Down
7 changes: 5 additions & 2 deletions cpp/src/reader/block/prepared_series_tsblock_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,10 +41,12 @@ class PreparedSeriesTsBlockReader final : public TsBlockReader {

int init(TsFileIOReader* io_reader,
const std::shared_ptr<PreparedSeries>& 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<std::shared_ptr<PreparedSeries>>& 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;
Expand All @@ -64,6 +66,7 @@ class PreparedSeriesTsBlockReader final : public TsBlockReader {
common::TsBlock* block_ = nullptr;
common::TSDataType value_data_type_ = common::INVALID_DATATYPE;
std::vector<common::TSDataType> value_data_types_;
uint32_t block_rows_ = 0;
int remaining_limit_ = -1;
bool block_ready_ = false;
bool exhausted_ = false;
Expand Down
8 changes: 4 additions & 4 deletions cpp/src/reader/tsfile_executor.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ int TsFileExecutor::prepare_series(
int TsFileExecutor::execute_prepared(
const std::shared_ptr<PreparedSeries>& 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) {
Expand All @@ -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;
}
Expand All @@ -150,7 +150,7 @@ int TsFileExecutor::execute_prepared(
int TsFileExecutor::execute_prepared_multi(
const std::vector<std::shared_ptr<PreparedSeries>>& 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) {
Expand All @@ -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;
}
Expand Down
4 changes: 2 additions & 2 deletions cpp/src/reader/tsfile_executor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,11 +48,11 @@ class TsFileExecutor // : public QueryExecutor
int execute_prepared(const std::shared_ptr<PreparedSeries>& 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<std::shared_ptr<PreparedSeries>>& 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);
Expand Down
120 changes: 116 additions & 4 deletions cpp/src/reader/tsfile_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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<PreparedSeries>& 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<TableSchema> 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<std::shared_ptr<IDeviceID>, 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<std::string> measurement_names;
measurement_names.insert(to_lower(column_name));
std::vector<ITimeseriesIndex*> 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<AlignedTimeseriesIndex*>(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<uint64_t>(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<uint64_t>(time_offset);
locator.time_metadata_length = time_length;
}
}
} else if (auto* value = dynamic_cast<TimeseriesIndex*>(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<uint64_t>(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<uint64_t>(read_file_->file_size());
return io_reader->prepare_series(generation, locator, prepared);
}

int TsFileReader::query_prepared(
const std::shared_ptr<PreparedSeries>& 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<std::shared_ptr<PreparedSeries>>& 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,
Expand Down
Loading
Loading