You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2025/05/11 16:17:36 UTC
(tsfile) 01/01: add timestamp check.
This is an automated email from the ASF dual-hosted git repository.
colinlee pushed a commit to branch colin_fix_timestamp_check
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit c34969a65ca39fc774971258fa51e0b7f19c0498
Author: ColinLee <sh...@163.com>
AuthorDate: Mon May 12 00:17:04 2025 +0800
add timestamp check.
---
cpp/src/common/statistic.h | 5 +++
cpp/src/writer/time_page_writer.h | 4 ++
cpp/src/writer/tsfile_writer.cc | 4 +-
.../writer/table_view/tsfile_writer_table_test.cc | 47 ++++++++++++++++++++++
4 files changed, 59 insertions(+), 1 deletion(-)
diff --git a/cpp/src/common/statistic.h b/cpp/src/common/statistic.h
index 8eb866e2..7590a6e0 100644
--- a/cpp/src/common/statistic.h
+++ b/cpp/src/common/statistic.h
@@ -164,6 +164,11 @@ class Statistic {
ASSERT(false);
return 0;
}
+
+ int get_count() const { return count_; }
+
+ int64_t get_end_time() const { return end_time_; }
+
virtual int deserialize_from(common::ByteStream &in) {
int ret = common::E_OK;
if (RET_FAIL(common::SerializationUtil::read_var_uint(
diff --git a/cpp/src/writer/time_page_writer.h b/cpp/src/writer/time_page_writer.h
index bc5b8212..bbf70165 100644
--- a/cpp/src/writer/time_page_writer.h
+++ b/cpp/src/writer/time_page_writer.h
@@ -73,6 +73,10 @@ class TimePageWriter {
FORCE_INLINE int write(int64_t timestamp) {
int ret = common::E_OK;
+ if (statistic_->count_ != 0 && is_inited_ &&
+ timestamp <= statistic_->end_time_) {
+ return common::E_OUT_OF_ORDER;
+ }
if (RET_FAIL(time_encoder_->encode(timestamp, time_out_stream_))) {
} else {
statistic_->update(timestamp);
diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc
index a570fc54..5b031189 100644
--- a/cpp/src/writer/tsfile_writer.cc
+++ b/cpp/src/writer/tsfile_writer.cc
@@ -730,7 +730,9 @@ int TsFileWriter::write_table(Tablet &tablet) {
return ret;
}
for (int i = start_idx; i < end_idx; i++) {
- time_chunk_writer->write(tablet.timestamps_[i]);
+ if (RET_FAIL(time_chunk_writer->write(tablet.timestamps_[i]))) {
+ return ret;
+ }
}
uint32_t field_col_count = 0;
for (uint32_t i = 0; i < tablet.get_column_count(); ++i) {
diff --git a/cpp/test/writer/table_view/tsfile_writer_table_test.cc b/cpp/test/writer/table_view/tsfile_writer_table_test.cc
index ac2f4040..121e63a2 100644
--- a/cpp/test/writer/table_view/tsfile_writer_table_test.cc
+++ b/cpp/test/writer/table_view/tsfile_writer_table_test.cc
@@ -139,6 +139,53 @@ TEST_F(TsFileWriterTableTest, WriteTableTest) {
delete table_schema;
}
+TEST_F(TsFileWriterTableTest, WriteDisorderTest) {
+ auto table_schema = gen_table_schema(0);
+ auto tsfile_table_writer_ =
+ std::make_shared<TsFileTableWriter>(&write_file_, table_schema);
+
+ int device_num = 1;
+ int num_timestamp_per_device = 10;
+ int offset = 0;
+ storage::Tablet tablet(table_schema->get_measurement_names(),
+ table_schema->get_data_types(),
+ device_num * num_timestamp_per_device);
+
+ char* literal = new char[std::strlen("device_id") + 1];
+ std::strcpy(literal, "device_id");
+ String literal_str(literal, std::strlen("device_id"));
+ for (int i = 0; i < device_num; i++) {
+ for (int l = 0; l < num_timestamp_per_device; l++) {
+ int row_index = i * num_timestamp_per_device + l;
+ // disordered timestamp.
+ tablet.add_timestamp(row_index, l > num_timestamp_per_device / 2 ? l - num_timestamp_per_device : offset + l);
+ auto column_schemas = table_schema->get_measurement_schemas();
+ for (const auto& column_schema : column_schemas) {
+ switch (column_schema->data_type_) {
+ case TSDataType::INT64:
+ tablet.add_value(row_index,
+ column_schema->measurement_name_,
+ static_cast<int64_t>(i));
+ break;
+ case TSDataType::STRING:
+ tablet.add_value(row_index,
+ column_schema->measurement_name_,
+ literal_str);
+ break;
+ default:
+ break;
+ }
+ }
+ }
+ }
+ delete[] literal;
+
+ ASSERT_EQ(tsfile_table_writer_->write_table(tablet), common::E_OUT_OF_ORDER);
+ ASSERT_EQ(tsfile_table_writer_->flush(), common::E_OK);
+ ASSERT_EQ(tsfile_table_writer_->close(), common::E_OK);
+ delete table_schema;
+}
+
TEST_F(TsFileWriterTableTest, WriteTableTestMultiFlush) {
auto table_schema = gen_table_schema(0);
auto tsfile_table_writer_ = std::make_shared<TsFileTableWriter>(