diff --git a/docs/en/12-taos-sql/10-function.md b/docs/en/12-taos-sql/10-function.md
index 18c7ffc345..2ea144c56a 100644
--- a/docs/en/12-taos-sql/10-function.md
+++ b/docs/en/12-taos-sql/10-function.md
@@ -539,7 +539,8 @@ TO_CHAR(ts, format_str_literal)
 - When `ms`,`us`,`ns` are used in `to_char`, like `to_char(ts, 'yyyy-mm-dd hh:mi:ss.ms.us.ns')`, The time of `ms`,`us`,`ns` corresponds to the same fraction seconds. When ts is `1697182085123`, the output of `ms` is `123`, `us` is `123000`, `ns` is `123000000`.
 - If we want to output some characters of format without converting, surround it with double quotes. `to_char(ts, 'yyyy-mm-dd "is formated by yyyy-mm-dd"')`. If want to output double quotes, add a back slash before double quote, like `to_char(ts, '\"yyyy-mm-dd\"')` will output `"2023-10-10"`.
 - For formats that output digits, the uppercase and lowercase formats are the same.
-- It's recommended to put time zone in the format, if not, the default time zone zone will be that in server or client.
+- It's recommended to put time zone in the format, if not, the default time zone will be that in server or client.
+- The precision of the input timestamp will be recognized automatically according to the precision of the table used, milliseconds will be used if no table is specified.
 
 #### TO_TIMESTAMP
 
@@ -564,9 +565,10 @@ TO_TIMESTAMP(ts_str_literal, format_str_literal)
 - The uppercase or lowercase of `MONTH`, `MON`, `DAY`, `DY` and formtas that output digits have same effect when used in `to_timestamp`, like `to_timestamp('2023-JANUARY-01', 'YYYY-month-dd')`, `month` can be replaced by `MONTH`, or `month`. The cases are ignored.
 - If multi times are specified for one component, the previous will be overwritten. Like `to_timestamp('2023-22-10-10', 'yyyy-yy-MM-dd')`, the output year will be `2022`.
 - To avoid unexpected time zone used during the convertion, it's recommended to put time zone in the ts string, e.g. '2023-10-10 10:10:10+08'. If time zone not specified, default will be that in server or client.
-- The default timestamp if some components are not specified will be: `1970-01-01 00:00:00` with specified or default local timezone.
+- The default timestamp if some components are not specified will be: `1970-01-01 00:00:00` with the timezone specified or default to local timezone.
 - If `AM` or `PM` is specified in formats, the Hour must between `1-12`.
 - In some cases, `to_timestamp` can convert correctly even the format and the timestamp string are not totally matched. Like `to_timetamp('200101/2', 'yyyyMM1/dd')`, the digit `1` in format string are ignored, and the output timestsamp is `2001-01-02 00:00:00`. Spaces and tabs in formats and tiemstamp string are also ignored automatically.
+- The precision of the output timestamp will be the same as the table in SELECT stmt, millisecond will be used if no table is specified. The output of `select to_timestamp('2023-08-1 10:10:10.123456789', 'yyyy-mm-dd hh:mi:ss.ns')` will be truncated to millisecond precision. If a nano precision table is specified, no truncation will be applied. Like `select to_timestamp('2023-08-1 10:10:10.123456789', 'yyyy-mm-dd hh:mi:ss.ns') from db_ns.table_ns limit 1`.
 
 
 ### Time and Date Functions
diff --git a/docs/zh/12-taos-sql/10-function.md b/docs/zh/12-taos-sql/10-function.md
index 4371124623..c1dc6a6363 100644
--- a/docs/zh/12-taos-sql/10-function.md
+++ b/docs/zh/12-taos-sql/10-function.md
@@ -540,6 +540,7 @@ TO_CHAR(ts, format_str_literal)
 - 时间格式中无法匹配规则的内容会直接输出. 如果想要在格式串中指定某些能够匹配规则的部分不做转换, 可以使用双引号, 如`to_char(ts, 'yyyy-mm-dd "is formated by yyyy-mm-dd"')`. 如果想要输出双引号, 那么在双引号之前加一个反斜杠, 如 `to_char(ts, '\"yyyy-mm-dd\"')` 将会输出 `"2023-10-10"`.
 - 那些输出是数字的格式, 如`YYYY`, `DD`, 大写与小写意义相同, 即`yyyy` 和 `YYYY` 可以互换.
 - 推荐在时间格式中带时区信息，如果不带则默认输出的时区为服务端或客户端所配置的时区.
+- 输入时间戳的精度由所查询表的精度确定, 若未指定表, 则精度为毫秒.
 
 #### TO_TIMESTAMP
 
@@ -560,13 +561,14 @@ TO_TIMESTAMP(ts_str_literal, format_str_literal)
 **支持的格式**: 与`to_char`相同
 
 **使用说明**:
-- 若`ms`, `us`, `ns`同时指定, 那么结果时间戳包含上述三个字段的和. 如 `to_timestamp('2023-10-10 10:10:10.123.000456.000000789', 'yyyy-mm-dd hh:mi:ss.ms.us.ns')` 输出是 `2023-10-10 10:10:10.123456789`.
+- 若`ms`, `us`, `ns`同时指定, 那么结果时间戳包含上述三个字段的和. 如 `to_timestamp('2023-10-10 10:10:10.123.000456.000000789', 'yyyy-mm-dd hh:mi:ss.ms.us.ns')` 输出为 `2023-10-10 10:10:10.123456789`对应的时间戳.
 - `MONTH`, `MON`, `DAY`, `DY` 以及其他输出为数字的格式的大小写意义相同, 如 `to_timestamp('2023-JANUARY-01', 'YYYY-month-dd')`, `month`可以被替换为`MONTH` 或者`Month`.
 - 如果同一字段被指定了多次, 那么前面的指定将会被覆盖.  如 `to_timestamp('2023-22-10-10', 'yyyy-yy-MM-dd')`, 输出年份是`2022`.
 - 为避免转换时使用了非预期的时区，推荐在时间中携带时区信息，例如'2023-10-10 10:10:10+08'，如果未指定时区则默认时区为服务端或客户端指定的时区。
 - 如果没有指定完整的时间，那么默认时间值为指定或默认时区的 `1970-01-01 00:00:00`, 未指定部分使用该默认值中的对应部分.
 - 如果格式串中有`AM`, `PM`等, 那么小时必须是12小时制, 范围必须是01-12.
 - `to_timestamp`转换具有一定的容错机制, 在格式串和时间戳串不完全对应时, 有时也可转换, 如: `to_timestamp('200101/2', 'yyyyMM1/dd')`, 格式串中多出来的1会被丢弃. 格式串与时间戳串中多余的空格字符(空格, tab等)也会被 自动忽略. 如`to_timestamp('  23 年 - 1 月 - 01 日  ', 'yy 年-MM月-dd日')` 可以被成功转换. 虽然`MM`等字段需要两个数字对应(只有一位时前面补0), 在`to_timestamp`时, 一个数字也可以成功转换.
+- 输出时间戳的精度与查询表的精度相同, 若查询未指定表, 则输出精度为毫秒. 如`select to_timestamp('2023-08-1 10:10:10.123456789', 'yyyy-mm-dd hh:mi:ss.ns')`的输出将会把微妙和纳秒进行截断. 如果指定一张纳秒表, 那么就不会发生截断, 如`select to_timestamp('2023-08-1 10:10:10.123456789', 'yyyy-mm-dd hh:mi:ss.ns') from db_ns.table_ns limit 1`.
 
 
 ### 时间和日期函数
diff --git a/include/common/tglobal.h b/include/common/tglobal.h
index 1d0da9e150..b9bc046504 100644
--- a/include/common/tglobal.h
+++ b/include/common/tglobal.h
@@ -82,6 +82,10 @@ extern int32_t tsHeartbeatTimeout;
 // vnode
 extern int64_t tsVndCommitMaxIntervalMs;
 
+// snode
+extern char tsSnodeIp[];
+extern char tsCheckpointBackupDir[];
+
 // mnode
 extern int64_t tsMndSdbWriteDelta;
 extern int64_t tsMndLogRetention;
diff --git a/source/dnode/vnode/src/inc/vndCos.h b/include/util/cos.h
similarity index 92%
rename from source/dnode/vnode/src/inc/vndCos.h
rename to include/util/cos.h
index bb4d284f0e..4d676d9ca7 100644
--- a/source/dnode/vnode/src/inc/vndCos.h
+++ b/include/util/cos.h
@@ -16,7 +16,7 @@
 #ifndef _TD_VND_COS_H_
 #define _TD_VND_COS_H_
 
-#include "vnd.h"
+#include "os.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -24,9 +24,12 @@ extern "C" {
 
 #define S3_BLOCK_CACHE
 
+extern int8_t  tsS3StreamEnabled;
 extern int8_t  tsS3Enabled;
 extern int32_t tsS3BlockSize;
 extern int32_t tsS3BlockCacheSize;
+extern int32_t tsS3PageCacheSize;
+extern int32_t tsS3UploadDelaySec;
 
 int32_t s3Init();
 void    s3CleanUp();
diff --git a/include/util/rsync.h b/include/util/rsync.h
new file mode 100644
index 0000000000..6cce645d1e
--- /dev/null
+++ b/include/util/rsync.h
@@ -0,0 +1,24 @@
+//
+// Created by mingming wanng on 2023/11/2.
+//
+
+#ifndef TDENGINE_RSYNC_H
+#define TDENGINE_RSYNC_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "tarray.h"
+
+void stopRsync();
+void startRsync();
+int  uploadRsync(char* id, char* path);
+int  downloadRsync(char* id, char* path);
+int  deleteRsync(char* id);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // TDENGINE_RSYNC_H
diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index db809f1af2..980d5b9177 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -752,7 +752,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_FUNC_FUNTION_PARA_VALUE       TAOS_DEF_ERROR_CODE(0, 0x2803)
 #define TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION      TAOS_DEF_ERROR_CODE(0, 0x2804)
 #define TSDB_CODE_FUNC_DUP_TIMESTAMP            TAOS_DEF_ERROR_CODE(0, 0x2805)
-#define TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED      TAOS_DEF_ERROR_CODE(0, 0x2806)
+#define TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED_FORMAT_ERR TAOS_DEF_ERROR_CODE(0, 0x2806)
+#define TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED_TS_ERR TAOS_DEF_ERROR_CODE(0, 0x2807)
 
 //udf
 #define TSDB_CODE_UDF_STOPPING                  TAOS_DEF_ERROR_CODE(0, 0x2901)
diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index d74a704894..3593665de9 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -105,9 +105,9 @@ bool     tsEnableAuditCreateTable = true;
 
 // telem
 #ifdef TD_ENTERPRISE
