From 42c9e3457c3a0a6d2e697f58b372aef961dc881c Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 11 Nov 2026 12:32:17 +0800 Subject: [PATCH 1/7] feat: allow non-decreasing (duplicate) timestamps behind an opt-in flag (#1056) Add an opt-in `allowNonDecreasingTimestamps` flag (default off) across all three implementations so a writer can accept a timestamp equal to the last written one while still rejecting strictly decreasing timestamps. Java: `TSFileConfig` flag threaded through the aligned/non-aligned chunk group writers and the tablet write context. C++: `ConfigValue` field plus a C API setter, applied in TimePageWriter's single and batch paths and the recovered-last-time ordering checks. Python: `allow_non_decreasing_timestamps` keyword argument on the writer constructors. Duplicate timestamps are preserved as separate rows (no write-time dedup). Tests cover flag on/off in Java, C++, and Python, plus a cross-language round-trip (Java writes duplicates, C++ reads them back). --- cpp/src/common/config/config.h | 7 + cpp/src/common/global.h | 8 + cpp/src/cwrapper/tsfile_cwrapper.cc | 5 + cpp/src/cwrapper/tsfile_cwrapper.h | 14 ++ cpp/src/writer/time_page_writer.h | 15 +- cpp/src/writer/tsfile_writer.cc | 74 ++++---- cpp/src/writer/tsfile_writer.h | 8 + .../duplicate_timestamps_roundtrip_test.cc | 68 +++++++ .../fixtures/duplicate_timestamps.tsfile | Bin 0 -> 292 bytes cpp/test/writer/tsfile_writer_test.cc | 169 ++++++++++++++++++ .../tsfile/common/conf/TSFileConfig.java | 16 ++ .../chunk/AlignedChunkGroupWriterImpl.java | 7 +- .../chunk/NonAlignedChunkGroupWriterImpl.java | 9 +- .../write/chunk/TabletWriteContext.java | 9 +- .../AllowNonDecreasingTimestampsTest.java | 160 +++++++++++++++++ .../test_allow_non_decreasing_timestamps.py | 105 +++++++++++ python/tsfile/tsfile_cpp.pxd | 2 + python/tsfile/tsfile_py_cpp.pyx | 5 + python/tsfile/tsfile_table_writer.py | 15 +- python/tsfile/tsfile_writer.pyx | 5 +- 20 files changed, 650 insertions(+), 51 deletions(-) create mode 100644 cpp/test/reader/duplicate_timestamps_roundtrip_test.cc create mode 100644 cpp/test/reader/fixtures/duplicate_timestamps.tsfile create mode 100644 java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java create mode 100644 python/tests/test_allow_non_decreasing_timestamps.py diff --git a/cpp/src/common/config/config.h b/cpp/src/common/config/config.h index 1293ebecb..8230614a3 100644 --- a/cpp/src/common/config/config.h +++ b/cpp/src/common/config/config.h @@ -68,6 +68,13 @@ typedef struct ConfigValue { // throughput: writes return success as soon as data is in the page cache. // Only set to false if the caller drives its own fsync policy. bool sync_on_close_ = true; + // When true, a timestamp equal to the previous one is accepted; only a + // strictly decreasing timestamp is rejected with E_OUT_OF_ORDER. Equal + // timestamps are preserved as separate rows (no write-time dedup). Like + // thread_count_/sync_on_close_ it keeps its in-class default rather than + // being reset by init_config_value(), so a setter called before + // libtsfile_init() is not clobbered. + bool allow_non_decreasing_timestamps_ = false; } ConfigValue; extern void init_config_value(); diff --git a/cpp/src/common/global.h b/cpp/src/common/global.h index 89538c9dc..b7fdaa396 100644 --- a/cpp/src/common/global.h +++ b/cpp/src/common/global.h @@ -209,6 +209,14 @@ FORCE_INLINE bool get_parallel_write_enabled() { return g_config_value_.parallel_write_enabled_; } +FORCE_INLINE void set_allow_non_decreasing_timestamps(bool enabled) { + g_config_value_.allow_non_decreasing_timestamps_ = enabled; +} + +FORCE_INLINE bool get_allow_non_decreasing_timestamps() { + return g_config_value_.allow_non_decreasing_timestamps_; +} + // Select the backend used by subsequently opened local files. Existing // LocalRandomAccessReadFile instances retain the backend selected when opened. // This setting deliberately lives outside exported ConfigValue so adding it diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc index be97ef484..6eaffff55 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.cc +++ b/cpp/src/cwrapper/tsfile_cwrapper.cc @@ -146,6 +146,11 @@ int set_global_compression(uint8_t compression) { return common::set_global_compression(compression); } +int set_allow_non_decreasing_timestamps(bool enabled) { + common::set_allow_non_decreasing_timestamps(enabled); + return common::E_OK; +} + ERRNO tsfile_set_file_read_backend(int32_t backend) { switch (backend) { case TSFILE_READ_BACKEND_AUTO: diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h index 6d9985c01..baa662142 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.h +++ b/cpp/src/cwrapper/tsfile_cwrapper.h @@ -433,6 +433,20 @@ int set_datatype_encoding(uint8_t data_type, uint8_t encoding); */ int set_global_compression(uint8_t compression); +/** + * @brief Allow equal (non-decreasing) timestamps when writing. + * + * When enabled, a timestamp equal to the previously written timestamp is + * accepted; only a strictly decreasing timestamp is rejected with + * E_OUT_OF_ORDER. Duplicate timestamps are preserved as separate rows (no + * write-time deduplication). Default is false, which rejects both equal and + * decreasing timestamps. + * + * @param enabled true to allow equal timestamps, false to keep strict ordering + * @return E_OK + */ +int set_allow_non_decreasing_timestamps(bool enabled); + /*--------------------------TsFile Reader and Writer------------------------ */ /** diff --git a/cpp/src/writer/time_page_writer.h b/cpp/src/writer/time_page_writer.h index bda9a5023..33d6860ed 100644 --- a/cpp/src/writer/time_page_writer.h +++ b/cpp/src/writer/time_page_writer.h @@ -20,6 +20,7 @@ #define WRITER_PAGE_TIME_WRITER_H #include "common/allocator/byte_stream.h" +#include "common/global.h" #include "common/statistic.h" #include "compress/compressor.h" #include "encoding/encoder.h" @@ -74,7 +75,7 @@ class TimePageWriter { FORCE_INLINE int write(int64_t timestamp) { int ret = common::E_OK; if (statistic_->count_ != 0 && is_inited_ && - timestamp <= statistic_->end_time_) { + is_out_of_order(timestamp, statistic_->end_time_)) { return common::E_OUT_OF_ORDER; } if (RET_FAIL(time_encoder_->encode(timestamp, time_out_stream_))) { @@ -89,12 +90,12 @@ class TimePageWriter { if (count == 0) return ret; // Check order: first timestamp vs existing end_time if (statistic_->count_ != 0 && is_inited_ && - timestamps[0] <= statistic_->end_time_) { + is_out_of_order(timestamps[0], statistic_->end_time_)) { return common::E_OUT_OF_ORDER; } // Check monotonicity within batch for (uint32_t i = 1; i < count; i++) { - if (timestamps[i] <= timestamps[i - 1]) { + if (is_out_of_order(timestamps[i], timestamps[i - 1])) { return common::E_OUT_OF_ORDER; } } @@ -135,6 +136,14 @@ class TimePageWriter { void destroy_page_data() { cur_page_data_.destroy(); } private: + // Reject a timestamp only when it is strictly less than the previous one + // (allow_non_decreasing_timestamps_ enabled), or at most equal to it + // (flag disabled, the historical behavior). + FORCE_INLINE static bool is_out_of_order(int64_t time, int64_t last_time) { + return common::g_config_value_.allow_non_decreasing_timestamps_ + ? time < last_time + : time <= last_time; + } FORCE_INLINE int prepare_end_page() { int ret = common::E_OK; if (RET_FAIL(time_encoder_->flush(time_out_stream_))) { diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index ed3e09be7..6fbc806d7 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -785,18 +785,28 @@ int TsFileWriter::check_memory_size_and_may_flush_chunks() { return ret; } +bool TsFileWriter::recovered_last_time_order_violated( + const std::shared_ptr& device_id, int64_t timestamp) const { + if (!enforce_recovered_last_time_order_) { + return false; + } + auto schema_it = schemas_.find(device_id); + if (schema_it == schemas_.end() || schema_it->second == nullptr) { + return false; + } + return common::g_config_value_.allow_non_decreasing_timestamps_ + ? timestamp < schema_it->second->last_time_ + : timestamp <= schema_it->second->last_time_; +} + int TsFileWriter::write_record(const TsRecord& record) { if (UNLIKELY(unrecoverable_)) return E_DATA_INCONSISTENCY; int ret = E_OK; auto device_id = std::make_shared(record.device_id_); // After recovery, refuse writes whose timestamp would land at or before // any already-flushed chunk's end_time for this device. - if (enforce_recovered_last_time_order_) { - auto schema_it = schemas_.find(device_id); - if (schema_it != schemas_.end() && schema_it->second != nullptr && - record.timestamp_ <= schema_it->second->last_time_) { - return E_OUT_OF_ORDER; - } + if (recovered_last_time_order_violated(device_id, record.timestamp_)) { + return E_OUT_OF_ORDER; } // std::vector chunk_writers; SimpleVector chunk_writers; @@ -834,12 +844,8 @@ int TsFileWriter::write_record_aligned(const TsRecord& record) { if (UNLIKELY(unrecoverable_)) return E_DATA_INCONSISTENCY; int ret = E_OK; auto device_id = std::make_shared(record.device_id_); - if (enforce_recovered_last_time_order_) { - auto schema_it = schemas_.find(device_id); - if (schema_it != schemas_.end() && schema_it->second != nullptr && - record.timestamp_ <= schema_it->second->last_time_) { - return E_OUT_OF_ORDER; - } + if (recovered_last_time_order_violated(device_id, record.timestamp_)) { + return E_OUT_OF_ORDER; } SimpleVector value_chunk_writers; SimpleVector data_types; @@ -1056,13 +1062,9 @@ int TsFileWriter::write_tablet_aligned(const Tablet& tablet) { auto device_id = std::make_shared(tablet.insert_target_name_); const uint32_t total_rows = tablet.get_cur_row_size(); - if (enforce_recovered_last_time_order_ && total_rows > 0 && - tablet.timestamps_ != nullptr) { - auto schema_it = schemas_.find(device_id); - if (schema_it != schemas_.end() && schema_it->second != nullptr && - tablet.timestamps_[0] <= schema_it->second->last_time_) { - return E_OUT_OF_ORDER; - } + if (total_rows > 0 && tablet.timestamps_ != nullptr && + recovered_last_time_order_violated(device_id, tablet.timestamps_[0])) { + return E_OUT_OF_ORDER; } SimpleVector value_chunk_writers; TimeChunkWriter* time_chunk_writer = nullptr; @@ -1206,13 +1208,9 @@ int TsFileWriter::write_tablet(const Tablet& tablet) { // and would let uninitialized timestamps/values past the live range leak // into the chunk. const uint32_t total_rows = tablet.get_cur_row_size(); - if (enforce_recovered_last_time_order_ && total_rows > 0 && - tablet.timestamps_ != nullptr) { - auto schema_it = schemas_.find(device_id); - if (schema_it != schemas_.end() && schema_it->second != nullptr && - tablet.timestamps_[0] <= schema_it->second->last_time_) { - return E_OUT_OF_ORDER; - } + if (total_rows > 0 && tablet.timestamps_ != nullptr && + recovered_last_time_order_violated(device_id, tablet.timestamps_[0])) { + return E_OUT_OF_ORDER; } SimpleVector chunk_writers; SimpleVector data_types; @@ -1336,14 +1334,10 @@ int TsFileWriter::write_table(Tablet& tablet) { // Recovery: refuse any segment whose first timestamp would land // at or before a flushed chunk's end_time for this device. This // mirrors the per-record / per-tablet check on the tree path. - if (enforce_recovered_last_time_order_ && tablet.timestamps_ && - ei > si) { - auto schema_it = schemas_.find(device_id); - if (schema_it != schemas_.end() && - schema_it->second != nullptr && - tablet.timestamps_[si] <= schema_it->second->last_time_) { - return E_OUT_OF_ORDER; - } + if (tablet.timestamps_ && ei > si && + recovered_last_time_order_violated(device_id, + tablet.timestamps_[si])) { + return E_OUT_OF_ORDER; } auto idx_it = device_ctx_index.find(device_id); if (idx_it == device_ctx_index.end()) { @@ -1565,14 +1559,10 @@ int TsFileWriter::write_table(Tablet& tablet) { if (end_idx == 0) continue; const uint32_t si = static_cast(start_idx); - if (enforce_recovered_last_time_order_ && tablet.timestamps_ && - end_idx > start_idx) { - auto schema_it = schemas_.find(device_id); - if (schema_it != schemas_.end() && - schema_it->second != nullptr && - tablet.timestamps_[si] <= schema_it->second->last_time_) { - return E_OUT_OF_ORDER; - } + if (tablet.timestamps_ && end_idx > start_idx && + recovered_last_time_order_violated(device_id, + tablet.timestamps_[si])) { + return E_OUT_OF_ORDER; } MeasurementNamesFromTablet mnames_getter(tablet); SimpleVector chunk_writers; diff --git a/cpp/src/writer/tsfile_writer.h b/cpp/src/writer/tsfile_writer.h index 0606a7a33..b5fb86702 100644 --- a/cpp/src/writer/tsfile_writer.h +++ b/cpp/src/writer/tsfile_writer.h @@ -227,6 +227,14 @@ class TsFileWriter { // flush/close refuse — can be unit-tested directly. friend class TsFileWriterUnrecoverableTest; + // After recovery (enforce_recovered_last_time_order_), returns true when + // `timestamp` must be rejected because it would land at or before a + // flushed chunk's end_time for this device. With + // allow_non_decreasing_timestamps_ enabled only strictly decreasing + // timestamps are rejected; otherwise equality is rejected too. + bool recovered_last_time_order_violated( + const std::shared_ptr& device_id, int64_t timestamp) const; + int write_column_batch(storage::ChunkWriter* chunk_writer, const Tablet& tablet, int col_idx, uint32_t start_idx, uint32_t end_idx); diff --git a/cpp/test/reader/duplicate_timestamps_roundtrip_test.cc b/cpp/test/reader/duplicate_timestamps_roundtrip_test.cc new file mode 100644 index 000000000..b397eb8b8 --- /dev/null +++ b/cpp/test/reader/duplicate_timestamps_roundtrip_test.cc @@ -0,0 +1,68 @@ +/* + * 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. + */ + +#include + +#include +#include +#include + +#include "common/global.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" + +using namespace common; +using namespace storage; + +#ifndef TSFILE_CPP_SOURCE_DIR +#define TSFILE_CPP_SOURCE_DIR "." +#endif + +// The fixture was written by the Java implementation with +// allowNonDecreasingTimestamps enabled: a single non-aligned series +// root.sg.d1.s1 (INT32, PLAIN, default TS_2DIFF time encoding) holding +// timestamps {0, 1, 1, 2}. The C++ reader must return the duplicate +// timestamps as separate rows and in order. +TEST(DuplicateTimestampsRoundTripTest, ReadsJavaWrittenDuplicateTimestamps) { + libtsfile_init(); + const std::string fixture = + std::string(TSFILE_CPP_SOURCE_DIR) + + "/test/reader/fixtures/duplicate_timestamps.tsfile"; + TsFileReader reader; + ASSERT_EQ(common::E_OK, reader.open(fixture)); + + ResultSet* result = nullptr; + std::vector paths{"root.sg.d1.s1"}; + ASSERT_EQ(common::E_OK, reader.query(paths, INT64_MIN, INT64_MAX, result)); + ASSERT_NE(result, nullptr); + + const int64_t expected[] = {0, 1, 1, 2}; + int64_t cur_row = 0; + bool has_next = false; + while (result->next(has_next) == common::E_OK && has_next) { + ASSERT_LT(cur_row, static_cast(4)); + RowRecord* rec = result->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(expected[cur_row], rec->get_timestamp()); + ++cur_row; + } + EXPECT_EQ(cur_row, 4); + reader.destroy_query_data_set(result); + ASSERT_EQ(common::E_OK, reader.close()); +} diff --git a/cpp/test/reader/fixtures/duplicate_timestamps.tsfile b/cpp/test/reader/fixtures/duplicate_timestamps.tsfile new file mode 100644 index 0000000000000000000000000000000000000000..2ee39a6abcc0eab3a82f800de6bb1fe086180782 GIT binary patch literal 292 zcmWG3cFW93Wnp0AE6UF=(JM}8Nik$)DK->kWM_~QUMviLZ3E?tFaotf*+x)0lbI1_D #endif +#include "common/global.h" #include "common/path.h" #include "common/record.h" #include "common/schema.h" @@ -2418,3 +2419,171 @@ TEST_F(TsFileWriterTest, AlignedRecordPointOrderDoesNotMatter) { reader.destroy_query_data_set(qds); ASSERT_EQ(reader.close(), E_OK); } + +namespace { +// The allow_non_decreasing_timestamps_ flag is process-global; restore it to +// false even when an ASSERT_* aborts a test early. +class ScopedAllowNonDecreasingTimestamps { + public: + explicit ScopedAllowNonDecreasingTimestamps(bool value) { + common::set_allow_non_decreasing_timestamps(value); + } + ~ScopedAllowNonDecreasingTimestamps() { + common::set_allow_non_decreasing_timestamps(false); + } +}; +} // namespace + +// With the flag on, equal timestamps must be accepted on the aligned batch +// path (TimePageWriter::write_batch) and read back as separate rows. +TEST_F(TsFileWriterTest, AllowNonDecreasingAlignedTabletWriteAndRead) { + ScopedAllowNonDecreasingTimestamps guard(true); + std::string device_name = "device_dup_aligned"; + std::vector schema_vec; + schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); + schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); + { + std::vector reg; + for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); + ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, reg), + E_OK); + } + + const int row_num = 4; + const int64_t times[row_num] = {10, 11, 11, 12}; + Tablet tablet(device_name, + std::make_shared>(schema_vec), + row_num); + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ(tablet.add_timestamp(i, times[i]), E_OK); + ASSERT_EQ(tablet.add_value(i, 0u, static_cast(i)), E_OK); + ASSERT_EQ(tablet.add_value(i, 1u, static_cast(i * 2)), E_OK); + } + ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + select_list.emplace_back(device_name + ".v0"); + select_list.emplace_back(device_name + ".v1"); + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + ASSERT_NE(tmp_qds, nullptr); + + bool has_next = false; + int64_t cur_row = 0; + while (IS_SUCC(tmp_qds->next(has_next)) && has_next) { + auto* rec = tmp_qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), times[cur_row]); + EXPECT_EQ(field_to_string(rec->get_field(1)), std::to_string(cur_row)); + EXPECT_EQ(field_to_string(rec->get_field(2)), + std::to_string(cur_row * 2)); + ++cur_row; + } + EXPECT_EQ(cur_row, row_num); + reader.destroy_query_data_set(tmp_qds); + ASSERT_EQ(reader.close(), E_OK); +} + +// With the flag on, equal timestamps must be accepted on the aligned row path +// (TimePageWriter::write) and read back as separate rows. +TEST_F(TsFileWriterTest, AllowNonDecreasingAlignedRecordWriteAndRead) { + ScopedAllowNonDecreasingTimestamps guard(true); + std::string device_name = "device_dup_aligned_record"; + std::vector schema_vec; + schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); + { + std::vector reg; + reg.push_back(new MeasurementSchema(schema_vec[0])); + ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, reg), + E_OK); + } + + const int64_t times[4] = {5, 6, 6, 7}; + for (int i = 0; i < 4; ++i) { + TsRecord record(times[i], device_name); + record.add_point("v0", static_cast(i)); + ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + select_list.emplace_back(device_name + ".v0"); + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + ASSERT_NE(tmp_qds, nullptr); + + bool has_next = false; + int64_t cur_row = 0; + while (IS_SUCC(tmp_qds->next(has_next)) && has_next) { + auto* rec = tmp_qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), times[cur_row]); + ++cur_row; + } + EXPECT_EQ(cur_row, 4); + reader.destroy_query_data_set(tmp_qds); + ASSERT_EQ(reader.close(), E_OK); +} + +// Strictly decreasing timestamps must still be rejected with the flag on. +TEST_F(TsFileWriterTest, AllowNonDecreasingStillRejectsStrictlyDecreasing) { + ScopedAllowNonDecreasingTimestamps guard(true); + std::string device_name = "device_dup_decreasing"; + std::vector schema_vec; + schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); + { + std::vector reg; + reg.push_back(new MeasurementSchema(schema_vec[0])); + ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, reg), + E_OK); + } + + const int row_num = 3; + const int64_t times[row_num] = {10, 11, 10}; + Tablet tablet(device_name, + std::make_shared>(schema_vec), + row_num); + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ(tablet.add_timestamp(i, times[i]), E_OK); + ASSERT_EQ(tablet.add_value(i, 0u, static_cast(i)), E_OK); + } + EXPECT_NE(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); +} + +// With the flag off (default), equal timestamps must still be rejected. +TEST_F(TsFileWriterTest, EqualTimestampsRejectedAlignedWhenFlagOff) { + ScopedAllowNonDecreasingTimestamps guard(false); + std::string device_name = "device_dup_flag_off"; + std::vector schema_vec; + schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); + { + std::vector reg; + reg.push_back(new MeasurementSchema(schema_vec[0])); + ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, reg), + E_OK); + } + + const int row_num = 3; + const int64_t times[row_num] = {10, 11, 11}; + Tablet tablet(device_name, + std::make_shared>(schema_vec), + row_num); + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ(tablet.add_timestamp(i, times[i]), E_OK); + ASSERT_EQ(tablet.add_value(i, 0u, static_cast(i)), E_OK); + } + EXPECT_NE(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/common/conf/TSFileConfig.java b/java/tsfile/src/main/java/org/apache/tsfile/common/conf/TSFileConfig.java index 6c79d6567..cffade760 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/common/conf/TSFileConfig.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/common/conf/TSFileConfig.java @@ -258,6 +258,14 @@ public class TSFileConfig implements Serializable { private boolean lz4UseJni = true; + /** + * When true, writing a timestamp equal to the previously written timestamp is accepted; only a + * strictly decreasing timestamp is rejected as out-of-order. Duplicate timestamps are preserved + * as separate rows (no write-time deduplication). Default is false, which rejects both equal and + * decreasing timestamps, matching the historical behavior. + */ + private boolean allowNonDecreasingTimestamps = false; + /** customizedProperties, this should be empty by default. */ private Properties customizedProperties = new Properties(); @@ -530,6 +538,14 @@ public void setEndian(String endian) { this.endian = endian; } + public boolean isAllowNonDecreasingTimestamps() { + return allowNonDecreasingTimestamps; + } + + public void setAllowNonDecreasingTimestamps(boolean allowNonDecreasingTimestamps) { + this.allowNonDecreasingTimestamps = allowNonDecreasingTimestamps; + } + public boolean isUseKerberos() { return useKerberos; } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java index 00e97cb71..24eae13ee 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java @@ -331,7 +331,12 @@ protected void sealAllChunks() { } protected void checkIsHistoryData(long time) throws WriteProcessException { - if (isInitLastTime && time <= lastTime) { + final boolean outOfOrder = + isInitLastTime + && (TSFileDescriptor.getInstance().getConfig().isAllowNonDecreasingTimestamps() + ? time < lastTime + : time <= lastTime); + if (outOfOrder) { throw new WriteProcessException( Messages.format( "error.write.chunk_group_aligned_out_of_order", diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java index 64f4b6297..ae1e6aeb0 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java @@ -18,6 +18,7 @@ */ package org.apache.tsfile.write.chunk; +import org.apache.tsfile.common.conf.TSFileDescriptor; import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; @@ -192,7 +193,7 @@ private void sealAllChunks() { private void checkIsHistoryData(String measurementId, long time) throws WriteProcessException { final Long lastTime = lastTimeMap.get(measurementId); - if (lastTime != null && time <= lastTime) { + if (lastTime != null && isOutOfOrder(time, lastTime)) { throw new WriteProcessException( Messages.format( "error.write.chunk_group_non_aligned_out_of_order", @@ -203,6 +204,12 @@ private void checkIsHistoryData(String measurementId, long time) throws WritePro } } + private static boolean isOutOfOrder(long time, long lastTime) { + return TSFileDescriptor.getInstance().getConfig().isAllowNonDecreasingTimestamps() + ? time < lastTime + : time <= lastTime; + } + public Map getLastTimeMap() { return this.lastTimeMap; } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TabletWriteContext.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TabletWriteContext.java index 3a97619dc..647ffe9bb 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TabletWriteContext.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TabletWriteContext.java @@ -19,6 +19,7 @@ package org.apache.tsfile.write.chunk; +import org.apache.tsfile.common.conf.TSFileDescriptor; import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.exception.write.WriteProcessException; import org.apache.tsfile.file.metadata.IDeviceID; @@ -40,7 +41,7 @@ public TabletWriteContext(IDeviceID deviceId, String measurementId, Long lastTim } public void checkTime(long time) throws WriteProcessException { - if (hasLastTime && time <= lastTime) { + if (hasLastTime && isOutOfOrder(time, lastTime)) { throw new WriteProcessException( Messages.format( "error.write.chunk_group_non_aligned_out_of_order", @@ -51,6 +52,12 @@ public void checkTime(long time) throws WriteProcessException { } } + private static boolean isOutOfOrder(long time, long lastTime) { + return TSFileDescriptor.getInstance().getConfig().isAllowNonDecreasingTimestamps() + ? time < lastTime + : time <= lastTime; + } + /** Advance only after a successful write, so failures retain the last accepted timestamp. */ public void recordTime(long time) { lastTime = time; diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java new file mode 100644 index 000000000..4575365a6 --- /dev/null +++ b/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java @@ -0,0 +1,160 @@ +/* + * 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. + */ + +package org.apache.tsfile.write; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.common.conf.TSFileDescriptor; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.exception.write.WriteProcessException; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TsFileReader; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.expression.QueryExpression; +import org.apache.tsfile.read.query.dataset.QueryDataSet; +import org.apache.tsfile.write.record.TSRecord; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.record.datapoint.IntDataPoint; +import org.apache.tsfile.write.schema.MeasurementSchema; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertThrows; + +public class AllowNonDecreasingTimestampsTest { + + private final File file = new File("target/AllowNonDecreasingTimestampsTest.tsfile"); + private final String deviceId = "root.sg.d1"; + private final MeasurementSchema s1 = + new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.PLAIN); + private final MeasurementSchema s2 = + new MeasurementSchema("s2", TSDataType.INT32, TSEncoding.PLAIN); + private TSFileConfig config; + + @Before + public void setUp() { + config = TSFileDescriptor.getInstance().getConfig(); + config.setAllowNonDecreasingTimestamps(false); + if (file.exists() && !file.delete()) { + throw new RuntimeException("can not delete " + file.getAbsolutePath()); + } + } + + @After + public void tearDown() { + config.setAllowNonDecreasingTimestamps(false); + if (file.exists()) { + file.delete(); + } + } + + @Test + public void writeNonAlignedTabletWithDuplicates() throws IOException, WriteProcessException { + config.setAllowNonDecreasingTimestamps(true); + try (TsFileWriter writer = new TsFileWriter(file)) { + writer.registerTimeseries(new Path(deviceId), Arrays.asList(s1)); + Tablet tablet = new Tablet(deviceId, Collections.singletonList(s1), 4); + long[] times = {0, 1, 1, 2}; + for (int row = 0; row < times.length; row++) { + tablet.addTimestamp(row, times[row]); + tablet.addValue("s1", row, row * 10); + } + writer.writeTree(tablet); + } + + List readTimes = readTimestamps(deviceId, "s1"); + Assert.assertEquals(Arrays.asList(0L, 1L, 1L, 2L), readTimes); + } + + @Test + public void writeAlignedRecordWithDuplicates() throws IOException, WriteProcessException { + config.setAllowNonDecreasingTimestamps(true); + try (TsFileWriter writer = new TsFileWriter(file)) { + writer.registerAlignedTimeseries(new Path(deviceId), Arrays.asList(s1, s2)); + long[] times = {5, 6, 6, 7}; + for (int i = 0; i < times.length; i++) { + TSRecord record = new TSRecord(deviceId, times[i]); + record.addTuple(new IntDataPoint("s1", i)); + record.addTuple(new IntDataPoint("s2", i * 2)); + writer.writeRecord(record); + } + } + + List readTimes = readTimestamps(deviceId, "s1"); + Assert.assertEquals(Arrays.asList(5L, 6L, 6L, 7L), readTimes); + } + + @Test + public void strictlyDecreasingStillRejectedWhenFlagOn() + throws IOException, WriteProcessException { + config.setAllowNonDecreasingTimestamps(true); + try (TsFileWriter writer = new TsFileWriter(file)) { + writer.registerTimeseries(new Path(deviceId), Arrays.asList(s1)); + Tablet tablet = new Tablet(deviceId, Collections.singletonList(s1), 3); + long[] times = {0, 1, 0}; + for (int row = 0; row < times.length; row++) { + tablet.addTimestamp(row, times[row]); + tablet.addValue("s1", row, row); + } + assertThrows(WriteProcessException.class, () -> writer.writeTree(tablet)); + } + } + + @Test + public void duplicateStillRejectedWhenFlagOff() throws IOException, WriteProcessException { + try (TsFileWriter writer = new TsFileWriter(file)) { + writer.registerTimeseries(new Path(deviceId), Arrays.asList(s1)); + Tablet tablet = new Tablet(deviceId, Collections.singletonList(s1), 3); + long[] times = {0, 1, 1}; + for (int row = 0; row < times.length; row++) { + tablet.addTimestamp(row, times[row]); + tablet.addValue("s1", row, row); + } + assertThrows(WriteProcessException.class, () -> writer.writeTree(tablet)); + } + } + + private List readTimestamps(String device, String measurement) throws IOException { + List times = new ArrayList<>(); + try (TsFileSequenceReader sequenceReader = new TsFileSequenceReader(file.getAbsolutePath()); + TsFileReader reader = new TsFileReader(sequenceReader)) { + QueryExpression query = + QueryExpression.create( + Collections.singletonList(new Path(device, measurement, false)), null); + QueryDataSet dataSet = reader.query(query); + while (dataSet.hasNext()) { + RowRecord record = dataSet.next(); + times.add(record.getTimestamp()); + } + } + return times; + } +} diff --git a/python/tests/test_allow_non_decreasing_timestamps.py b/python/tests/test_allow_non_decreasing_timestamps.py new file mode 100644 index 000000000..3c4492f34 --- /dev/null +++ b/python/tests/test_allow_non_decreasing_timestamps.py @@ -0,0 +1,105 @@ +# 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 os + +import pytest + +from tsfile import ( + TIME_COLUMN, + ColumnCategory, + ColumnSchema, + TableSchema, + TSDataType, + Tablet, + TsFileTableWriter, + to_dataframe, +) +from tsfile.exceptions import OutOfOrderError + +TABLE_NAME = "dup_table" +FILE_NAME = "allow_non_decreasing_timestamps.tsfile" + + +def _table_schema(): + return TableSchema( + TABLE_NAME, + [ + ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG), + ColumnSchema("value", TSDataType.INT64, ColumnCategory.FIELD), + ], + ) + + +def _build_tablet(times): + tablet = Tablet( + ["device", "value"], [TSDataType.STRING, TSDataType.INT64], len(times) + ) + tablet.set_table_name(TABLE_NAME) + for i, timestamp in enumerate(times): + tablet.add_timestamp(i, timestamp) + tablet.add_value_by_name("device", i, "d0") + tablet.add_value_by_name("value", i, i) + return tablet + + +def _remove_file(): + if os.path.exists(FILE_NAME): + os.remove(FILE_NAME) + + +def test_table_writer_duplicates_allowed(): + try: + with TsFileTableWriter( + FILE_NAME, _table_schema(), allow_non_decreasing_timestamps=True + ) as writer: + writer.write_table(_build_tablet([10, 11, 11, 12])) + df = to_dataframe(FILE_NAME, table_name=TABLE_NAME) + assert df.shape[0] == 4 + assert list(df[TIME_COLUMN]) == [10, 11, 11, 12] + finally: + _remove_file() + + +def test_table_writer_duplicates_rejected_by_default(): + _assert_write_rejected(allow_non_decreasing_timestamps=False, times=[10, 11, 11]) + + +def test_table_writer_strictly_decreasing_rejected(): + _assert_write_rejected(allow_non_decreasing_timestamps=True, times=[10, 11, 10]) + + +def _assert_write_rejected(allow_non_decreasing_timestamps, times): + try: + writer = TsFileTableWriter( + FILE_NAME, + _table_schema(), + allow_non_decreasing_timestamps=allow_non_decreasing_timestamps, + ) + try: + with pytest.raises(OutOfOrderError): + writer.write_table(_build_tablet(times)) + finally: + # A rejected time column leaves the C++ writer unrecoverable, so + # close() may raise DataInconsistencyError; ignore it here. + try: + writer.close() + except Exception: + pass + finally: + _remove_file() diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index d5f2539cf..e4f75b868 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -526,6 +526,7 @@ cdef extern from "common/config/config.h" namespace "common": uint8_t double_encoding_type_; uint8_t string_encoding_type_; uint8_t default_compression_type_; + bint allow_non_decreasing_timestamps_; cdef extern from "common/global.h" namespace "common": ConfigValue g_config_value_ @@ -534,3 +535,4 @@ cdef extern from "common/global.h" namespace "common": 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); + void set_allow_non_decreasing_timestamps(bint enabled); diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index b6a7795b6..ae1083b4d 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -914,6 +914,7 @@ cpdef object get_tsfile_config(): "string_encoding_type_": TSEncodingPy(int(g_config_value_.string_encoding_type_)), "default_compression_type_": CompressorPy(int(g_config_value_.default_compression_type_)), "file_read_backend_": FileReadBackendPy(int(tsfile_get_file_read_backend())), + "allow_non_decreasing_timestamps_": g_config_value_.allow_non_decreasing_timestamps_, } cpdef object get_file_read_backend(): @@ -1014,6 +1015,10 @@ cpdef void set_tsfile_config(dict new_config): check_error(code) if "file_read_backend_" in new_config: set_file_read_backend(new_config["file_read_backend_"]) + if "allow_non_decreasing_timestamps_" in new_config: + _check_bool(new_config["allow_non_decreasing_timestamps_"]) + set_allow_non_decreasing_timestamps( + new_config["allow_non_decreasing_timestamps_"]) cdef _check_uint32(value): if not isinstance(value, int) or value < 0 or value > 0xFFFFFFFF: diff --git a/python/tsfile/tsfile_table_writer.py b/python/tsfile/tsfile_table_writer.py index e4fceef97..6e525ff98 100644 --- a/python/tsfile/tsfile_table_writer.py +++ b/python/tsfile/tsfile_table_writer.py @@ -98,14 +98,25 @@ class TsFileTableWriter: """ def __init__( - self, path: str, table_schema: TableSchema, memory_threshold=128 * 1024 * 1024 + self, + path: str, + table_schema: TableSchema, + memory_threshold=128 * 1024 * 1024, + allow_non_decreasing_timestamps=False, ): """ :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. + :param allow_non_decreasing_timestamps: when True, a timestamp equal to + the previously written timestamp is accepted; only a strictly + decreasing timestamp is rejected as out-of-order. Duplicate + timestamps are preserved as separate rows (no write-time dedup). + This is a process-wide setting. """ - self.writer = TsFileWriter(path, memory_threshold) + self.writer = TsFileWriter( + path, memory_threshold, allow_non_decreasing_timestamps + ) self.writer.register_table(table_schema) self.tableSchema = table_schema diff --git a/python/tsfile/tsfile_writer.pyx b/python/tsfile/tsfile_writer.pyx index 7ae95c635..7f0bd57ff 100644 --- a/python/tsfile/tsfile_writer.pyx +++ b/python/tsfile/tsfile_writer.pyx @@ -33,8 +33,11 @@ from .tsfile_py_cpp cimport * cdef class TsFileWriterPy: cdef TsFileWriter writer - def __init__(self, pathname: str, memory_threshold: int = 128 * 1024 * 1024): + def __init__(self, pathname: str, memory_threshold: int = 128 * 1024 * 1024, + allow_non_decreasing_timestamps: bool = False): self.writer = NULL + set_allow_non_decreasing_timestamps( + allow_non_decreasing_timestamps) self.writer = tsfile_writer_new_c(pathname, memory_threshold) def register_timeseries(self, device_name : str, timeseries_schema : TimeseriesSchemaPy): From b61de61795e0a773fe68c754c9c5de6713f74e92 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 11 Nov 2026 14:13:22 +0800 Subject: [PATCH 2/7] test(python): use a unique temp dir for duplicate-timestamp rejection tests The rejected write leaves the C++ writer unrecoverable, so its file handle stays open on Windows; deleting a fixed-named file then fails with WinError 32 and leaks into the next test. Write to a unique temp dir and clean up best-effort. --- .../test_allow_non_decreasing_timestamps.py | 26 ++++++++++++------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/python/tests/test_allow_non_decreasing_timestamps.py b/python/tests/test_allow_non_decreasing_timestamps.py index 3c4492f34..50166ab69 100644 --- a/python/tests/test_allow_non_decreasing_timestamps.py +++ b/python/tests/test_allow_non_decreasing_timestamps.py @@ -17,6 +17,8 @@ # import os +import shutil +import tempfile import pytest @@ -33,7 +35,6 @@ from tsfile.exceptions import OutOfOrderError TABLE_NAME = "dup_table" -FILE_NAME = "allow_non_decreasing_timestamps.tsfile" def _table_schema(): @@ -58,22 +59,23 @@ def _build_tablet(times): return tablet -def _remove_file(): - if os.path.exists(FILE_NAME): - os.remove(FILE_NAME) +def _new_tmpdir(): + return tempfile.mkdtemp(prefix="tsfile_allow_non_decreasing_") def test_table_writer_duplicates_allowed(): + tmpdir = _new_tmpdir() + path = os.path.join(tmpdir, "dup.tsfile") try: with TsFileTableWriter( - FILE_NAME, _table_schema(), allow_non_decreasing_timestamps=True + path, _table_schema(), allow_non_decreasing_timestamps=True ) as writer: writer.write_table(_build_tablet([10, 11, 11, 12])) - df = to_dataframe(FILE_NAME, table_name=TABLE_NAME) + df = to_dataframe(path, table_name=TABLE_NAME) assert df.shape[0] == 4 assert list(df[TIME_COLUMN]) == [10, 11, 11, 12] finally: - _remove_file() + shutil.rmtree(tmpdir, ignore_errors=True) def test_table_writer_duplicates_rejected_by_default(): @@ -85,9 +87,11 @@ def test_table_writer_strictly_decreasing_rejected(): def _assert_write_rejected(allow_non_decreasing_timestamps, times): + tmpdir = _new_tmpdir() + path = os.path.join(tmpdir, "dup.tsfile") try: writer = TsFileTableWriter( - FILE_NAME, + path, _table_schema(), allow_non_decreasing_timestamps=allow_non_decreasing_timestamps, ) @@ -96,10 +100,12 @@ def _assert_write_rejected(allow_non_decreasing_timestamps, times): writer.write_table(_build_tablet(times)) finally: # A rejected time column leaves the C++ writer unrecoverable, so - # close() may raise DataInconsistencyError; ignore it here. + # close() raises DataInconsistencyError instead of releasing the + # file; ignore it here. The handle stays open on Windows, so the + # rmtree below cleans up best-effort. try: writer.close() except Exception: pass finally: - _remove_file() + shutil.rmtree(tmpdir, ignore_errors=True) From 7f3cbe8dff4b01dd36ff57aa481a258bca994d13 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 11 Nov 2026 14:52:03 +0800 Subject: [PATCH 3/7] test(cpp): generate duplicate-timestamp fixture inline instead of committing a binary Replace the committed Java-written binary fixture with a self-contained non-aligned write+read round-trip test in TsFileWriterTest, and remove the now-unused standalone reader test and binary fixture. The new test writes {0, 1, 1, 2} through the non-aligned row path with allow_non_decreasing_timestamps enabled and asserts the duplicate timestamps are read back as separate rows. --- .../duplicate_timestamps_roundtrip_test.cc | 68 ------------------ .../fixtures/duplicate_timestamps.tsfile | Bin 292 -> 0 bytes cpp/test/writer/tsfile_writer_test.cc | 42 +++++++++++ 3 files changed, 42 insertions(+), 68 deletions(-) delete mode 100644 cpp/test/reader/duplicate_timestamps_roundtrip_test.cc delete mode 100644 cpp/test/reader/fixtures/duplicate_timestamps.tsfile diff --git a/cpp/test/reader/duplicate_timestamps_roundtrip_test.cc b/cpp/test/reader/duplicate_timestamps_roundtrip_test.cc deleted file mode 100644 index b397eb8b8..000000000 --- a/cpp/test/reader/duplicate_timestamps_roundtrip_test.cc +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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. - */ - -#include - -#include -#include -#include - -#include "common/global.h" -#include "reader/result_set.h" -#include "reader/tsfile_reader.h" - -using namespace common; -using namespace storage; - -#ifndef TSFILE_CPP_SOURCE_DIR -#define TSFILE_CPP_SOURCE_DIR "." -#endif - -// The fixture was written by the Java implementation with -// allowNonDecreasingTimestamps enabled: a single non-aligned series -// root.sg.d1.s1 (INT32, PLAIN, default TS_2DIFF time encoding) holding -// timestamps {0, 1, 1, 2}. The C++ reader must return the duplicate -// timestamps as separate rows and in order. -TEST(DuplicateTimestampsRoundTripTest, ReadsJavaWrittenDuplicateTimestamps) { - libtsfile_init(); - const std::string fixture = - std::string(TSFILE_CPP_SOURCE_DIR) + - "/test/reader/fixtures/duplicate_timestamps.tsfile"; - TsFileReader reader; - ASSERT_EQ(common::E_OK, reader.open(fixture)); - - ResultSet* result = nullptr; - std::vector paths{"root.sg.d1.s1"}; - ASSERT_EQ(common::E_OK, reader.query(paths, INT64_MIN, INT64_MAX, result)); - ASSERT_NE(result, nullptr); - - const int64_t expected[] = {0, 1, 1, 2}; - int64_t cur_row = 0; - bool has_next = false; - while (result->next(has_next) == common::E_OK && has_next) { - ASSERT_LT(cur_row, static_cast(4)); - RowRecord* rec = result->get_row_record(); - ASSERT_NE(rec, nullptr); - EXPECT_EQ(expected[cur_row], rec->get_timestamp()); - ++cur_row; - } - EXPECT_EQ(cur_row, 4); - reader.destroy_query_data_set(result); - ASSERT_EQ(common::E_OK, reader.close()); -} diff --git a/cpp/test/reader/fixtures/duplicate_timestamps.tsfile b/cpp/test/reader/fixtures/duplicate_timestamps.tsfile deleted file mode 100644 index 2ee39a6abcc0eab3a82f800de6bb1fe086180782..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 292 zcmWG3cFW93Wnp0AE6UF=(JM}8Nik$)DK->kWM_~QUMviLZ3E?tFaotf*+x)0lbI1_Dwrite_tablet_aligned(tablet), E_OK); ASSERT_EQ(tsfile_writer_->close(), E_OK); } + +// With the flag on, equal timestamps must be accepted on the non-aligned row +// path (TimePageWriter::write) and read back as separate rows. +TEST_F(TsFileWriterTest, AllowNonDecreasingNonAlignedRecordWriteAndRead) { + ScopedAllowNonDecreasingTimestamps guard(true); + std::string device_name = "device_dup_non_aligned"; + ASSERT_EQ( + tsfile_writer_->register_timeseries( + device_name, MeasurementSchema("v0", INT64, PLAIN, UNCOMPRESSED)), + E_OK); + + const int64_t times[4] = {0, 1, 1, 2}; + for (int i = 0; i < 4; ++i) { + TsRecord record(times[i], device_name); + record.add_point("v0", static_cast(i)); + ASSERT_EQ(tsfile_writer_->write_record(record), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + select_list.emplace_back(device_name + ".v0"); + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + ASSERT_NE(tmp_qds, nullptr); + + bool has_next = false; + int64_t cur_row = 0; + while (IS_SUCC(tmp_qds->next(has_next)) && has_next) { + auto* rec = tmp_qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), times[cur_row]); + ++cur_row; + } + EXPECT_EQ(cur_row, 4); + reader.destroy_query_data_set(tmp_qds); + ASSERT_EQ(reader.close(), E_OK); +} From 39f08f9eb59031212b72b78800cb5b8673904dad Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 11 Nov 2026 15:29:38 +0800 Subject: [PATCH 4/7] test(compat): add Java/C++ cross-language duplicate-timestamp round-trip Add an optional trailing timestampProfile column to the compatibility matrix. The duplicates profile writes timestamps as row/2 (each value appears twice) with allow_non_decreasing_timestamps enabled. Two new INT32/PLAIN cases (table + tree) let each side write the fixture and validate the other side's file returns the duplicates as separate rows. --- compat/README.md | 1 + compat/compatibility-matrix.csv | 4 +- ...encoding_compression_compatibility_test.cc | 111 +++++++++++++----- .../EncodingCompressionCompatibilityTest.java | 93 ++++++++++----- 4 files changed, 148 insertions(+), 61 deletions(-) diff --git a/compat/README.md b/compat/README.md index e0154a2ec..1a453fba8 100644 --- a/compat/README.md +++ b/compat/README.md @@ -52,6 +52,7 @@ summary. | `javaWrite`/`cppWrite` | Whether that language writes the fixture. | | `javaRead`/`cppRead` | Whether that language validates the fixture when the other side wrote it. | | `gapReason` | Required whenever any write/read cell is `no`; the reason is reported in every CI run. | +| `timestampProfile` | `increasing` (default; the timestamp equals the row index) or `duplicates` (the timestamp is `row/2`, so every value appears twice, written with `allow_non_decreasing_timestamps` enabled). This trailing column is optional; omitting it means `increasing`. | A case with a `no` in a write column is an explicit declared gap. A case absent from the file is not a gap: the completeness test in both suites enumerates the diff --git a/compat/compatibility-matrix.csv b/compat/compatibility-matrix.csv index 0167e4242..591d2caa2 100644 --- a/compat/compatibility-matrix.csv +++ b/compat/compatibility-matrix.csv @@ -53,7 +53,7 @@ # 'sprintz' profile (values within +/-1e9). Streaming numeric extremes are # covered under PLAIN, RLE, TS_2DIFF, GORILLA, CHIMP and RLBE. -id,model,dataType,encoding,compressions,rowCount,pageRows,chunks,devices,valueProfile,nullPatterns,javaWrite,cppWrite,javaRead,cppRead,gapReason +id,model,dataType,encoding,compressions,rowCount,pageRows,chunks,devices,valueProfile,nullPatterns,javaWrite,cppWrite,javaRead,cppRead,gapReason,timestampProfile table_int32_plain_r10,table,INT32,PLAIN,ALL,10,4,1,1,default,NONE,yes,yes,yes,yes, table_int32_plain_r1000,table,INT32,PLAIN,ALL,1000,100,1,1,default,NONE,yes,yes,yes,yes, table_int32_plain_r100000,table,INT32,PLAIN,ALL,100000,10000,1,1,default,NONE,yes,yes,yes,yes, @@ -342,3 +342,5 @@ table_two_pages,table,FLOAT,CHIMP,ALL,250,100,1,1,default,NONE,yes,yes,yes,yes, table_multi_chunk,table,INT64,TS_2DIFF,ALL,300,0,3,1,default,NONE,yes,yes,yes,yes, table_multi_chunk_group,table,INT32,RLE,ALL,50,0,1,3,runs,NONE,yes,yes,yes,yes, table_long_runs,table,INT64,RLE,ALL,1024,0,1,1,runs,NONE,yes,yes,yes,yes, +table_int32_plain_dups,table,INT32,PLAIN,UNCOMPRESSED,10,4,1,1,default,NONE,yes,yes,yes,yes,,duplicates +tree_int32_plain_dups,tree,INT32,PLAIN,UNCOMPRESSED,10,4,1,1,default,NONE,yes,yes,yes,yes,,duplicates diff --git a/cpp/test/reader/table_view/encoding_compression_compatibility_test.cc b/cpp/test/reader/table_view/encoding_compression_compatibility_test.cc index 84b00dac2..0e09fcf3d 100644 --- a/cpp/test/reader/table_view/encoding_compression_compatibility_test.cc +++ b/cpp/test/reader/table_view/encoding_compression_compatibility_test.cc @@ -95,7 +95,8 @@ const char* const kSummaryFile = "summary.md"; const char* const kSpecHeader = "id,model,dataType,encoding,compressions,rowCount,pageRows,chunks,devices," - "valueProfile,nullPatterns,javaWrite,cppWrite,javaRead,cppRead,gapReason"; + "valueProfile,nullPatterns,javaWrite,cppWrite,javaRead,cppRead,gapReason," + "timestampProfile"; const char* const kManifestHeader = "producer,id,model,dataType,encoding,compression,rowCount,pageRows,chunks," "devices,valueProfile,nullPattern"; @@ -127,6 +128,8 @@ enum class ValueProfile { BLOB }; +enum class TimestampProfile { INCREASING, DUPLICATES }; + const char* ModelName(Model model) { return model == Model::TABLE ? "TABLE" : "TREE"; } @@ -199,6 +202,7 @@ struct FixtureCase { bool java_read = false; bool cpp_read = false; std::string gap_reason; + TimestampProfile timestamp_profile = TimestampProfile::INCREASING; bool Writes(const std::string& language) const { return language == "java" ? java_write : cpp_write; @@ -721,6 +725,19 @@ bool ParseProfile(const std::string& value, ValueProfile& out) { return false; } +bool ParseTimestampProfile(const std::string& value, TimestampProfile& out) { + std::string upper = Upper(value); + if (upper.empty() || upper == "INCREASING") { + out = TimestampProfile::INCREASING; + return true; + } + if (upper == "DUPLICATES") { + out = TimestampProfile::DUPLICATES; + return true; + } + return false; +} + bool ParseYesNo(const std::string& value, bool& out) { if (value == "yes") { out = true; @@ -808,7 +825,10 @@ std::vector ReadSpec(const std::string& path) { header_seen = true; continue; } - if (columns.empty() || parts.size() != columns.size()) { + // The trailing timestampProfile column is optional; a row may omit it + // to mean the default INCREASING profile. + if (columns.empty() || (parts.size() != columns.size() && + parts.size() + 1 != columns.size())) { ADD_FAILURE() << "Bad spec line: " << line; continue; } @@ -837,6 +857,15 @@ std::vector ReadSpec(const std::string& path) { continue; } std::string gap_reason = parts[ColumnIndex(columns, "gapReason")]; + TimestampProfile timestamp_profile = TimestampProfile::INCREASING; + int ts_profile_index = ColumnIndex(columns, "timestampProfile"); + if (ts_profile_index >= 0 && + static_cast(ts_profile_index) < parts.size() && + !ParseTimestampProfile(parts[ts_profile_index], + timestamp_profile)) { + ADD_FAILURE() << "Bad timestampProfile in spec line: " << line; + continue; + } Model model; TSDataType data_type; @@ -893,6 +922,7 @@ std::vector ReadSpec(const std::string& path) { fixture_case.java_read = java_read; fixture_case.cpp_read = cpp_read; fixture_case.gap_reason = gap_reason; + fixture_case.timestamp_profile = timestamp_profile; cases.push_back(fixture_case); } } @@ -1169,6 +1199,23 @@ class ScopedPageConfig { uint32_t old_bytes_; }; +// Enable allow_non_decreasing_timestamps for the DUPLICATES profile and +// restore the process-wide flag when the fixture write finishes. +class ScopedAllowNonDecreasing { + public: + explicit ScopedAllowNonDecreasing(const FixtureCase& fixture_case) + : old_(g_config_value_.allow_non_decreasing_timestamps_) { + g_config_value_.allow_non_decreasing_timestamps_ = + fixture_case.timestamp_profile == TimestampProfile::DUPLICATES; + } + ~ScopedAllowNonDecreasing() { + g_config_value_.allow_non_decreasing_timestamps_ = old_; + } + + private: + bool old_; +}; + std::vector> Segments(const FixtureCase& fixture_case) { int chunks = std::max(1, fixture_case.chunks); std::vector> segments; @@ -1183,6 +1230,12 @@ std::vector> Segments(const FixtureCase& fixture_case) { return segments; } +int64_t TimeOf(const FixtureCase& fixture_case, int row) { + return fixture_case.timestamp_profile == TimestampProfile::DUPLICATES + ? static_cast(row) / 2 + : static_cast(row); +} + std::string DeviceId(int device) { return std::string(kDevicePrefix) + std::to_string(device); } @@ -1260,7 +1313,7 @@ Tablet CreateTablet(TableSchema* table_schema, const FixtureCase& fixture_case, static_cast(std::max(1, end - start))); uint32_t index = 0; for (int row = start; row < end; row++, index++) { - EXPECT_EQ(E_OK, tablet.add_timestamp(index, row)); + EXPECT_EQ(E_OK, tablet.add_timestamp(index, TimeOf(fixture_case, row))); EXPECT_EQ(E_OK, tablet.add_value(index, kTagColumn, common::String(DeviceId(device)))); if (!IsNull(fixture_case, row)) { @@ -1332,6 +1385,7 @@ void AddRecordPoint(TsRecord& record, std::string& keep_alive, void WriteTableFixture(const std::string& directory, const FixtureCase& fixture_case) { ScopedPageConfig page_config(fixture_case); + ScopedAllowNonDecreasing allow_non_decreasing(fixture_case); std::string path = JoinPath(directory, fixture_case.FileName()); WriteFile write_file; CreateWriteFile(path, write_file); @@ -1352,6 +1406,7 @@ void WriteTableFixture(const std::string& directory, void WriteTreeFixture(const std::string& directory, const FixtureCase& fixture_case) { ScopedPageConfig page_config(fixture_case); + ScopedAllowNonDecreasing allow_non_decreasing(fixture_case); std::string path = JoinPath(directory, fixture_case.FileName()); WriteFile write_file; CreateWriteFile(path, write_file); @@ -1375,7 +1430,7 @@ void WriteTreeFixture(const std::string& directory, std::string device_id = DeviceId(device); for (const std::pair& segment : Segments(fixture_case)) { for (int row = segment.first; row < segment.second; row++) { - TsRecord record(device_id, row); + TsRecord record(device_id, TimeOf(fixture_case, row)); std::string keep_alive; int points = 0; if (!IsNull(fixture_case, row)) { @@ -1640,7 +1695,7 @@ void ValidateTableRange(const std::string& directory, // from doing a million string copies and map inserts. const int range_len = range.end - range.start + 1; int device = 0; - int expected_time = range.start; + int row = range.start; int rows_in_device = 0; int total_rows = 0; bool tag_verified = false; @@ -1651,8 +1706,8 @@ void ValidateTableRange(const std::string& directory, break; } total_rows++; - SCOPED_TRACE(::testing::Message() - << "device=" << device << " t=" << expected_time); + SCOPED_TRACE(::testing::Message() << "device=" << device << " t=" + << TimeOf(fixture_case, row)); if (!tag_verified) { common::String* tag = owned_result->get_value(2); @@ -1661,25 +1716,23 @@ void ValidateTableRange(const std::string& directory, tag_verified = true; } - EXPECT_EQ(static_cast(expected_time), - owned_result->get_int64_at(1)); + EXPECT_EQ(TimeOf(fixture_case, row), owned_result->get_int64_at(1)); - const bool expected_null = IsNull(fixture_case, expected_time); + const bool expected_null = IsNull(fixture_case, row); EXPECT_EQ(expected_null, owned_result->is_null(3)); if (!expected_null) { - ASSERT_NO_FATAL_FAILURE(ExpectTableValue( - fixture_case, expected_time, owned_result.get())); + ASSERT_NO_FATAL_FAILURE( + ExpectTableValue(fixture_case, row, owned_result.get())); } if (with_anchor) { - EXPECT_EQ(static_cast(expected_time), - owned_result->get_int64_at(4)); + EXPECT_EQ(static_cast(row), owned_result->get_int64_at(4)); } - expected_time++; + row++; if (++rows_in_device == range_len) { device++; - expected_time = range.start; + row = range.start; rows_in_device = 0; tag_verified = false; } @@ -1714,40 +1767,38 @@ void ValidateTreeRange(const std::string& directory, }; std::unique_ptr owned_result(result, cleanup); ResultSetIterator iter = owned_result->iterator(); - int expected_time = range.start; + int row = range.start; while (iter.hasNext()) { RowRecord* record = iter.next(); ASSERT_NE(nullptr, record); if (!with_anchor) { - while (expected_time <= range.end && - IsNull(fixture_case, expected_time)) { - expected_time++; + while (row <= range.end && IsNull(fixture_case, row)) { + row++; } } - ASSERT_LE(expected_time, range.end); - EXPECT_EQ(static_cast(expected_time), record->get_timestamp()); + ASSERT_LE(row, range.end); + EXPECT_EQ(TimeOf(fixture_case, row), record->get_timestamp()); Field* value_field = record->get_field(1); ASSERT_NE(nullptr, value_field); bool is_null = value_field->type_ == NULL_TYPE; - EXPECT_EQ(IsNull(fixture_case, expected_time), is_null); + EXPECT_EQ(IsNull(fixture_case, row), is_null); if (!is_null) { - ExpectFieldValue(fixture_case, expected_time, value_field); + ExpectFieldValue(fixture_case, row, value_field); } if (with_anchor) { Field* anchor_field = record->get_field(2); ASSERT_NE(nullptr, anchor_field); - EXPECT_EQ(static_cast(expected_time), + EXPECT_EQ(static_cast(row), anchor_field->get_value()); } - expected_time++; + row++; } if (!with_anchor) { - while (expected_time <= range.end && - IsNull(fixture_case, expected_time)) { - expected_time++; + while (row <= range.end && IsNull(fixture_case, row)) { + row++; } } - EXPECT_EQ(range.end + 1, expected_time); + EXPECT_EQ(range.end + 1, row); owned_result.reset(); ASSERT_EQ(E_OK, reader.close()); } diff --git a/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java b/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java index 1268045aa..40979584f 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java @@ -127,7 +127,7 @@ private static int shardCount() { private static final String SPEC_HEADER = "id,model,dataType,encoding,compressions,rowCount,pageRows,chunks,devices,valueProfile," - + "nullPatterns,javaWrite,cppWrite,javaRead,cppRead,gapReason"; + + "nullPatterns,javaWrite,cppWrite,javaRead,cppRead,gapReason,timestampProfile"; private static final String MANIFEST_HEADER = "producer,id,model,dataType,encoding,compression,rowCount,pageRows,chunks,devices," + "valueProfile,nullPattern"; @@ -181,6 +181,11 @@ private enum ValueProfile { BLOB } + private enum TimestampProfile { + INCREASING, + DUPLICATES + } + private static final class FixtureCase { final String specId; final String id; @@ -199,6 +204,7 @@ private static final class FixtureCase { final boolean javaRead; final boolean cppRead; final String gapReason; + final TimestampProfile timestampProfile; FixtureCase( String specId, @@ -217,7 +223,8 @@ private static final class FixtureCase { boolean cppWrite, boolean javaRead, boolean cppRead, - String gapReason) { + String gapReason, + TimestampProfile timestampProfile) { this.specId = specId; this.id = id; this.model = model; @@ -235,6 +242,7 @@ private static final class FixtureCase { this.javaRead = javaRead; this.cppRead = cppRead; this.gapReason = gapReason; + this.timestampProfile = timestampProfile; } String fileName() { @@ -254,6 +262,10 @@ String describe() { } } + private static long timeOf(FixtureCase fixtureCase, int row) { + return fixtureCase.timestampProfile == TimestampProfile.DUPLICATES ? row / 2L : row; + } + // --------------------------------------------------------------------------- // Entry points // --------------------------------------------------------------------------- @@ -515,7 +527,9 @@ private static List readSpec(Path specPath) throws IOException { columns = indexColumns(parts, SPEC_HEADER, specPath); continue; } - if (parts.length != columns.size()) { + // The trailing timestampProfile column is optional: a row may omit it to + // mean the default INCREASING profile. + if (parts.length != columns.size() && parts.length != columns.size() - 1) { throw new IllegalArgumentException("Bad spec line (" + parts.length + " fields): " + line); } String id = part(parts, columns, "id"); @@ -534,6 +548,8 @@ private static List readSpec(Path specPath) throws IOException { boolean javaRead = parseYesNo(part(parts, columns, "javaRead")); boolean cppRead = parseYesNo(part(parts, columns, "cppRead")); String gapReason = part(parts, columns, "gapReason"); + TimestampProfile timestampProfile = + parseTimestampProfile(part(parts, columns, "timestampProfile")); List compressionList = "ALL".equals(compressions) @@ -569,7 +585,8 @@ private static List readSpec(Path specPath) throws IOException { cppWrite, javaRead, cppRead, - gapReason)); + gapReason, + timestampProfile)); } } } @@ -597,7 +614,8 @@ private static String part(String[] parts, Map columns, String if (index == null) { throw new IllegalArgumentException("Missing column " + name); } - return parts[index]; + // A spec row may omit the trailing timestampProfile column; default it. + return index < parts.length ? parts[index] : ""; } private static boolean parseYesNo(String value) { @@ -610,6 +628,13 @@ private static boolean parseYesNo(String value) { throw new IllegalArgumentException("Expected yes/no, got: " + value); } + private static TimestampProfile parseTimestampProfile(String value) { + if (value == null || value.isEmpty()) { + return TimestampProfile.INCREASING; + } + return TimestampProfile.valueOf(value.toUpperCase(Locale.ROOT)); + } + private static String capabilityKey(FixtureCase fixtureCase) { return fixtureCase.model + "/" @@ -1258,8 +1283,11 @@ private static void writeTableFixture(File file, FixtureCase fixtureCase) throws TSFileConfig config = TSFileDescriptor.getInstance().getConfig(); int oldPageRows = config.getMaxNumberOfPointsInPage(); int oldPageBytes = config.getPageSizeInByte(); + boolean oldAllowNonDecreasing = config.isAllowNonDecreasingTimestamps(); try { applyPageConfiguration(config, fixtureCase); + config.setAllowNonDecreasingTimestamps( + fixtureCase.timestampProfile == TimestampProfile.DUPLICATES); try (TsFileWriter writer = new TsFileWriter(file)) { writer.setGenerateTableSchema(true); writer.registerTableSchema(tableSchema); @@ -1273,6 +1301,7 @@ private static void writeTableFixture(File file, FixtureCase fixtureCase) throws } finally { config.setMaxNumberOfPointsInPage(oldPageRows); config.setPageSizeInByte(oldPageBytes); + config.setAllowNonDecreasingTimestamps(oldAllowNonDecreasing); } } @@ -1280,8 +1309,11 @@ private static void writeTreeFixture(File file, FixtureCase fixtureCase) throws TSFileConfig config = TSFileDescriptor.getInstance().getConfig(); int oldPageRows = config.getMaxNumberOfPointsInPage(); int oldPageBytes = config.getPageSizeInByte(); + boolean oldAllowNonDecreasing = config.isAllowNonDecreasingTimestamps(); try { applyPageConfiguration(config, fixtureCase); + config.setAllowNonDecreasingTimestamps( + fixtureCase.timestampProfile == TimestampProfile.DUPLICATES); try (TsFileWriter writer = new TsFileWriter(file)) { for (int device = 0; device < fixtureCase.devices; device++) { org.apache.tsfile.read.common.Path devicePath = @@ -1298,7 +1330,7 @@ private static void writeTreeFixture(File file, FixtureCase fixtureCase) throws String deviceId = deviceId(device); for (int[] segment : segments(fixtureCase)) { for (int row = segment[0]; row < segment[1]; row++) { - TSRecord record = new TSRecord(deviceId, row); + TSRecord record = new TSRecord(deviceId, timeOf(fixtureCase, row)); int points = 0; if (!isNull(fixtureCase, row)) { record.addTuple(dataPoint(fixtureCase, row)); @@ -1319,6 +1351,7 @@ private static void writeTreeFixture(File file, FixtureCase fixtureCase) throws } finally { config.setMaxNumberOfPointsInPage(oldPageRows); config.setPageSizeInByte(oldPageBytes); + config.setAllowNonDecreasingTimestamps(oldAllowNonDecreasing); } } @@ -1386,7 +1419,7 @@ private static Tablet tablet( Math.max(1, end - start)); int index = 0; for (int row = start; row < end; row++, index++) { - tablet.addTimestamp(index, row); + tablet.addTimestamp(index, timeOf(fixtureCase, row)); tablet.addValue(TAG_COLUMN, index, deviceId(device)); if (isNull(fixtureCase, row)) { tablet.addValue(VALUE_COLUMN, index, null); @@ -1551,7 +1584,7 @@ private static void validateTableRange( // inline instead of buffering every row, so a 1M-row fixture does not keep // millions of boxed values and arrays alive until the query finishes. int device = 0; - int expectedTime = start; + int row = start; int rowsInDevice = 0; int totalRows = 0; boolean tagVerified = false; @@ -1566,28 +1599,28 @@ private static void validateTableRange( } long time = resultSet.getLong(1); - assertEquals(fixtureCase.id + " timestamp", expectedTime, time); + assertEquals(fixtureCase.id + " timestamp", timeOf(fixtureCase, row), time); boolean isNull = resultSet.isNull(3); - boolean expectedNull = isNull(fixtureCase, expectedTime); - assertEquals(fixtureCase.id + " null at t=" + expectedTime, expectedNull, isNull); + boolean expectedNull = isNull(fixtureCase, row); + assertEquals(fixtureCase.id + " null at t=" + time, expectedNull, isNull); if (!expectedNull) { assertValue( fixtureCase, - expectedValue(fixtureCase, expectedTime), + expectedValue(fixtureCase, row), readResultValue(fixtureCase, resultSet, 3), - expectedTime); + time); } if (withAnchor) { assertFalse("anchor null at " + fixtureCase.id + " t=" + time, resultSet.isNull(4)); assertEquals( - fixtureCase.id + " anchor at t=" + expectedTime, expectedTime, resultSet.getLong(4)); + fixtureCase.id + " anchor at t=" + time, row, resultSet.getLong(4)); } - expectedTime++; + row++; if (++rowsInDevice == end - start + 1) { device++; - expectedTime = start; + row = start; rowsInDevice = 0; tagVerified = false; } @@ -1625,50 +1658,50 @@ private static void validateTreeRange( try (TsFileSequenceReader sequenceReader = new TsFileSequenceReader(file.getPath())) { try (TsFileReader reader = new TsFileReader(sequenceReader)) { QueryDataSet dataSet = reader.query(expression); - long expectedTime = start; + int row = start; while (dataSet.hasNext()) { RowRecord record = dataSet.next(); if (!withAnchor) { // Without the anchor, timestamps whose value is missing are not // returned at all; skip to the next present value. - while (expectedTime <= end && isNull(fixtureCase, (int) expectedTime)) { - expectedTime++; + while (row <= end && isNull(fixtureCase, row)) { + row++; } } assertEquals( fixtureCase.id + " tree timestamp for device " + device, - expectedTime, + timeOf(fixtureCase, row), record.getTimestamp()); Field valueField = record.getFields().get(0); boolean isNull = valueField == null || valueField.getDataType() == null; assertEquals( - fixtureCase.id + " tree null at t=" + expectedTime, - isNull(fixtureCase, (int) expectedTime), + fixtureCase.id + " tree null at t=" + timeOf(fixtureCase, row), + isNull(fixtureCase, row), isNull); if (!isNull) { assertFieldValue( fixtureCase, - expectedValue(fixtureCase, (int) expectedTime), + expectedValue(fixtureCase, row), valueField, - expectedTime); + timeOf(fixtureCase, row)); } if (withAnchor) { Field anchorField = record.getFields().get(1); assertNotNull(anchorField); assertEquals( - fixtureCase.id + " tree anchor at t=" + expectedTime, - expectedTime, + fixtureCase.id + " tree anchor at t=" + timeOf(fixtureCase, row), + row, anchorField.getLongV()); } - expectedTime++; + row++; } if (!withAnchor) { - while (expectedTime <= end && isNull(fixtureCase, (int) expectedTime)) { - expectedTime++; + while (row <= end && isNull(fixtureCase, row)) { + row++; } } assertEquals( - fixtureCase.id + " tree row count for device " + device, end + 1, expectedTime); + fixtureCase.id + " tree row count for device " + device, end + 1, row); } } } From 3620ac295025e9374f96a07cf966502f10307727 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 11 Nov 2026 15:32:03 +0800 Subject: [PATCH 5/7] style(compat): apply spotless formatting to EncodingCompressionCompatibilityTest --- .../EncodingCompressionCompatibilityTest.java | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java b/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java index 40979584f..b83bb5bb6 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java @@ -1613,8 +1613,7 @@ private static void validateTableRange( } if (withAnchor) { assertFalse("anchor null at " + fixtureCase.id + " t=" + time, resultSet.isNull(4)); - assertEquals( - fixtureCase.id + " anchor at t=" + time, row, resultSet.getLong(4)); + assertEquals(fixtureCase.id + " anchor at t=" + time, row, resultSet.getLong(4)); } row++; @@ -1680,10 +1679,7 @@ private static void validateTreeRange( isNull); if (!isNull) { assertFieldValue( - fixtureCase, - expectedValue(fixtureCase, row), - valueField, - timeOf(fixtureCase, row)); + fixtureCase, expectedValue(fixtureCase, row), valueField, timeOf(fixtureCase, row)); } if (withAnchor) { Field anchorField = record.getFields().get(1); @@ -1700,8 +1696,7 @@ private static void validateTreeRange( row++; } } - assertEquals( - fixtureCase.id + " tree row count for device " + device, end + 1, row); + assertEquals(fixtureCase.id + " tree row count for device " + device, end + 1, row); } } } From eddaf114c67fed12777f458e9749326a8cff4665 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 11 Nov 2026 19:19:25 +0800 Subject: [PATCH 6/7] feat: record the non-decreasing timestamp policy as a TsFile property When the writer runs with allowNonDecreasingTimestamps, register the file-level property "allowNonDecreasingTimestamps" = "true" so downstream applications (e.g. an IoTDB load) can tell the file may contain duplicate timestamps and must not assume strictly increasing times within a series. The property is added at writer initialization in both Java (TsFileIOWriter, including the append path in RestorableTsFileIOWriter) and C++ (TsFileIOWriter init, which both the tree and table writers use). Flag-off writes add nothing, so existing files are unchanged. --- cpp/src/common/constant/tsfile_constant.h | 8 +++ cpp/src/file/tsfile_io_writer.cc | 11 ++++ cpp/test/writer/tsfile_properties_test.cc | 52 +++++++++++++++++++ .../writer/RestorableTsFileIOWriter.java | 3 ++ .../tsfile/write/writer/TsFileIOWriter.java | 21 ++++++++ .../AllowNonDecreasingTimestampsTest.java | 32 ++++++++++++ .../test_allow_non_decreasing_timestamps.py | 20 +++++++ 7 files changed, 147 insertions(+) diff --git a/cpp/src/common/constant/tsfile_constant.h b/cpp/src/common/constant/tsfile_constant.h index 01ad1b2bc..0c9c3e8d0 100644 --- a/cpp/src/common/constant/tsfile_constant.h +++ b/cpp/src/common/constant/tsfile_constant.h @@ -41,6 +41,14 @@ static const std::string TIME_COLUMN_NAME = "time"; static const unsigned char VALUE_COLUMN_MASK = 0x40; static const int NO_STR_TO_READ = -1; +// File-level property written when the writer is created with +// allow_non_decreasing_timestamps_ enabled. Its presence (value "true") tells +// downstream readers that the file may contain duplicate timestamps, so they +// must not assume strictly increasing times within a series. +static const std::string TSFILE_PROP_ALLOW_NON_DECREASING_TIMESTAMPS = + "allowNonDecreasingTimestamps"; +static const std::string TSFILE_PROP_VALUE_TRUE = "true"; + // Identifier characters: ASCII alphanumerics / underscore plus any non-ASCII // (high) byte. The \x80-\xFF byte range covers UTF-8 lead/continuation bytes // (e.g. CJK characters) and is accepted by every std::regex implementation; diff --git a/cpp/src/file/tsfile_io_writer.cc b/cpp/src/file/tsfile_io_writer.cc index 29ddf0d90..7e2fc52d5 100644 --- a/cpp/src/file/tsfile_io_writer.cc +++ b/cpp/src/file/tsfile_io_writer.cc @@ -26,6 +26,7 @@ #include #include +#include "common/constant/tsfile_constant.h" #include "common/device_id.h" #include "common/global.h" #include "common/logger/elog.h" @@ -53,6 +54,16 @@ int TsFileIOWriter::init(WriteFile* write_file) { meta_allocator_.init(page_size, MOD_TSFILE_WRITER_META); chunk_meta_count_ = 0; file_ = write_file; + // Record the writer's timestamp policy so downstream readers (e.g. an + // IoTDB load) can tell that series in this file may contain duplicate + // timestamps. Only written when the flag is on, so flag-off files keep + // their metadata byte-for-byte unchanged. + if (common::g_config_value_.allow_non_decreasing_timestamps_) { + tsfile_properties_[TSFILE_PROP_ALLOW_NON_DECREASING_TIMESTAMPS] = + TsFilePropertyValue( + reinterpret_cast(TSFILE_PROP_VALUE_TRUE.data()), + static_cast(TSFILE_PROP_VALUE_TRUE.size())); + } // Re-arm destroy() for the new lifecycle. Without this, a writer that // was destroy()'d and then init()'d again would leak the fresh // meta_allocator_/write_stream_/file_ on its next destroy(). diff --git a/cpp/test/writer/tsfile_properties_test.cc b/cpp/test/writer/tsfile_properties_test.cc index 4590c1bcd..58dd37543 100644 --- a/cpp/test/writer/tsfile_properties_test.cc +++ b/cpp/test/writer/tsfile_properties_test.cc @@ -22,6 +22,7 @@ #include #include +#include "common/constant/tsfile_constant.h" #include "common/global.h" #include "reader/tsfile_reader.h" #include "writer/tsfile_writer.h" @@ -34,6 +35,18 @@ std::vector Bytes(const std::string& value) { return std::vector(value.begin(), value.end()); } +// allow_non_decreasing_timestamps_ is process-global; restore it on scope exit +// so an ASSERT failure cannot leak the setting into later tests. +class ScopedAllowNonDecreasingTimestamps { + public: + explicit ScopedAllowNonDecreasingTimestamps(bool value) { + common::set_allow_non_decreasing_timestamps(value); + } + ~ScopedAllowNonDecreasingTimestamps() { + common::set_allow_non_decreasing_timestamps(false); + } +}; + class TsFilePropertiesTest : public ::testing::Test { protected: void SetUp() override { @@ -107,6 +120,45 @@ TEST_F(TsFilePropertiesTest, WriterPreservesBinaryNullAndEmptyValues) { EXPECT_EQ(common::E_OK, reader.close()); } +// A writer created with allow_non_decreasing_timestamps_ must record the +// policy in the file metadata so downstream readers can tell the file may +// contain duplicate timestamps. +TEST_F(TsFilePropertiesTest, RecordsAllowNonDecreasingTimestampsPolicy) { + { + ScopedAllowNonDecreasingTimestamps guard(true); + TsFileWriter writer; + ASSERT_EQ(common::E_OK, writer.open(file_name_)); + ASSERT_EQ(common::E_OK, writer.close()); + } + + TsFileReader reader; + ASSERT_EQ(common::E_OK, reader.open(file_name_)); + TsFileProperties properties = reader.get_tsfile_properties(); + auto it = properties.find(TSFILE_PROP_ALLOW_NON_DECREASING_TIMESTAMPS); + ASSERT_NE(it, properties.end()); + EXPECT_FALSE(it->second.is_null); + EXPECT_EQ(Bytes("true"), it->second.value); + EXPECT_EQ(common::E_OK, reader.close()); +} + +// Flag off keeps the metadata free of the policy property, so existing files +// are byte-for-byte unchanged. +TEST_F(TsFilePropertiesTest, OmitsPolicyPropertyWhenFlagOff) { + { + ScopedAllowNonDecreasingTimestamps guard(false); + TsFileWriter writer; + ASSERT_EQ(common::E_OK, writer.open(file_name_)); + ASSERT_EQ(common::E_OK, writer.close()); + } + + TsFileReader reader; + ASSERT_EQ(common::E_OK, reader.open(file_name_)); + TsFileProperties properties = reader.get_tsfile_properties(); + EXPECT_EQ(properties.find(TSFILE_PROP_ALLOW_NON_DECREASING_TIMESTAMPS), + properties.end()); + EXPECT_EQ(common::E_OK, reader.close()); +} + } // namespace } // namespace storage diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java index 13dd66d47..df2d5a3d7 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java @@ -128,6 +128,9 @@ public RestorableTsFileIOWriter(File file, boolean truncate, EncryptParameter pa this.out = FSFactoryProducer.getFileOutputFactory().getTsFileOutput(file.getPath(), true); this.param = param != null && param.isTdePageAead() ? param.copy() : param; setEncryptParam(this.param); + // Appending may still introduce duplicate timestamps; record the policy so + // the rewritten metadata states how readers must interpret the file. + registerInterpretationProperties(); // file doesn't exist if (file.length() == 0) { diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java index abf45b076..b6c34ccc0 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java @@ -130,6 +130,14 @@ public class TsFileIOWriter implements AutoCloseable { public static final String CHUNK_METADATA_TEMP_FILE_SUFFIX = ".meta"; public static final String TABLE_POINT_COUNT_PROPERTY_PREFIX = "tablePointCount."; + /** + * File property written when {@link TSFileConfig#isAllowNonDecreasingTimestamps()} is enabled. + * Its presence (value {@code "true"}) tells downstream readers that the file may contain + * duplicate timestamps, so they must not assume strictly increasing times within a series. + */ + public static final String ALLOW_NON_DECREASING_TIMESTAMPS_PROPERTY = + "allowNonDecreasingTimestamps"; + private boolean generateTableSchema = false; protected String encryptLevel; @@ -381,9 +389,22 @@ protected void startFile() throws IOException { out.write(MAGIC_STRING_BYTES); out.write(VERSION_NUMBER_BYTE); fileStarted = true; + registerInterpretationProperties(); writeEncryptionHeaderIfNecessary(); } + /** + * Record writer options that change how readers must interpret the file. Called when the file is + * created and when an existing file is reopened for appending, so the final metadata always + * reflects the policy this writer ran with. + */ + protected void registerInterpretationProperties() { + if (TS_FILE_CONFIG.isAllowNonDecreasingTimestamps()) { + tsFileProperties.put( + ALLOW_NON_DECREASING_TIMESTAMPS_PROPERTY, "true".getBytes(TSFileConfig.STRING_CHARSET)); + } + } + public int startChunkGroup(IDeviceID deviceId) throws IOException { writeEncryptionHeaderIfNecessary(); updateTableSize(deviceId); diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java index 4575365a6..a7690cc88 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java @@ -34,6 +34,7 @@ import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.record.datapoint.IntDataPoint; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; import org.junit.After; import org.junit.Assert; @@ -142,6 +143,37 @@ public void duplicateStillRejectedWhenFlagOff() throws IOException, WriteProcess } } + @Test + public void recordsPolicyPropertyWhenFlagOn() throws IOException, WriteProcessException { + config.setAllowNonDecreasingTimestamps(true); + writeSingleRecord(); + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath())) { + byte[] value = + reader.getTsFileProperties().get(TsFileIOWriter.ALLOW_NON_DECREASING_TIMESTAMPS_PROPERTY); + Assert.assertArrayEquals("true".getBytes(TSFileConfig.STRING_CHARSET), value); + } + } + + @Test + public void omitsPolicyPropertyWhenFlagOff() throws IOException, WriteProcessException { + writeSingleRecord(); + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath())) { + Assert.assertFalse( + reader + .getTsFileProperties() + .containsKey(TsFileIOWriter.ALLOW_NON_DECREASING_TIMESTAMPS_PROPERTY)); + } + } + + private void writeSingleRecord() throws IOException, WriteProcessException { + try (TsFileWriter writer = new TsFileWriter(file)) { + writer.registerTimeseries(new Path(deviceId), Arrays.asList(s1)); + TSRecord record = new TSRecord(deviceId, 1); + record.addTuple(new IntDataPoint("s1", 1)); + writer.writeRecord(record); + } + } + private List readTimestamps(String device, String measurement) throws IOException { List times = new ArrayList<>(); try (TsFileSequenceReader sequenceReader = new TsFileSequenceReader(file.getAbsolutePath()); diff --git a/python/tests/test_allow_non_decreasing_timestamps.py b/python/tests/test_allow_non_decreasing_timestamps.py index 50166ab69..2fba75a37 100644 --- a/python/tests/test_allow_non_decreasing_timestamps.py +++ b/python/tests/test_allow_non_decreasing_timestamps.py @@ -29,6 +29,7 @@ TableSchema, TSDataType, Tablet, + TsFileReader, TsFileTableWriter, to_dataframe, ) @@ -74,6 +75,25 @@ def test_table_writer_duplicates_allowed(): df = to_dataframe(path, table_name=TABLE_NAME) assert df.shape[0] == 4 assert list(df[TIME_COLUMN]) == [10, 11, 11, 12] + with TsFileReader(path) as reader: + assert ( + reader.get_tsfile_properties()["allowNonDecreasingTimestamps"] + == b"true" + ) + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + +def test_policy_property_absent_when_flag_off(): + tmpdir = _new_tmpdir() + path = os.path.join(tmpdir, "strict.tsfile") + try: + with TsFileTableWriter( + path, _table_schema(), allow_non_decreasing_timestamps=False + ) as writer: + writer.write_table(_build_tablet([10, 11, 12])) + with TsFileReader(path) as reader: + assert "allowNonDecreasingTimestamps" not in reader.get_tsfile_properties() finally: shutil.rmtree(tmpdir, ignore_errors=True) From 421cd653edbc6ff2491e25a6180798d2fabc9cc6 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 11 Nov 2026 21:46:47 +0800 Subject: [PATCH 7/7] test(cpp): isolate TsFilePropertiesTest cases into per-test files CTest runs cases of one binary in parallel (--parallel 4). All cases in this fixture shared one fixed file name, so the new policy-property cases truncated or removed the file while WriterPreservesBinaryNullAndEmptyValues was still using it, failing that test on Linux and Windows CI. Derive the file name from the current test name instead. --- cpp/test/writer/tsfile_properties_test.cc | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/cpp/test/writer/tsfile_properties_test.cc b/cpp/test/writer/tsfile_properties_test.cc index 58dd37543..fd5eef718 100644 --- a/cpp/test/writer/tsfile_properties_test.cc +++ b/cpp/test/writer/tsfile_properties_test.cc @@ -51,7 +51,14 @@ class TsFilePropertiesTest : public ::testing::Test { protected: void SetUp() override { libtsfile_init(); - file_name_ = "tsfile_properties_test.tsfile"; + // CTest runs cases of one binary in parallel; a shared file name would + // let two cases truncate or remove each other's fixture mid-test. + const ::testing::TestInfo* info = + ::testing::UnitTest::GetInstance()->current_test_info(); + file_name_ = std::string("tsfile_properties_test_") + + (info == nullptr ? std::string("unknown") + : std::string(info->name())) + + ".tsfile"; std::remove(file_name_.c_str()); }