This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new a2fdf7116 Python error code alignment (#870)
a2fdf7116 is described below
commit a2fdf71161790af959a4311f49a88ff0c46abc4a
Author: Colin Lee <[email protected]>
AuthorDate: Mon Jul 27 15:40:11 2026 +0800
Python error code alignment (#870)
---
cpp/examples/c_examples/demo_read.c | 2 +-
cpp/src/cwrapper/errno_define_c.h | 54 +--
cpp/src/file/read_file.cc | 4 +-
cpp/src/reader/tsfile_executor.cc | 2 +-
cpp/src/reader/tsfile_reader.cc | 32 +-
cpp/src/utils/errno_define.h | 16 -
cpp/test/cwrapper/c_release_test.cc | 19 +-
cpp/test/cwrapper/query_by_row_cwrapper_test.cc | 27 ++
cpp/tools/format/output_format.cc | 2 -
python/tests/test_exceptions.py | 142 ++++++
python/tests/test_write_and_read.py | 8 +-
python/tsfile/exceptions.py | 145 ++++++-
python/tsfile/tsfile_cpp.pxd | 5 +
python/tsfile/tsfile_py_cpp.pxd | 4 +-
python/tsfile/tsfile_py_cpp.pyx | 554 ++++++++++++++++--------
python/tsfile/tsfile_reader.pyx | 18 +-
python/tsfile/tsfile_writer.pyx | 6 +-
python/tsfile/utils.py | 8 +-
18 files changed, 787 insertions(+), 261 deletions(-)
diff --git a/cpp/examples/c_examples/demo_read.c
b/cpp/examples/c_examples/demo_read.c
index 5ac6111e7..efa6b63d0 100644
--- a/cpp/examples/c_examples/demo_read.c
+++ b/cpp/examples/c_examples/demo_read.c
@@ -37,7 +37,7 @@ ERRNO read_tsfile() {
HANDLE_ERROR(code);
if (ret == NULL) {
- HANDLE_ERROR(RET_INVALID_QUERY);
+ HANDLE_ERROR(RET_INVALID_ARG);
}
// Get query result metadata: column name and datatype
diff --git a/cpp/src/cwrapper/errno_define_c.h
b/cpp/src/cwrapper/errno_define_c.h
index 3ceb06416..3ad1c1301 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
@@ -28,49 +28,33 @@
#define RET_OUT_OF_RANGE 5
#define RET_PARTIAL_READ 6
#define RET_INVALID_SCHEMA 8
-#define RET_NET_EPOLL_ERR 9
-#define RET_NET_EPOLL_WAIT_ERR 10
-#define RET_NET_RECV_ERR 11
-#define RET_NET_ACCEPT_ERR 12
-#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_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_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_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_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
+#endif /* CWRAPPER_ERRNO_DEFINE_H */
diff --git a/cpp/src/file/read_file.cc b/cpp/src/file/read_file.cc
index c6bfd547a..7d41d7095 100644
--- a/cpp/src/file/read_file.cc
+++ b/cpp/src/file/read_file.cc
@@ -40,7 +40,7 @@ using namespace common;
namespace storage {
void ReadFile::close() {
- if (fd_ > 0) {
+ if (fd_ >= 0) {
::close(fd_);
fd_ = -1;
}
@@ -65,7 +65,7 @@ int ReadFile::open(const std::string& file_path) {
} else if (RET_FAIL(check_file_magic())) {
}
if (IS_FAIL(ret)) {
- ::close(fd_);
+ close();
}
return ret;
}
diff --git a/cpp/src/reader/tsfile_executor.cc
b/cpp/src/reader/tsfile_executor.cc
index f2d34167d..15aaa161b 100644
--- a/cpp/src/reader/tsfile_executor.cc
+++ b/cpp/src/reader/tsfile_executor.cc
@@ -67,7 +67,7 @@ int TsFileExecutor::execute(QueryExpression* query_expr,
ResultSet*& ret_qds) {
if (query_exprs_->has_filter_) {
regular_expr = query_exprs_->optimize(origin_expr, paths);
if (regular_expr == nullptr) {
- return E_SDK_QUERY_OPTIMIZE_ERR;
+ return E_INVALID_ARG;
}
query_exprs_->set_expression(regular_expr);
}
diff --git a/cpp/src/reader/tsfile_reader.cc b/cpp/src/reader/tsfile_reader.cc
index 540674f33..33d0d8967 100644
--- a/cpp/src/reader/tsfile_reader.cc
+++ b/cpp/src/reader/tsfile_reader.cc
@@ -18,6 +18,8 @@
*/
#include "tsfile_reader.h"
+#include <stdexcept>
+
#include "common/schema.h"
#include "filter/time_operator.h"
#include "tsfile_executor.h"
@@ -33,6 +35,19 @@ struct DeviceMetaEntry {
int64_t end_offset;
};
+int parse_paths(const std::vector<std::string>& path_list,
+ std::vector<Path>& parsed_paths) {
+ try {
+ parsed_paths.reserve(path_list.size());
+ for (const auto& path : path_list) {
+ parsed_paths.emplace_back(path, true);
+ }
+ } catch (const std::runtime_error&) {
+ return E_INVALID_PATH;
+ }
+ return E_OK;
+}
+
int get_all_device_entries(std::vector<DeviceMetaEntry>& entries,
std::shared_ptr<MetaIndexNode> index_node,
ReadFile* read_file, PageArena& pa) {
@@ -152,14 +167,15 @@ int TsFileReader::query(QueryExpression* qe, ResultSet*&
ret_qds) {
int TsFileReader::query(std::vector<std::string>& path_list, int64_t
start_time,
int64_t end_time, ResultSet*& result_set) {
- int ret = E_OK;
+ std::vector<Path> path_list_vec;
+ int ret = parse_paths(path_list, path_list_vec);
+ if (ret != E_OK) {
+ return ret;
+ }
+
Filter* time_filter = new TimeBetween(start_time, end_time, false);
Expression* exp =
new storage::Expression(storage::GLOBALTIME_EXPR, time_filter);
- std::vector<Path> path_list_vec;
- for (const auto& path : path_list) {
- path_list_vec.emplace_back(Path(path, true));
- }
QueryExpression* query_expression =
QueryExpression::create(path_list_vec, exp);
ret = tsfile_executor_->execute(query_expression, result_set);
@@ -200,10 +216,10 @@ int TsFileReader::query(const std::string& table_name,
int TsFileReader::queryByRow(std::vector<std::string>& path_list, int offset,
int limit, ResultSet*& result_set) {
- int ret = E_OK;
std::vector<Path> path_list_vec;
- for (const auto& path : path_list) {
- path_list_vec.emplace_back(Path(path, true));
+ int ret = parse_paths(path_list, path_list_vec);
+ if (ret != E_OK) {
+ return ret;
}
QueryExpression* query_expression =
QueryExpression::create(path_list_vec, nullptr);
diff --git a/cpp/src/utils/errno_define.h b/cpp/src/utils/errno_define.h
index f52cead99..ca2d1397a 100644
--- a/cpp/src/utils/errno_define.h
+++ b/cpp/src/utils/errno_define.h
@@ -29,17 +29,6 @@ const int E_INVALID_ARG = 4;
const int E_OUT_OF_RANGE = 5;
const int E_PARTIAL_READ = 6;
const int E_INVALID_SCHEMA = 8;
-const int E_NET_EPOLL_ERR = 9;
-const int E_NET_EPOLL_WAIT_ERR = 10;
-const int E_NET_RECV_ERR = 11;
-const int E_NET_ACCEPT_ERR = 12;
-const int E_NET_FCNTL_ERR = 13;
-const int E_NET_LISTEN_ERR = 14;
-const int E_NET_SEND_ERR = 15;
-const int E_PIPE_ERR = 16;
-const int E_THREAD_CREATE_ERR = 17;
-const int E_MUTEX_ERR = 18;
-const int E_COND_ERR = 19;
const int E_OVERFLOW = 20;
const int E_NO_MORE_DATA = 21;
const int E_OUT_OF_ORDER = 22;
@@ -57,15 +46,10 @@ const int E_TSFILE_CORRUPTED = 35;
const int E_BUF_NOT_ENOUGH = 36;
const int E_INVALID_PATH = 37;
const int E_NOT_MATCH = 38;
-const int E_JSON_INVALID = 39;
const int E_NOT_SUPPORT = 40;
-const int E_PARSER_ERR = 41;
-const int E_ANALYZE_ERR = 42;
const int E_INVALID_DATA_POINT = 43;
const int E_DEVICE_NOT_EXIST = 44;
const int E_MEASUREMENT_NOT_EXIST = 45;
-const int E_INVALID_QUERY = 46;
-const int E_SDK_QUERY_OPTIMIZE_ERR = 47;
const int E_COMPRESS_ERR = 48;
const int E_TABLE_NOT_EXIST = 49;
const int E_COLUMN_NOT_EXIST = 50;
diff --git a/cpp/test/cwrapper/c_release_test.cc
b/cpp/test/cwrapper/c_release_test.cc
index bb21483f7..ae3f5cb3d 100644
--- a/cpp/test/cwrapper/c_release_test.cc
+++ b/cpp/test/cwrapper/c_release_test.cc
@@ -25,6 +25,7 @@
#endif
#include <utils/db_utils.h>
+#include <cstdio>
#include <cstring>
#include <string>
@@ -54,13 +55,27 @@ 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);
}
+TEST_F(CReleaseTest, RejectCorruptedFileWithoutDoubleClosingDescriptor) {
+ const char* file_name = "corrupted_empty_file.tsfile";
+ remove(file_name);
+ FILE* empty_file = fopen(file_name, "wb");
+ ASSERT_NE(nullptr, empty_file);
+ ASSERT_EQ(0, fclose(empty_file));
+
+ ERRNO error_no = RET_OK;
+ TsFileReader reader = tsfile_reader_new(file_name, &error_no);
+ EXPECT_EQ(nullptr, reader);
+ EXPECT_EQ(RET_TSFILE_CORRUPTED, error_no);
+
+ remove(file_name);
+}
+
TEST_F(CReleaseTest, TsFileWriterNew) {
ERRNO error_code = RET_OK;
diff --git a/cpp/test/cwrapper/query_by_row_cwrapper_test.cc
b/cpp/test/cwrapper/query_by_row_cwrapper_test.cc
index 4983c57ea..6f36c491c 100644
--- a/cpp/test/cwrapper/query_by_row_cwrapper_test.cc
+++ b/cpp/test/cwrapper/query_by_row_cwrapper_test.cc
@@ -193,6 +193,33 @@ TEST_F(CWrapperQueryByRowTest, TreeByRowOffsetLimit) {
storage::libtsfile_destroy();
}
+TEST_F(CWrapperQueryByRowTest, InvalidTreePathReturnsErrorCode) {
+ storage::libtsfile_init();
+
+ const char* file_name = "cwrapper_invalid_tree_path_test.tsfile";
+ remove(file_name);
+ write_tree_tsfile(file_name, {"root.d1"}, {"s1"}, 1);
+
+ ERRNO code = RET_OK;
+ TsFileReader reader = tsfile_reader_new(file_name, &code);
+ ASSERT_EQ(code, RET_OK);
+ ASSERT_NE(reader, nullptr);
+
+ char device_id[] = "root.d1";
+ char invalid_measurement[] = "a*%";
+ char* device_ids[] = {device_id};
+ char* measurement_ids[] = {invalid_measurement};
+ ResultSet result = tsfile_reader_query_tree_by_row(
+ reader, device_ids, 1, measurement_ids, 1, 0, -1, &code);
+
+ EXPECT_EQ(code, RET_INVALID_PATH);
+ EXPECT_EQ(result, nullptr);
+ EXPECT_EQ(tsfile_reader_close(reader), RET_OK);
+ remove(file_name);
+
+ storage::libtsfile_destroy();
+}
+
TEST_F(CWrapperQueryByRowTest, TableByRowOffsetLimit) {
storage::libtsfile_init();
diff --git a/cpp/tools/format/output_format.cc
b/cpp/tools/format/output_format.cc
index e9ae8cb0b..2962cda25 100644
--- a/cpp/tools/format/output_format.cc
+++ b/cpp/tools/format/output_format.cc
@@ -58,8 +58,6 @@ const char* error_code_message(int code) {
return "table does not exist";
case common::E_COLUMN_NOT_EXIST:
return "column does not exist";
- case common::E_INVALID_QUERY:
- return "invalid query";
case common::E_TYPE_NOT_SUPPORTED:
return "data type not supported";
case common::E_TYPE_NOT_MATCH:
diff --git a/python/tests/test_exceptions.py b/python/tests/test_exceptions.py
new file mode 100644
index 000000000..dbc12e792
--- /dev/null
+++ b/python/tests/test_exceptions.py
@@ -0,0 +1,142 @@
+# 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,
+ InvalidPathError,
+ 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"
+
+ retired_codes = {
+ *range(9, 20),
+ 23,
+ 25,
+ 39,
+ 41,
+ 42,
+ 46,
+ 47,
+ }
+ assert retired_codes.isdisjoint(error.value for error in ErrorCode)
+ for code in retired_codes:
+ retired = get_exception(code)
+ assert type(retired) is LibraryError
+ assert retired.code == code
+ assert retired.message == "Unknown library error"
+
+ invalid_path = get_exception(37)
+ assert isinstance(invalid_path, InvalidPathError)
+ assert invalid_path.code == ErrorCode.INVALID_PATH
+
+
+def test_invalid_tree_path_propagates_native_error(tmp_path):
+ path = tmp_path / "invalid-path.tsfile"
+ with TsFileWriter(str(path)) as writer:
+ writer.register_timeseries(
+ "root.device", TimeseriesSchema("value", TSDataType.INT64)
+ )
+
+ with TsFileReader(str(path)) as reader:
+ with pytest.raises(InvalidPathError) as exc_info:
+ reader.query_timeseries("root.device", ["a*%"], 0, 2)
+
+ assert exc_info.value.code == ErrorCode.INVALID_PATH
+
+
+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..c0a963862 100644
--- a/python/tsfile/exceptions.py
+++ b/python/tsfile/exceptions.py
@@ -17,6 +17,48 @@
#
+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
+ OVERFLOW = 20
+ NO_MORE_DATA = 21
+ OUT_OF_ORDER = 22
+ DATA_INCONSISTENCY = 24
+ 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
+ NOT_SUPPORTED = 40
+ INVALID_DATA_POINT = 43
+ DEVICE_NOT_EXIST = 44
+ MEASUREMENT_NOT_EXIST = 45
+ 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 +102,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 +157,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
@@ -110,11 +192,6 @@ class MeasurementNotExistError(LibraryError):
_default_code = 45
-class InvalidQueryError(LibraryError):
- _default_message = "Malformed query syntax"
- _default_code = 46
-
-
class CompressionError(LibraryError):
_default_message = "Data compression/decompression failed"
_default_code = 48
@@ -140,6 +217,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 +244,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 +257,41 @@ ERROR_MAPPING = {
31: FileReadError,
32: FileSyncError,
33: MetadataError,
+ 34: FileStatError,
+ 35: TsFileCorruptedError,
36: BufferNotEnoughError,
+ 37: InvalidPathError,
40: NotSupportedError,
44: DeviceNotExistError,
45: MeasurementNotExistError,
- 46: InvalidQueryError,
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..936c74af1 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,18 +765,18 @@ 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 TsFileWriter writer = NULL
cdef bytes encoded_path = PyUnicode_AsUTF8String(pathname)
cdef const char * c_path = encoded_path
writer = _tsfile_writer_new(c_path, memory_threshold, &errno)
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 TsFileReader reader = NULL
cdef bytes encoded_path = PyUnicode_AsUTF8String(pathname)
cdef const char * c_path = encoded_path
reader = tsfile_reader_new(c_path, &errno)
diff --git a/python/tsfile/tsfile_reader.pyx b/python/tsfile/tsfile_reader.pyx
index f94d42053..36374adde 100644
--- a/python/tsfile/tsfile_reader.pyx
+++ b/python/tsfile/tsfile_reader.pyx
@@ -58,6 +58,7 @@ cdef class ResultSetPy:
cdef object is_tree
def __init__(self, tsfile_reader : TsFileReaderPy, is_tree: bint = False):
+ self.result = NULL
self.metadata = None
self.valid = True
self.tsfile_reader = weakref.ref(tsfile_reader)
@@ -164,9 +165,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:
@@ -296,7 +297,10 @@ cdef class ResultSetPy:
self.close()
def __dealloc__(self):
- self.close()
+ try:
+ self.close()
+ except Exception:
+ pass
def __enter__(self):
return self
@@ -321,8 +325,9 @@ cdef class TsFileReaderPy:
"""
Initialize a TsFile reader for the specified file path.
"""
- self.init_reader(pathname)
+ self.reader = NULL
self.activate_result_set_list = weakref.WeakSet()
+ self.init_reader(pathname)
cdef init_reader(self, pathname):
self.reader = tsfile_reader_new_c(pathname)
@@ -533,7 +538,10 @@ cdef class TsFileReaderPy:
return self.activate_result_set_list
def __dealloc__(self):
- self.close()
+ try:
+ self.close()
+ except Exception:
+ pass
def __enter__(self):
return self
diff --git a/python/tsfile/tsfile_writer.pyx b/python/tsfile/tsfile_writer.pyx
index 30a29d3bb..9e84d83c0 100644
--- a/python/tsfile/tsfile_writer.pyx
+++ b/python/tsfile/tsfile_writer.pyx
@@ -33,6 +33,7 @@ cdef class TsFileWriterPy:
cdef TsFileWriter writer
def __init__(self, pathname: str, memory_threshold: int = 128 * 1024 *
1024):
+ self.writer = NULL
self.writer = tsfile_writer_new_c(pathname, memory_threshold)
def register_timeseries(self, device_name : str, timeseries_schema :
TimeseriesSchemaPy):
@@ -185,7 +186,10 @@ cdef class TsFileWriterPy:
check_error(errno)
def __dealloc__(self):
- self.close()
+ try:
+ self.close()
+ except Exception:
+ pass
def __enter__(self):
return self
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