-bool     tsEnableTelem = false;
+bool tsEnableTelem = false;
 #else
-bool     tsEnableTelem = true;
+bool    tsEnableTelem = true;
 #endif
 int32_t  tsTelemInterval = 43200;
 char     tsTelemServer[TSDB_FQDN_LEN] = "telemetry.tdengine.com";
@@ -115,9 +115,9 @@ uint16_t tsTelemPort = 80;
 char    *tsTelemUri = "/report";
 
 #ifdef TD_ENTERPRISE
-bool  tsEnableCrashReport = false;
+bool tsEnableCrashReport = false;
 #else
-bool  tsEnableCrashReport = true;
+bool    tsEnableCrashReport = true;
 #endif
 char *tsClientCrashReportUri = "/ccrashreport";
 char *tsSvrCrashReportUri = "/dcrashreport";
@@ -128,11 +128,19 @@ char tsSmlTsDefaultName[TSDB_COL_NAME_LEN] = "_ts";
 char tsSmlTagName[TSDB_COL_NAME_LEN] = "_tag_null";
 char tsSmlChildTableName[TSDB_TABLE_NAME_LEN] = "";  // user defined child table name can be specified in tag value.
 char tsSmlAutoChildTableNameDelimiter[TSDB_TABLE_NAME_LEN] = "";
-                                                                   // If set to empty system will generate table name using MD5 hash.
+// If set to empty system will generate table name using MD5 hash.
 // true means that the name and order of cols in each line are the same(only for influx protocol)
 // bool    tsSmlDataFormat = false;
 // int32_t tsSmlBatchSize = 10000;
 
+// checkpoint backup
+char tsSnodeIp[TSDB_FQDN_LEN] = {0};
+#ifdef WINDOWS
+char tsCheckpointBackupDir[PATH_MAX] = "C:\\TDengine\\data\\backup\\checkpoint\\";
+#else
+char tsCheckpointBackupDir[PATH_MAX] = "/var/lib/taos/backup/checkpoint/";
+#endif
+
 // tmq
 int32_t tmqMaxTopicNum = 20;
 // query
@@ -276,12 +284,15 @@ char   tsS3AccessKeySecret[TSDB_FQDN_LEN] = "<accesskeysecrect>";
 char   tsS3BucketName[TSDB_FQDN_LEN] = "<bucketname>";
 char   tsS3AppId[TSDB_FQDN_LEN] = "<appid>";
 int8_t tsS3Enabled = false;
+int8_t tsS3StreamEnabled = false;
 
 int8_t tsS3Https = true;
 char   tsS3Hostname[TSDB_FQDN_LEN] = "<hostname>";
 
-int32_t tsS3BlockSize = 4096;     // number of tsdb pages
-int32_t tsS3BlockCacheSize = 16;  // number of blocks
+int32_t tsS3BlockSize = -1;        // number of tsdb pages (4096)
+int32_t tsS3BlockCacheSize = 16;   // number of blocks
+int32_t tsS3PageCacheSize = 4096;  // number of pages
+int32_t tsS3UploadDelaySec = 60 * 60;
 
 #ifndef _STORAGE
 int32_t taosSetTfsCfg(SConfig *pCfg) {
@@ -337,9 +348,10 @@ int32_t taosSetS3Cfg(SConfig *pCfg) {
       tstrncpy(tsS3AppId, appid + 1, TSDB_FQDN_LEN);
     }
   }
-  if (tsS3BucketName[0] != '<' && tsDiskCfgNum > 1) {
+  if (tsS3BucketName[0] != '<') {
 #if defined(USE_COS) || defined(USE_S3)
-    tsS3Enabled = true;
+    if(tsDiskCfgNum > 1) tsS3Enabled = true;
+    tsS3StreamEnabled = true;
 #endif
   }
 
@@ -460,7 +472,8 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddBool(pCfg, "queryUseNodeAllocator", tsQueryUseNodeAllocator, CFG_SCOPE_CLIENT) != 0) return -1;
   if (cfgAddBool(pCfg, "keepColumnName", tsKeepColumnName, CFG_SCOPE_CLIENT) != 0) return -1;
   if (cfgAddString(pCfg, "smlChildTableName", tsSmlChildTableName, CFG_SCOPE_CLIENT) != 0) return -1;
-  if (cfgAddString(pCfg, "smlAutoChildTableNameDelimiter", tsSmlAutoChildTableNameDelimiter, CFG_SCOPE_CLIENT) != 0) return -1;
+  if (cfgAddString(pCfg, "smlAutoChildTableNameDelimiter", tsSmlAutoChildTableNameDelimiter, CFG_SCOPE_CLIENT) != 0)
+    return -1;
   if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, CFG_SCOPE_CLIENT) != 0) return -1;
   if (cfgAddString(pCfg, "smlTsDefaultName", tsSmlTsDefaultName, CFG_SCOPE_CLIENT) != 0) return -1;
   if (cfgAddBool(pCfg, "smlDot2Underline", tsSmlDot2Underline, CFG_SCOPE_CLIENT) != 0) return -1;
@@ -648,6 +661,9 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddString(pCfg, "telemetryServer", tsTelemServer, CFG_SCOPE_BOTH) != 0) return -1;
   if (cfgAddInt32(pCfg, "telemetryPort", tsTelemPort, 1, 65056, CFG_SCOPE_BOTH) != 0) return -1;
 
+  if (cfgAddString(pCfg, "snodeIp", tsSnodeIp, CFG_SCOPE_SERVER) != 0) return -1;
+  if (cfgAddString(pCfg, "checkpointBackupDir", tsCheckpointBackupDir, CFG_SCOPE_SERVER) != 0) return -1;
+
   if (cfgAddInt32(pCfg, "tmqMaxTopicNum", tmqMaxTopicNum, 1, 10000, CFG_SCOPE_SERVER) != 0) return -1;
 
   if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, CFG_SCOPE_SERVER) != 0) return -1;
@@ -675,7 +691,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt64(pCfg, "checkpointInterval", tsStreamCheckpointInterval, 60, 1200, CFG_SCOPE_SERVER) != 0) return -1;
   if (cfgAddFloat(pCfg, "streamSinkDataRate", tsSinkDataRate, 0.1, 5, CFG_SCOPE_SERVER) != 0) return -1;
 
-  if (cfgAddInt32(pCfg, "cacheLazyLoadThreshold", tsCacheLazyLoadThreshold, 0, 100000, CFG_SCOPE_SERVER) != 0) return -1;
+  if (cfgAddInt32(pCfg, "cacheLazyLoadThreshold", tsCacheLazyLoadThreshold, 0, 100000, CFG_SCOPE_SERVER) != 0)
+    return -1;
 
   if (cfgAddString(pCfg, "lossyColumns", tsLossyColumns, CFG_SCOPE_SERVER) != 0) return -1;
   if (cfgAddFloat(pCfg, "fPrecision", tsFPrecision, 0.0f, 100000.0f, CFG_SCOPE_SERVER) != 0) return -1;
@@ -693,8 +710,11 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddString(pCfg, "s3Accesskey", tsS3AccessKey, CFG_SCOPE_SERVER) != 0) return -1;
   if (cfgAddString(pCfg, "s3Endpoint", tsS3Endpoint, CFG_SCOPE_SERVER) != 0) return -1;
   if (cfgAddString(pCfg, "s3BucketName", tsS3BucketName, CFG_SCOPE_SERVER) != 0) return -1;
-  if (cfgAddInt32(pCfg, "s3BlockSize", tsS3BlockSize, 2048, 1024 * 1024, CFG_SCOPE_SERVER) != 0) return -1;
+  if (cfgAddInt32(pCfg, "s3BlockSize", tsS3BlockSize, -100, 1024 * 1024, CFG_SCOPE_SERVER) != 0) return -1;
   if (cfgAddInt32(pCfg, "s3BlockCacheSize", tsS3BlockCacheSize, 4, 1024 * 1024, CFG_SCOPE_SERVER) != 0) return -1;
+  if (cfgAddInt32(pCfg, "s3PageCacheSize", tsS3PageCacheSize, 4, 1024 * 1024 * 1024, CFG_SCOPE_SERVER) != 0) return -1;
+  if (cfgAddInt32(pCfg, "s3UploadDelaySec", tsS3UploadDelaySec, 60 * 10, 60 * 60 * 24 * 30, CFG_SCOPE_SERVER) != 0)
+    return -1;
 
   // min free disk space used to check if the disk is full [50MB, 1GB]
   if (cfgAddInt64(pCfg, "minDiskFreeSize", tsMinDiskFreeSize, TFS_MIN_DISK_FREE_SIZE, 1024 * 1024 * 1024,
@@ -953,7 +973,8 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
     return -1;
   }
 
-  tstrncpy(tsSmlAutoChildTableNameDelimiter, cfgGetItem(pCfg, "smlAutoChildTableNameDelimiter")->str, TSDB_TABLE_NAME_LEN);
+  tstrncpy(tsSmlAutoChildTableNameDelimiter, cfgGetItem(pCfg, "smlAutoChildTableNameDelimiter")->str,
+           TSDB_TABLE_NAME_LEN);
   tstrncpy(tsSmlChildTableName, cfgGetItem(pCfg, "smlChildTableName")->str, TSDB_TABLE_NAME_LEN);
   tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN);
   tstrncpy(tsSmlTsDefaultName, cfgGetItem(pCfg, "smlTsDefaultName")->str, TSDB_COL_NAME_LEN);
@@ -1067,6 +1088,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsTtlFlushThreshold = cfgGetItem(pCfg, "ttlFlushThreshold")->i32;
   tsTelemInterval = cfgGetItem(pCfg, "telemetryInterval")->i32;
   tstrncpy(tsTelemServer, cfgGetItem(pCfg, "telemetryServer")->str, TSDB_FQDN_LEN);
+  tstrncpy(tsSnodeIp, cfgGetItem(pCfg, "snodeIp")->str, TSDB_FQDN_LEN);
+  tstrncpy(tsCheckpointBackupDir, cfgGetItem(pCfg, "checkpointBackupDir")->str, PATH_MAX);
   tsTelemPort = (uint16_t)cfgGetItem(pCfg, "telemetryPort")->i32;
 
   tmqMaxTopicNum = cfgGetItem(pCfg, "tmqMaxTopicNum")->i32;
@@ -1125,6 +1148,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
 
   tsS3BlockSize = cfgGetItem(pCfg, "s3BlockSize")->i32;
   tsS3BlockCacheSize = cfgGetItem(pCfg, "s3BlockCacheSize")->i32;
