You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/10/10 11:28:36 UTC
(tsfile) 01/08: fix(cpp): propagate table read failures without silent data loss (#970)
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch rc/2.5.0
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 6e33e434c50768816d1fc64e0897213530a2a6c0
Author: Colin Lee <10...@users.noreply.github.com>
AuthorDate: Sat Oct 10 09:41:12 2026 +0800
fix(cpp): propagate table read failures without silent data loss (#970)
---
cpp/examples/cpp_examples/demo_read.cpp | 4 +-
cpp/src/cwrapper/tsfile_cwrapper.cc | 1082 ++++++++++++--------
cpp/src/cwrapper/tsfile_cwrapper.h | 107 +-
cpp/src/reader/aligned_chunk_reader.cc | 35 +-
.../reader/block/device_ordered_tsblock_reader.cc | 5 +-
cpp/src/reader/chunk_reader.cc | 9 +
cpp/src/reader/device_meta_iterator.cc | 60 +-
cpp/src/reader/device_meta_iterator.h | 11 +-
cpp/src/reader/imeta_data_querier.h | 2 +-
cpp/src/reader/meta_data_querier.cc | 6 +-
cpp/src/reader/meta_data_querier.h | 5 +-
cpp/src/reader/table_query_executor.cc | 25 +-
cpp/src/reader/table_result_set.cc | 20 +
cpp/src/reader/table_result_set.h | 4 +
cpp/src/reader/task/device_task_iterator.cc | 4 +-
cpp/src/reader/task/device_task_iterator.h | 2 +-
cpp/src/reader/tsfile_reader.cc | 144 ++-
cpp/src/reader/tsfile_reader.h | 50 +
cpp/test/cwrapper/c_release_test.cc | 35 +-
cpp/test/cwrapper/cwrapper_test.cc | 42 +
cpp/test/reader/table_read_failure_test.cc | 1005 ++++++++++++++++++
.../reader/table_view/tsfile_reader_table_test.cc | 5 +-
cpp/test/tools/cli_requirements_test.cc | 236 +++++
cpp/tools/cli/run_cli.cc | 41 +-
cpp/tools/commands/cmd_count.cc | 22 +-
cpp/tools/commands/cmd_ls.cc | 28 +-
cpp/tools/commands/cmd_meta.cc | 6 +-
cpp/tools/commands/cmd_schema.cc | 7 +-
cpp/tools/commands/cmd_stats.cc | 64 +-
cpp/tools/commands/commands.h | 10 +-
cpp/tools/commands/row_query.cc | 65 +-
cpp/tools/commands/statistics.cc | 43 +-
cpp/tools/commands/statistics.h | 4 +-
python/tests/test_reader_metadata.py | 44 +
python/tests/test_reader_sources.py | 115 ++-
python/tests/test_tag_filter_query.py | 23 +
python/tsfile/tsfile_cpp.pxd | 5 +
python/tsfile/tsfile_py_cpp.pyx | 40 +-
python/tsfile/tsfile_reader.pyx | 54 +-
39 files changed, 2823 insertions(+), 646 deletions(-)
diff --git a/cpp/examples/cpp_examples/demo_read.cpp b/cpp/examples/cpp_examples/demo_read.cpp
index 91eb7c461..90e24c676 100644
--- a/cpp/examples/cpp_examples/demo_read.cpp
+++ b/cpp/examples/cpp_examples/demo_read.cpp
@@ -18,6 +18,7 @@
*/
#include <iostream>
+#include <memory>
#include <string>
#include <vector>
@@ -40,7 +41,8 @@ int demo_read() {
columns.emplace_back("id2");
columns.emplace_back("s1");
- auto table_schema = reader.get_table_schema(table_name);
+ std::shared_ptr<storage::TableSchema> table_schema;
+ HANDLE_ERROR(reader.get_table_schema(table_name, table_schema));
storage::Filter* tag_filter1 =
storage::TagFilterBuilder(table_schema.get()).eq("id1", "id1_filed_1");
storage::Filter* tag_filter2 =
diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc
index 5cae8f01d..ff70990ca 100644
--- a/cpp/src/cwrapper/tsfile_cwrapper.cc
+++ b/cpp/src/cwrapper/tsfile_cwrapper.cc
@@ -46,6 +46,7 @@
#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)
@@ -1060,81 +1061,141 @@ int tsfile_result_set_metadata_get_column_num(ResultSetMetaData result_set) {
TableSchema tsfile_reader_get_table_schema(TsFileReader reader,
const char* table_name) {
- auto* r = static_cast<storage::TsFileReader*>(reader);
- auto table_shcema = r->get_table_schema(table_name);
- TableSchema ret_schema;
- ret_schema.table_name = strdup(table_shcema->get_table_name().c_str());
- int column_num = table_shcema->get_columns_num();
- ret_schema.column_num = column_num;
- ret_schema.column_schemas =
- static_cast<ColumnSchema*>(malloc(sizeof(ColumnSchema) * column_num));
- for (int i = 0; i < column_num; i++) {
- auto column_schema = table_shcema->get_measurement_schemas()[i];
- ret_schema.column_schemas[i].column_name =
- strdup(column_schema->measurement_name_.c_str());
- ret_schema.column_schemas[i].data_type =
- static_cast<TSDataType>(column_schema->data_type_);
- ret_schema.column_schemas[i].column_category =
- static_cast<ColumnCategory>(
- table_shcema->get_column_categories()[i]);
+ 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 (!src || out_schema == nullptr) {
- return common::E_TABLE_NOT_EXIST;
+ if (out_schema == nullptr) {
+ return common::E_INVALID_ARG;
}
*out_schema = TableSchema{};
- out_schema->table_name = strdup(src->get_table_name().c_str());
- if (out_schema->table_name == nullptr) {
- return common::E_OOM;
+ if (!src) {
+ return common::E_TABLE_NOT_EXIST;
}
- out_schema->column_num = src->get_columns_num();
- if (out_schema->column_num == 0) {
+ 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;
- }
- out_schema->column_schemas = static_cast<ColumnSchema*>(calloc(
- static_cast<size_t>(out_schema->column_num), sizeof(ColumnSchema)));
- if (out_schema->column_schemas == nullptr) {
+ } 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;
}
- const auto& measurements = src->get_measurement_schemas();
- const auto& categories = src->get_column_categories();
- for (int i = 0; i < out_schema->column_num; ++i) {
- 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]);
+}
+
+static void free_table_schema_array(TableSchema* schemas, size_t size) {
+ if (schemas == nullptr) {
+ return;
}
- return common::E_OK;
+ 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 (reader == nullptr || table_name == nullptr || out_schema == nullptr) {
+ if (out_schema == nullptr) {
return common::E_INVALID_ARG;
}
*out_schema = TableSchema{};
+ if (reader == nullptr || table_name == nullptr) {
+ return common::E_INVALID_ARG;
+ }
try {
- auto schema =
+ std::shared_ptr<storage::TableSchema> schema;
+ const int ret =
static_cast<storage::TsFileReader*>(reader)->get_table_schema(
- table_name);
+ 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;
}
}
@@ -1155,66 +1216,50 @@ TableSchema* tsfile_reader_get_all_table_schemas_with_error(TsFileReader reader,
if (error_code == nullptr) {
return nullptr;
}
- *error_code = common::E_INVALID_ARG;
- if (reader == nullptr || size == nullptr) {
- return nullptr;
- }
- auto* r = static_cast<storage::TsFileReader*>(reader);
- std::vector<std::shared_ptr<storage::TableSchema>> table_schemas;
- *error_code = r->get_all_table_schemas(table_schemas);
- if (*error_code != common::E_OK || table_schemas.empty()) {
- return nullptr;
- }
- size_t table_num = table_schemas.size();
- TableSchema* ret =
- static_cast<TableSchema*>(malloc(sizeof(TableSchema) * table_num));
- for (size_t i = 0; i < table_schemas.size(); i++) {
- ret[i].table_name = strdup(table_schemas[i]->get_table_name().c_str());
- int column_num = table_schemas[i]->get_columns_num();
- ret[i].column_num = column_num;
- ret[i].column_schemas = static_cast<ColumnSchema*>(
- malloc(column_num * sizeof(ColumnSchema)));
- auto column_schemas = table_schemas[i]->get_measurement_schemas();
- for (int j = 0; j < column_num; j++) {
- ret[i].column_schemas[j].column_name =
- strdup(column_schemas[j]->measurement_name_.c_str());
- ret[i].column_schemas[j].data_type =
- static_cast<TSDataType>(column_schemas[j]->data_type_);
- ret[i].column_schemas[j].column_category =
- static_cast<ColumnCategory>(
- table_schemas[i]->get_column_categories()[j]);
- }
- }
- *size = table_num;
- return ret;
+ 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;
}
- *out_schemas = nullptr;
- *out_size = 0;
+ TableSchema* copied = nullptr;
+ size_t initialized = 0;
try {
- auto schemas = static_cast<storage::TsFileReader*>(reader)
- ->get_all_table_schemas();
+ 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;
}
- TableSchema* copied = static_cast<TableSchema*>(
+ 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) {
- for (size_t j = 0; j < i; ++j) {
- free_table_schema(copied[j]);
- }
- free(copied);
+ free_table_schema_array(copied, initialized);
return ret;
}
}
@@ -1222,63 +1267,117 @@ ERRNO tsfile_reader_get_all_table_schemas_checked(TsFileReader reader,
*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;
}
}
-DeviceSchema* tsfile_reader_get_all_timeseries_schemas(TsFileReader reader,
- uint32_t* size) {
- auto* r = static_cast<storage::TsFileReader*>(reader);
- auto device_ids = r->get_all_device_ids();
- if (size == nullptr) {
- return nullptr;
+ERRNO tsfile_reader_get_all_timeseries_schemas_checked(
+ TsFileReader reader, DeviceSchema** out_schemas, uint32_t* out_size) {
+ if (out_schemas != nullptr) {
+ *out_schemas = nullptr;
}
- *size = static_cast<uint32_t>(device_ids.size());
- if (device_ids.empty()) {
- return nullptr;
+ if (out_size != nullptr) {
+ *out_size = 0;
}
-
- DeviceSchema* device_schema = static_cast<DeviceSchema*>(
- malloc(sizeof(DeviceSchema) * device_ids.size()));
- if (device_schema == nullptr) {
- *size = 0;
- return nullptr;
+ 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) {
- 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());
- cur_schema.timeseries_num = 0;
- cur_schema.timeseries_schema = nullptr;
+ 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 || schemas.empty()) {
- continue;
- }
+ 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_num = static_cast<int>(schemas.size());
- cur_schema.timeseries_schema = static_cast<TimeseriesSchema*>(
- malloc(sizeof(TimeseriesSchema) * 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());
- 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_);
+ 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;
}
- return device_schema;
+}
+
+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) {
@@ -1648,7 +1747,7 @@ void free_device_timeseries_metadata_entries_partial(
/**
* Copies path, table name, and segment strings from IDeviceID into heap
- * buffers. On failure, frees any partial allocations and returns E_OOM.
+ * 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,
@@ -1658,70 +1757,90 @@ int duplicate_ideviceid_to_device_fields(storage::IDeviceID* id,
*out_table_name = nullptr;
*out_segment_count = 0;
*out_segments = nullptr;
- if (id == nullptr) {
- *out_path = strdup("");
- *out_table_name = strdup("");
- if (*out_path == nullptr || *out_table_name == 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);
- 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;
- }
- 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;
- }
- 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));
- const auto& segs = id->get_segments();
- 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;
+ const int n = id->segment_num();
+ if (n <= 0) {
+ return common::E_OK;
}
- 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);
+ 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;
}
- *out_segment_count = static_cast<uint32_t>(n);
- *out_segments = seg_arr;
- return common::E_OK;
}
int fill_device_id_from_ideviceid(storage::IDeviceID* id, DeviceID* out) {
@@ -1730,176 +1849,154 @@ int fill_device_id_from_ideviceid(storage::IDeviceID* id, DeviceID* out) {
id, &out->path, &out->table_name, &out->segment_count, &out->segments);
}
-void clear_metadata_entry_device_only(DeviceTimeseriesMetadataEntry* e) {
- if (e == nullptr) {
- return;
- }
- tsfile_device_id_free_contents(&e->device);
-}
-
ERRNO populate_c_metadata_map_from_cpp(
storage::DeviceTimeseriesMetadataMap& cpp_map,
DeviceTimeseriesMetadataMap* out_map) {
- 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);
- 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) {
- free_device_timeseries_metadata_entries_partial(entries, di);
- 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;
+ try {
+ if (cpp_map.empty()) {
+ return common::E_OK;
}
- e.timeseries = static_cast<TimeseriesMetadata*>(
- malloc(sizeof(TimeseriesMetadata) * e.timeseries_count));
- if (e.timeseries == nullptr) {
- clear_metadata_entry_device_only(&e);
- free_device_timeseries_metadata_entries_partial(entries, di);
+ 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(e.timeseries, 0,
- sizeof(TimeseriesMetadata) * e.timeseries_count);
- uint32_t slot = 0;
- for (const auto& idx : vec) {
- if (idx == nullptr) {
- continue;
+ 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;
}
- 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) {
- for (uint32_t u = 0; u < slot; u++) {
- free_timeseries_statistic_heap(&e.timeseries[u].statistic);
- free_timeseries_statistic_heap(
- &e.timeseries[u].timeline_statistic);
- free(e.timeseries[u].measurement_name);
+ const auto& vec = kv.second;
+ uint32_t n_ts = 0;
+ for (const auto& idx_nz : vec) {
+ if (idx_nz != nullptr) {
+ n_ts++;
}
- free(e.timeseries);
+ }
+ e.timeseries_count = n_ts;
+ if (e.timeseries_count == 0) {
e.timeseries = nullptr;
- clear_metadata_entry_device_only(&e);
- free_device_timeseries_metadata_entries_partial(entries, di);
+ di++;
+ continue;
+ }
+ e.timeseries = static_cast<TimeseriesMetadata*>(
+ malloc(sizeof(TimeseriesMetadata) * e.timeseries_count));
+ if (e.timeseries == nullptr) {
return common::E_OOM;
}
- 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();
+ memset(e.timeseries, 0,
+ sizeof(TimeseriesMetadata) * e.timeseries_count);
+ uint32_t slot = 0;
+ for (const auto& idx : vec) {
+ if (idx == nullptr) {
+ continue;
}
- 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();
+ 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;
}
- 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();
+#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());
+ 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());
}
- 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);
+ m.chunk_meta_count = chunk_cnt;
+ if (chunk_cnt >= 0 && m.value_metadata_length > 0) {
+ m.locator_flags |= 1;
}
- }
- const int st_rc = fill_timeseries_statistic(st, &m.statistic);
- if (st_rc != common::E_OK) {
- for (uint32_t u = 0; u < slot; u++) {
- free_timeseries_statistic_heap(&e.timeseries[u].statistic);
- free_timeseries_statistic_heap(
- &e.timeseries[u].timeline_statistic);
- free(e.timeseries[u].measurement_name);
+ 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);
+ }
}
- free_timeseries_statistic_heap(&m.statistic);
- free_timeseries_statistic_heap(&m.timeline_statistic);
- free(m.measurement_name);
- free(e.timeseries);
- e.timeseries = nullptr;
- clear_metadata_entry_device_only(&e);
- free_device_timeseries_metadata_entries_partial(entries, di);
- return st_rc;
- }
- const int timeline_st_rc =
- fill_timeline_statistic(idx.get(), &m.timeline_statistic);
- if (timeline_st_rc != common::E_OK) {
- for (uint32_t u = 0; u < slot; u++) {
- free_timeseries_statistic_heap(&e.timeseries[u].statistic);
- free_timeseries_statistic_heap(
- &e.timeseries[u].timeline_statistic);
- free(e.timeseries[u].measurement_name);
+ const int st_rc = fill_timeseries_statistic(st, &m.statistic);
+ if (st_rc != common::E_OK) {
+ return st_rc;
}
- free_timeseries_statistic_heap(&m.statistic);
- free_timeseries_statistic_heap(&m.timeline_statistic);
- free(m.measurement_name);
- free(e.timeseries);
- e.timeseries = nullptr;
- clear_metadata_entry_device_only(&e);
- free_device_timeseries_metadata_entries_partial(entries, di);
- return timeline_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++;
}
- slot++;
+ di++;
}
- 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;
}
- out_map->entries = entries;
- out_map->device_count = dev_n;
- return common::E_OK;
}
} // namespace
@@ -1949,24 +2046,36 @@ ERRNO tsfile_reader_get_all_devices(TsFileReader reader, DeviceID** out_devices,
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;
}
- out_map->entries = nullptr;
- out_map->device_count = 0;
- auto* r = static_cast<storage::TsFileReader*>(reader);
- storage::DeviceTimeseriesMetadataMap cpp_map = r->get_timeseries_metadata();
- return populate_c_metadata_map_from_cpp(cpp_map, out_map);
+ 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;
}
- out_map->entries = nullptr;
- out_map->device_count = 0;
if (length == 0) {
return common::E_OK;
}
@@ -1974,20 +2083,48 @@ ERRNO tsfile_reader_get_timeseries_metadata_for_devices(
return common::E_INVALID_ARG;
}
for (uint32_t i = 0; i < length; i++) {
- if (devices[i].path == nullptr) {
+ 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;
}
}
- 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++) {
- query_ids.push_back(std::make_shared<storage::StringArrayDeviceID>(
- std::string(devices[i].path)));
+ 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;
}
- storage::DeviceTimeseriesMetadataMap cpp_map =
- r->get_timeseries_metadata(query_ids);
- return populate_c_metadata_map_from_cpp(cpp_map, out_map);
}
void tsfile_free_device_timeseries_metadata_map(
@@ -2508,37 +2645,93 @@ ResultSet _tsfile_reader_query_device(TsFileReader reader,
// ============== Tag Filter API Implementation ==============
-// Helper macro to avoid repetition in tag filter factory functions.
-// The shared_ptr must stay alive while TagFilterBuilder accesses the schema.
-// Every C-API entry must validate its pointers: a null reader would deref
-// during the static_cast, and null table/column/value would feed std::string
-// a null pointer (UB / crash).
-// The function-name suffix and the TagFilterBuilder method are always the same
-// operator, so the macro takes a single argument used for both.
-#define DEFINE_TAG_FILTER_FACTORY(op) \
- TagFilterHandle tsfile_tag_filter_##op( \
+#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) { \
- if (reader == nullptr || table_name == nullptr || \
- column_name == nullptr || value == nullptr) { \
- return nullptr; \
- } \
- auto* r = static_cast<storage::TsFileReader*>(reader); \
- auto schema = r->get_table_schema(table_name); \
- if (!schema) return nullptr; \
- storage::TagFilterBuilder builder(schema.get()); \
- return builder.op(column_name, value); \
- }
-
-DEFINE_TAG_FILTER_FACTORY(eq)
-DEFINE_TAG_FILTER_FACTORY(neq)
-DEFINE_TAG_FILTER_FACTORY(lt)
-DEFINE_TAG_FILTER_FACTORY(lteq)
-DEFINE_TAG_FILTER_FACTORY(gt)
-DEFINE_TAG_FILTER_FACTORY(gteq)
+ 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,
@@ -2547,60 +2740,49 @@ TagFilterHandle tsfile_tag_filter_create(TsFileReader reader,
if (err_code == nullptr) {
return nullptr;
}
- if (reader == nullptr || table_name == nullptr || column_name == nullptr ||
- value == 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 nullptr;
}
- auto* r = static_cast<storage::TsFileReader*>(reader);
- auto schema = r->get_table_schema(table_name);
- if (!schema) {
- *err_code = common::E_INVALID_ARG;
- return nullptr;
+ 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;
}
- 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:
- *err_code = common::E_INVALID_ARG;
- return nullptr;
+ if (reader == nullptr || table_name == nullptr || column_name == nullptr ||
+ lower == nullptr || upper == nullptr || out_filter == nullptr) {
+ return common::E_INVALID_ARG;
}
- if (filter == nullptr) {
- *err_code = common::E_COLUMN_NOT_EXIST;
- return nullptr;
+ 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;
}
- *err_code = common::E_OK;
- return static_cast<void*>(filter);
}
TagFilterHandle tsfile_tag_filter_between(TsFileReader reader,
@@ -2611,49 +2793,47 @@ TagFilterHandle tsfile_tag_filter_between(TsFileReader reader,
if (err_code == nullptr) {
return nullptr;
}
- if (reader == nullptr || table_name == nullptr || column_name == nullptr ||
- lower == nullptr || upper == 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 nullptr;
}
- auto* r = static_cast<storage::TsFileReader*>(reader);
- auto schema = r->get_table_schema(table_name);
- if (!schema) {
- *err_code = common::E_INVALID_ARG;
- return nullptr;
- }
- 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) {
- *err_code = common::E_COLUMN_NOT_EXIST;
- return nullptr;
- }
- *err_code = common::E_OK;
- return static_cast<void*>(filter);
+ return *err_code == common::E_OK ? filter : nullptr;
}
TagFilterHandle tsfile_tag_filter_and(TagFilterHandle left,
TagFilterHandle right) {
if (!left || !right) return nullptr;
- return storage::TagFilterBuilder::and_filter(
- static_cast<storage::Filter*>(left),
- static_cast<storage::Filter*>(right));
+ 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;
- return storage::TagFilterBuilder::or_filter(
- static_cast<storage::Filter*>(left),
- static_cast<storage::Filter*>(right));
+ 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;
- return storage::TagFilterBuilder::not_filter(
- static_cast<storage::Filter*>(filter));
+ try {
+ return storage::TagFilterBuilder::not_filter(
+ static_cast<storage::Filter*>(filter));
+ } catch (...) {
+ return nullptr;
+ }
}
void tsfile_tag_filter_free(TagFilterHandle filter) {
diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h
index 003e59198..4a87db90c 100644
--- a/cpp/src/cwrapper/tsfile_cwrapper.h
+++ b/cpp/src/cwrapper/tsfile_cwrapper.h
@@ -552,7 +552,9 @@ ERRNO tsfile_reader_get_timeseries_metadata_all(
*
* @param devices NULL and length>0 is E_INVALID_ARG. length==0: empty result
* (E_OK); @p devices is not read.
- * For each entry, @p path must be non-NULL (canonical device path).
+ * Nonempty @p segments are used as the exact device ID, preserving null tags.
+ * The segment array and its first (table name) entry must be non-NULL.
+ * With segment_count==0, @p path must be a non-NULL legacy tree device path.
*/
ERRNO tsfile_reader_get_timeseries_metadata_for_devices(
TsFileReader reader, const DeviceID* devices, uint32_t length,
@@ -1017,54 +1019,81 @@ int tsfile_result_set_metadata_get_column_num(ResultSetMetaData result_set);
// const char* device_id);
/**
- * @brief Gets specific table's schema in the tsfile.
- *
- * @return TableSchema, contains table and column info.
- * @note Caller should call free_table_schema to free the tableschema.
+ * @brief Gets a table schema using the legacy value-returning API.
+ * @return A populated schema, or a zero-initialized schema when the lookup
+ * fails. Use tsfile_reader_get_table_schema_checked() when the error code is
+ * required.
+ * @note Release the returned schema's contents with free_table_schema().
*/
TableSchema tsfile_reader_get_table_schema(TsFileReader reader,
const char* table_name);
-/** Retrieves one table schema and reports missing tables through ERRNO. */
+/**
+ * @brief Gets one table schema and returns the error code directly.
+ * @param out_schema Required output storage with no owned allocations.
+ * It is zero-initialized on failure; partial results are released internally.
+ * @return RET_OK on success, or the lookup, read, or allocation error.
+ * @note On success, release the contents with free_table_schema(*out_schema).
+ * The caller owns the output storage itself.
+ */
ERRNO tsfile_reader_get_table_schema_checked(TsFileReader reader,
const char* table_name,
TableSchema* out_schema);
+
/**
- * @brief Gets all table schema in the tsfile.
- *
- * @return TableSchema, contains table and column info.
- * @note Caller should call free_table_schema and free to free the ptr.
- * @note Use tsfile_reader_get_all_table_schemas_with_error to distinguish
- * metadata read failures from an empty schema list.
+ * @brief Gets all table schemas using the legacy pointer-returning API.
+ * @return Schema array, or NULL when there are no tables or an error occurs.
+ * Use tsfile_reader_get_all_table_schemas_checked() when the error code is
+ * required.
+ * @note size is zero on failure. Release each schema with free_table_schema(),
+ * then release the array with free().
*/
TableSchema* tsfile_reader_get_all_table_schemas(TsFileReader reader,
uint32_t* size);
-/** Retrieves every table schema into a caller-freed array. */
+/**
+ * @brief Gets all table schemas and returns the error code directly.
+ * Both output parameters are required. Non-null outputs are set to NULL/0
+ * before work begins and remain empty on failure. An empty file result is
+ * successful with RET_OK and NULL/0 outputs.
+ * @note On success, release each schema with free_table_schema(), then free()
+ * the array. Partial results are released internally on failure.
+ */
ERRNO tsfile_reader_get_all_table_schemas_checked(TsFileReader reader,
TableSchema** out_schemas,
uint32_t* out_size);
/**
- * @brief Gets all table schemas and reports metadata read failures.
- * @return Schema array, or NULL when there are no tables or on error. Check
- * error_code to distinguish these cases; size is zero on error.
- * @note Caller must free each schema with free_table_schema, then free the
- * array.
+ * @brief Compatibility API reporting errors through a required error_code.
+ * Returns NULL for an empty result or an error; error_code distinguishes them.
+ * size is zero on failure. Ownership is the same as the checked array API.
*/
TableSchema* tsfile_reader_get_all_table_schemas_with_error(TsFileReader reader,
uint32_t* size,
ERRNO* error_code);
/**
- * @brief Gets all timeseries schema in the tsfile.
- *
- * @return DeviceSchema list, contains timeseries info.
- * @note Caller should call free_device_schema and free to free the ptr.
+ * @brief Gets all timeseries schemas using the legacy pointer-returning API.
+ * Use tsfile_reader_get_all_timeseries_schemas_checked() when the error code
+ * is required.
+ * @return Array on success, or NULL for an empty result or an error.
+ * @note size is zero on failure. Release each schema with free_device_schema(),
+ * then release the array with free().
*/
DeviceSchema* tsfile_reader_get_all_timeseries_schemas(TsFileReader reader,
uint32_t* size);
+/**
+ * @brief Gets all timeseries schemas and returns the error code directly.
+ * Both output parameters are required. Non-null outputs are set to NULL/0
+ * before work begins and remain empty on failure. No devices is a successful
+ * result with RET_OK and NULL/0 outputs.
+ * @note On success, release each schema with free_device_schema(), then free()
+ * the array. Partial results are released internally on failure.
+ */
+ERRNO tsfile_reader_get_all_timeseries_schemas_checked(
+ TsFileReader reader, DeviceSchema** out_schemas, uint32_t* out_size);
+
// ---------- Tag Filter API ----------
/**
@@ -1111,6 +1140,29 @@ TagFilterHandle tsfile_tag_filter_between(TsFileReader reader,
const char* lower, const char* upper,
bool is_not, ERRNO* err_code);
+/**
+ * Create a tag filter and return the error code directly. out_filter is
+ * required and is set to NULL on failure. Unlike the legacy create API,
+ * a missing table is reported as RET_TABLE_NOT_EXIST, not RET_INVALID_ARG.
+ * Release a successful result with tsfile_tag_filter_free().
+ */
+ERRNO tsfile_tag_filter_create_checked(TsFileReader reader,
+ const char* table_name,
+ const char* column_name,
+ const char* value, TagFilterOp op,
+ TagFilterHandle* out_filter);
+
+/**
+ * Create a BETWEEN tag filter and return the error code directly.
+ * Output ownership and missing-table handling match the checked create API.
+ */
+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);
+
/**
* @brief Create a tag equality filter: column == value.
*
@@ -1119,6 +1171,8 @@ TagFilterHandle tsfile_tag_filter_between(TsFileReader reader,
* @param column_name [in] Tag column name.
* @param value [in] Value to compare against.
* @return TagFilterHandle on success, NULL on failure.
+ * @note Retained for backward compatibility. New code should use the checked
+ * create API and check its error code. Free with tsfile_tag_filter_free().
*/
TagFilterHandle tsfile_tag_filter_eq(TsFileReader reader,
const char* table_name,
@@ -1151,19 +1205,22 @@ TagFilterHandle tsfile_tag_filter_gteq(TsFileReader reader,
const char* value);
/**
- * @brief Logical AND of two tag filters. Takes ownership of left and right.
+ * @brief Logical AND of two tag filters. Takes ownership of left and right
+ * only on success. Returns NULL on failure; the caller still owns the inputs.
*/
TagFilterHandle tsfile_tag_filter_and(TagFilterHandle left,
TagFilterHandle right);
/**
- * @brief Logical OR of two tag filters. Takes ownership of left and right.
+ * @brief Logical OR of two tag filters. Takes ownership of left and right
+ * only on success. Returns NULL on failure; the caller still owns the inputs.
*/
TagFilterHandle tsfile_tag_filter_or(TagFilterHandle left,
TagFilterHandle right);
/**
- * @brief Logical NOT of a tag filter. Takes ownership of filter.
+ * @brief Logical NOT of a tag filter. Takes ownership of filter only on
+ * success. Returns NULL on failure; the caller still owns the input.
*/
TagFilterHandle tsfile_tag_filter_not(TagFilterHandle filter);
diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc
index 1c5b838d7..059570265 100644
--- a/cpp/src/reader/aligned_chunk_reader.cc
+++ b/cpp/src/reader/aligned_chunk_reader.cc
@@ -259,6 +259,13 @@ int AlignedChunkReader::load_by_aligned_meta(ChunkMeta* time_chunk_meta,
ret = read_file_->read(time_chunk_meta_->offset_of_chunk_header_,
time_file_data_buf, file_data_time_buf_size_,
ret_read_len);
+ if (IS_SUCC(ret) &&
+ ret_read_len <
+ UTIL_MIN(static_cast<int64_t>(file_data_time_buf_size_),
+ read_file_->file_size() -
+ time_chunk_meta_->offset_of_chunk_header_)) {
+ ret = E_FILE_READ_ERR;
+ }
if (!IS_SUCC(ret)) {
mem_free(time_file_data_buf);
return ret;
@@ -286,6 +293,13 @@ int AlignedChunkReader::load_by_aligned_meta(ChunkMeta* time_chunk_meta,
ret = read_file_->read(value_chunk_meta_->offset_of_chunk_header_,
value_file_data_buf, file_data_value_buf_size_,
ret_read_len);
+ if (IS_SUCC(ret) &&
+ ret_read_len <
+ UTIL_MIN(static_cast<int64_t>(file_data_value_buf_size_),
+ read_file_->file_size() -
+ value_chunk_meta_->offset_of_chunk_header_)) {
+ ret = E_FILE_READ_ERR;
+ }
if (!IS_SUCC(ret)) {
mem_free(value_file_data_buf);
return ret;
@@ -478,6 +492,9 @@ int AlignedChunkReader::read_from_file_and_rewrap(
int ret_read_len = 0;
if (RET_FAIL(
read_file_->read(offset, file_data_buf, read_size, ret_read_len))) {
+ } else if (ret_read_len < UTIL_MIN(static_cast<int64_t>(read_size),
+ read_file_->file_size() - offset)) {
+ ret = E_FILE_READ_ERR;
} else {
in_stream_.wrap_from(file_data_buf, ret_read_len);
#ifdef DEBUG_SE
@@ -1216,6 +1233,13 @@ int AlignedChunkReader::load_by_aligned_meta_multi(
ret = read_file_->read(time_chunk_meta_->offset_of_chunk_header_,
time_file_data_buf, file_data_time_buf_size_,
ret_read_len);
+ if (IS_SUCC(ret) &&
+ ret_read_len <
+ UTIL_MIN(static_cast<int64_t>(file_data_time_buf_size_),
+ read_file_->file_size() -
+ time_chunk_meta_->offset_of_chunk_header_)) {
+ ret = E_FILE_READ_ERR;
+ }
if (!IS_SUCC(ret)) {
mem_free(time_file_data_buf);
return ret;
@@ -1264,6 +1288,13 @@ int AlignedChunkReader::load_by_aligned_meta_multi(
ret = read_file_->read(col->chunk_meta->offset_of_chunk_header_, vbuf,
col->file_data_buf_size, ret_read_len);
+ if (IS_SUCC(ret) &&
+ ret_read_len <
+ UTIL_MIN(static_cast<int64_t>(col->file_data_buf_size),
+ read_file_->file_size() -
+ col->chunk_meta->offset_of_chunk_header_)) {
+ ret = E_FILE_READ_ERR;
+ }
if (!IS_SUCC(ret)) {
mem_free(vbuf);
return ret;
@@ -1390,7 +1421,7 @@ int AlignedChunkReader::decode_time_page_with(const ChunkPageInfo& page_info,
// bytes would feed garbage to the decompressor.
if (read_len != static_cast<int32_t>(page_info.time_compressed_size)) {
if (heap) common::mem_free(compressed_buf);
- return E_TSFILE_CORRUPTED;
+ return E_FILE_READ_ERR;
}
char* uncompressed_buf = nullptr;
@@ -1637,7 +1668,7 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx,
if (read_len !=
static_cast<int32_t>(page_info.value_compressed_sizes[col_idx])) {
if (heap) common::mem_free(compressed_buf);
- return E_TSFILE_CORRUPTED;
+ return E_FILE_READ_ERR;
}
char* uncompressed_buf = nullptr;
diff --git a/cpp/src/reader/block/device_ordered_tsblock_reader.cc b/cpp/src/reader/block/device_ordered_tsblock_reader.cc
index 38cb63f7e..5c30d9bc6 100644
--- a/cpp/src/reader/block/device_ordered_tsblock_reader.cc
+++ b/cpp/src/reader/block/device_ordered_tsblock_reader.cc
@@ -51,7 +51,10 @@ int DeviceOrderedTsBlockReader::has_next(bool& has_next) {
has_next = false;
return common::E_OK;
}
- if (!device_task_iterator_->has_next()) {
+ if (RET_FAIL(device_task_iterator_->has_next(has_next))) {
+ return ret;
+ }
+ if (!has_next) {
break;
}
DeviceQueryTask* task = nullptr;
diff --git a/cpp/src/reader/chunk_reader.cc b/cpp/src/reader/chunk_reader.cc
index 271b5c206..abeff8375 100644
--- a/cpp/src/reader/chunk_reader.cc
+++ b/cpp/src/reader/chunk_reader.cc
@@ -114,6 +114,12 @@ int ChunkReader::load_by_meta(ChunkMeta* meta) {
}
ret = read_file_->read(chunk_meta_->offset_of_chunk_header_, file_data_buf,
file_data_buf_size_, ret_read_len);
+ if (IS_SUCC(ret) &&
+ ret_read_len < UTIL_MIN(static_cast<int64_t>(file_data_buf_size_),
+ read_file_->file_size() -
+ chunk_meta_->offset_of_chunk_header_)) {
+ ret = E_FILE_READ_ERR;
+ }
if (!IS_SUCC(ret)) {
mem_free(file_data_buf);
return ret;
@@ -275,6 +281,9 @@ int ChunkReader::read_from_file_and_rewrap(int want_size) {
int ret_read_len = 0;
if (RET_FAIL(
read_file_->read(offset, file_data_buf, read_size, ret_read_len))) {
+ } else if (ret_read_len < UTIL_MIN(static_cast<int64_t>(read_size),
+ read_file_->file_size() - offset)) {
+ ret = E_FILE_READ_ERR;
} else {
in_stream_.wrap_from(file_data_buf, ret_read_len);
// DEBUG_hex_dump_buf("wrapped buf = ", file_data_buf, 256);
diff --git a/cpp/src/reader/device_meta_iterator.cc b/cpp/src/reader/device_meta_iterator.cc
index 6edb413eb..e57111e32 100644
--- a/cpp/src/reader/device_meta_iterator.cc
+++ b/cpp/src/reader/device_meta_iterator.cc
@@ -35,33 +35,55 @@ void DeviceMetaIterator::destroy_remaining_cached_devices() {
DeviceMetaIterator::~DeviceMetaIterator() {
destroy_remaining_cached_devices();
+ while (!meta_index_nodes_.empty()) {
+ auto pending = meta_index_nodes_.front();
+ meta_index_nodes_.pop();
+ if (pending.second) {
+ pending.first->~MetaIndexNode();
+ }
+ }
pa_.destroy();
}
-bool DeviceMetaIterator::has_next() {
+int DeviceMetaIterator::has_next(bool& has_next) {
+ int ret = common::E_OK;
+ has_next = false;
+ if (read_error_ != common::E_OK) {
+ return read_error_;
+ }
if (!result_cache_.empty()) {
- return true;
+ has_next = true;
+ return common::E_OK;
}
if (direct_device_id_ != nullptr) {
if (direct_lookup_done_) {
- return false;
+ return common::E_OK;
}
- if (load_results_direct() != common::E_OK) {
- return false;
+ if (RET_FAIL(load_results_direct())) {
+ read_error_ = ret;
+ return ret;
}
- return !result_cache_.empty();
+ has_next = !result_cache_.empty();
+ return ret;
}
- if (load_results() != common::E_OK) {
- return false;
+ if (RET_FAIL(load_results())) {
+ read_error_ = ret;
+ return ret;
}
- return !result_cache_.empty();
+ has_next = !result_cache_.empty();
+ return ret;
}
int DeviceMetaIterator::next(
std::pair<std::shared_ptr<IDeviceID>, MetaIndexNode*>& ret_meta) {
- if (!has_next()) {
+ int ret = common::E_OK;
+ bool available = false;
+ if (RET_FAIL(has_next(available))) {
+ return ret;
+ }
+ if (!available) {
return common::E_NO_MORE_DATA;
}
@@ -71,21 +93,25 @@ int DeviceMetaIterator::next(
}
int DeviceMetaIterator::load_results() {
- int root_num = meta_index_nodes_.size();
while (!meta_index_nodes_.empty()) {
- auto meta_data_index_node = meta_index_nodes_.front();
+ auto pending = meta_index_nodes_.front();
meta_index_nodes_.pop();
+ auto meta_data_index_node = pending.first;
const auto& node_type = meta_data_index_node->node_type_;
+ int ret = common::E_OK;
if (node_type == MetaIndexNodeType::LEAF_DEVICE) {
- load_leaf_device(meta_data_index_node);
+ ret = load_leaf_device(meta_data_index_node);
} else if (node_type == MetaIndexNodeType::INTERNAL_DEVICE) {
- load_internal_node(meta_data_index_node);
+ ret = load_internal_node(meta_data_index_node);
} else {
- return common::E_INVALID_NODE_TYPE;
+ ret = common::E_INVALID_NODE_TYPE;
}
- if (root_num-- <= 0) {
+ if (pending.second) {
meta_data_index_node->~MetaIndexNode();
}
+ if (ret != common::E_OK) {
+ return ret;
+ }
}
return common::E_OK;
}
@@ -136,7 +162,7 @@ int DeviceMetaIterator::load_internal_node(MetaIndexNode* meta_index_node) {
start_offset, end_offset, pa_, child_node, false))) {
return ret;
} else {
- meta_index_nodes_.push(child_node);
+ meta_index_nodes_.push({child_node, true});
}
}
return ret;
diff --git a/cpp/src/reader/device_meta_iterator.h b/cpp/src/reader/device_meta_iterator.h
index 9f42819d3..00d5a8b86 100644
--- a/cpp/src/reader/device_meta_iterator.h
+++ b/cpp/src/reader/device_meta_iterator.h
@@ -41,7 +41,7 @@ class DeviceMetaIterator {
// A valid schema-only table has no device index. Treat a null root as
// an empty iterator instead of dereferencing it during has_next().
if (meat_index_node != nullptr) {
- meta_index_nodes_.push(meat_index_node);
+ meta_index_nodes_.push({meat_index_node, false});
}
pa_.init(512, common::MOD_DEVICE_META_ITER);
try_setup_direct_lookup(meat_index_node);
@@ -54,7 +54,7 @@ class DeviceMetaIterator {
id_filter_(id_filter),
direct_lookup_done_(false) {
for (auto meta_index_node : meta_index_node_list) {
- meta_index_nodes_.push(meta_index_node);
+ meta_index_nodes_.push({meta_index_node, false});
}
should_split_device_name = true;
pa_.init(512, common::MOD_DEVICE_META_ITER);
@@ -64,7 +64,7 @@ class DeviceMetaIterator {
void destroy_remaining_cached_devices();
- bool has_next();
+ int has_next(bool& has_next);
int next(std::pair<std::shared_ptr<IDeviceID>, MetaIndexNode*>& ret_meta);
@@ -77,13 +77,16 @@ class DeviceMetaIterator {
int load_results_direct();
TsFileIOReader* io_reader_;
- std::queue<MetaIndexNode*> meta_index_nodes_;
+ // The bool tracks ownership: roots borrowed from file metadata are false;
+ // descendants allocated in pa_ are true and need explicit destruction.
+ std::queue<std::pair<MetaIndexNode*, bool>> meta_index_nodes_;
std::queue<std::pair<std::shared_ptr<IDeviceID>, MetaIndexNode*>>
result_cache_;
const Filter* id_filter_;
common::PageArena pa_;
bool should_split_device_name;
+ int read_error_ = common::E_OK;
bool direct_lookup_done_;
std::shared_ptr<IDeviceID> direct_device_id_;
MetaIndexNode* direct_root_node_ = nullptr;
diff --git a/cpp/src/reader/imeta_data_querier.h b/cpp/src/reader/imeta_data_querier.h
index c034f9151..d7b5d7deb 100644
--- a/cpp/src/reader/imeta_data_querier.h
+++ b/cpp/src/reader/imeta_data_querier.h
@@ -43,7 +43,7 @@ class IMetadataQuerier {
virtual std::map<Path, std::vector<std::shared_ptr<ChunkMeta>>>
get_chunk_metadata_map(const std::vector<Path>& paths) const = 0;
- virtual int get_whole_file_metadata(TsFileMeta* tsfile_meta) const = 0;
+ virtual int get_whole_file_metadata(TsFileMeta*& tsfile_meta) const = 0;
virtual void load_chunk_metadatas(const std::vector<Path>& paths) = 0;
diff --git a/cpp/src/reader/meta_data_querier.cc b/cpp/src/reader/meta_data_querier.cc
index 0accbdde9..31f3698d1 100644
--- a/cpp/src/reader/meta_data_querier.cc
+++ b/cpp/src/reader/meta_data_querier.cc
@@ -25,7 +25,6 @@ namespace storage {
MetadataQuerier::MetadataQuerier(TsFileIOReader* tsfile_io_reader)
: io_reader_(tsfile_io_reader) {
- file_metadata_ = io_reader_->get_tsfile_meta();
device_chunk_meta_cache_ = std::unique_ptr<
common::Cache<std::string /*ToDO: Device ID*/,
std::vector<std::shared_ptr<ChunkMeta>>, std::mutex>>(
@@ -68,9 +67,8 @@ MetadataQuerier::get_chunk_metadata_map(const std::vector<Path>& paths) const {
return {};
}
-int MetadataQuerier::get_whole_file_metadata(TsFileMeta* tsfile_meta) const {
- tsfile_meta = io_reader_->get_tsfile_meta();
- return common::E_OK;
+int MetadataQuerier::get_whole_file_metadata(TsFileMeta*& tsfile_meta) const {
+ return io_reader_->get_tsfile_meta(tsfile_meta);
}
void MetadataQuerier::load_chunk_metadatas(const std::vector<Path>& paths) {
diff --git a/cpp/src/reader/meta_data_querier.h b/cpp/src/reader/meta_data_querier.h
index be575323f..70dc7121e 100644
--- a/cpp/src/reader/meta_data_querier.h
+++ b/cpp/src/reader/meta_data_querier.h
@@ -50,7 +50,7 @@ class MetadataQuerier : public IMetadataQuerier {
std::map<Path, std::vector<std::shared_ptr<ChunkMeta>>>
get_chunk_metadata_map(const std::vector<Path>& paths) const override;
- int get_whole_file_metadata(TsFileMeta* tsfile_meta) const override;
+ int get_whole_file_metadata(TsFileMeta*& tsfile_meta) const override;
void load_chunk_metadatas(const std::vector<Path>& paths) override;
@@ -70,7 +70,6 @@ class MetadataQuerier : public IMetadataQuerier {
private:
TsFileIOReader* io_reader_;
- TsFileMeta* file_metadata_;
std::unique_ptr<
common::Cache<std::string, /*Todo std::pair<IDeviceID, std::string>*/
std::vector<std::shared_ptr<ChunkMeta>>, std::mutex>>
@@ -84,4 +83,4 @@ class MetadataQuerier : public IMetadataQuerier {
};
} // end namespace storage
-#endif // READER_META_DATA_QUERIER_H
\ No newline at end of file
+#endif // READER_META_DATA_QUERIER_H
diff --git a/cpp/src/reader/table_query_executor.cc b/cpp/src/reader/table_query_executor.cc
index ec3807b31..5d708baae 100644
--- a/cpp/src/reader/table_query_executor.cc
+++ b/cpp/src/reader/table_query_executor.cc
@@ -28,7 +28,11 @@ int TableQueryExecutor::query(const std::string& table_name,
Filter* field_filter, ResultSet*& ret_qds) {
int ret = common::E_OK;
TsFileMeta* file_metadata = nullptr;
- file_metadata = tsfile_io_reader_->get_tsfile_meta();
+ ret_qds = nullptr;
+ if (RET_FAIL(tsfile_io_reader_->get_tsfile_meta(file_metadata))) {
+ delete time_filter;
+ return ret;
+ }
common::PageArena pa;
pa.init(512, common::MOD_TSFILE_READER);
MetaIndexNode* table_root = nullptr;
@@ -98,7 +102,11 @@ int TableQueryExecutor::query(const std::string& table_name,
ResultSet*& ret_qds) {
int ret = common::E_OK;
TsFileMeta* file_metadata = nullptr;
- file_metadata = tsfile_io_reader_->get_tsfile_meta();
+ ret_qds = nullptr;
+ if (RET_FAIL(tsfile_io_reader_->get_tsfile_meta(file_metadata))) {
+ delete time_filter;
+ return ret;
+ }
common::PageArena pa;
pa.init(512, common::MOD_TSFILE_READER);
MetaIndexNode* table_root = nullptr;
@@ -165,13 +173,20 @@ int TableQueryExecutor::query_on_tree(
common::PageArena pa;
pa.init(512, common::MOD_TSFILE_READER);
int ret = common::E_OK;
- TsFileMeta* file_meta = tsfile_io_reader_->get_tsfile_meta();
+ ret_qds = nullptr;
+ TsFileMeta* file_meta = nullptr;
+ if (RET_FAIL(tsfile_io_reader_->get_tsfile_meta(file_meta))) {
+ delete time_filter;
+ return ret;
+ }
std::unordered_set<MetaIndexNode*> table_inodes;
for (auto const& device : devices) {
- MetaIndexNode* table_inode;
+ MetaIndexNode* table_inode = nullptr;
if (RET_FAIL(file_meta->get_table_metaindex_node(
device->get_table_name(), table_inode))) {
- };
+ delete time_filter;
+ return ret;
+ }
table_inodes.insert(table_inode);
}
diff --git a/cpp/src/reader/table_result_set.cc b/cpp/src/reader/table_result_set.cc
index 1a8d2a687..ce54c98f8 100644
--- a/cpp/src/reader/table_result_set.cc
+++ b/cpp/src/reader/table_result_set.cc
@@ -39,6 +39,21 @@ void TableResultSet::init() {
TableResultSet::~TableResultSet() { close(); }
int TableResultSet::next(bool& has_next) {
+ int ret = common::E_OK;
+ has_next = false;
+ if (read_error_ != common::E_OK) {
+ return read_error_;
+ }
+ if (RET_FAIL(next_internal(has_next))) {
+ read_error_ = ret;
+ has_next = false;
+ row_ready_ = false;
+ row_materialized_ = false;
+ }
+ return ret;
+}
+
+int TableResultSet::next_internal(bool& has_next) {
if (return_mode_ != RETURN_ROW) {
return tsblock_reader_->has_next(has_next);
}
@@ -177,6 +192,9 @@ std::shared_ptr<ResultSetMetadata> TableResultSet::get_metadata() {
int TableResultSet::get_next_tsblock(common::TsBlock*& block) {
int ret = common::E_OK;
block = nullptr;
+ if (read_error_ != common::E_OK) {
+ return read_error_;
+ }
if (return_mode_ == RETURN_ROW) {
return common::E_INVALID_ARG;
@@ -184,6 +202,7 @@ int TableResultSet::get_next_tsblock(common::TsBlock*& block) {
bool has_next = false;
if (RET_FAIL(tsblock_reader_->has_next(has_next))) {
+ read_error_ = ret;
return ret;
}
@@ -192,6 +211,7 @@ int TableResultSet::get_next_tsblock(common::TsBlock*& block) {
}
if (RET_FAIL(tsblock_reader_->next(tsblock_))) {
+ read_error_ = ret;
return ret;
}
diff --git a/cpp/src/reader/table_result_set.h b/cpp/src/reader/table_result_set.h
index d92072934..b42b69c3f 100644
--- a/cpp/src/reader/table_result_set.h
+++ b/cpp/src/reader/table_result_set.h
@@ -61,6 +61,7 @@ class TableResultSet : public ResultSet {
private:
void init();
+ int next_internal(bool& has_next);
// Lazy materialization: fill row_record_ from the current row when a
// caller actually requests the RowRecord (or a non-fast accessor).
void materialize_current_row();
@@ -74,6 +75,9 @@ class TableResultSet : public ResultSet {
std::vector<common::TSDataType> data_types_;
const int return_mode_;
bool closed_ = false;
+ // A failed read may have advanced device/page state. Never resume it as
+ // EOF.
+ int read_error_ = common::E_OK;
// True when row_iterator_ points at a row that hasn't been consumed yet.
bool row_ready_ = false;
// True when row_record_ has been populated for the current row.
diff --git a/cpp/src/reader/task/device_task_iterator.cc b/cpp/src/reader/task/device_task_iterator.cc
index e22fefb06..5b20f1715 100644
--- a/cpp/src/reader/task/device_task_iterator.cc
+++ b/cpp/src/reader/task/device_task_iterator.cc
@@ -25,8 +25,8 @@ void DeviceTaskIterator::flush_remaining_device_meta_cache() {
device_meta_iterator_->destroy_remaining_cached_devices();
}
-bool DeviceTaskIterator::has_next() const {
- return device_meta_iterator_->has_next();
+int DeviceTaskIterator::has_next(bool& has_next) const {
+ return device_meta_iterator_->has_next(has_next);
}
int DeviceTaskIterator::next(DeviceQueryTask*& task) {
diff --git a/cpp/src/reader/task/device_task_iterator.h b/cpp/src/reader/task/device_task_iterator.h
index cc5a75562..ad86dbc1d 100644
--- a/cpp/src/reader/task/device_task_iterator.h
+++ b/cpp/src/reader/task/device_task_iterator.h
@@ -72,7 +72,7 @@ class DeviceTaskIterator {
void flush_remaining_device_meta_cache();
- bool has_next() const;
+ int has_next(bool& has_next) const;
int next(DeviceQueryTask*& task);
diff --git a/cpp/src/reader/tsfile_reader.cc b/cpp/src/reader/tsfile_reader.cc
index f860a06e5..03755e4c7 100644
--- a/cpp/src/reader/tsfile_reader.cc
+++ b/cpp/src/reader/tsfile_reader.cc
@@ -427,19 +427,38 @@ void TsFileReader::destroy_query_data_set(storage::ResultSet* qds) {
std::vector<std::shared_ptr<IDeviceID>> TsFileReader::get_all_devices(
std::string table_name) {
- TsFileMeta* tsfile_meta = tsfile_executor_->get_tsfile_meta();
std::vector<std::shared_ptr<IDeviceID>> device_ids;
- if (tsfile_meta != nullptr) {
- PageArena pa;
- pa.init(512, MOD_TSFILE_READER);
- to_lowercase_inplace(table_name);
- auto it = tsfile_meta->table_metadata_index_node_map_.find(table_name);
- if (it != tsfile_meta->table_metadata_index_node_map_.end() &&
- it->second != nullptr) {
- get_all_devices(device_ids, it->second, pa);
+ // The error code is ignored to preserve the legacy interface.
+ // New code should use the error-reporting overload and check its return
+ // code.
+ get_all_devices(table_name, device_ids);
+ return device_ids;
+}
+
+int TsFileReader::get_all_devices(
+ std::string table_name,
+ std::vector<std::shared_ptr<IDeviceID>>& device_ids) {
+ device_ids.clear();
+ if (tsfile_executor_ == nullptr) {
+ return E_INVALID_ARG;
+ }
+ TsFileMeta* tsfile_meta = nullptr;
+ int ret = E_OK;
+ if (RET_FAIL(tsfile_executor_->get_tsfile_meta(tsfile_meta))) {
+ return ret;
+ }
+ PageArena pa;
+ pa.init(512, MOD_TSFILE_READER);
+ to_lowercase_inplace(table_name);
+ auto it = tsfile_meta->table_metadata_index_node_map_.find(table_name);
+ if (it != tsfile_meta->table_metadata_index_node_map_.end() &&
+ it->second != nullptr) {
+ if (RET_FAIL(get_all_devices(device_ids, it->second, pa))) {
+ device_ids.clear();
+ return ret;
}
}
- return device_ids;
+ return E_OK;
}
std::vector<std::shared_ptr<IDeviceID>> TsFileReader::get_all_device_ids() {
@@ -554,7 +573,7 @@ int read_chunk_header_codec(RandomAccessReadFile* read_file,
return ret;
}
if (read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
- return E_TSFILE_CORRUPTED;
+ return E_FILE_READ_ERR;
}
common::ByteStream in;
in.wrap_from(buf.data(), read_len);
@@ -653,6 +672,21 @@ int TsFileReader::get_timeseries_metadata_impl(
DeviceTimeseriesMetadataMap TsFileReader::get_timeseries_metadata(
const std::vector<std::shared_ptr<IDeviceID>>& device_ids) {
+ DeviceTimeseriesMetadataMap result;
+ // The error code is ignored to preserve the legacy interface.
+ // New code should use the error-reporting overload and check its return
+ // code.
+ get_timeseries_metadata(device_ids, result);
+ return result;
+}
+
+int TsFileReader::get_timeseries_metadata(
+ const std::vector<std::shared_ptr<IDeviceID>>& device_ids,
+ DeviceTimeseriesMetadataMap& result) {
+ result.clear();
+ if (tsfile_executor_ == nullptr) {
+ return E_INVALID_ARG;
+ }
// Reset the shared meta arena up front: every call writes fresh
// timeseries-index metadata into it via _impl(), and the previous
// implementation only ever appended. A long-lived reader that repeats
@@ -662,22 +696,44 @@ DeviceTimeseriesMetadataMap TsFileReader::get_timeseries_metadata(
// shared_ptrs handed back use a noop deleter pointing into this arena.
tsfile_reader_meta_pa_.destroy();
tsfile_reader_meta_pa_.init(512, MOD_TSFILE_READER);
- DeviceTimeseriesMetadataMap result;
+ DeviceTimeseriesMetadataMap collected;
+ int ret = E_OK;
for (const auto& device_id : device_ids) {
+ if (!device_id) {
+ return E_INVALID_ARG;
+ }
std::vector<std::shared_ptr<ITimeseriesIndex>> list;
- if (get_timeseries_metadata_impl(device_id, list) == E_OK) {
- result.insert(std::make_pair(device_id, std::move(list)));
+ if (RET_FAIL(get_timeseries_metadata_impl(device_id, list))) {
+ if (ret == E_DEVICE_NOT_EXIST) {
+ continue;
+ }
+ return ret;
+ } else {
+ collected.insert(std::make_pair(device_id, std::move(list)));
}
- // Skip non-existent devices (not inserted)
}
- return result;
+ result.swap(collected);
+ return E_OK;
}
DeviceTimeseriesMetadataMap TsFileReader::get_timeseries_metadata() {
DeviceTimeseriesMetadataMap result;
- TsFileMeta* tsfile_meta = tsfile_executor_->get_tsfile_meta();
- if (tsfile_meta == nullptr) {
- return result;
+ // The error code is ignored to preserve the legacy interface.
+ // New code should use the error-reporting overload and check its return
+ // code.
+ get_timeseries_metadata(result);
+ return result;
+}
+
+int TsFileReader::get_timeseries_metadata(DeviceTimeseriesMetadataMap& result) {
+ result.clear();
+ if (tsfile_executor_ == nullptr) {
+ return E_INVALID_ARG;
+ }
+ TsFileMeta* tsfile_meta = nullptr;
+ int ret = E_OK;
+ if (RET_FAIL(tsfile_executor_->get_tsfile_meta(tsfile_meta))) {
+ return ret;
}
// Same arena-reset rationale as the device_ids overload above.
@@ -688,29 +744,33 @@ DeviceTimeseriesMetadataMap TsFileReader::get_timeseries_metadata() {
pa.init(512, MOD_TSFILE_READER);
std::vector<DeviceMetaEntry> entries;
for (auto& table_entry : tsfile_meta->table_metadata_index_node_map_) {
- if (get_all_device_entries(entries, table_entry.second,
- read_file_.get(), pa) != E_OK) {
- return result;
+ if (RET_FAIL(get_all_device_entries(entries, table_entry.second,
+ read_file_.get(), pa))) {
+ return ret;
}
}
auto noop_deleter = [](ITimeseriesIndex*) {};
+ DeviceTimeseriesMetadataMap collected;
for (auto& device_entry : entries) {
std::vector<ITimeseriesIndex*> raw_ts_indexes;
- if (tsfile_executor_->get_tsfile_io_reader()
- ->get_device_timeseries_meta_by_offset(
- device_entry.start_offset, device_entry.end_offset,
- raw_ts_indexes, tsfile_reader_meta_pa_) == E_OK) {
+ if (RET_FAIL(tsfile_executor_->get_tsfile_io_reader()
+ ->get_device_timeseries_meta_by_offset(
+ device_entry.start_offset, device_entry.end_offset,
+ raw_ts_indexes, tsfile_reader_meta_pa_))) {
+ return ret;
+ } else {
std::vector<std::shared_ptr<ITimeseriesIndex>> list;
for (auto ts_idx : raw_ts_indexes) {
list.emplace_back(
std::shared_ptr<ITimeseriesIndex>(ts_idx, noop_deleter));
}
- result.insert(
+ collected.insert(
std::make_pair(device_entry.device_id, std::move(list)));
}
}
- return result;
+ result.swap(collected);
+ return E_OK;
}
TsFileProperties TsFileReader::get_tsfile_properties() {
@@ -730,19 +790,39 @@ ResultSet* TsFileReader::read_timeseries(
std::shared_ptr<TableSchema> TsFileReader::get_table_schema(
const std::string& table_name) {
- TsFileMeta* file_metadata = tsfile_executor_->get_tsfile_meta();
std::shared_ptr<TableSchema> table_schema;
+ // Errors are reduced to nullptr to preserve the legacy interface.
+ // New code should use the error-reporting overload and check its return
+ // code.
+ if (get_table_schema(table_name, table_schema) != E_OK) {
+ return nullptr;
+ }
+ return table_schema;
+}
+
+int TsFileReader::get_table_schema(const std::string& table_name,
+ std::shared_ptr<TableSchema>& table_schema) {
+ table_schema.reset();
+ if (tsfile_executor_ == nullptr) {
+ return E_INVALID_ARG;
+ }
+ TsFileMeta* file_metadata = nullptr;
+ int ret = E_OK;
+ if (RET_FAIL(tsfile_executor_->get_tsfile_meta(file_metadata))) {
+ return ret;
+ }
// A schema-only table has no device-level metadata index. Schema lookup
// must therefore be independent of the presence of data pages; callers
// can still construct an empty result set from the returned schema.
- if (file_metadata == nullptr) return table_schema;
- file_metadata->get_table_schema(to_lower(table_name), table_schema);
- return table_schema;
+ return file_metadata->get_table_schema(to_lower(table_name), table_schema);
}
std::vector<std::shared_ptr<TableSchema>>
TsFileReader::get_all_table_schemas() {
std::vector<std::shared_ptr<TableSchema>> table_schemas;
+ // The error code is ignored to preserve the legacy interface.
+ // New code should use the error-reporting overload and check its return
+ // code.
get_all_table_schemas(table_schemas);
return table_schemas;
}
diff --git a/cpp/src/reader/tsfile_reader.h b/cpp/src/reader/tsfile_reader.h
index 6493a4af7..38a27a760 100644
--- a/cpp/src/reader/tsfile_reader.h
+++ b/cpp/src/reader/tsfile_reader.h
@@ -199,12 +199,24 @@ class TsFileReader {
/**
* @brief get all devices in the tsfile
*
+ * @note Retained for backward compatibility. This signature cannot report
+ * error codes; read failures return an empty list. Avoid this overload in
+ * new code: use the error-reporting overload and check its return code.
+ *
* @param table_name the table name
* @return std::vector<std::shared_ptr<IDeviceID>> the device id list
*/
std::vector<std::shared_ptr<IDeviceID>> get_all_devices(
std::string table_name);
+ /**
+ * Error-reporting table-scoped overload. The output is empty on failure.
+ * A missing table or a table without devices returns E_OK and an empty
+ * list, matching the legacy overload.
+ */
+ int get_all_devices(std::string table_name,
+ std::vector<std::shared_ptr<IDeviceID>>& device_ids);
+
/**
* @brief get all devices in the tsfile
*
@@ -239,33 +251,71 @@ class TsFileReader {
* Only devices that exist in the file are included in the result.
* If device_ids is empty, returns an empty map.
*
+ * @note Retained for backward compatibility. This signature cannot report
+ * error codes; read failures return an empty map. Avoid this overload in
+ * new code: use the error-reporting overload and check its return code.
+ *
* @param device_ids device list to query
* @return map: IDeviceID -> list of timeseries metadata (only existing)
*/
DeviceTimeseriesMetadataMap get_timeseries_metadata(
const std::vector<std::shared_ptr<IDeviceID>>& device_ids);
+ /** Error-reporting overload. Missing devices are skipped; the output is
+ * empty on read failure. Metadata remains valid until the next metadata
+ * call or until the reader is closed, as with the legacy overload. */
+ int get_timeseries_metadata(
+ const std::vector<std::shared_ptr<IDeviceID>>& device_ids,
+ DeviceTimeseriesMetadataMap& result);
+
/**
* @brief Get timeseries metadata for all devices in the file.
*
+ * @note Retained for backward compatibility. This signature cannot report
+ * error codes; read failures return an empty map. Avoid this overload in
+ * new code: use the error-reporting overload and check its return code.
+ *
* @return map: IDeviceID -> list of timeseries metadata
*/
DeviceTimeseriesMetadataMap get_timeseries_metadata();
+ /** Error-reporting overload. The output is empty on failure. */
+ int get_timeseries_metadata(DeviceTimeseriesMetadataMap& result);
+
/** Return a copy of all file-level properties, preserving null values. */
TsFileProperties get_tsfile_properties();
/**
* @brief get the table schema by the table name
*
+ * @note Retained for backward compatibility. This signature cannot report
+ * error codes; both a missing table and a read failure return null. Avoid
+ * this overload in new code: use the error-reporting overload and check
+ * its return code.
+ *
* @param table_name the table name
* @return std::shared_ptr<TableSchema> the table schema
*/
std::shared_ptr<TableSchema> get_table_schema(
const std::string& table_name);
+
+ /**
+ * @brief Get the table schema by table name.
+ *
+ * @param table_name the table name
+ * @param[out] table_schema the resolved schema, null on failure
+ * @return Returns 0 on success, E_TABLE_NOT_EXIST when the table is
+ * absent, or a non-zero read error code on metadata failure.
+ */
+ int get_table_schema(const std::string& table_name,
+ std::shared_ptr<TableSchema>& table_schema);
/**
* @brief get all table schemas in the tsfile
*
+ * @note Retained for backward compatibility. This signature cannot report
+ * error codes; read failures return an empty list. Avoid this overload in
+ * new code: use the error-reporting overload and check its return code.
+ *
* @return std::vector<std::shared_ptr<TableSchema>> the table schema list
*/
std::vector<std::shared_ptr<TableSchema>> get_all_table_schemas();
diff --git a/cpp/test/cwrapper/c_release_test.cc b/cpp/test/cwrapper/c_release_test.cc
index c8ac72346..27f02d56c 100644
--- a/cpp/test/cwrapper/c_release_test.cc
+++ b/cpp/test/cwrapper/c_release_test.cc
@@ -407,16 +407,45 @@ TEST_F(CReleaseTest, TsFileWriterConfTest) {
free_write_file(&file);
TsFileReader reader = tsfile_reader_new("plain_file.tsfile", &err_no);
ASSERT_EQ(RET_OK, err_no);
- TableSchema schema = tsfile_reader_get_table_schema(reader, "plain_table");
+ ERRNO schema_error = RET_OK;
+ TableSchema schema{};
+ schema_error =
+ tsfile_reader_get_table_schema_checked(reader, "plain_table", &schema);
+ ASSERT_EQ(schema_error, RET_OK);
ASSERT_EQ(schema.column_num, 2);
uint32_t size = 0;
- DeviceSchema* device_schema =
- tsfile_reader_get_all_timeseries_schemas(reader, &size);
+ ERRNO timeseries_schema_error = RET_OK;
+ DeviceSchema* device_schema = nullptr;
+ timeseries_schema_error = tsfile_reader_get_all_timeseries_schemas_checked(
+ reader, &device_schema, &size);
+ ASSERT_EQ(timeseries_schema_error, RET_OK);
ASSERT_EQ(1, size);
ASSERT_EQ(1, device_schema->timeseries_num);
ASSERT_EQ(device_schema->timeseries_schema[0].encoding, TS_ENCODING_PLAIN);
ASSERT_EQ(device_schema->timeseries_schema[0].compression,
TS_COMPRESSION_UNCOMPRESSED);
+
+ TableSchema legacy_schema =
+ tsfile_reader_get_table_schema(reader, "plain_table");
+ ASSERT_EQ(legacy_schema.column_num, 2);
+ free_table_schema(legacy_schema);
+
+ uint32_t legacy_table_count = 0;
+ TableSchema* legacy_tables =
+ tsfile_reader_get_all_table_schemas(reader, &legacy_table_count);
+ ASSERT_EQ(legacy_table_count, 1);
+ ASSERT_NE(legacy_tables, nullptr);
+ free_table_schema(legacy_tables[0]);
+ free(legacy_tables);
+
+ uint32_t legacy_device_count = 0;
+ DeviceSchema* legacy_devices =
+ tsfile_reader_get_all_timeseries_schemas(reader, &legacy_device_count);
+ ASSERT_EQ(legacy_device_count, 1);
+ ASSERT_NE(legacy_devices, nullptr);
+ free_device_schema(legacy_devices[0]);
+ free(legacy_devices);
+
tsfile_reader_close(reader);
free_table_schema(schema);
free_device_schema(*device_schema);
diff --git a/cpp/test/cwrapper/cwrapper_test.cc b/cpp/test/cwrapper/cwrapper_test.cc
index 1d5309fd8..26bd424f7 100644
--- a/cpp/test/cwrapper/cwrapper_test.cc
+++ b/cpp/test/cwrapper/cwrapper_test.cc
@@ -512,4 +512,46 @@ TEST(TagFilterCApiTest, RejectsNullInputs) {
nullptr);
}
+TEST(CApiCompatibilityTest, KeepsLegacySchemaAndTagFilterEntryPoints) {
+ TableSchema legacy_schema =
+ tsfile_reader_get_table_schema(nullptr, "missing");
+ EXPECT_EQ(legacy_schema.table_name, nullptr);
+ EXPECT_EQ(legacy_schema.column_num, 0);
+ EXPECT_EQ(legacy_schema.column_schemas, nullptr);
+
+ uint32_t table_count = 7;
+ EXPECT_EQ(tsfile_reader_get_all_table_schemas(nullptr, &table_count),
+ nullptr);
+ EXPECT_EQ(table_count, 0);
+
+ uint32_t device_count = 7;
+ EXPECT_EQ(tsfile_reader_get_all_timeseries_schemas(nullptr, &device_count),
+ nullptr);
+ EXPECT_EQ(device_count, 0);
+
+ TableSchema checked_schema{};
+ EXPECT_EQ(tsfile_reader_get_table_schema_checked(nullptr, "missing",
+ &checked_schema),
+ common::E_INVALID_ARG);
+
+ TableSchema* checked_tables = nullptr;
+ table_count = 7;
+ EXPECT_EQ(tsfile_reader_get_all_table_schemas_checked(
+ nullptr, &checked_tables, &table_count),
+ common::E_INVALID_ARG);
+ EXPECT_EQ(checked_tables, nullptr);
+ EXPECT_EQ(table_count, 0);
+
+ DeviceSchema* checked_devices = nullptr;
+ device_count = 7;
+ EXPECT_EQ(tsfile_reader_get_all_timeseries_schemas_checked(
+ nullptr, &checked_devices, &device_count),
+ common::E_INVALID_ARG);
+ EXPECT_EQ(checked_devices, nullptr);
+ EXPECT_EQ(device_count, 0);
+
+ EXPECT_EQ(tsfile_tag_filter_eq(nullptr, "missing", "tag", "value"),
+ nullptr);
+}
+
} // namespace cwrapper
diff --git a/cpp/test/reader/table_read_failure_test.cc b/cpp/test/reader/table_read_failure_test.cc
new file mode 100644
index 000000000..9e8076612
--- /dev/null
+++ b/cpp/test/reader/table_read_failure_test.cc
@@ -0,0 +1,1005 @@
+/*
+ * 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 <gtest/gtest.h>
+
+#include <algorithm>
+#include <chrono>
+#include <cstdio>
+#include <cstring>
+#include <fstream>
+#include <iterator>
+#include <memory>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include "common/allocator/alloc_base.h"
+#include "common/config/config.h"
+#include "common/tablet.h"
+#include "cwrapper/tsfile_cwrapper.h"
+#include "file/tsfile_io_reader.h"
+#include "file/write_file.h"
+#include "reader/filter/tag_filter.h"
+#include "reader/meta_data_querier.h"
+#include "reader/table_result_set.h"
+#include "reader/tsfile_reader.h"
+#include "utils/injection.h"
+#include "writer/tsfile_table_writer.h"
+
+namespace {
+
+class InjectionGuard {
+ public:
+ explicit InjectionGuard(const char* point, int count_down = 0)
+ : point_(point) {
+ common::enable_injection(point_, count_down);
+ }
+ ~InjectionGuard() { common::disable_injection(point_); }
+
+ private:
+ const char* point_;
+};
+
+class FailingReadFile : public storage::RandomAccessReadFile {
+ public:
+ explicit FailingReadFile(const std::vector<char>& bytes) : bytes_(bytes) {}
+ bool is_opened() const override { return opened_; }
+ int64_t file_size() const override { return bytes_.size(); }
+ const std::string& file_path() const override { return path_; }
+ int generation(uint64_t& size, uint64_t& fingerprint) const override {
+ size = bytes_.size();
+ fingerprint = 0;
+ return common::E_OK;
+ }
+ int read(int64_t offset, char* buffer, int32_t size,
+ int32_t& read_size) override {
+ ++reads;
+ read_size = 0;
+ if (fail_at > 0 && (persistent ? reads >= fail_at : reads == fail_at)) {
+ failed = true;
+ return short_read ? common::E_OK : common::E_FILE_READ_ERR;
+ }
+ if (offset < 0 || size < 0) return common::E_INVALID_ARG;
+ if (offset >= static_cast<int64_t>(bytes_.size())) return common::E_OK;
+ read_size = static_cast<int32_t>(
+ std::min<int64_t>(size, bytes_.size() - offset));
+ std::memcpy(buffer, bytes_.data() + offset, read_size);
+ return common::E_OK;
+ }
+ void close() override { opened_ = false; }
+
+ int reads = 0;
+ int fail_at = 0;
+ bool persistent = false;
+ bool short_read = false;
+ bool failed = false;
+
+ private:
+ const std::vector<char>& bytes_;
+ bool opened_ = true;
+ std::string path_ = "memory://table-read-failure";
+};
+
+// One tag exercises TagEq's direct lookup; two tags exercise filtered
+// traversal. Two devices fit in a leaf; five force multiple levels of internal
+// index nodes.
+class TableReadFailureTest
+ : public ::testing::TestWithParam<std::tuple<int, int>> {
+ protected:
+ void SetUp() override {
+ storage::libtsfile_init();
+ saved_index_degree_ = common::g_config_value_.max_degree_of_index_node_;
+ ASSERT_EQ(storage::set_max_degree_of_index_node(2), common::E_OK);
+ filename_ =
+ "table_read_failure_" +
+ std::to_string(
+ std::chrono::steady_clock::now().time_since_epoch().count()) +
+ ".tsfile";
+ storage::WriteFile file;
+ int flags = O_WRONLY | O_CREAT | O_TRUNC;
+#ifdef _WIN32
+ flags |= O_BINARY;
+#endif
+ ASSERT_EQ(file.create(filename_, flags, 0666), common::E_OK);
+ std::vector<common::ColumnSchema> columns;
+ for (int i = 0; i < std::get<0>(GetParam()); ++i) {
+ columns.emplace_back("id" + std::to_string(i), common::STRING,
+ common::UNCOMPRESSED, common::PLAIN,
+ common::ColumnCategory::TAG);
+ }
+ columns.emplace_back("value", common::INT64, common::UNCOMPRESSED,
+ common::PLAIN, common::ColumnCategory::FIELD);
+ storage::TableSchema schema("test", columns);
+ storage::TsFileTableWriter writer(&file, &schema);
+ storage::Tablet tablet(
+ "test", schema.get_measurement_names(), schema.get_data_types(),
+ schema.get_column_categories(), std::get<1>(GetParam()) * 30);
+ for (int device = 0; device < std::get<1>(GetParam()); ++device) {
+ for (int t = 0; t < 30; ++t) {
+ const int row = device * 30 + t;
+ const std::string id = "d" + std::to_string(device);
+ ASSERT_EQ(tablet.add_timestamp(row, t), common::E_OK);
+ ASSERT_EQ(tablet.add_value(row, "id0", id.c_str()),
+ common::E_OK);
+ if (std::get<0>(GetParam()) == 2) {
+ ASSERT_EQ(tablet.add_value(row, "id1", "tag"),
+ common::E_OK);
+ }
+ ASSERT_EQ(
+ tablet.add_value(row, "value", static_cast<int64_t>(row)),
+ common::E_OK);
+ }
+ }
+ ASSERT_EQ(writer.write_table(tablet), common::E_OK);
+ ASSERT_EQ(writer.flush(), common::E_OK);
+ ASSERT_EQ(writer.close(), common::E_OK);
+ std::ifstream input(filename_, std::ios::binary);
+ ASSERT_TRUE(input.is_open());
+ bytes_.assign(std::istreambuf_iterator<char>(input),
+ std::istreambuf_iterator<char>());
+ }
+
+ void TearDown() override {
+ storage::set_max_degree_of_index_node(saved_index_degree_);
+ std::remove(filename_.c_str());
+ storage::libtsfile_destroy();
+ }
+
+ struct ScanResult {
+ int ret = common::E_OK;
+ int rows = 0;
+ int open_reads = 0;
+ int reads = 0;
+ bool failed = false;
+ };
+
+ void scan(bool filter, int batch_size, int fail_at, bool persistent,
+ bool short_read, ScanResult& out) {
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ source->fail_at = fail_at;
+ source->persistent = persistent;
+ source->short_read = short_read;
+ ASSERT_EQ(
+ reader.open(std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ out.open_reads = source->reads;
+ storage::TagEq eq(1, "d0");
+ storage::ResultSet* result = nullptr;
+ out.ret = reader.query("test", {"id0", "value"}, 0, INT64_MAX, result,
+ filter ? &eq : nullptr, batch_size);
+ if (out.ret == common::E_OK) {
+ ASSERT_NE(result, nullptr);
+ bool next = false;
+ while ((out.ret = result->next(next)) == common::E_OK && next) {
+ if (batch_size == 0) {
+ ++out.rows;
+ } else {
+ common::TsBlock* block = nullptr;
+ out.ret = result->get_next_tsblock(block);
+ if (out.ret != common::E_OK) break;
+ ASSERT_NE(block, nullptr);
+ out.rows += block->get_row_count();
+ }
+ }
+ }
+ out.reads = source->reads;
+ out.failed = source->failed;
+ if (out.ret != common::E_OK && result != nullptr) {
+ source->fail_at = 0;
+ bool next = true;
+ EXPECT_EQ(result->next(next), out.ret);
+ EXPECT_FALSE(next);
+ if (batch_size != 0) {
+ common::TsBlock* block = nullptr;
+ EXPECT_EQ(result->get_next_tsblock(block), out.ret);
+ EXPECT_EQ(block, nullptr);
+ }
+ EXPECT_EQ(source->reads, out.reads);
+ }
+ if (result != nullptr) reader.destroy_query_data_set(result);
+ }
+
+ std::vector<char> bytes_;
+ std::string filename_;
+ uint32_t saved_index_degree_ = 0;
+};
+
+TEST_P(TableReadFailureTest, EveryReadFailureReachesCaller) {
+ for (bool filter : {false, true}) {
+ for (int batch_size : {0, 16}) {
+ ScanResult baseline;
+ ASSERT_NO_FATAL_FAILURE(
+ scan(filter, batch_size, 0, false, false, baseline));
+ ASSERT_EQ(baseline.ret, common::E_OK);
+ ASSERT_EQ(baseline.rows,
+ filter ? 30 : std::get<1>(GetParam()) * 30);
+ for (bool persistent : {false, true}) {
+ for (bool short_read : {false, true}) {
+ for (int fail_at = baseline.open_reads + 1;
+ fail_at <= baseline.reads; ++fail_at) {
+ SCOPED_TRACE(::testing::Message()
+ << filter << ":" << batch_size << ":"
+ << persistent << ":" << short_read << ":"
+ << fail_at);
+ ScanResult failed;
+ ASSERT_NO_FATAL_FAILURE(scan(filter, batch_size,
+ fail_at, persistent,
+ short_read, failed));
+ EXPECT_TRUE(failed.failed);
+ EXPECT_EQ(failed.ret, common::E_FILE_READ_ERR);
+ }
+ }
+ }
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, MetadataApisPreserveReadErrors) {
+ for (int operation = 0; operation < 5; ++operation) {
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ source->fail_at = source->reads + 1;
+ source->persistent = true;
+ TableSchema schema{};
+ ERRNO schema_error = common::E_OK;
+ if (operation == 0) {
+ schema_error = tsfile_reader_get_table_schema_checked(
+ &reader, "test", &schema);
+ EXPECT_EQ(schema_error, common::E_FILE_READ_ERR);
+ EXPECT_EQ(schema.table_name, nullptr);
+ EXPECT_EQ(schema.column_num, 0);
+ EXPECT_EQ(schema.column_schemas, nullptr);
+ } else if (operation == 3) {
+ uint32_t count = 0;
+ DeviceSchema* device_schemas = nullptr;
+ schema_error = tsfile_reader_get_all_timeseries_schemas_checked(
+ &reader, &device_schemas, &count);
+ EXPECT_EQ(schema_error, common::E_FILE_READ_ERR);
+ EXPECT_EQ(count, 0u);
+ EXPECT_EQ(device_schemas, nullptr);
+ } else if (operation == 4) {
+ uint32_t count = 0;
+ TableSchema* schemas = nullptr;
+ schema_error = tsfile_reader_get_all_table_schemas_checked(
+ &reader, &schemas, &count);
+ EXPECT_EQ(schema_error, common::E_FILE_READ_ERR);
+ EXPECT_EQ(count, 0u);
+ EXPECT_EQ(schemas, nullptr);
+ } else {
+ ERRNO error = common::E_OK;
+ TagFilterHandle filter =
+ operation == 1
+ ? tsfile_tag_filter_create(&reader, "test", "id0", "d0",
+ TAG_FILTER_EQ, &error)
+ : tsfile_tag_filter_between(&reader, "test", "id0", "d0",
+ "d0", false, &error);
+ EXPECT_EQ(error, common::E_FILE_READ_ERR);
+ EXPECT_EQ(filter, nullptr);
+ if (filter != nullptr) tsfile_tag_filter_free(filter);
+ }
+ EXPECT_TRUE(source->failed);
+ // Metadata failures must not poison the reader or cache an empty
+ // schema.
+ source->fail_at = 0;
+ if (operation == 3) {
+ uint32_t count = 0;
+ DeviceSchema* device_schemas = nullptr;
+ schema_error = tsfile_reader_get_all_timeseries_schemas_checked(
+ &reader, &device_schemas, &count);
+ ASSERT_EQ(schema_error, common::E_OK);
+ ASSERT_NE(device_schemas, nullptr);
+ ASSERT_GT(count, 0u);
+ for (uint32_t i = 0; i < count; ++i) {
+ free_device_schema(device_schemas[i]);
+ }
+ free(device_schemas);
+ } else if (operation == 4) {
+ uint32_t count = 0;
+ TableSchema* schemas = nullptr;
+ schema_error = tsfile_reader_get_all_table_schemas_checked(
+ &reader, &schemas, &count);
+ ASSERT_EQ(schema_error, common::E_OK);
+ ASSERT_NE(schemas, nullptr);
+ ASSERT_GT(count, 0u);
+ for (uint32_t i = 0; i < count; ++i) {
+ free_table_schema(schemas[i]);
+ }
+ free(schemas);
+ } else {
+ schema_error = tsfile_reader_get_table_schema_checked(
+ &reader, "test", &schema);
+ ASSERT_EQ(schema_error, common::E_OK);
+ EXPECT_STREQ(schema.table_name, "test");
+ free_table_schema(schema);
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, CheckedMetadataApisPropagateEveryReadFailure) {
+ // Enumerating every read also covers failures after device-schema
+ // entries or device lists have been partially initialized.
+ for (int operation = 0; operation < 4; ++operation) {
+ auto invoke = [operation](storage::TsFileReader& reader,
+ bool expect_failure) {
+ if (operation == 3) {
+ std::vector<std::shared_ptr<storage::IDeviceID>> devices(1);
+ const int ret = reader.get_all_devices("TEST", devices);
+ if (expect_failure) {
+ EXPECT_TRUE(devices.empty());
+ } else {
+ EXPECT_FALSE(devices.empty());
+ }
+ return ret;
+ }
+ if (operation == 0) {
+ auto* schemas = reinterpret_cast<DeviceSchema*>(&reader);
+ uint32_t count = 7;
+ const ERRNO ret =
+ tsfile_reader_get_all_timeseries_schemas_checked(
+ &reader, &schemas, &count);
+ if (expect_failure) {
+ EXPECT_EQ(schemas, nullptr);
+ EXPECT_EQ(count, 0u);
+ } else {
+ EXPECT_NE(schemas, nullptr);
+ EXPECT_GT(count, 0u);
+ }
+ if (schemas != nullptr) {
+ for (uint32_t i = 0; i < count; ++i) {
+ free_device_schema(schemas[i]);
+ }
+ free(schemas);
+ }
+ return ret;
+ }
+ TagFilterHandle filter = &reader;
+ const ERRNO ret =
+ operation == 1
+ ? tsfile_tag_filter_create_checked(
+ &reader, "test", "id0", "d0", TAG_FILTER_EQ, &filter)
+ : tsfile_tag_filter_between_checked(
+ &reader, "test", "id0", "d0", "d1", false, &filter);
+ if (expect_failure) {
+ EXPECT_EQ(filter, nullptr);
+ } else {
+ EXPECT_NE(filter, nullptr);
+ }
+ tsfile_tag_filter_free(filter);
+ return ret;
+ };
+
+ int first_read = 0;
+ int last_read = 0;
+ {
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ first_read = source->reads + 1;
+ ASSERT_EQ(invoke(reader, false), common::E_OK);
+ last_read = source->reads;
+ ASSERT_GE(last_read, first_read);
+ }
+ for (bool short_read : {false, true}) {
+ for (int fail_at = first_read; fail_at <= last_read; ++fail_at) {
+ SCOPED_TRACE(::testing::Message()
+ << operation << ":" << short_read << ":"
+ << fail_at);
+ const int64_t allocated_before =
+ common::ModStat::get_instance().get_stat(
+ common::MOD_TSFILE_READER);
+ {
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(
+ source)),
+ common::E_OK);
+ source->fail_at = fail_at;
+ source->short_read = short_read;
+ EXPECT_EQ(invoke(reader, true), common::E_FILE_READ_ERR);
+ EXPECT_TRUE(source->failed);
+ if (operation == 3) {
+ storage::TsFileReader legacy_reader;
+ auto* legacy_source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ legacy_reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(
+ legacy_source)),
+ common::E_OK);
+ legacy_source->fail_at = fail_at;
+ legacy_source->short_read = short_read;
+ EXPECT_TRUE(
+ legacy_reader.get_all_devices("TEST").empty());
+ EXPECT_TRUE(legacy_source->failed);
+ source->fail_at = 0;
+ EXPECT_EQ(invoke(reader, false), common::E_OK);
+ }
+ }
+ EXPECT_EQ(common::ModStat::get_instance().get_stat(
+ common::MOD_TSFILE_READER),
+ allocated_before);
+ }
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, TableScopedDeviceEnumerationValidatesInputs) {
+ storage::TsFileReader reader;
+ std::vector<std::shared_ptr<storage::IDeviceID>> devices(1);
+ EXPECT_EQ(reader.get_all_devices("test", devices), common::E_INVALID_ARG);
+ EXPECT_TRUE(devices.empty());
+ EXPECT_TRUE(reader.get_all_devices("test").empty());
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ ASSERT_EQ(reader.get_all_devices("TEST", devices), common::E_OK);
+ EXPECT_EQ(devices.size(), static_cast<size_t>(std::get<1>(GetParam())));
+ EXPECT_EQ(reader.get_all_devices("TEST").size(), devices.size());
+ ASSERT_EQ(reader.get_all_devices("missing", devices), common::E_OK);
+ EXPECT_TRUE(devices.empty());
+}
+
+TEST_P(TableReadFailureTest, LegacySchemaApisReturnEmptyOnReadFailure) {
+ for (int operation = 0; operation < 4; ++operation) {
+ SCOPED_TRACE(operation);
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ source->fail_at = source->reads + 1;
+ source->persistent = true;
+ if (operation == 0) {
+ TableSchema schema =
+ tsfile_reader_get_table_schema(&reader, "test");
+ EXPECT_EQ(schema.table_name, nullptr);
+ EXPECT_EQ(schema.column_num, 0);
+ EXPECT_EQ(schema.column_schemas, nullptr);
+ free_table_schema(schema);
+ } else if (operation == 1) {
+ uint32_t count = 7;
+ TableSchema* schemas =
+ tsfile_reader_get_all_table_schemas(&reader, &count);
+ EXPECT_EQ(schemas, nullptr);
+ EXPECT_EQ(count, 0u);
+ if (schemas != nullptr) {
+ for (uint32_t i = 0; i < count; ++i)
+ free_table_schema(schemas[i]);
+ free(schemas);
+ }
+ } else if (operation == 2) {
+ uint32_t count = 7;
+ DeviceSchema* schemas =
+ tsfile_reader_get_all_timeseries_schemas(&reader, &count);
+ EXPECT_EQ(schemas, nullptr);
+ EXPECT_EQ(count, 0u);
+ if (schemas != nullptr) {
+ for (uint32_t i = 0; i < count; ++i)
+ free_device_schema(schemas[i]);
+ free(schemas);
+ }
+ } else {
+ EXPECT_EQ(reader.get_table_schema("test"), nullptr);
+ }
+ EXPECT_TRUE(source->failed);
+ source->fail_at = 0;
+ auto schema = reader.get_table_schema("test");
+ ASSERT_NE(schema, nullptr);
+ EXPECT_EQ(schema->get_table_name(), "test");
+ }
+}
+
+TEST_P(TableReadFailureTest, TimeseriesMetadataPropagatesEveryReadFailure) {
+ for (bool selected_devices : {false, true}) {
+ auto invoke = [this, selected_devices](
+ storage::TsFileReader& reader,
+ storage::DeviceTimeseriesMetadataMap& result) {
+ if (!selected_devices)
+ return reader.get_timeseries_metadata(result);
+ std::vector<std::shared_ptr<storage::IDeviceID>> devices;
+ for (int i = 0; i < 5; ++i) {
+ std::vector<std::string> segments{"test",
+ "d" + std::to_string(i)};
+ if (std::get<0>(GetParam()) == 2) segments.push_back("tag");
+ devices.emplace_back(
+ new storage::StringArrayDeviceID(segments));
+ }
+ return reader.get_timeseries_metadata(devices, result);
+ };
+ int first_read = 0;
+ int last_read = 0;
+ {
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ first_read = source->reads + 1;
+ storage::DeviceTimeseriesMetadataMap result;
+ ASSERT_EQ(invoke(reader, result), common::E_OK);
+ ASSERT_EQ(result.size(),
+ static_cast<size_t>(std::get<1>(GetParam())));
+ last_read = source->reads;
+ ASSERT_GE(last_read, first_read);
+ }
+ for (bool persistent : {false, true}) {
+ for (bool short_read : {false, true}) {
+ for (int fail_at = first_read; fail_at <= last_read;
+ ++fail_at) {
+ SCOPED_TRACE(::testing::Message()
+ << selected_devices << ":" << persistent << ":"
+ << short_read << ":" << fail_at);
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(
+ source)),
+ common::E_OK);
+ source->fail_at = fail_at;
+ source->persistent = persistent;
+ source->short_read = short_read;
+ storage::DeviceTimeseriesMetadataMap result;
+ result[nullptr] = {};
+ EXPECT_EQ(invoke(reader, result), common::E_FILE_READ_ERR);
+ EXPECT_TRUE(source->failed);
+ EXPECT_TRUE(result.empty());
+ source->fail_at = 0;
+ ASSERT_EQ(invoke(reader, result), common::E_OK);
+ EXPECT_EQ(result.size(),
+ static_cast<size_t>(std::get<1>(GetParam())));
+ }
+ }
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, CMetadataPropagatesEveryReadFailure) {
+ std::vector<std::string> names;
+ for (int i = 0; i < 5; ++i) names.push_back("d" + std::to_string(i));
+ DeviceID devices[5]{};
+ char* segments[5][3]{};
+ for (int i = 0; i < 5; ++i) {
+ segments[i][0] = const_cast<char*>("test");
+ segments[i][1] = const_cast<char*>(names[i].c_str());
+ segments[i][2] = const_cast<char*>("tag");
+ devices[i].segment_count = 1 + std::get<0>(GetParam());
+ devices[i].segments = segments[i];
+ }
+ for (bool selected : {false, true}) {
+ auto invoke = [&](storage::TsFileReader& reader) {
+ DeviceTimeseriesMetadataMap result{};
+ const int ret =
+ selected ? tsfile_reader_get_timeseries_metadata_for_devices(
+ &reader, devices, 5, &result)
+ : tsfile_reader_get_timeseries_metadata_all(&reader,
+ &result);
+ if (ret == common::E_OK) {
+ EXPECT_EQ(result.device_count,
+ static_cast<uint32_t>(std::get<1>(GetParam())));
+ EXPECT_NE(result.entries, nullptr);
+ } else {
+ EXPECT_EQ(result.device_count, 0u);
+ EXPECT_EQ(result.entries, nullptr);
+ }
+ tsfile_free_device_timeseries_metadata_map(&result);
+ return ret;
+ };
+ int first_read = 0;
+ int last_read = 0;
+ {
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ first_read = source->reads + 1;
+ ASSERT_EQ(invoke(reader), common::E_OK);
+ last_read = source->reads;
+ ASSERT_GE(last_read, first_read);
+ }
+ for (bool persistent : {false, true}) {
+ for (int mode = 0; mode < 2; ++mode) {
+ for (int fail_at = first_read; fail_at <= last_read;
+ ++fail_at) {
+ SCOPED_TRACE(::testing::Message()
+ << selected << ":" << persistent << ":" << mode
+ << ":" << fail_at);
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(
+ source)),
+ common::E_OK);
+ source->fail_at = fail_at;
+ source->persistent = persistent;
+ source->short_read = mode == 1;
+ EXPECT_EQ(invoke(reader), common::E_FILE_READ_ERR);
+ EXPECT_TRUE(source->failed);
+ source->fail_at = 0;
+ ASSERT_EQ(invoke(reader), common::E_OK);
+ }
+ }
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, CMetadataCopyExceptionsClearOutput) {
+ storage::TsFileReader reader;
+ ASSERT_EQ(reader.open(std::unique_ptr<storage::RandomAccessReadFile>(
+ new FailingReadFile(bytes_))),
+ common::E_OK);
+ DeviceID* devices = nullptr;
+ uint32_t count = 0;
+ ASSERT_EQ(tsfile_reader_get_all_devices(&reader, &devices, &count),
+ common::E_OK);
+ ASSERT_GE(count, 2u);
+ for (bool selected : {false, true}) {
+ auto invoke = [&](DeviceTimeseriesMetadataMap* result) {
+ return selected ? tsfile_reader_get_timeseries_metadata_for_devices(
+ &reader, devices, count, result)
+ : tsfile_reader_get_timeseries_metadata_all(&reader,
+ result);
+ };
+ for (bool device_copy : {false, true}) {
+ for (bool oom : {false, true}) {
+ DeviceTimeseriesMetadataMap result{};
+ {
+ // Fail after one entry has been copied to exercise cleanup
+ // of both completed and partially populated C structures.
+ const char* point =
+ device_copy ? (oom ? "device_id_copy_oom"
+ : "device_id_copy_error")
+ : (oom ? "timeseries_metadata_copy_oom"
+ : "timeseries_metadata_copy_error");
+ InjectionGuard injection(point, 1);
+ EXPECT_EQ(invoke(&result),
+ oom ? common::E_OOM : common::E_FILE_READ_ERR);
+ EXPECT_EQ(result.entries, nullptr);
+ EXPECT_EQ(result.device_count, 0u);
+ tsfile_free_device_timeseries_metadata_map(&result);
+ }
+ ASSERT_EQ(invoke(&result), common::E_OK);
+ EXPECT_EQ(result.device_count, count);
+ tsfile_free_device_timeseries_metadata_map(&result);
+ }
+ }
+ }
+ tsfile_free_device_id_array(devices, count);
+}
+
+TEST_P(TableReadFailureTest, TimeseriesMetadataValidatesInputs) {
+ storage::TsFileReader reader;
+ storage::DeviceTimeseriesMetadataMap result;
+ result[nullptr] = {};
+ EXPECT_EQ(reader.get_timeseries_metadata(result), common::E_INVALID_ARG);
+ EXPECT_TRUE(result.empty());
+ const std::vector<std::shared_ptr<storage::IDeviceID>> no_devices;
+ result[nullptr] = {};
+ EXPECT_EQ(reader.get_timeseries_metadata(no_devices, result),
+ common::E_INVALID_ARG);
+ EXPECT_TRUE(result.empty());
+ ASSERT_EQ(reader.open(std::unique_ptr<storage::RandomAccessReadFile>(
+ new FailingReadFile(bytes_))),
+ common::E_OK);
+ EXPECT_EQ(reader.get_timeseries_metadata(no_devices, result), common::E_OK);
+ EXPECT_TRUE(result.empty());
+ result[nullptr] = {};
+ EXPECT_EQ(reader.get_timeseries_metadata({nullptr}, result),
+ common::E_INVALID_ARG);
+ EXPECT_TRUE(result.empty());
+ DeviceTimeseriesMetadataMap c_result{};
+ EXPECT_EQ(tsfile_reader_get_timeseries_metadata_all(nullptr, &c_result),
+ common::E_INVALID_ARG);
+ EXPECT_EQ(c_result.entries, nullptr);
+ EXPECT_EQ(c_result.device_count, 0u);
+ char* invalid_segments[] = {nullptr, const_cast<char*>("d0")};
+ DeviceID invalid_device{};
+ invalid_device.path = const_cast<char*>("test.d0");
+ invalid_device.segment_count = 2;
+ invalid_device.segments = invalid_segments;
+ EXPECT_EQ(tsfile_reader_get_timeseries_metadata_for_devices(
+ &reader, &invalid_device, 1, &c_result),
+ common::E_INVALID_ARG);
+ invalid_device.segments = nullptr;
+ EXPECT_EQ(tsfile_reader_get_timeseries_metadata_for_devices(
+ &reader, &invalid_device, 1, &c_result),
+ common::E_INVALID_ARG);
+}
+
+TEST_P(TableReadFailureTest, ColumnArrayAllocationFailureReturnsOOM) {
+ for (bool all_tables : {false, true}) {
+ SCOPED_TRACE(all_tables);
+ storage::TsFileReader reader;
+ ASSERT_EQ(reader.open(std::unique_ptr<storage::RandomAccessReadFile>(
+ new FailingReadFile(bytes_))),
+ common::E_OK);
+ InjectionGuard injection("table_schema_column_alloc_fail");
+ if (all_tables) {
+ TableSchema* schemas = nullptr;
+ uint32_t size = 1;
+ EXPECT_EQ(tsfile_reader_get_all_table_schemas_checked(
+ &reader, &schemas, &size),
+ common::E_OOM);
+ EXPECT_EQ(schemas, nullptr);
+ EXPECT_EQ(size, 0u);
+ } else {
+ TableSchema schema{};
+ EXPECT_EQ(tsfile_reader_get_table_schema_checked(&reader, "test",
+ &schema),
+ common::E_OOM);
+ EXPECT_EQ(schema.table_name, nullptr);
+ EXPECT_EQ(schema.column_num, 0);
+ EXPECT_EQ(schema.column_schemas, nullptr);
+ free_table_schema(schema);
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, SchemaCopyExceptionsClearOutput) {
+ for (bool oom : {false, true}) {
+ for (bool all_tables : {false, true}) {
+ SCOPED_TRACE(all_tables);
+ storage::TsFileReader reader;
+ ASSERT_EQ(
+ reader.open(std::unique_ptr<storage::RandomAccessReadFile>(
+ new FailingReadFile(bytes_))),
+ common::E_OK);
+ InjectionGuard injection(oom ? "table_schema_copy_oom"
+ : "table_schema_copy_error");
+ if (all_tables) {
+ TableSchema* schemas = nullptr;
+ uint32_t size = 1;
+ EXPECT_EQ(tsfile_reader_get_all_table_schemas_checked(
+ &reader, &schemas, &size),
+ oom ? common::E_OOM : common::E_FILE_READ_ERR);
+ EXPECT_EQ(schemas, nullptr);
+ EXPECT_EQ(size, 0u);
+ } else {
+ TableSchema schema{};
+ EXPECT_EQ(tsfile_reader_get_table_schema_checked(
+ &reader, "test", &schema),
+ oom ? common::E_OOM : common::E_FILE_READ_ERR);
+ EXPECT_EQ(schema.table_name, nullptr);
+ EXPECT_EQ(schema.column_num, 0);
+ EXPECT_EQ(schema.column_schemas, nullptr);
+ free_table_schema(schema);
+ }
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, WholeFileMetadataUpdatesCallerOutput) {
+ for (bool fail : {false, true}) {
+ SCOPED_TRACE(fail);
+ FailingReadFile source(bytes_);
+ source.fail_at = fail ? 1 : 0;
+ storage::TsFileIOReader io_reader;
+ ASSERT_EQ(io_reader.init(&source), common::E_OK);
+ storage::MetadataQuerier querier(&io_reader);
+ storage::IMetadataQuerier& interface = querier;
+ storage::TsFileMeta* metadata = nullptr;
+ if (fail) {
+ metadata = reinterpret_cast<storage::TsFileMeta*>(&querier);
+ EXPECT_EQ(interface.get_whole_file_metadata(metadata),
+ common::E_FILE_READ_ERR);
+ EXPECT_EQ(metadata, nullptr);
+ } else {
+ ASSERT_EQ(interface.get_whole_file_metadata(metadata),
+ common::E_OK);
+ ASSERT_NE(metadata, nullptr);
+ EXPECT_EQ(metadata->table_schemas_.count("test"), 1u);
+ }
+ }
+}
+
+TEST_P(TableReadFailureTest, NullTagOperatorsAcceptNullValues) {
+ storage::TsFileReader reader;
+ ASSERT_EQ(reader.open(std::unique_ptr<storage::RandomAccessReadFile>(
+ new FailingReadFile(bytes_))),
+ common::E_OK);
+ for (TagFilterOp op : {TAG_FILTER_IS_NULL, TAG_FILTER_IS_NOT_NULL}) {
+ for (bool checked : {false, true}) {
+ SCOPED_TRACE(::testing::Message() << op << ":" << checked);
+ TagFilterHandle filter = nullptr;
+ ERRNO ret = common::E_OK;
+ if (checked) {
+ ret = tsfile_tag_filter_create_checked(&reader, "test", "id0",
+ nullptr, op, &filter);
+ } else {
+ filter = tsfile_tag_filter_create(&reader, "test", "id0",
+ nullptr, op, &ret);
+ }
+ ASSERT_EQ(ret, common::E_OK);
+ ASSERT_NE(filter, nullptr);
+ storage::ResultSet* result = nullptr;
+ ret = reader.query("test", {"value"}, 0, INT64_MAX, result,
+ static_cast<storage::Filter*>(filter));
+ EXPECT_EQ(ret, common::E_OK);
+ int rows = 0;
+ bool next = false;
+ if (result != nullptr) {
+ while ((ret = result->next(next)) == common::E_OK && next) {
+ ++rows;
+ }
+ reader.destroy_query_data_set(result);
+ }
+ tsfile_tag_filter_free(filter);
+ EXPECT_EQ(ret, common::E_OK);
+ EXPECT_EQ(rows, op == TAG_FILTER_IS_NULL
+ ? 0
+ : std::get<1>(GetParam()) * 30);
+ }
+ }
+ for (TagFilterOp op : {TAG_FILTER_EQ, TAG_FILTER_NEQ, TAG_FILTER_LT,
+ TAG_FILTER_LTEQ, TAG_FILTER_GT, TAG_FILTER_GTEQ,
+ TAG_FILTER_REGEXP, TAG_FILTER_NOT_REGEXP}) {
+ TagFilterHandle filter = &reader;
+ EXPECT_EQ(tsfile_tag_filter_create_checked(&reader, "test", "id0",
+ nullptr, op, &filter),
+ common::E_INVALID_ARG);
+ EXPECT_EQ(filter, nullptr);
+ }
+}
+
+TEST_P(TableReadFailureTest, LegacyTagFactoriesReturnNullOnReadFailure) {
+ using Factory = TagFilterHandle (*)(TsFileReader, const char*, const char*,
+ const char*);
+ const Factory factories[] = {tsfile_tag_filter_eq, tsfile_tag_filter_neq,
+ tsfile_tag_filter_lt, tsfile_tag_filter_lteq,
+ tsfile_tag_filter_gt, tsfile_tag_filter_gteq};
+ for (auto factory : factories) {
+ storage::TsFileReader reader;
+ auto* source = new FailingReadFile(bytes_);
+ ASSERT_EQ(
+ reader.open(std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ source->fail_at = source->reads + 1;
+ source->persistent = true;
+ TagFilterHandle filter = factory(&reader, "test", "id0", "d0");
+ EXPECT_EQ(filter, nullptr);
+ tsfile_tag_filter_free(filter);
+ EXPECT_TRUE(source->failed);
+ source->fail_at = 0;
+ filter = factory(&reader, "test", "id0", "d0");
+ ASSERT_NE(filter, nullptr);
+ tsfile_tag_filter_free(filter);
+ }
+}
+
+TEST_P(TableReadFailureTest, CheckedTagFactoriesPreserveMissingTableError) {
+ storage::TsFileReader reader;
+ ASSERT_EQ(reader.open(std::unique_ptr<storage::RandomAccessReadFile>(
+ new FailingReadFile(bytes_))),
+ common::E_OK);
+ TagFilterHandle filter = &reader;
+ EXPECT_EQ(tsfile_tag_filter_create_checked(&reader, "missing", "id0", "d0",
+ TAG_FILTER_EQ, &filter),
+ common::E_TABLE_NOT_EXIST);
+ ASSERT_EQ(filter, nullptr);
+ filter = &reader;
+ EXPECT_EQ(tsfile_tag_filter_between_checked(&reader, "missing", "id0", "d0",
+ "d1", false, &filter),
+ common::E_TABLE_NOT_EXIST);
+ ASSERT_EQ(filter, nullptr);
+
+ ERRNO code = common::E_OK;
+ EXPECT_EQ(tsfile_tag_filter_create(&reader, "missing", "id0", "d0",
+ TAG_FILTER_EQ, &code),
+ nullptr);
+ EXPECT_EQ(code, common::E_INVALID_ARG);
+ code = common::E_OK;
+ EXPECT_EQ(tsfile_tag_filter_between(&reader, "missing", "id0", "d0", "d1",
+ false, &code),
+ nullptr);
+ EXPECT_EQ(code, common::E_INVALID_ARG);
+}
+
+INSTANTIATE_TEST_SUITE_P(LeafAndInternalDeviceIndexes, TableReadFailureTest,
+ ::testing::Combine(::testing::Values(1, 2),
+ ::testing::Values(2, 5)));
+
+TEST(TableMetadataReadFailureTest, ResizedBufferIsReleasedOnReadFailure) {
+ // A metadata tail larger than the initial 1 KiB buffer forces another
+ // read before deserialization. Inject the failure into that second read.
+ std::vector<char> bytes(2048, 0);
+ const uint32_t metadata_size = 1536;
+ for (int i = 0; i < 4; ++i) {
+ bytes[bytes.size() - 10 + i] =
+ static_cast<char>(metadata_size >> (8 * (3 - i)));
+ }
+ for (int failure = 0; failure < 3; ++failure) {
+ SCOPED_TRACE(failure);
+ const int64_t allocated_before =
+ common::ModStat::get_instance().get_stat(common::MOD_TSFILE_READER);
+ {
+ FailingReadFile source(bytes);
+ source.fail_at = 2;
+ source.short_read = failure == 1;
+ storage::TsFileIOReader reader;
+ ASSERT_EQ(reader.init(&source), common::E_OK);
+ storage::TsFileMeta* metadata = nullptr;
+ if (failure == 2) {
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(reader.get_tsfile_meta(metadata), common::E_OOM);
+ } else {
+ EXPECT_EQ(reader.get_tsfile_meta(metadata),
+ common::E_FILE_READ_ERR);
+ }
+ EXPECT_EQ(source.reads, failure == 2 ? 1 : 2);
+ }
+ EXPECT_EQ(
+ common::ModStat::get_instance().get_stat(common::MOD_TSFILE_READER),
+ allocated_before);
+ }
+}
+
+class FailingBlockReader : public storage::TsBlockReader {
+ public:
+ int has_next(bool& next) override {
+ next = true;
+ return common::E_OK;
+ }
+ int next(common::TsBlock*& block) override {
+ ++reads;
+ block = nullptr;
+ return common::E_FILE_READ_ERR;
+ }
+ void close() override {}
+ int reads = 0;
+};
+
+TEST(TableResultReadFailureTest, FailureWhileFetchingBlockIsTerminal) {
+ storage::libtsfile_init();
+ for (int mode : {storage::RETURN_ROW, storage::RETURN_BATCH}) {
+ auto* source = new FailingBlockReader();
+ storage::TableResultSet result(
+ std::unique_ptr<storage::TsBlockReader>(source), {"value"},
+ {common::INT64}, mode);
+ bool next = true;
+ common::TsBlock* block = nullptr;
+ if (mode == storage::RETURN_ROW) {
+ EXPECT_EQ(result.next(next), common::E_FILE_READ_ERR);
+ EXPECT_FALSE(next);
+ } else {
+ EXPECT_EQ(result.get_next_tsblock(block), common::E_FILE_READ_ERR);
+ EXPECT_EQ(block, nullptr);
+ }
+ EXPECT_EQ(result.next(next), common::E_FILE_READ_ERR);
+ EXPECT_FALSE(next);
+ EXPECT_EQ(result.get_next_tsblock(block), common::E_FILE_READ_ERR);
+ EXPECT_EQ(block, nullptr);
+ EXPECT_EQ(source->reads, 1);
+ }
+ storage::libtsfile_destroy();
+}
+
+} // namespace
diff --git a/cpp/test/reader/table_view/tsfile_reader_table_test.cc b/cpp/test/reader/table_view/tsfile_reader_table_test.cc
index 0fdfa8820..dfbff0576 100644
--- a/cpp/test/reader/table_view/tsfile_reader_table_test.cc
+++ b/cpp/test/reader/table_view/tsfile_reader_table_test.cc
@@ -539,7 +539,10 @@ TEST_F(TsFileTableReaderTest, TableModelGetSchema) {
}
}
- auto table_schema = reader.get_table_schema("testtable0");
+ std::shared_ptr<storage::TableSchema> table_schema;
+ ASSERT_EQ(reader.get_table_schema("testtable0", table_schema),
+ common::E_OK);
+ ASSERT_NE(table_schema, nullptr);
ASSERT_EQ(table_schema->get_table_name(), "testtable0");
for (int i = 0; i < 5; i++) {
ASSERT_EQ(table_schema->get_data_types()[i], TSDataType::STRING);
diff --git a/cpp/test/tools/cli_requirements_test.cc b/cpp/test/tools/cli_requirements_test.cc
index f48cec0f5..438f7507e 100644
--- a/cpp/test/tools/cli_requirements_test.cc
+++ b/cpp/test/tools/cli_requirements_test.cc
@@ -30,6 +30,10 @@
#include "cli/run_cli.h"
#include "cli_test_util.h"
+#include "commands/commands.h"
+#include "commands/statistics.h"
+#include "common/config/config.h"
+#include "file/local_random_access_read_file.h"
#include "reader/tsfile_reader.h"
#ifndef TSFILE_CPP_SOURCE_DIR
@@ -48,6 +52,29 @@ struct MultiTableFixture {
~MultiTableFixture() { std::remove(path.c_str()); }
};
+class FailingMetadataReadFile : public storage::LocalRandomAccessReadFile {
+ public:
+ int read(int64_t offset, char* buffer, int32_t size,
+ int32_t& read_size) override {
+ ++reads;
+ if (fail_reads || (fail_at > 0 && (persistent ? reads >= fail_at
+ : reads == fail_at))) {
+ failed = true;
+ fail_reads = persistent;
+ read_size = 0;
+ return short_read ? common::E_OK : common::E_FILE_READ_ERR;
+ }
+ return LocalRandomAccessReadFile::read(offset, buffer, size, read_size);
+ }
+
+ bool fail_reads = false;
+ bool persistent = false;
+ bool failed = false;
+ bool short_read = false;
+ int reads = 0;
+ int fail_at = 0;
+};
+
bool file_exists(const std::string& path) {
std::ifstream in(path.c_str());
return in.good();
@@ -93,6 +120,215 @@ std::string global_usage() {
} // namespace
+TEST(CliRequirements, FileSummaryPropagatesEveryMetadataReadFailure) {
+ for (bool internal_indexes : {false, true}) {
+ TableFixture fixture;
+ if (internal_indexes) {
+ const uint32_t saved =
+ common::g_config_value_.max_degree_of_index_node_;
+ ASSERT_EQ(storage::set_max_degree_of_index_node(2), common::E_OK);
+ std::remove(fixture.path.c_str());
+ fixture.path = tsfile_cli_test::write_tag_filter_fixture();
+ ASSERT_EQ(storage::set_max_degree_of_index_node(saved),
+ common::E_OK);
+ }
+ tsfile_cli::ParsedArgs args;
+ args.file = fixture.path;
+ int first_read = 0;
+ int last_read = 0;
+ {
+ auto* source = new FailingMetadataReadFile;
+ ASSERT_EQ(source->open(fixture.path), common::E_OK);
+ storage::TsFileReader reader;
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ first_read = source->reads + 1;
+ tsfile_cli::FileSummary summary;
+ std::ostringstream err;
+ ASSERT_EQ(
+ tsfile_cli::collect_file_summary(args, reader, summary, err), 0)
+ << err.str();
+ EXPECT_EQ(summary.device_count, internal_indexes ? 3 : 1);
+ EXPECT_GT(summary.series_count, 0);
+ last_read = source->reads;
+ }
+ for (bool persistent : {false, true}) {
+ for (bool short_read : {false, true}) {
+ for (int fail_at = first_read; fail_at <= last_read;
+ ++fail_at) {
+ SCOPED_TRACE(::testing::Message()
+ << internal_indexes << ":" << persistent << ":"
+ << short_read << ":" << fail_at);
+ auto* source = new FailingMetadataReadFile;
+ ASSERT_EQ(source->open(fixture.path), common::E_OK);
+ storage::TsFileReader reader;
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(
+ source)),
+ common::E_OK);
+ source->fail_at = fail_at;
+ source->persistent = persistent;
+ source->short_read = short_read;
+ tsfile_cli::FileSummary summary;
+ summary.file = "unchanged";
+ summary.device_count = -1;
+ std::ostringstream err;
+ EXPECT_EQ(tsfile_cli::collect_file_summary(args, reader,
+ summary, err),
+ 2);
+ EXPECT_TRUE(source->failed);
+ EXPECT_EQ(summary.file, "unchanged");
+ EXPECT_EQ(summary.device_count, -1);
+ EXPECT_NE(err.str().find("failed to read"),
+ std::string::npos)
+ << err.str();
+ }
+ }
+ }
+ }
+}
+
+TEST(CliRequirements, TableStatsPropagateEveryMetadataReadFailure) {
+ TableFixture fixture;
+ const uint32_t saved = common::g_config_value_.max_degree_of_index_node_;
+ ASSERT_EQ(storage::set_max_degree_of_index_node(2), common::E_OK);
+ std::remove(fixture.path.c_str());
+ fixture.path = tsfile_cli_test::write_tag_filter_fixture();
+ ASSERT_EQ(storage::set_max_degree_of_index_node(saved), common::E_OK);
+ tsfile_cli::ParsedArgs args;
+ args.file = fixture.path;
+ args.table = "t1";
+ int first_read = 0;
+ int last_read = 0;
+ for (int fail_at = 0; fail_at <= last_read; ++fail_at) {
+ if (fail_at > 0 && fail_at < first_read) {
+ continue;
+ }
+ for (bool persistent : {false, true}) {
+ for (bool short_read : {false, true}) {
+ SCOPED_TRACE(::testing::Message()
+ << persistent << ":" << short_read << ":"
+ << fail_at);
+ auto* source = new FailingMetadataReadFile;
+ ASSERT_EQ(source->open(fixture.path), common::E_OK);
+ storage::TsFileReader reader;
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ // Match run_cli's model validation before command dispatch.
+ std::vector<std::shared_ptr<storage::TableSchema>> schemas;
+ std::vector<std::shared_ptr<storage::IDeviceID>> devices;
+ ASSERT_EQ(reader.get_all_table_schemas(schemas), common::E_OK);
+ ASSERT_EQ(reader.get_all_devices(devices), common::E_OK);
+ first_read = source->reads + 1;
+ source->fail_at = fail_at;
+ source->persistent = persistent;
+ source->short_read = short_read;
+ std::ostringstream out;
+ std::ostringstream err;
+ const int ret = tsfile_cli::cmd_stats(
+ args, reader, tsfile_cli::OutputFormat::kCsv, out, err);
+ if (fail_at == 0) {
+ ASSERT_EQ(ret, 0) << err.str();
+ EXPECT_TRUE(err.str().empty()) << err.str();
+ EXPECT_NE(out.str().find("table,t1,dev_a,s1,INT64,1,"),
+ std::string::npos);
+ EXPECT_NE(out.str().find("table,t1,dev_b,s1,INT64,2,"),
+ std::string::npos);
+ EXPECT_NE(out.str().find("table,t1,dev_c,s1,INT64,1,"),
+ std::string::npos);
+ last_read = source->reads;
+ ASSERT_GE(last_read, first_read);
+ } else {
+ EXPECT_EQ(ret, 2);
+ EXPECT_TRUE(source->failed);
+ EXPECT_TRUE(out.str().empty()) << out.str();
+ EXPECT_NE(err.str().find("failed to read"),
+ std::string::npos)
+ << err.str();
+ }
+ }
+ }
+ }
+}
+
+TEST(CliRequirements, ModelDetectionPropagatesMetadataReadErrors) {
+ TableFixture fixture;
+ using Command =
+ int (*)(const tsfile_cli::ParsedArgs&, storage::TsFileReader&,
+ tsfile_cli::OutputFormat, std::ostream&, std::ostream&);
+ const Command commands[] = {tsfile_cli::cmd_ls, tsfile_cli::cmd_schema,
+ tsfile_cli::cmd_meta, tsfile_cli::cmd_stats,
+ tsfile_cli::cmd_count, tsfile_cli::cmd_head,
+ tsfile_cli::cmd_cat};
+ for (size_t i = 0; i < sizeof(commands) / sizeof(commands[0]); ++i) {
+ for (bool persistent : {false, true}) {
+ SCOPED_TRACE(::testing::Message() << i << ":" << persistent);
+ auto* source = new FailingMetadataReadFile;
+ ASSERT_EQ(source->open(fixture.path), common::E_OK);
+ storage::TsFileReader reader;
+ ASSERT_EQ(
+ reader.open(
+ std::unique_ptr<storage::RandomAccessReadFile>(source)),
+ common::E_OK);
+ source->fail_reads = true;
+ source->persistent = persistent;
+ tsfile_cli::ParsedArgs args;
+ args.file = fixture.path;
+ std::ostringstream out;
+ std::ostringstream err;
+ EXPECT_EQ(commands[i](args, reader, tsfile_cli::OutputFormat::kCsv,
+ out, err),
+ 2);
+ EXPECT_TRUE(source->failed);
+ EXPECT_TRUE(out.str().empty()) << out.str();
+ EXPECT_NE(err.str().find("failed to read"), std::string::npos)
+ << err.str();
+ }
+ }
+}
+
+TEST(CliRequirements, MetadataFailurePrecedesModelOptionValidation) {
+ TableFixture fixture;
+ const std::string bytes = read_file(fixture.path);
+ ASSERT_GT(bytes.size(), 17u);
+ // Keep both magic strings and the metadata-size tail, but remove the
+ // metadata body. Opening succeeds; the first schema read must fail.
+ {
+ std::ofstream file(fixture.path, std::ios::binary | std::ios::trunc);
+ file.write(bytes.data(), 7);
+ file.write(bytes.data() + bytes.size() - 10, 10);
+ }
+ storage::TsFileReader reader;
+ ASSERT_EQ(reader.open(fixture.path), common::E_OK);
+ reader.close();
+ for (const std::string& command :
+ {"ls", "schema", "meta", "stats", "count", "head", "cat"}) {
+ for (bool table_option : {false, true}) {
+ if (table_option && command != "head" && command != "cat" &&
+ command != "schema") {
+ continue;
+ }
+ SCOPED_TRACE(command + (table_option ? ":table" : ":auto"));
+ std::vector<std::string> args = {command};
+ if (table_option) {
+ args.insert(args.end(), {"-t", "table1"});
+ }
+ args.push_back(fixture.path);
+ std::ostringstream out;
+ std::ostringstream err;
+ EXPECT_EQ(tsfile_cli::run_cli(args, out, err), 2);
+ EXPECT_TRUE(out.str().empty()) << out.str();
+ EXPECT_NE(err.str().find("failed to read"), std::string::npos)
+ << err.str();
+ }
+ }
+}
+
TEST(CliRequirements, HelpListsExactlyCurrentCommandSurface) {
std::ostringstream out;
std::ostringstream err;
diff --git a/cpp/tools/cli/run_cli.cc b/cpp/tools/cli/run_cli.cc
index d46d681a8..aa9513935 100644
--- a/cpp/tools/cli/run_cli.cc
+++ b/cpp/tools/cli/run_cli.cc
@@ -215,10 +215,16 @@ CSV examples: \\N -> literal \N; \\path -> \path; \\\path -> \\path.
}
}
-bool has_mixed_model_metadata(storage::TsFileReader& reader) {
- const auto schemas = reader.get_all_table_schemas();
+int validate_file_model(storage::TsFileReader& reader, std::ostream& err) {
+ std::vector<std::shared_ptr<storage::TableSchema>> schemas;
+ const int schemas_ret = reader.get_all_table_schemas(schemas);
+ if (schemas_ret != common::E_OK) {
+ err << "Error: failed to read table schemas: "
+ << error_code_message(schemas_ret) << "\n";
+ return kExitFile;
+ }
if (schemas.empty()) {
- return false;
+ return kExitOk;
}
std::set<std::string> table_names;
@@ -227,7 +233,14 @@ bool has_mixed_model_metadata(storage::TsFileReader& reader) {
table_names.insert(storage::to_lower(schema->get_table_name()));
}
}
- for (const auto& device : reader.get_all_device_ids()) {
+ std::vector<std::shared_ptr<storage::IDeviceID>> devices;
+ const int devices_ret = reader.get_all_devices(devices);
+ if (devices_ret != common::E_OK) {
+ err << "Error: failed to read devices: "
+ << error_code_message(devices_ret) << "\n";
+ return kExitFile;
+ }
+ for (const auto& device : devices) {
if (device == nullptr) {
continue;
}
@@ -237,10 +250,12 @@ bool has_mixed_model_metadata(storage::TsFileReader& reader) {
// the CLI and must be reported as an input error.
if (table_names.find(storage::to_lower(device->get_table_name())) ==
table_names.end()) {
- return true;
+ err << "Error: input TsFile must be a pure tree-model or pure "
+ "table-model file\n";
+ return kExitFile;
}
}
- return false;
+ return kExitOk;
}
bool is_known_command(const std::string& c) {
@@ -608,18 +623,22 @@ int run_cli(const std::vector<std::string>& args, std::ostream& out,
return kExitFile;
}
- if (has_mixed_model_metadata(reader)) {
- err << "Error: input TsFile must be a pure tree-model or pure "
- "table-model file\n";
+ const int file_model_ret = validate_file_model(reader, err);
+ if (file_model_ret != kExitOk) {
reader.close();
- return kExitFile;
+ return file_model_ret;
}
// head/cat/export/schema dispatch on the data model and would silently
// ignore the scope flag of the other model; reject that instead.
if (p.command == "head" || p.command == "cat" || p.command == "export" ||
p.command == "schema") {
- const bool table_model = is_table_model(p, reader);
+ bool table_model = false;
+ const int model_ret = resolve_table_model(p, reader, table_model, err);
+ if (model_ret != kExitOk) {
+ reader.close();
+ return model_ret;
+ }
if (table_model && !p.device.empty()) {
err << "Error: -d/--device does not apply to the table model; "
"use -t/--table (or force --model tree)\n";
diff --git a/cpp/tools/commands/cmd_count.cc b/cpp/tools/commands/cmd_count.cc
index 9d8c75df6..e3ff9212f 100644
--- a/cpp/tools/commands/cmd_count.cc
+++ b/cpp/tools/commands/cmd_count.cc
@@ -93,11 +93,16 @@ int collect_table_count(const ParsedArgs& args, storage::TsFileReader& reader,
TableCountSummary& summary, std::ostream& err,
bool require_all_measurements) {
std::string table_name = storage::to_lower(args.table);
- std::shared_ptr<storage::TableSchema> schema =
- reader.get_table_schema(table_name);
- if (!schema) {
- err << "Error: table '" << args.table << "' does not exist\n";
- return kExitUsage;
+ std::shared_ptr<storage::TableSchema> schema;
+ const int schema_ret = reader.get_table_schema(table_name, schema);
+ if (schema_ret != common::E_OK) {
+ if (schema_ret == common::E_TABLE_NOT_EXIST) {
+ err << "Error: table '" << args.table << "' does not exist\n";
+ return kExitUsage;
+ }
+ err << "Error: failed to read schema for table '" << args.table
+ << "': " << error_code_message(schema_ret) << "\n";
+ return kExitFile;
}
summary.table_name = schema->get_table_name();
@@ -228,6 +233,11 @@ int collect_table_count(const ParsedArgs& args, storage::TsFileReader& reader,
int cmd_count(const ParsedArgs& args, storage::TsFileReader& reader,
OutputFormat fmt, std::ostream& out, std::ostream& err) {
+ bool table_model = false;
+ const int model_ret = resolve_table_model(args, reader, table_model, err);
+ if (model_ret != kExitOk) {
+ return model_ret;
+ }
RowWriter w(
out, fmt,
{"model", "object", "column", "category", "row_count", "entity_count",
@@ -237,7 +247,7 @@ int cmd_count(const ParsedArgs& args, storage::TsFileReader& reader,
common::INT64, common::INT64, common::STRING},
args.no_header);
- if (is_table_model(args, reader)) {
+ if (table_model) {
const std::vector<std::shared_ptr<storage::TableSchema>> all_schemas =
sorted_table_schemas(reader);
if (args.table.empty()) {
diff --git a/cpp/tools/commands/cmd_ls.cc b/cpp/tools/commands/cmd_ls.cc
index f04ae0495..8ee523f45 100644
--- a/cpp/tools/commands/cmd_ls.cc
+++ b/cpp/tools/commands/cmd_ls.cc
@@ -27,14 +27,25 @@
namespace tsfile_cli {
-bool is_table_model(const ParsedArgs& args, storage::TsFileReader& reader) {
+int resolve_table_model(const ParsedArgs& args, storage::TsFileReader& reader,
+ bool& table_model, std::ostream& err) {
+ table_model = false;
if (args.model == "tree") {
- return false;
+ return kExitOk;
}
if (args.model == "table") {
- return true;
+ table_model = true;
+ return kExitOk;
}
- return !reader.get_all_table_schemas().empty();
+ std::vector<std::shared_ptr<storage::TableSchema>> schemas;
+ const int ret = reader.get_all_table_schemas(schemas);
+ if (ret != common::E_OK) {
+ err << "Error: failed to read table schemas: "
+ << error_code_message(ret) << "\n";
+ return kExitFile;
+ }
+ table_model = !schemas.empty();
+ return kExitOk;
}
std::vector<std::shared_ptr<storage::TableSchema>> sorted_table_schemas(
@@ -52,8 +63,13 @@ std::vector<std::shared_ptr<storage::TableSchema>> sorted_table_schemas(
int cmd_ls(const ParsedArgs& args, storage::TsFileReader& reader,
OutputFormat fmt, std::ostream& out, std::ostream& err) {
+ bool table_model = false;
+ const int model_ret = resolve_table_model(args, reader, table_model, err);
+ if (model_ret != kExitOk) {
+ return model_ret;
+ }
std::vector<std::string> names;
- if (is_table_model(args, reader)) {
+ if (table_model) {
for (auto& ts : sorted_table_schemas(reader)) {
if (ts) {
names.push_back(ts->get_table_name());
@@ -67,7 +83,7 @@ int cmd_ls(const ParsedArgs& args, storage::TsFileReader& reader,
}
}
- const std::string model = is_table_model(args, reader) ? "table" : "tree";
+ const std::string model = table_model ? "table" : "tree";
RowWriter w(out, fmt, {"model", "object"}, {common::STRING, common::STRING},
false);
for (const std::string& n : names) {
diff --git a/cpp/tools/commands/cmd_meta.cc b/cpp/tools/commands/cmd_meta.cc
index 5f923b313..a18d7dfe2 100644
--- a/cpp/tools/commands/cmd_meta.cc
+++ b/cpp/tools/commands/cmd_meta.cc
@@ -28,7 +28,11 @@ namespace tsfile_cli {
int cmd_meta(const ParsedArgs& args, storage::TsFileReader& reader,
OutputFormat fmt, std::ostream& out, std::ostream& err) {
- FileSummary s = collect_file_summary(args, reader);
+ FileSummary s;
+ const int summary_ret = collect_file_summary(args, reader, s, err);
+ if (summary_ret != kExitOk) {
+ return summary_ret;
+ }
RowWriter w(out, fmt, {"size_bytes", "format_version", "model"},
{common::INT64, common::INT64, common::STRING}, false);
if (!w.write({std::to_string(s.file_size_bytes),
diff --git a/cpp/tools/commands/cmd_schema.cc b/cpp/tools/commands/cmd_schema.cc
index f402a58b3..8238436e5 100644
--- a/cpp/tools/commands/cmd_schema.cc
+++ b/cpp/tools/commands/cmd_schema.cc
@@ -132,8 +132,13 @@ int collect_table_schema_rows(const ParsedArgs& args,
int cmd_schema(const ParsedArgs& args, storage::TsFileReader& reader,
OutputFormat fmt, std::ostream& out, std::ostream& err) {
+ bool table_model = false;
+ const int model_ret = resolve_table_model(args, reader, table_model, err);
+ if (model_ret != kExitOk) {
+ return model_ret;
+ }
std::vector<SchemaRow> rows;
- if (is_table_model(args, reader)) {
+ if (table_model) {
int ret = collect_table_schema_rows(args, reader, rows, err);
if (ret != kExitOk) {
return ret;
diff --git a/cpp/tools/commands/cmd_stats.cc b/cpp/tools/commands/cmd_stats.cc
index df5371f1b..e84d7101d 100644
--- a/cpp/tools/commands/cmd_stats.cc
+++ b/cpp/tools/commands/cmd_stats.cc
@@ -318,10 +318,21 @@ int collect_table_stats(const ParsedArgs& args,
}
}
- std::vector<std::shared_ptr<storage::IDeviceID>> devices =
- reader.get_all_devices(schema->get_table_name());
- storage::DeviceTimeseriesMetadataMap metadata =
- reader.get_timeseries_metadata(devices);
+ std::vector<std::shared_ptr<storage::IDeviceID>> devices;
+ const int devices_ret =
+ reader.get_all_devices(schema->get_table_name(), devices);
+ if (devices_ret != common::E_OK) {
+ err << "Error: failed to read devices: "
+ << error_code_message(devices_ret) << "\n";
+ return kExitFile;
+ }
+ storage::DeviceTimeseriesMetadataMap metadata;
+ const int metadata_ret = reader.get_timeseries_metadata(devices, metadata);
+ if (metadata_ret != common::E_OK) {
+ err << "Error: failed to read timeseries metadata: "
+ << error_code_message(metadata_ret) << "\n";
+ return kExitFile;
+ }
for (const auto& device : devices) {
if (!device) {
continue;
@@ -415,14 +426,38 @@ int cmd_table_stats(const ParsedArgs& args, storage::TsFileReader& reader,
OutputFormat fmt, std::ostream& out, std::ostream& err) {
std::vector<std::shared_ptr<storage::TableSchema>> schemas;
if (!args.table.empty()) {
- schemas.push_back(
- reader.get_table_schema(storage::to_lower(args.table)));
+ std::shared_ptr<storage::TableSchema> schema;
+ const int schema_ret =
+ reader.get_table_schema(storage::to_lower(args.table), schema);
+ if (schema_ret != common::E_OK) {
+ if (schema_ret == common::E_TABLE_NOT_EXIST) {
+ err << "Error: table '" << args.table << "' does not exist\n";
+ return kExitUsage;
+ }
+ err << "Error: failed to read schema for table '" << args.table
+ << "': " << error_code_message(schema_ret) << "\n";
+ return kExitFile;
+ }
+ schemas.push_back(schema);
} else {
- schemas = sorted_table_schemas(reader);
- }
- if (schemas.empty() || !schemas[0]) {
- err << "Error: table '" << args.table << "' does not exist\n";
- return kExitUsage;
+ const int schemas_ret = reader.get_all_table_schemas(schemas);
+ if (schemas_ret != common::E_OK) {
+ err << "Error: failed to read table schemas: "
+ << error_code_message(schemas_ret) << "\n";
+ return kExitFile;
+ }
+ std::sort(
+ schemas.begin(), schemas.end(),
+ [](const std::shared_ptr<storage::TableSchema>& lhs,
+ const std::shared_ptr<storage::TableSchema>& rhs) {
+ if (!lhs) return false;
+ if (!rhs) return true;
+ return lhs->get_table_name() < rhs->get_table_name();
+ });
+ if (schemas.empty() || !schemas[0]) {
+ err << "Error: table '" << args.table << "' does not exist\n";
+ return kExitUsage;
+ }
}
if (args.table.empty()) {
@@ -623,7 +658,12 @@ int cmd_table_stats(const ParsedArgs& args, storage::TsFileReader& reader,
int cmd_stats(const ParsedArgs& args, storage::TsFileReader& reader,
OutputFormat fmt, std::ostream& out, std::ostream& err) {
- if (is_table_model(args, reader)) {
+ bool table_model = false;
+ const int model_ret = resolve_table_model(args, reader, table_model, err);
+ if (model_ret != kExitOk) {
+ return model_ret;
+ }
+ if (table_model) {
return cmd_table_stats(args, reader, fmt, out, err);
}
diff --git a/cpp/tools/commands/commands.h b/cpp/tools/commands/commands.h
index eb786cdd4..cd4ae9630 100644
--- a/cpp/tools/commands/commands.h
+++ b/cpp/tools/commands/commands.h
@@ -36,7 +36,8 @@ class TsFileReader;
namespace tsfile_cli {
-bool is_table_model(const ParsedArgs& args, storage::TsFileReader& reader);
+int resolve_table_model(const ParsedArgs& args, storage::TsFileReader& reader,
+ bool& table_model, std::ostream& err);
std::vector<std::shared_ptr<storage::TableSchema>> sorted_table_schemas(
storage::TsFileReader& reader);
@@ -49,9 +50,10 @@ std::vector<std::shared_ptr<storage::TableSchema>> sorted_table_schemas(
std::vector<std::string> collect_tree_query_paths(
const ParsedArgs& args, storage::TsFileReader& reader);
-std::unique_ptr<storage::Filter> build_table_tag_filter(
- const ParsedArgs& args, storage::TsFileReader& reader,
- const std::string& table_name, std::ostream& err);
+int build_table_tag_filter(const ParsedArgs& args,
+ storage::TsFileReader& reader,
+ const std::string& table_name, std::ostream& err,
+ std::unique_ptr<storage::Filter>& ret_filter);
int run_row_query(const ParsedArgs& args, storage::TsFileReader& reader,
OutputFormat fmt, std::ostream& out, std::ostream& err,
diff --git a/cpp/tools/commands/row_query.cc b/cpp/tools/commands/row_query.cc
index bcee4e7c4..0c606c924 100644
--- a/cpp/tools/commands/row_query.cc
+++ b/cpp/tools/commands/row_query.cc
@@ -152,16 +152,23 @@ int resolve_tree_paths(const ParsedArgs& args, storage::TsFileReader& reader,
} // namespace
-std::unique_ptr<storage::Filter> build_table_tag_filter(
- const ParsedArgs& args, storage::TsFileReader& reader,
- const std::string& table_name, std::ostream& err) {
+int build_table_tag_filter(const ParsedArgs& args,
+ storage::TsFileReader& reader,
+ const std::string& table_name, std::ostream& err,
+ std::unique_ptr<storage::Filter>& ret_filter) {
if (!args.has_tag_filter) {
- return std::unique_ptr<storage::Filter>();
+ return kExitOk;
}
- auto schema = reader.get_table_schema(table_name);
- if (!schema) {
- err << "Error: no schema found for table " << table_name << "\n";
- return std::unique_ptr<storage::Filter>();
+ std::shared_ptr<storage::TableSchema> schema;
+ const int schema_ret = reader.get_table_schema(table_name, schema);
+ if (schema_ret != common::E_OK) {
+ if (schema_ret == common::E_TABLE_NOT_EXIST) {
+ err << "Error: no schema found for table " << table_name << "\n";
+ return kExitUsage;
+ }
+ err << "Error: failed to read schema for table " << table_name << ": "
+ << error_code_message(schema_ret) << "\n";
+ return kExitFile;
}
storage::TagFilterBuilder builder(schema.get());
@@ -182,7 +189,7 @@ std::unique_ptr<storage::Filter> build_table_tag_filter(
} catch (const std::regex_error&) {
err << "Error: invalid regular expression for TAG '"
<< spec.column << "'\n";
- return std::unique_ptr<storage::Filter>();
+ return kExitUsage;
}
{
int tag_order = schema->find_id_column_order(spec.column);
@@ -204,7 +211,7 @@ std::unique_ptr<storage::Filter> build_table_tag_filter(
if (filter == nullptr) {
err << "Error: invalid tag filter column '" << spec.column
<< "' for table " << table_name << "\n";
- return std::unique_ptr<storage::Filter>();
+ return kExitUsage;
}
if (!combined) {
combined.reset(filter);
@@ -216,7 +223,8 @@ std::unique_ptr<storage::Filter> build_table_tag_filter(
combined.release(), filter));
}
}
- return combined;
+ ret_filter = std::move(combined);
+ return kExitOk;
}
std::vector<std::string> collect_tree_query_paths(
@@ -282,10 +290,21 @@ int run_row_query(const ParsedArgs& args, storage::TsFileReader& reader,
std::unique_ptr<storage::Filter> tag_filter;
std::string tree_device_prefix;
- if (is_table_model(args, reader)) {
+ bool table_model = false;
+ const int model_ret = resolve_table_model(args, reader, table_model, err);
+ if (model_ret != kExitOk) {
+ return model_ret;
+ }
+ if (table_model) {
std::string table_name = args.table;
if (table_name.empty()) {
- auto schemas = reader.get_all_table_schemas();
+ std::vector<std::shared_ptr<storage::TableSchema>> schemas;
+ const int schemas_ret = reader.get_all_table_schemas(schemas);
+ if (schemas_ret != common::E_OK) {
+ err << "Error: failed to read table schemas: "
+ << error_code_message(schemas_ret) << "\n";
+ return kExitFile;
+ }
if (schemas.empty() || !schemas[0]) {
err << "Error: no table found in file\n";
return kExitRuntime;
@@ -297,15 +316,27 @@ int run_row_query(const ParsedArgs& args, storage::TsFileReader& reader,
}
table_name = schemas[0]->get_table_name();
}
- auto table_schema = reader.get_table_schema(table_name);
+ std::shared_ptr<storage::TableSchema> table_schema;
+ const int schema_ret =
+ reader.get_table_schema(table_name, table_schema);
+ if (schema_ret != common::E_OK) {
+ if (schema_ret == common::E_TABLE_NOT_EXIST) {
+ err << "Error: table '" << table_name << "' does not exist\n";
+ return kExitUsage;
+ }
+ err << "Error: failed to read schema for table '" << table_name
+ << "': " << error_code_message(schema_ret) << "\n";
+ return kExitFile;
+ }
std::vector<std::string> cols;
int selection_ret = resolve_table_fields(args, table_schema, cols, err);
if (selection_ret != kExitOk) {
return selection_ret;
}
- tag_filter = build_table_tag_filter(args, reader, table_name, err);
- if (args.has_tag_filter && tag_filter == nullptr) {
- return kExitUsage;
+ int filter_ret =
+ build_table_tag_filter(args, reader, table_name, err, tag_filter);
+ if (filter_ret != kExitOk) {
+ return filter_ret;
}
if (push_down) {
qret = reader.queryByRow(
diff --git a/cpp/tools/commands/statistics.cc b/cpp/tools/commands/statistics.cc
index 2a8656c6e..b4b95101e 100644
--- a/cpp/tools/commands/statistics.cc
+++ b/cpp/tools/commands/statistics.cc
@@ -26,6 +26,7 @@
#include <map>
#include <set>
#include <sstream>
+#include <utility>
#include "cli/exit_codes.h"
#include "commands/commands.h"
@@ -315,18 +316,41 @@ int collect_series_stats(const ParsedArgs& args, storage::TsFileReader& reader,
return kExitOk;
}
-FileSummary collect_file_summary(const ParsedArgs& args,
- storage::TsFileReader& reader) {
+int collect_file_summary(const ParsedArgs& args, storage::TsFileReader& reader,
+ FileSummary& summary, std::ostream& err) {
+ bool table_model = false;
+ const int model_ret = resolve_table_model(args, reader, table_model, err);
+ if (model_ret != kExitOk) {
+ return model_ret;
+ }
FileSummary s;
s.file = args.file;
- s.model = is_table_model(args, reader) ? "table" : "tree";
- s.device_count = static_cast<long long>(reader.get_all_device_ids().size());
- s.table_count =
- static_cast<long long>(reader.get_all_table_schemas().size());
+ s.model = table_model ? "table" : "tree";
+ std::vector<std::shared_ptr<storage::IDeviceID>> devices;
+ const int devices_ret = reader.get_all_devices(devices);
+ if (devices_ret != common::E_OK) {
+ err << "Error: failed to read devices: "
+ << error_code_message(devices_ret) << "\n";
+ return kExitFile;
+ }
+ std::vector<std::shared_ptr<storage::TableSchema>> schemas;
+ const int schemas_ret = reader.get_all_table_schemas(schemas);
+ if (schemas_ret != common::E_OK) {
+ err << "Error: failed to read table schemas: "
+ << error_code_message(schemas_ret) << "\n";
+ return kExitFile;
+ }
+ s.device_count = static_cast<long long>(devices.size());
+ s.table_count = static_cast<long long>(schemas.size());
s.file_size_bytes = file_size(args.file);
- storage::DeviceTimeseriesMetadataMap metadata =
- reader.get_timeseries_metadata();
+ storage::DeviceTimeseriesMetadataMap metadata;
+ const int metadata_ret = reader.get_timeseries_metadata(metadata);
+ if (metadata_ret != common::E_OK) {
+ err << "Error: failed to read timeseries metadata: "
+ << error_code_message(metadata_ret) << "\n";
+ return kExitFile;
+ }
int64_t min_start = std::numeric_limits<int64_t>::max();
int64_t max_end = std::numeric_limits<int64_t>::min();
for (const auto& device : metadata) {
@@ -348,7 +372,8 @@ FileSummary collect_file_summary(const ParsedArgs& args,
s.start_time = min_start;
s.end_time = max_end;
}
- return s;
+ summary = std::move(s);
+ return kExitOk;
}
} // namespace tsfile_cli
diff --git a/cpp/tools/commands/statistics.h b/cpp/tools/commands/statistics.h
index a29df39a5..72e3d6e90 100644
--- a/cpp/tools/commands/statistics.h
+++ b/cpp/tools/commands/statistics.h
@@ -66,8 +66,8 @@ struct FileSummary {
StatisticCells statistic_value_cells(storage::Statistic* st);
int collect_series_stats(const ParsedArgs& args, storage::TsFileReader& reader,
std::vector<SeriesStatRow>& rows, std::ostream& err);
-FileSummary collect_file_summary(const ParsedArgs& args,
- storage::TsFileReader& reader);
+int collect_file_summary(const ParsedArgs& args, storage::TsFileReader& reader,
+ FileSummary& summary, std::ostream& err);
} // namespace tsfile_cli
diff --git a/python/tests/test_reader_metadata.py b/python/tests/test_reader_metadata.py
index fd00d7f74..4ab59784b 100644
--- a/python/tests/test_reader_metadata.py
+++ b/python/tests/test_reader_metadata.py
@@ -269,3 +269,47 @@ def test_get_timeseries_metadata_table_timeline_statistic_keeps_null_rows():
os.unlink(path)
except OSError:
pass
+
+
+def test_selected_metadata_preserves_table_device_segments(tmp_path):
+ path = tmp_path / "structured_device_metadata.tsfile"
+ schema = TableSchema(
+ "weather",
+ [
+ ColumnSchema("region", TSDataType.STRING, ColumnCategory.TAG),
+ ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
+ ColumnSchema("value", TSDataType.INT64, ColumnCategory.FIELD),
+ ],
+ )
+ frame = pd.DataFrame(
+ {
+ "time": [0, 1, 2, 3],
+ "region": ["north.east", "north.east", "north.east", ""],
+ "device": [None, "null", "a.b", "d"],
+ "value": [10, 11, 12, 13],
+ }
+ )
+ with TsFileTableWriter(str(path), schema) as writer:
+ writer.write_dataframe(frame)
+
+ with TsFileReader(str(path)) as reader:
+ devices = reader.get_all_devices()
+ expected = reader.get_timeseries_metadata()
+ assert set(expected) == {
+ ("weather", "north.east", None),
+ ("weather", "north.east", "null"),
+ ("weather", "north.east", "a.b"),
+ ("weather", "", "d"),
+ }
+ selected = reader.get_timeseries_metadata(devices)
+ assert set(selected) == set(expected)
+ for device in devices:
+ # Structured segments must win even if the display path is absent
+ # or ambiguous because a tag contains a dot.
+ for display_path in (None, "unrelated.path"):
+ query = DeviceID(display_path, None, device.segments)
+ group = reader.get_timeseries_metadata([query])[device.segments]
+ assert group.segments == expected[device.segments].segments
+ assert group.timeseries[0].statistic.sum == pytest.approx(
+ expected[device.segments].timeseries[0].statistic.sum
+ )
diff --git a/python/tests/test_reader_sources.py b/python/tests/test_reader_sources.py
index 24961f85e..11c3c5095 100644
--- a/python/tests/test_reader_sources.py
+++ b/python/tests/test_reader_sources.py
@@ -39,6 +39,7 @@ from tsfile import (
TsFileWriter,
)
from tsfile.exceptions import FileOpenError, FileReadError
+from tsfile.tag_filter import BetweenTagFilter, ComparisonTagFilter
RESOURCES = Path(__file__).parent / "resources"
@@ -404,7 +405,16 @@ def test_tree_query_propagates_data_block_read_errors(tmp_path, measurements, st
assert source.close_calls == 0
-@pytest.mark.parametrize("method", ["get_all_devices", "get_all_table_schemas"])
+@pytest.mark.parametrize(
+ "method",
+ [
+ "get_all_devices",
+ "get_all_table_schemas",
+ "get_table_schema",
+ "get_all_timeseries_schemas",
+ "get_timeseries_metadata",
+ ],
+)
def test_metadata_read_failure_is_not_an_empty_result(method):
class FailingBytesIO(TrackingBytesIO):
fail_reads = False
@@ -420,8 +430,11 @@ def test_metadata_read_failure_is_not_an_empty_result(method):
source.seek(17)
with TsFileReader(source) as reader:
source.fail_reads = True
+ args = ("test",) if method == "get_table_schema" else ()
with pytest.raises(FileReadError):
- getattr(reader, method)()
+ getattr(reader, method)(*args)
+ source.fail_reads = False
+ assert getattr(reader, method)(*args) is not None
assert source.failed
assert source.tell() == 17
assert source.close_calls == 0
@@ -461,3 +474,101 @@ def test_device_index_read_failure_does_not_return_partial_devices(tmp_path):
assert source.failed
assert source.tell() == 17
assert source.close_calls == 0
+
+
+@pytest.mark.parametrize(
+ "method,columns,offset",
+ [
+ ("query_table", ["s0", "s2"], 0),
+ ("query_table_by_row", ["s0", "s2"], 0),
+ ("query_table", ["s0", "s2", "s3"], 0),
+ ("query_table_by_row", ["s0", "s2", "s3"], 0),
+ ("query_table_by_row", ["s0", "s2"], 1),
+ ("query_table_by_row", ["s0", "s2", "s3"], 1),
+ ],
+)
+@pytest.mark.parametrize("tag_kind", [None, "eq", "between"])
+@pytest.mark.parametrize("batch_size", [0, 16])
+@pytest.mark.parametrize("short_read", [False, True])
+@pytest.mark.parametrize("persistent", [False, True])
+def test_table_queries_propagate_every_read_failure(
+ method, columns, offset, tag_kind, batch_size, short_read, persistent
+):
+ class FailingBytesIO(TrackingBytesIO):
+ read_calls = 0
+ fail_at = None
+ failed = False
+
+ def read(self, size=-1):
+ self.read_calls += 1
+ if self.fail_at is not None and (
+ self.read_calls >= self.fail_at
+ if persistent
+ else self.read_calls == self.fail_at
+ ):
+ self.failed = True
+ if short_read:
+ return b""
+ raise OSError("table read failed")
+ return super().read(size)
+
+ data = (RESOURCES / "simple_table_t1.tsfile").read_bytes()
+
+ def query(reader):
+ tag_filter = None
+ if tag_kind == "eq":
+ tag_filter = ComparisonTagFilter("s0", "a", ComparisonTagFilter.EQ)
+ elif tag_kind == "between":
+ tag_filter = BetweenTagFilter("s0", "a", "a")
+ kwargs = {"offset": offset} if method == "query_table_by_row" else {}
+ return getattr(reader, method)(
+ "test", columns, tag_filter=tag_filter, batch_size=batch_size, **kwargs
+ )
+
+ def consume(result):
+ rows = 0
+ if batch_size:
+ while True:
+ batch = result.read_arrow_record_batch()
+ if batch is None:
+ break
+ rows += batch.num_rows
+ else:
+ while result.next():
+ rows += 1
+ return rows
+
+ baseline = FailingBytesIO(data)
+ with TsFileReader(baseline) as reader:
+ open_reads = baseline.read_calls
+ with query(reader) as result:
+ assert consume(result) == (60 if tag_kind is None else 30) - offset
+
+ # Sweep actual reads rather than hard-coding call numbers: metadata may
+ # be prefetched or cached differently as the implementation evolves.
+ for fail_at in range(open_reads + 1, baseline.read_calls + 1):
+ source = FailingBytesIO(data)
+ source.seek(17)
+ with TsFileReader(source) as reader:
+ source.fail_at = fail_at
+ result = None
+ try:
+ with pytest.raises(FileReadError):
+ result = query(reader)
+ consume(result)
+ assert source.failed, f"read {fail_at} was not reached"
+ if result is not None:
+ # Recovery of the source must not turn a failed result
+ # into successful EOF or resume it after skipped devices.
+ source.fail_at = None
+ with pytest.raises(FileReadError):
+ result.next()
+ if batch_size:
+ with pytest.raises(FileReadError):
+ result.read_arrow_record_batch()
+ finally:
+ if result is not None:
+ result.close()
+ assert source.tell() == 17
+ assert not source.closed
+ assert source.close_calls == 0
diff --git a/python/tests/test_tag_filter_query.py b/python/tests/test_tag_filter_query.py
index 1d2482b5d..3d037a9c0 100644
--- a/python/tests/test_tag_filter_query.py
+++ b/python/tests/test_tag_filter_query.py
@@ -33,6 +33,8 @@ from tsfile import (
tag_gteq,
TIME_COLUMN,
)
+from tsfile.exceptions import ColumnNotExistError
+from tsfile.tag_filter import AndTagFilter, OrTagFilter, NotTagFilter
TSFILE_PATH = "test_tag_filter_query.tsfile"
@@ -44,6 +46,27 @@ COLUMNS = [
]
SCHEMA = TableSchema(TABLE_NAME, COLUMNS)
+
+@pytest.mark.parametrize("combine", [AndTagFilter, OrTagFilter])
+@pytest.mark.parametrize("nested", [False, True])
+@pytest.mark.parametrize("invalid_type", [False, True])
+def test_failed_compound_filter_construction_allows_recovery(
+ combine, nested, invalid_type
+):
+ good = tag_eq("region", "north")
+ bad = object() if invalid_type else tag_eq("missing", "north")
+ left = good & tag_eq("device", "dev_a") if nested else good
+ tag_filter = combine(left, NotTagFilter(bad) if nested else bad)
+ error = TypeError if invalid_type else ColumnNotExistError
+ with TsFileReader(TSFILE_PATH) as reader:
+ for _ in range(20):
+ with pytest.raises(error):
+ reader.query_table(TABLE_NAME, ["value"], tag_filter=tag_filter)
+ assert not reader.get_active_query_result()
+ with reader.query_table(TABLE_NAME, ["value"], tag_filter=good) as result:
+ assert sum(1 for _ in iter(result.next, False)) == 10
+
+
# Test data:
# region | device | timestamps | values
# north | dev_a | 0..4 | 0.0..4.0
diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd
index b0b2d6b78..a30de676d 100644
--- a/python/tsfile/tsfile_cpp.pxd
+++ b/python/tsfile/tsfile_cpp.pxd
@@ -365,6 +365,11 @@ cdef extern from "cwrapper/tsfile_cwrapper.h":
DeviceSchema * tsfile_reader_get_all_timeseries_schemas(TsFileReader reader,
uint32_t * size);
+ ErrorCode tsfile_reader_get_table_schema_checked(
+ TsFileReader reader, const char * table_name, TableSchema * out_schema);
+ ErrorCode tsfile_reader_get_all_timeseries_schemas_checked(
+ TsFileReader reader, DeviceSchema ** out_schemas, uint32_t * out_size);
+
void tsfile_device_id_free_contents(DeviceID * d)
ErrorCode tsfile_reader_get_all_devices(TsFileReader reader,
diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx
index c8f155ae8..8290add50 100644
--- a/python/tsfile/tsfile_py_cpp.pyx
+++ b/python/tsfile/tsfile_py_cpp.pyx
@@ -1236,7 +1236,11 @@ cdef ResultSet tsfile_reader_query_table_with_tag_filter_c(TsFileReader reader,
cdef object get_table_schema(TsFileReader reader, object table_name):
cdef bytes table_name_bytes = PyUnicode_AsUTF8String(table_name)
cdef const char * table_name_c = table_name_bytes
- cdef TableSchema schema = tsfile_reader_get_table_schema(reader, table_name_c)
+ cdef TableSchema schema
+ cdef ErrorCode code = 0
+ code = tsfile_reader_get_table_schema_checked(
+ reader, table_name_c, &schema)
+ check_error(code)
return from_c_table_schema(schema)
cdef object get_all_table_schema(TsFileReader reader):
@@ -1257,10 +1261,13 @@ cdef object get_all_table_schema(TsFileReader reader):
cdef object get_all_timeseries_schema(TsFileReader reader):
cdef uint32_t device_num = 0
cdef DeviceSchema * schemas
+ cdef ErrorCode error_code = 0
cdef int i
device_schemas = {}
- schemas = tsfile_reader_get_all_timeseries_schemas(reader, &device_num)
+ error_code = tsfile_reader_get_all_timeseries_schemas_checked(
+ reader, &schemas, &device_num)
+ check_error(error_code)
for i in range(device_num):
schema_py = from_c_device_schema(schemas[i])
device_schemas.update([(schema_py.get_device_name(), schema_py)])
@@ -1423,6 +1430,7 @@ cdef public api object reader_get_timeseries_metadata_c(TsFileReader reader,
cdef DeviceID* q = NULL
cdef uint32_t qlen = 0
cdef uint32_t i
+ cdef uint32_t j
cdef int err
cdef bytes bpath
cdef const char* raw
@@ -1447,18 +1455,30 @@ cdef public api object reader_get_timeseries_metadata_c(TsFileReader reader,
path_s = dev.path
except AttributeError:
path_s = str(dev)
- bpath = path_s.encode('utf-8')
- raw = PyBytes_AsString(bpath)
- q[i].path = strdup(raw)
- if q[i].path == NULL:
- raise MemoryError()
+ if path_s is not None:
+ bpath = path_s.encode('utf-8')
+ raw = PyBytes_AsString(bpath)
+ q[i].path = strdup(raw)
+ if q[i].path == NULL:
+ raise MemoryError()
+ segments = getattr(dev, 'segments', ())
+ if segments:
+ q[i].segment_count = <uint32_t>len(segments)
+ q[i].segments = <char**>malloc(sizeof(char*) * len(segments))
+ if q[i].segments == NULL:
+ raise MemoryError()
+ memset(q[i].segments, 0, sizeof(char*) * len(segments))
+ for j in range(q[i].segment_count):
+ if segments[j] is not None:
+ bpath = segments[j].encode('utf-8')
+ q[i].segments[j] = strdup(PyBytes_AsString(bpath))
+ if q[i].segments[j] == NULL:
+ raise MemoryError()
err = tsfile_reader_get_timeseries_metadata_for_devices(
reader, q, qlen, &mmap)
check_error(err)
finally:
- for i in range(qlen):
- free(q[i].path)
- free(q)
+ tsfile_free_device_id_array(q, qlen)
try:
return device_timeseries_metadata_map_to_py(&mmap)
finally:
diff --git a/python/tsfile/tsfile_reader.pyx b/python/tsfile/tsfile_reader.pyx
index f6509a4b7..c7f0f1a61 100644
--- a/python/tsfile/tsfile_reader.pyx
+++ b/python/tsfile/tsfile_reader.pyx
@@ -391,8 +391,10 @@ cdef class TsFileReaderPy:
"""
cdef ResultSet result
cdef TagFilterHandle c_tag_filter = NULL
+ pyresult = ResultSetPy(self)
if tag_filter is not None:
c_tag_filter = self._build_c_tag_filter(table_name.lower(), tag_filter)
+ pyresult._tag_filter_handle = c_tag_filter
if batch_size <= 0:
result = tsfile_reader_query_table_with_tag_filter_c(
self.reader, table_name.lower(),
@@ -403,8 +405,6 @@ cdef class TsFileReaderPy:
self.reader, table_name.lower(),
[column_name.lower() for column_name in column_names],
start_time, end_time, c_tag_filter, batch_size)
- pyresult = ResultSetPy(self)
- pyresult._tag_filter_handle = c_tag_filter
pyresult.init_c(result, table_name)
self.activate_result_set_list.add(pyresult)
return pyresult
@@ -413,6 +413,9 @@ cdef class TsFileReaderPy:
"""Recursively build C TagFilterHandle from Python TagFilter tree."""
cdef ErrorCode code = 0
cdef TagFilterHandle handle = NULL
+ cdef TagFilterHandle left = NULL
+ cdef TagFilterHandle right = NULL
+ cdef TagFilterHandle inner = NULL
cdef bytes table_bytes
cdef bytes col_bytes
cdef bytes val_bytes
@@ -441,16 +444,43 @@ cdef class TsFileReaderPy:
check_error(code)
return handle
elif isinstance(tag_filter, AndTagFilter):
- left = self._build_c_tag_filter(table_name, tag_filter.left)
- right = self._build_c_tag_filter(table_name, tag_filter.right)
- return tsfile_tag_filter_and(left, right)
+ try:
+ left = self._build_c_tag_filter(table_name, tag_filter.left)
+ right = self._build_c_tag_filter(table_name, tag_filter.right)
+ handle = tsfile_tag_filter_and(left, right)
+ if handle == NULL:
+ raise MemoryError("Unable to allocate compound tag filter")
+ # The compound filter now owns both children.
+ left = NULL
+ right = NULL
+ return handle
+ finally:
+ tsfile_tag_filter_free(left)
+ tsfile_tag_filter_free(right)
elif isinstance(tag_filter, OrTagFilter):
- left = self._build_c_tag_filter(table_name, tag_filter.left)
- right = self._build_c_tag_filter(table_name, tag_filter.right)
- return tsfile_tag_filter_or(left, right)
+ try:
+ left = self._build_c_tag_filter(table_name, tag_filter.left)
+ right = self._build_c_tag_filter(table_name, tag_filter.right)
+ handle = tsfile_tag_filter_or(left, right)
+ if handle == NULL:
+ raise MemoryError("Unable to allocate compound tag filter")
+ # The compound filter now owns both children.
+ left = NULL
+ right = NULL
+ return handle
+ finally:
+ tsfile_tag_filter_free(left)
+ tsfile_tag_filter_free(right)
elif isinstance(tag_filter, NotTagFilter):
- inner = self._build_c_tag_filter(table_name, tag_filter.filter)
- return tsfile_tag_filter_not(inner)
+ try:
+ inner = self._build_c_tag_filter(table_name, tag_filter.filter)
+ handle = tsfile_tag_filter_not(inner)
+ if handle == NULL:
+ raise MemoryError("Unable to allocate compound tag filter")
+ inner = NULL
+ return handle
+ finally:
+ tsfile_tag_filter_free(inner)
else:
raise TypeError(f"Unknown tag filter type: {type(tag_filter)}")
def query_table_on_tree(self, column_names : List[str],
@@ -494,13 +524,13 @@ cdef class TsFileReaderPy:
"""
cdef ResultSet result
cdef TagFilterHandle c_tag_filter = NULL
+ pyresult = ResultSetPy(self)
if tag_filter is not None:
c_tag_filter = self._build_c_tag_filter(table_name.lower(), tag_filter)
+ pyresult._tag_filter_handle = c_tag_filter
result = tsfile_reader_query_table_by_row_c(self.reader, table_name.lower(),
[column_name.lower() for column_name in column_names],
offset, limit, c_tag_filter, batch_size)
- pyresult = ResultSetPy(self)
- pyresult._tag_filter_handle = c_tag_filter
pyresult.init_c(result, table_name)
self.activate_result_set_list.add(pyresult)
return pyresult