diff --git a/compat/README.md b/compat/README.md
index e0154a2ec9..1a453fba87 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 0167e4242f..591d2caa2d 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/src/common/config/config.h b/cpp/src/common/config/config.h
index 1293ebecba..8230614a33 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/constant/tsfile_constant.h b/cpp/src/common/constant/tsfile_constant.h
index 01ad1b2bc0..0c9c3e8d00 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/common/global.h b/cpp/src/common/global.h
index 89538c9dcb..b7fdaa3966 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 be97ef484f..6eaffff552 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 6d9985c01a..baa6621421 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/file/tsfile_io_writer.cc b/cpp/src/file/tsfile_io_writer.cc
index 29ddf0d90a..7e2fc52d54 100644
--- a/cpp/src/file/tsfile_io_writer.cc
+++ b/cpp/src/file/tsfile_io_writer.cc
@@ -26,6 +26,7 @@
 #include <limits>
 #include <memory>
 
+#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<const uint8_t*>(TSFILE_PROP_VALUE_TRUE.data()),
+                static_cast<uint32_t>(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/src/writer/time_page_writer.h b/cpp/src/writer/time_page_writer.h
index bda9a50232..33d6860edf 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 ed3e09be70..6fbc806d7d 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<IDeviceID>& 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<StringArrayDeviceID>(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<ChunkWriter*> chunk_writers;
     SimpleVector<ChunkWriter*> 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<StringArrayDeviceID>(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<ValueChunkWriter*> value_chunk_writers;
     SimpleVector<common::TSDataType> data_types;
@@ -1056,13 +1062,9 @@ int TsFileWriter::write_tablet_aligned(const Tablet& tablet) {
     auto device_id =
         std::make_shared<StringArrayDeviceID>(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<ValueChunkWriter*> 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<ChunkWriter*> chunk_writers;
     SimpleVector<common::TSDataType> 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<uint32_t>(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<ChunkWriter*> chunk_writers;
diff --git a/cpp/src/writer/tsfile_writer.h b/cpp/src/writer/tsfile_writer.h
index 0606a7a33e..b5fb86702a 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<IDeviceID>& 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/table_view/encoding_compression_compatibility_test.cc b/cpp/test/reader/table_view/encoding_compression_compatibility_test.cc
index 84b00dac27..0e09fcf3d0 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<FixtureCase> 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<FixtureCase> 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<size_t>(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<FixtureCase> 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<std::pair<int, int>> Segments(const FixtureCase& fixture_case) {
     int chunks = std::max(1, fixture_case.chunks);
     std::vector<std::pair<int, int>> segments;
@@ -1183,6 +1230,12 @@ std::vector<std::pair<int, int>> 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<int64_t>(row) / 2
+               : static_cast<int64_t>(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<uint32_t>(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<int, int>& 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<common::String*>(2);
@@ -1661,25 +1716,23 @@ void ValidateTableRange(const std::string& directory,
             tag_verified = true;
         }
 
-        EXPECT_EQ(static_cast<int64_t>(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<int64_t>(expected_time),
-                      owned_result->get_int64_at(4));
+            EXPECT_EQ(static_cast<int64_t>(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<ResultSet, decltype(cleanup)> 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<int64_t>(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<int64_t>(expected_time),
+            EXPECT_EQ(static_cast<int64_t>(row),
                       anchor_field->get_value<int64_t>());
         }
-        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/cpp/test/writer/tsfile_properties_test.cc b/cpp/test/writer/tsfile_properties_test.cc
index 4590c1bcde..fd5eef7185 100644
--- a/cpp/test/writer/tsfile_properties_test.cc
+++ b/cpp/test/writer/tsfile_properties_test.cc
@@ -22,6 +22,7 @@
 #include <cstdio>
 #include <vector>
 
+#include "common/constant/tsfile_constant.h"
 #include "common/global.h"
 #include "reader/tsfile_reader.h"
 #include "writer/tsfile_writer.h"
@@ -34,11 +35,30 @@ std::vector<uint8_t> Bytes(const std::string& value) {
     return std::vector<uint8_t>(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 {
         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());
     }
 
@@ -107,6 +127,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/cpp/test/writer/tsfile_writer_test.cc b/cpp/test/writer/tsfile_writer_test.cc
index 1727fa5b8d..860d6a1648 100644
--- a/cpp/test/writer/tsfile_writer_test.cc
+++ b/cpp/test/writer/tsfile_writer_test.cc
@@ -32,6 +32,7 @@
 #include <unistd.h>
 #endif
 
+#include "common/global.h"
 #include "common/path.h"
 #include "common/record.h"
 #include "common/schema.h"
@@ -2418,3 +2419,213 @@ 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<MeasurementSchema> schema_vec;
+    schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+    schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED);
+    {
+        std::vector<MeasurementSchema*> 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<std::vector<MeasurementSchema>>(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<int64_t>(i)), E_OK);
+        ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(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<storage::Path> 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<MeasurementSchema> schema_vec;
+    schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+    {
+        std::vector<MeasurementSchema*> 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<int64_t>(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<storage::Path> 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<MeasurementSchema> schema_vec;
+    schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+    {
+        std::vector<MeasurementSchema*> 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<std::vector<MeasurementSchema>>(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<int64_t>(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<MeasurementSchema> schema_vec;
+    schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+    {
+        std::vector<MeasurementSchema*> 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<std::vector<MeasurementSchema>>(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<int64_t>(i)), E_OK);
+    }
+    EXPECT_NE(tsfile_writer_->write_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<int64_t>(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<storage::Path> 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);
+}
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 6c79d65677..cffade760b 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 00e97cb717..24eae13ee8 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 64f4b62970..ae1e6aeb0d 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<String, Long> 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 3a97619dc6..647ffe9bb9 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/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java
index 13dd66d471..df2d5a3d76 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 abf45b0766..b6c34ccc04 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/compatibility/EncodingCompressionCompatibilityTest.java b/java/tsfile/src/test/java/org/apache/tsfile/compatibility/EncodingCompressionCompatibilityTest.java
index 1268045aaf..b83bb5bb60 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<FixtureCase> 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<FixtureCase> 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<String> compressionList =
           "ALL".equals(compressions)
@@ -569,7 +585,8 @@ private static List<FixtureCase> readSpec(Path specPath) throws IOException {
                   cppWrite,
                   javaRead,
                   cppRead,
-                  gapReason));
+                  gapReason,
+                  timestampProfile));
         }
       }
     }
@@ -597,7 +614,8 @@ private static String part(String[] parts, Map<String, Integer> 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,27 @@ 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));
+          assertEquals(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 +1657,46 @@ 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),
-                valueField,
-                expectedTime);
+                fixtureCase, expectedValue(fixtureCase, row), valueField, 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);
+        assertEquals(fixtureCase.id + " tree row count for device " + device, end + 1, row);
       }
     }
   }
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 0000000000..a7690cc887
--- /dev/null
+++ b/java/tsfile/src/test/java/org/apache/tsfile/write/AllowNonDecreasingTimestampsTest.java
@@ -0,0 +1,192 @@
+/*
+ * 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.apache.tsfile.write.writer.TsFileIOWriter;
+
+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<Long> 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<Long> 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));
+    }
+  }
+
+  @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<Long> readTimestamps(String device, String measurement) throws IOException {
+    List<Long> 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 0000000000..2fba75a37d
--- /dev/null
+++ b/python/tests/test_allow_non_decreasing_timestamps.py
@@ -0,0 +1,131 @@
+# 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 shutil
+import tempfile
+
+import pytest
+
+from tsfile import (
+    TIME_COLUMN,
+    ColumnCategory,
+    ColumnSchema,
+    TableSchema,
+    TSDataType,
+    Tablet,
+    TsFileReader,
+    TsFileTableWriter,
+    to_dataframe,
+)
+from tsfile.exceptions import OutOfOrderError
+
+TABLE_NAME = "dup_table"
+
+
+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 _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(
+            path, _table_schema(), allow_non_decreasing_timestamps=True
+        ) as writer:
+            writer.write_table(_build_tablet([10, 11, 11, 12]))
+        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)
+
+
+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):
+    tmpdir = _new_tmpdir()
+    path = os.path.join(tmpdir, "dup.tsfile")
+    try:
+        writer = TsFileTableWriter(
+            path,
+            _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() 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:
+        shutil.rmtree(tmpdir, ignore_errors=True)
diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd
index d5f2539cf5..e4f75b8684 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 b6a7795b6e..ae1083b4d3 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(
+            <bint> 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 e4fceef976..6e525ff982 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 7ae95c635d..7f0bd57ff7 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(
+            <bint> allow_non_decreasing_timestamps)
         self.writer = tsfile_writer_new_c(pathname, memory_threshold)
 
     def register_timeseries(self, device_name : str, timeseries_schema : TimeseriesSchemaPy):