+  tsS3PageCacheSize = cfgGetItem(pCfg, "s3PageCacheSize")->i32;
+  tsS3UploadDelaySec = cfgGetItem(pCfg, "s3UploadDelaySec")->i32;
 
   GRANT_CFG_GET;
   return 0;
@@ -1422,7 +1447,8 @@ int32_t taosApplyLocalCfg(SConfig *pCfg, char *name) {
       } else if (strcasecmp("smlChildTableName", name) == 0) {
         tstrncpy(tsSmlChildTableName, cfgGetItem(pCfg, "smlChildTableName")->str, TSDB_TABLE_NAME_LEN);
       } else if (strcasecmp("smlAutoChildTableNameDelimiter", name) == 0) {
-        tstrncpy(tsSmlAutoChildTableNameDelimiter, cfgGetItem(pCfg, "smlAutoChildTableNameDelimiter")->str, TSDB_TABLE_NAME_LEN);
+        tstrncpy(tsSmlAutoChildTableNameDelimiter, cfgGetItem(pCfg, "smlAutoChildTableNameDelimiter")->str,
+                 TSDB_TABLE_NAME_LEN);
       } else if (strcasecmp("smlTagName", name) == 0) {
         tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN);
         //      } else if (strcasecmp("smlDataFormat", name) == 0) {
@@ -1717,6 +1743,20 @@ void taosCfgDynamicOptions(const char *option, const char *value) {
     return;
   }
 
+  if (strcasecmp(option, "s3PageCacheSize") == 0) {
+    int32_t newS3PageCacheSize = atoi(value);
+    uInfo("s3PageCacheSize set from %d to %d", tsS3PageCacheSize, newS3PageCacheSize);
+    tsS3PageCacheSize = newS3PageCacheSize;
+    return;
+  }
+
+  if (strcasecmp(option, "s3UploadDelaySec") == 0) {
+    int32_t newS3UploadDelaysec = atoi(value);
+    uInfo("s3UploadDelaySec set from %d to %d", tsS3UploadDelaySec, newS3UploadDelaysec);
+    tsS3UploadDelaySec = newS3UploadDelaysec;
+    return;
+  }
+
   if (strcasecmp(option, "ttlPushInterval") == 0) {
     int32_t newTtlPushInterval = atoi(value);
     uInfo("ttlPushInterval set from %d to %d", tsTtlPushIntervalSec, newTtlPushInterval);
diff --git a/source/common/src/ttime.c b/source/common/src/ttime.c
index 6b5bb8680e..4b0848e5e9 100644
--- a/source/common/src/ttime.c
+++ b/source/common/src/ttime.c
@@ -1320,7 +1320,7 @@ static void tm2char(const SArray* formats, const struct STm* tm, char* s, int32_
         s += 4;
         break;
       case TSFKW_DDD:
-        sprintf(s, "%d", tm->tm.tm_yday);
+        sprintf(s, "%03d", tm->tm.tm_yday + 1);
         s += strlen(s);
         break;
       case TSFKW_DD:
diff --git a/source/common/test/commonTests.cpp b/source/common/test/commonTests.cpp
index dc320ebcb2..c65d8761b7 100644
--- a/source/common/test/commonTests.cpp
+++ b/source/common/test/commonTests.cpp
@@ -344,7 +344,7 @@ TEST(timeTest, ts2char) {
                "day-\"日\"",
                TSDB_TIME_PRECISION_MILLI,
                "2023-023-23-3-2023-023-23-3-年-OCTOBER  -OCT-October  -Oct-october  "
-               "-oct-月-285-13-6-285-13-6-FRIDAY   -Friday   -friday   -日");
+               "-oct-月-286-13-6-286-13-6-FRIDAY   -Friday   -friday   -日");
 #endif
   ts = 1697182085123L; // Friday, October 13, 2023 3:28:05.123 PM GMT+08:00
   test_ts2char(ts, "HH24:hh24:HH12:hh12:HH:hh:MI:mi:SS:ss:MS:ms:US:us:NS:ns:PM:AM:pm:am", TSDB_TIME_PRECISION_MILLI,
diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c
index 7de6e860f9..c32625009e 100644
--- a/source/dnode/mnode/impl/src/mndDnode.c
+++ b/source/dnode/mnode/impl/src/mndDnode.c
@@ -721,7 +721,7 @@ static int32_t mndProcessNotifyReq(SRpcMsg *pReq) {
       mndReleaseVgroup(pMnode, pVgroup);
     }
   }
-    mndUpdClusterInfo(pReq);
+  mndUpdClusterInfo(pReq);
 _OVER:
   tFreeSNotifyReq(&notifyReq);
   return code;
@@ -1228,6 +1228,70 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
 
     strcpy(dcfgReq.config, "monitor");
     snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
+  } else if (strncasecmp(cfgReq.config, "s3blocksize", 11) == 0) {
+    int32_t optLen = strlen("s3blocksize");
+    int32_t flag = -1;
+    int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag);
+    if (code < 0) return code;
+
+    if (flag > 1024 * 1024) {
+      mError("dnode:%d, failed to config s3blocksize since value:%d. Valid range: [4, 1024 * 1024]", cfgReq.dnodeId,
+             flag);
+      terrno = TSDB_CODE_INVALID_CFG;
+      tFreeSMCfgDnodeReq(&cfgReq);
+      return -1;
+    }
+
+    strcpy(dcfgReq.config, "s3blocksize");
+    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
+  } else if (strncasecmp(cfgReq.config, "s3blockcachesize", 16) == 0) {
+    int32_t optLen = strlen("s3blockcachesize");
+    int32_t flag = -1;
+    int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag);
+    if (code < 0) return code;
+
+    if (flag < 4 || flag > 1024 * 1024) {
+      mError("dnode:%d, failed to config s3BlockCacheSize since value:%d. Valid range: [4, 1024 * 1024]",
+             cfgReq.dnodeId, flag);
+      terrno = TSDB_CODE_INVALID_CFG;
+      tFreeSMCfgDnodeReq(&cfgReq);
+      return -1;
+    }
+
+    strcpy(dcfgReq.config, "s3blockcachesize");
+    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
+  } else if (strncasecmp(cfgReq.config, "s3pagecachesize", 16) == 0) {
+    int32_t optLen = strlen("s3pagecachesize");
+    int32_t flag = -1;
+    int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag);
+    if (code < 0) return code;
+
+    if (flag < 4 || flag > 1024 * 1024 * 1024) {
+      mError("dnode:%d, failed to config s3PageCacheSize since value:%d. Valid range: [4, 1024 * 1024]", cfgReq.dnodeId,
+             flag);
+      terrno = TSDB_CODE_INVALID_CFG;
+      tFreeSMCfgDnodeReq(&cfgReq);
+      return -1;
+    }
+
+    strcpy(dcfgReq.config, "s3pagecachesize");
+    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
+  } else if (strncasecmp(cfgReq.config, "s3uploaddelaysec", 16) == 0) {
+    int32_t optLen = strlen("s3uploaddelaysec");
+    int32_t flag = -1;
+    int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag);
+    if (code < 0) return code;
+
+    if (flag < 600 || flag > 60 * 60 * 24 * 30) {
+      mError("dnode:%d, failed to config s3UploadDelaySec since value:%d. Valid range: [600, 60 * 60 * 24 * 30]",
+             cfgReq.dnodeId, flag);
+      terrno = TSDB_CODE_INVALID_CFG;
+      tFreeSMCfgDnodeReq(&cfgReq);
+      return -1;
+    }
+
+    strcpy(dcfgReq.config, "s3uploaddelaysec");
+    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
   } else if (strncasecmp(cfgReq.config, "ttlpushinterval", 14) == 0) {
     int32_t optLen = strlen("ttlpushinterval");
     int32_t flag = -1;
diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c
index 0cc283e1a9..ff147fa37b 100644
--- a/source/dnode/snode/src/snode.c
+++ b/source/dnode/snode/src/snode.c
@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "rsync.h"
 #include "executor.h"
 #include "sndInt.h"
 #include "tstream.h"
@@ -126,6 +127,9 @@ SSnode *sndOpen(const char *path, const SSnodeOpt *pOption) {
     goto FAIL;
   }
 
+  stopRsync();
+  startRsync();
+
   // todo fix it: send msg to mnode to rollback to an existed checkpoint
   streamMetaInitForSnode(pSnode->pMeta);
   return pSnode;
diff --git a/source/dnode/vnode/CMakeLists.txt b/source/dnode/vnode/CMakeLists.txt
index dcc9f9a115..27cb0f93f7 100644
--- a/source/dnode/vnode/CMakeLists.txt
+++ b/source/dnode/vnode/CMakeLists.txt
@@ -8,7 +8,6 @@ set(
   "src/vnd/vnodeCommit.c"
   "src/vnd/vnodeQuery.c"
   "src/vnd/vnodeModule.c"
-  "src/vnd/vnodeCos.c"
   "src/vnd/vnodeSvr.c"
   "src/vnd/vnodeSync.c"
   "src/vnd/vnodeSnapshot.c"
diff --git a/source/dnode/vnode/src/inc/tsdb.h b/source/dnode/vnode/src/inc/tsdb.h
index 1efcd9417d..ec69ae5ca7 100644
--- a/source/dnode/vnode/src/inc/tsdb.h
+++ b/source/dnode/vnode/src/inc/tsdb.h
@@ -382,6 +382,8 @@ struct STsdb {
   TdThreadMutex        biMutex;
   SLRUCache           *bCache;
   TdThreadMutex        bMutex;
+  SLRUCache           *pgCache;
+  TdThreadMutex        pgMutex;
   struct STFileSystem *pFS;  // new
   SRocksCache          rCache;
 };
@@ -909,7 +911,9 @@ int32_t tsdbCacheGetBlockIdx(SLRUCache *pCache, SDataFReader *pFileReader, LRUHa
 int32_t tsdbBICacheRelease(SLRUCache *pCache, LRUHandle *h);
 
 int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle);
-int32_t tsdbBCacheRelease(SLRUCache *pCache, LRUHandle *h);
+int32_t tsdbCacheGetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, LRUHandle **handle);
+int32_t tsdbCacheSetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, uint8_t *pPage);
+int32_t tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h);
 
 int32_t tsdbCacheDeleteLastrow(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey);
 int32_t tsdbCacheDeleteLast(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey);
diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c
index ca3fb7027f..aa83478105 100644
--- a/source/dnode/vnode/src/tsdb/tsdbCache.c
+++ b/source/dnode/vnode/src/tsdb/tsdbCache.c
@@ -12,11 +12,11 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
+#include "cos.h"
 #include "tsdb.h"
 #include "tsdbDataFileRW.h"
 #include "tsdbReadUtil.h"
 #include "vnd.h"
