This is an automated email from the ASF dual-hosted git repository. colinlee pushed a commit to branch colin_support_read_tree in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit c5d141089a9f08f9cdfa88c0cd166b72ad36e91c Author: ColinLee <[email protected]> AuthorDate: Sun Nov 30 12:03:38 2025 +0800 support get all timeseries. --- cpp/src/cwrapper/tsfile_cwrapper.cc | 80 ++++++++++++------ cpp/src/cwrapper/tsfile_cwrapper.h | 9 ++ python/tests/test_write_and_read.py | 70 ++++++++++++++- python/tsfile/schema.py | 6 ++ python/tsfile/tsfile_cpp.pxd | 2 + python/tsfile/tsfile_py_cpp.pxd | 2 +- python/tsfile/tsfile_py_cpp.pyx | 164 +++++++++++++++++++++--------------- python/tsfile/tsfile_reader.pyx | 10 +-- 8 files changed, 239 insertions(+), 104 deletions(-) diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc index 0b7585b9..8a8f462b 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.cc +++ b/cpp/src/cwrapper/tsfile_cwrapper.cc @@ -546,34 +546,58 @@ TableSchema* tsfile_reader_get_all_table_schemas(TsFileReader reader, return ret; } -// TimeseriesSchema* tsfile_reader_get_all_timeseries_schemas(TsFileReader -// reader, -// uint32_t* size) { -// auto* r = static_cast<storage::TsFileReader*>(reader); -// auto table_schemas = r->get_all_devices(); -// 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; -// } +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; + } + *size = static_cast<uint32_t>(device_ids.size()); + if (device_ids.empty()) { + return nullptr; + } + + DeviceSchema* device_schema = static_cast<DeviceSchema*>( + malloc(sizeof(DeviceSchema) * device_ids.size())); + if (device_schema == nullptr) { + *size = 0; + return nullptr; + } + + 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; + + std::vector<storage::MeasurementSchema> schemas; + int ret = r->get_timeseries_schema(device_id, schemas); + if (ret != common::E_OK || schemas.empty()) { + continue; + } + + 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_); + } + } + return device_schema; +} // delete pointer void _free_tsfile_ts_record(TsRecord* record) { diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h index f94325aa..32f85aa4 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.h +++ b/cpp/src/cwrapper/tsfile_cwrapper.h @@ -576,6 +576,15 @@ TableSchema tsfile_reader_get_table_schema(TsFileReader reader, TableSchema* tsfile_reader_get_all_table_schemas(TsFileReader reader, uint32_t* size); +/** + * @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. + */ +DeviceSchema* tsfile_reader_get_all_timeseries_schemas(TsFileReader reader, + uint32_t* size); + // Close and free resource. void free_tablet(Tablet* tablet); void free_tsfile_result_set(ResultSet* result_set); diff --git a/python/tests/test_write_and_read.py b/python/tests/test_write_and_read.py index cf163c1f..e22f31c0 100644 --- a/python/tests/test_write_and_read.py +++ b/python/tests/test_write_and_read.py @@ -21,6 +21,8 @@ import os import numpy as np import pandas as pd import pytest +from pandas.core.dtypes.common import is_integer_dtype + from tsfile import ColumnSchema, TableSchema, TSEncoding from tsfile import Compressor from tsfile import TSDataType @@ -211,6 +213,70 @@ def test_tree_query_to_dataframe_variants(): os.remove(file_path) +def test_get_all_timeseries_schemas(): + file_path = "get_all_timeseries_schema.tsfile" + device_ids = [ + "root.db1.t1", + "root.db2.t1", + "root.db3.t2.t3", + "root.db3.t3", + "device", + "device.ln", + "device2.ln1.tmp", + "device3.ln2.tmp.v1.v2", + "device3.ln2.tmp.v1.v3", + ] + measurement_ids1 = ["temperature", "hudi", "level"] + measurement_ids2 = ["level", "vol"] + rows_per_device = 2 + + try: + writer = TsFileWriter(file_path) + for idx, device_id in enumerate(device_ids): + measurements = measurement_ids1 if idx % 2 == 0 else measurement_ids2 + for measurement in measurements: + writer.register_timeseries( + device_id, TimeseriesSchema(measurement, TSDataType.INT32) + ) + for ts in range(rows_per_device): + fields = [] + for measurement in measurements: + fields.append( + Field( + measurement, + idx * 100 + ts * 10 + len(fields), + TSDataType.INT32, + ) + ) + writer.write_row_record(RowRecord(device_id, ts, fields)) + writer.close() + + reader = TsFileReader(file_path) + device_schema_map = reader.get_all_timeseries_schemas() + expected_devices = {device_id.lower() for device_id in device_ids} + assert set(device_schema_map.keys()) == expected_devices + print(device_schema_map) + + for idx, device_id in enumerate(device_ids): + measurements = measurement_ids1 if idx % 2 == 0 else measurement_ids2 + normalized_device = device_id.lower() + assert normalized_device in device_schema_map + device_schema = device_schema_map[normalized_device] + assert device_schema.get_device_name() == normalized_device + timeseries_list = device_schema.get_timeseries_list() + assert len(timeseries_list) == len(measurements) + actual_measurements = { + ts_schema.get_timeseries_name() for ts_schema in timeseries_list + } + assert actual_measurements == {m.lower() for m in measurements} + for ts_schema in timeseries_list: + assert ts_schema.get_data_type() == TSDataType.INT32 + reader.close() + finally: + if os.path.exists(file_path): + os.remove(file_path) + + @pytest.mark.skip(reason="API not match") def test_tablet_write_and_read(): try: @@ -439,9 +505,9 @@ def test_tsfile_to_df(): df1 = to_dataframe("table_write_to_df.tsfile") assert df1.shape == (4097, 4) assert df1["value2"].sum() == 100 * (1 + 4096) / 2 * 4096 - assert df1["time"].dtype == np.int64 + assert is_integer_dtype(df1["time"]) assert df1["value"].dtype == np.float64 - assert df1["value2"].dtype == np.int64 + assert is_integer_dtype(df1["value2"]) df2 = to_dataframe("table_write_to_df.tsfile", column_names=["device", "value2"]) assert df2.shape == (4097, 3) assert df1["value2"].equals(df2["value2"]) diff --git a/python/tsfile/schema.py b/python/tsfile/schema.py index ae7960c5..3aa1313c 100644 --- a/python/tsfile/schema.py +++ b/python/tsfile/schema.py @@ -49,6 +49,10 @@ class TimeseriesSchema: def get_compression_type(self): return self.compression_type + def __repr__(self): + return f"TimeseriesSchema({self.timeseries_name}, {self.data_type.name}, {self.encoding_type.name}, {self.compression_type.name})" + + class DeviceSchema: """Represents a device entity containing multiple time series.""" @@ -66,6 +70,8 @@ class DeviceSchema: def get_timeseries_list(self): return self.timeseries_list + def __repr__(self): + return f"DeviceSchema({self.device_name}, {self.timeseries_list})" class ColumnSchema: """Defines schema for a table column (name, datatype, category).""" diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index e9a67364..d35be96e 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -183,6 +183,8 @@ cdef extern from "./tsfile_cwrapper.h": TableSchema * tsfile_reader_get_all_table_schemas(TsFileReader reader, uint32_t * size); + DeviceSchema * tsfile_reader_get_all_timeseries_schemas(TsFileReader reader, + uint32_t * size); # resultSet : get data from resultSet bint tsfile_result_set_next(ResultSet result_set, ErrorCode * err_code); diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd index 584ff0b1..e44bb588 100644 --- a/python/tsfile/tsfile_py_cpp.pxd +++ b/python/tsfile/tsfile_py_cpp.pxd @@ -55,6 +55,6 @@ cdef public api ResultSet tsfile_reader_query_paths_c(TsFileReader reader, objec int64_t end_time) cdef public api object get_table_schema(TsFileReader reader, object table_name) cdef public api object get_all_table_schema(TsFileReader reader) -# cdef public api object get_all_timeseries_schema(TsFileReader reader) +cdef public api object get_all_timeseries_schema(TsFileReader reader) cpdef public api object get_tsfile_config() cpdef public api void set_tsfile_config(dict new_config) \ No newline at end of file diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index 1d7af385..7a9aa889 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -15,7 +15,6 @@ # specific language governing permissions and limitations # under the License. # - #cython: language_level=3 from .tsfile_cpp cimport * @@ -32,9 +31,10 @@ from tsfile.schema import ResultSetMetaData as ResultSetMetaDataPy from tsfile.schema import TSDataType as TSDataTypePy, TSEncoding as TSEncodingPy from tsfile.schema import Compressor as CompressorPy, ColumnCategory as CategoryPy from tsfile.schema import TableSchema as TableSchemaPy, ColumnSchema as ColumnSchemaPy +from tsfile.schema import DeviceSchema as DeviceSchemaPy, TimeseriesSchema as TimeseriesSchemaPy # check exception and set py exception object -cdef inline void check_error(int errcode, const char* context=NULL) except *: +cdef inline void check_error(int errcode, const char * context=NULL) except*: cdef: object exc_type object exc_instance @@ -74,7 +74,21 @@ cdef object from_c_table_schema(TableSchema schema): free_c_table_schema(&schema) return TableSchemaPy(table_name, columns) +cdef object from_c_timeseries_schema(TimeseriesSchema schema): + timeseries_name = schema.timeseries_name.decode('utf-8') + data_type = TSDataTypePy(schema.data_type) + encoding = TSEncodingPy(schema.encoding) + compression = CompressorPy(schema.compression) + return TimeseriesSchemaPy(timeseries_name, data_type, encoding, compression) +cdef object from_c_device_schema(DeviceSchema schema): + cdef int i + device_name = schema.device_name.decode('utf-8') + timeseries = [] + for i in range(schema.timeseries_num): + timeseries.append(from_c_timeseries_schema(schema.timeseries_schema[i])) + free_c_device_schema(&schema) + return DeviceSchemaPy(device_name, timeseries) # Convert from python to c struct cdef dict TS_DATA_TYPE_MAP = { @@ -128,7 +142,6 @@ cdef ColumnCategory to_c_category_type(object category): except KeyError: raise ValueError(f"Unsupported Python Column Category: {category}") - cdef TSEncoding to_c_encoding_type(object encoding_type): try: return TS_ENCODING_MAP[encoding_type] @@ -141,8 +154,8 @@ cdef CompressionType to_c_compression_type(object compression_type): except KeyError: raise ValueError(f"Unsupported Python Compressor: {compression_type}") -cdef TimeseriesSchema* to_c_timeseries_schema(object py_schema): - cdef TimeseriesSchema* c_schema +cdef TimeseriesSchema * to_c_timeseries_schema(object py_schema): + cdef TimeseriesSchema * c_schema c_schema = <TimeseriesSchema *> malloc(sizeof(TimeseriesSchema)) c_schema.timeseries_name = strdup(py_schema.timeseries_name.encode('utf-8')) if py_schema.data_type is not None: @@ -159,30 +172,30 @@ cdef TimeseriesSchema* to_c_timeseries_schema(object py_schema): raise ValueError("compression_type cannot be None") return c_schema - -cdef DeviceSchema* to_c_device_schema(object py_schema): - cdef DeviceSchema* c_schema +cdef DeviceSchema * to_c_device_schema(object py_schema): + cdef DeviceSchema * c_schema c_schema = <DeviceSchema *> malloc(sizeof(DeviceSchema)) c_schema.device_name = strdup(py_schema.device_name.encode('utf-8')) c_schema.timeseries_num = len(py_schema.timeseries_list) c_schema.timeseries_schema = <TimeseriesSchema *> malloc(c_schema.timeseries_num * sizeof(TimeseriesSchema)) for i in range(c_schema.timeseries_num): - c_schema.timeseries_schema[i].timeseries_name = strdup(py_schema.timeseries_list[i].timeseries_name.encode('utf-8')) + c_schema.timeseries_schema[i].timeseries_name = strdup( + py_schema.timeseries_list[i].timeseries_name.encode('utf-8')) c_schema.timeseries_schema[i].data_type = to_c_data_type(py_schema.timeseries_list[i].data_type) c_schema.timeseries_schema[i].encoding = to_c_encoding_type(py_schema.timeseries_list[i].encoding_type) c_schema.timeseries_schema[i].compression = to_c_compression_type(py_schema.timeseries_list[i].compression_type) return c_schema -cdef ColumnSchema* to_c_column_schema(object py_schema): - cdef ColumnSchema* c_schema - c_schema = <ColumnSchema*> malloc(sizeof(ColumnSchema)) +cdef ColumnSchema * to_c_column_schema(object py_schema): + cdef ColumnSchema * c_schema + c_schema = <ColumnSchema *> malloc(sizeof(ColumnSchema)) c_schema.data_type = to_c_data_type(py_schema.data_type) c_schema.column_category = py_schema.category c_schema.column_name = strdup(py_schema.column_name.encode('utf-8')) return c_schema -cdef TableSchema* to_c_table_schema(object py_schema): - cdef TableSchema* c_schema +cdef TableSchema * to_c_table_schema(object py_schema): + cdef TableSchema * c_schema c_schema = <TableSchema *> malloc(sizeof(TableSchema)) c_schema.table_name = strdup(py_schema.table_name.encode('utf-8')) c_schema.column_num = len(py_schema.columns) @@ -193,7 +206,6 @@ cdef TableSchema* to_c_table_schema(object py_schema): c_schema.column_schemas[i].data_type = to_c_data_type(py_schema.columns[i].data_type) return c_schema - cdef Tablet to_c_tablet(object tablet): cdef Tablet ctablet cdef int max_row_num @@ -202,7 +214,7 @@ cdef Tablet to_c_tablet(object tablet): cdef bytes device_id_bytes cdef const char * device_id_c cdef char** columns_names - cdef TSDataType* column_types + cdef TSDataType * column_types cdef bytes row_bytes cdef const char *row_str @@ -222,7 +234,7 @@ cdef Tablet to_c_tablet(object tablet): max_row_num = tablet.get_max_row_num() ctablet = _tablet_new_with_target_name(device_id_c, columns_names, columns_types, column_num, - max_row_num) + max_row_num) free(columns_types) for i in range(column_num): free(columns_names[i]) @@ -275,15 +287,13 @@ cdef Tablet to_c_tablet(object tablet): row_str = PyBytes_AsString(row_bytes) tablet_add_value_by_index_string(ctablet, row, col, row_str) - return ctablet - cdef TsRecord to_c_record(object row_record): cdef int field_num = row_record.get_fields_num() - cdef int64_t timestamp = <int64_t>row_record.get_timestamp() + cdef int64_t timestamp = <int64_t> row_record.get_timestamp() cdef bytes device_id_bytes = PyUnicode_AsUTF8String(row_record.get_device_id()) - cdef const char* device_id = device_id_bytes + cdef const char * device_id = device_id_bytes cdef TsRecord record cdef int i cdef TSDataType data_type @@ -292,32 +302,37 @@ cdef TsRecord to_c_record(object row_record): field = row_record.get_fields()[i] data_type = to_c_data_type(field.get_data_type()) if data_type == TS_DATATYPE_BOOLEAN: - _insert_data_into_ts_record_by_name_bool(record, PyUnicode_AsUTF8(field.get_field_name()), field.get_bool_value()) + _insert_data_into_ts_record_by_name_bool(record, PyUnicode_AsUTF8(field.get_field_name()), + field.get_bool_value()) elif data_type == TS_DATATYPE_INT32: - _insert_data_into_ts_record_by_name_int32_t(record, PyUnicode_AsUTF8(field.get_field_name()), field.get_int_value()) + _insert_data_into_ts_record_by_name_int32_t(record, PyUnicode_AsUTF8(field.get_field_name()), + field.get_int_value()) elif data_type == TS_DATATYPE_INT64: - _insert_data_into_ts_record_by_name_int64_t(record, PyUnicode_AsUTF8(field.get_field_name()), field.get_long_value()) + _insert_data_into_ts_record_by_name_int64_t(record, PyUnicode_AsUTF8(field.get_field_name()), + field.get_long_value()) elif data_type == TS_DATATYPE_DOUBLE: - _insert_data_into_ts_record_by_name_double(record, PyUnicode_AsUTF8(field.get_field_name()), field.get_double_value()) + _insert_data_into_ts_record_by_name_double(record, PyUnicode_AsUTF8(field.get_field_name()), + field.get_double_value()) elif data_type == TS_DATATYPE_FLOAT: - _insert_data_into_ts_record_by_name_float(record, PyUnicode_AsUTF8(field.get_field_name()), field.get_float_value()) + _insert_data_into_ts_record_by_name_float(record, PyUnicode_AsUTF8(field.get_field_name()), + field.get_float_value()) return record # Free c structs' space -cdef void free_c_table_schema(TableSchema* c_schema): +cdef void free_c_table_schema(TableSchema * c_schema): free(c_schema.table_name) for i in range(c_schema.column_num): free_c_column_schema(&(c_schema.column_schemas[i])) free(c_schema.column_schemas) -cdef void free_c_column_schema(ColumnSchema* c_schema): +cdef void free_c_column_schema(ColumnSchema * c_schema): free(c_schema.column_name) -cdef void free_c_timeseries_schema(TimeseriesSchema* c_schema): +cdef void free_c_timeseries_schema(TimeseriesSchema * c_schema): free(c_schema.timeseries_name) -cdef void free_c_device_schema(DeviceSchema* c_schema): +cdef void free_c_device_schema(DeviceSchema * c_schema): free(c_schema.device_name) for i in range(c_schema.timeseries_num): free_c_timeseries_schema(&(c_schema.timeseries_schema[i])) @@ -334,7 +349,7 @@ cdef TsFileWriter tsfile_writer_new_c(object pathname, uint64_t memory_threshold cdef ErrorCode errno = 0 cdef TsFileWriter writer cdef bytes encoded_path = PyUnicode_AsUTF8String(pathname) - cdef const char* c_path = encoded_path + cdef const char * c_path = encoded_path writer = _tsfile_writer_new(c_path, memory_threshold, &errno) check_error(errno) return writer @@ -343,8 +358,8 @@ cdef TsFileReader tsfile_reader_new_c(object pathname) except +: cdef ErrorCode errno = 0 cdef TsFileReader reader cdef bytes encoded_path = PyUnicode_AsUTF8String(pathname) - cdef const char* c_path = encoded_path - reader = tsfile_reader_new(c_path, &errno) + cdef const char * c_path = encoded_path + reader = tsfile_reader_new(c_path, &errno) check_error(errno) return reader @@ -356,13 +371,13 @@ cpdef object get_tsfile_config(): "page_writer_max_memory_bytes_": g_config_value_.page_writer_max_memory_bytes_, "max_degree_of_index_node_": g_config_value_.max_degree_of_index_node_, "tsfile_index_bloom_filter_error_percent_": g_config_value_.tsfile_index_bloom_filter_error_percent_, - "time_encoding_type_":TSEncodingPy(int(g_config_value_.time_encoding_type_)), + "time_encoding_type_": TSEncodingPy(int(g_config_value_.time_encoding_type_)), "time_data_type_": TSDataTypePy(int(g_config_value_.time_data_type_)), "time_compress_type_": CompressorPy(int(g_config_value_.time_compress_type_)), "chunk_group_size_threshold_": g_config_value_.chunk_group_size_threshold_, - "record_count_for_next_mem_check_":g_config_value_.record_count_for_next_mem_check_, - "encrypt_flag_":g_config_value_.encrypt_flag_, - "boolean_encoding_type_":TSEncodingPy(int(g_config_value_.boolean_encoding_type_)), + "record_count_for_next_mem_check_": g_config_value_.record_count_for_next_mem_check_, + "encrypt_flag_": g_config_value_.encrypt_flag_, + "boolean_encoding_type_": TSEncodingPy(int(g_config_value_.boolean_encoding_type_)), "int32_encoding_type_": TSEncodingPy(int(g_config_value_.int32_encoding_type_)), "int64_encoding_type_": TSEncodingPy(int(g_config_value_.int64_encoding_type_)), "float_encoding_type_": TSEncodingPy(int(g_config_value_.float_encoding_type_)), @@ -371,7 +386,6 @@ cpdef object get_tsfile_config(): "default_compression_type_": CompressorPy(int(g_config_value_.default_compression_type_)), } - cpdef void set_tsfile_config(dict new_config): if "tsblock_mem_inc_step_size_" in new_config: _check_uint32(new_config["tsblock_mem_inc_step_size_"]) @@ -390,21 +404,22 @@ cpdef void set_tsfile_config(dict new_config): g_config_value_.max_degree_of_index_node_ = new_config["max_degree_of_index_node_"] if "tsfile_index_bloom_filter_error_percent_" in new_config: _check_double(new_config["tsfile_index_bloom_filter_error_percent_"]) - g_config_value_.tsfile_index_bloom_filter_error_percent_ = new_config["tsfile_index_bloom_filter_error_percent_"] + g_config_value_.tsfile_index_bloom_filter_error_percent_ = new_config[ + "tsfile_index_bloom_filter_error_percent_"] if "time_encoding_type_" in new_config: if not isinstance(new_config["time_encoding_type_"], TSEncodingPy): raise TypeError(f"Unsupported TSEncoding: {new_config['time_encoding_type_']}") - code = set_global_time_encoding(<uint8_t>(new_config["time_encoding_type_"].value)) + code = set_global_time_encoding(<uint8_t> (new_config["time_encoding_type_"].value)) check_error(code) if "time_data_type_" in new_config: if not isinstance(new_config["time_data_type_"], TSDataTypePy): raise TypeError(f"Unsupported TSDataType: {new_config['time_data_type_']}") - code = set_global_time_data_type(<uint8_t>(new_config["time_data_type_"].value)) + code = set_global_time_data_type(<uint8_t> (new_config["time_data_type_"].value)) check_error(code) if "time_compress_type_" in new_config: if not isinstance(new_config["time_compress_type_"], CompressorPy): raise TypeError(f"Unsupported Compressor: {new_config['time_compress_type_']}") - code = set_global_time_compression(<uint8_t>(new_config["time_compress_type_"].value)) + code = set_global_time_compression(<uint8_t> (new_config["time_compress_type_"].value)) check_error(code) if "chunk_group_size_threshold_" in new_config: _check_uint32(new_config["chunk_group_size_threshold_"]) @@ -414,7 +429,7 @@ cpdef void set_tsfile_config(dict new_config): g_config_value_.record_count_for_next_mem_check_ = new_config["record_count_for_next_mem_check_"] if "encrypt_flag_" in new_config: _check_bool(new_config["encrypt_flag_"]) - g_config_value_.encrypt_flag_ = <bint>new_config["encrypt_flag_"] + g_config_value_.encrypt_flag_ = <bint> new_config["encrypt_flag_"] if "boolean_encoding_type_" in new_config: if not isinstance(new_config["boolean_encoding_type_"], TSEncodingPy): @@ -474,7 +489,7 @@ cdef ErrorCode tsfile_writer_register_timeseries_py_cpp(TsFileWriter writer, obj TimeseriesSchema *schema): cdef ErrorCode errno cdef bytes encoded_device_name = PyUnicode_AsUTF8String(device_name) - cdef const char* c_device_name = encoded_device_name + cdef const char * c_device_name = encoded_device_name errno = _tsfile_writer_register_timeseries(writer, c_device_name, schema) return errno @@ -485,48 +500,48 @@ cdef ErrorCode tsfile_writer_register_table_py_cpp(TsFileWriter writer, TableSch cdef bint tsfile_result_set_is_null_by_name_c(ResultSet result_set, object name): cdef bytes encoded_name = PyUnicode_AsUTF8String(name) - cdef const char* c_name = encoded_name + cdef const char * c_name = encoded_name return tsfile_result_set_is_null_by_name(result_set, c_name) cdef ResultSet tsfile_reader_query_table_c(TsFileReader reader, object table_name, object column_list, - int64_t start_time, int64_t end_time): + int64_t start_time, int64_t end_time): cdef ResultSet result cdef int column_num = len(column_list) cdef bytes table_name_bytes = PyUnicode_AsUTF8String(table_name) - cdef const char* table_name_c = table_name_bytes - cdef char** columns = <char**> malloc(sizeof(char*) * column_num) + cdef const char * table_name_c = table_name_bytes + cdef char** columns = <char**> malloc(sizeof(char *) * column_num) cdef int i cdef ErrorCode code = 0 if columns == NULL: raise MemoryError("Failed to allocate memory for columns") try: for i in range(column_num): - columns[i] = strdup((<str>column_list[i]).encode('utf-8')) + columns[i] = strdup((<str> column_list[i]).encode('utf-8')) if columns[i] == NULL: raise MemoryError("Failed to allocate memory for column name") - result = tsfile_query_table(reader, table_name_c, columns, column_num, start_time, end_time, &code) + result = tsfile_query_table(reader, table_name_c, columns, column_num, start_time, end_time, &code) check_error(code) return result finally: if columns != NULL: for i in range(column_num): - free(<void*>columns[i]) + free(<void *> columns[i]) columns[i] = NULL - free(<void*>columns) + free(<void *> columns) columns = NULL cdef ResultSet tsfile_reader_query_table_on_tree_c(TsFileReader reader, object column_list, - int64_t start_time, int64_t end_time): + int64_t start_time, int64_t end_time): cdef ResultSet result cdef int column_num = len(column_list) - cdef char** columns = <char**> malloc(sizeof(char*) * column_num) + cdef char** columns = <char**> malloc(sizeof(char *) * column_num) cdef int i cdef ErrorCode code = 0 if columns == NULL: raise MemoryError("Failed to allocate memory for columns") try: for i in range(column_num): - columns[i] = strdup((<str>column_list[i]).encode('utf-8')) + columns[i] = strdup((<str> column_list[i]).encode('utf-8')) if columns[i] == NULL: raise MemoryError("Failed to allocate memory for column name") result = tsfile_query_table_on_tree(reader, columns, column_num, start_time, end_time, &code) @@ -535,49 +550,50 @@ cdef ResultSet tsfile_reader_query_table_on_tree_c(TsFileReader reader, object c finally: if columns != NULL: for i in range(column_num): - free(<void*>columns[i]) + free(<void *> columns[i]) columns[i] = NULL - free(<void*>columns) + free(<void *> columns) columns = NULL - -cdef ResultSet tsfile_reader_query_paths_c(TsFileReader reader, object device_name, object sensor_list, int64_t start_time, - int64_t end_time): +cdef ResultSet tsfile_reader_query_paths_c(TsFileReader reader, object device_name, object sensor_list, + int64_t start_time, + int64_t end_time): cdef ResultSet result cdef int path_num = len(sensor_list) - cdef char** sensor_list_c = <char**> malloc(sizeof(char*) * path_num) + cdef char** sensor_list_c = <char**> malloc(sizeof(char *) * path_num) cdef bytes device_name_bytes = PyUnicode_AsUTF8String(device_name) - cdef const char* device_name_c = device_name_bytes + cdef const char * device_name_c = device_name_bytes cdef int i cdef ErrorCode code = 0 if sensor_list_c == NULL: raise MemoryError("Failed to allocate memory for paths") try: for i in range(path_num): - sensor_list_c[i] = strdup((<str>sensor_list[i]).encode('utf-8')) + sensor_list_c[i] = strdup((<str> sensor_list[i]).encode('utf-8')) if sensor_list_c[i] == NULL: raise MemoryError("Failed to allocate memory for path") - result = _tsfile_reader_query_device(reader, device_name_c, sensor_list_c, path_num, start_time, end_time, &code) + result = _tsfile_reader_query_device(reader, device_name_c, sensor_list_c, path_num, start_time, end_time, + &code) check_error(code) return result finally: if sensor_list_c != NULL: for i in range(path_num): if sensor_list_c[i] != NULL: - free(<void*>sensor_list_c[i]) + free(<void *> sensor_list_c[i]) sensor_list_c[i] = NULL - free(<void*>sensor_list_c) + free(<void *> sensor_list_c) sensor_list_c = NULL 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 const char * table_name_c = table_name_bytes cdef TableSchema schema = tsfile_reader_get_table_schema(reader, table_name_c) return from_c_table_schema(schema) cdef object get_all_table_schema(TsFileReader reader): cdef uint32_t table_num = 0 - cdef TableSchema* schemas + cdef TableSchema * schemas cdef int i table_schemas = {} @@ -588,3 +604,15 @@ cdef object get_all_table_schema(TsFileReader reader): free(schemas) return table_schemas +cdef object get_all_timeseries_schema(TsFileReader reader): + cdef uint32_t device_num = 0 + cdef DeviceSchema * schemas + cdef int i + + device_schemas = {} + schemas = tsfile_reader_get_all_timeseries_schemas(reader, &device_num) + for i in range(device_num): + schema_py = from_c_device_schema(schemas[i]) + device_schemas.update([(schema_py.get_device_name(), schema_py)]) + free(schemas) + return device_schemas diff --git a/python/tsfile/tsfile_reader.pyx b/python/tsfile/tsfile_reader.pyx index f8ed6adc..eef0add8 100644 --- a/python/tsfile/tsfile_reader.pyx +++ b/python/tsfile/tsfile_reader.pyx @@ -339,11 +339,11 @@ cdef class TsFileReaderPy: """ return get_all_table_schema(self.reader) - # def get_all_timeseries_schemas(self): - # """ - # Get all timeseries schemas - # """ - # return get_all_timeseries_schema(self.reader) + def get_all_timeseries_schemas(self): + """ + Get all timeseries schemas + """ + return get_all_timeseries_schema(self.reader) def close(self): """
