This is an automated email from the ASF dual-hosted git repository. ColinLeeo pushed a commit to branch codex/python-error-code-alignment in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 514de3d70216222a3860831582bfa6a7399f4205 Author: ColinLee <[email protected]> AuthorDate: Mon Jul 20 17:43:13 2026 +0800 fix(python): align native error handling --- cpp/src/cwrapper/errno_define_c.h | 67 +++-- cpp/test/cwrapper/c_release_test.cc | 3 +- python/tests/test_exceptions.py | 106 +++++++ python/tests/test_write_and_read.py | 8 +- python/tsfile/exceptions.py | 163 ++++++++++- python/tsfile/tsfile_cpp.pxd | 5 + python/tsfile/tsfile_py_cpp.pxd | 4 +- python/tsfile/tsfile_py_cpp.pyx | 550 +++++++++++++++++++++++++----------- python/tsfile/tsfile_reader.pyx | 4 +- python/tsfile/utils.py | 8 +- 10 files changed, 712 insertions(+), 206 deletions(-) diff --git a/cpp/src/cwrapper/errno_define_c.h b/cpp/src/cwrapper/errno_define_c.h index 3ceb06416..af9cd1ca1 100644 --- a/cpp/src/cwrapper/errno_define_c.h +++ b/cpp/src/cwrapper/errno_define_c.h @@ -17,8 +17,8 @@ * under the License. */ -#ifndef CWRAPPER_ERRNO_DEFINRET_H -#define CWRAPPER_ERRNO_DEFINRET_H +#ifndef CWRAPPER_ERRNO_DEFINE_H +#define CWRAPPER_ERRNO_DEFINE_H #define RET_OK 0 #define RET_OOM 1 @@ -35,42 +35,65 @@ #define RET_NET_FCNTL_ERR 13 #define RET_NET_LISTEN_ERR 14 #define RET_NET_SEND_ERR 15 -#define RET_PIPRET_ERR 16 -#define RET_THREAD_CREATRET_ERR 17 +#define RET_PIPE_ERR 16 +#define RET_THREAD_CREATE_ERR 17 #define RET_MUTEX_ERR 18 #define RET_COND_ERR 19 #define RET_OVERFLOW 20 -#define RET_NO_MORRET_DATA 21 +#define RET_NO_MORE_DATA 21 #define RET_OUT_OF_ORDER 22 -#define RET_TSBLOCK_TYPRET_NOT_SUPPORTED 23 +#define RET_TSBLOCK_TYPE_NOT_SUPPORTED 23 #define RET_TSBLOCK_DATA_INCONSISTENCY 24 #define RET_DDL_UNKNOWN_TYPE 25 -#define RET_TYPRET_NOT_SUPPORTED 26 -#define RET_TYPRET_NOT_MATCH 27 -#define RET_FILRET_OPEN_ERR 28 -#define RET_FILRET_CLOSRET_ERR 29 -#define RET_FILRET_WRITRET_ERR 30 -#define RET_FILRET_READ_ERR 31 -#define RET_FILRET_SYNC_ERR 32 -#define RET_TSFILRET_WRITER_META_ERR 33 -#define RET_FILRET_STAT_ERR 34 -#define RET_TSFILRET_CORRUPTED 35 +#define RET_TYPE_NOT_SUPPORTED 26 +#define RET_TYPE_NOT_MATCH 27 +#define RET_FILE_OPEN_ERR 28 +#define RET_FILE_CLOSE_ERR 29 +#define RET_FILE_WRITE_ERR 30 +#define RET_FILE_READ_ERR 31 +#define RET_FILE_SYNC_ERR 32 +#define RET_TSFILE_WRITER_META_ERR 33 +#define RET_FILE_STAT_ERR 34 +#define RET_TSFILE_CORRUPTED 35 #define RET_BUF_NOT_ENOUGH 36 #define RET_INVALID_PATH 37 #define RET_NOT_MATCH 38 #define RET_JSON_INVALID 39 #define RET_NOT_SUPPORT 40 #define RET_PARSER_ERR 41 -#define RET_ANALYZRET_ERR 42 +#define RET_ANALYZE_ERR 42 #define RET_INVALID_DATA_POINT 43 -#define RET_DEVICRET_NOT_EXIST 44 +#define RET_DEVICE_NOT_EXIST 44 #define RET_MEASUREMENT_NOT_EXIST 45 #define RET_INVALID_QUERY 46 -#define RET_SDK_QUERY_OPTIMIZRET_ERR 47 +#define RET_SDK_QUERY_OPTIMIZE_ERR 47 #define RET_COMPRESS_ERR 48 -#define RET_TABLRET_NOT_EXIST 49 +#define RET_TABLE_NOT_EXIST 49 #define RET_COLUMN_NOT_EXIST 50 #define RET_UNSUPPORTED_ORDER 51 -#define RET_INVALID_NODRET_TYPE 52 +#define RET_INVALID_NODE_TYPE 52 +#define RET_ENCODE_ERR 53 +#define RET_DECODE_ERR 54 -#endif /* CWRAPPER_ERRNO_DEFINRET_H */ \ No newline at end of file +/* Backward-compatible aliases for identifiers published with misspellings. */ +#define RET_PIPRET_ERR RET_PIPE_ERR +#define RET_THREAD_CREATRET_ERR RET_THREAD_CREATE_ERR +#define RET_NO_MORRET_DATA RET_NO_MORE_DATA +#define RET_TSBLOCK_TYPRET_NOT_SUPPORTED RET_TSBLOCK_TYPE_NOT_SUPPORTED +#define RET_TYPRET_NOT_SUPPORTED RET_TYPE_NOT_SUPPORTED +#define RET_TYPRET_NOT_MATCH RET_TYPE_NOT_MATCH +#define RET_FILRET_OPEN_ERR RET_FILE_OPEN_ERR +#define RET_FILRET_CLOSRET_ERR RET_FILE_CLOSE_ERR +#define RET_FILRET_WRITRET_ERR RET_FILE_WRITE_ERR +#define RET_FILRET_READ_ERR RET_FILE_READ_ERR +#define RET_FILRET_SYNC_ERR RET_FILE_SYNC_ERR +#define RET_TSFILRET_WRITER_META_ERR RET_TSFILE_WRITER_META_ERR +#define RET_FILRET_STAT_ERR RET_FILE_STAT_ERR +#define RET_TSFILRET_CORRUPTED RET_TSFILE_CORRUPTED +#define RET_ANALYZRET_ERR RET_ANALYZE_ERR +#define RET_DEVICRET_NOT_EXIST RET_DEVICE_NOT_EXIST +#define RET_SDK_QUERY_OPTIMIZRET_ERR RET_SDK_QUERY_OPTIMIZE_ERR +#define RET_TABLRET_NOT_EXIST RET_TABLE_NOT_EXIST +#define RET_INVALID_NODRET_TYPE RET_INVALID_NODE_TYPE + +#endif /* CWRAPPER_ERRNO_DEFINE_H */ diff --git a/cpp/test/cwrapper/c_release_test.cc b/cpp/test/cwrapper/c_release_test.cc index bb21483f7..f27e049e0 100644 --- a/cpp/test/cwrapper/c_release_test.cc +++ b/cpp/test/cwrapper/c_release_test.cc @@ -54,8 +54,7 @@ TEST_F(CReleaseTest, TestCreateFile) { // Folder: rejected either as an open error (POSIX) or as already-existing // (Windows / filesystems where the directory already exists). file = write_file_new("test/", &error_no); - ASSERT_TRUE(error_no == RET_FILRET_OPEN_ERR || - error_no == RET_ALREADY_EXIST); + ASSERT_TRUE(error_no == RET_FILE_OPEN_ERR || error_no == RET_ALREADY_EXIST); remove("create_file1.tsfile"); free_write_file(&file); diff --git a/python/tests/test_exceptions.py b/python/tests/test_exceptions.py new file mode 100644 index 000000000..7d025dc27 --- /dev/null +++ b/python/tests/test_exceptions.py @@ -0,0 +1,106 @@ +# 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. + +import pytest + +from tsfile import Field, RowRecord, Tablet, TimeseriesSchema, TSDataType +from tsfile import TsFileReader, TsFileWriter +from tsfile.exceptions import ( + AlreadyExistsError, + ErrorCode, + FileOpenError, + InvalidArgumentError, + LibraryError, + TsFileCorruptedError, + get_exception, +) + + +def test_get_exception_preserves_known_and_unknown_codes(): + corrupted = get_exception(ErrorCode.TSFILE_CORRUPTED) + assert isinstance(corrupted, TsFileCorruptedError) + assert corrupted.code == ErrorCode.TSFILE_CORRUPTED + assert corrupted.message == "TsFile is corrupted" + + unknown = get_exception(999, "opening reader") + assert type(unknown) is LibraryError + assert unknown.code == 999 + assert unknown.message == "opening reader: Unknown library error" + + +def test_writer_constructor_propagates_native_error(tmp_path, capsys): + path = tmp_path / "already-exists.tsfile" + path.touch() + + with pytest.raises(AlreadyExistsError) as exc_info: + TsFileWriter(str(path)) + + assert exc_info.value.code == ErrorCode.ALREADY_EXIST + assert exc_info.value.message == "Resource already exists" + assert capsys.readouterr().out == "" + + +def test_reader_constructor_propagates_native_errors(tmp_path): + missing = tmp_path / "missing.tsfile" + with pytest.raises(FileOpenError) as exc_info: + TsFileReader(str(missing)) + assert exc_info.value.code == ErrorCode.FILE_OPEN_ERROR + + corrupted = tmp_path / "corrupted.tsfile" + corrupted.touch() + with pytest.raises(TsFileCorruptedError) as exc_info: + TsFileReader(str(corrupted)) + assert exc_info.value.code == ErrorCode.TSFILE_CORRUPTED + + +def test_duplicate_tablet_columns_do_not_silently_drop_rows(tmp_path): + path = tmp_path / "duplicate-columns.tsfile" + with TsFileWriter(str(path)) as writer: + writer.register_timeseries( + "root.device", TimeseriesSchema("value", TSDataType.INT64) + ) + tablet = Tablet( + ["value", "VALUE"], + [TSDataType.INT64, TSDataType.INT64], + max_row_num=1, + ) + tablet.set_table_name("root.device") + tablet.add_timestamp(0, 1) + tablet.add_value_by_index(0, 0, 10) + tablet.add_value_by_index(1, 0, 20) + + with pytest.raises(InvalidArgumentError) as exc_info: + writer.write_tablet(tablet) + + assert exc_info.value.code == ErrorCode.INVALID_ARGUMENT + assert "unique" in exc_info.value.message + + +def test_row_record_still_writes_after_error_handling_changes(tmp_path): + path = tmp_path / "row-record.tsfile" + with TsFileWriter(str(path)) as writer: + writer.register_timeseries( + "root.device", TimeseriesSchema("value", TSDataType.INT64) + ) + writer.write_row_record( + RowRecord("root.device", 1, [Field("value", 10, TSDataType.INT64)]) + ) + + with TsFileReader(str(path)) as reader: + with reader.query_timeseries("root.device", ["value"], 0, 2) as result: + assert result.next() + assert result.get_value_by_index(2) == 10 diff --git a/python/tests/test_write_and_read.py b/python/tests/test_write_and_read.py index a1b3b609e..fda145efd 100644 --- a/python/tests/test_write_and_read.py +++ b/python/tests/test_write_and_read.py @@ -639,10 +639,14 @@ def test_tsfile_to_df(): max_row_num=8000, ) assert df3.shape == (4097, 3) - with pytest.raises(TableNotExistError): + with pytest.raises(TableNotExistError) as table_error: to_dataframe("table_write_to_df.tsfile", "test_tb") - with pytest.raises(ColumnNotExistError): + assert table_error.value.code == 49 + assert "test_tb" in table_error.value.message + with pytest.raises(ColumnNotExistError) as column_error: to_dataframe("table_write_to_df.tsfile", "test_table", ["device1"]) + assert column_error.value.code == 50 + assert "device1" in column_error.value.message finally: os.remove("table_write_to_df.tsfile") diff --git a/python/tsfile/exceptions.py b/python/tsfile/exceptions.py index a2e930bad..8f355c21c 100644 --- a/python/tsfile/exceptions.py +++ b/python/tsfile/exceptions.py @@ -17,6 +17,66 @@ # +from enum import IntEnum + + +class ErrorCode(IntEnum): + OK = 0 + OOM = 1 + NOT_EXIST = 2 + ALREADY_EXIST = 3 + INVALID_ARGUMENT = 4 + OUT_OF_RANGE = 5 + PARTIAL_READ = 6 + INVALID_SCHEMA = 8 + NET_EPOLL_ERROR = 9 + NET_EPOLL_WAIT_ERROR = 10 + NET_RECEIVE_ERROR = 11 + NET_ACCEPT_ERROR = 12 + NET_FCNTL_ERROR = 13 + NET_LISTEN_ERROR = 14 + NET_SEND_ERROR = 15 + PIPE_ERROR = 16 + THREAD_CREATE_ERROR = 17 + MUTEX_ERROR = 18 + CONDITION_ERROR = 19 + OVERFLOW = 20 + NO_MORE_DATA = 21 + OUT_OF_ORDER = 22 + TSBLOCK_TYPE_NOT_SUPPORTED = 23 + DATA_INCONSISTENCY = 24 + DDL_UNKNOWN_TYPE = 25 + TYPE_NOT_SUPPORTED = 26 + TYPE_MISMATCH = 27 + FILE_OPEN_ERROR = 28 + FILE_CLOSE_ERROR = 29 + FILE_WRITE_ERROR = 30 + FILE_READ_ERROR = 31 + FILE_SYNC_ERROR = 32 + WRITER_METADATA_ERROR = 33 + FILE_STAT_ERROR = 34 + TSFILE_CORRUPTED = 35 + BUFFER_NOT_ENOUGH = 36 + INVALID_PATH = 37 + NOT_MATCH = 38 + JSON_INVALID = 39 + NOT_SUPPORTED = 40 + PARSER_ERROR = 41 + ANALYZE_ERROR = 42 + INVALID_DATA_POINT = 43 + DEVICE_NOT_EXIST = 44 + MEASUREMENT_NOT_EXIST = 45 + INVALID_QUERY = 46 + QUERY_OPTIMIZE_ERROR = 47 + COMPRESSION_ERROR = 48 + TABLE_NOT_EXIST = 49 + COLUMN_NOT_EXIST = 50 + UNSUPPORTED_ORDER = 51 + INVALID_NODE_TYPE = 52 + ENCODE_ERROR = 53 + DECODE_ERROR = 54 + + class LibraryError(Exception): _default_message = "Unknown error occurred" _default_code = -1 @@ -60,6 +120,31 @@ class PartialReadError(LibraryError): _default_code = 6 +class InvalidSchemaError(LibraryError): + _default_message = "Invalid schema" + _default_code = 8 + + +class TsFileOverflowError(LibraryError): + _default_message = "Buffer or value overflow" + _default_code = 20 + + +class NoMoreDataError(LibraryError): + _default_message = "No more data" + _default_code = 21 + + +class OutOfOrderError(LibraryError): + _default_message = "Data is out of order" + _default_code = 22 + + +class DataInconsistencyError(LibraryError): + _default_message = "Data is inconsistent" + _default_code = 24 + + class FileOpenError(LibraryError): _default_message = "Failed to open file" _default_code = 28 @@ -90,11 +175,26 @@ class MetadataError(LibraryError): _default_code = 33 +class FileStatError(LibraryError): + _default_message = "Failed to inspect file metadata" + _default_code = 34 + + +class TsFileCorruptedError(LibraryError): + _default_message = "TsFile is corrupted" + _default_code = 35 + + class BufferNotEnoughError(LibraryError): _default_message = "Insufficient buffer space" _default_code = 36 +class InvalidPathError(LibraryError): + _default_message = "Invalid path" + _default_code = 37 + + class NotSupportedError(LibraryError): _default_message = "Not support yet" _default_code = 40 @@ -115,6 +215,11 @@ class InvalidQueryError(LibraryError): _default_code = 46 +class QueryOptimizeError(LibraryError): + _default_message = "Failed to optimize query" + _default_code = 47 + + class CompressionError(LibraryError): _default_message = "Data compression/decompression failed" _default_code = 48 @@ -140,6 +245,26 @@ class ColumnNotExistError(LibraryError): _default_code = 50 +class UnsupportedOrderError(LibraryError): + _default_message = "Unsupported ordering" + _default_code = 51 + + +class InvalidNodeTypeError(LibraryError): + _default_message = "Invalid node type" + _default_code = 52 + + +class EncodeError(LibraryError): + _default_message = "Failed to encode data" + _default_code = 53 + + +class DecodeError(LibraryError): + _default_message = "Failed to decode data" + _default_code = 54 + + ERROR_MAPPING = { 1: OOMError, 2: NotExistsError, @@ -147,6 +272,11 @@ ERROR_MAPPING = { 4: InvalidArgumentError, 5: OutOfRangeError, 6: PartialReadError, + 8: InvalidSchemaError, + 20: TsFileOverflowError, + 21: NoMoreDataError, + 22: OutOfOrderError, + 24: DataInconsistencyError, 26: TypeNotSupportedError, 27: TypeMismatchError, 28: FileOpenError, @@ -155,24 +285,43 @@ ERROR_MAPPING = { 31: FileReadError, 32: FileSyncError, 33: MetadataError, + 34: FileStatError, + 35: TsFileCorruptedError, 36: BufferNotEnoughError, + 37: InvalidPathError, 40: NotSupportedError, 44: DeviceNotExistError, 45: MeasurementNotExistError, 46: InvalidQueryError, + 47: QueryOptimizeError, 48: CompressionError, 49: TableNotExistError, 50: ColumnNotExistError, + 51: UnsupportedOrderError, + 52: InvalidNodeTypeError, + 53: EncodeError, + 54: DecodeError, +} + + +ERROR_MESSAGES = { + error.value: error.name.replace("_", " ").lower() for error in ErrorCode } +ERROR_MESSAGES.update( + { + code: exception_type._default_message + for code, exception_type in ERROR_MAPPING.items() + } +) def get_exception(code: int, context: str = None): - if code == 0: + code = int(code) + if code == ErrorCode.OK: return None - exc_type = ERROR_MAPPING.get(code) - if not exc_type: - return LibraryError( - code=code, context=f"Unmapped error code: {code}, message: {context}" - ) - return exc_type(code=code, context=context) + exc_type = ERROR_MAPPING.get(code, LibraryError) + message = ERROR_MESSAGES.get(code, "Unknown library error") + if context: + message = f"{context}: {message}" + return exc_type(code=code, context=message) diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index 4e90dd483..cc14f4034 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -21,6 +21,11 @@ from libc.stdint cimport uint32_t, int32_t, int64_t, uint64_t, uint8_t ctypedef int32_t ErrorCode +cdef extern from "cwrapper/errno_define_c.h": + enum: + RET_OK + RET_NO_MORE_DATA + # import symbols from tsfile_cwrapper.h cdef extern from "cwrapper/tsfile_cwrapper.h": # common diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd index 7286e78fb..0e2f91cbc 100644 --- a/python/tsfile/tsfile_py_cpp.pxd +++ b/python/tsfile/tsfile_py_cpp.pxd @@ -41,8 +41,8 @@ cdef public api void free_c_timeseries_schema(TimeseriesSchema* c_schema) cdef public api void free_c_device_schema(DeviceSchema* c_schema) cdef public api void free_c_tablet(Tablet tablet) cdef public api void free_c_row_record(TsRecord record) -cdef public api TsFileWriter tsfile_writer_new_c(object pathname, uint64_t memory_threshold) except + -cdef public api TsFileReader tsfile_reader_new_c(object pathname) except + +cdef public api TsFileWriter tsfile_writer_new_c(object pathname, uint64_t memory_threshold) except NULL +cdef public api TsFileReader tsfile_reader_new_c(object pathname) except NULL cdef public api ErrorCode tsfile_writer_register_device_py_cpp(TsFileWriter writer, DeviceSchema *schema) cdef public api ErrorCode tsfile_writer_register_timeseries_py_cpp(TsFileWriter writer, object device_name, TimeseriesSchema *schema) diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index 038336a42..e36b548ac 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -27,11 +27,10 @@ from libc.stdlib cimport free from libc.stdlib cimport malloc from libc.string cimport strdup from libc.string cimport memset -from cpython.exc cimport PyErr_SetObject from cpython.unicode cimport PyUnicode_AsUTF8String, PyUnicode_AsUTF8, PyUnicode_AsUTF8AndSize from cpython.bytes cimport PyBytes_AsString, PyBytes_AsStringAndSize -from tsfile.exceptions import ERROR_MAPPING, TypeMismatchError +from tsfile.exceptions import get_exception, InvalidArgumentError, TypeMismatchError 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 @@ -49,17 +48,11 @@ from tsfile.schema import TimeseriesMetadata as TimeseriesMetadataPy # check exception and set py exception object cdef inline void check_error(int errcode, const char * context=NULL) except*: - cdef: - object exc_type - object exc_instance - - if errcode == 0: + if errcode == RET_OK: return - exc_type = ERROR_MAPPING.get(errcode) - print(exc_type) - exc_instance = exc_type(errcode, "") - PyErr_SetObject(exc_type, exc_instance) + py_context = context.decode('utf-8') if context != NULL else None + raise get_exception(errcode, py_context) # convert from c to python cdef object from_c_result_set_meta_data(ResultSetMetaData schema): @@ -226,18 +219,35 @@ cdef TableSchema * to_c_table_schema(object py_schema): return c_schema cdef Tablet to_c_tablet(object tablet): - cdef Tablet ctablet + cdef Tablet ctablet = NULL cdef int max_row_num + cdef int column_num + cdef int i, row, col cdef TSDataType data_type cdef int64_t timestamp cdef bytes device_id_bytes cdef const char * device_id_c - cdef char** columns_names - cdef TSDataType * column_types - cdef bytes row_bytes - cdef char *raw_str + cdef char** columns_names = NULL + cdef TSDataType * columns_types = NULL + cdef char *raw_str = NULL cdef const char * str_ptr cdef Py_ssize_t raw_len + cdef object column_name_list = tablet.get_column_name_list() + cdef object data_type_list = tablet.get_data_type_list() + cdef object value_list = tablet.get_value_list() + + column_num = len(column_name_list) + max_row_num = tablet.get_max_row_num() + if column_num != len(data_type_list) or column_num != len(value_list): + raise InvalidArgumentError( + context="Tablet column names, data types, and values must have equal lengths" + ) + if len({(<str> name).lower() for name in column_name_list}) != column_num: + raise InvalidArgumentError( + context="Tablet column names must be unique (case-insensitive)" + ) + if max_row_num <= 0: + raise InvalidArgumentError(context="Tablet max_row_num must be positive") if tablet.get_target_name() is not None: device_id_bytes = PyUnicode_AsUTF8String(tablet.get_target_name()) @@ -245,90 +255,153 @@ cdef Tablet to_c_tablet(object tablet): else: device_id_c = NULL - column_num = len(tablet.get_column_name_list()) columns_names = <char**> malloc(sizeof(char *) * column_num) columns_types = <TSDataType *> malloc(sizeof(TSDataType) * column_num) - for i in range(column_num): - columns_names[i] = strdup(tablet.get_column_name_list()[i].encode('utf-8')) - columns_types[i] = to_c_data_type(tablet.get_data_type_list()[i]) - - 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) - free(columns_types) - for i in range(column_num): - free(columns_names[i]) - free(columns_names) - - for row in range(max_row_num): - timestamp_py = tablet.get_timestamp_list()[row] - if timestamp_py is None: - continue - timestamp = timestamp_py - tablet_add_timestamp(ctablet, row, timestamp) - - for col in range(column_num): - data_type = to_c_data_type(tablet.get_data_type_list()[col]) - value = tablet.get_value_list()[col] - # BOOLEAN - if data_type == TS_DATATYPE_BOOLEAN: - for row in range(max_row_num): - if value[row] is not None: - tablet_add_value_by_index_bool(ctablet, row, col, value[row]) - # INT32 - elif data_type == TS_DATATYPE_INT32: - for row in range(max_row_num): - if value[row] is not None: - tablet_add_value_by_index_int32_t(ctablet, row, col, value[row]) - - # INT64 - elif data_type == TS_DATATYPE_INT64 or data_type == TS_DATATYPE_TIMESTAMP: - for row in range(max_row_num): - if value[row] is not None: - tablet_add_value_by_index_int64_t(ctablet, row, col, value[row]) - # FLOAT - elif data_type == TS_DATATYPE_FLOAT: - for row in range(max_row_num): - if value[row] is not None: - tablet_add_value_by_index_float(ctablet, row, col, value[row]) + if columns_names == NULL or columns_types == NULL: + free(columns_names) + free(columns_types) + raise MemoryError("Failed to allocate tablet schema arrays") + memset(columns_names, 0, sizeof(char *) * column_num) + try: + for i in range(column_num): + columns_names[i] = strdup((<str> column_name_list[i]).encode('utf-8')) + if columns_names[i] == NULL: + raise MemoryError("Failed to allocate tablet column name") + columns_types[i] = to_c_data_type(data_type_list[i]) + ctablet = _tablet_new_with_target_name( + device_id_c, columns_names, columns_types, column_num, max_row_num + ) + finally: + free(columns_types) + for i in range(column_num): + free(columns_names[i]) + free(columns_names) - # DOUBLE - elif data_type == TS_DATATYPE_DOUBLE: - for row in range(max_row_num): - if value[row] is not None: - tablet_add_value_by_index_double(ctablet, row, col, value[row]) + if ctablet == NULL: + raise MemoryError("Failed to allocate native tablet") - elif data_type == TS_DATATYPE_DATE: - for row in range(max_row_num): - if value[row] is not None: - tablet_add_value_by_index_int32_t(ctablet, row, col, parse_date_to_int(value[row])) + try: + for row in range(max_row_num): + timestamp_py = tablet.get_timestamp_list()[row] + if timestamp_py is None: + continue + timestamp = timestamp_py + check_error( + tablet_add_timestamp(ctablet, row, timestamp), + b"Failed to add tablet timestamp", + ) + + for col in range(column_num): + data_type = to_c_data_type(data_type_list[col]) + value = value_list[col] + # BOOLEAN + if data_type == TS_DATATYPE_BOOLEAN: + for row in range(max_row_num): + if value[row] is not None: + check_error( + tablet_add_value_by_index_bool( + ctablet, row, col, value[row] + ), + b"Failed to add BOOLEAN tablet value", + ) + # INT32 + elif data_type == TS_DATATYPE_INT32: + for row in range(max_row_num): + if value[row] is not None: + check_error( + tablet_add_value_by_index_int32_t( + ctablet, row, col, value[row] + ), + b"Failed to add INT32 tablet value", + ) + # INT64 + elif data_type == TS_DATATYPE_INT64 or data_type == TS_DATATYPE_TIMESTAMP: + for row in range(max_row_num): + if value[row] is not None: + check_error( + tablet_add_value_by_index_int64_t( + ctablet, row, col, value[row] + ), + b"Failed to add INT64 tablet value", + ) + # FLOAT + elif data_type == TS_DATATYPE_FLOAT: + for row in range(max_row_num): + if value[row] is not None: + check_error( + tablet_add_value_by_index_float( + ctablet, row, col, value[row] + ), + b"Failed to add FLOAT tablet value", + ) + # DOUBLE + elif data_type == TS_DATATYPE_DOUBLE: + for row in range(max_row_num): + if value[row] is not None: + check_error( + tablet_add_value_by_index_double( + ctablet, row, col, value[row] + ), + b"Failed to add DOUBLE tablet value", + ) + elif data_type == TS_DATATYPE_DATE: + for row in range(max_row_num): + if value[row] is not None: + check_error( + tablet_add_value_by_index_int32_t( + ctablet, row, col, parse_date_to_int(value[row]) + ), + b"Failed to add DATE tablet value", + ) + # STRING or TEXT + elif data_type == TS_DATATYPE_STRING or data_type == TS_DATATYPE_TEXT: + for row in range(max_row_num): + if value[row] is not None: + py_value = value[row] + str_ptr = PyUnicode_AsUTF8AndSize(py_value, &raw_len) + check_error( + tablet_add_value_by_index_string_with_len( + ctablet, row, col, str_ptr, raw_len + ), + b"Failed to add STRING tablet value", + ) + elif data_type == TS_DATATYPE_BLOB: + for row in range(max_row_num): + if value[row] is not None: + if PyBytes_AsStringAndSize( + value[row], &raw_str, &raw_len + ) < 0: + raise TypeError("BLOB tablet values must be bytes") + check_error( + tablet_add_value_by_index_string_with_len( + ctablet, row, col, raw_str, raw_len + ), + b"Failed to add BLOB tablet value", + ) + except: + free_tablet(&ctablet) + raise - # STRING or TEXT - elif data_type == TS_DATATYPE_STRING or data_type == TS_DATATYPE_TEXT: - for row in range(max_row_num): - if value[row] is not None: - py_value = value[row] - str_ptr = PyUnicode_AsUTF8AndSize(py_value, &raw_len) - tablet_add_value_by_index_string_with_len(ctablet, row, col, str_ptr, raw_len) + return ctablet - elif data_type == TS_DATATYPE_BLOB: - for row in range(max_row_num): - if value[row] is not None: - PyBytes_AsStringAndSize(value[row], &raw_str, &raw_len) - tablet_add_value_by_index_string_with_len(ctablet, row, col, raw_str, raw_len) +cdef inline void check_dataframe_tablet_error( + Tablet tablet, int errcode, const char * context +) except *: + if errcode == RET_OK: + return + free_tablet(&tablet) + check_error(errcode, context) - return ctablet cdef Tablet dataframe_to_c_tablet(object target_name, object dataframe, object table_schema): - cdef Tablet ctablet + cdef Tablet ctablet = NULL cdef int max_row_num cdef TSDataType data_type cdef int64_t timestamp cdef const char * device_id_c = NULL - cdef char** columns_names - cdef TSDataType * columns_types - cdef char *raw_str + cdef char** columns_names = NULL + cdef TSDataType * columns_types = NULL + cdef char *raw_str = NULL cdef const char * str_ptr cdef Py_ssize_t raw_len cdef int column_num @@ -359,20 +432,68 @@ cdef Tablet dataframe_to_c_tablet(object target_name, object dataframe, object t data_type = table_schema.get_column(column).get_data_type() column_types_list.append(data_type) + # Validate object-backed columns before allocating the native Tablet so a + # validation exception cannot leak the native allocation. + for col in range(column_num): + col_name = data_columns[col] + data_type = column_types_list[col] + if data_type in ( + TS_DATATYPE_DATE, + TS_DATATYPE_STRING, + TS_DATATYPE_TEXT, + TS_DATATYPE_BLOB, + ): + col_series = dataframe[col_name] + first_valid_idx = col_series.first_valid_index() + if first_valid_idx is not None: + value = col_series[first_valid_idx] + if data_type == TS_DATATYPE_DATE and not isinstance(value, date_type): + raise TypeMismatchError( + context=( + f"Column '{col_name}': expected DATE (datetime.date), " + f"got {type(value).__name__}: {value!r}" + ) + ) + if data_type in (TS_DATATYPE_STRING, TS_DATATYPE_TEXT) and not isinstance(value, str): + raise TypeMismatchError( + context=( + f"Column '{col_name}': expected STRING/TEXT, " + f"got {type(value).__name__}: {value!r}" + ) + ) + if data_type == TS_DATATYPE_BLOB and not isinstance(value, bytes): + raise TypeMismatchError( + context=( + f"Column '{col_name}': expected BLOB (bytes), " + f"got {type(value).__name__}: {value!r}" + ) + ) + columns_names = <char**> malloc(sizeof(char *) * column_num) columns_types = <TSDataType *> malloc(sizeof(TSDataType) * column_num) + if columns_names == NULL or columns_types == NULL: + free(columns_names) + free(columns_types) + raise MemoryError("Failed to allocate DataFrame tablet schema arrays") + memset(columns_names, 0, sizeof(char *) * column_num) + try: + for i in range(column_num): + columns_names[i] = strdup(data_columns[i].lower().encode('utf-8')) + if columns_names[i] == NULL: + raise MemoryError("Failed to allocate DataFrame tablet column name") + columns_types[i] = to_c_data_type(column_types_list[i]) - for i in range(column_num): - columns_names[i] = strdup(data_columns[i].lower().encode('utf-8')) - columns_types[i] = column_types_list[i] - - ctablet = _tablet_new_with_target_name(device_id_c, columns_names, columns_types, column_num, - max_row_num) + ctablet = _tablet_new_with_target_name( + device_id_c, columns_names, columns_types, column_num, max_row_num + ) + finally: + free(columns_types) + for i in range(column_num): + free(columns_names[i]) + free(columns_names) - free(columns_types) - for i in range(column_num): - free(columns_names[i]) - free(columns_names) + if ctablet == NULL: + raise MemoryError("Failed to allocate native DataFrame tablet") if use_id_as_time: for row in range(max_row_num): @@ -380,7 +501,11 @@ cdef Tablet dataframe_to_c_tablet(object target_name, object dataframe, object t if pd.isna(timestamp_py): continue timestamp = <int64_t> timestamp_py - tablet_add_timestamp(ctablet, row, timestamp) + check_dataframe_tablet_error( + ctablet, + tablet_add_timestamp(ctablet, row, timestamp), + b"Failed to add DataFrame index timestamp", + ) else: time_values = dataframe[time_column.get_column_name()].values for row in range(max_row_num): @@ -388,74 +513,89 @@ cdef Tablet dataframe_to_c_tablet(object target_name, object dataframe, object t if pd.isna(timestamp_py): continue timestamp = <int64_t> timestamp_py - tablet_add_timestamp(ctablet, row, timestamp) + check_dataframe_tablet_error( + ctablet, + tablet_add_timestamp(ctablet, row, timestamp), + b"Failed to add DataFrame timestamp", + ) for col in range(column_num): col_name = data_columns[col] data_type = column_types_list[col] column_values = dataframe[col_name].values - # Per-column validation for object types (check first non-null value only) - if data_type in (TS_DATATYPE_DATE, TS_DATATYPE_STRING, TS_DATATYPE_TEXT, TS_DATATYPE_BLOB): - col_series = dataframe[col_name] - first_valid_idx = col_series.first_valid_index() - if first_valid_idx is not None: - value = col_series[first_valid_idx] - if data_type == TS_DATATYPE_DATE: - if not isinstance(value, date_type): - raise TypeMismatchError(context= - f"Column '{col_name}': expected DATE (datetime.date), " - f"got {type(value).__name__}: {value!r}" - ) - elif data_type in (TS_DATATYPE_STRING, TS_DATATYPE_TEXT): - if not isinstance(value, str): - raise TypeMismatchError(context= - f"Column '{col_name}': expected STRING/TEXT, " - f"got {type(value).__name__}: {value!r}" - ) - elif data_type == TS_DATATYPE_BLOB: - if not isinstance(value, bytes): - raise TypeMismatchError(context= - f"Column '{col_name}': expected BLOB (bytes or bytearray), " - f"got {type(value).__name__}: {value!r}" - ) - # BOOLEAN if data_type == TS_DATATYPE_BOOLEAN: for row in range(max_row_num): value = column_values[row] if not pd.isna(value): - tablet_add_value_by_index_bool(ctablet, row, col, <bint> value) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_bool( + ctablet, row, col, <bint> value + ), + b"Failed to add BOOLEAN DataFrame value", + ) # INT32 elif data_type == TS_DATATYPE_INT32: for row in range(max_row_num): value = column_values[row] if not pd.isna(value): - tablet_add_value_by_index_int32_t(ctablet, row, col, <int32_t> value) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_int32_t( + ctablet, row, col, <int32_t> value + ), + b"Failed to add INT32 DataFrame value", + ) # INT64 elif data_type == TS_DATATYPE_INT64 or data_type == TS_DATATYPE_TIMESTAMP: for row in range(max_row_num): value = column_values[row] if not pd.isna(value): - tablet_add_value_by_index_int64_t(ctablet, row, col, <int64_t> value) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_int64_t( + ctablet, row, col, <int64_t> value + ), + b"Failed to add INT64 DataFrame value", + ) # FLOAT elif data_type == TS_DATATYPE_FLOAT: for row in range(max_row_num): value = column_values[row] if not pd.isna(value): - tablet_add_value_by_index_float(ctablet, row, col, <float> value) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_float( + ctablet, row, col, <float> value + ), + b"Failed to add FLOAT DataFrame value", + ) # DOUBLE elif data_type == TS_DATATYPE_DOUBLE: for row in range(max_row_num): value = column_values[row] if not pd.isna(value): - tablet_add_value_by_index_double(ctablet, row, col, <double> value) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_double( + ctablet, row, col, <double> value + ), + b"Failed to add DOUBLE DataFrame value", + ) # DATE (validated per-column above) elif data_type == TS_DATATYPE_DATE: for row in range(max_row_num): value = column_values[row] if not pd.isna(value): - tablet_add_value_by_index_int32_t(ctablet, row, col, parse_date_to_int(value)) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_int32_t( + ctablet, row, col, parse_date_to_int(value) + ), + b"Failed to add DATE DataFrame value", + ) # STRING or TEXT (validated per-column above) elif data_type == TS_DATATYPE_STRING or data_type == TS_DATATYPE_TEXT: for row in range(max_row_num): @@ -463,7 +603,13 @@ cdef Tablet dataframe_to_c_tablet(object target_name, object dataframe, object t if not pd.isna(value): py_value = str(value) str_ptr = PyUnicode_AsUTF8AndSize(py_value, &raw_len) - tablet_add_value_by_index_string_with_len(ctablet, row, col, str_ptr, raw_len) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_string_with_len( + ctablet, row, col, str_ptr, raw_len + ), + b"Failed to add STRING DataFrame value", + ) # BLOB (validated per-column above) elif data_type == TS_DATATYPE_BLOB: for row in range(max_row_num): @@ -471,11 +617,23 @@ cdef Tablet dataframe_to_c_tablet(object target_name, object dataframe, object t if not pd.isna(value): if isinstance(value, bytes): PyBytes_AsStringAndSize(value, &raw_str, &raw_len) - tablet_add_value_by_index_string_with_len(ctablet, row, col, raw_str, raw_len) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_string_with_len( + ctablet, row, col, raw_str, raw_len + ), + b"Failed to add BLOB DataFrame value", + ) else: value_bytes = bytes(value) PyBytes_AsStringAndSize(value_bytes, &raw_str, &raw_len) - tablet_add_value_by_index_string_with_len(ctablet, row, col, raw_str, raw_len) + check_dataframe_tablet_error( + ctablet, + tablet_add_value_by_index_string_with_len( + ctablet, row, col, raw_str, raw_len + ), + b"Failed to add BLOB DataFrame value", + ) return ctablet @@ -485,42 +643,100 @@ cdef TsRecord to_c_record(object row_record): cdef bytes device_id_bytes = PyUnicode_AsUTF8String(row_record.get_device_id()) cdef const char * device_id = device_id_bytes cdef const char * str_ptr - cdef char * blob_ptr + cdef char * blob_ptr = NULL cdef Py_ssize_t str_len - cdef TsRecord record + cdef TsRecord record = NULL cdef int i cdef TSDataType data_type record = _ts_record_new(device_id, timestamp, field_num) - for i in range(field_num): - 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()) - elif data_type == TS_DATATYPE_INT32 or data_type == TS_DATATYPE_DATE: - _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()) - elif data_type == TS_DATATYPE_TIMESTAMP: - _insert_data_into_ts_record_by_name_int64_t(record, PyUnicode_AsUTF8(field.get_field_name()), - field.get_timestamp_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()) - 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()) - elif data_type == TS_DATATYPE_TEXT or data_type == TS_DATATYPE_STRING: - str_ptr = PyUnicode_AsUTF8AndSize(field.get_string_value(), &str_len) - _insert_data_into_ts_record_by_name_string_with_len(record, PyUnicode_AsUTF8(field.get_field_name()), - str_ptr, str_len) - elif data_type == TS_DATATYPE_BLOB: - if PyBytes_AsStringAndSize(field.get_string_value(), &blob_ptr, &str_len) < 0: - raise ValueError("blob not legal") - _insert_data_into_ts_record_by_name_string_with_len(record, PyUnicode_AsUTF8(field.get_field_name()), - <const char *> blob_ptr, <uint32_t> str_len) + if record == NULL: + raise MemoryError("Failed to allocate native row record") + try: + for i in range(field_num): + field = row_record.get_fields()[i] + data_type = to_c_data_type(field.get_data_type()) + if data_type == TS_DATATYPE_BOOLEAN: + check_error( + _insert_data_into_ts_record_by_name_bool( + record, + PyUnicode_AsUTF8(field.get_field_name()), + field.get_bool_value(), + ), + b"Failed to add BOOLEAN row-record field", + ) + elif data_type == TS_DATATYPE_INT32 or data_type == TS_DATATYPE_DATE: + check_error( + _insert_data_into_ts_record_by_name_int32_t( + record, + PyUnicode_AsUTF8(field.get_field_name()), + field.get_int_value(), + ), + b"Failed to add INT32 row-record field", + ) + elif data_type == TS_DATATYPE_INT64: + check_error( + _insert_data_into_ts_record_by_name_int64_t( + record, + PyUnicode_AsUTF8(field.get_field_name()), + field.get_long_value(), + ), + b"Failed to add INT64 row-record field", + ) + elif data_type == TS_DATATYPE_TIMESTAMP: + check_error( + _insert_data_into_ts_record_by_name_int64_t( + record, + PyUnicode_AsUTF8(field.get_field_name()), + field.get_timestamp_value(), + ), + b"Failed to add TIMESTAMP row-record field", + ) + elif data_type == TS_DATATYPE_DOUBLE: + check_error( + _insert_data_into_ts_record_by_name_double( + record, + PyUnicode_AsUTF8(field.get_field_name()), + field.get_double_value(), + ), + b"Failed to add DOUBLE row-record field", + ) + elif data_type == TS_DATATYPE_FLOAT: + check_error( + _insert_data_into_ts_record_by_name_float( + record, + PyUnicode_AsUTF8(field.get_field_name()), + field.get_float_value(), + ), + b"Failed to add FLOAT row-record field", + ) + elif data_type == TS_DATATYPE_TEXT or data_type == TS_DATATYPE_STRING: + str_ptr = PyUnicode_AsUTF8AndSize(field.get_string_value(), &str_len) + check_error( + _insert_data_into_ts_record_by_name_string_with_len( + record, + PyUnicode_AsUTF8(field.get_field_name()), + str_ptr, + str_len, + ), + b"Failed to add STRING row-record field", + ) + elif data_type == TS_DATATYPE_BLOB: + if PyBytes_AsStringAndSize( + field.get_string_value(), &blob_ptr, &str_len + ) < 0: + raise TypeError("BLOB row-record values must be bytes") + check_error( + _insert_data_into_ts_record_by_name_string_with_len( + record, + PyUnicode_AsUTF8(field.get_field_name()), + <const char *> blob_ptr, + <uint32_t> str_len, + ), + b"Failed to add BLOB row-record field", + ) + except: + _free_tsfile_ts_record(&record) + raise return record # Free c structs' space @@ -549,7 +765,7 @@ cdef void free_c_row_record(TsRecord record): _free_tsfile_ts_record(&record) # Reader and writer new. -cdef TsFileWriter tsfile_writer_new_c(object pathname, uint64_t memory_threshold) except +: +cdef TsFileWriter tsfile_writer_new_c(object pathname, uint64_t memory_threshold) except NULL: cdef ErrorCode errno = 0 cdef TsFileWriter writer cdef bytes encoded_path = PyUnicode_AsUTF8String(pathname) @@ -558,7 +774,7 @@ cdef TsFileWriter tsfile_writer_new_c(object pathname, uint64_t memory_threshold check_error(errno) return writer -cdef TsFileReader tsfile_reader_new_c(object pathname) except +: +cdef TsFileReader tsfile_reader_new_c(object pathname) except NULL: cdef ErrorCode errno = 0 cdef TsFileReader reader cdef bytes encoded_path = PyUnicode_AsUTF8String(pathname) diff --git a/python/tsfile/tsfile_reader.pyx b/python/tsfile/tsfile_reader.pyx index f94d42053..4ca1ca714 100644 --- a/python/tsfile/tsfile_reader.pyx +++ b/python/tsfile/tsfile_reader.pyx @@ -164,9 +164,9 @@ cdef class ResultSetPy: code = tsfile_result_set_get_next_tsblock_as_arrow(self.result, &arrow_array, &arrow_schema) - if code == 21: # E_NO_MORE_DATA + if code == RET_NO_MORE_DATA: return None - if code != 0: + if code != RET_OK: check_error(code) if arrow_schema.release == NULL or arrow_array.release == NULL: diff --git a/python/tsfile/utils.py b/python/tsfile/utils.py index ea9263149..0ec54b3a7 100644 --- a/python/tsfile/utils.py +++ b/python/tsfile/utils.py @@ -142,7 +142,9 @@ def to_dataframe( else: _table_name = _table_name.lower() if _table_name.lower() not in table_schema: - raise TableNotExistError(_table_name) + raise TableNotExistError( + context=f"Table '{_table_name}' does not exist" + ) table_schema = table_schema[_table_name] column_names_in_file = [] @@ -155,7 +157,9 @@ def to_dataframe( if _column_names is not None: for column in _column_names: if column not in column_names_in_file and column != time_column: - raise ColumnNotExistError(column) + raise ColumnNotExistError( + context=f"Column '{column}' does not exist" + ) if ( table_schema.get_column(column).get_category() == ColumnCategory.FIELD