-#include "vndCos.h"
 
 #define ROCKS_BATCH_SIZE (4096)
 
@@ -87,6 +87,41 @@ static void tsdbCloseBCache(STsdb *pTsdb) {
   }
 }
 
+static int32_t tsdbOpenPgCache(STsdb *pTsdb) {
+  int32_t code = 0;
+  // SLRUCache *pCache = taosLRUCacheInit(10 * 1024 * 1024, 0, .5);
+  int32_t szPage = pTsdb->pVnode->config.tsdbPageSize;
+
+  SLRUCache *pCache = taosLRUCacheInit((int64_t)tsS3PageCacheSize * szPage, 0, .5);
+  if (pCache == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _err;
+  }
+
+  taosLRUCacheSetStrictCapacity(pCache, false);
+
+  taosThreadMutexInit(&pTsdb->pgMutex, NULL);
+
+_err:
+  pTsdb->pgCache = pCache;
+  return code;
+}
+
+static void tsdbClosePgCache(STsdb *pTsdb) {
+  SLRUCache *pCache = pTsdb->pgCache;
+  if (pCache) {
+    int32_t elems = taosLRUCacheGetElems(pCache);
+    tsdbTrace("vgId:%d, elems: %d", TD_VID(pTsdb->pVnode), elems);
+    taosLRUCacheEraseUnrefEntries(pCache);
+    elems = taosLRUCacheGetElems(pCache);
+    tsdbTrace("vgId:%d, elems: %d", TD_VID(pTsdb->pVnode), elems);
+
+    taosLRUCacheCleanup(pCache);
+
+    taosThreadMutexDestroy(&pTsdb->bMutex);
+  }
+}
+
 #define ROCKS_KEY_LEN (sizeof(tb_uid_t) + sizeof(int16_t) + sizeof(int8_t))
 
 typedef struct {
@@ -1191,6 +1226,12 @@ int32_t tsdbOpenCache(STsdb *pTsdb) {
     goto _err;
   }
 
+  code = tsdbOpenPgCache(pTsdb);
+  if (code != TSDB_CODE_SUCCESS) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _err;
+  }
+
   code = tsdbOpenRocksCache(pTsdb);
   if (code != TSDB_CODE_SUCCESS) {
     code = TSDB_CODE_OUT_OF_MEMORY;
@@ -1221,6 +1262,7 @@ void tsdbCloseCache(STsdb *pTsdb) {
 
   tsdbCloseBICache(pTsdb);
   tsdbCloseBCache(pTsdb);
+  tsdbClosePgCache(pTsdb);
   tsdbCloseRocksCache(pTsdb);
 }
 
@@ -3057,7 +3099,6 @@ static int32_t tsdbCacheLoadBlockS3(STsdbFD *pFD, uint8_t **ppBlock) {
   }
   */
   int64_t block_offset = (pFD->blkno - 1) * tsS3BlockSize * pFD->szPage;
-  // int64_t size = 4096;
   code = s3GetObjectBlock(pFD->objName, block_offset, tsS3BlockSize * pFD->szPage, ppBlock);
   if (code != TSDB_CODE_SUCCESS) {
     // taosMemoryFree(pBlock);
@@ -3123,10 +3164,42 @@ int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle)
   return code;
 }
 
