This is an automated email from the ASF dual-hosted git repository.
jiangtian 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 6f08fb13 Colin fix config (#493)
6f08fb13 is described below
commit 6f08fb13faf35f21802f93138aa3bfc2ecd311c0
Author: Colin Lee <[email protected]>
AuthorDate: Tue May 13 16:42:49 2025 +0800
Colin fix config (#493)
* add set and get config in tsfile py.
* fix compile on linux.
* add type and encoding check.
---
.gitignore | 1 +
cpp/src/common/config/config.h | 7 ++
cpp/src/common/db_common.h | 42 ---------
cpp/src/common/global.cc | 26 ++++--
cpp/src/common/global.h | 92 ++++++++++++++++++++
cpp/src/common/schema.h | 6 +-
cpp/src/encoding/gorilla_decoder.h | 2 +-
cpp/src/reader/tsfile_series_scan_iterator.cc | 9 +-
cpp/src/utils/db_utils.h | 6 +-
cpp/test/cwrapper/c_release_test.cc | 8 +-
python/setup.py | 46 ++++++----
python/tests/test_write_and_read.py | 60 ++++++++++++-
python/tsfile/__init__.py | 1 +
python/tsfile/exceptions.py | 5 ++
python/tsfile/tsfile_cpp.pxd | 34 +++++++-
python/tsfile/tsfile_py_cpp.pxd | 4 +-
python/tsfile/tsfile_py_cpp.pyx | 117 ++++++++++++++++++++++++++
python/tsfile/tsfile_table_writer.py | 5 +-
python/tsfile/tsfile_writer.pyx | 2 +-
19 files changed, 384 insertions(+), 89 deletions(-)
diff --git a/.gitignore b/.gitignore
index b9070546..4fa45012 100644
--- a/.gitignore
+++ b/.gitignore
@@ -35,6 +35,7 @@ python/data
python/venv/*
python/tests/__pycache__/*
python/tests/*.tsfile
+python/tsfile/include
cpp/cmake-build-debug-mingw/
cpp/third_party/googletest-release-1.12.1.zip
diff --git a/cpp/src/common/config/config.h b/cpp/src/common/config/config.h
index 1b9b27d2..e18f6d9c 100644
--- a/cpp/src/common/config/config.h
+++ b/cpp/src/common/config/config.h
@@ -45,6 +45,13 @@ typedef struct ConfigValue {
int32_t chunk_group_size_threshold_;
int32_t record_count_for_next_mem_check_;
bool encrypt_flag_ = false;
+ TSEncoding boolean_encoding_type_;
+ TSEncoding int32_encoding_type_;
+ TSEncoding int64_encoding_type_;
+ TSEncoding float_encoding_type_;
+ TSEncoding double_encoding_type_;
+ TSEncoding string_encoding_type_;
+ CompressionType default_compression_type_;
} ConfigValue;
extern void init_config_value();
diff --git a/cpp/src/common/db_common.h b/cpp/src/common/db_common.h
index b06c8fe4..5fe8b4fa 100644
--- a/cpp/src/common/db_common.h
+++ b/cpp/src/common/db_common.h
@@ -100,48 +100,6 @@ FORCE_INLINE const char*
get_compression_name(CompressionType type) {
return s_compression_names[type];
}
-FORCE_INLINE TSEncoding get_default_encoding_for_type(TSDataType type) {
- if (type == common::BOOLEAN) {
- return PLAIN;
- } else if (type == common::INT32) {
- return PLAIN;
- } else if (type == common::INT64) {
- return PLAIN;
- } else if (type == common::FLOAT) {
- return PLAIN;
- } else if (type == common::DOUBLE) {
- return PLAIN;
- } else if (type == common::TEXT) {
- return PLAIN;
- } else if (type == common::STRING) {
- return PLAIN;
- } else {
- ASSERT(false);
- }
- return INVALID_ENCODING;
-}
-
-FORCE_INLINE CompressionType get_default_compression_for_type(TSDataType type)
{
- if (type == common::BOOLEAN) {
- return UNCOMPRESSED;
- } else if (type == common::INT32) {
- return UNCOMPRESSED;
- } else if (type == common::INT64) {
- return UNCOMPRESSED;
- } else if (type == common::FLOAT) {
- return UNCOMPRESSED;
- } else if (type == common::DOUBLE) {
- return UNCOMPRESSED;
- } else if (type == common::TEXT) {
- return UNCOMPRESSED;
- } else if (type == common::STRING) {
- return UNCOMPRESSED;
- } else {
- ASSERT(false);
- }
- return INVALID_COMPRESSION;
-}
-
enum Ordering { DESC, ASC };
template <typename T>
diff --git a/cpp/src/common/global.cc b/cpp/src/common/global.cc
index c8ea574c..32b93712 100644
--- a/cpp/src/common/global.cc
+++ b/cpp/src/common/global.cc
@@ -44,24 +44,32 @@ void init_config_value() {
g_config_value_.time_encoding_type_ = TS_2DIFF;
g_config_value_.time_data_type_ = INT64;
g_config_value_.time_compress_type_ = LZ4;
+ // Not support RLE yet.
+ g_config_value_.boolean_encoding_type_ = PLAIN;
+ g_config_value_.int32_encoding_type_ = TS_2DIFF;
+ g_config_value_.int64_encoding_type_ = TS_2DIFF;
+ g_config_value_.float_encoding_type_ = GORILLA;
+ g_config_value_.double_encoding_type_ = GORILLA;
+ // Default compression type is LZ4
+ g_config_value_.default_compression_type_ = LZ4;
}
extern TSEncoding get_value_encoder(TSDataType data_type) {
switch (data_type) {
case BOOLEAN:
- return TSEncoding::RLE;
+ return g_config_value_.boolean_encoding_type_;
case INT32:
- return TSEncoding::TS_2DIFF;
+ return g_config_value_.int32_encoding_type_;
case INT64:
- return TSEncoding::TS_2DIFF;
+ return g_config_value_.int64_encoding_type_;
case FLOAT:
- return TSEncoding::GORILLA;
+ return g_config_value_.float_encoding_type_;
case DOUBLE:
- return TSEncoding::GORILLA;
+ return g_config_value_.double_encoding_type_;
case TEXT:
- return TSEncoding::PLAIN;
+ return g_config_value_.string_encoding_type_;
case STRING:
- return TSEncoding::PLAIN;
+ return g_config_value_.string_encoding_type_;
case VECTOR:
break;
case NULL_TYPE:
@@ -75,7 +83,7 @@ extern TSEncoding get_value_encoder(TSDataType data_type) {
}
extern CompressionType get_default_compressor() {
- return LZ4;
+ return g_config_value_.default_compression_type_;
}
void config_set_page_max_point_count(uint32_t page_max_point_count) {
@@ -87,7 +95,7 @@ void config_set_max_degree_of_index_node(uint32_t
max_degree_of_index_node) {
}
void set_config_value() {}
-const char* s_data_type_names[8] = {"BOOLEAN", "INT32", "INT64", "FLOAT",
+const char* s_data_type_names[8] = {"BOOLEAN", "INT32", "INT64", "FLOAT",
"DOUBLE", "TEXT", "VECTOR", "STRING"};
const char* s_encoding_names[12] = {
diff --git a/cpp/src/common/global.h b/cpp/src/common/global.h
index 3a288cd6..3f331460 100644
--- a/cpp/src/common/global.h
+++ b/cpp/src/common/global.h
@@ -28,6 +28,98 @@ namespace common {
extern ConfigValue g_config_value_;
extern ColumnSchema g_time_column_schema;
+
+FORCE_INLINE int set_global_time_data_type(uint8_t data_type) {
+ ASSERT(data_type >= BOOLEAN && data_type <= STRING);
+ if (data_type != INT64) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.time_data_type_ = static_cast<TSDataType>(data_type);
+ return E_OK;
+}
+
+FORCE_INLINE int set_global_time_encoding(uint8_t encoding) {
+ ASSERT(encoding >= PLAIN && encoding <= FREQ);
+ if (encoding != TS_2DIFF && encoding != PLAIN) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.time_encoding_type_ = static_cast<TSEncoding>(encoding);
+ return E_OK;
+}
+
+FORCE_INLINE int set_global_time_compression(uint8_t compression) {
+ ASSERT(compression >= UNCOMPRESSED && compression <= LZ4);
+ if (compression != UNCOMPRESSED && compression != LZ4) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.time_compress_type_ =
+ static_cast<CompressionType>(compression);
+ return E_OK;
+}
+
+FORCE_INLINE int set_datatype_encoding(uint8_t data_type, uint8_t encoding) {
+ int code = E_OK;
+ TSDataType dtype = static_cast<TSDataType>(data_type);
+ ASSERT(dtype >= BOOLEAN && dtype <= STRING);
+ TSEncoding encoding_type = static_cast<TSEncoding>(encoding);
+ ASSERT(encoding >= PLAIN && encoding <= FREQ);
+ switch (dtype) {
+ case BOOLEAN:
+ if (encoding_type != PLAIN) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.boolean_encoding_type_ = encoding_type;
+ break;
+ case INT32:
+ if (encoding_type != PLAIN && encoding_type != TS_2DIFF &&
+ encoding_type != GORILLA) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.int32_encoding_type_ = encoding_type;
+ break;
+ case INT64:
+ if (encoding_type != PLAIN && encoding_type != TS_2DIFF &&
+ encoding_type != GORILLA) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.int64_encoding_type_ = encoding_type;
+ break;
+ case STRING:
+ if (encoding_type != PLAIN) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.string_encoding_type_ = encoding_type;
+ break;
+ case FLOAT:
+ if (encoding_type != PLAIN && encoding_type != TS_2DIFF &&
+ encoding_type != GORILLA) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.float_encoding_type_ = encoding_type;
+ break;
+ case DOUBLE:
+ if (encoding_type != PLAIN && encoding_type != TS_2DIFF &&
+ encoding_type != GORILLA) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.double_encoding_type_ = encoding_type;
+ break;
+ default:
+ break;
+ }
+ return E_OK;
+}
+
+FORCE_INLINE int set_global_compression(uint8_t compression) {
+ ASSERT(compression >= UNCOMPRESSED && compression <= LZ4);
+ if (compression != UNCOMPRESSED && compression != LZ4) {
+ return E_NOT_SUPPORT;
+ }
+ g_config_value_.default_compression_type_ =
+ static_cast<CompressionType>(compression);
+ return E_OK;
+}
+
extern int init_common();
extern bool is_timestamp_column_name(const char *time_col_name);
extern void cols_to_json(ByteStream *byte_stream,
diff --git a/cpp/src/common/schema.h b/cpp/src/common/schema.h
index 72fd028f..5deffad9 100644
--- a/cpp/src/common/schema.h
+++ b/cpp/src/common/schema.h
@@ -62,8 +62,8 @@ struct MeasurementSchema {
common::TSDataType data_type)
: measurement_name_(measurement_name),
data_type_(data_type),
- encoding_(get_default_encoding_for_type(data_type)),
- compression_type_(common::UNCOMPRESSED),
+ encoding_(common::get_value_encoder(data_type)),
+ compression_type_(common::get_default_compressor()),
chunk_writer_(nullptr),
value_chunk_writer_(nullptr) {}
@@ -238,7 +238,6 @@ class TableSchema {
column_schemas_(std::move(other.column_schemas_)),
column_categories_(std::move(other.column_categories_)) {}
-
TableSchema(const TableSchema &other) noexcept
: table_name_(other.table_name_),
column_categories_(other.column_categories_) {
@@ -410,7 +409,6 @@ class TableSchema {
}
private:
-
std::string table_name_;
std::vector<std::shared_ptr<MeasurementSchema> > column_schemas_;
std::vector<common::ColumnCategory> column_categories_;
diff --git a/cpp/src/encoding/gorilla_decoder.h
b/cpp/src/encoding/gorilla_decoder.h
index 5b241de4..f374b32e 100644
--- a/cpp/src/encoding/gorilla_decoder.h
+++ b/cpp/src/encoding/gorilla_decoder.h
@@ -44,7 +44,7 @@ class GorillaDecoder : public Decoder {
stored_trailing_zeros_ = 0;
bits_left_ = 0;
first_value_was_read_ = false;
- has_next_ = true;
+ has_next_ = false;
buffer_ = 0;
}
diff --git a/cpp/src/reader/tsfile_series_scan_iterator.cc
b/cpp/src/reader/tsfile_series_scan_iterator.cc
index eb3bd1ed..b1be6835 100644
--- a/cpp/src/reader/tsfile_series_scan_iterator.cc
+++ b/cpp/src/reader/tsfile_series_scan_iterator.cc
@@ -79,8 +79,7 @@ int TsFileSeriesScanIterator::get_next(TsBlock *&ret_tsblock,
bool alloc,
if (alloc) {
ret_tsblock = alloc_tsblock();
}
- ret = chunk_reader_->get_next_page(ret_tsblock, filter,
- *data_pa_);
+ ret = chunk_reader_->get_next_page(ret_tsblock, filter, *data_pa_);
}
return ret;
}
@@ -139,8 +138,10 @@ TsBlock *TsFileSeriesScanIterator::alloc_tsblock() {
ChunkHeader &ch = chunk_reader_->get_chunk_header();
// TODO config
- ColumnSchema time_cd("time", common::INT64, common::SNAPPY,
common::TS_2DIFF);
- ColumnSchema value_cd(ch.measurement_name_, ch.data_type_,
ch.compression_type_, ch.encoding_type_);
+ ColumnSchema time_cd("time", common::INT64, common::SNAPPY,
+ common::TS_2DIFF);
+ ColumnSchema value_cd(ch.measurement_name_, ch.data_type_,
+ ch.compression_type_, ch.encoding_type_);
tuple_desc_.push_back(time_cd);
tuple_desc_.push_back(value_cd);
diff --git a/cpp/src/utils/db_utils.h b/cpp/src/utils/db_utils.h
index 508d3a29..4b5aca9b 100644
--- a/cpp/src/utils/db_utils.h
+++ b/cpp/src/utils/db_utils.h
@@ -34,6 +34,8 @@
#include "utils/util_define.h"
namespace common {
+extern TSEncoding get_value_encoder(TSDataType data_type);
+extern CompressionType get_default_compressor();
typedef struct FileID {
int64_t seq_; // timestamp when create
@@ -285,8 +287,8 @@ struct ColumnSchema {
ColumnCategory column_category = ColumnCategory::FIELD)
: column_name_(std::move(column_name)),
data_type_(data_type),
- compression_(get_default_compression_for_type(data_type)),
- encoding_(get_default_encoding_for_type(data_type)),
+ compression_(get_default_compressor()),
+ encoding_(get_value_encoder(data_type)),
column_category_(column_category) {}
const std::string &get_column_name() const { return column_name_; }
diff --git a/cpp/test/cwrapper/c_release_test.cc
b/cpp/test/cwrapper/c_release_test.cc
index 86f1b697..bb73fb9d 100644
--- a/cpp/test/cwrapper/c_release_test.cc
+++ b/cpp/test/cwrapper/c_release_test.cc
@@ -112,9 +112,10 @@ TEST_F(CReleaseTest, TsFileWriterNew) {
}
TEST_F(CReleaseTest, TsFileWriterWriteDataAbnormalColumn) {
+ remove("TsFileWriterWriteDataAbnormalColumn.tsfile");
ERRNO error_code = RET_OK;
WriteFile file = write_file_new(
- "TsFileWriterWriteDataAbnormalColumn_3_100.tsfile", &error_code);
+ "TsFileWriterWriteDataAbnormalColumn.tsfile", &error_code);
TableSchema abnormal_schema;
abnormal_schema.table_name = strdup("!@#$%^*()_+-=");
@@ -184,7 +185,7 @@ TEST_F(CReleaseTest, TsFileWriterWriteDataAbnormalColumn) {
free_write_file(&file);
TsFileReader reader = tsfile_reader_new(
- "TsFileWriterWriteDataAbnormalColumn_3_100.tsfile", &error_code);
+ "TsFileWriterWriteDataAbnormalColumn.tsfile", &error_code);
ASSERT_EQ(RET_OK, error_code);
int i = 0;
ResultSet result_set = tsfile_query_table(
@@ -212,11 +213,12 @@ TEST_F(CReleaseTest, TsFileWriterWriteDataAbnormalColumn)
{
free_tablet(&tablet);
free_tsfile_result_set(&result_set);
tsfile_reader_close(reader);
- remove("TsFileWriterWriteDataAbnormalColumn_3_100.tsfile");
+ remove("TsFileWriterWriteDataAbnormalColumn.tsfile");
}
TEST_F(CReleaseTest, TsFileWriterMultiDataType) {
ERRNO error_code = RET_OK;
+ remove("TsFileWriterMultiDataType.tsfile");
WriteFile file = write_file_new(
"TsFileWriterMultiDataType.tsfile", &error_code);
ASSERT_EQ(RET_OK, error_code);
diff --git a/python/setup.py b/python/setup.py
index 6edeea0b..329cc2aa 100644
--- a/python/setup.py
+++ b/python/setup.py
@@ -16,17 +16,19 @@
# under the License.
#
-from setuptools import setup, Extension
-from setuptools.command.build_ext import build_ext
-from Cython.Build import cythonize
-import numpy as np
+import os
import platform
import shutil
-import os
+
+import numpy as np
+from Cython.Build import cythonize
+from setuptools import setup, Extension
+from setuptools.command.build_ext import build_ext
version = "2.1.0.dev0"
system = platform.system()
+
def copy_tsfile_lib(source_dir, target_dir, suffix):
lib_file_name = f"libtsfile.{suffix}"
source = os.path.join(source_dir, lib_file_name)
@@ -47,19 +49,32 @@ def copy_tsfile_lib(source_dir, target_dir, suffix):
os.symlink(lib_file_name, link_name)
-def copy_tsfile_header(source, target):
- if os.path.exists(source):
- shutil.copyfile(source, target)
-
project_dir = os.path.dirname(os.path.abspath(__file__))
+tsfile_py_include = os.path.join(project_dir, "tsfile", "include")
+
+if os.path.exists(tsfile_py_include):
+ shutil.rmtree(tsfile_py_include)
+
+shutil.copytree(
+ os.path.join(project_dir, "..", "cpp", "target", "build", "include"),
+ os.path.join(tsfile_py_include, ""),
+)
+
+
+def copy_tsfile_header(source):
+ for file in source:
+ if os.path.exists(file):
+ target = os.path.join(tsfile_py_include, os.path.basename(file))
+ shutil.copyfile(file, target)
+
## Copy C wrapper header.
# tsfile/cpp/src/cwrapper/tsfile_cwrapper.h
-tsfile_c_include_file = os.path.join(
- project_dir, "..", "cpp", "src", "cwrapper", "tsfile_cwrapper.h"
-)
-tsfile_py_include_file = os.path.join(project_dir, "tsfile",
"tsfile_cwrapper.h")
-copy_tsfile_header(tsfile_c_include_file, tsfile_py_include_file)
+source_headers = [
+ os.path.join(project_dir, "..", "cpp", "src", "cwrapper",
"tsfile_cwrapper.h"),
+]
+
+copy_tsfile_header(source_headers)
## Copy shared library
tsfile_shared_source_dir = os.path.join(project_dir, "..", "cpp", "target",
"build", "lib")
@@ -72,8 +87,7 @@ elif system == "Linux":
else:
copy_tsfile_lib(tsfile_shared_source_dir, tsfile_shared_dir, "dll")
-tsfile_include_dir=os.path.join(project_dir, "tsfile")
-
+tsfile_include_dir = os.path.join(project_dir, "tsfile", "include")
ext_modules_tsfile = [
# utils: from python to c or c to python.
diff --git a/python/tests/test_write_and_read.py
b/python/tests/test_write_and_read.py
index 6d0dbe7a..78731899 100644
--- a/python/tests/test_write_and_read.py
+++ b/python/tests/test_write_and_read.py
@@ -20,12 +20,13 @@ import os
import pytest
-from tsfile import ColumnSchema, TableSchema
+from tsfile import ColumnSchema, TableSchema, TSEncoding, NotSupportedError
from tsfile import TSDataType
from tsfile import Tablet, RowRecord, Field
from tsfile import TimeseriesSchema
from tsfile import TsFileTableWriter
from tsfile import TsFileWriter, TsFileReader, ColumnCategory
+from tsfile import Compressor
def test_row_record_write_and_read():
@@ -59,7 +60,7 @@ def test_row_record_write_and_read():
if os.path.exists("record_write_and_read.tsfile"):
os.remove("record_write_and_read.tsfile")
-
[email protected](reason="API not match")
def test_tablet_write_and_read():
try:
if os.path.exists("record_write_and_read.tsfile"):
@@ -92,6 +93,8 @@ def test_tablet_write_and_read():
while result.next():
assert result.is_null_by_index(1) == False
assert result.get_value_by_index(1) == row_num
+ # Here, the data retrieval uses the table model's API,
+ # which might be incompatible. Therefore, it is better to skip it
for now.
assert result.get_value_by_name("level0") == row_num
row_num = row_num + 1
@@ -201,7 +204,7 @@ def test_lower_case_name():
for i in range(100):
tablet.add_timestamp(i, i)
tablet.add_value_by_name("device", i, "device" + str(i))
- tablet.add_value_by_name("valuE", i, i * 1.1)
+ tablet.add_value_by_name("valuE", i, i * 1.1)
writer.write_table(tablet)
@@ -214,3 +217,54 @@ def test_lower_case_name():
assert data_frame["value"].sum() == 5445.0
+def test_tsfile_config():
+ from tsfile import get_tsfile_config, set_tsfile_config
+
+ config = get_tsfile_config()
+
+ table = TableSchema("tEst_Table",
+ [ColumnSchema("Device", TSDataType.STRING,
ColumnCategory.TAG),
+ ColumnSchema("vAlue", TSDataType.DOUBLE,
ColumnCategory.FIELD)])
+ if os.path.exists("test1.tsfile"):
+ os.remove("test1.tsfile")
+ with TsFileTableWriter("test1.tsfile", table) as writer:
+ tablet = Tablet(["device", "VALUE"], [TSDataType.STRING,
TSDataType.DOUBLE])
+ for i in range(100):
+ tablet.add_timestamp(i, i)
+ tablet.add_value_by_name("device", i, "device" + str(i))
+ tablet.add_value_by_name("valuE", i, i * 1.1)
+
+ writer.write_table(tablet)
+
+ config_normal = get_tsfile_config()
+ print(config_normal)
+ assert config_normal["chunk_group_size_threshold_"] == 128 * 1024 * 1024
+
+ os.remove("test1.tsfile")
+ with TsFileTableWriter("test1.tsfile", table, 100 * 100) as writer:
+ tablet = Tablet(["device", "VALUE"], [TSDataType.STRING,
TSDataType.DOUBLE])
+ for i in range(100):
+ tablet.add_timestamp(i, i)
+ tablet.add_value_by_name("device", i, "device" + str(i))
+ tablet.add_value_by_name("valuE", i, i * 1.1)
+
+ writer.write_table(tablet)
+ config_modified = get_tsfile_config()
+ assert config_normal != config_modified
+ assert config_modified["chunk_group_size_threshold_"] == 100 * 100
+ set_tsfile_config({'chunk_group_size_threshold_': 100 * 20})
+ assert get_tsfile_config()["chunk_group_size_threshold_"] == 100 * 20
+ with pytest.raises(TypeError):
+ set_tsfile_config({"time_compress_type_": TSDataType.DOUBLE})
+ with pytest.raises(TypeError):
+ set_tsfile_config({'chunk_group_size_threshold_': -1 * 100 * 20})
+
+ set_tsfile_config({'float_encoding_type_': TSEncoding.PLAIN})
+ assert get_tsfile_config()["float_encoding_type_"] == TSEncoding.PLAIN
+
+ with pytest.raises(TypeError):
+ set_tsfile_config({"float_encoding_type_": -1 * 100 * 20})
+ with pytest.raises(NotSupportedError):
+ set_tsfile_config({"float_encoding_type_": TSEncoding.BITMAP})
+ with pytest.raises(NotSupportedError):
+ set_tsfile_config({"time_compress_type_": Compressor.PAA})
diff --git a/python/tsfile/__init__.py b/python/tsfile/__init__.py
index df51bcfa..0c5081fa 100644
--- a/python/tsfile/__init__.py
+++ b/python/tsfile/__init__.py
@@ -32,4 +32,5 @@ from .date_utils import *
from .exceptions import *
from .tsfile_reader import TsFileReaderPy as TsFileReader, ResultSetPy as
ResultSet
from .tsfile_writer import TsFileWriterPy as TsFileWriter
+from .tsfile_py_cpp import get_tsfile_config, set_tsfile_config
from .tsfile_table_writer import TsFileTableWriter
\ No newline at end of file
diff --git a/python/tsfile/exceptions.py b/python/tsfile/exceptions.py
index 182186a3..15575954 100644
--- a/python/tsfile/exceptions.py
+++ b/python/tsfile/exceptions.py
@@ -79,6 +79,10 @@ class BufferNotEnoughError(LibraryError):
_default_message = "Insufficient buffer space"
_default_code = 36
+class NotSupportedError(LibraryError):
+ _default_message = "Not support yet"
+ _default_code = 40
+
class DeviceNotExistError(LibraryError):
_default_message = "Requested device does not exist"
_default_code = 44
@@ -124,6 +128,7 @@ ERROR_MAPPING = {
32: FileSyncError,
33: MetadataError,
36: BufferNotEnoughError,
+ 40: NotSupportedError,
44: DeviceNotExistError,
45: MeasurementNotExistError,
46: InvalidQueryError,
diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd
index e65cd214..1b04051c 100644
--- a/python/tsfile/tsfile_cpp.pxd
+++ b/python/tsfile/tsfile_cpp.pxd
@@ -17,7 +17,7 @@
#
#cython: language_level=3
-from libc.stdint cimport uint32_t, int32_t, int64_t, uint64_t
+from libc.stdint cimport uint32_t, int32_t, int64_t, uint64_t, uint8_t
ctypedef int32_t ErrorCode
@@ -193,3 +193,35 @@ cdef extern from "./tsfile_cwrapper.h":
ResultSetMetaData tsfile_result_set_get_metadata(ResultSet result_set);
void free_result_set_meta_data(ResultSetMetaData result_set_meta_data);
+
+
+
+cdef extern from "./common/config/config.h" namespace "common":
+ cdef cppclass ConfigValue:
+ uint32_t tsblock_mem_inc_step_size_
+ uint32_t tsblock_max_memory_
+ uint32_t page_writer_max_point_num_
+ uint32_t page_writer_max_memory_bytes_
+ uint32_t max_degree_of_index_node_
+ double tsfile_index_bloom_filter_error_percent_
+ uint8_t time_encoding_type_
+ uint8_t time_data_type_
+ uint8_t time_compress_type_
+ int32_t chunk_group_size_threshold_
+ int32_t record_count_for_next_mem_check_
+ bint encrypt_flag_
+ uint8_t boolean_encoding_type_;
+ uint8_t int32_encoding_type_;
+ uint8_t int64_encoding_type_;
+ uint8_t float_encoding_type_;
+ uint8_t double_encoding_type_;
+ uint8_t string_encoding_type_;
+ uint8_t default_compression_type_;
+
+cdef extern from "./common/global.h" namespace "common":
+ ConfigValue g_config_value_
+ int set_datatype_encoding(uint8_t data_type, uint8_t encoding)
+ int set_global_compression(uint8_t compression)
+ int set_global_time_data_type(uint8_t data_type);
+ int set_global_time_encoding(uint8_t encoding);
+ int set_global_time_compression(uint8_t compression);
diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd
index c7310d7c..ce907a79 100644
--- a/python/tsfile/tsfile_py_cpp.pxd
+++ b/python/tsfile/tsfile_py_cpp.pxd
@@ -52,4 +52,6 @@ cdef public api ResultSet
tsfile_reader_query_table_c(TsFileReader reader, objec
cdef public api ResultSet tsfile_reader_query_paths_c(TsFileReader reader,
object device_name, object sensor_list, int64_t start_time,
int64_t end_time)
cdef public api object get_table_schema(TsFileReader reader, object table_name)
-cdef public api object get_all_table_schema(TsFileReader reader)
\ No newline at end of file
+cdef public api object get_all_table_schema(TsFileReader reader)
+cpdef public api object get_tsfile_config()
+cpdef public api void set_tsfile_config(dict new_config)
\ No newline at end of file
diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx
index 90b12b4a..e1743039 100644
--- a/python/tsfile/tsfile_py_cpp.pyx
+++ b/python/tsfile/tsfile_py_cpp.pyx
@@ -43,6 +43,7 @@ cdef inline void check_error(int errcode, const char*
context=NULL) except *:
return
exc_type = ERROR_MAPPING.get(errcode)
+ print(exc_type)
exc_instance = exc_type(errcode, "")
PyErr_SetObject(exc_type, exc_instance)
@@ -347,6 +348,122 @@ cdef TsFileReader tsfile_reader_new_c(object pathname)
except +:
check_error(errno)
return reader
+cpdef object get_tsfile_config():
+ return {
+ "tsblock_mem_inc_step_size_":
g_config_value_.tsblock_mem_inc_step_size_,
+ "tsblock_max_memory_": g_config_value_.tsblock_max_memory_,
+ "page_writer_max_point_num_":
g_config_value_.page_writer_max_point_num_,
+ "page_writer_max_memory_bytes_":
g_config_value_.page_writer_max_memory_bytes_,
+ "max_degree_of_index_node_": g_config_value_.max_degree_of_index_node_,
+ "tsfile_index_bloom_filter_error_percent_":
g_config_value_.tsfile_index_bloom_filter_error_percent_,
+
"time_encoding_type_":TSEncodingPy(int(g_config_value_.time_encoding_type_)),
+ "time_data_type_": TSDataTypePy(int(g_config_value_.time_data_type_)),
+ "time_compress_type_":
CompressorPy(int(g_config_value_.time_compress_type_)),
+ "chunk_group_size_threshold_":
g_config_value_.chunk_group_size_threshold_,
+
"record_count_for_next_mem_check_":g_config_value_.record_count_for_next_mem_check_,
+ "encrypt_flag_":g_config_value_.encrypt_flag_,
+
"boolean_encoding_type_":TSEncodingPy(int(g_config_value_.boolean_encoding_type_)),
+ "int32_encoding_type_":
TSEncodingPy(int(g_config_value_.int32_encoding_type_)),
+ "int64_encoding_type_":
TSEncodingPy(int(g_config_value_.int64_encoding_type_)),
+ "float_encoding_type_":
TSEncodingPy(int(g_config_value_.float_encoding_type_)),
+ "double_encoding_type_":
TSEncodingPy(int(g_config_value_.double_encoding_type_)),
+ "string_encoding_type_":
TSEncodingPy(int(g_config_value_.string_encoding_type_)),
+ "default_compression_type_":
CompressorPy(int(g_config_value_.default_compression_type_)),
+ }
+
+
+cpdef void set_tsfile_config(dict new_config):
+ if "tsblock_mem_inc_step_size_" in new_config:
+ _check_uint32(new_config["tsblock_mem_inc_step_size_"])
+ g_config_value_.tsblock_max_memory_ =
new_config["tsblock_mem_inc_step_size_"]
+ if "tsblock_max_memory_" in new_config:
+ _check_uint32(new_config["tsblock_max_memory_"])
+ g_config_value_.tsblock_max_memory_ = new_config["tsblock_max_memory_"]
+ if "page_writer_max_point_num_" in new_config:
+ _check_uint32(new_config["page_writer_max_point_num_"])
+ g_config_value_.page_writer_max_point_num_ =
new_config["page_writer_max_point_num_"]
+ if "page_writer_max_memory_bytes_" in new_config:
+ _check_uint32(new_config["page_writer_max_memory_bytes_"])
+ g_config_value_.page_writer_max_memory_bytes_ =
new_config["page_writer_max_memory_bytes_"]
+ if "max_degree_of_index_node_" in new_config:
+ _check_uint32(new_config["max_degree_of_index_node_"])
+ g_config_value_.max_degree_of_index_node_ =
new_config["max_degree_of_index_node_"]
+ if "tsfile_index_bloom_filter_error_percent_" in new_config:
+ _check_double(new_config["tsfile_index_bloom_filter_error_percent_"])
+ g_config_value_.tsfile_index_bloom_filter_error_percent_ =
new_config["tsfile_index_bloom_filter_error_percent_"]
+ if "time_encoding_type_" in new_config:
+ if not isinstance(new_config["time_encoding_type_"], TSEncodingPy):
+ raise TypeError(f"Unsupported TSEncoding:
{new_config['time_encoding_type_']}")
+ code =
set_global_time_encoding(<uint8_t>(new_config["time_encoding_type_"].value))
+ check_error(code)
+ if "time_data_type_" in new_config:
+ if not isinstance(new_config["time_data_type_"], TSDataTypePy):
+ raise TypeError(f"Unsupported TSDataType:
{new_config['time_data_type_']}")
+ code =
set_global_time_data_type(<uint8_t>(new_config["time_data_type_"].value))
+ check_error(code)
+ if "time_compress_type_" in new_config:
+ if not isinstance(new_config["time_compress_type_"], CompressorPy):
+ raise TypeError(f"Unsupported Compressor:
{new_config['time_compress_type_']}")
+ code =
set_global_time_compression(<uint8_t>(new_config["time_compress_type_"].value))
+ check_error(code)
+ if "chunk_group_size_threshold_" in new_config:
+ _check_uint32(new_config["chunk_group_size_threshold_"])
+ g_config_value_.chunk_group_size_threshold_ =
new_config["chunk_group_size_threshold_"]
+ if "record_count_for_next_mem_check_" in new_config:
+ _check_uint32(new_config["record_count_for_next_mem_check_"])
+ g_config_value_.record_count_for_next_mem_check_ =
new_config["record_count_for_next_mem_check_"]
+ if "encrypt_flag_" in new_config:
+ _check_bool(new_config["encrypt_flag_"])
+ g_config_value_.encrypt_flag_ = <bint>new_config["encrypt_flag_"]
+
+ if "boolean_encoding_type_" in new_config:
+ if not isinstance(new_config["boolean_encoding_type_"], TSEncodingPy):
+ raise TypeError(f"Unsupported TSEncodingType:
{new_config['boolean_encoding_type_']}")
+ code = set_datatype_encoding(TSDataTypePy.BOOLEAN.value,
new_config['boolean_encoding_type_'].value)
+ check_error(code)
+ if "int32_encoding_type_" in new_config:
+ if not isinstance(new_config["int32_encoding_type_"], TSEncodingPy):
+ raise TypeError(f"Unsupported TSEncodingType:
{new_config['int32_encoding_type_']}")
+ code = set_datatype_encoding(TSDataTypePy.INT32.value,
new_config['int32_encoding_type_'].value)
+ check_error(code)
+ if "int64_encoding_type_" in new_config:
+ if not isinstance(new_config["int64_encoding_type_"], TSEncodingPy):
+ raise TypeError(f"Unsupported TSEncodingType:
{new_config['int64_encoding_type_']}")
+ code = set_datatype_encoding(TSDataTypePy.INT64.value,
new_config['int64_encoding_type_'].value)
+ check_error(code)
+ if "float_encoding_type_" in new_config:
+ if not isinstance(new_config["float_encoding_type_"], TSEncodingPy):
+ raise TypeError(f"Unsupported TSEncodingType:
{new_config['float_encoding_type_']}")
+ code = set_datatype_encoding(TSDataTypePy.FLOAT.value,
new_config['float_encoding_type_'].value)
+ check_error(code)
+ if "double_encoding_type_" in new_config:
+ if not isinstance(new_config["double_encoding_type_"], TSEncodingPy):
+ raise TypeError(f"Unsupported TSEncodingType:
{new_config['double_encoding_type_']}")
+ code = set_datatype_encoding(TSDataTypePy.DOUBLE.value,
new_config['double_encoding_type_'].value)
+ check_error(code)
+ if "string_encoding_type_" in new_config:
+ if not isinstance(new_config["string_encoding_type_"], TSEncodingPy):
+ raise TypeError(f"Unsupported TSEncodingType:
{new_config['string_encoding_type_']}")
+ code = set_datatype_encoding(TSDataTypePy.STRING.value,
new_config['string_encoding_type_'].value)
+ check_error(code)
+ if "default_compression_type_" in new_config:
+ if not isinstance(new_config["default_compression_type_"],
CompressorPy):
+ raise TypeError(f"Unsupported CompressionType:
{new_config['default_compression_type_']}")
+ code =
set_global_compression(new_config["default_compression_type_"].value)
+ check_error(code)
+
+cdef _check_uint32(value):
+ if not isinstance(value, int) or value < 0 or value > 0xFFFFFFFF:
+ raise TypeError(f"Expected uint32, got {type(value)}")
+
+cdef _check_double(value):
+ if not isinstance(value, (int, float)):
+ raise TypeError(f"Expected float, got {type(value)}")
+
+cdef _check_bool(value):
+ if not isinstance(value, bool):
+ raise TypeError(f"Expected bool, got {type(value)}")
+
# Register table and device
cdef ErrorCode tsfile_writer_register_device_py_cpp(TsFileWriter writer,
DeviceSchema *schema):
cdef ErrorCode errno
diff --git a/python/tsfile/tsfile_table_writer.py
b/python/tsfile/tsfile_table_writer.py
index c7ab9492..28193360 100644
--- a/python/tsfile/tsfile_table_writer.py
+++ b/python/tsfile/tsfile_table_writer.py
@@ -31,12 +31,13 @@ class TsFileTableWriter:
according to that schema, and serialize this data into a TsFile.
"""
- def __init__(self, path: str, table_schema: TableSchema):
+ def __init__(self, path: str, table_schema: TableSchema, memory_threshold
= 128 * 1024 * 1024):
"""
:param path: The path of tsfile, will create if it doesn't exist.
:param table_schema: describes the schema of the tables they want to
write.
+ :param memory_threshold(Byte): memory usage threshold for flushing
data.
"""
- self.writer = TsFileWriter(path)
+ self.writer = TsFileWriter(path, memory_threshold)
self.writer.register_table(table_schema)
self.exclusive_table_name_ = table_schema.get_table_name()
diff --git a/python/tsfile/tsfile_writer.pyx b/python/tsfile/tsfile_writer.pyx
index 4d1f53ed..20199195 100644
--- a/python/tsfile/tsfile_writer.pyx
+++ b/python/tsfile/tsfile_writer.pyx
@@ -29,7 +29,7 @@ from tsfile.tablet import Tablet as TabletPy
cdef class TsFileWriterPy:
cdef TsFileWriter writer
- def __init__(self, pathname, memory_threshold = 128 * 1024 * 1024):
+ def __init__(self, pathname:str, memory_threshold:int = 128 * 1024 * 1024):
self.writer = tsfile_writer_new_c(pathname, memory_threshold)
def register_timeseries(self, device_name : str, timeseries_schema :
TimeseriesSchemaPy):
