| /* |
| * 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 "cwrapper/tsfile_cwrapper.h" |
| |
| #include <fcntl.h> |
| #include <file/write_file.h> |
| #include <reader/qds_without_timegenerator.h> |
| #include <writer/tsfile_table_writer.h> |
| |
| #ifdef _WIN32 |
| #include <io.h> |
| #else |
| #include <unistd.h> |
| #endif |
| |
| #include <cstring> |
| #include <limits> |
| #include <memory> |
| #include <new> |
| #include <set> |
| #include <vector> |
| |
| #include "common/device_id.h" |
| #include "common/global.h" |
| #include "common/statistic.h" |
| #include "common/tablet.h" |
| #include "common/tsfile_common.h" |
| #include "reader/filter/tag_filter.h" |
| #include "reader/result_set.h" |
| #include "reader/table_result_set.h" |
| #include "reader/tsfile_reader.h" |
| #include "utils/injection.h" |
| #include "writer/tsfile_writer.h" |
| |
| // Forward declarations for arrow namespace functions (defined in arrow_c.cc) |
| namespace arrow { |
| int TsBlockToArrowStruct(common::TsBlock& tsblock, ArrowArray* out_array, |
| ArrowSchema* out_schema); |
| int ArrowStructToTablet(const char* table_name, const ArrowArray* in_array, |
| const ArrowSchema* in_schema, |
| const storage::TableSchema* reg_schema, |
| storage::Tablet** out_tablet, int time_col_index); |
| } // namespace arrow |
| |
| namespace { |
| |
| bool is_supported_table_type(TSDataType type) { |
| switch (type) { |
| case TS_DATATYPE_BOOLEAN: |
| case TS_DATATYPE_INT32: |
| case TS_DATATYPE_INT64: |
| case TS_DATATYPE_FLOAT: |
| case TS_DATATYPE_DOUBLE: |
| case TS_DATATYPE_TEXT: |
| case TS_DATATYPE_TIMESTAMP: |
| case TS_DATATYPE_DATE: |
| case TS_DATATYPE_BLOB: |
| case TS_DATATYPE_STRING: |
| return true; |
| default: |
| return false; |
| } |
| } |
| |
| ERRNO validate_table_schema(const TableSchema* schema) { |
| if (schema == nullptr || schema->table_name == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| if (schema->table_name[0] == '\0' || schema->column_num <= 0) { |
| return common::E_INVALID_SCHEMA; |
| } |
| if (schema->column_schemas == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| std::set<std::string> names; |
| for (int i = 0; i < schema->column_num; ++i) { |
| const ColumnSchema& column = schema->column_schemas[i]; |
| if (column.column_name == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| if (column.column_name[0] == '\0' || |
| !names.insert(storage::to_lower(column.column_name)).second || |
| !is_supported_table_type(column.data_type) || |
| (column.column_category != TAG && |
| column.column_category != FIELD) || |
| (column.column_category == TAG && |
| column.data_type != TS_DATATYPE_STRING)) { |
| return common::E_INVALID_SCHEMA; |
| } |
| } |
| return common::E_OK; |
| } |
| |
| } // namespace |
| |
| #ifdef __cplusplus |
| extern "C" { |
| #endif |
| |
| void init_tsfile_config() { storage::libtsfile_init(); } |
| |
| uint8_t get_global_time_encoding() { |
| return common::get_global_time_encoding(); |
| } |
| |
| uint8_t get_global_time_compression() { |
| return common::get_global_time_compression(); |
| } |
| |
| uint8_t get_datatype_encoding(uint8_t data_type) { |
| return common::get_datatype_encoding(data_type); |
| } |
| |
| uint8_t get_global_compression() { return common::get_global_compression(); } |
| |
| int set_global_time_encoding(uint8_t encoding) { |
| return common::set_global_time_encoding(encoding); |
| } |
| |
| int set_global_time_compression(uint8_t compression) { |
| return common::set_global_time_compression(compression); |
| } |
| |
| int set_datatype_encoding(uint8_t data_type, uint8_t encoding) { |
| return common::set_datatype_encoding(data_type, encoding); |
| } |
| |
| 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<TsFileReadBackend>(common::get_file_read_backend()); |
| } |
| |
| WriteFile write_file_new(const char* pathname, ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| if (pathname == nullptr || pathname[0] == '\0') { |
| *err_code = common::E_INVALID_ARG; |
| return nullptr; |
| } |
| int ret; |
| init_tsfile_config(); |
| |
| int flags = O_RDWR | O_CREAT | O_TRUNC; |
| #ifdef _WIN32 |
| flags |= O_BINARY; |
| #endif |
| mode_t mode = 0666; |
| storage::WriteFile* file = new (std::nothrow) storage::WriteFile; |
| if (file == nullptr) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } |
| ret = file->create(pathname, flags, mode); |
| if (ret != common::E_OK) { |
| delete file; |
| *err_code = ret; |
| return nullptr; |
| } |
| *err_code = ret; |
| return file; |
| } |
| |
| TsFileWriter tsfile_writer_new(WriteFile file, TableSchema* schema, |
| ERRNO* err_code) { |
| // C API: every public entry must defend against null callers — a null |
| // schema or err_code would crash the host process the moment it's |
| // dereferenced. The tag-filter helpers already follow this pattern. |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| if (file == nullptr) { |
| *err_code = common::E_INVALID_ARG; |
| return nullptr; |
| } |
| *err_code = validate_table_schema(schema); |
| if (*err_code != common::E_OK) { |
| return nullptr; |
| } |
| |
| try { |
| init_tsfile_config(); |
| std::vector<common::ColumnSchema> column_schemas; |
| for (int i = 0; i < schema->column_num; i++) { |
| ColumnSchema cur_schema = schema->column_schemas[i]; |
| column_schemas.emplace_back( |
| cur_schema.column_name, |
| static_cast<common::TSDataType>(cur_schema.data_type), |
| static_cast<common::ColumnCategory>( |
| cur_schema.column_category)); |
| } |
| |
| std::unique_ptr<storage::TableSchema> table_schema( |
| new storage::TableSchema(schema->table_name, column_schemas)); |
| std::unique_ptr<storage::TsFileTableWriter> table_writer( |
| new storage::TsFileTableWriter( |
| static_cast<storage::WriteFile*>(file), table_schema.get())); |
| *err_code = table_writer->get_error(); |
| if (*err_code != common::E_OK) { |
| return nullptr; |
| } |
| return table_writer.release(); |
| } catch (const std::bad_alloc&) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } catch (...) { |
| *err_code = common::E_INVALID_SCHEMA; |
| return nullptr; |
| } |
| } |
| |
| TsFileWriter tsfile_writer_new_with_memory_threshold(WriteFile file, |
| TableSchema* schema, |
| uint64_t memory_threshold, |
| ERRNO* err_code) { |
| // See tsfile_writer_new() above for the null-guard rationale. |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| if (file == nullptr || memory_threshold == 0) { |
| *err_code = common::E_INVALID_ARG; |
| return nullptr; |
| } |
| *err_code = validate_table_schema(schema); |
| if (*err_code != common::E_OK) { |
| return nullptr; |
| } |
| try { |
| init_tsfile_config(); |
| std::vector<common::ColumnSchema> column_schemas; |
| for (int i = 0; i < schema->column_num; i++) { |
| ColumnSchema cur_schema = schema->column_schemas[i]; |
| column_schemas.emplace_back( |
| cur_schema.column_name, |
| static_cast<common::TSDataType>(cur_schema.data_type), |
| static_cast<common::ColumnCategory>( |
| cur_schema.column_category)); |
| } |
| |
| std::unique_ptr<storage::TableSchema> table_schema( |
| new storage::TableSchema(schema->table_name, column_schemas)); |
| std::unique_ptr<storage::TsFileTableWriter> table_writer( |
| new storage::TsFileTableWriter( |
| static_cast<storage::WriteFile*>(file), table_schema.get(), |
| memory_threshold)); |
| *err_code = table_writer->get_error(); |
| if (*err_code != common::E_OK) { |
| return nullptr; |
| } |
| return table_writer.release(); |
| } catch (const std::bad_alloc&) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } catch (...) { |
| *err_code = common::E_INVALID_SCHEMA; |
| return nullptr; |
| } |
| } |
| |
| TsFileReader tsfile_reader_new(const char* pathname, ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| if (pathname == nullptr || pathname[0] == '\0') { |
| *err_code = common::E_INVALID_ARG; |
| return nullptr; |
| } |
| init_tsfile_config(); |
| auto reader = new (std::nothrow) storage::TsFileReader(); |
| if (reader == nullptr) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } |
| int ret = reader->open(pathname); |
| if (ret != common::E_OK) { |
| *err_code = ret; |
| delete reader; |
| return nullptr; |
| } |
| return reader; |
| } |
| |
| ERRNO tsfile_writer_close(TsFileWriter writer) { |
| if (writer == nullptr) { |
| return common::E_OK; |
| } |
| auto* w = static_cast<storage::TsFileTableWriter*>(writer); |
| int ret = w->close(); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| delete w; |
| return ret; |
| } |
| |
| ERRNO tsfile_writer_add_tsfile_property(TsFileWriter writer, const char* key, |
| uint32_t key_len, const uint8_t* value, |
| uint32_t value_len) { |
| if (writer == nullptr || key == nullptr || |
| (value == nullptr && value_len > 0)) { |
| return common::E_INVALID_ARG; |
| } |
| if (key_len > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()) || |
| value_len > |
| static_cast<uint32_t>(std::numeric_limits<int32_t>::max())) { |
| return common::E_OUT_OF_RANGE; |
| } |
| try { |
| auto* w = static_cast<storage::TsFileTableWriter*>(writer); |
| return w->add_tsfile_property(std::string(key, key_len), value, |
| value_len); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ERRNO tsfile_writer_flush(TsFileWriter writer) { |
| if (writer == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| return static_cast<storage::TsFileTableWriter*>(writer)->flush(); |
| } |
| |
| ERRNO tsfile_reader_close(TsFileReader reader) { |
| auto* ts_reader = static_cast<storage::TsFileReader*>(reader); |
| delete ts_reader; |
| return common::E_OK; |
| } |
| |
| Tablet tablet_new(char** column_name_list, TSDataType* data_types, |
| uint32_t column_num, uint32_t max_rows, ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_INVALID_ARG; |
| if (column_num == 0 || max_rows == 0 || column_name_list == nullptr || |
| data_types == nullptr || max_rows >= (1u << 30)) { |
| return nullptr; |
| } |
| try { |
| std::vector<std::string> measurement_list; |
| std::vector<common::TSDataType> data_type_list; |
| std::set<std::string> names; |
| for (uint32_t i = 0; i < column_num; i++) { |
| if (column_name_list[i] == nullptr || |
| column_name_list[i][0] == '\0') { |
| return nullptr; |
| } |
| if (!is_supported_table_type(data_types[i])) { |
| *err_code = common::E_TYPE_NOT_SUPPORTED; |
| return nullptr; |
| } |
| std::string name = storage::to_lower(column_name_list[i]); |
| if (!names.insert(name).second) { |
| *err_code = common::E_INVALID_SCHEMA; |
| return nullptr; |
| } |
| measurement_list.emplace_back(std::move(name)); |
| data_type_list.push_back( |
| static_cast<common::TSDataType>(*(data_types + i))); |
| } |
| auto* tablet = new (std::nothrow) |
| storage::Tablet(measurement_list, data_type_list, max_rows); |
| if (tablet == nullptr) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } |
| if (tablet->err_code_ != common::E_OK) { |
| *err_code = tablet->err_code_; |
| delete tablet; |
| return nullptr; |
| } |
| *err_code = common::E_OK; |
| return tablet; |
| } catch (const std::bad_alloc&) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } catch (...) { |
| *err_code = common::E_INVALID_SCHEMA; |
| return nullptr; |
| } |
| } |
| |
| uint32_t tablet_get_cur_row_size(Tablet tablet) { |
| return static_cast<storage::Tablet*>(tablet)->get_cur_row_size(); |
| } |
| |
| ERRNO tablet_add_timestamp(Tablet tablet, uint32_t row_index, |
| Timestamp timestamp) { |
| return static_cast<storage::Tablet*>(tablet)->add_timestamp(row_index, |
| timestamp); |
| } |
| |
| #define TABLET_ADD_VALUE_BY_NAME_DEF(type) \ |
| ERRNO tablet_add_value_by_name_##type(Tablet tablet, uint32_t row_index, \ |
| const char* column_name, \ |
| const type value) { \ |
| return static_cast<storage::Tablet*>(tablet)->add_value( \ |
| row_index, storage::to_lower(column_name), value); \ |
| } |
| TABLET_ADD_VALUE_BY_NAME_DEF(int32_t); |
| TABLET_ADD_VALUE_BY_NAME_DEF(int64_t); |
| TABLET_ADD_VALUE_BY_NAME_DEF(float); |
| TABLET_ADD_VALUE_BY_NAME_DEF(double); |
| TABLET_ADD_VALUE_BY_NAME_DEF(bool); |
| |
| ERRNO tablet_add_value_by_name_string_with_len(Tablet tablet, |
| uint32_t row_index, |
| const char* column_name, |
| const char* value, |
| int value_len) { |
| return static_cast<storage::Tablet*>(tablet)->add_value( |
| row_index, storage::to_lower(column_name), |
| common::String(value, value_len)); |
| } |
| |
| #define TABLE_ADD_VALUE_BY_INDEX_DEF(type) \ |
| ERRNO tablet_add_value_by_index_##type(Tablet tablet, uint32_t row_index, \ |
| uint32_t column_index, \ |
| const type value) { \ |
| return static_cast<storage::Tablet*>(tablet)->add_value( \ |
| row_index, column_index, value); \ |
| } |
| |
| ERRNO tablet_add_value_by_index_string_with_len(Tablet tablet, |
| uint32_t row_index, |
| uint32_t column_index, |
| const char* value, |
| int value_len) { |
| return static_cast<storage::Tablet*>(tablet)->add_value( |
| row_index, column_index, common::String(value, value_len)); |
| } |
| |
| TABLE_ADD_VALUE_BY_INDEX_DEF(int32_t); |
| TABLE_ADD_VALUE_BY_INDEX_DEF(int64_t); |
| TABLE_ADD_VALUE_BY_INDEX_DEF(float); |
| TABLE_ADD_VALUE_BY_INDEX_DEF(double); |
| TABLE_ADD_VALUE_BY_INDEX_DEF(bool); |
| |
| // TsRecord API |
| TsRecord _ts_record_new(const char* device_id, Timestamp timestamp, |
| int timeseries_num) { |
| auto* record = new storage::TsRecord(timestamp, device_id, timeseries_num); |
| return record; |
| } |
| |
| #define INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(type) \ |
| ERRNO _insert_data_into_ts_record_by_name_##type( \ |
| TsRecord data, const char* measurement_name, type value) { \ |
| auto* record = (storage::TsRecord*)data; \ |
| storage::DataPoint point(measurement_name, value); \ |
| if (record->points_.size() + 1 > record->points_.capacity()) \ |
| return common::E_BUF_NOT_ENOUGH; \ |
| record->points_.push_back(point); \ |
| return common::E_OK; \ |
| } |
| |
| ERRNO _insert_data_into_ts_record_by_name_string_with_len( |
| TsRecord data, const char* measurement_name, const char* value, |
| const uint32_t value_len) { |
| auto* record = (storage::TsRecord*)data; |
| if (record->points_.size() + 1 > record->points_.capacity()) |
| return common::E_BUF_NOT_ENOUGH; |
| common::String str_value; |
| str_value.dup_from(value, value_len, record->pa); |
| record->add_point(measurement_name, str_value); |
| return common::E_OK; |
| } |
| |
| INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(int32_t); |
| INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(int64_t); |
| INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(bool); |
| INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(float); |
| INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(double); |
| /* |
| TsFileWriter tsfile_writer_new_with_conf(const char *pathname, |
| const mode_t flag, ERRNO *err_code, |
| TsFileConf *conf) { |
| init_tsfile_config(); |
| auto *writer = new storage::TsFileWriter(); |
| const int ret = writer->open(pathname, O_CREAT | O_RDWR, flag); |
| if (ret != common::E_OK) { |
| delete writer; |
| *err_code = ret; |
| return nullptr; |
| } |
| return writer; |
| } |
| |
| */ |
| ERRNO tsfile_writer_write(TsFileWriter writer, Tablet tablet) { |
| if (writer == nullptr || tablet == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| auto* w = static_cast<storage::TsFileTableWriter*>(writer); |
| auto* tbl = static_cast<storage::Tablet*>(tablet); |
| return w->write_table(*tbl); |
| } |
| |
| ERRNO tsfile_writer_write_arrow(TsFileWriter writer, ArrowArray* array, |
| ArrowSchema* schema, int time_col_index) { |
| if (writer == nullptr || array == nullptr || schema == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| auto* w = static_cast<storage::TsFileTableWriter*>(writer); |
| std::shared_ptr<storage::TableSchema> registered_schema = |
| w->get_table_schema(); |
| if (!registered_schema) { |
| return common::E_INVALID_SCHEMA; |
| } |
| storage::Tablet* raw_tablet = nullptr; |
| int ret = arrow::ArrowStructToTablet(w->get_table_name().c_str(), array, |
| schema, registered_schema.get(), |
| &raw_tablet, time_col_index); |
| std::unique_ptr<storage::Tablet> tablet(raw_tablet); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| return w->write_table(*tablet); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } catch (...) { |
| return common::E_INVALID_ARG; |
| } |
| } |
| |
| // ERRNO tsfile_writer_flush_data(TsFileWriter writer) { |
| // auto *w = static_cast<storage::TsFileWriter *>(writer); |
| // return w->flush(); |
| // } |
| |
| // Query |
| |
| PreparedSeriesHandle tsfile_reader_prepare_series( |
| TsFileReader reader, const TsFilePreparedLocator* locator, bool trust_index, |
| ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_INVALID_ARG; |
| if (reader == nullptr || locator == nullptr) { |
| return nullptr; |
| } |
| storage::FileGeneration generation; |
| generation.mapped_index_identity = locator->mapped_index_identity; |
| generation.file_id = locator->file_id; |
| generation.file_size = locator->file_size; |
| generation.file_fingerprint = locator->file_fingerprint; |
| generation.trust_index = trust_index; |
| storage::PreparedLocator native_locator; |
| native_locator.locator_id = locator->locator_id; |
| native_locator.layout = locator->layout; |
| native_locator.flags = locator->flags; |
| native_locator.value_metadata_offset = locator->value_metadata_offset; |
| native_locator.value_metadata_length = locator->value_metadata_length; |
| native_locator.time_metadata_offset = locator->time_metadata_offset; |
| native_locator.time_metadata_length = locator->time_metadata_length; |
| std::shared_ptr<storage::PreparedSeries> prepared; |
| *err_code = static_cast<storage::TsFileReader*>(reader)->prepare_series( |
| generation, native_locator, 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; |
| } |
| |
| PreparedSeriesHandle tsfile_reader_prepare_series_with_time_owner( |
| TsFileReader reader, const TsFilePreparedLocator* locator, |
| PreparedSeriesHandle aligned_time_owner, bool trust_index, |
| ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_INVALID_ARG; |
| if (reader == nullptr || locator == nullptr || |
| aligned_time_owner == nullptr) { |
| return nullptr; |
| } |
| |
| storage::FileGeneration generation; |
| generation.mapped_index_identity = locator->mapped_index_identity; |
| generation.file_id = locator->file_id; |
| generation.file_size = locator->file_size; |
| generation.file_fingerprint = locator->file_fingerprint; |
| generation.trust_index = trust_index; |
| storage::PreparedLocator native_locator; |
| native_locator.locator_id = locator->locator_id; |
| native_locator.layout = locator->layout; |
| native_locator.flags = locator->flags; |
| native_locator.value_metadata_offset = locator->value_metadata_offset; |
| native_locator.value_metadata_length = locator->value_metadata_length; |
| native_locator.time_metadata_offset = locator->time_metadata_offset; |
| native_locator.time_metadata_length = locator->time_metadata_length; |
| |
| auto* owner = static_cast<std::shared_ptr<storage::PreparedSeries>*>( |
| aligned_time_owner); |
| std::shared_ptr<storage::PreparedSeries> prepared; |
| *err_code = static_cast<storage::TsFileReader*>(reader)->prepare_series( |
| generation, native_locator, *owner, 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; |
| } |
| |
| void tsfile_prepared_series_free(PreparedSeriesHandle prepared) { |
| delete static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared); |
| } |
| |
| ResultSet tsfile_reader_query_prepared(TsFileReader reader, |
| PreparedSeriesHandle prepared, |
| Timestamp start_time, Timestamp end_time, |
| int offset, int limit, ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_INVALID_ARG; |
| if (reader == nullptr || prepared == nullptr) { |
| return nullptr; |
| } |
| auto* handle = |
| 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); |
| 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) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_INVALID_ARG; |
| if (reader == nullptr || prepared == nullptr || prepared_count == 0) { |
| return nullptr; |
| } |
| |
| std::vector<std::shared_ptr<storage::PreparedSeries>> native_prepared; |
| native_prepared.reserve(prepared_count); |
| for (uint32_t i = 0; i < prepared_count; i++) { |
| if (prepared[i] == nullptr) { |
| return nullptr; |
| } |
| auto* handle = |
| static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared[i]); |
| native_prepared.push_back(*handle); |
| } |
| |
| storage::ResultSet* result = nullptr; |
| *err_code = |
| static_cast<storage::TsFileReader*>(reader)->query_prepared_multi( |
| native_prepared, start_time, end_time, offset, limit, result); |
| return result; |
| } |
| |
| ResultSet tsfile_query_table(TsFileReader reader, const char* table_name, |
| char** columns, uint32_t column_num, |
| Timestamp start_time, Timestamp end_time, |
| ERRNO* err_code) { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::ResultSet* table_result_set = nullptr; |
| std::vector<std::string> column_names; |
| for (uint32_t i = 0; i < column_num; i++) { |
| column_names.emplace_back(columns[i]); |
| } |
| *err_code = r->query(table_name, column_names, start_time, end_time, |
| table_result_set); |
| return table_result_set; |
| } |
| |
| ResultSet tsfile_query_table_on_tree(TsFileReader reader, char** columns, |
| uint32_t column_num, Timestamp start_time, |
| Timestamp end_time, ERRNO* err_code) { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::ResultSet* table_result_set = nullptr; |
| std::vector<std::string> column_names; |
| for (uint32_t i = 0; i < column_num; i++) { |
| column_names.emplace_back(columns[i]); |
| } |
| *err_code = r->query_table_on_tree(column_names, start_time, end_time, |
| table_result_set); |
| return table_result_set; |
| } |
| |
| ResultSet tsfile_reader_query_tree(TsFileReader reader, char** paths, |
| uint32_t path_num, Timestamp start_time, |
| Timestamp end_time, ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_INVALID_ARG; |
| if (reader == nullptr || paths == nullptr || path_num == 0) { |
| return nullptr; |
| } |
| try { |
| std::vector<std::string> selected_paths; |
| selected_paths.reserve(path_num); |
| for (uint32_t i = 0; i < path_num; i++) { |
| if (paths[i] == nullptr) { |
| return nullptr; |
| } |
| selected_paths.emplace_back(paths[i]); |
| } |
| storage::ResultSet* result_set = nullptr; |
| *err_code = static_cast<storage::TsFileReader*>(reader)->query( |
| selected_paths, start_time, end_time, result_set); |
| return result_set; |
| } catch (const std::bad_alloc&) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } |
| } |
| |
| ResultSet tsfile_reader_query_tree_by_row(TsFileReader reader, |
| char** device_ids, int device_ids_len, |
| char** measurement_names, |
| int measurement_names_len, int offset, |
| int limit, ERRNO* err_code) { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::ResultSet* result_set = nullptr; |
| |
| std::vector<std::string> path_list; |
| if (device_ids_len > 0 && measurement_names_len > 0) { |
| path_list.reserve(static_cast<size_t>(device_ids_len) * |
| static_cast<size_t>(measurement_names_len)); |
| } |
| |
| for (int i = 0; i < device_ids_len; i++) { |
| const char* device_id = device_ids[i]; |
| if (device_id == nullptr) { |
| continue; |
| } |
| for (int j = 0; j < measurement_names_len; j++) { |
| const char* measurement_name = measurement_names[j]; |
| if (measurement_name == nullptr) { |
| continue; |
| } |
| path_list.emplace_back(std::string(device_id) + "." + |
| std::string(measurement_name)); |
| } |
| } |
| |
| *err_code = r->queryByRow(path_list, offset, limit, result_set); |
| return result_set; |
| } |
| |
| ResultSet tsfile_reader_query_table_by_row( |
| TsFileReader reader, const char* table_name, char** column_names, |
| int column_names_len, int offset, int limit, TagFilterHandle tag_filter, |
| int batch_size, ERRNO* err_code) { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::ResultSet* result_set = nullptr; |
| |
| std::vector<std::string> columns; |
| if (column_names_len > 0) { |
| columns.reserve(static_cast<size_t>(column_names_len)); |
| } |
| for (int i = 0; i < column_names_len; i++) { |
| const char* name = column_names[i]; |
| columns.emplace_back(name == nullptr ? "" : std::string(name)); |
| } |
| |
| *err_code = r->queryByRow( |
| table_name == nullptr ? "" : table_name, columns, offset, limit, |
| result_set, static_cast<storage::Filter*>(tag_filter), batch_size); |
| return result_set; |
| } |
| |
| ResultSet tsfile_reader_query_table(TsFileReader reader, const char* table_name, |
| char** column_names, |
| uint32_t column_names_len, |
| Timestamp start_time, Timestamp end_time, |
| int offset, int limit, |
| TagFilterHandle tag_filter, int batch_size, |
| ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_INVALID_ARG; |
| if (reader == nullptr || table_name == nullptr || column_names == nullptr || |
| column_names_len == 0 || offset < 0 || batch_size < 0) { |
| return nullptr; |
| } |
| try { |
| std::vector<std::string> columns; |
| columns.reserve(column_names_len); |
| for (uint32_t i = 0; i < column_names_len; ++i) { |
| if (column_names[i] == nullptr) { |
| return nullptr; |
| } |
| columns.emplace_back(column_names[i]); |
| } |
| storage::ResultSet* result_set = nullptr; |
| const int normalized_limit = limit < 0 ? -1 : limit; |
| *err_code = static_cast<storage::TsFileReader*>(reader)->query( |
| table_name, columns, start_time, end_time, offset, normalized_limit, |
| result_set, static_cast<storage::Filter*>(tag_filter), batch_size); |
| return result_set; |
| } catch (const std::bad_alloc&) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } catch (...) { |
| *err_code = common::E_FILE_READ_ERR; |
| return nullptr; |
| } |
| } |
| |
| ResultSet tsfile_query_table_batch(TsFileReader reader, const char* table_name, |
| char** columns, uint32_t column_num, |
| Timestamp start_time, Timestamp end_time, |
| TagFilterHandle tag_filter, int batch_size, |
| ERRNO* err_code) { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::ResultSet* table_result_set = nullptr; |
| std::vector<std::string> column_names; |
| for (uint32_t i = 0; i < column_num; i++) { |
| column_names.emplace_back(columns[i]); |
| } |
| *err_code = r->query(table_name, column_names, start_time, end_time, |
| table_result_set, |
| static_cast<storage::Filter*>(tag_filter), batch_size); |
| return table_result_set; |
| } |
| |
| bool tsfile_result_set_next(ResultSet result_set, ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return false; |
| } |
| if (result_set == nullptr) { |
| *err_code = common::E_INVALID_ARG; |
| return false; |
| } |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| bool has_next = true; |
| int ret = common::E_OK; |
| ret = r->next(has_next); |
| *err_code = ret; |
| if (ret != common::E_OK) { |
| return false; |
| } |
| return has_next; |
| } |
| |
| ERRNO tsfile_result_set_get_next_tsblock_as_arrow(ResultSet result_set, |
| ArrowArray* out_array, |
| ArrowSchema* out_schema) { |
| if (result_set == nullptr || out_array == nullptr || |
| out_schema == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| std::memset(out_array, 0, sizeof(*out_array)); |
| std::memset(out_schema, 0, sizeof(*out_schema)); |
| |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| auto* table_result_set = dynamic_cast<storage::TableResultSet*>(r); |
| if (table_result_set == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| |
| common::TsBlock* tsblock = nullptr; |
| int ret = table_result_set->get_next_tsblock(tsblock); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| |
| if (tsblock == nullptr) { |
| return common::E_NO_MORE_DATA; |
| } |
| |
| ret = arrow::TsBlockToArrowStruct(*tsblock, out_array, out_schema); |
| return ret; |
| } |
| |
| #define TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(type) \ |
| type tsfile_result_set_get_value_by_name_##type(ResultSet result_set, \ |
| const char* column_name) { \ |
| auto* r = static_cast<storage::ResultSet*>(result_set); \ |
| std::string column_name_(column_name); \ |
| return r->get_value<type>(column_name_); \ |
| } |
| |
| TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(bool); |
| TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(int32_t); |
| TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(int64_t); |
| TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(float); |
| TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(double); |
| char* tsfile_result_set_get_value_by_name_string(ResultSet result_set, |
| const char* column_name) { |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| std::string column_name_(column_name); |
| common::String* ret = r->get_value<common::String*>(column_name_); |
| if (ret == nullptr) { |
| return nullptr; |
| } |
| // Caller should free return's char* 's space. |
| char* dup = (char*)malloc(ret->len_ + 1); |
| if (dup) { |
| memcpy(dup, ret->buf_, ret->len_); |
| dup[ret->len_] = '\0'; |
| } |
| return dup; |
| } |
| |
| #define TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(type) \ |
| type tsfile_result_set_get_value_by_index_##type(ResultSet result_set, \ |
| uint32_t column_index) { \ |
| auto* r = static_cast<storage::ResultSet*>(result_set); \ |
| return r->get_value<type>(column_index); \ |
| } |
| |
| TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(int32_t); |
| TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(int64_t); |
| TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(float); |
| TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(double); |
| TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(bool); |
| |
| char* tsfile_result_set_get_value_by_index_string(ResultSet result_set, |
| uint32_t column_index) { |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| common::String* ret = r->get_value<common::String*>(column_index); |
| if (ret == nullptr) { |
| return nullptr; |
| } |
| // Caller should free return's char* 's space. |
| char* dup = (char*)malloc(ret->len_ + 1); |
| if (dup) { |
| memcpy(dup, ret->buf_, ret->len_); |
| dup[ret->len_] = '\0'; |
| } |
| return dup; |
| } |
| |
| ERRNO tsfile_result_set_get_value_by_index_binary(ResultSet result_set, |
| uint32_t column_index, |
| uint8_t** out_value, |
| uint32_t* out_length) { |
| if (result_set == nullptr || column_index == 0 || out_value == nullptr || |
| out_length == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| *out_value = nullptr; |
| *out_length = 0; |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| common::String* value = r->get_value<common::String*>(column_index); |
| if (value == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| if (value->len_ == 0) { |
| return common::E_OK; |
| } |
| uint8_t* copied = static_cast<uint8_t*>(malloc(value->len_)); |
| if (copied == nullptr) { |
| return common::E_OOM; |
| } |
| memcpy(copied, value->buf_, value->len_); |
| *out_value = copied; |
| *out_length = value->len_; |
| return common::E_OK; |
| } |
| |
| bool tsfile_result_set_is_null_by_name(ResultSet result_set, |
| const char* column_name) { |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| return r->is_null(column_name); |
| } |
| |
| bool tsfile_result_set_is_null_by_index(const ResultSet result_set, |
| const uint32_t column_index) { |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| return r->is_null(column_index); |
| } |
| |
| ResultSetMetaData tsfile_result_set_get_metadata(ResultSet result_set) { |
| auto* r = static_cast<storage::ResultSet*>(result_set); |
| if (result_set == NULL) { |
| return ResultSetMetaData(); |
| } |
| |
| ResultSetMetaData meta_data; |
| std::shared_ptr<storage::ResultSetMetadata> result_set_metadata = |
| r->get_metadata(); |
| meta_data.column_num = result_set_metadata->get_column_count(); |
| meta_data.column_names = |
| static_cast<char**>(malloc(meta_data.column_num * sizeof(char*))); |
| meta_data.data_types = static_cast<TSDataType*>( |
| malloc(meta_data.column_num * sizeof(TSDataType))); |
| for (int i = 0; i < meta_data.column_num; i++) { |
| meta_data.column_names[i] = |
| strdup(result_set_metadata->get_column_name(i + 1).c_str()); |
| meta_data.data_types[i] = static_cast<TSDataType>( |
| result_set_metadata->get_column_type(i + 1)); |
| } |
| return meta_data; |
| } |
| |
| char* tsfile_result_set_metadata_get_column_name(ResultSetMetaData result_set, |
| uint32_t column_index) { |
| if (column_index < 1 || column_index > (uint32_t)result_set.column_num) { |
| return nullptr; |
| } |
| return result_set.column_names[column_index - 1]; |
| } |
| |
| TSDataType tsfile_result_set_metadata_get_data_type( |
| ResultSetMetaData result_set, uint32_t column_index) { |
| if (column_index < 1 || |
| column_index > static_cast<uint32_t>(result_set.column_num)) { |
| return TS_DATATYPE_INVALID; |
| } |
| return result_set.data_types[column_index - 1]; |
| } |
| |
| int tsfile_result_set_metadata_get_column_num(ResultSetMetaData result_set) { |
| return result_set.column_num; |
| } |
| |
| TableSchema tsfile_reader_get_table_schema(TsFileReader reader, |
| const char* table_name) { |
| TableSchema ret_schema{}; |
| if (tsfile_reader_get_table_schema_checked(reader, table_name, |
| &ret_schema) != common::E_OK) { |
| return TableSchema{}; |
| } |
| return ret_schema; |
| } |
| |
| static ColumnSchema* allocate_table_schema_columns(size_t count) { |
| #ifdef ENABLE_TEST |
| using common::g_all_inject_points; |
| using common::InjectPoint; |
| DBUG_EXECUTE_IF("table_schema_column_alloc_fail", return nullptr;); |
| #endif |
| return static_cast<ColumnSchema*>(calloc(count, sizeof(ColumnSchema))); |
| } |
| |
| // MSVC /EHsc assumes C-linkage helpers do not let exceptions escape. |
| static ERRNO copy_table_schema(const std::shared_ptr<storage::TableSchema>& src, |
| TableSchema* out_schema) { |
| if (out_schema == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| *out_schema = TableSchema{}; |
| if (!src) { |
| return common::E_TABLE_NOT_EXIST; |
| } |
| try { |
| out_schema->table_name = strdup(src->get_table_name().c_str()); |
| if (out_schema->table_name == nullptr) { |
| return common::E_OOM; |
| } |
| const int column_num = src->get_columns_num(); |
| if (column_num == 0) { |
| return common::E_OK; |
| } |
| out_schema->column_schemas = |
| allocate_table_schema_columns(static_cast<size_t>(column_num)); |
| if (out_schema->column_schemas == nullptr) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_OOM; |
| } |
| out_schema->column_num = column_num; |
| #ifdef ENABLE_TEST |
| using common::g_all_inject_points; |
| using common::InjectPoint; |
| DBUG_EXECUTE_IF("table_schema_copy_oom", throw std::bad_alloc();); |
| DBUG_EXECUTE_IF("table_schema_copy_error", |
| throw common::E_FILE_READ_ERR;); |
| #endif |
| const auto& measurements = src->get_measurement_schemas(); |
| const auto& categories = src->get_column_categories(); |
| if (measurements.size() < static_cast<size_t>(out_schema->column_num) || |
| categories.size() < static_cast<size_t>(out_schema->column_num)) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_INVALID_SCHEMA; |
| } |
| for (int i = 0; i < out_schema->column_num; ++i) { |
| if (!measurements[i]) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_INVALID_SCHEMA; |
| } |
| out_schema->column_schemas[i].column_name = |
| strdup(measurements[i]->measurement_name_.c_str()); |
| if (out_schema->column_schemas[i].column_name == nullptr) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_OOM; |
| } |
| out_schema->column_schemas[i].data_type = |
| static_cast<TSDataType>(measurements[i]->data_type_); |
| out_schema->column_schemas[i].column_category = |
| static_cast<ColumnCategory>(categories[i]); |
| } |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_OOM; |
| } catch (...) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| static void free_table_schema_array(TableSchema* schemas, size_t size) { |
| if (schemas == nullptr) { |
| return; |
| } |
| for (size_t i = 0; i < size; ++i) { |
| free_table_schema(schemas[i]); |
| } |
| free(schemas); |
| } |
| |
| static void free_device_schema_array(DeviceSchema* schemas, size_t size) { |
| if (schemas == nullptr) { |
| return; |
| } |
| for (size_t i = 0; i < size; ++i) { |
| free_device_schema(schemas[i]); |
| } |
| free(schemas); |
| } |
| |
| ERRNO tsfile_reader_get_table_schema_checked(TsFileReader reader, |
| const char* table_name, |
| TableSchema* out_schema) { |
| if (out_schema == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| *out_schema = TableSchema{}; |
| if (reader == nullptr || table_name == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| std::shared_ptr<storage::TableSchema> schema; |
| const int ret = |
| static_cast<storage::TsFileReader*>(reader)->get_table_schema( |
| table_name, schema); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| return copy_table_schema(schema, out_schema); |
| } catch (const std::bad_alloc&) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_OOM; |
| } catch (...) { |
| free_table_schema(*out_schema); |
| *out_schema = TableSchema{}; |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| 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; |
| } |
| TableSchema* schemas = nullptr; |
| *error_code = |
| tsfile_reader_get_all_table_schemas_checked(reader, &schemas, size); |
| return *error_code == common::E_OK ? schemas : nullptr; |
| } |
| |
| ERRNO tsfile_reader_get_all_table_schemas_checked(TsFileReader reader, |
| TableSchema** out_schemas, |
| uint32_t* out_size) { |
| if (out_schemas != nullptr) { |
| *out_schemas = nullptr; |
| } |
| if (out_size != nullptr) { |
| *out_size = 0; |
| } |
| if (reader == nullptr || out_schemas == nullptr || out_size == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| TableSchema* copied = nullptr; |
| size_t initialized = 0; |
| try { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| std::vector<std::shared_ptr<storage::TableSchema>> schemas; |
| const ERRNO read_ret = r->get_all_table_schemas(schemas); |
| if (read_ret != common::E_OK) { |
| return read_ret; |
| } |
| if (schemas.empty()) { |
| return common::E_OK; |
| } |
| const size_t table_num = schemas.size(); |
| if (table_num > std::numeric_limits<uint32_t>::max()) { |
| return common::E_OVERFLOW; |
| } |
| copied = static_cast<TableSchema*>( |
| calloc(schemas.size(), sizeof(TableSchema))); |
| if (copied == nullptr) { |
| return common::E_OOM; |
| } |
| for (size_t i = 0; i < schemas.size(); ++i) { |
| initialized = i + 1; |
| ERRNO ret = copy_table_schema(schemas[i], &copied[i]); |
| if (ret != common::E_OK) { |
| free_table_schema_array(copied, initialized); |
| return ret; |
| } |
| } |
| *out_schemas = copied; |
| *out_size = static_cast<uint32_t>(schemas.size()); |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| free_table_schema_array(copied, initialized); |
| return common::E_OOM; |
| } catch (...) { |
| free_table_schema_array(copied, initialized); |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| ERRNO tsfile_reader_get_all_timeseries_schemas_checked( |
| TsFileReader reader, DeviceSchema** out_schemas, uint32_t* out_size) { |
| if (out_schemas != nullptr) { |
| *out_schemas = nullptr; |
| } |
| if (out_size != nullptr) { |
| *out_size = 0; |
| } |
| if (reader == nullptr || out_schemas == nullptr || out_size == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| DeviceSchema* device_schema = nullptr; |
| size_t initialized = 0; |
| try { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| std::vector<std::shared_ptr<storage::IDeviceID>> device_ids; |
| const ERRNO read_ret = r->get_all_devices(device_ids); |
| if (read_ret != common::E_OK) { |
| return read_ret; |
| } |
| if (device_ids.empty()) { |
| return common::E_OK; |
| } |
| const size_t device_count = device_ids.size(); |
| if (device_count > std::numeric_limits<uint32_t>::max()) { |
| return common::E_OVERFLOW; |
| } |
| device_schema = static_cast<DeviceSchema*>( |
| calloc(device_count, sizeof(DeviceSchema))); |
| if (device_schema == nullptr) { |
| return common::E_OOM; |
| } |
| |
| size_t device_index = 0; |
| for (const auto& device_id : device_ids) { |
| initialized = device_index + 1; |
| DeviceSchema& cur_schema = device_schema[device_index++]; |
| std::string device_name = |
| device_id == nullptr ? "" : device_id->get_device_name(); |
| cur_schema.device_name = strdup(device_name.c_str()); |
| if (cur_schema.device_name == nullptr) { |
| free_device_schema_array(device_schema, initialized); |
| return common::E_OOM; |
| } |
| |
| std::vector<storage::MeasurementSchema> schemas; |
| int ret = r->get_timeseries_schema(device_id, schemas); |
| if (ret != common::E_OK) { |
| free_device_schema_array(device_schema, initialized); |
| return ret; |
| } |
| if (schemas.empty()) { |
| continue; |
| } |
| if (schemas.size() > |
| static_cast<size_t>(std::numeric_limits<int>::max())) { |
| free_device_schema_array(device_schema, initialized); |
| return common::E_OVERFLOW; |
| } |
| |
| cur_schema.timeseries_schema = static_cast<TimeseriesSchema*>( |
| calloc(schemas.size(), sizeof(TimeseriesSchema))); |
| if (cur_schema.timeseries_schema == nullptr) { |
| free_device_schema_array(device_schema, initialized); |
| return common::E_OOM; |
| } |
| cur_schema.timeseries_num = static_cast<int>(schemas.size()); |
| for (size_t i = 0; i < schemas.size(); ++i) { |
| const auto& measurement_schema = schemas[i]; |
| cur_schema.timeseries_schema[i].timeseries_name = |
| strdup(measurement_schema.measurement_name_.c_str()); |
| if (cur_schema.timeseries_schema[i].timeseries_name == |
| nullptr) { |
| free_device_schema_array(device_schema, initialized); |
| return common::E_OOM; |
| } |
| cur_schema.timeseries_schema[i].data_type = |
| static_cast<TSDataType>(measurement_schema.data_type_); |
| cur_schema.timeseries_schema[i].encoding = |
| static_cast<TSEncoding>(measurement_schema.encoding_); |
| cur_schema.timeseries_schema[i].compression = |
| static_cast<CompressionType>( |
| measurement_schema.compression_type_); |
| } |
| } |
| *out_schemas = device_schema; |
| *out_size = static_cast<uint32_t>(device_count); |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| free_device_schema_array(device_schema, initialized); |
| return common::E_OOM; |
| } catch (...) { |
| free_device_schema_array(device_schema, initialized); |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| DeviceSchema* tsfile_reader_get_all_timeseries_schemas(TsFileReader reader, |
| uint32_t* size) { |
| DeviceSchema* schemas = nullptr; |
| const ERRNO ret = tsfile_reader_get_all_timeseries_schemas_checked( |
| reader, &schemas, size); |
| return ret == common::E_OK ? schemas : nullptr; |
| } |
| |
| void tsfile_device_id_free_contents(DeviceID* d) { |
| if (d == nullptr) { |
| return; |
| } |
| free(d->path); |
| d->path = nullptr; |
| free(d->table_name); |
| d->table_name = nullptr; |
| if (d->segments != nullptr) { |
| for (uint32_t k = 0; k < d->segment_count; k++) { |
| free(d->segments[k]); |
| } |
| free(d->segments); |
| d->segments = nullptr; |
| } |
| d->segment_count = 0; |
| } |
| |
| namespace { |
| |
| char* dup_common_string_to_cstr(const common::String& s) { |
| if (s.buf_ == nullptr || s.len_ == 0) { |
| return strdup(""); |
| } |
| char* p = static_cast<char*>(malloc(static_cast<size_t>(s.len_) + 1U)); |
| if (p == nullptr) { |
| return nullptr; |
| } |
| memcpy(p, s.buf_, static_cast<size_t>(s.len_)); |
| p[s.len_] = '\0'; |
| return p; |
| } |
| |
| static TSDataType cpp_stat_type_to_c(common::TSDataType t) { |
| return static_cast<TSDataType>(static_cast<uint8_t>(t)); |
| } |
| |
| void free_timeseries_statistic_heap(TimeseriesStatistic* s) { |
| if (s == nullptr) { |
| return; |
| } |
| TsFileStatisticBase* b = tsfile_statistic_base(s); |
| if (!b->has_statistic) { |
| return; |
| } |
| switch (b->type) { |
| case TS_DATATYPE_STRING: |
| free(s->u.string_s.str_min); |
| s->u.string_s.str_min = nullptr; |
| free(s->u.string_s.str_max); |
| s->u.string_s.str_max = nullptr; |
| free(s->u.string_s.str_first); |
| s->u.string_s.str_first = nullptr; |
| free(s->u.string_s.str_last); |
| s->u.string_s.str_last = nullptr; |
| break; |
| case TS_DATATYPE_TEXT: |
| free(s->u.text_s.str_first); |
| s->u.text_s.str_first = nullptr; |
| free(s->u.text_s.str_last); |
| s->u.text_s.str_last = nullptr; |
| break; |
| default: |
| break; |
| } |
| } |
| |
| void clear_timeseries_statistic(TimeseriesStatistic* s) { |
| memset(s, 0, sizeof(*s)); |
| tsfile_statistic_base(s)->type = TS_DATATYPE_INVALID; |
| } |
| |
| /** |
| * Fills @p out from C++ Statistic. On allocation failure returns E_OOM and |
| * clears/frees any partial string fields in @p out. |
| */ |
| int fill_timeseries_statistic(storage::Statistic* st, |
| TimeseriesStatistic* out) { |
| clear_timeseries_statistic(out); |
| if (st == nullptr) { |
| return common::E_OK; |
| } |
| const common::TSDataType t = st->get_type(); |
| switch (t) { |
| case common::BOOLEAN: { |
| auto* bs = static_cast<storage::BooleanStatistic*>(st); |
| TsFileBoolStatistic* p = &out->u.bool_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::BOOLEAN); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->sum = static_cast<double>(bs->sum_value_); |
| p->first_bool = bs->first_value_; |
| p->last_bool = bs->last_value_; |
| break; |
| } |
| case common::INT32: { |
| auto* is = static_cast<storage::Int32Statistic*>(st); |
| TsFileIntStatistic* p = &out->u.int_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::INT32); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->sum = static_cast<double>(is->sum_value_); |
| if (p->base.row_count > 0) { |
| p->min_int64 = static_cast<int64_t>(is->min_value_); |
| p->max_int64 = static_cast<int64_t>(is->max_value_); |
| p->first_int64 = static_cast<int64_t>(is->first_value_); |
| p->last_int64 = static_cast<int64_t>(is->last_value_); |
| } |
| break; |
| } |
| case common::DATE: { |
| auto* is = static_cast<storage::Int32Statistic*>(st); |
| TsFileIntStatistic* p = &out->u.int_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::DATE); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->sum = static_cast<double>(is->sum_value_); |
| if (p->base.row_count > 0) { |
| p->min_int64 = static_cast<int64_t>(is->min_value_); |
| p->max_int64 = static_cast<int64_t>(is->max_value_); |
| p->first_int64 = static_cast<int64_t>(is->first_value_); |
| p->last_int64 = static_cast<int64_t>(is->last_value_); |
| } |
| break; |
| } |
| case common::INT64: { |
| auto* ls = static_cast<storage::Int64Statistic*>(st); |
| TsFileIntStatistic* p = &out->u.int_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::INT64); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->sum = ls->sum_value_; |
| if (p->base.row_count > 0) { |
| p->min_int64 = ls->min_value_; |
| p->max_int64 = ls->max_value_; |
| p->first_int64 = ls->first_value_; |
| p->last_int64 = ls->last_value_; |
| } |
| break; |
| } |
| case common::TIMESTAMP: { |
| auto* ls = static_cast<storage::Int64Statistic*>(st); |
| TsFileIntStatistic* p = &out->u.int_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::TIMESTAMP); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->sum = ls->sum_value_; |
| if (p->base.row_count > 0) { |
| p->min_int64 = ls->min_value_; |
| p->max_int64 = ls->max_value_; |
| p->first_int64 = ls->first_value_; |
| p->last_int64 = ls->last_value_; |
| } |
| break; |
| } |
| case common::FLOAT: { |
| auto* fs = static_cast<storage::FloatStatistic*>(st); |
| TsFileFloatStatistic* p = &out->u.float_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::FLOAT); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->sum = static_cast<double>(fs->sum_value_); |
| if (p->base.row_count > 0) { |
| p->min_float64 = static_cast<double>(fs->min_value_); |
| p->max_float64 = static_cast<double>(fs->max_value_); |
| p->first_float64 = static_cast<double>(fs->first_value_); |
| p->last_float64 = static_cast<double>(fs->last_value_); |
| } |
| break; |
| } |
| case common::DOUBLE: { |
| auto* ds = static_cast<storage::DoubleStatistic*>(st); |
| TsFileFloatStatistic* p = &out->u.float_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::DOUBLE); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->sum = ds->sum_value_; |
| if (p->base.row_count > 0) { |
| p->min_float64 = ds->min_value_; |
| p->max_float64 = ds->max_value_; |
| p->first_float64 = ds->first_value_; |
| p->last_float64 = ds->last_value_; |
| } |
| break; |
| } |
| case common::STRING: { |
| auto* ss = static_cast<storage::StringStatistic*>(st); |
| TsFileStringStatistic* p = &out->u.string_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::STRING); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->str_min = dup_common_string_to_cstr(ss->min_value_); |
| if (p->str_min == nullptr) { |
| free_timeseries_statistic_heap(out); |
| clear_timeseries_statistic(out); |
| return common::E_OOM; |
| } |
| p->str_max = dup_common_string_to_cstr(ss->max_value_); |
| if (p->str_max == nullptr) { |
| free_timeseries_statistic_heap(out); |
| clear_timeseries_statistic(out); |
| return common::E_OOM; |
| } |
| p->str_first = dup_common_string_to_cstr(ss->first_value_); |
| if (p->str_first == nullptr) { |
| free_timeseries_statistic_heap(out); |
| clear_timeseries_statistic(out); |
| return common::E_OOM; |
| } |
| p->str_last = dup_common_string_to_cstr(ss->last_value_); |
| if (p->str_last == nullptr) { |
| free_timeseries_statistic_heap(out); |
| clear_timeseries_statistic(out); |
| return common::E_OOM; |
| } |
| break; |
| } |
| case common::TEXT: { |
| auto* ts = static_cast<storage::TextStatistic*>(st); |
| TsFileTextStatistic* p = &out->u.text_s; |
| p->base.has_statistic = true; |
| p->base.type = cpp_stat_type_to_c(common::TEXT); |
| p->base.row_count = st->get_count(); |
| p->base.start_time = st->start_time_; |
| p->base.end_time = st->get_end_time(); |
| p->str_first = dup_common_string_to_cstr(ts->first_value_); |
| if (p->str_first == nullptr) { |
| free_timeseries_statistic_heap(out); |
| clear_timeseries_statistic(out); |
| return common::E_OOM; |
| } |
| p->str_last = dup_common_string_to_cstr(ts->last_value_); |
| if (p->str_last == nullptr) { |
| free_timeseries_statistic_heap(out); |
| clear_timeseries_statistic(out); |
| return common::E_OOM; |
| } |
| break; |
| } |
| default: { |
| TsFileStatisticBase* b = tsfile_statistic_base(out); |
| b->has_statistic = true; |
| b->type = TS_DATATYPE_INVALID; |
| b->row_count = st->get_count(); |
| b->start_time = st->start_time_; |
| b->end_time = st->get_end_time(); |
| break; |
| } |
| } |
| return common::E_OK; |
| } |
| |
| int fill_timeline_statistic(storage::ITimeseriesIndex* idx, |
| TimeseriesStatistic* out) { |
| clear_timeseries_statistic(out); |
| if (idx == nullptr) { |
| return common::E_OK; |
| } |
| |
| auto* aligned_idx = dynamic_cast<storage::AlignedTimeseriesIndex*>(idx); |
| if (aligned_idx != nullptr && aligned_idx->time_ts_idx_ != nullptr && |
| aligned_idx->time_ts_idx_->get_statistic() != nullptr) { |
| auto* st = aligned_idx->time_ts_idx_->get_statistic(); |
| TsFileStatisticBase* b = tsfile_statistic_base(out); |
| b->has_statistic = true; |
| b->type = TS_DATATYPE_VECTOR; |
| b->row_count = st->get_count(); |
| b->start_time = st->start_time_; |
| b->end_time = st->get_end_time(); |
| return common::E_OK; |
| } |
| |
| if (idx->get_statistic() != nullptr && |
| idx->get_time_chunk_meta_list() == nullptr) { |
| auto* st = idx->get_statistic(); |
| TsFileStatisticBase* b = tsfile_statistic_base(out); |
| b->has_statistic = true; |
| b->type = TS_DATATYPE_VECTOR; |
| b->row_count = st->get_count(); |
| b->start_time = st->start_time_; |
| b->end_time = st->get_end_time(); |
| return common::E_OK; |
| } |
| |
| auto* list = idx->get_time_chunk_meta_list(); |
| if (list == nullptr) { |
| list = idx->get_chunk_meta_list(); |
| } |
| if (list == nullptr) { |
| return common::E_OK; |
| } |
| |
| int64_t row_count = 0; |
| int64_t start_time = 0; |
| int64_t end_time = 0; |
| bool has_statistic = false; |
| for (auto it = list->begin(); it != list->end(); it++) { |
| auto* chunk_meta = it.get(); |
| if (chunk_meta == nullptr || chunk_meta->statistic_ == nullptr || |
| chunk_meta->statistic_->count_ <= 0) { |
| continue; |
| } |
| if (!has_statistic) { |
| start_time = chunk_meta->statistic_->start_time_; |
| end_time = chunk_meta->statistic_->end_time_; |
| has_statistic = true; |
| } else { |
| start_time = |
| std::min(start_time, chunk_meta->statistic_->start_time_); |
| end_time = std::max(end_time, chunk_meta->statistic_->end_time_); |
| } |
| row_count += chunk_meta->statistic_->count_; |
| } |
| |
| if (!has_statistic) { |
| return common::E_OK; |
| } |
| |
| TsFileStatisticBase* b = tsfile_statistic_base(out); |
| b->has_statistic = true; |
| b->type = TS_DATATYPE_VECTOR; |
| b->row_count = row_count; |
| b->start_time = start_time; |
| b->end_time = end_time; |
| return common::E_OK; |
| } |
| |
| void free_device_timeseries_metadata_entries_partial( |
| DeviceTimeseriesMetadataEntry* entries, size_t filled_count) { |
| if (entries == nullptr) { |
| return; |
| } |
| for (size_t i = 0; i < filled_count; i++) { |
| tsfile_device_id_free_contents(&entries[i].device); |
| if (entries[i].timeseries != nullptr) { |
| for (uint32_t j = 0; j < entries[i].timeseries_count; j++) { |
| free_timeseries_statistic_heap( |
| &entries[i].timeseries[j].statistic); |
| free_timeseries_statistic_heap( |
| &entries[i].timeseries[j].timeline_statistic); |
| free(entries[i].timeseries[j].measurement_name); |
| } |
| free(entries[i].timeseries); |
| entries[i].timeseries = nullptr; |
| } |
| } |
| free(entries); |
| } |
| |
| /** |
| * Copies path, table name, and segment strings from IDeviceID into heap |
| * buffers. On failure, frees any partial allocations and returns an error code. |
| */ |
| int duplicate_ideviceid_to_device_fields(storage::IDeviceID* id, |
| char** out_path, char** out_table_name, |
| uint32_t* out_segment_count, |
| char*** out_segments) { |
| *out_path = nullptr; |
| *out_table_name = nullptr; |
| *out_segment_count = 0; |
| *out_segments = nullptr; |
| auto cleanup = [&]() { |
| free(*out_path); |
| free(*out_table_name); |
| *out_path = nullptr; |
| *out_table_name = nullptr; |
| }; |
| try { |
| if (id == nullptr) { |
| *out_path = strdup(""); |
| *out_table_name = strdup(""); |
| if (*out_path == nullptr || *out_table_name == nullptr) { |
| free(*out_path); |
| free(*out_table_name); |
| *out_path = nullptr; |
| *out_table_name = nullptr; |
| return common::E_OOM; |
| } |
| return common::E_OK; |
| } |
| const std::string dname = id->get_device_name(); |
| *out_path = strdup(dname.c_str()); |
| if (*out_path == nullptr) { |
| return common::E_OOM; |
| } |
| #ifdef ENABLE_TEST |
| using common::g_all_inject_points; |
| using common::InjectPoint; |
| DBUG_EXECUTE_IF("device_id_copy_oom", throw std::bad_alloc();); |
| DBUG_EXECUTE_IF("device_id_copy_error", throw common::E_FILE_READ_ERR;); |
| #endif |
| const std::string tname = id->get_table_name(); |
| *out_table_name = strdup(tname.c_str()); |
| if (*out_table_name == nullptr) { |
| free(*out_path); |
| *out_path = nullptr; |
| return common::E_OOM; |
| } |
| const int n = id->segment_num(); |
| if (n <= 0) { |
| return common::E_OK; |
| } |
| const auto& segs = id->get_segments(); |
| auto* seg_arr = |
| static_cast<char**>(malloc(sizeof(char*) * static_cast<size_t>(n))); |
| if (seg_arr == nullptr) { |
| free(*out_table_name); |
| *out_table_name = nullptr; |
| free(*out_path); |
| *out_path = nullptr; |
| return common::E_OOM; |
| } |
| memset(seg_arr, 0, sizeof(char*) * static_cast<size_t>(n)); |
| for (int i = 0; i < n; i++) { |
| const std::string* ps = |
| (static_cast<size_t>(i) < segs.size()) ? segs[i] : nullptr; |
| // A null tag segment is exposed as a NULL pointer so callers can |
| // distinguish a missing/null tag from the literal string "null". |
| if (ps == nullptr) { |
| seg_arr[i] = nullptr; |
| continue; |
| } |
| seg_arr[i] = strdup(ps->c_str()); |
| if (seg_arr[i] == nullptr) { |
| for (int j = 0; j < i; j++) { |
| free(seg_arr[j]); |
| } |
| free(seg_arr); |
| free(*out_table_name); |
| *out_table_name = nullptr; |
| free(*out_path); |
| *out_path = nullptr; |
| return common::E_OOM; |
| } |
| } |
| *out_segment_count = static_cast<uint32_t>(n); |
| *out_segments = seg_arr; |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| cleanup(); |
| return common::E_OOM; |
| } catch (...) { |
| cleanup(); |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| int fill_device_id_from_ideviceid(storage::IDeviceID* id, DeviceID* out) { |
| memset(out, 0, sizeof(*out)); |
| return duplicate_ideviceid_to_device_fields( |
| id, &out->path, &out->table_name, &out->segment_count, &out->segments); |
| } |
| |
| ERRNO populate_c_metadata_map_from_cpp( |
| storage::DeviceTimeseriesMetadataMap& cpp_map, |
| DeviceTimeseriesMetadataMap* out_map) { |
| try { |
| if (cpp_map.empty()) { |
| return common::E_OK; |
| } |
| const uint32_t dev_n = static_cast<uint32_t>(cpp_map.size()); |
| auto* entries = static_cast<DeviceTimeseriesMetadataEntry*>( |
| malloc(sizeof(DeviceTimeseriesMetadataEntry) * dev_n)); |
| if (entries == nullptr) { |
| return common::E_OOM; |
| } |
| memset(entries, 0, sizeof(DeviceTimeseriesMetadataEntry) * dev_n); |
| auto cleanup = [dev_n](DeviceTimeseriesMetadataEntry* data) { |
| free_device_timeseries_metadata_entries_partial(data, dev_n); |
| }; |
| std::unique_ptr<DeviceTimeseriesMetadataEntry, decltype(cleanup)> guard( |
| entries, cleanup); |
| size_t di = 0; |
| for (const auto& kv : cpp_map) { |
| DeviceTimeseriesMetadataEntry& e = entries[di]; |
| const int dup_rc = fill_device_id_from_ideviceid( |
| kv.first ? kv.first.get() : nullptr, &e.device); |
| if (dup_rc != common::E_OK) { |
| return dup_rc; |
| } |
| const auto& vec = kv.second; |
| uint32_t n_ts = 0; |
| for (const auto& idx_nz : vec) { |
| if (idx_nz != nullptr) { |
| n_ts++; |
| } |
| } |
| e.timeseries_count = n_ts; |
| if (e.timeseries_count == 0) { |
| e.timeseries = nullptr; |
| di++; |
| continue; |
| } |
| e.timeseries = static_cast<TimeseriesMetadata*>( |
| malloc(sizeof(TimeseriesMetadata) * e.timeseries_count)); |
| if (e.timeseries == nullptr) { |
| return common::E_OOM; |
| } |
| memset(e.timeseries, 0, |
| sizeof(TimeseriesMetadata) * e.timeseries_count); |
| uint32_t slot = 0; |
| for (const auto& idx : vec) { |
| if (idx == nullptr) { |
| continue; |
| } |
| TimeseriesMetadata& m = e.timeseries[slot]; |
| common::String mn = idx->get_measurement_name(); |
| m.measurement_name = strdup(mn.to_std_string().c_str()); |
| if (m.measurement_name == nullptr) { |
| return common::E_OOM; |
| } |
| #ifdef ENABLE_TEST |
| using common::g_all_inject_points; |
| using common::InjectPoint; |
| DBUG_EXECUTE_IF("timeseries_metadata_copy_oom", |
| throw std::bad_alloc();); |
| DBUG_EXECUTE_IF("timeseries_metadata_copy_error", |
| throw common::E_FILE_READ_ERR;); |
| #endif |
| auto* aligned_idx = |
| dynamic_cast<storage::AlignedTimeseriesIndex*>(idx.get()); |
| if (aligned_idx != nullptr && |
| aligned_idx->value_ts_idx_ != nullptr) { |
| m.data_type = static_cast<TSDataType>( |
| aligned_idx->value_ts_idx_->get_data_type()); |
| const storage::TimeseriesIndex* value_idx = |
| aligned_idx->value_ts_idx_; |
| const storage::TimeseriesIndex* time_idx = |
| aligned_idx->time_ts_idx_; |
| if (value_idx->get_metadata_offset() >= 0) { |
| m.value_metadata_offset = static_cast<uint64_t>( |
| value_idx->get_metadata_offset()); |
| m.value_metadata_length = |
| value_idx->get_metadata_length(); |
| } |
| if (time_idx != nullptr && |
| time_idx->get_metadata_offset() >= 0) { |
| m.time_metadata_offset = static_cast<uint64_t>( |
| time_idx->get_metadata_offset()); |
| m.time_metadata_length = |
| time_idx->get_metadata_length(); |
| } |
| m.layout = 1; |
| } else { |
| m.data_type = static_cast<TSDataType>(idx->get_data_type()); |
| const storage::TimeseriesIndex* value_idx = |
| dynamic_cast<const storage::TimeseriesIndex*>( |
| idx.get()); |
| if (value_idx != nullptr && |
| value_idx->get_metadata_offset() >= 0) { |
| m.value_metadata_offset = static_cast<uint64_t>( |
| value_idx->get_metadata_offset()); |
| m.value_metadata_length = |
| value_idx->get_metadata_length(); |
| } |
| } |
| storage::Statistic* st = idx->get_statistic(); |
| int32_t chunk_cnt = 0; |
| auto* cl = aligned_idx != nullptr |
| ? idx->get_value_chunk_meta_list() |
| : idx->get_chunk_meta_list(); |
| if (cl != nullptr) { |
| chunk_cnt = static_cast<int32_t>(cl->size()); |
| } |
| m.chunk_meta_count = chunk_cnt; |
| if (chunk_cnt >= 0 && m.value_metadata_length > 0) { |
| m.locator_flags |= 1; |
| } |
| if (aligned_idx != nullptr) { |
| auto* time_chunks = idx->get_time_chunk_meta_list(); |
| if (time_chunks != nullptr) { |
| m.time_chunk_meta_count = |
| static_cast<uint32_t>(time_chunks->size()); |
| } |
| if (m.time_metadata_length == 0 || |
| m.time_chunk_meta_count != |
| static_cast<uint32_t>(chunk_cnt)) { |
| m.locator_flags &= ~static_cast<uint16_t>(1); |
| } |
| } |
| const int st_rc = fill_timeseries_statistic(st, &m.statistic); |
| if (st_rc != common::E_OK) { |
| return st_rc; |
| } |
| const int timeline_st_rc = |
| fill_timeline_statistic(idx.get(), &m.timeline_statistic); |
| if (timeline_st_rc != common::E_OK) { |
| return timeline_st_rc; |
| } |
| slot++; |
| } |
| di++; |
| } |
| out_map->entries = guard.release(); |
| out_map->device_count = dev_n; |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } catch (...) { |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| } // namespace |
| |
| void tsfile_free_device_id_array(DeviceID* devices, uint32_t length) { |
| if (devices == nullptr) { |
| return; |
| } |
| for (uint32_t i = 0; i < length; i++) { |
| tsfile_device_id_free_contents(&devices[i]); |
| } |
| free(devices); |
| } |
| |
| ERRNO tsfile_reader_get_all_devices(TsFileReader reader, DeviceID** out_devices, |
| uint32_t* out_length) { |
| if (reader == nullptr || out_devices == nullptr || out_length == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| *out_devices = nullptr; |
| *out_length = 0; |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| 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; |
| } |
| auto* arr = static_cast<DeviceID*>(malloc(sizeof(DeviceID) * ids.size())); |
| if (arr == nullptr) { |
| return common::E_OOM; |
| } |
| memset(arr, 0, sizeof(DeviceID) * ids.size()); |
| for (size_t i = 0; i < ids.size(); i++) { |
| const int rc = fill_device_id_from_ideviceid(ids[i].get(), &arr[i]); |
| if (rc != common::E_OK) { |
| tsfile_free_device_id_array(arr, static_cast<uint32_t>(i)); |
| return rc; |
| } |
| } |
| *out_devices = arr; |
| *out_length = static_cast<uint32_t>(ids.size()); |
| return common::E_OK; |
| } |
| |
| ERRNO tsfile_reader_get_timeseries_metadata_all( |
| TsFileReader reader, DeviceTimeseriesMetadataMap* out_map) { |
| if (out_map != nullptr) { |
| out_map->entries = nullptr; |
| out_map->device_count = 0; |
| } |
| if (reader == nullptr || out_map == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::DeviceTimeseriesMetadataMap cpp_map; |
| const int ret = r->get_timeseries_metadata(cpp_map); |
| if (ret != common::E_OK) return ret; |
| return populate_c_metadata_map_from_cpp(cpp_map, out_map); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } catch (...) { |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| ERRNO tsfile_reader_get_timeseries_metadata_for_devices( |
| TsFileReader reader, const DeviceID* devices, uint32_t length, |
| DeviceTimeseriesMetadataMap* out_map) { |
| if (out_map != nullptr) { |
| out_map->entries = nullptr; |
| out_map->device_count = 0; |
| } |
| if (reader == nullptr || out_map == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| if (length == 0) { |
| return common::E_OK; |
| } |
| if (devices == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| for (uint32_t i = 0; i < length; i++) { |
| if (devices[i].segment_count > 0) { |
| if (devices[i].segments == nullptr || |
| devices[i].segments[0] == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| } else if (devices[i].path == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| } |
| try { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| std::vector<std::shared_ptr<storage::IDeviceID>> query_ids; |
| query_ids.reserve(length); |
| for (uint32_t i = 0; i < length; i++) { |
| const DeviceID& device = devices[i]; |
| if (device.segment_count == 0) { |
| query_ids.push_back( |
| std::make_shared<storage::StringArrayDeviceID>( |
| std::string(device.path))); |
| continue; |
| } |
| std::vector<std::string> values; |
| values.reserve(device.segment_count); |
| std::vector<std::string*> segments(device.segment_count, nullptr); |
| for (uint32_t j = 0; j < device.segment_count; ++j) { |
| if (device.segments[j] != nullptr) { |
| values.emplace_back(device.segments[j]); |
| segments[j] = &values.back(); |
| } |
| } |
| query_ids.push_back( |
| std::make_shared<storage::StringArrayDeviceID>(segments)); |
| } |
| storage::DeviceTimeseriesMetadataMap cpp_map; |
| const int ret = r->get_timeseries_metadata(query_ids, cpp_map); |
| if (ret != common::E_OK) return ret; |
| return populate_c_metadata_map_from_cpp(cpp_map, out_map); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } catch (...) { |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| void tsfile_free_device_timeseries_metadata_map( |
| DeviceTimeseriesMetadataMap* map) { |
| if (map == nullptr) { |
| return; |
| } |
| free_device_timeseries_metadata_entries_partial(map->entries, |
| map->device_count); |
| map->entries = nullptr; |
| map->device_count = 0; |
| } |
| |
| void tsfile_free_tsfile_properties(TsFileProperty* properties, |
| uint32_t length) { |
| if (properties == nullptr) { |
| return; |
| } |
| for (uint32_t i = 0; i < length; i++) { |
| free(properties[i].key); |
| properties[i].key = nullptr; |
| free(properties[i].value); |
| properties[i].value = nullptr; |
| properties[i].key_len = 0; |
| properties[i].value_len = 0; |
| properties[i].is_null = false; |
| } |
| free(properties); |
| } |
| |
| ERRNO tsfile_reader_get_tsfile_properties(TsFileReader reader, |
| TsFileProperty** out_properties, |
| uint32_t* out_length) { |
| if (out_properties == nullptr || out_length == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| *out_properties = nullptr; |
| *out_length = 0; |
| if (reader == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| |
| try { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::TsFileProperties cpp_properties = r->get_tsfile_properties(); |
| if (cpp_properties.size() > |
| static_cast<size_t>(std::numeric_limits<uint32_t>::max()) || |
| cpp_properties.size() > |
| std::numeric_limits<size_t>::max() / sizeof(TsFileProperty)) { |
| return common::E_OUT_OF_RANGE; |
| } |
| if (cpp_properties.empty()) { |
| return common::E_OK; |
| } |
| |
| auto* properties = static_cast<TsFileProperty*>( |
| malloc(sizeof(TsFileProperty) * cpp_properties.size())); |
| if (properties == nullptr) { |
| return common::E_OOM; |
| } |
| memset(properties, 0, sizeof(TsFileProperty) * cpp_properties.size()); |
| |
| uint32_t property_index = 0; |
| for (const auto& cpp_property : cpp_properties) { |
| TsFileProperty& property = properties[property_index]; |
| if (cpp_property.first.size() > |
| static_cast<size_t>(std::numeric_limits<uint32_t>::max())) { |
| tsfile_free_tsfile_properties(properties, property_index); |
| return common::E_OUT_OF_RANGE; |
| } |
| property.key_len = static_cast<uint32_t>(cpp_property.first.size()); |
| property.key = static_cast<char*>( |
| malloc(static_cast<size_t>(property.key_len) + 1U)); |
| if (property.key == nullptr) { |
| tsfile_free_tsfile_properties(properties, property_index + 1); |
| return common::E_OOM; |
| } |
| if (property.key_len > 0) { |
| memcpy(property.key, cpp_property.first.data(), |
| property.key_len); |
| } |
| property.key[property.key_len] = '\0'; |
| |
| const storage::TsFilePropertyValue& cpp_value = cpp_property.second; |
| property.is_null = cpp_value.is_null; |
| if (!cpp_value.is_null) { |
| if (cpp_value.value.size() > |
| static_cast<size_t>(std::numeric_limits<uint32_t>::max())) { |
| tsfile_free_tsfile_properties(properties, |
| property_index + 1); |
| return common::E_OUT_OF_RANGE; |
| } |
| property.value_len = |
| static_cast<uint32_t>(cpp_value.value.size()); |
| if (property.value_len > 0) { |
| property.value = |
| static_cast<uint8_t*>(malloc(property.value_len)); |
| if (property.value == nullptr) { |
| tsfile_free_tsfile_properties(properties, |
| property_index + 1); |
| return common::E_OOM; |
| } |
| memcpy(property.value, cpp_value.value.data(), |
| property.value_len); |
| } |
| } |
| property_index++; |
| } |
| |
| *out_properties = properties; |
| *out_length = static_cast<uint32_t>(cpp_properties.size()); |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| // delete pointer |
| void _free_tsfile_ts_record(TsRecord* record) { |
| if (*record != nullptr) { |
| delete static_cast<storage::TsRecord*>(*record); |
| } |
| *record = nullptr; |
| } |
| |
| void free_tablet(Tablet* tablet) { |
| if (*tablet != nullptr) { |
| delete static_cast<storage::Tablet*>(*tablet); |
| } |
| *tablet = nullptr; |
| } |
| |
| void free_tsfile_result_set(ResultSet* result_set) { |
| if (*result_set != nullptr) { |
| delete static_cast<storage::ResultSet*>(*result_set); |
| } |
| *result_set = nullptr; |
| } |
| |
| void free_result_set_meta_data(ResultSetMetaData result_set_meta_data) { |
| for (int i = 0; i < result_set_meta_data.column_num; i++) { |
| free(result_set_meta_data.column_names[i]); |
| } |
| free(result_set_meta_data.column_names); |
| free(result_set_meta_data.data_types); |
| } |
| |
| void free_device_schema(DeviceSchema schema) { |
| free(schema.device_name); |
| for (int i = 0; i < schema.timeseries_num; i++) { |
| free_timeseries_schema(schema.timeseries_schema[i]); |
| } |
| free(schema.timeseries_schema); |
| } |
| void free_timeseries_schema(TimeseriesSchema schema) { |
| free(schema.timeseries_name); |
| } |
| void free_table_schema(TableSchema schema) { |
| free(schema.table_name); |
| for (int i = 0; i < schema.column_num; i++) { |
| free_column_schema(schema.column_schemas[i]); |
| } |
| if (schema.column_num > 0) { |
| free(schema.column_schemas); |
| } |
| } |
| void free_column_schema(ColumnSchema schema) { free(schema.column_name); } |
| |
| void free_write_file(WriteFile* write_file) { |
| auto f = static_cast<storage::WriteFile*>(*write_file); |
| delete f; |
| *write_file = nullptr; |
| } |
| |
| // For Python API |
| TsFileWriter _tsfile_writer_new(const char* pathname, uint64_t memory_threshold, |
| ERRNO* err_code) { |
| init_tsfile_config(); |
| auto writer = new storage::TsFileWriter(); |
| int flags = O_WRONLY | O_CREAT | O_TRUNC; |
| #ifdef _WIN32 |
| flags |= O_BINARY; |
| #endif |
| int ret = writer->open(pathname, flags, 0644); |
| common::g_config_value_.chunk_group_size_threshold_ = memory_threshold; |
| if (ret != common::E_OK) { |
| delete writer; |
| *err_code = ret; |
| return nullptr; |
| } |
| return writer; |
| } |
| |
| Tablet _tablet_new_with_target_name(const char* device_id, |
| char** column_name_list, |
| TSDataType* data_types, int column_num, |
| int max_rows) { |
| std::vector<std::string> measurement_list; |
| std::vector<common::TSDataType> data_type_list; |
| for (int i = 0; i < column_num; i++) { |
| measurement_list.emplace_back(column_name_list[i]); |
| data_type_list.push_back( |
| static_cast<common::TSDataType>(*(data_types + i))); |
| } |
| if (device_id != nullptr) { |
| return new storage::Tablet(device_id, &measurement_list, |
| &data_type_list, max_rows); |
| } else { |
| return new storage::Tablet(measurement_list, data_type_list, max_rows); |
| } |
| } |
| |
| ERRNO _tsfile_writer_register_table(TsFileWriter writer, TableSchema* schema) { |
| if (writer == nullptr || schema == nullptr || |
| schema->column_schemas == nullptr || schema->table_name == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| if (schema->column_num <= 0) { |
| return common::E_INVALID_SCHEMA; |
| } |
| std::vector<storage::MeasurementSchema*> measurement_schemas; |
| std::vector<common::ColumnCategory> column_categories; |
| measurement_schemas.resize(schema->column_num); |
| for (int i = 0; i < schema->column_num; i++) { |
| ColumnSchema* cur_schema = schema->column_schemas + i; |
| measurement_schemas[i] = new storage::MeasurementSchema( |
| cur_schema->column_name, |
| static_cast<common::TSDataType>(cur_schema->data_type)); |
| column_categories.push_back( |
| static_cast<common::ColumnCategory>(cur_schema->column_category)); |
| } |
| auto tsfile_writer = static_cast<storage::TsFileWriter*>(writer); |
| return tsfile_writer->register_table(std::make_shared<storage::TableSchema>( |
| schema->table_name, measurement_schemas, column_categories)); |
| } |
| |
| ERRNO _tsfile_writer_register_timeseries(TsFileWriter writer, |
| const char* device_id, |
| const TimeseriesSchema* schema) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| |
| int ret = w->register_timeseries( |
| device_id, |
| storage::MeasurementSchema( |
| schema->timeseries_name, |
| static_cast<common::TSDataType>(schema->data_type), |
| static_cast<common::TSEncoding>(schema->encoding), |
| static_cast<common::CompressionType>(schema->compression))); |
| return ret; |
| } |
| |
| ERRNO _tsfile_writer_register_device(TsFileWriter writer, |
| const device_schema* device_schema) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| for (int column_id = 0; column_id < device_schema->timeseries_num; |
| column_id++) { |
| TimeseriesSchema schema = device_schema->timeseries_schema[column_id]; |
| const ERRNO ret = w->register_timeseries( |
| device_schema->device_name, |
| storage::MeasurementSchema( |
| schema.timeseries_name, |
| static_cast<common::TSDataType>(schema.data_type), |
| static_cast<common::TSEncoding>(schema.encoding), |
| static_cast<common::CompressionType>(schema.compression))); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| } |
| return common::E_OK; |
| } |
| |
| ERRNO _tsfile_writer_write_tablet(TsFileWriter writer, Tablet tablet) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| const auto* tbl = static_cast<storage::Tablet*>(tablet); |
| return w->write_tablet(*tbl); |
| } |
| |
| ERRNO _tsfile_writer_write_table(TsFileWriter writer, Tablet tablet) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| auto* tbl = static_cast<storage::Tablet*>(tablet); |
| return w->write_table(*tbl); |
| } |
| |
| ERRNO _tsfile_writer_write_arrow_table(TsFileWriter writer, |
| const char* table_name, |
| ArrowArray* array, ArrowSchema* schema, |
| int time_col_index) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| std::shared_ptr<storage::TableSchema> reg_schema = |
| w->get_table_schema(table_name ? std::string(table_name) : ""); |
| storage::Tablet* tablet = nullptr; |
| int ret = arrow::ArrowStructToTablet( |
| table_name, array, schema, reg_schema.get(), &tablet, time_col_index); |
| if (ret != common::E_OK) return ret; |
| ret = w->write_table(*tablet); |
| delete tablet; |
| return ret; |
| } |
| |
| ERRNO _tsfile_writer_write_ts_record(TsFileWriter writer, TsRecord data) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| const storage::TsRecord* record = static_cast<storage::TsRecord*>(data); |
| const int ret = w->write_record(*record); |
| return ret; |
| } |
| |
| ERRNO _tsfile_writer_close(TsFileWriter writer) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| int ret = w->close(); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| delete w; |
| return ret; |
| } |
| |
| ERRNO _tsfile_writer_flush(TsFileWriter writer) { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| return w->flush(); |
| } |
| |
| ERRNO _tsfile_writer_add_tsfile_property(TsFileWriter writer, const char* key, |
| uint32_t key_len, const uint8_t* value, |
| uint32_t value_len) { |
| if (writer == nullptr || key == nullptr || |
| (value == nullptr && value_len > 0)) { |
| return common::E_INVALID_ARG; |
| } |
| if (key_len > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()) || |
| value_len > |
| static_cast<uint32_t>(std::numeric_limits<int32_t>::max())) { |
| return common::E_OUT_OF_RANGE; |
| } |
| try { |
| auto* w = static_cast<storage::TsFileWriter*>(writer); |
| return w->add_tsfile_property(std::string(key, key_len), value, |
| value_len); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| TsFileGenericWriter tsfile_generic_writer_new(const char* pathname, |
| uint64_t memory_threshold, |
| ERRNO* err_code) { |
| // Public C entry point: guard null arguments before the private delegate |
| // dereferences them, and keep std::bad_alloc from crossing the C ABI. |
| // See tsfile_writer_new() above for the null-guard rationale. |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| *err_code = common::E_OK; |
| if (pathname == nullptr) { |
| *err_code = common::E_INVALID_ARG; |
| return nullptr; |
| } |
| try { |
| return _tsfile_writer_new(pathname, memory_threshold, err_code); |
| } catch (const std::bad_alloc&) { |
| *err_code = common::E_OOM; |
| return nullptr; |
| } |
| } |
| |
| Tablet tablet_new_with_target_name(const char* target_name, |
| char** column_name_list, |
| TSDataType* data_types, int column_num, |
| int max_rows) { |
| // Public C entry point: reject arguments that would make the private |
| // delegate read out-of-bounds memory (negative column_num, null lists or |
| // null column names) before touching caller buffers. The native Tablet |
| // constructor asserts 0 < max_rows < (1 << 30); reject values outside |
| // that domain here instead of letting an assertion abort the host |
| // process. Returns NULL on invalid arguments or allocation failure. |
| if (column_num < 0 || max_rows <= 0 || max_rows >= (1 << 30)) { |
| return nullptr; |
| } |
| if (column_num > 0) { |
| if (column_name_list == nullptr || data_types == nullptr) { |
| return nullptr; |
| } |
| for (int i = 0; i < column_num; i++) { |
| if (column_name_list[i] == nullptr) { |
| return nullptr; |
| } |
| } |
| } |
| try { |
| return _tablet_new_with_target_name(target_name, column_name_list, |
| data_types, column_num, max_rows); |
| } catch (const std::bad_alloc&) { |
| return nullptr; |
| } |
| } |
| |
| ERRNO tsfile_generic_writer_register_table(TsFileGenericWriter writer, |
| TableSchema* schema) { |
| if (writer == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| return _tsfile_writer_register_table(writer, schema); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ERRNO tsfile_generic_writer_register_timeseries( |
| TsFileGenericWriter writer, const char* device_id, |
| const TimeseriesSchema* schema) { |
| if (writer == nullptr || device_id == nullptr || schema == nullptr || |
| schema->timeseries_name == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| return _tsfile_writer_register_timeseries(writer, device_id, schema); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ERRNO tsfile_generic_writer_register_device(TsFileGenericWriter writer, |
| const DeviceSchema* device_schema) { |
| if (writer == nullptr || device_schema == nullptr || |
| device_schema->device_name == nullptr || |
| device_schema->timeseries_num < 0) { |
| return common::E_INVALID_ARG; |
| } |
| if (device_schema->timeseries_num > 0 && |
| device_schema->timeseries_schema == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| for (int i = 0; i < device_schema->timeseries_num; i++) { |
| if (device_schema->timeseries_schema[i].timeseries_name == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| } |
| try { |
| return _tsfile_writer_register_device(writer, device_schema); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ERRNO tsfile_generic_writer_write_tree_tablet(TsFileGenericWriter writer, |
| Tablet tablet) { |
| if (writer == nullptr || tablet == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| return _tsfile_writer_write_tablet(writer, tablet); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ERRNO tsfile_generic_writer_write_table_tablet(TsFileGenericWriter writer, |
| Tablet tablet) { |
| if (writer == nullptr || tablet == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| return _tsfile_writer_write_table(writer, tablet); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ERRNO tsfile_generic_writer_flush(TsFileGenericWriter writer) { |
| if (writer == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| return _tsfile_writer_flush(writer); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ERRNO tsfile_generic_writer_add_tsfile_property(TsFileGenericWriter writer, |
| const char* key, |
| uint32_t key_len, |
| const uint8_t* value, |
| uint32_t value_len) { |
| return _tsfile_writer_add_tsfile_property(writer, key, key_len, value, |
| value_len); |
| } |
| |
| ERRNO tsfile_generic_writer_close(TsFileGenericWriter writer) { |
| // Matches the public tsfile_writer_close(): a null handle is treated as |
| // already closed. |
| if (writer == nullptr) { |
| return common::E_OK; |
| } |
| try { |
| return _tsfile_writer_close(writer); |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } |
| } |
| |
| ResultSet _tsfile_reader_query_device(TsFileReader reader, |
| const char* device_name, |
| char** sensor_name, uint32_t sensor_num, |
| Timestamp start_time, Timestamp end_time, |
| ERRNO* err_code) { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| std::vector<std::string> selected_paths; |
| selected_paths.reserve(sensor_num); |
| for (uint32_t i = 0; i < sensor_num; i++) { |
| selected_paths.push_back(std::string(device_name) + "." + |
| std::string(sensor_name[i])); |
| } |
| storage::ResultSet* qds = nullptr; |
| *err_code = r->query(selected_paths, start_time, end_time, qds); |
| return qds; |
| } |
| |
| // ============== Tag Filter API Implementation ============== |
| |
| #define DEFINE_TAG_FILTER_FACTORY(name, op) \ |
| TagFilterHandle tsfile_tag_filter_##name( \ |
| TsFileReader reader, const char* table_name, const char* column_name, \ |
| const char* value) { \ |
| ERRNO error_code = common::E_OK; \ |
| return tsfile_tag_filter_create(reader, table_name, column_name, \ |
| value, TAG_FILTER_##op, &error_code); \ |
| } |
| |
| DEFINE_TAG_FILTER_FACTORY(eq, EQ) |
| DEFINE_TAG_FILTER_FACTORY(neq, NEQ) |
| DEFINE_TAG_FILTER_FACTORY(lt, LT) |
| DEFINE_TAG_FILTER_FACTORY(lteq, LTEQ) |
| DEFINE_TAG_FILTER_FACTORY(gt, GT) |
| DEFINE_TAG_FILTER_FACTORY(gteq, GTEQ) |
| |
| #undef DEFINE_TAG_FILTER_FACTORY |
| |
| ERRNO tsfile_tag_filter_create_checked(TsFileReader reader, |
| const char* table_name, |
| const char* column_name, |
| const char* value, TagFilterOp op, |
| TagFilterHandle* out_filter) { |
| if (out_filter != nullptr) { |
| *out_filter = nullptr; |
| } |
| if (reader == nullptr || table_name == nullptr || column_name == nullptr || |
| out_filter == nullptr || |
| (value == nullptr && op != TAG_FILTER_IS_NULL && |
| op != TAG_FILTER_IS_NOT_NULL)) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| std::shared_ptr<storage::TableSchema> schema; |
| const int ret = r->get_table_schema(table_name, schema); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| storage::TagFilterBuilder builder(schema.get()); |
| storage::Filter* filter = nullptr; |
| switch (op) { |
| case TAG_FILTER_EQ: |
| filter = builder.eq(column_name, value); |
| break; |
| case TAG_FILTER_NEQ: |
| filter = builder.neq(column_name, value); |
| break; |
| case TAG_FILTER_LT: |
| filter = builder.lt(column_name, value); |
| break; |
| case TAG_FILTER_LTEQ: |
| filter = builder.lteq(column_name, value); |
| break; |
| case TAG_FILTER_GT: |
| filter = builder.gt(column_name, value); |
| break; |
| case TAG_FILTER_GTEQ: |
| filter = builder.gteq(column_name, value); |
| break; |
| case TAG_FILTER_REGEXP: |
| filter = builder.reg_exp(column_name, value); |
| break; |
| case TAG_FILTER_NOT_REGEXP: |
| filter = builder.not_reg_exp(column_name, value); |
| break; |
| case TAG_FILTER_IS_NULL: |
| filter = builder.is_null(column_name); |
| break; |
| case TAG_FILTER_IS_NOT_NULL: |
| filter = builder.is_not_null(column_name); |
| break; |
| default: |
| return common::E_INVALID_ARG; |
| } |
| if (filter == nullptr) { |
| return common::E_COLUMN_NOT_EXIST; |
| } |
| *out_filter = static_cast<void*>(filter); |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } catch (...) { |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| TagFilterHandle tsfile_tag_filter_create(TsFileReader reader, |
| const char* table_name, |
| const char* column_name, |
| const char* value, TagFilterOp op, |
| ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| TagFilterHandle filter = nullptr; |
| *err_code = tsfile_tag_filter_create_checked( |
| reader, table_name, column_name, value, op, &filter); |
| if (*err_code == common::E_TABLE_NOT_EXIST) { |
| *err_code = common::E_INVALID_ARG; |
| } |
| return *err_code == common::E_OK ? filter : nullptr; |
| } |
| |
| ERRNO tsfile_tag_filter_between_checked(TsFileReader reader, |
| const char* table_name, |
| const char* column_name, |
| const char* lower, const char* upper, |
| bool is_not, |
| TagFilterHandle* out_filter) { |
| if (out_filter != nullptr) { |
| *out_filter = nullptr; |
| } |
| if (reader == nullptr || table_name == nullptr || column_name == nullptr || |
| lower == nullptr || upper == nullptr || out_filter == nullptr) { |
| return common::E_INVALID_ARG; |
| } |
| try { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| std::shared_ptr<storage::TableSchema> schema; |
| const int ret = r->get_table_schema(table_name, schema); |
| if (ret != common::E_OK) { |
| return ret; |
| } |
| storage::TagFilterBuilder builder(schema.get()); |
| storage::Filter* filter = |
| is_not ? builder.not_between_and(column_name, lower, upper) |
| : builder.between_and(column_name, lower, upper); |
| if (filter == nullptr) { |
| return common::E_COLUMN_NOT_EXIST; |
| } |
| *out_filter = static_cast<void*>(filter); |
| return common::E_OK; |
| } catch (const std::bad_alloc&) { |
| return common::E_OOM; |
| } catch (...) { |
| return common::E_FILE_READ_ERR; |
| } |
| } |
| |
| TagFilterHandle tsfile_tag_filter_between(TsFileReader reader, |
| const char* table_name, |
| const char* column_name, |
| const char* lower, const char* upper, |
| bool is_not, ERRNO* err_code) { |
| if (err_code == nullptr) { |
| return nullptr; |
| } |
| TagFilterHandle filter = nullptr; |
| *err_code = tsfile_tag_filter_between_checked( |
| reader, table_name, column_name, lower, upper, is_not, &filter); |
| if (*err_code == common::E_TABLE_NOT_EXIST) { |
| *err_code = common::E_INVALID_ARG; |
| } |
| return *err_code == common::E_OK ? filter : nullptr; |
| } |
| |
| TagFilterHandle tsfile_tag_filter_and(TagFilterHandle left, |
| TagFilterHandle right) { |
| if (!left || !right) return nullptr; |
| try { |
| return storage::TagFilterBuilder::and_filter( |
| static_cast<storage::Filter*>(left), |
| static_cast<storage::Filter*>(right)); |
| } catch (...) { |
| return nullptr; |
| } |
| } |
| |
| TagFilterHandle tsfile_tag_filter_or(TagFilterHandle left, |
| TagFilterHandle right) { |
| if (!left || !right) return nullptr; |
| try { |
| return storage::TagFilterBuilder::or_filter( |
| static_cast<storage::Filter*>(left), |
| static_cast<storage::Filter*>(right)); |
| } catch (...) { |
| return nullptr; |
| } |
| } |
| |
| TagFilterHandle tsfile_tag_filter_not(TagFilterHandle filter) { |
| if (!filter) return nullptr; |
| try { |
| return storage::TagFilterBuilder::not_filter( |
| static_cast<storage::Filter*>(filter)); |
| } catch (...) { |
| return nullptr; |
| } |
| } |
| |
| void tsfile_tag_filter_free(TagFilterHandle filter) { |
| if (filter) { |
| delete static_cast<storage::Filter*>(filter); |
| } |
| } |
| |
| ResultSet tsfile_query_table_with_tag_filter( |
| TsFileReader reader, const char* table_name, char** columns, |
| uint32_t column_num, Timestamp start_time, Timestamp end_time, |
| TagFilterHandle tag_filter, int batch_size, ERRNO* err_code) { |
| auto* r = static_cast<storage::TsFileReader*>(reader); |
| storage::ResultSet* table_result_set = nullptr; |
| std::vector<std::string> column_names; |
| for (uint32_t i = 0; i < column_num; i++) { |
| column_names.emplace_back(columns[i]); |
| } |
| *err_code = r->query(table_name, column_names, start_time, end_time, |
| table_result_set, |
| static_cast<storage::Filter*>(tag_filter), batch_size); |
| return table_result_set; |
| } |
| |
| #ifdef __cplusplus |
| } |
| #endif |