-int32_t tsdbBCacheRelease(SLRUCache *pCache, LRUHandle *h) {
+int32_t tsdbCacheGetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, LRUHandle **handle) {
   int32_t code = 0;
+  char    key[128] = {0};
+  int     keyLen = 0;
 
-  taosLRUCacheRelease(pCache, h, false);
+  getBCacheKey(pFD->fid, pFD->cid, pgno, key, &keyLen);
+  *handle = taosLRUCacheLookup(pCache, key, keyLen);
+
+  return code;
+}
+
+int32_t tsdbCacheSetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, uint8_t *pPage) {
+  int32_t    code = 0;
+  char       key[128] = {0};
+  int        keyLen = 0;
+  LRUHandle *handle = NULL;
+
+  getBCacheKey(pFD->fid, pFD->cid, pgno, key, &keyLen);
+  taosThreadMutexLock(&pFD->pTsdb->pgMutex);
+  handle = taosLRUCacheLookup(pFD->pTsdb->pgCache, key, keyLen);
+  if (!handle) {
+    size_t              charge = pFD->szPage;
+    _taos_lru_deleter_t deleter = deleteBCache;
+    uint8_t            *pPg = taosMemoryMalloc(charge);
+    memcpy(pPg, pPage, charge);
+
+    LRUStatus status =
+        taosLRUCacheInsert(pCache, key, keyLen, pPg, charge, deleter, &handle, TAOS_LRU_PRIORITY_LOW, NULL);
+    if (status != TAOS_LRU_STATUS_OK) {
+      // ignore cache updating if not ok
+      // code = TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+  taosThreadMutexUnlock(&pFD->pTsdb->pgMutex);
+
+  tsdbCacheRelease(pFD->pTsdb->pgCache, handle);
 
   return code;
 }
diff --git a/source/dnode/vnode/src/tsdb/tsdbCommit2.c b/source/dnode/vnode/src/tsdb/tsdbCommit2.c
index 79ecdab15c..46e3aff0d4 100644
--- a/source/dnode/vnode/src/tsdb/tsdbCommit2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbCommit2.c
@@ -46,6 +46,7 @@ typedef struct {
     STFileSet *fset;
     TABLEID    tbid[1];
     bool       hasTSData;
+    bool       skipTsRow;
   } ctx[1];
 
   // reader
@@ -127,18 +128,18 @@ static int32_t tsdbCommitTSData(SCommitter2 *committer) {
         continue;
       }
     }
-
+    /*
     extern int8_t tsS3Enabled;
 
     int32_t nlevel = tfsGetLevel(committer->tsdb->pVnode->pTfs);
-    bool    skipRow = false;
+    committer->ctx->skipTsRow = false;
     if (tsS3Enabled && nlevel > 1 && committer->ctx->did.level == nlevel - 1) {
-      skipRow = true;
+      committer->ctx->skipTsRow = true;
     }
-
+    */
     int64_t ts = TSDBROW_TS(&row->row);
 
-    if (skipRow && ts <= committer->ctx->maxKey) {
+    if (committer->ctx->skipTsRow && ts <= committer->ctx->maxKey) {
       ts = committer->ctx->maxKey + 1;
     }
 
@@ -402,6 +403,32 @@ static int32_t tsdbCommitFileSetBegin(SCommitter2 *committer) {
   // reset nextKey
   committer->ctx->nextKey = TSKEY_MAX;
 
+  committer->ctx->skipTsRow = false;
+
+  extern int8_t  tsS3Enabled;
+  extern int32_t tsS3UploadDelaySec;
+  long           s3Size(const char *object_name);
+  int32_t        nlevel = tfsGetLevel(committer->tsdb->pVnode->pTfs);
+  committer->ctx->skipTsRow = false;
+  if (tsS3Enabled && nlevel > 1 && committer->ctx->fset) {
+    STFileObj *fobj = committer->ctx->fset->farr[TSDB_FTYPE_DATA];
+    if (fobj && fobj->f->did.level == nlevel - 1) {
+      // if exists on s3 or local mtime < committer->ctx->now - tsS3UploadDelay
+      const char *object_name = taosDirEntryBaseName((char *)fobj->fname);
+
+      if (taosCheckExistFile(fobj->fname)) {
+        int32_t mtime = 0;
+        taosStatFile(fobj->fname, NULL, &mtime, NULL);
+        if (mtime < committer->ctx->now - tsS3UploadDelaySec) {
+          committer->ctx->skipTsRow = true;
+        }
+      } else if (s3Size(object_name) > 0) {
+        committer->ctx->skipTsRow = true;
+      }
+    }
+    // new fset can be written with ts data
+  }
+
 _exit:
   if (code) {
     TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
diff --git a/source/dnode/vnode/src/tsdb/tsdbFS2.c b/source/dnode/vnode/src/tsdb/tsdbFS2.c
index 38d221d978..348397272d 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFS2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbFS2.c
@@ -14,9 +14,9 @@
  */
 
 #include "tsdbFS2.h"
+#include "cos.h"
 #include "tsdbUpgrade.h"
 #include "vnd.h"
-#include "vndCos.h"
 
 #define BLOCK_COMMIT_FACTOR 3
 
diff --git a/source/dnode/vnode/src/tsdb/tsdbFile2.c b/source/dnode/vnode/src/tsdb/tsdbFile2.c
index 963c5bad34..9edb03d35b 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFile2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbFile2.c
@@ -303,6 +303,7 @@ bool tsdbIsSameTFile(const STFile *f1, const STFile *f2) {
   if (f1->did.id != f2->did.id) return false;
   if (f1->fid != f2->fid) return false;
   if (f1->cid != f2->cid) return false;
+  if (f1->s3flag != f2->s3flag) return false;
   return true;
 }
 
diff --git a/source/dnode/vnode/src/tsdb/tsdbFile2.h b/source/dnode/vnode/src/tsdb/tsdbFile2.h
index 33d8ac5478..9da198c1f0 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFile2.h
+++ b/source/dnode/vnode/src/tsdb/tsdbFile2.h
@@ -58,6 +58,7 @@ int32_t tsdbTFileObjCmpr(const STFileObj **fobj1, const STFileObj **fobj2);
 struct STFile {
   tsdb_ftype_t type;
   SDiskID      did;  // disk id
+  int32_t      s3flag;
   int32_t      fid;  // file id
   int64_t      cid;  // commit id
   int64_t      size;
diff --git a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
index f3bcfef703..151c6419cc 100644
--- a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
+++ b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "cos.h"
 #include "tsdb.h"
-#include "vndCos.h"
 
 static int32_t tsdbOpenFileImpl(STsdbFD *pFD) {
   int32_t     code = 0;
@@ -131,6 +131,7 @@ static int32_t tsdbWriteFilePage(STsdbFD *pFD) {
   }
 
   if (pFD->s3File) {
+    tsdbWarn("%s file: %s", __func__, pFD->path);
     return code;
   }
   if (pFD->pgno > 0) {
@@ -177,7 +178,7 @@ static int32_t tsdbReadFilePage(STsdbFD *pFD, int64_t pgno) {
     pFD->blkno = (pgno + tsS3BlockSize - 1) / tsS3BlockSize;
     code = tsdbCacheGetBlockS3(pFD->pTsdb->bCache, pFD, &handle);
     if (code != TSDB_CODE_SUCCESS || handle == NULL) {
-      tsdbBCacheRelease(pFD->pTsdb->bCache, handle);
+      tsdbCacheRelease(pFD->pTsdb->bCache, handle);
       if (code == TSDB_CODE_SUCCESS && !handle) {
         code = TSDB_CODE_OUT_OF_MEMORY;
       }
@@ -189,7 +190,7 @@ static int32_t tsdbReadFilePage(STsdbFD *pFD, int64_t pgno) {
     int64_t blk_offset = (pFD->blkno - 1) * tsS3BlockSize * pFD->szPage;
     memcpy(pFD->pBuf, pBlock + (offset - blk_offset), pFD->szPage);
 
-    tsdbBCacheRelease(pFD->pTsdb->bCache, handle);
+    tsdbCacheRelease(pFD->pTsdb->bCache, handle);
   } else {
     // seek
     int64_t n = taosLSeekFile(pFD->pFD, offset, SEEK_SET);
@@ -253,7 +254,7 @@ _exit:
   return code;
 }
 
-int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) {
+static int32_t tsdbReadFileImp(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) {
   int32_t code = 0;
   int64_t n = 0;
   int64_t fOffset = LOGIC_TO_FILE_OFFSET(offset, pFD->szPage);
@@ -282,10 +283,122 @@ _exit:
   return code;
 }
 
+static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) {
+  int32_t code = 0;
+  int64_t n = 0;
+  int32_t szPgCont = PAGE_CONTENT_SIZE(pFD->szPage);
+  int64_t fOffset = LOGIC_TO_FILE_OFFSET(offset, pFD->szPage);
+  int64_t pgno = OFFSET_PGNO(fOffset, pFD->szPage);
+  int64_t bOffset = fOffset % pFD->szPage;
+
+  ASSERT(bOffset < szPgCont);
+
+  // 1, find pgnoStart & pgnoEnd to fetch from s3, if all pgs are local, no need to fetch
+  // 2, fetch pgnoStart ~ pgnoEnd from s3
+  // 3, store pgs to pcache & last pg to pFD->pBuf
+  // 4, deliver pgs to [pBuf, pBuf + size)
+
+  while (n < size) {
+    if (pFD->pgno != pgno) {
+      LRUHandle *handle = NULL;
+      code = tsdbCacheGetPageS3(pFD->pTsdb->pgCache, pFD, pgno, &handle);
+      if (code != TSDB_CODE_SUCCESS) {
+        if (handle) {
+          tsdbCacheRelease(pFD->pTsdb->pgCache, handle);
+        }
+        goto _exit;
+      }
+
+      if (!handle) {
+        break;
+      }
+
+      uint8_t *pPage = (uint8_t *)taosLRUCacheValue(pFD->pTsdb->pgCache, handle);
+      memcpy(pFD->pBuf, pPage, pFD->szPage);
+      tsdbCacheRelease(pFD->pTsdb->pgCache, handle);
+
+      // check
+      if (pgno > 1 && !taosCheckChecksumWhole(pFD->pBuf, pFD->szPage)) {
+        code = TSDB_CODE_FILE_CORRUPTED;
+        goto _exit;
+      }
+
+      pFD->pgno = pgno;
+    }
+
+    int64_t nRead = TMIN(szPgCont - bOffset, size - n);
+    memcpy(pBuf + n, pFD->pBuf + bOffset, nRead);
+
+    n += nRead;
+    pgno++;
+    bOffset = 0;
+  }
+
+  if (n < size) {
+    // 2, retrieve pgs from s3
+    uint8_t *pBlock = NULL;
+    int64_t  retrieve_offset = PAGE_OFFSET(pgno, pFD->szPage);
+    int64_t  pgnoEnd = pgno - 1 + (size - n + szPgCont - 1) / szPgCont;
+    int64_t  retrieve_size = (pgnoEnd - pgno + 1) * pFD->szPage;
+    code = s3GetObjectBlock(pFD->objName, retrieve_offset, retrieve_size, &pBlock);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _exit;
+    }
+
+    // 3, Store Pages in Cache
+    int nPage = pgnoEnd - pgno + 1;
+    for (int i = 0; i < nPage; ++i) {
+      tsdbCacheSetPageS3(pFD->pTsdb->pgCache, pFD, pgno, pBlock + i * pFD->szPage);
+
+      memcpy(pFD->pBuf, pBlock + i * pFD->szPage, pFD->szPage);
+
+      // check
+      if (pgno > 1 && !taosCheckChecksumWhole(pFD->pBuf, pFD->szPage)) {
+        code = TSDB_CODE_FILE_CORRUPTED;
+        goto _exit;
+      }
+
+      pFD->pgno = pgno;
+
+      int64_t nRead = TMIN(szPgCont - bOffset, size - n);
+      memcpy(pBuf + n, pFD->pBuf + bOffset, nRead);
+
+      n += nRead;
+      pgno++;
+      bOffset = 0;
+    }
+
+    taosMemoryFree(pBlock);
+  }
+
+_exit:
+  return code;
+}
+
+int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) {
+  int32_t code = 0;
+  if (!pFD->pFD) {
+    code = tsdbOpenFileImpl(pFD);
+    if (code) {
+      goto _exit;
+    }
+  }
+
+  if (pFD->s3File && tsS3BlockSize < 0) {
+    return tsdbReadFileS3(pFD, offset, pBuf, size);
+  } else {
+    return tsdbReadFileImp(pFD, offset, pBuf, size);
+  }
+
+_exit:
+  return code;
+}
+
 int32_t tsdbFsyncFile(STsdbFD *pFD) {
   int32_t code = 0;
 
   if (pFD->s3File) {
+    tsdbWarn("%s file: %s", __func__, pFD->path);
     return code;
   }
   code = tsdbWriteFilePage(pFD);
diff --git a/source/dnode/vnode/src/tsdb/tsdbRetention.c b/source/dnode/vnode/src/tsdb/tsdbRetention.c
index 6c41b46c73..1908f16529 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRetention.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRetention.c
@@ -15,7 +15,8 @@
 
 #include "tsdb.h"
 #include "tsdbFS2.h"
-#include "vndCos.h"
+#include "cos.h"
+#include "vnd.h"
 
 typedef struct {
   STsdb  *tsdb;
@@ -206,6 +207,8 @@ static int32_t tsdbMigrateDataFileS3(SRTNer *rtner, const STFileObj *fobj, const
           },
   };
 
+  op.nf.s3flag = true;
+
   code = TARRAY2_APPEND(rtner->fopArr, op);
   TSDB_CHECK_CODE(code, lino, _exit);
 
@@ -322,27 +325,40 @@ static int32_t tsdbDoRetentionOnFileSet(SRTNer *rtner, STFileSet *fset) {
     for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX && (fobj = fset->farr[ftype], 1); ++ftype) {
       if (fobj == NULL) continue;
 
-      if (fobj->f->did.level == did.level) continue;
-
       int32_t nlevel = tfsGetLevel(rtner->tsdb->pVnode->pTfs);
-      if (tsS3Enabled && nlevel > 1 && TSDB_FTYPE_DATA == ftype && did.level == nlevel - 1) {
-        code = tsdbMigrateDataFileS3(rtner, fobj, &did);
-        TSDB_CHECK_CODE(code, lino, _exit);
-      } else {
-        if (tsS3Enabled) {
-          int64_t fsize = 0;
-          if (taosStatFile(fobj->fname, &fsize, NULL, NULL) < 0) {
-            code = TAOS_SYSTEM_ERROR(terrno);
-            tsdbError("vgId:%d %s failed since file:%s stat failed, reason:%s", TD_VID(rtner->tsdb->pVnode), __func__,
-                      fobj->fname, tstrerror(code));
+
+      if (fobj->f->did.level == did.level) {
+        if (tsS3Enabled && nlevel > 1 && TSDB_FTYPE_DATA == ftype && did.level == nlevel - 1 &&
+            taosCheckExistFile(fobj->fname)) {
+          int32_t mtime = 0;
+          taosStatFile(fobj->fname, NULL, &mtime, NULL);
+          if (mtime < rtner->now - tsS3UploadDelaySec) {
+            code = tsdbMigrateDataFileS3(rtner, fobj, &did);
             TSDB_CHECK_CODE(code, lino, _exit);
           }
-          s3EvictCache(fobj->fname, fsize * 2);
         }
 
-        code = tsdbDoMigrateFileObj(rtner, fobj, &did);
-        TSDB_CHECK_CODE(code, lino, _exit);
+        continue;
       }
+      /*
+    if (tsS3Enabled && nlevel > 1 && TSDB_FTYPE_DATA == ftype && did.level == nlevel - 1) {
+      code = tsdbMigrateDataFileS3(rtner, fobj, &did);
+      TSDB_CHECK_CODE(code, lino, _exit);
+    } else {
+
+      if (tsS3Enabled) {
+        int64_t fsize = 0;
+        if (taosStatFile(fobj->fname, &fsize, NULL, NULL) < 0) {
+          code = TAOS_SYSTEM_ERROR(terrno);
+          tsdbError("vgId:%d %s failed since file:%s stat failed, reason:%s", TD_VID(rtner->tsdb->pVnode),
+      __func__, fobj->fname, tstrerror(code)); TSDB_CHECK_CODE(code, lino, _exit);
+        }
+        s3EvictCache(fobj->fname, fsize * 2);
+      }
+      */
+      code = tsdbDoMigrateFileObj(rtner, fobj, &did);
+      TSDB_CHECK_CODE(code, lino, _exit);
+      //}
     }
 
     // stt
diff --git a/source/dnode/vnode/src/tsdb/tsdbWrite.c b/source/dnode/vnode/src/tsdb/tsdbWrite.c
index 5949b103d5..43025cbe91 100644
--- a/source/dnode/vnode/src/tsdb/tsdbWrite.c
+++ b/source/dnode/vnode/src/tsdb/tsdbWrite.c
@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "cos.h"
 #include "tsdb.h"
-#include "vndCos.h"
 
 /**
  * @brief max key by precision
@@ -80,8 +80,8 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq2 *pMsg) {
   TSKEY         minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep2;
   TSKEY         maxKey = tsMaxKeyByPrecision[pCfg->precision];
   int32_t       size = taosArrayGetSize(pMsg->aSubmitTbData);
-  int32_t       nlevel = tfsGetLevel(pTsdb->pVnode->pTfs);
-
+  /*
+  int32_t nlevel = tfsGetLevel(pTsdb->pVnode->pTfs);
   if (nlevel > 1 && tsS3Enabled) {
     if (nlevel == 3) {
       minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep1;
@@ -89,7 +89,7 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq2 *pMsg) {
       minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep0;
     }
   }
-
+  */
   for (int32_t i = 0; i < size; ++i) {
     SSubmitTbData *pData = TARRAY_GET_ELEM(pMsg->aSubmitTbData, i);
     if (pData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
diff --git a/source/dnode/vnode/src/vnd/vnodeModule.c b/source/dnode/vnode/src/vnd/vnodeModule.c
index 6ccce5c9d7..df08fb8a2b 100644
--- a/source/dnode/vnode/src/vnd/vnodeModule.c
+++ b/source/dnode/vnode/src/vnd/vnodeModule.c
@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "cos.h"
 #include "vnd.h"
-#include "vndCos.h"
 
 typedef struct SVnodeTask SVnodeTask;
 struct SVnodeTask {
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 3bdecee79b..ff79e83d72 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -13,10 +13,10 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "cos.h"
 #include "sync.h"
 #include "tsdb.h"
 #include "vnd.h"
-#include "vndCos.h"
 
 int32_t vnodeGetPrimaryDir(const char *relPath, int32_t diskPrimary, STfs *pTfs, char *buf, size_t bufLen) {
   if (pTfs) {
diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c
index 12883a34af..3f58de773a 100644
--- a/source/dnode/vnode/src/vnd/vnodeSvr.c
+++ b/source/dnode/vnode/src/vnd/vnodeSvr.c
@@ -13,14 +13,14 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "audit.h"
 #include "tencode.h"
 #include "tmsg.h"
+#include "tstrbuild.h"
 #include "vnd.h"
-#include "vndCos.h"
+#include "cos.h"
 #include "vnode.h"
 #include "vnodeInt.h"
-#include "audit.h"
-#include "tstrbuild.h"
 
 static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
 static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
@@ -178,7 +178,7 @@ static int32_t vnodePreProcessDropTtlMsg(SVnode *pVnode, SRpcMsg *pMsg) {
     ttlReq.pTbUids = tbUids;
   }
 
-  { // prepare new content
+  {  // prepare new content
     int32_t reqLenNew = tSerializeSVDropTtlTableReq(NULL, 0, &ttlReq);
     int32_t contLenNew = reqLenNew + sizeof(SMsgHead);
 
@@ -263,8 +263,9 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
     now *= 1000000;
   }
 
-  int32_t nlevel = tfsGetLevel(pVnode->pTfs);
   int32_t keep = pVnode->config.tsdbCfg.keep2;
+  /*
+  int32_t nlevel = tfsGetLevel(pVnode->pTfs);
   if (nlevel > 1 && tsS3Enabled) {
     if (nlevel == 3) {
       keep = pVnode->config.tsdbCfg.keep1;
@@ -272,6 +273,7 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
       keep = pVnode->config.tsdbCfg.keep0;
     }
   }
+  */
 
   TSKEY minKey = now - tsTickPerMin[pVnode->config.tsdbCfg.precision] * keep;
   TSKEY maxKey = tsMaxKeyByPrecision[pVnode->config.tsdbCfg.precision];
@@ -904,7 +906,7 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
 
   rsp.pArray = taosArrayInit(req.nReqs, sizeof(cRsp));
   tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
-  tbNames = taosArrayInit(req.nReqs, sizeof(char*));
+  tbNames = taosArrayInit(req.nReqs, sizeof(char *));
   if (rsp.pArray == NULL || tbUids == NULL || tbNames == NULL) {
     rcode = -1;
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -950,8 +952,8 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
 
     taosArrayPush(rsp.pArray, &cRsp);
 
-    if(tsEnableAuditCreateTable){
-      char* str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
+    if (tsEnableAuditCreateTable) {
+      char *str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
       strcpy(str, pCreateReq->name);
       taosArrayPush(tbNames, &str);
     }
@@ -976,24 +978,24 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
   tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
   tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
 
-  if(tsEnableAuditCreateTable){
+  if (tsEnableAuditCreateTable) {
     int64_t clusterId = pVnode->config.syncCfg.nodeInfo[0].clusterId;
 
     SName name = {0};
     tNameFromString(&name, pVnode->config.dbname, T_NAME_ACCT | T_NAME_DB);
 
     SStringBuilder sb = {0};
-    for(int32_t iReq = 0; iReq < req.nReqs; iReq++){
-      char** key = (char**)taosArrayGet(tbNames, iReq);
+    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
+      char **key = (char **)taosArrayGet(tbNames, iReq);
       taosStringBuilderAppendStringLen(&sb, *key, strlen(*key));
-      if(iReq < req.nReqs - 1){
+      if (iReq < req.nReqs - 1) {
         taosStringBuilderAppendChar(&sb, ',');
       }
       taosMemoryFreeClear(*key);
     }
 
-    size_t    len = 0;
-    char*     keyJoined = taosStringBuilderGetResult(&sb, &len);
+    size_t len = 0;
+    char  *keyJoined = taosStringBuilderGetResult(&sb, &len);
 
     auditRecord(NULL, clusterId, "createTable", name.dbname, "", keyJoined, len);
 
@@ -1005,7 +1007,7 @@ _exit:
     pCreateReq = req.pReqs + iReq;
     taosMemoryFree(pCreateReq->sql);
     taosMemoryFree(pCreateReq->comment);
-    taosArrayDestroy(pCreateReq->ctb.tagName);    
+    taosArrayDestroy(pCreateReq->ctb.tagName);
   }
   taosArrayDestroyEx(rsp.pArray, tFreeSVCreateTbRsp);
   taosArrayDestroy(tbUids);
@@ -1164,7 +1166,7 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, in
   // process req
   tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
   rsp.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbRsp));
-  tbNames = taosArrayInit(req.nReqs, sizeof(char*));
+  tbNames = taosArrayInit(req.nReqs, sizeof(char *));
   if (tbUids == NULL || rsp.pArray == NULL || tbNames == NULL) goto _exit;
 
   for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
@@ -1186,9 +1188,9 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, in
     }
 
     taosArrayPush(rsp.pArray, &dropTbRsp);
-    
-    if(tsEnableAuditCreateTable){
-      char* str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
+
+    if (tsEnableAuditCreateTable) {
+      char *str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
       strcpy(str, pDropTbReq->name);
       taosArrayPush(tbNames, &str);
     }
@@ -1197,30 +1199,30 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, in
   tqUpdateTbUidList(pVnode->pTq, tbUids, false);
   tdUpdateTbUidList(pVnode->pSma, pStore, false);
 
-  if(tsEnableAuditCreateTable){
+  if (tsEnableAuditCreateTable) {
     int64_t clusterId = pVnode->config.syncCfg.nodeInfo[0].clusterId;
 
     SName name = {0};
     tNameFromString(&name, pVnode->config.dbname, T_NAME_ACCT | T_NAME_DB);
 
     SStringBuilder sb = {0};
-    for(int32_t iReq = 0; iReq < req.nReqs; iReq++){
-      char** key = (char**)taosArrayGet(tbNames, iReq);
+    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
+      char **key = (char **)taosArrayGet(tbNames, iReq);
       taosStringBuilderAppendStringLen(&sb, *key, strlen(*key));
-      if(iReq < req.nReqs - 1){
+      if (iReq < req.nReqs - 1) {
         taosStringBuilderAppendChar(&sb, ',');
       }
       taosMemoryFreeClear(*key);
     }
 
-    size_t    len = 0;
-    char*     keyJoined = taosStringBuilderGetResult(&sb, &len);
+    size_t len = 0;
+    char  *keyJoined = taosStringBuilderGetResult(&sb, &len);
 
     auditRecord(NULL, clusterId, "dropTable", name.dbname, "", keyJoined, len);
 
     taosStringBuilderDestroy(&sb);
   }
-  
+
 _exit:
   taosArrayDestroy(tbUids);
   tdUidStoreFree(pStore);
@@ -1518,7 +1520,6 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, in
       int32_t nRow = TARRAY_SIZE(pSubmitTbData->aRowP);
       SRow  **aRow = (SRow **)TARRAY_DATA(pSubmitTbData->aRowP);
       for (int32_t iRow = 0; iRow < nRow; ++iRow) {
-
         if (aRow[iRow]->ts < minKey || aRow[iRow]->ts > maxKey || (iRow > 0 && aRow[iRow]->ts <= aRow[iRow - 1]->ts)) {
           code = TSDB_CODE_INVALID_MSG;
           vError("vgId:%d %s failed since %s, version:%" PRId64, TD_VID(pVnode), __func__, tstrerror(code), ver);
@@ -1627,7 +1628,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, in
         if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
           code = terrno;
           vError("vgId:%d failed to create table:%s, code:%s", TD_VID(pVnode), pSubmitTbData->pCreateTbReq->name,
-              tstrerror(terrno));
+                 tstrerror(terrno));
           goto _exit;
         }
         terrno = 0;
diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c
index 84aff9fa88..fdbc0b4038 100644
--- a/source/libs/function/src/builtins.c
+++ b/source/libs/function/src/builtins.c
@@ -3315,7 +3315,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "to_timestamp",
     .type = FUNCTION_TYPE_TO_TIMESTAMP,
-    .classification = FUNC_MGT_SCALAR_FUNC,
+    .classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
     .translateFunc = translateToTimestamp,
     .getEnvFunc = NULL,
     .initFunc = NULL,
diff --git a/source/libs/scalar/src/sclfunc.c b/source/libs/scalar/src/sclfunc.c
index 48886b1eec..dbdd79cc65 100644
--- a/source/libs/scalar/src/sclfunc.c
+++ b/source/libs/scalar/src/sclfunc.c
@@ -1230,7 +1230,7 @@ int32_t toTimestampFunction(SScalarParam* pInput, int32_t inputNum, SScalarParam
     code = taosChar2Ts(format, &formats, tsStr, &ts, precision, errMsg, 128);
     if (code) {
       qError("func to_timestamp failed %s", errMsg);
-      code = TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED;
+      code = code == -1 ? TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED_FORMAT_ERR : TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED_TS_ERR;
       break;
     }
     colDataSetVal(pOutput->columnData, i, (char *)&ts, false);
diff --git a/source/libs/stream/inc/streamInt.h b/source/libs/stream/inc/streamInt.h
index 63e97fca19..feaacb7969 100644
--- a/source/libs/stream/inc/streamInt.h
+++ b/source/libs/stream/inc/streamInt.h
@@ -151,6 +151,17 @@ void*         streamQueueNextItem(SStreamQueue* pQueue);
 void          streamFreeQitem(SStreamQueueItem* data);
 int32_t       streamQueueGetItemSize(const SStreamQueue* pQueue);
 
+//#define CHECKPOINT_PATH_LEN 128
+//typedef struct SChekpointDataHeader{
+//  int64_t size;
+//  char    name[CHECKPOINT_PATH_LEN];
+//  char    id[CHECKPOINT_PATH_LEN];
+//} SChekpointDataHeader;
+
+int uploadCheckpoint(char* id, char* path);
+int downloadCheckpoint(char* id, char* path);
+int deleteCheckpoint(char* id);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/source/libs/stream/src/streamCheckpoint.c b/source/libs/stream/src/streamCheckpoint.c
index 966edf7f60..3a52e68e93 100644
--- a/source/libs/stream/src/streamCheckpoint.c
+++ b/source/libs/stream/src/streamCheckpoint.c
@@ -14,6 +14,7 @@
  */
 
 #include "streamInt.h"
+#include "rsync.h"
 
 int32_t tEncodeStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq) {
   if (tStartEncode(pEncoder) < 0) return -1;
@@ -374,3 +375,134 @@ int32_t streamTaskBuildCheckpoint(SStreamTask* pTask) {
 
   return code;
 }
+
+
+//static int64_t kBlockSize = 64 * 1024;
+//static int sendCheckpointToS3(char* id, SArray* fileList){
+//  code = s3PutObjectFromFile2(from->fname, object_name);
+//  return 0;
+//}
+//static int sendCheckpointToSnode(char* id, SArray* fileList){
+//  if(strlen(id) >= CHECKPOINT_PATH_LEN){
+//    tqError("uploadCheckpoint id name too long, name:%s", id);
+//    return -1;
+//  }
+//  uint8_t* buf = taosMemoryCalloc(1, sizeof(SChekpointDataHeader) + kBlockSize);
+//  if(buf == NULL){
+//    tqError("uploadCheckpoint malloc failed");
+//    return -1;
+//  }
+//
+//  SChekpointDataHeader* pHdr = (SChekpointDataHeader*)buf;
+//  strcpy(pHdr->id, id);
+//
+//  TdFilePtr fd = NULL;
+//  for(int i = 0; i < taosArrayGetSize(fileList); i++){
+//    char* name = (char*)taosArrayGetP(fileList, i);
+//    if(strlen(name) >= CHECKPOINT_PATH_LEN){
+//      tqError("uploadCheckpoint file name too long, name:%s", name);
+//      return -1;
+//    }
+//    int64_t  offset = 0;
+//
+//    fd = taosOpenFile(name, TD_FILE_READ);
+//    tqDebug("uploadCheckpoint open file %s, file index: %d", name, i);
+//
+//    while(1){
+//      int64_t  nread = taosPReadFile(fd, buf + sizeof(SChekpointDataHeader), kBlockSize, offset);
+//      if (nread == -1) {
+//        taosCloseFile(&fd);
+//        taosMemoryFree(buf);
+//        tqError("uploadCheckpoint failed to read file name:%s,reason:%d", name, errno);
+//        return -1;
+//      } else if (nread == 0){
+//        tqDebug("uploadCheckpoint no data read, close file:%s, move to next file, open and read", name);
+//        taosCloseFile(&fd);
+//        break;
+//      } else if (nread == kBlockSize){
+//        offset += nread;
+//      } else {
+//        taosCloseFile(&fd);
+//        offset = 0;
+//      }
+//      tqDebug("uploadCheckpoint read file %s, size:%" PRId64 ", current offset:%" PRId64, name, nread, offset);
+//
+//
+//      pHdr->size = nread;
+//      strcpy(pHdr->name, name);
+//
+//      SRpcMsg rpcMsg = {0};
+//      int32_t bytes = sizeof(SChekpointDataHeader) + nread;
+//      rpcMsg.pCont = rpcMallocCont(bytes);
+//      rpcMsg.msgType = TDMT_SYNC_SNAPSHOT_SEND;
+//      rpcMsg.contLen = bytes;
+//      if (rpcMsg.pCont == NULL) {
+//        tqError("uploadCheckpoint malloc failed");
+//        taosCloseFile(&fd);
+//        taosMemoryFree(buf);
+//        return -1;
+//      }
+//      memcpy(rpcMsg.pCont, buf, bytes);
+//      int try = 3;
+//      int32_t code = 0;
+//      while(try-- > 0){
+//        code = tmsgSendReq(pEpSet, &rpcMsg);
+//        if(code == 0)
+//          break;
+//        taosMsleep(10);
+//      }
+//      if(code != 0){
+//        tqError("uploadCheckpoint send request failed code:%d", code);
+//        taosCloseFile(&fd);
+//        taosMemoryFree(buf);
+//        return -1;
+//      }
+//
+//      if(offset == 0){
+//        break;
+//      }
+//    }
+//  }
+//
+//  taosMemoryFree(buf);
+
+//}
+
+int uploadCheckpoint(char* id, char* path){
+  if(id == NULL || path == NULL || strlen(id) == 0 || strlen(path) == 0 || strlen(path) >= PATH_MAX){
+    stError("uploadCheckpoint parameters invalid");
+    return -1;
+  }
+  if(strlen(tsSnodeIp) != 0){
+    uploadRsync(id, path);
+//  }else if(tsS3StreamEnabled){
+
+  }
+  return 0;
+}
+
+int downloadCheckpoint(char* id, char* path){
+  if(id == NULL || path == NULL || strlen(id) == 0 || strlen(path) == 0 || strlen(path) >= PATH_MAX){
+    stError("downloadCheckpoint parameters invalid");
+    return -1;
+  }
+  if(strlen(tsSnodeIp) != 0){
+    downloadRsync(id, path);
+//  }else if(tsS3StreamEnabled){
+
+  }
+  return 0;
+}
+
+int deleteCheckpoint(char* id){
+  if(id == NULL || strlen(id) == 0){
+    stError("deleteCheckpoint parameters invalid");
+    return -1;
+  }
+  if(strlen(tsSnodeIp) != 0){
+    deleteRsync(id);
+//  }else if(tsS3StreamEnabled){
+
+  }
+  return 0;
+}
diff --git a/source/libs/stream/test/CMakeLists.txt b/source/libs/stream/test/CMakeLists.txt
index 629b04ae51..d756b71e29 100644
--- a/source/libs/stream/test/CMakeLists.txt
+++ b/source/libs/stream/test/CMakeLists.txt
@@ -18,6 +18,17 @@ TARGET_INCLUDE_DIRECTORIES(
   PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
 )
 
+ADD_EXECUTABLE(checkpointTest checkpointTest.cpp)
+TARGET_LINK_LIBRARIES(
+        checkpointTest
+        PUBLIC os common gtest stream executor qcom index transport util
+)
+
+TARGET_INCLUDE_DIRECTORIES(
+        checkpointTest
+        PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
+)
+
 add_test(
   NAME streamUpdateTest
   COMMAND streamUpdateTest
diff --git a/source/libs/stream/test/checkpointTest.cpp b/source/libs/stream/test/checkpointTest.cpp
new file mode 100644
index 0000000000..6b53f13c71
--- /dev/null
+++ b/source/libs/stream/test/checkpointTest.cpp
@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <gtest/gtest.h>
+
+#include <taoserror.h>
+#include <tglobal.h>
+#include <iostream>
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wwrite-strings"
+#pragma GCC diagnostic ignored "-Wunused-function"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+
+#include "rsync.h"
+#include "streamInt.h"
+
+int main(int argc, char **argv) {
+  testing::InitGoogleTest(&argc, argv);
+
+  strcpy(tsSnodeIp, "127.0.0.1");
+  return RUN_ALL_TESTS();
+}
+
+TEST(testCase, checkpointUpload_Test) {
+  stopRsync();
+  startRsync();
+
+  taosSsleep(5);
+  char* id = "2013892036";
+
+  uploadCheckpoint(id, "/Users/mingmingwanng/rsync/");
+}
+
+TEST(testCase, checkpointDownload_Test) {
+  char* id = "2013892036";
+  downloadRsync(id, "/Users/mingmingwanng/rsync/tmp");
+}
+
+TEST(testCase, checkpointDelete_Test) {
+  char* id = "2013892036";
+  deleteRsync(id);
+}
diff --git a/source/dnode/vnode/src/vnd/vnodeCos.c b/source/util/src/cos.c
similarity index 99%
rename from source/dnode/vnode/src/vnd/vnodeCos.c
rename to source/util/src/cos.c
index 6e36739f5a..23b2d53990 100644
--- a/source/dnode/vnode/src/vnd/vnodeCos.c
+++ b/source/util/src/cos.c
@@ -1,6 +1,6 @@
 #define ALLOW_FORBID_FUNC
 
-#include "vndCos.h"
+#include "cos.h"
 
 extern char   tsS3Endpoint[];
 extern char   tsS3AccessKeyId[];
diff --git a/source/util/src/rsync.c b/source/util/src/rsync.c
new file mode 100644
index 0000000000..cbb447a33e
--- /dev/null
+++ b/source/util/src/rsync.c
@@ -0,0 +1,235 @@
+//
+// Created by mingming wanng on 2023/11/2.
+//
+#include "rsync.h"
+#include <stdlib.h>
+#include "tglobal.h"
+
+#define ERRNO_ERR_FORMAT  "errno:%d,msg:%s"
+#define ERRNO_ERR_DATA    errno,strerror(errno)
+
+// deleteRsync function produce empty directories, traverse base directory to remove them
+static void removeEmptyDir(){
+  TdDirPtr pDir = taosOpenDir(tsCheckpointBackupDir);
+  if (pDir == NULL) return;
+
+  TdDirEntryPtr de = NULL;
+  while ((de = taosReadDir(pDir)) != NULL) {
+    if (!taosDirEntryIsDir(de)) {
+      continue;
+    }
+
+    if (strcmp(taosGetDirEntryName(de), ".") == 0 || strcmp(taosGetDirEntryName(de), "..") == 0) continue;
+
+    char filename[PATH_MAX] = {0};
+    snprintf(filename, sizeof(filename), "%s%s", tsCheckpointBackupDir, taosGetDirEntryName(de));
+
+    TdDirPtr pDirTmp = taosOpenDir(filename);
+    TdDirEntryPtr deTmp = NULL;
+    bool empty = true;
+    while ((deTmp = taosReadDir(pDirTmp)) != NULL){
+      if (strcmp(taosGetDirEntryName(deTmp), ".") == 0 || strcmp(taosGetDirEntryName(deTmp), "..") == 0) continue;
+      empty = false;
+    }
+    if(empty) taosRemoveDir(filename);
+    taosCloseDir(&pDirTmp);
+  }
+
+  taosCloseDir(&pDir);
+}
+
+#ifdef WINDOWS
+// C:\TDengine\data\backup\checkpoint\ -> /c/TDengine/data/backup/checkpoint/
+static void changeDirFromWindowsToLinux(char* from, char* to){
+  to[0] = '/';
+  to[1] = from[0];
+  for(int i = 2; i < strlen(from); i++) {
+    if (from[i] == '\\') {
+      to[i] = '/';
+    } else {
+      to[i] = from[i];
+    }
+  }
+}
+#endif
+
+static int generateConfigFile(char* confDir){
+  TdFilePtr pFile = taosOpenFile(confDir, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  if (pFile == NULL) {
+    uError("[rsync] open conf file error, dir:%s,"ERRNO_ERR_FORMAT, confDir, ERRNO_ERR_DATA);
+    return -1;
+  }
+
+#ifdef WINDOWS
+  char path[PATH_MAX] = {0};
+  changeDirFromWindowsToLinux(tsCheckpointBackupDir, path);
+#endif
+
+  char confContent[PATH_MAX*4] = {0};
+  snprintf(confContent, PATH_MAX*4,
+#ifndef WINDOWS
+           "uid = root\n"
+           "gid = root\n"
+#endif
+           "use chroot = false\n"
+           "max connections = 200\n"
+           "timeout = 100\n"
+           "lock file = %srsync.lock\n"
+           "log file = %srsync.log\n"
+           "ignore errors = true\n"
+           "read only = false\n"
+           "list = false\n"
+           "[checkpoint]\n"
+           "path = %s", tsCheckpointBackupDir, tsCheckpointBackupDir,
+#ifdef WINDOWS
+           path
+#else
+           tsCheckpointBackupDir
+#endif
+           );
+  uDebug("[rsync] conf:%s", confContent);
+  if (taosWriteFile(pFile, confContent, strlen(confContent)) <= 0){
+    uError("[rsync] write conf file error,"ERRNO_ERR_FORMAT, ERRNO_ERR_DATA);
+    taosCloseFile(&pFile);
+    return -1;
+  }
+
+  taosCloseFile(&pFile);
+  return 0;
+}
+
+static int execCommand(char* command){
+  int try = 3;
+  int32_t code = 0;
+  while(try-- > 0) {
+    code = system(command);
+    if (code == 0) {
+      break;
+    }
+    taosMsleep(10);
+  }
+  return code;
+}
+
+void stopRsync(){
+  int code =
+#ifdef WINDOWS
+  system("taskkill /f /im rsync.exe");
+#else
+  system("pkill rsync");
+#endif
+  if(code != 0){
+    uError("[rsync] stop rsync server failed,"ERRNO_ERR_FORMAT, ERRNO_ERR_DATA);
+    return;
+  }
+  uDebug("[rsync] stop rsync server successful");
+}
+
+void startRsync(){
+  if(taosMulMkDir(tsCheckpointBackupDir) != 0){
+    uError("[rsync] build checkpoint backup dir failed, dir:%s,"ERRNO_ERR_FORMAT, tsCheckpointBackupDir, ERRNO_ERR_DATA);
+    return;
+  }
+  removeEmptyDir();
+
+  char confDir[PATH_MAX] = {0};
+  snprintf(confDir, PATH_MAX, "%srsync.conf", tsCheckpointBackupDir);
+
+  int code = generateConfigFile(confDir);
+  if(code != 0){
+    return;
+  }
+
+  char cmd[PATH_MAX] = {0};
+  snprintf(cmd, PATH_MAX, "rsync --daemon --config=%s", confDir);
+  // start rsync service to backup checkpoint
+  code = system(cmd);
+  if(code != 0){
+    uError("[rsync] start server failed, code:%d,"ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
+    return;
+  }
+  uDebug("[rsync] start server successful");
+}
+
+int uploadRsync(char* id, char* path){
+#ifdef WINDOWS
+  char pathTransform[PATH_MAX] = {0};
+  changeDirFromWindowsToLinux(path, pathTransform);
+#endif
+  char command[PATH_MAX] = {0};
+#ifdef WINDOWS
+  if(pathTransform[strlen(pathTransform) - 1] != '/'){
+#else
+  if(path[strlen(path) - 1] != '/'){
+#endif
+    snprintf(command, PATH_MAX, "rsync -av --delete --timeout=10 --bwlimit=100000 %s/ rsync://%s/checkpoint/%s/",
+#ifdef WINDOWS
+             pathTransform
+#else
+             path
+#endif
+             , tsSnodeIp, id);
+  }else{
+    snprintf(command, PATH_MAX, "rsync -av --delete --timeout=10 --bwlimit=100000 %s rsync://%s/checkpoint/%s/",
+#ifdef WINDOWS
+             pathTransform
+#else
+             path
+#endif
+             , tsSnodeIp, id);
+  }
+
+  int code = execCommand(command);
+  if(code != 0){
+    uError("[rsync] send failed code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
+    return -1;
+  }
+  uDebug("[rsync] upload data:%s successful", id);
+  return 0;
+}
+
+int downloadRsync(char* id, char* path){
+#ifdef WINDOWS
+  char pathTransform[PATH_MAX] = {0};
+  changeDirFromWindowsToLinux(path, pathTransform);
+#endif
+  char command[PATH_MAX] = {0};
+  snprintf(command, PATH_MAX, "rsync -av --timeout=10 --bwlimit=100000 rsync://%s/checkpoint/%s/ %s",
+           tsSnodeIp, id,
+#ifdef WINDOWS
+           pathTransform
+#else
+           path
+#endif
+           );
+
+  int code = execCommand(command);
+  if(code != 0){
+    uError("[rsync] get failed code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
+    return -1;
+  }
+  uDebug("[rsync] down data:%s successful", id);
+  return 0;
+}
+
+int deleteRsync(char* id){
+  char* tmp = "./tmp_empty/";
+  int code = taosMkDir(tmp);
+  if(code != 0){
+    uError("[rsync] make tmp dir failed. code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
+    return -1;
+  }
+  char command[PATH_MAX] = {0};
+  snprintf(command, PATH_MAX, "rsync -av --delete --timeout=10 %s rsync://%s/checkpoint/%s/",
+           tmp, tsSnodeIp, id);
+
+  code = execCommand(command);
+  taosRemoveDir(tmp);
+  if(code != 0){
+    uError("[rsync] get failed code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
+    return -1;
+  }
+  uDebug("[rsync] delete data:%s successful", id);
+
+  return 0;
+}
\ No newline at end of file
diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index 58022cb8f1..6c964c3da0 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -613,7 +613,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_FUNTION_PARA_TYPE,        "Invalid function par
 TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_FUNTION_PARA_VALUE,       "Invalid function para value")
 TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION,      "Not buildin function")
 TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_DUP_TIMESTAMP,            "Duplicate timestamps not allowed in function")
-TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED,      "Func to_timestamp failed, check log for detail")
+TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED_FORMAT_ERR, "Func to_timestamp failed, format mismatch")
+TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_TO_TIMESTAMP_FAILED_TS_ERR, "Func to_timestamp failed, wrong timestamp")
 
 //udf
 TAOS_DEFINE_ERROR(TSDB_CODE_UDF_STOPPING,                   "udf is stopping")
diff --git a/tests/system-test/2-query/func_to_char_timestamp.py b/tests/system-test/2-query/func_to_char_timestamp.py
index 639811d275..d955e00a82 100644
--- a/tests/system-test/2-query/func_to_char_timestamp.py
+++ b/tests/system-test/2-query/func_to_char_timestamp.py
@@ -166,6 +166,44 @@ class TDTestCase:
     def run(self):
         self.prepareTestEnv()
         self.test_to_timestamp()
+        self.test_ns_to_timestamp()
+
+    def create_tables(self):
+        tdSql.execute("create database if not exists test_us precision 'us'")
+        tdSql.execute("create database if not exists test_ns precision 'ns'")
+        tdSql.execute("use test_us")
+        tdSql.execute(f"CREATE STABLE `meters_us` (`ts` TIMESTAMP, `ip_value` FLOAT, `ip_quality` INT, `ts2` timestamp) TAGS (`t1` INT)")
+        tdSql.execute(f"CREATE TABLE `ctb1_us` USING `meters_us` (`t1`) TAGS (1)")
+        tdSql.execute(f"CREATE TABLE `ctb2_us` USING `meters_us` (`t1`) TAGS (2)")
+        tdSql.execute("use test_ns")
+        tdSql.execute(f"CREATE STABLE `meters_ns` (`ts` TIMESTAMP, `ip_value` FLOAT, `ip_quality` INT, `ts2` timestamp) TAGS (`t1` INT)")
+        tdSql.execute(f"CREATE TABLE `ctb1_ns` USING `meters_ns` (`t1`) TAGS (1)")
+        tdSql.execute(f"CREATE TABLE `ctb2_ns` USING `meters_ns` (`t1`) TAGS (2)")
+
+    def insert_ns_data(self):
+        tdLog.debug("start to insert data ............")
+        tdSql.execute(f"INSERT INTO `test_us`.`ctb1_us` VALUES ('2023-07-01 00:00:00.123456', 10.30000, 100, '2023-07-01 00:00:00.123456')")
+        tdSql.execute(f"INSERT INTO `test_us`.`ctb2_us` VALUES ('2023-08-01 00:00:00.123456', 20.30000, 200, '2023-07-01 00:00:00.123456')")
+        tdSql.execute(f"INSERT INTO `test_ns`.`ctb1_ns` VALUES ('2023-07-01 00:00:00.123456789', 10.30000, 100, '2023-07-01 00:00:00.123456000')")
+        tdSql.execute(f"INSERT INTO `test_ns`.`ctb2_ns` VALUES ('2023-08-01 00:00:00.123456789', 20.30000, 200, '2023-08-01 00:00:00.123456789')")
+        tdLog.debug("insert data ............ [OK]")
+
+    def test_ns_to_timestamp(self):
+        self.create_tables()
+        self.insert_ns_data()
+        tdSql.query("select to_timestamp('2023-08-1 10:10:10.123456789', 'yyyy-mm-dd hh:mi:ss.ns')", queryTimes=1)
+        tdSql.checkData(0, 0, 1690855810123)
+        tdSql.execute('use test_ns', queryTimes=1)
+        tdSql.query("select to_timestamp('2023-08-1 10:10:10.123456789', 'yyyy-mm-dd hh:mi:ss.ns')", queryTimes=1)
+        tdSql.checkData(0, 0, 1690855810123)
+        tdSql.query("select to_char(ts2, 'yyyy-mm-dd hh:mi:ss.ns') from meters_ns", queryTimes=1)
+        tdSql.checkData(0, 0, '2023-07-01 12:00:00.123456000')
+        tdSql.checkData(1, 0, '2023-08-01 12:00:00.123456789')
+
+        tdSql.query("select to_timestamp(to_char(ts2, 'yyyy-mm-dd hh:mi:ss.ns'), 'yyyy-mm-dd hh:mi:ss.ns') from meters_ns", queryTimes=1)
+        tdSql.checkData(0, 0, 1688140800123456000)
+        tdSql.checkData(1, 0, 1690819200123456789)
+
 
     def stop(self):
         tdSql.close()