Merge remote-tracking branch 'origin/3.0' into feat/TD-24700

docs/en/12-taos-sql/01-data-type.md

@@ -42,10 +42,20 @@ In TDengine, the data types below can be used when specifying a column or tag.
 | 14  |       NCHAR       | User Defined | Multi-byte string that can include multi byte characters like Chinese characters. Each character of NCHAR type consumes 4 bytes storage. The string value should be quoted with single quotes. Literal single quote inside the string must be preceded with backslash, like `\'`. The length must be specified when defining a column or tag of NCHAR type, for example nchar(10) means it can store at most 10 characters of nchar type and will consume fixed storage of 40 bytes. An error will be reported if the string value exceeds the length defined. |
 | 15  |       JSON        |              | JSON type can only be used on tags. A tag of json type is excluded with any other tags of any other type.                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
 | 16  |      VARCHAR      | User-defined | Alias of BINARY                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
+| 16  |      GEOMETRY     | User-defined | Geometry                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
 :::note
 
+- Each row of the table cannot be longer than 48KB (64KB since version 3.0.5.0) (note that each BINARY/NCHAR/GEOMETRY column takes up an additional 2 bytes of storage space).
 - Only ASCII visible characters are suggested to be used in a column or tag of BINARY type. Multi-byte characters must be stored in NCHAR type.
 - The length of BINARY can be up to 16,374(data column is 65,517 and tag column is 16,382 since version 3.0.5.0) bytes. The string value must be quoted with single quotes. You must specify a length in bytes for a BINARY value, for example binary(20) for up to twenty single-byte characters. If the data exceeds the specified length, an error will occur. The literal single quote inside the string must be preceded with back slash like `\'`
+- The maximum length of the GEOMETRY data column is 65,517 bytes, and the maximum length of the tag column is 16,382 bytes. Supports POINT, LINESTRING, and POLYGON subtypes of 2D. The following table describes the length calculation method:
+
+    | # | **Syntax**                           | **MinLen** | **MaxLen** | **Growth of each point** |
+    |---|--------------------------------------|------------|------------|--------------------------|
+    | 1 | POINT(1.0 1.0)                       | 21         | 21         | NA                       |
+    | 2 | LINESTRING(1.0 1.0, 2.0 2.0)         | 9+2*16     | 9+4094*16  | +16                      |
+    | 3 | POLYGON((1.0 1.0, 2.0 2.0, 1.0 1.0)) | 13+3*16    | 13+4094*16 | +16                      |
+
 - Numeric values in SQL statements will be determined as integer or float type according to whether there is decimal point or whether scientific notation is used, so attention must be paid to avoid overflow. For example, 9999999999999999999 will be considered as overflow because it exceeds the upper limit of long integer, but 9999999999999999999.0 will be considered as a legal float number.
 
 :::

docs/en/12-taos-sql/03-table.md

@@ -45,9 +45,9 @@ table_option: {
 
 1. The first column of a table MUST be of type TIMESTAMP. It is automatically set as the primary key.
 2. The maximum length of the table name is 192 bytes.
-3. The maximum length of each row is 48k(64k since version 3.0.5.0) bytes, please note that the extra 2 bytes used by each BINARY/NCHAR column are also counted.
+3. The maximum length of each row is 48k(64k since version 3.0.5.0) bytes, please note that the extra 2 bytes used by each BINARY/NCHAR/GEOMETRY column are also counted.
 4. The name of the subtable can only consist of characters from the English alphabet, digits and underscore. Table names can't start with a digit. Table names are case insensitive.
-5. The maximum length in bytes must be specified when using BINARY or NCHAR types.
+5. The maximum length in bytes must be specified when using BINARY/NCHAR/GEOMETRY types.
 6. Escape character "\`" can be used to avoid the conflict between table names and reserved keywords, above rules will be bypassed when using escape character on table names, but the upper limit for the name length is still valid. The table names specified using escape character are case sensitive.
    For example \`aBc\` and \`abc\` are different table names but `abc` and `aBc` are same table names because they are both converted to `abc` internally.
    Only ASCII visible characters can be used with escape character.

docs/en/12-taos-sql/16-operators.md

@@ -39,7 +39,7 @@ TDengine supports the `UNION` and `UNION ALL` operations. UNION ALL collects all
 | 3   |      \>, <       | All types except BLOB, MEDIUMBLOB, and JSON                             | Greater than and less than           |
 | 4   |     \>=, <=      | All types except BLOB, MEDIUMBLOB, and JSON                             | Greater than or equal to and less than or equal to   |
 | 5   |   IS [NOT] NULL   | All types                                                             | Indicates whether the value is null           |
-| 6   | [NOT] BETWEEN AND | All types except BLOB, MEDIUMBLOB, and JSON                       | Closed interval comparison           |
+| 6   | [NOT] BETWEEN AND | All types except BLOB, MEDIUMBLOB, JSON and GEOMETRY                  | Closed interval comparison           |
 | 7   |        IN         | All types except BLOB, MEDIUMBLOB, and JSON; the primary key (timestamp) is also not supported | Equal to any value in the list |
 | 8   |       LIKE        | BINARY, NCHAR, and VARCHAR                                             | Wildcard match           |
 | 9   |   MATCH, NMATCH   | BINARY, NCHAR, and VARCHAR                                             | Regular expression match       |

docs/en/12-taos-sql/29-changes.md

@@ -18,6 +18,7 @@ description: This document describes how TDengine SQL has changed in version 3.0
 | 8 | Mixed operations | Enhanced | Mixing scalar and vector operations in queries has been enhanced and is supported in all SELECT clauses.
 | 9 | Tag operations | Added | Tag columns can be used in queries and clauses like data columns.
 | 10 | Timeline clauses and time functions in supertables | Enhanced | When PARTITION BY is not used, data in supertables is merged into a single timeline.
+| 11 | GEOMETRY | Added | Geometry
 
 ## SQL Syntax
 

docs/en/14-reference/12-config/index.md

@@ -166,7 +166,7 @@ Please note the `taoskeeper` needs to be installed and running to create the `lo
 
 | Attribute     | Description                                                                  |
 | ------------- | ---------------------------------------------------------------------------- |
-| Applicable    | Server Only                                                                  |
+| Applicable    | Server and Client                                                            |
 | Meaning       | Switch for allowing TDengine to collect and report service usage information |
 | Value Range   | 0: Not allowed; 1: Allowed                                                   |
 | Default Value | 1                                                                            |
@@ -174,7 +174,7 @@ Please note the `taoskeeper` needs to be installed and running to create the `lo
 
 | Attribute     | Description                                                                  |
 | ------------- | ---------------------------------------------------------------------------- |
-| Applicable    | Server Only                                                                  |
+| Applicable    | Server and Client                                                            |
 | Meaning       | Switch for allowing TDengine to collect and report crash related information |
 | Value Range   | 0,1   0: Not allowed; 1: allowed                                             |
 | Default Value | 1                                                                            |

docs/zh/12-taos-sql/01-data-type.md

@@ -42,12 +42,21 @@ CREATE DATABASE db_name PRECISION 'ns';
 | 14  |       NCHAR       | 自定义    | 记录包含多字节字符在内的字符串，如中文字符。每个 NCHAR 字符占用 4 字节的存储空间。字符串两端使用单引号引用，字符串内的单引号需用转义字符 `\'`。NCHAR 使用时须指定字符串大小，类型为 NCHAR(10) 的列表示此列的字符串最多存储 10 个 NCHAR 字符。如果用户字符串长度超出声明长度，将会报错。 |
 | 15  |       JSON        |           | JSON 数据类型， 只有 Tag 可以是 JSON 格式                                                                                                                                                                                                                                                                         |
 | 16  |      VARCHAR      | 自定义    | BINARY 类型的别名                                                                                                                                                                                                                                                                                                 |
+| 17  |      GEOMETRY     | 自定义    | 几何类型                                                                                                                                                                                                                                                                                                 |
 
 :::note
 
-- 表的每行长度不能超过 48KB（从 3.0.5.0 版本开始为 64KB）（注意：每个 BINARY/NCHAR 类型的列还会额外占用 2 个字节的存储位置）。
+- 表的每行长度不能超过 48KB（从 3.0.5.0 版本开始为 64KB）（注意：每个 BINARY/NCHAR/GEOMETRY 类型的列还会额外占用 2 个字节的存储位置）。
 - 虽然 BINARY 类型在底层存储上支持字节型的二进制字符，但不同编程语言对二进制数据的处理方式并不保证一致，因此建议在 BINARY 类型中只存储 ASCII 可见字符，而避免存储不可见字符。多字节的数据，例如中文字符，则需要使用 NCHAR 类型进行保存。如果强行使用 BINARY 类型保存中文字符，虽然有时也能正常读写，但并不带有字符集信息，很容易出现数据乱码甚至数据损坏等情况。
 - BINARY 类型理论上最长可以有 16,374（从 3.0.5.0 版本开始，数据列为 65,517，标签列为 16,382） 字节。BINARY 仅支持字符串输入，字符串两端需使用单引号引用。使用时须指定大小，如 BINARY(20) 定义了最长为 20 个单字节字符的字符串，每个字符占 1 字节的存储空间，总共固定占用 20 字节的空间，此时如果用户字符串超出 20 字节将会报错。对于字符串内的单引号，可以用转义字符反斜线加单引号来表示，即 `\'`。
+- GEOMETRY 类型数据列为最大长度为 65,517 字节，标签列最大长度为 16,382 字节。支持 2D 的 POINT、LINESTRING 和 POLYGON 子类型数据。长度计算方式如下表所示：
+
+    | # | **语法**                               | **最小长度** | **最大长度**   | **每组坐标长度增长** |
+    |---|--------------------------------------|----------|------------|--------------|
+    | 1 | POINT(1.0 1.0)                       | 21       | 21         | 无            |
+    | 2 | LINESTRING(1.0 1.0, 2.0 2.0)         | 9+2*16   | 9+4094*16  | +16          |
+    | 3 | POLYGON((1.0 1.0, 2.0 2.0, 1.0 1.0)) | 13+3*16  | 13+4094*16 | +16          |
+
 - SQL 语句中的数值类型将依据是否存在小数点，或使用科学计数法表示，来判断数值类型是否为整型或者浮点型，因此在使用时要注意相应类型越界的情况。例如，9999999999999999999 会认为超过长整型的上边界而溢出，而 9999999999999999999.0 会被认为是有效的浮点数。
 
 :::

docs/zh/12-taos-sql/03-table.md

@@ -43,9 +43,9 @@ table_option: {
 
 1. 表的第一个字段必须是 TIMESTAMP，并且系统自动将其设为主键；
 2. 表名最大长度为 192；
-3. 表的每行长度不能超过 48KB（从 3.0.5.0 版本开始为 64KB）;（注意：每个 BINARY/NCHAR 类型的列还会额外占用 2 个字节的存储位置）
+3. 表的每行长度不能超过 48KB（从 3.0.5.0 版本开始为 64KB）;（注意：每个 BINARY/NCHAR/GEOMETRY 类型的列还会额外占用 2 个字节的存储位置）
 4. 子表名只能由字母、数字和下划线组成，且不能以数字开头，不区分大小写
-5. 使用数据类型 binary 或 nchar，需指定其最长的字节数，如 binary(20)，表示 20 字节；
+5. 使用数据类型 BINARY/NCHAR/GEOMETRY，需指定其最长的字节数，如 BINARY(20)，表示 20 字节；
 6. 为了兼容支持更多形式的表名，TDengine 引入新的转义符 "\`"，可以让表名与关键词不冲突，同时不受限于上述表名称合法性约束检查。但是同样具有长度限制要求。使用转义字符以后，不再对转义字符中的内容进行大小写统一。
    例如：\`aBc\` 和 \`abc\` 是不同的表名，但是 abc 和 aBc 是相同的表名。
    需要注意的是转义字符中的内容必须是可打印字符。

docs/zh/12-taos-sql/16-operators.md

@@ -39,7 +39,7 @@ TDengine 支持 `UNION ALL` 和 `UNION` 操作符。UNION ALL 将查询返回的
 | 3   |      \>, <       | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于，小于           |
 | 4   |     \>=, <=      | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于等于，小于等于   |
 | 5   |   IS [NOT] NULL   | 所有类型                                                             | 是否为空值           |
-| 6   | [NOT] BETWEEN AND | 除 BOOL、BLOB、MEDIUMBLOB 和 JSON 外的所有类型                       | 闭区间比较           |
+| 6   | [NOT] BETWEEN AND | 除 BOOL、BLOB、MEDIUMBLOB、JSON 和 GEOMETRY 外的所有类型                | 闭区间比较           |
 | 7   |        IN         | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型，且不可以为表的时间戳主键列 | 与列表内的任意值相等 |
 | 8   |       LIKE        | BINARY、NCHAR 和 VARCHAR                                             | 通配符匹配           |
 | 9   |   MATCH, NMATCH   | BINARY、NCHAR 和 VARCHAR                                             | 正则表达式匹配       |

docs/zh/12-taos-sql/29-changes.md

@@ -18,6 +18,7 @@ description: "TDengine 3.0 版本的语法变更说明"
 | 8 | 混合运算 | 增强 | 查询中的混合运算（标量运算和矢量运算混合）全面增强，SELECT的各个子句均全面支持符合语法语义的混合运算。
 | 9 | 标签运算 | 新增 |在查询中，标签列可以像普通列一样参与各种运算，用于各种子句。
 | 10 | 时间线子句和时间函数用于超级表查询 | 增强 |没有PARTITION BY时，超级表的数据会被合并成一条时间线。
+| 11 | GEOMETRY | 新增 | 几何类型。
 
 ## SQL 语句变更
 

docs/zh/14-reference/12-config/index.md

@@ -184,7 +184,7 @@ taos -C
 
 | 属性     | 说明                     |
 | -------- | ------------------------ |
-| 适用范围 | 仅服务端适用             |
+| 适用范围 | 客户端和服务端都适用     |
 | 含义     | 是否上传 telemetry       |
 | 取值范围 | 0,1   0: 不上传；1：上传 |
 | 缺省值   | 1                        |
@@ -193,7 +193,7 @@ taos -C
 
 | 属性     | 说明                     |
 | -------- | ------------------------ |
-| 适用范围 | 仅服务端适用             |
+| 适用范围 | 客户端和服务端都适用     |
 | 含义     | 是否上传 crash 信息      |
 | 取值范围 | 0,1   0: 不上传；1：上传 |
 | 缺省值   | 1                        |

include/common/tmsg.h

@@ -1703,6 +1703,7 @@ int32_t tDeserializeSShowVariablesReq(void* buf, int32_t bufLen, SShowVariablesR
 typedef struct {
   char name[TSDB_CONFIG_OPTION_LEN + 1];
   char value[TSDB_CONFIG_VALUE_LEN + 1];
+  char scope[TSDB_CONFIG_SCOPE_LEN + 1];
 } SVariablesInfo;
 
 typedef struct {

include/libs/nodes/cmdnodes.h

@@ -36,9 +36,10 @@ extern "C" {
 #define SHOW_CREATE_TB_RESULT_FIELD1_LEN (TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE)
 #define SHOW_CREATE_TB_RESULT_FIELD2_LEN (TSDB_MAX_ALLOWED_SQL_LEN * 3)
 
-#define SHOW_LOCAL_VARIABLES_RESULT_COLS       2
+#define SHOW_LOCAL_VARIABLES_RESULT_COLS       3
 #define SHOW_LOCAL_VARIABLES_RESULT_FIELD1_LEN (TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE)
 #define SHOW_LOCAL_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
+#define SHOW_LOCAL_VARIABLES_RESULT_FIELD3_LEN (TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE)
 
 #define SHOW_ALIVE_RESULT_COLS 1
 

include/libs/nodes/plannodes.h

@@ -56,6 +56,7 @@ typedef struct SLogicNode {
   EGroupAction       groupAction;
   EOrder             inputTsOrder;
   EOrder             outputTsOrder;
+  bool               forceCreateNonBlockingOptr;  // true if the operator can use non-blocking(pipeline) mode
 } SLogicNode;
 
 typedef enum EScanType {
@@ -106,6 +107,7 @@ typedef struct SScanLogicNode {
   bool          hasNormalCols;  // neither tag column nor primary key tag column
   bool          sortPrimaryKey;
   bool          igLastNull;
+  bool          groupOrderScan;
 } SScanLogicNode;
 
 typedef struct SJoinLogicNode {
@@ -343,6 +345,7 @@ typedef struct SPhysiNode {
   struct SPhysiNode*  pParent;
   SNode*              pLimit;
   SNode*              pSlimit;
+  bool                forceCreateNonBlockingOptr;
 } SPhysiNode;
 
 typedef struct SScanPhysiNode {
@@ -353,6 +356,7 @@ typedef struct SScanPhysiNode {
   uint64_t   suid;
   int8_t     tableType;
   SName      tableName;
+  bool       groupOrderScan;
 } SScanPhysiNode;
 
 typedef SScanPhysiNode STagScanPhysiNode;

include/libs/stream/tstream.h

@@ -45,8 +45,8 @@ enum {
   TASK_STATUS__FAIL,
   TASK_STATUS__STOP,
   TASK_STATUS__SCAN_HISTORY,  // stream task scan history data by using tsdbread in the stream scanner
-  TASK_STATUS__HALT,          // stream task will handle all data in the input queue, and then paused
-  TASK_STATUS__PAUSE,
+  TASK_STATUS__HALT,          // pause, but not be manipulated by user command
+  TASK_STATUS__PAUSE,         // pause
 };
 
 enum {
@@ -272,6 +272,7 @@ typedef struct SStreamStatus {
   int8_t        keepTaskStatus;
   bool          transferState;
   int8_t        timerActive;     // timer is active
+  int8_t        pauseAllowed;    // allowed task status to be set to be paused
 } SStreamStatus;
 
 typedef struct SHistDataRange {
@@ -296,15 +297,15 @@ typedef struct SDispatchMsgInfo {
 } SDispatchMsgInfo;
 
 typedef struct {
-  int8_t        outputType;
-  int8_t        outputStatus;
-  SStreamQueue* outputQueue;
-} SSTaskOutputInfo;
+  int8_t        type;
+  int8_t        status;
+  SStreamQueue* queue;
+} STaskOutputInfo;
 
 struct SStreamTask {
   SStreamId        id;
   SSTaskBasicInfo  info;
-  int8_t           outputType;
+  STaskOutputInfo  outputInfo;
   SDispatchMsgInfo msgInfo;
   SStreamStatus    status;
   SCheckpointInfo  chkInfo;
@@ -315,7 +316,7 @@ struct SStreamTask {
   SArray*          pUpstreamEpInfoList;  // SArray<SStreamChildEpInfo*>, // children info
   int32_t          nextCheckId;
   SArray*          checkpointInfo;  // SArray<SStreamCheckpointInfo>
-
+  int64_t          initTs;
   // output
   union {
     STaskDispatcherFixedEp fixedEpDispatcher;
@@ -326,9 +327,7 @@ struct SStreamTask {
   };
 
   int8_t        inputStatus;
-  int8_t        outputStatus;
   SStreamQueue* inputQueue;
-  SStreamQueue* outputQueue;
 
   // trigger
   int8_t        triggerStatus;
@@ -337,6 +336,8 @@ struct SStreamTask {
   void*         launchTaskTimer;
   SMsgCb*       pMsgCb;  // msg handle
   SStreamState* pState;  // state backend
+  SArray*       pRspMsgList;
+  TdThreadMutex lock;
 
   // the followings attributes don't be serialized
   int32_t             notReadyTasks;
@@ -458,7 +459,9 @@ typedef struct {
 
 typedef struct {
   int64_t streamId;
-  int32_t taskId;
+  int32_t upstreamTaskId;
+  int32_t downstreamTaskId;
+  int32_t upstreamNodeId;
   int32_t childId;
 } SStreamScanHistoryFinishReq, SStreamTransferReq;
 
@@ -519,6 +522,17 @@ int32_t tDecodeSStreamCheckpointReq(SDecoder* pDecoder, SStreamCheckpointReq* pR
 int32_t tEncodeSStreamCheckpointRsp(SEncoder* pEncoder, const SStreamCheckpointRsp* pRsp);
 int32_t tDecodeSStreamCheckpointRsp(SDecoder* pDecoder, SStreamCheckpointRsp* pRsp);
 
+typedef struct {
+  int64_t streamId;
+  int32_t upstreamTaskId;
+  int32_t upstreamNodeId;
+  int32_t downstreamId;
+  int32_t downstreamNode;
+} SStreamCompleteHistoryMsg;
+
+int32_t tEncodeCompleteHistoryDataMsg(SEncoder* pEncoder, const SStreamCompleteHistoryMsg* pReq);
+int32_t tDecodeCompleteHistoryDataMsg(SDecoder* pDecoder, SStreamCompleteHistoryMsg* pReq);
+
 typedef struct {
   int64_t streamId;
   int32_t downstreamTaskId;
@@ -559,7 +573,6 @@ int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, S
 int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, int32_t code);
 
 int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq, SRpcMsg* pMsg);
-// int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp);
 
 void    streamTaskInputFail(SStreamTask* pTask);
 int32_t streamTryExec(SStreamTask* pTask);
@@ -569,17 +582,20 @@ bool    streamTaskShouldStop(const SStreamStatus* pStatus);
 bool    streamTaskShouldPause(const SStreamStatus* pStatus);
 bool    streamTaskIsIdle(const SStreamTask* pTask);
 
+SStreamChildEpInfo * streamTaskGetUpstreamTaskEpInfo(SStreamTask* pTask, int32_t taskId);
 int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz);
 
 char*   createStreamTaskIdStr(int64_t streamId, int32_t taskId);
 
 // recover and fill history
-void    streamPrepareNdoCheckDownstream(SStreamTask* pTask);
-int32_t streamTaskCheckDownstreamTasks(SStreamTask* pTask);
+void    streamTaskCheckDownstreamTasks(SStreamTask* pTask);
+int32_t streamTaskDoCheckDownstreamTasks(SStreamTask* pTask);
 int32_t streamTaskLaunchScanHistory(SStreamTask* pTask);
 int32_t streamTaskCheckStatus(SStreamTask* pTask);
+int32_t streamSendCheckRsp(const SStreamMeta* pMeta, const SStreamTaskCheckReq* pReq, SStreamTaskCheckRsp* pRsp,
+                           SRpcHandleInfo* pRpcInfo, int32_t taskId);
 int32_t streamProcessCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* pRsp);
-int32_t streamCheckHistoryTaskDownstream(SStreamTask* pTask);
+int32_t streamLaunchFillHistoryTask(SStreamTask* pTask);
 int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
 int32_t streamStartRecoverTask(SStreamTask* pTask, int8_t igUntreated);
 void    streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask);
@@ -593,6 +609,12 @@ int32_t     streamSetParamForScanHistory(SStreamTask* pTask);
 int32_t     streamRestoreParam(SStreamTask* pTask);
 int32_t     streamSetStatusNormal(SStreamTask* pTask);
 const char* streamGetTaskStatusStr(int32_t status);
+void        streamTaskPause(SStreamTask* pTask);
+void        streamTaskResume(SStreamTask* pTask);
+void        streamTaskHalt(SStreamTask* pTask);
+void        streamTaskResumeFromHalt(SStreamTask* pTask);
+void        streamTaskDisablePause(SStreamTask* pTask);
+void        streamTaskEnablePause(SStreamTask* pTask);
 
 // source level
 int32_t streamSetParamForStreamScannerStep1(SStreamTask* pTask, SVersionRange* pVerRange, STimeWindow* pWindow);
@@ -604,8 +626,9 @@ int32_t streamDispatchScanHistoryFinishMsg(SStreamTask* pTask);
 int32_t streamDispatchTransferStateMsg(SStreamTask* pTask);
 
 // agg level
-int32_t streamAggScanHistoryPrepare(SStreamTask* pTask);
-int32_t streamProcessScanHistoryFinishReq(SStreamTask* pTask, int32_t taskId, int32_t childId);
+int32_t streamTaskScanHistoryPrepare(SStreamTask* pTask);
+int32_t streamProcessScanHistoryFinishReq(SStreamTask* pTask, SStreamScanHistoryFinishReq *pReq, SRpcHandleInfo* pRpcInfo);
+int32_t streamProcessScanHistoryFinishRsp(SStreamTask* pTask);
 
 // stream task meta
 void         streamMetaInit();

include/libs/tfs/tfs.h

@@ -69,6 +69,13 @@ void tfsUpdateSize(STfs *pTfs);
  */
 SDiskSize tfsGetSize(STfs *pTfs);
 
+/**
+ * @brief Get the number of disks at level of multi-tier storage.
+ *
+ * @param pTfs
+ * @return int32_t
+ */
+int32_t tfsGetDisksAtLevel(STfs *pTfs, int32_t level);
 /**
  * @brief Get level of multi-tier storage.
  *
@@ -123,6 +130,15 @@ int32_t tfsMkdir(STfs *pTfs, const char *rname);
  */
 int32_t tfsMkdirAt(STfs *pTfs, const char *rname, SDiskID diskId);
 
+/**
+ * @brief Recursive make directory at all levels in tfs.
+ *
+ * @param pTfs The fs object.
+ * @param rname The rel name of directory.
+ * @return int32_t 0 for success, -1 for failure.
+ */
+int32_t tfsMkdirRecur(STfs *pTfs, const char *rname);
+
 /**
  * @brief Recursive create directories in tfs.
  *
@@ -160,7 +176,17 @@ int32_t tfsRmdir(STfs *pTfs, const char *rname);
  * @param nrname The rel name of new file.
  * @return int32_t 0 for success, -1 for failure.
  */
-int32_t tfsRename(STfs *pTfs, const char *orname, const char *nrname);
+int32_t tfsRename(STfs *pTfs, int32_t diskPrimary, const char *orname, const char *nrname);
+
+/**
+ * @brief Search fname in level of tfs
+ *
+ * @param pTfs The fs object.
+ * @param level The level to search on
+ * @param fname The relative file name to be searched
+ * @param int32_t diskId for successs, -1 for failure
+ */
+int32_t tfsSearch(STfs *pTfs, int32_t level, const char *fname);
 
 /**
  * @brief Init file object in tfs.

include/util/taoserror.h

@@ -416,7 +416,7 @@ int32_t* taosGetErrno();
 // #define TSDB_CODE_VND_MSG_NOT_PROCESSED      TAOS_DEF_ERROR_CODE(0, 0x0501) // 2.x
 // #define TSDB_CODE_VND_ACTION_NEED_REPROCESS. TAOS_DEF_ERROR_CODE(0, 0x0502) // 2.x
 #define TSDB_CODE_VND_INVALID_VGROUP_ID         TAOS_DEF_ERROR_CODE(0, 0x0503)
-// #define TSDB_CODE_VND_INIT_FAILED            TAOS_DEF_ERROR_CODE(0, 0x0504) // 2.x
+#define TSDB_CODE_VND_INIT_FAILED               TAOS_DEF_ERROR_CODE(0, 0x0504)
 // #define TSDB_CODE_VND_NO_DISKSPACE           TAOS_DEF_ERROR_CODE(0, 0x0505) // 2.x
 // #define TSDB_CODE_VND_NO_DISK_PERMISSIONS    TAOS_DEF_ERROR_CODE(0, 0x0506) // 2.x
 // #define TSDB_CODE_VND_NO_SUCH_FILE_OR_DIR    TAOS_DEF_ERROR_CODE(0, 0x0507) // 2.x

include/util/tarray.h

@@ -22,7 +22,7 @@
 extern "C" {
 #endif
 
-#define TARRAY_MIN_SIZE               8
+#define TARRAY_MIN_SIZE               4
 #define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
 #define TARRAY_ELEM_IDX(array, ele)   (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
 
@@ -138,7 +138,7 @@ size_t taosArrayGetSize(const SArray* pArray);
  * @param index
  * @param pData
  */
-void* taosArrayInsert(SArray* pArray, size_t index, void* pData);
+void* taosArrayInsert(SArray* pArray, size_t index, const void* pData);
 
 /**
  * set data in array

include/util/tarray2.h

+/*
+ * 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 "talgo.h"
+
+#ifndef _TD_UTIL_TARRAY2_H_
+#define _TD_UTIL_TARRAY2_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// a: a
+// e: element
+// ep: element pointer
+// cmp: compare function
+// idx: index
+// cb: callback function
+
+#define TARRAY2(TYPE) \
+  struct {            \
+    int32_t size;     \
+    int32_t capacity; \
+    TYPE   *data;     \
+  }
+
+typedef void (*TArray2Cb)(void *);
+
+#define TARRAY2_SIZE(a)       ((a)->size)
+#define TARRAY2_CAPACITY(a)   ((a)->capacity)
+#define TARRAY2_DATA(a)       ((a)->data)
+#define TARRAY2_GET(a, i)     ((a)->data[i])
+#define TARRAY2_GET_PTR(a, i) ((a)->data + i)
+#define TARRAY2_FIRST(a)      ((a)->data[0])
+#define TARRAY2_LAST(a)       ((a)->data[(a)->size - 1])
+#define TARRAY2_DATA_LEN(a)   ((a)->size * sizeof(((a)->data[0])))
+
+static FORCE_INLINE int32_t tarray2_make_room(void *arr, int32_t expSize, int32_t eleSize) {
+  TARRAY2(void) *a = arr;
+
+  int32_t capacity = (a->capacity > 0) ? (a->capacity << 1) : 32;
+  while (capacity < expSize) {
+    capacity <<= 1;
+  }
+  void *p = taosMemoryRealloc(a->data, capacity * eleSize);
+  if (p == NULL) return TSDB_CODE_OUT_OF_MEMORY;
+  a->capacity = capacity;
+  a->data = p;
+  return 0;
+}
+
+static FORCE_INLINE int32_t tarray2InsertBatch(void *arr, int32_t idx, const void *elePtr, int32_t numEle,
+                                               int32_t eleSize) {
+  TARRAY2(uint8_t) *a = arr;
+
+  int32_t ret = 0;
+  if (a->size + numEle > a->capacity) {
+    ret = tarray2_make_room(a, a->size + numEle, eleSize);
+  }
+  if (ret == 0) {
+    if (idx < a->size) {
+      memmove(a->data + (idx + numEle) * eleSize, a->data + idx * eleSize, (a->size - idx) * eleSize);
+    }
+    memcpy(a->data + idx * eleSize, elePtr, numEle * eleSize);
+    a->size += numEle;
+  }
+  return ret;
+}
+
+static FORCE_INLINE void *tarray2Search(void *arr, const void *elePtr, int32_t eleSize, __compar_fn_t compar,
+                                        int32_t flag) {
+  TARRAY2(void) *a = arr;
+  return taosbsearch(elePtr, a->data, a->size, eleSize, compar, flag);
+}
+
+static FORCE_INLINE int32_t tarray2SearchIdx(void *arr, const void *elePtr, int32_t eleSize, __compar_fn_t compar,
+                                             int32_t flag) {
+  TARRAY2(void) *a = arr;
+  void *p = taosbsearch(elePtr, a->data, a->size, eleSize, compar, flag);
+  if (p == NULL) {
+    return -1;
+  } else {
+    return (int32_t)(((uint8_t *)p - (uint8_t *)a->data) / eleSize);
+  }
+}
+
+static FORCE_INLINE int32_t tarray2SortInsert(void *arr, const void *elePtr, int32_t eleSize, __compar_fn_t compar) {
+  TARRAY2(void) *a = arr;
+  int32_t idx = tarray2SearchIdx(arr, elePtr, eleSize, compar, TD_GT);
+  return tarray2InsertBatch(arr, idx < 0 ? a->size : idx, elePtr, 1, eleSize);
+}
+
+#define TARRAY2_INIT_EX(a, size_, capacity_, data_) \
+  do {                                              \
+    (a)->size = (size_);                            \
+    (a)->capacity = (capacity_);                    \
+    (a)->data = (data_);                            \
+  } while (0)
+
+#define TARRAY2_INIT(a) TARRAY2_INIT_EX(a, 0, 0, NULL)
+
+#define TARRAY2_CLEAR(a, cb)                    \
+  do {                                          \
+    if ((cb) && (a)->size > 0) {                \
+      TArray2Cb cb_ = (TArray2Cb)(cb);          \
+      for (int32_t i = 0; i < (a)->size; ++i) { \
+        cb_((a)->data + i);                     \
+      }                                         \
+    }                                           \
+    (a)->size = 0;                              \
+  } while (0)
+
+#define TARRAY2_DESTROY(a, cb)   \
+  do {                           \
+    TARRAY2_CLEAR(a, cb);        \
+    if ((a)->data) {             \
+      taosMemoryFree((a)->data); \
+      (a)->data = NULL;          \
+    }                            \
+    (a)->capacity = 0;           \
+  } while (0)
+
+#define TARRAY2_INSERT_PTR(a, idx, ep) tarray2InsertBatch(a, idx, ep, 1, sizeof((a)->data[0]))
+#define TARRAY2_APPEND_PTR(a, ep)      tarray2InsertBatch(a, (a)->size, ep, 1, sizeof((a)->data[0]))
+#define TARRAY2_APPEND_BATCH(a, ep, n) tarray2InsertBatch(a, (a)->size, ep, n, sizeof((a)->data[0]))
+#define TARRAY2_APPEND(a, e)           TARRAY2_APPEND_PTR(a, &(e))
+
+// return (TYPE *)
+#define TARRAY2_SEARCH(a, ep, cmp, flag) tarray2Search(a, ep, sizeof(((a)->data[0])), (__compar_fn_t)cmp, flag)
+
+#define TARRAY2_SEARCH_IDX(a, ep, cmp, flag) tarray2SearchIdx(a, ep, sizeof(((a)->data[0])), (__compar_fn_t)cmp, flag)
+
+#define TARRAY2_SORT_INSERT(a, e, cmp)    tarray2SortInsert(a, &(e), sizeof(((a)->data[0])), (__compar_fn_t)cmp)
+#define TARRAY2_SORT_INSERT_P(a, ep, cmp) tarray2SortInsert(a, ep, sizeof(((a)->data[0])), (__compar_fn_t)cmp)
+
+#define TARRAY2_REMOVE(a, idx, cb)                                                                       \
+  do {                                                                                                   \
+    if ((idx) < (a)->size) {                                                                             \
+      if (cb) {                                                                                          \
+        TArray2Cb cb_ = (TArray2Cb)(cb);                                                                 \
+        cb_((a)->data + (idx));                                                                          \
+      }                                                                                                  \
+      if ((idx) < (a)->size - 1) {                                                                       \
+        memmove((a)->data + (idx), (a)->data + (idx) + 1, sizeof((*(a)->data)) * ((a)->size - (idx)-1)); \
+      }                                                                                                  \
+      (a)->size--;                                                                                       \
+    }                                                                                                    \
+  } while (0)
+
+#define TARRAY2_FOREACH(a, e)         for (int32_t __i = 0; __i < (a)->size && ((e) = (a)->data[__i], 1); __i++)
+#define TARRAY2_FOREACH_REVERSE(a, e) for (int32_t __i = (a)->size - 1; __i >= 0 && ((e) = (a)->data[__i], 1); __i--)
+#define TARRAY2_FOREACH_PTR(a, ep)    for (int32_t __i = 0; __i < (a)->size && ((ep) = &(a)->data[__i], 1); __i++)
+#define TARRAY2_FOREACH_PTR_REVERSE(a, ep) \
+  for (int32_t __i = (a)->size - 1; __i >= 0 && ((ep) = &(a)->data[__i], 1); __i--)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_UTIL_TARRAY2_H_*/

include/util/tconfig.h

@@ -50,10 +50,16 @@ typedef enum {
   CFG_DTYPE_TIMEZONE
 } ECfgDataType;
 
+typedef enum {
+  CFG_SCOPE_SERVER,
+  CFG_SCOPE_CLIENT,
+  CFG_SCOPE_BOTH
+} ECfgScopeType;
+
 typedef struct SConfigItem {
   ECfgSrcType  stype;
   ECfgDataType dtype;
-  bool         tsc;
+  int8_t       scope;
   char         *name;
   union {
     bool    bval;
@@ -92,20 +98,21 @@ int32_t      cfgGetSize(SConfig *pCfg);
 SConfigItem *cfgGetItem(SConfig *pCfg, const char *name);
 int32_t      cfgSetItem(SConfig *pCfg, const char *name, const char *value, ECfgSrcType stype);
 
-int32_t cfgAddBool(SConfig *pCfg, const char *name, bool defaultVal, bool tsc);
-int32_t cfgAddInt32(SConfig *pCfg, const char *name, int32_t defaultVal, int64_t minval, int64_t maxval, bool tsc);
-int32_t cfgAddInt64(SConfig *pCfg, const char *name, int64_t defaultVal, int64_t minval, int64_t maxval, bool tsc);
-int32_t cfgAddFloat(SConfig *pCfg, const char *name, float defaultVal, double minval, double maxval, bool tsc);
-int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, bool tsc);
-int32_t cfgAddDir(SConfig *pCfg, const char *name, const char *defaultVal, bool tsc);
-int32_t cfgAddLocale(SConfig *pCfg, const char *name, const char *defaultVal);
-int32_t cfgAddCharset(SConfig *pCfg, const char *name, const char *defaultVal);
-int32_t cfgAddTimezone(SConfig *pCfg, const char *name, const char *defaultVal);
+int32_t cfgAddBool(SConfig *pCfg, const char *name, bool defaultVal, int8_t scope);
+int32_t cfgAddInt32(SConfig *pCfg, const char *name, int32_t defaultVal, int64_t minval, int64_t maxval, int8_t scope);
+int32_t cfgAddInt64(SConfig *pCfg, const char *name, int64_t defaultVal, int64_t minval, int64_t maxval, int8_t scope);
+int32_t cfgAddFloat(SConfig *pCfg, const char *name, float defaultVal, double minval, double maxval, int8_t scope);
+int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);
+int32_t cfgAddDir(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);
+int32_t cfgAddLocale(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);
+int32_t cfgAddCharset(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);
+int32_t cfgAddTimezone(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);
 
 const char *cfgStypeStr(ECfgSrcType type);
 const char *cfgDtypeStr(ECfgDataType type);
 
 void cfgDumpItemValue(SConfigItem *pItem, char *buf, int32_t bufSize, int32_t *pLen);
+void cfgDumpItemScope(SConfigItem *pItem, char *buf, int32_t bufSize, int32_t *pLen);
 
 void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump);
 

include/util/tdef.h

@@ -370,7 +370,7 @@ typedef enum ELogicConditionType {
 #define TSDB_DEFAULT_DB_SCHEMALESS      TSDB_DB_SCHEMALESS_OFF
 #define TSDB_MIN_STT_TRIGGER            1
 #define TSDB_MAX_STT_TRIGGER            16
-#define TSDB_DEFAULT_SST_TRIGGER        1
+#define TSDB_DEFAULT_SST_TRIGGER        2
 #define TSDB_MIN_HASH_PREFIX            (2 - TSDB_TABLE_NAME_LEN)
 #define TSDB_MAX_HASH_PREFIX            (TSDB_TABLE_NAME_LEN - 2)
 #define TSDB_DEFAULT_HASH_PREFIX        0
@@ -492,6 +492,7 @@ enum {
 
 #define TSDB_CONFIG_OPTION_LEN 32
 #define TSDB_CONFIG_VALUE_LEN  64
+#define TSDB_CONFIG_SCOPE_LEN  8
 #define TSDB_CONFIG_NUMBER     8
 
 #define QUERY_ID_SIZE      20

include/util/tlist.h

@@ -241,6 +241,54 @@ void       tdListNodeGetData(SList *list, SListNode *node, void *target);
 void       tdListInitIter(SList *list, SListIter *pIter, TD_LIST_DIRECTION_T direction);
 SListNode *tdListNext(SListIter *pIter);
 
+// macros ====================================================================================
+
+// q: for queue
+// n: for node
+// m: for member
+
+#define LISTD(TYPE)    \
+  struct {             \
+    TYPE *next, *prev; \
+  }
+
+#define LISTD_NEXT(n, m)      ((n)->m.next)
+#define LISTD_PREV(n, m)      ((n)->m.prev)
+#define LISTD_INIT(q, m)      (LISTD_NEXT(q, m) = LISTD_PREV(q, m) = (q))
+#define LISTD_HEAD(q, m)      (LISTD_NEXT(q, m))
+#define LISTD_TAIL(q, m)      (LISTD_PREV(q, m))
+#define LISTD_PREV_NEXT(n, m) (LISTD_NEXT(LISTD_PREV(n, m), m))
+#define LISTD_NEXT_PREV(n, m) (LISTD_PREV(LISTD_NEXT(n, m), m))
+
+#define LISTD_INSERT_HEAD(q, n, m)       \
+  do {                                   \
+    LISTD_NEXT(n, m) = LISTD_NEXT(q, m); \
+    LISTD_PREV(n, m) = (q);              \
+    LISTD_NEXT_PREV(n, m) = (n);         \
+    LISTD_NEXT(q, m) = (n);              \
+  } while (0)
+
+#define LISTD_INSERT_TAIL(q, n, m)       \
+  do {                                   \
+    LISTD_NEXT(n, m) = (q);              \
+    LISTD_PREV(n, m) = LISTD_PREV(q, m); \
+    LISTD_PREV_NEXT(n, m) = (n);         \
+    LISTD_PREV(q, m) = (n);              \
+  } while (0)
+
+#define LISTD_REMOVE(n, m)                    \
+  do {                                        \
+    LISTD_PREV_NEXT(n, m) = LISTD_NEXT(n, m); \
+    LISTD_NEXT_PREV(n, m) = LISTD_PREV(n, m); \
+  } while (0)
+
+#define LISTD_FOREACH(q, n, m)         for ((n) = LISTD_HEAD(q, m); (n) != (q); (n) = LISTD_NEXT(n, m))
+#define LISTD_FOREACH_REVERSE(q, n, m) for ((n) = LISTD_TAIL(q, m); (n) != (q); (n) = LISTD_PREV(n, m))
+#define LISTD_FOREACH_SAFE(q, n, t, m) \
+  for ((n) = LISTD_HEAD(q, m), (t) = LISTD_NEXT(n, m); (n) != (q); (n) = (t), (t) = LISTD_NEXT(n, m))
+#define LISTD_FOREACH_REVERSE_SAFE(q, n, t, m) \
+  for ((n) = LISTD_TAIL(q, m), (t) = LISTD_PREV(n, m); (n) != (q); (n) = (t), (t) = LISTD_PREV(n, m))
+
 #ifdef __cplusplus
 }
 #endif

include/util/trbtree.h

@@ -39,7 +39,7 @@ void         tRBTreeDrop(SRBTree *pTree, SRBTreeNode *z);
 SRBTreeNode *tRBTreeDropByKey(SRBTree *pTree, void *pKey);
 SRBTreeNode *tRBTreeDropMin(SRBTree *pTree);
 SRBTreeNode *tRBTreeDropMax(SRBTree *pTree);
-SRBTreeNode *tRBTreeGet(SRBTree *pTree, const SRBTreeNode *pKeyNode);
+SRBTreeNode *tRBTreeGet(const SRBTree *pTree, const SRBTreeNode *pKeyNode);
 
 // SRBTreeIter =============================================
 #define tRBTreeIterCreate(tree, ascend) \
@@ -68,7 +68,7 @@ struct SRBTree {
 
 struct SRBTreeIter {
   int8_t         asc;
-  SRBTree     *pTree;
+  const SRBTree *pTree;
   SRBTreeNode   *pNode;
 };
 

include/util/tutil.h

@@ -29,7 +29,7 @@ extern "C" {
 int32_t strdequote(char *src);
 size_t  strtrim(char *src);
 char   *strnchr(const char *haystack, char needle, int32_t len, bool skipquote);
-TdUcs4* wcsnchr(const TdUcs4* haystack, TdUcs4 needle, size_t len);
+TdUcs4 *wcsnchr(const TdUcs4 *haystack, TdUcs4 needle, size_t len);
 
 char  **strsplit(char *src, const char *delim, int32_t *num);
 char   *strtolower(char *dst, const char *src);
@@ -81,7 +81,8 @@ static FORCE_INLINE void taosEncryptPass_c(uint8_t *inBuf, size_t len, char *tar
 
 static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen, int32_t method, int32_t prefix,
                                              int32_t suffix) {
-  if ((prefix == 0 && suffix == 0) || (tblen <= (prefix + suffix)) || (tblen <= -1 * (prefix + suffix)) || prefix * suffix < 0) {
+  if ((prefix == 0 && suffix == 0) || (tblen <= (prefix + suffix)) || (tblen <= -1 * (prefix + suffix)) ||
+      prefix * suffix < 0) {
     return MurmurHash3_32(tbname, tblen);
   } else if (prefix > 0 || suffix > 0) {
     return MurmurHash3_32(tbname + prefix, tblen - prefix - suffix);
@@ -94,20 +95,33 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
     }
     if (suffix < 0) {
       strncpy(tbName + offset, tbname + tblen + suffix, -1 * suffix);
-      offset += -1 *suffix;
+      offset += -1 * suffix;
     }
     return MurmurHash3_32(tbName, offset);
   }
 }
 
 #define TSDB_CHECK_CODE(CODE, LINO, LABEL) \
-  if (CODE) {                              \
+  do {                                     \
+    if ((CODE)) {                          \
       LINO = __LINE__;                     \
       goto LABEL;                          \
+    }                                      \
+  } while (0)
+
+#define TSDB_CHECK_NULL(ptr, CODE, LINO, LABEL, ERRNO) \
+  if ((ptr) == NULL) {                                 \
+    (CODE) = (ERRNO);                                  \
+    (LINO) = __LINE__;                                 \
+    goto LABEL;                                        \
   }
 
+#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
+
 #define VND_CHECK_CODE(CODE, LINO, LABEL) TSDB_CHECK_CODE(CODE, LINO, LABEL)
 
+#define TCONTAINER_OF(ptr, type, member) ((type *)((char *)(ptr)-offsetof(type, member)))
+
 #ifdef __cplusplus
 }
 #endif

source/client/inc/clientInt.h

@@ -46,9 +46,10 @@ enum {
   RES_TYPE__TMQ_METADATA,
 };
 
-#define SHOW_VARIABLES_RESULT_COLS       2
+#define SHOW_VARIABLES_RESULT_COLS       3
 #define SHOW_VARIABLES_RESULT_FIELD1_LEN (TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE)
 #define SHOW_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
+#define SHOW_VARIABLES_RESULT_FIELD3_LEN (TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE)
 
 #define TD_RES_QUERY(res)        (*(int8_t*)res == RES_TYPE__QUERY)
 #define TD_RES_TMQ(res)          (*(int8_t*)res == RES_TYPE__TMQ)

source/client/src/clientMsgHandler.c

@@ -428,13 +428,16 @@ static int32_t buildShowVariablesBlock(SArray* pVars, SSDataBlock** block) {
   SColumnInfoData infoData = {0};
   infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
   infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD1_LEN;
-
   taosArrayPush(pBlock->pDataBlock, &infoData);
 
   infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
   infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD2_LEN;
   taosArrayPush(pBlock->pDataBlock, &infoData);
 
+  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
+  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD3_LEN;
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
   int32_t numOfCfg = taosArrayGetSize(pVars);
   blockDataEnsureCapacity(pBlock, numOfCfg);
 
@@ -450,6 +453,11 @@ static int32_t buildShowVariablesBlock(SArray* pVars, SSDataBlock** block) {
     STR_WITH_MAXSIZE_TO_VARSTR(value, pInfo->value, TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE);
     pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
     colDataSetVal(pColInfo, i, value, false);
+
+    char scope[TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE] = {0};
+    STR_WITH_MAXSIZE_TO_VARSTR(scope, pInfo->scope, TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE);
+    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
+    colDataSetVal(pColInfo, i, scope, false);
   }
 
   pBlock->info.rows = numOfCfg;

source/common/src/systable.c

@@ -271,6 +271,7 @@ static const SSysDbTableSchema variablesSchema[] = {
     {.name = "dnode_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "name", .bytes = TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
     {.name = "value", .bytes = TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "scope", .bytes = TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
 };
 
 static const SSysDbTableSchema topicSchema[] = {

source/common/src/tdatablock.c

@@ -699,7 +699,10 @@ int32_t blockDataToBuf(char* buf, const SSDataBlock* pBlock) {
         pStart += colSize;
       }
     } else {
+      if (dataSize != 0) {
+        // ubsan reports error if pCol->pData==NULL && dataSize==0
         memcpy(pStart, pCol->pData, dataSize);
+      }
       pStart += dataSize;
     }
   }
@@ -751,8 +754,10 @@ int32_t blockDataFromBuf(SSDataBlock* pBlock, const char* buf) {
         return TSDB_CODE_FAILED;
       }
     }
-
+    if (colLength != 0) {
+      // ubsan reports error if colLength==0 && pCol->pData == 0
       memcpy(pCol->pData, pStart, colLength);
+    }
     pStart += colLength;
   }
 

source/common/src/tdataformat.c

@@ -2245,15 +2245,18 @@ static int32_t tColDataUpdateValue72(SColData *pColData, uint8_t *pData, uint32_
   }
   return 0;
 }
+static FORCE_INLINE int32_t tColDataUpdateNothing(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
+  return 0;
+}
 static int32_t (*tColDataUpdateValueImpl[8][3])(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) = {
     {NULL, NULL, NULL},                                                     // 0
-    {tColDataUpdateValue10, NULL, tColDataUpdateValue12},  // HAS_NONE
-    {tColDataUpdateValue20, NULL, NULL},                   // HAS_NULL
-    {tColDataUpdateValue30, NULL, tColDataUpdateValue32},  // HAS_NULL|HAS_NONE
-    {tColDataUpdateValue40, NULL, tColDataUpdateValue42},  // HAS_VALUE
-    {tColDataUpdateValue50, NULL, tColDataUpdateValue52},  // HAS_VALUE|HAS_NONE
-    {tColDataUpdateValue60, NULL, tColDataUpdateValue62},  // HAS_VALUE|HAS_NULL
-    {tColDataUpdateValue70, NULL, tColDataUpdateValue72},  // HAS_VALUE|HAS_NULL|HAS_NONE
+    {tColDataUpdateValue10, tColDataUpdateNothing, tColDataUpdateValue12},  // HAS_NONE
+    {tColDataUpdateValue20, tColDataUpdateNothing, tColDataUpdateNothing},  // HAS_NULL
+    {tColDataUpdateValue30, tColDataUpdateNothing, tColDataUpdateValue32},  // HAS_NULL|HAS_NONE
+    {tColDataUpdateValue40, tColDataUpdateNothing, tColDataUpdateValue42},  // HAS_VALUE
+    {tColDataUpdateValue50, tColDataUpdateNothing, tColDataUpdateValue52},  // HAS_VALUE|HAS_NONE
+    {tColDataUpdateValue60, tColDataUpdateNothing, tColDataUpdateValue62},  // HAS_VALUE|HAS_NULL
+    {tColDataUpdateValue70, tColDataUpdateNothing, tColDataUpdateValue72},  // HAS_VALUE|HAS_NULL|HAS_NONE
 
     //    VALUE             NONE        NULL
 };

source/common/src/tmsg.c

@@ -3484,12 +3484,14 @@ int32_t tDeserializeSShowVariablesReq(void *buf, int32_t bufLen, SShowVariablesR
 int32_t tEncodeSVariablesInfo(SEncoder *pEncoder, SVariablesInfo *pInfo) {
   if (tEncodeCStr(pEncoder, pInfo->name) < 0) return -1;
   if (tEncodeCStr(pEncoder, pInfo->value) < 0) return -1;
+  if (tEncodeCStr(pEncoder, pInfo->scope) < 0) return -1;
   return 0;
 }
 
 int32_t tDecodeSVariablesInfo(SDecoder *pDecoder, SVariablesInfo *pInfo) {
   if (tDecodeCStrTo(pDecoder, pInfo->name) < 0) return -1;
   if (tDecodeCStrTo(pDecoder, pInfo->value) < 0) return -1;
+  if (tDecodeCStrTo(pDecoder, pInfo->scope) < 0) return -1;
   return 0;
 }
 

source/dnode/mgmt/mgmt_dnode/src/dmHandle.c

@@ -265,6 +265,12 @@ int32_t dmAppendVariablesToBlock(SSDataBlock *pBlock, int32_t dnodeId) {
     pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
     colDataSetVal(pColInfo, i, value, false);
 
+    char scope[TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE] = {0};
+    cfgDumpItemScope(pItem, &scope[VARSTR_HEADER_SIZE], TSDB_CONFIG_SCOPE_LEN, &valueLen);
+    varDataSetLen(scope, valueLen);
+    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
+    colDataSetVal(pColInfo, i, scope, false);
+
     numOfRows++;
   }
 

source/dnode/mgmt/mgmt_snode/src/smHandle.c

@@ -79,6 +79,7 @@ SArray *smGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECK, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECK_RSP, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_SCAN_HISTORY_FINISH, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_STREAM_SCAN_HISTORY_FINISH_RSP, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
 
   code = 0;
 _OVER:

source/dnode/mgmt/mgmt_vnode/inc/vmInt.h

@@ -46,6 +46,7 @@ typedef struct {
   int32_t vgId;
   int32_t vgVersion;
   int8_t  dropped;
+  int32_t diskPrimary;
   int32_t toVgId;
   char    path[PATH_MAX + 20];
 } SWrapperCfg;
@@ -56,6 +57,7 @@ typedef struct {
   int32_t       refCount;
   int8_t        dropped;
   int8_t        disable;
+  int32_t       diskPrimary;
   int32_t       toVgId;
   char         *path;
   SVnode       *pImpl;
@@ -81,6 +83,7 @@ typedef struct {
 } SVnodeThread;
 
 // vmInt.c
+int32_t    vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId);
 SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId);
 void       vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode);
 int32_t    vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl);

source/dnode/mgmt/mgmt_vnode/src/vmFile.c

@@ -71,6 +71,8 @@ static int32_t vmDecodeVnodeList(SJson *pJson, SVnodeMgmt *pMgmt, SWrapperCfg **
     if (code < 0) goto _OVER;
     tjsonGetInt32ValueFromDouble(vnode, "vgVersion", pCfg->vgVersion, code);
     if (code < 0) goto _OVER;
+    tjsonGetInt32ValueFromDouble(vnode, "diskPrimary", pCfg->diskPrimary, code);
+    if (code < 0) goto _OVER;
     tjsonGetInt32ValueFromDouble(vnode, "toVgId", pCfg->toVgId, code);
     if (code < 0) goto _OVER;
 
@@ -167,6 +169,7 @@ static int32_t vmEncodeVnodeList(SJson *pJson, SVnodeObj **ppVnodes, int32_t num
     if (tjsonAddDoubleToObject(vnode, "vgId", pVnode->vgId) < 0) return -1;
     if (tjsonAddDoubleToObject(vnode, "dropped", pVnode->dropped) < 0) return -1;
     if (tjsonAddDoubleToObject(vnode, "vgVersion", pVnode->vgVersion) < 0) return -1;
+    if (tjsonAddDoubleToObject(vnode, "diskPrimary", pVnode->diskPrimary) < 0) return -1;
     if (pVnode->toVgId && tjsonAddDoubleToObject(vnode, "toVgId", pVnode->toVgId) < 0) return -1;
     if (tjsonAddItemToArray(vnodes, vnode) < 0) return -1;
   }

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -263,16 +263,19 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return 0;
   }
 
+  wrapperCfg.diskPrimary = vmAllocPrimaryDisk(pMgmt, vnodeCfg.vgId);
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
+
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
 
-  if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
+  if (vnodeCreate(path, &vnodeCfg, diskPrimary, pMgmt->pTfs) < 0) {
     tFreeSCreateVnodeReq(&req);
     dError("vgId:%d, failed to create vnode since %s", req.vgId, terrstr());
     code = terrno;
     goto _OVER;
   }
 
-  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode since %s", req.vgId, terrstr());
     code = terrno;
@@ -403,21 +406,23 @@ int32_t vmProcessAlterVnodeTypeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
       .dropped = pVnode->dropped,
       .vgId = pVnode->vgId,
       .vgVersion = pVnode->vgVersion,
+      .diskPrimary = pVnode->diskPrimary,
   };
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
   vmCloseVnode(pMgmt, pVnode, false);
 
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
   char path[TSDB_FILENAME_LEN] = {0};
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vgId);
 
   dInfo("vgId:%d, start to alter vnode replica at %s", vgId, path);
-  if (vnodeAlterReplica(path, &req, pMgmt->pTfs) < 0) {
+  if (vnodeAlterReplica(path, &req, diskPrimary, pMgmt->pTfs) < 0) {
     dError("vgId:%d, failed to alter vnode at %s since %s", vgId, path, terrstr());
     return -1;
   }
 
   dInfo("vgId:%d, begin to open vnode", vgId);
-  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode at %s since %s", vgId, path, terrstr());
     return -1;
@@ -490,6 +495,7 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
       .dropped = pVnode->dropped,
       .vgId = dstVgId,
       .vgVersion = pVnode->vgVersion,
+      .diskPrimary = pVnode->diskPrimary,
   };
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
 
@@ -503,19 +509,20 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   dInfo("vgId:%d, close vnode", srcVgId);
   vmCloseVnode(pMgmt, pVnode, true);
 
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
   char srcPath[TSDB_FILENAME_LEN] = {0};
   char dstPath[TSDB_FILENAME_LEN] = {0};
   snprintf(srcPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, srcVgId);
   snprintf(dstPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, dstVgId);
 
   dInfo("vgId:%d, alter vnode hashrange at %s", srcVgId, srcPath);
-  if (vnodeAlterHashRange(srcPath, dstPath, &req, pMgmt->pTfs) < 0) {
+  if (vnodeAlterHashRange(srcPath, dstPath, &req, diskPrimary, pMgmt->pTfs) < 0) {
     dError("vgId:%d, failed to alter vnode hashrange since %s", srcVgId, terrstr());
     return -1;
   }
 
   dInfo("vgId:%d, open vnode", dstVgId);
-  SVnode *pImpl = vnodeOpen(dstPath, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(dstPath, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode at %s since %s", dstVgId, dstPath, terrstr());
     return -1;
@@ -602,21 +609,23 @@ int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
       .dropped = pVnode->dropped,
       .vgId = pVnode->vgId,
       .vgVersion = pVnode->vgVersion,
+      .diskPrimary = pVnode->diskPrimary,
   };
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
   vmCloseVnode(pMgmt, pVnode, false);
 
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
   char path[TSDB_FILENAME_LEN] = {0};
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vgId);
 
   dInfo("vgId:%d, start to alter vnode replica at %s", vgId, path);
-  if (vnodeAlterReplica(path, &alterReq, pMgmt->pTfs) < 0) {
+  if (vnodeAlterReplica(path, &alterReq, diskPrimary, pMgmt->pTfs) < 0) {
     dError("vgId:%d, failed to alter vnode at %s since %s", vgId, path, terrstr());
     return -1;
   }
 
   dInfo("vgId:%d, begin to open vnode", vgId);
-  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode at %s since %s", vgId, path, terrstr());
     return -1;
@@ -731,6 +740,7 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_SCAN_HISTORY_FINISH, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_STREAM_SCAN_HISTORY_FINISH_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TRANSFER_STATE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECK, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECK_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;

source/dnode/mgmt/mgmt_vnode/src/vmInt.c

@@ -15,8 +15,64 @@
 
 #define _DEFAULT_SOURCE
 #include "vmInt.h"
+#include "tfs.h"
 #include "vnd.h"
 
+int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {
+  STfs   *pTfs = pMgmt->pTfs;
+  int32_t diskId = 0;
+  if (!pTfs) {
+    return diskId;
+  }
+
+  // search fs
+  char vnodePath[TSDB_FILENAME_LEN] = {0};
+  snprintf(vnodePath, TSDB_FILENAME_LEN - 1, "vnode%svnode%d", TD_DIRSEP, vgId);
+  char fname[TSDB_FILENAME_LEN] = {0};
+  char fnameTmp[TSDB_FILENAME_LEN] = {0};
+  snprintf(fname, TSDB_FILENAME_LEN - 1, "%s%s%s", vnodePath, TD_DIRSEP, VND_INFO_FNAME);
+  snprintf(fnameTmp, TSDB_FILENAME_LEN - 1, "%s%s%s", vnodePath, TD_DIRSEP, VND_INFO_FNAME_TMP);
+
+  diskId = tfsSearch(pTfs, 0, fname);
+  if (diskId >= 0) {
+    return diskId;
+  }
+  diskId = tfsSearch(pTfs, 0, fnameTmp);
+  if (diskId >= 0) {
+    return diskId;
+  }
+
+  // alloc
+  int32_t     disks[TFS_MAX_DISKS_PER_TIER] = {0};
+  int32_t     numOfVnodes = 0;
+  SVnodeObj **ppVnodes = vmGetVnodeListFromHash(pMgmt, &numOfVnodes);
+  for (int32_t v = 0; v < numOfVnodes; v++) {
+    SVnodeObj *pVnode = ppVnodes[v];
+    disks[pVnode->diskPrimary] += 1;
+  }
+
+  int32_t minVal = INT_MAX;
+  int32_t ndisk = tfsGetDisksAtLevel(pTfs, 0);
+  diskId = 0;
+  for (int32_t id = 0; id < ndisk; id++) {
+    if (minVal > disks[id]) {
+      minVal = disks[id];
+      diskId = id;
+    }
+  }
+
+  for (int32_t i = 0; i < numOfVnodes; ++i) {
+    if (ppVnodes == NULL || ppVnodes[i] == NULL) continue;
+    vmReleaseVnode(pMgmt, ppVnodes[i]);
+  }
+  if (ppVnodes != NULL) {
+    taosMemoryFree(ppVnodes);
+  }
+
+  dInfo("vgId:%d, alloc disk:%d of level 0. ndisk:%d, vnodes: %d", vgId, diskId, ndisk, numOfVnodes);
+  return diskId;
+}
+
 SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
   SVnodeObj *pVnode = NULL;
 
@@ -52,6 +108,7 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
 
   pVnode->vgId = pCfg->vgId;
   pVnode->vgVersion = pCfg->vgVersion;
+  pVnode->diskPrimary = pCfg->diskPrimary;
   pVnode->refCount = 0;
   pVnode->dropped = 0;
   pVnode->path = taosStrdup(pCfg->path);
@@ -169,7 +226,8 @@ static int32_t vmRestoreVgroupId(SWrapperCfg *pCfg, STfs *pTfs) {
   snprintf(srcPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, srcVgId);
   snprintf(dstPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, dstVgId);
 
-  int32_t vgId = vnodeRestoreVgroupId(srcPath, dstPath, srcVgId, dstVgId, pTfs);
+  int32_t diskPrimary = pCfg->diskPrimary;
+  int32_t vgId = vnodeRestoreVgroupId(srcPath, dstPath, srcVgId, dstVgId, diskPrimary, pTfs);
   if (vgId <= 0) {
     dError("vgId:%d, failed to restore vgroup id. srcPath: %s", pCfg->vgId, srcPath);
     return -1;
@@ -205,11 +263,12 @@ static void *vmOpenVnodeInThread(void *param) {
       pThread->updateVnodesList = true;
     }
 
+    int32_t diskPrimary = pCfg->diskPrimary;
     snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, pCfg->vgId);
 
-    SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+    SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
     if (pImpl == NULL) {
-      dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
+      dError("vgId:%d, failed to open vnode by thread:%d since %s", pCfg->vgId, pThread->threadIndex, terrstr());
       pThread->failed++;
       continue;
     }
@@ -296,6 +355,7 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) {
 
   if (pMgmt->state.openVnodes != pMgmt->state.totalVnodes) {
     dError("there are total vnodes:%d, opened:%d", pMgmt->state.totalVnodes, pMgmt->state.openVnodes);
+    terrno = TSDB_CODE_VND_INIT_FAILED;
     return -1;
   }
 
@@ -518,7 +578,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   tmsgReportStartup("vnode-worker", "initialized");
 
   if (vmOpenVnodes(pMgmt) != 0) {
-    dError("failed to open vnode since %s", terrstr());
+    dError("failed to open all vnodes since %s", terrstr());
     goto _OVER;
   }
   tmsgReportStartup("vnode-vnodes", "initialized");

source/dnode/mnode/impl/inc/mndStb.h

@@ -39,6 +39,7 @@ int32_t mndBuildSMCreateStbRsp(SMnode *pMnode, char *dbFName, char *stbFName, vo
 
 void mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst);
 void mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst);
+void mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst);
 void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize);
 
 const char *mndGetStbStr(const char *src);

source/dnode/mnode/impl/src/mndDnode.c

@@ -70,6 +70,8 @@ static void    mndCancelGetNextConfig(SMnode *pMnode, void *pIter);
 static int32_t mndRetrieveDnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
 static void    mndCancelGetNextDnode(SMnode *pMnode, void *pIter);
 
+static int32_t mndMCfgGetValInt32(SMCfgDnodeReq *pInMCfgReq, int32_t opLen, int32_t *pOutValue);
+
 int32_t mndInitDnode(SMnode *pMnode) {
   SSdbTable table = {
       .sdbType = SDB_DNODE,
@@ -783,18 +785,22 @@ static int32_t mndProcessShowVariablesReq(SRpcMsg *pReq) {
 
   strcpy(info.name, "statusInterval");
   snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%d", tsStatusInterval);
+  strcpy(info.scope, "server");
   taosArrayPush(rsp.variables, &info);
 
   strcpy(info.name, "timezone");
   snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%s", tsTimezoneStr);
+  strcpy(info.scope, "both");
   taosArrayPush(rsp.variables, &info);
 
   strcpy(info.name, "locale");
   snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%s", tsLocale);
+  strcpy(info.scope, "both");
   taosArrayPush(rsp.variables, &info);
 
   strcpy(info.name, "charset");
   snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%s", tsCharset);
+  strcpy(info.scope, "both");
   taosArrayPush(rsp.variables, &info);
 
   int32_t rspLen = tSerializeSShowVariablesRsp(NULL, 0, &rsp);
@@ -1061,22 +1067,19 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
     strcpy(dcfgReq.config, "monitor");
     snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
   } else if (strncasecmp(cfgReq.config, "keeptimeoffset", 14) == 0) {
-    if (' ' != cfgReq.config[14] && 0 != cfgReq.config[14]) {
-      mError("dnode:%d, failed to config keeptimeoffset since invalid conf:%s", cfgReq.dnodeId, cfgReq.config);
-      terrno = TSDB_CODE_INVALID_CFG;
-      return -1;
-    }
+    int32_t optLen = strlen("keeptimeoffset");
+    int32_t flag = -1;
+    int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag);
+    if (code < 0) return code;
 
-    const char *value = cfgReq.value;
-    int32_t     offset = atoi(value);
-    if (offset < 0 || offset > 23) {
-      mError("dnode:%d, failed to config keepTimeOffset since value:%d. Valid range: [0, 23]", cfgReq.dnodeId, offset);
+    if (flag < 0 || flag > 23) {
+      mError("dnode:%d, failed to config keepTimeOffset since value:%d. Valid range: [0, 23]", cfgReq.dnodeId, flag);
       terrno = TSDB_CODE_INVALID_CFG;
       return -1;
     }
 
     strcpy(dcfgReq.config, "keeptimeoffset");
-    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", offset);
+    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
 #ifdef TD_ENTERPRISE
   } else if (strncasecmp(cfgReq.config, "activeCode", 10) == 0 || strncasecmp(cfgReq.config, "cActiveCode", 11) == 0) {
     int8_t opt = strncasecmp(cfgReq.config, "a", 1) == 0 ? DND_ACTIVE_CODE : DND_CONN_ACTIVE_CODE;
@@ -1309,3 +1312,28 @@ static void mndCancelGetNextDnode(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
 }
+
+// get int32_t value from 'SMCfgDnodeReq'
+static int32_t mndMCfgGetValInt32(SMCfgDnodeReq *pMCfgReq, int32_t opLen, int32_t *pOutValue) {
+  terrno = 0;
+  if (' ' != pMCfgReq->config[opLen] && 0 != pMCfgReq->config[opLen]) {
+    goto _err;
+  }
+
+  if (' ' == pMCfgReq->config[opLen]) {
+    // 'key value'
+    if (strlen(pMCfgReq->value) != 0) goto _err;
+    *pOutValue = atoi(pMCfgReq->config + opLen + 1);
+  } else {
+    // 'key' 'value'
+    if (strlen(pMCfgReq->value) == 0) goto _err;
+    *pOutValue = atoi(pMCfgReq->value);
+  }
+
+  return 0;
+
+_err:
+  mError("dnode:%d, failed to config keeptimeoffset since invalid conf:%s", pMCfgReq->dnodeId, pMCfgReq->config);
+  terrno = TSDB_CODE_INVALID_CFG;
+  return -1;
+}

source/dnode/mnode/impl/src/mndScheduler.c

@@ -25,6 +25,7 @@
 #define SINK_NODE_LEVEL (0)
 extern bool tsDeployOnSnode;
 
+static int32_t setTaskUpstreamEpInfo(const SStreamTask* pTask, SStreamTask* pDownstream);
 static int32_t mndAddSinkTaskToStream(SStreamObj* pStream, SArray* pTaskList, SMnode* pMnode, int32_t vgId,
                                       SVgObj* pVgroup, int32_t fillHistory);
 static void    setFixedDownstreamEpInfo(SStreamTask* pDstTask, const SStreamTask* pTask);
@@ -87,10 +88,10 @@ END:
 
 int32_t mndSetSinkTaskInfo(SStreamObj* pStream, SStreamTask* pTask) {
   if (pStream->smaId != 0) {
-    pTask->outputType = TASK_OUTPUT__SMA;
+    pTask->outputInfo.type = TASK_OUTPUT__SMA;
     pTask->smaSink.smaId = pStream->smaId;
   } else {
-    pTask->outputType = TASK_OUTPUT__TABLE;
+    pTask->outputInfo.type = TASK_OUTPUT__TABLE;
     pTask->tbSink.stbUid = pStream->targetStbUid;
     memcpy(pTask->tbSink.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN);
     pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
@@ -110,7 +111,7 @@ int32_t mndAddDispatcherForInternalTask(SMnode* pMnode, SStreamObj* pStream, SAr
     SDbObj* pDb = mndAcquireDb(pMnode, pStream->targetDb);
     if (pDb != NULL && pDb->cfg.numOfVgroups > 1) {
       isShuffle = true;
-      pTask->outputType = TASK_OUTPUT__SHUFFLE_DISPATCH;
+      pTask->outputInfo.type = TASK_OUTPUT__SHUFFLE_DISPATCH;
       pTask->msgInfo.msgType = TDMT_STREAM_TASK_DISPATCH;
       if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) {
         return -1;
@@ -267,10 +268,15 @@ static int32_t addSourceStreamTask(SMnode* pMnode, SVgObj* pVgroup, SArray* pTas
     return terrno;
   }
 
+  for(int32_t i = 0; i < taosArrayGetSize(pSinkTaskList); ++i) {
+    SStreamTask* pSinkTask = taosArrayGetP(pSinkTaskList, i);
+    setTaskUpstreamEpInfo(pTask, pSinkTask);
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
-static SStreamChildEpInfo* createStreamTaskEpInfo(SStreamTask* pTask) {
+static SStreamChildEpInfo* createStreamTaskEpInfo(const SStreamTask* pTask) {
   SStreamChildEpInfo* pEpInfo = taosMemoryMalloc(sizeof(SStreamChildEpInfo));
   if (pEpInfo == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -291,11 +297,11 @@ void setFixedDownstreamEpInfo(SStreamTask* pDstTask, const SStreamTask* pTask) {
   pDispatcher->nodeId = pTask->info.nodeId;
   pDispatcher->epSet = pTask->info.epSet;
 
-  pDstTask->outputType = TASK_OUTPUT__FIXED_DISPATCH;
+  pDstTask->outputInfo.type = TASK_OUTPUT__FIXED_DISPATCH;
   pDstTask->msgInfo.msgType = TDMT_STREAM_TASK_DISPATCH;
 }
 
-int32_t setEpToDownstreamTask(SStreamTask* pTask, SStreamTask* pDownstream) {
+int32_t setTaskUpstreamEpInfo(const SStreamTask* pTask, SStreamTask* pDownstream) {
   SStreamChildEpInfo* pEpInfo = createStreamTaskEpInfo(pTask);
   if (pEpInfo == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;
@@ -418,7 +424,7 @@ static int32_t doAddSourceTask(SArray* pTaskList, int8_t fillHistory, int64_t ui
     return -1;
   }
 
-  return setEpToDownstreamTask(pTask, pDownstreamTask);
+  return setTaskUpstreamEpInfo(pTask, pDownstreamTask);
 }
 
 static int32_t doAddAggTask(uint64_t uid, SArray* pTaskList, SArray* pSinkNodeList, SMnode* pMnode, SStreamObj* pStream,
@@ -586,6 +592,14 @@ static int32_t addSinkTasks(SArray* pTasksList, SMnode* pMnode, SStreamObj* pStr
   return TSDB_CODE_SUCCESS;
 }
 
+static void setSinkTaskUpstreamInfo(SArray* pTasksList, const SStreamTask* pUpstreamTask) {
+  SArray* pSinkTaskList = taosArrayGetP(pTasksList, SINK_NODE_LEVEL);
+  for(int32_t i = 0; i < taosArrayGetSize(pSinkTaskList); ++i) {
+    SStreamTask* pSinkTask = taosArrayGetP(pSinkTaskList, i);
+    setTaskUpstreamEpInfo(pUpstreamTask, pSinkTask);
+  }
+}
+
 static int32_t doScheduleStream(SStreamObj* pStream, SMnode* pMnode, SQueryPlan* pPlan, int64_t nextWindowSkey) {
   SSdb*   pSdb = pMnode->pSdb;
   int32_t numOfPlanLevel = LIST_LENGTH(pPlan->pSubplans);
@@ -637,6 +651,9 @@ static int32_t doScheduleStream(SStreamObj* pStream, SMnode* pMnode, SQueryPlan*
       return code;
     }
 
+    setSinkTaskUpstreamInfo(pStream->tasks, pAggTask);
+    setSinkTaskUpstreamInfo(pStream->pHTasksList, pHAggTask);
+
     // source level
     return addSourceTasksForMultiLevelStream(pMnode, pPlan, pStream, pAggTask, pHAggTask, nextWindowSkey);
   } else if (numOfPlanLevel == 1) {

source/dnode/mnode/impl/src/mndStb.c

@@ -2498,12 +2498,14 @@ static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
-  if (0 == strcmp(cfgReq.dbFName, TSDB_INFORMATION_SCHEMA_DB)) {
+  char dbName[TSDB_DB_NAME_LEN] = {0};
+  mndExtractShortDbNameFromDbFullName(cfgReq.dbFName, dbName);
+  if (0 == strcmp(dbName, TSDB_INFORMATION_SCHEMA_DB)) {
     mInfo("information_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
     if (mndBuildInsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp) != 0) {
       goto _OVER;
     }
-  } else if (0 == strcmp(cfgReq.dbFName, TSDB_PERFORMANCE_SCHEMA_DB)) {
+  } else if (0 == strcmp(dbName, TSDB_PERFORMANCE_SCHEMA_DB)) {
     mInfo("performance_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
     if (mndBuildPerfsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp) != 0) {
       goto _OVER;
@@ -2672,6 +2674,13 @@ void mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst) {
   tNameGetDbName(&name, dst);
 }
 
+void mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst) {
+  SName name = {0};
+  tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB);
+
+  tNameGetDbName(&name, dst);
+}
+
 void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize) {
   int32_t pos = -1;
   int32_t num = 0;

source/dnode/snode/src/snode.c

@@ -66,14 +66,15 @@ int32_t sndExpandTask(SSnode *pSnode, SStreamTask *pTask, int64_t ver) {
 
   pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;
   pTask->inputQueue = streamQueueOpen(512 << 10);
-  pTask->outputQueue = streamQueueOpen(512 << 10);
+  pTask->outputInfo.queue = streamQueueOpen(512 << 10);
 
-  if (pTask->inputQueue == NULL || pTask->outputQueue == NULL) {
+  if (pTask->inputQueue == NULL || pTask->outputInfo.queue == NULL) {
     return -1;
   }
 
+  pTask->initTs = taosGetTimestampMs();
   pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
-  pTask->outputStatus = TASK_OUTPUT_STATUS__NORMAL;
+  pTask->outputInfo.status = TASK_OUTPUT_STATUS__NORMAL;
   pTask->pMsgCb = &pSnode->msgCb;
   pTask->chkInfo.version = ver;
   pTask->pMeta = pSnode->pMeta;
@@ -90,6 +91,7 @@ int32_t sndExpandTask(SSnode *pSnode, SStreamTask *pTask, int64_t ver) {
   pTask->exec.pExecutor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle, 0);
   ASSERT(pTask->exec.pExecutor);
 
+  taosThreadMutexInit(&pTask->lock, NULL);
   streamSetupScheduleTrigger(pTask);
 
   qDebug("snode:%d expand stream task on snode, s-task:%s, checkpoint ver:%" PRId64 " child id:%d, level:%d", SNODE_HANDLE,
@@ -166,11 +168,10 @@ int32_t sndProcessTaskDeployReq(SSnode *pSnode, char *msg, int32_t msgLen) {
 
   int32_t numOfTasks = streamMetaGetNumOfTasks(pSnode->pMeta);
   taosWUnLockLatch(&pSnode->pMeta->lock);
-
-  streamPrepareNdoCheckDownstream(pTask);
   qDebug("snode:%d s-task:%s is deployed on snode and add into meta, status:%s, numOfTasks:%d", SNODE_HANDLE, pTask->id.idStr,
          streamGetTaskStatusStr(pTask->status.taskStatus), numOfTasks);
 
+  streamTaskCheckDownstreamTasks(pTask);
   return 0;
 }
 
@@ -274,7 +275,7 @@ int32_t sndProcessWriteMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg *pRsp) {
   return 0;
 }
 
-int32_t sndProcessTaskRecoverFinishReq(SSnode *pSnode, SRpcMsg *pMsg) {
+int32_t sndProcessStreamTaskScanHistoryFinishReq(SSnode *pSnode, SRpcMsg *pMsg) {
   char   *msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
   int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
 
@@ -287,12 +288,12 @@ int32_t sndProcessTaskRecoverFinishReq(SSnode *pSnode, SRpcMsg *pMsg) {
   tDecoderClear(&decoder);
 
   // find task
-  SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, req.taskId);
+  SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, req.downstreamTaskId);
   if (pTask == NULL) {
     return -1;
   }
   // do process request
-  if (streamProcessScanHistoryFinishReq(pTask, req.taskId, req.childId) < 0) {
+  if (streamProcessScanHistoryFinishReq(pTask, &req, &pMsg->info) < 0) {
     streamMetaReleaseTask(pSnode->pMeta, pTask);
     return -1;
   }
@@ -415,7 +416,7 @@ int32_t sndProcessStreamMsg(SSnode *pSnode, SRpcMsg *pMsg) {
     case TDMT_STREAM_RETRIEVE_RSP:
       return sndProcessTaskRetrieveRsp(pSnode, pMsg);
     case TDMT_STREAM_SCAN_HISTORY_FINISH:
-      return sndProcessTaskRecoverFinishReq(pSnode, pMsg);
+      return sndProcessStreamTaskScanHistoryFinishReq(pSnode, pMsg);
     case TDMT_STREAM_SCAN_HISTORY_FINISH_RSP:
       return sndProcessTaskRecoverFinishRsp(pSnode, pMsg);
     case TDMT_STREAM_TASK_CHECK:

source/dnode/vnode/CMakeLists.txt

 # vnode
 add_library(vnode STATIC "")
-target_sources(
-    vnode
-    PRIVATE
-
-    # vnode
+set(
+  VNODE_SOURCE_FILES
   "src/vnd/vnodeOpen.c"
   "src/vnd/vnodeBufPool.c"
   "src/vnd/vnodeCfg.c"
@@ -38,23 +35,23 @@ target_sources(
     "src/sma/smaSnapshot.c"
     "src/sma/smaTimeRange.c"
 
-    # tsdb
-    "src/tsdb/tsdbCommit.c"
-    "src/tsdb/tsdbFile.c"
-    "src/tsdb/tsdbFS.c"
-    "src/tsdb/tsdbOpen.c"
-    "src/tsdb/tsdbMemTable.c"
-    "src/tsdb/tsdbRead.c"
-    "src/tsdb/tsdbCache.c"
-    "src/tsdb/tsdbWrite.c"
-    "src/tsdb/tsdbReaderWriter.c"
-    "src/tsdb/tsdbUtil.c"
-    "src/tsdb/tsdbSnapshot.c"
-    "src/tsdb/tsdbCacheRead.c"
-    "src/tsdb/tsdbRetention.c"
-    "src/tsdb/tsdbDiskData.c"
-    "src/tsdb/tsdbMergeTree.c"
-    "src/tsdb/tsdbDataIter.c"
+  # # tsdb
+  # "src/tsdb/tsdbCommit.c"
+  # "src/tsdb/tsdbFile.c"
+  # "src/tsdb/tsdbFS.c"
+  # "src/tsdb/tsdbOpen.c"
+  # "src/tsdb/tsdbMemTable.c"
+  # "src/tsdb/tsdbRead.c"
+  # "src/tsdb/tsdbCache.c"
+  # "src/tsdb/tsdbWrite.c"
+  # "src/tsdb/tsdbReaderWriter.c"
+  # "src/tsdb/tsdbUtil.c"
+  # "src/tsdb/tsdbSnapshot.c"
+  # "src/tsdb/tsdbCacheRead.c"
+  # "src/tsdb/tsdbRetention.c"
+  # "src/tsdb/tsdbDiskData.c"
+  # "src/tsdb/tsdbMergeTree.c"
+  # "src/tsdb/tsdbDataIter.c"
 
     # tq
     "src/tq/tq.c"
@@ -71,6 +68,19 @@ target_sources(
     "src/tq/tqOffsetSnapshot.c"
 )
 
+aux_source_directory("src/tsdb/" TSDB_SOURCE_FILES)
+list(
+  APPEND
+  VNODE_SOURCE_FILES
+  ${TSDB_SOURCE_FILES}
+)
+
+target_sources(
+    vnode
+    PRIVATE
+    ${VNODE_SOURCE_FILES}
+)
+
 IF (TD_VNODE_PLUGINS)
   target_sources(
     vnode

source/dnode/vnode/inc/vnode.h

@@ -51,12 +51,14 @@ extern const SVnodeCfg vnodeCfgDefault;
 
 int32_t vnodeInit(int32_t nthreads);
 void    vnodeCleanup();
-int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs);
-int32_t vnodeAlterReplica(const char *path, SAlterVnodeReplicaReq *pReq, STfs *pTfs);
-int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnodeHashRangeReq *pReq, STfs *pTfs);
-int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId, STfs *pTfs);
+int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, int32_t diskPrimary, STfs *pTfs);
+int32_t vnodeAlterReplica(const char *path, SAlterVnodeReplicaReq *pReq, int32_t diskPrimary, STfs *pTfs);
+int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnodeHashRangeReq *pReq,
+                            int32_t diskPrimary, STfs *pTfs);
+int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId,
+                             int32_t diskPrimary, STfs *pTfs);
 void    vnodeDestroy(const char *path, STfs *pTfs);
-SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb);
+SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgCb);
 void    vnodePreClose(SVnode *pVnode);
 void    vnodePostClose(SVnode *pVnode);
 void    vnodeSyncCheckTimeout(SVnode *pVnode);
@@ -166,6 +168,27 @@ uint64_t     tsdbGetReaderMaxVersion(STsdbReader *pReader);
 void         tsdbReaderSetCloseFlag(STsdbReader *pReader);
 int64_t      tsdbGetLastTimestamp(SVnode *pVnode, void *pTableList, int32_t numOfTables, const char *pIdStr);
 
+//======================================================================================================================
+int32_t      tsdbReaderOpen2(void *pVnode, SQueryTableDataCond *pCond, void *pTableList, int32_t numOfTables,
+                            SSDataBlock *pResBlock, void **ppReader, const char *idstr, bool countOnly,
+                            SHashObj **pIgnoreTables);
+int32_t      tsdbSetTableList2(STsdbReader *pReader, const void *pTableList, int32_t num);
+void         tsdbReaderSetId2(STsdbReader *pReader, const char *idstr);
+void         tsdbReaderClose2(STsdbReader *pReader);
+int32_t      tsdbNextDataBlock2(STsdbReader *pReader, bool *hasNext);
+int32_t      tsdbRetrieveDatablockSMA2(STsdbReader *pReader, SSDataBlock *pDataBlock, bool *allHave, bool *hasNullSMA);
+void         tsdbReleaseDataBlock2(STsdbReader *pReader);
+SSDataBlock *tsdbRetrieveDataBlock2(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);
+int32_t      tsdbReaderReset2(STsdbReader *pReader, SQueryTableDataCond *pCond);
+int32_t      tsdbGetFileBlocksDistInfo2(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
+int64_t      tsdbGetNumOfRowsInMemTable2(STsdbReader *pHandle);
+void        *tsdbGetIdx2(SMeta *pMeta);
+void        *tsdbGetIvtIdx2(SMeta *pMeta);
+uint64_t     tsdbGetReaderMaxVersion2(STsdbReader *pReader);
+void         tsdbReaderSetCloseFlag2(STsdbReader *pReader);
+int64_t      tsdbGetLastTimestamp2(SVnode *pVnode, void *pTableList, int32_t numOfTables, const char *pIdStr);
+//======================================================================================================================
+
 int32_t tsdbReuseCacherowsReader(void *pReader, void *pTableIdList, int32_t numOfTables);
 int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, void *pTableIdList, int32_t numOfTables, int32_t numOfCols,
                                 SArray *pCidList, int32_t *pSlotIds, uint64_t suid, void **pReader, const char *idstr);

source/dnode/vnode/src/inc/tsdb.h

@@ -16,6 +16,9 @@
 #ifndef _TD_VNODE_TSDB_H_
 #define _TD_VNODE_TSDB_H_
 
+// #include "../tsdb/tsdbFile2.h"
+// #include "../tsdb/tsdbMerge.h"
+// #include "../tsdb/tsdbSttFileRW.h"
 #include "tsimplehash.h"
 #include "vnodeInt.h"
 
@@ -76,7 +79,6 @@ typedef struct STsdbFilterInfo  STsdbFilterInfo;
 #define TSDBROW_COL_FMT ((int8_t)0x1)
 
 #define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)
-#define TSDB_MAX_SUBBLOCKS 8
 #define TSDB_FHDR_SIZE 512
 
 #define VERSION_MIN 0
@@ -165,6 +167,7 @@ void    tBlockDataDestroy(SBlockData *pBlockData);
 int32_t tBlockDataInit(SBlockData *pBlockData, TABLEID *pId, STSchema *pTSchema, int16_t *aCid, int32_t nCid);
 void    tBlockDataReset(SBlockData *pBlockData);
 int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid);
+int32_t tBlockDataUpdateRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema);
 int32_t tBlockDataTryUpsertRow(SBlockData *pBlockData, TSDBROW *pRow, int64_t uid);
 int32_t tBlockDataUpsertRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid);
 void    tBlockDataClear(SBlockData *pBlockData);
@@ -198,7 +201,7 @@ int32_t tMapDataToArray(SMapData *pMapData, int32_t itemSize, int32_t (*tGetItem
 // other
 int32_t tsdbKeyFid(TSKEY key, int32_t minutes, int8_t precision);
 void    tsdbFidKeyRange(int32_t fid, int32_t minutes, int8_t precision, TSKEY *minKey, TSKEY *maxKey);
-int32_t tsdbFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int64_t now);
+int32_t tsdbFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int64_t nowSec);
 int32_t tsdbBuildDeleteSkyline(SArray *aDelData, int32_t sidx, int32_t eidx, SArray *aSkyline);
 int32_t tPutColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg);
 int32_t tGetColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg);
@@ -302,8 +305,11 @@ int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx);
 // tsdbRead.c ==============================================================================================
 int32_t tsdbTakeReadSnap(STsdbReader *pReader, _query_reseek_func_t reseek, STsdbReadSnap **ppSnap);
 void    tsdbUntakeReadSnap(STsdbReader *pReader, STsdbReadSnap *pSnap, bool proactive);
+
+int32_t tsdbTakeReadSnap2(STsdbReader *pReader, _query_reseek_func_t reseek, STsdbReadSnap **ppSnap);
+void    tsdbUntakeReadSnap2(STsdbReader *pReader, STsdbReadSnap *pSnap, bool proactive);
 // tsdbMerge.c ==============================================================================================
-int32_t tsdbMerge(STsdb *pTsdb);
+int32_t tsdbMerge(void *arg);
 
 // tsdbDiskData ==============================================================================================
 int32_t tDiskDataBuilderCreate(SDiskDataBuilder **ppBuilder);
@@ -368,12 +374,13 @@ struct STsdb {
   TdThreadRwlock       rwLock;
   SMemTable           *mem;
   SMemTable           *imem;
-  STsdbFS          fs;
+  STsdbFS              fs;  // old
   SLRUCache           *lruCache;
   SCacheFlushState     flushState;
   TdThreadMutex        lruMutex;
   SLRUCache           *biCache;
   TdThreadMutex        biMutex;
+  struct STFileSystem *pFS;  // new
   SRocksCache          rCache;
 };
 
@@ -410,6 +417,7 @@ struct STbData {
   SDelData    *pTail;
   SMemSkipList sl;
   STbData     *next;
+  SRBTreeNode  rbtn[1];
 };
 
 struct SMemTable {
@@ -423,11 +431,10 @@ struct SMemTable {
   TSKEY            maxKey;
   int64_t          nRow;
   int64_t          nDel;
-  struct {
   int32_t          nTbData;
   int32_t          nBucket;
   STbData        **aBucket;
-  };
+  SRBTree          tbDataTree[1];
 };
 
 struct TSDBROW {
@@ -500,7 +507,7 @@ struct SDataBlk {
   int32_t    nRow;
   int8_t     hasDup;
   int8_t     nSubBlock;
-  SBlockInfo aSubBlock[TSDB_MAX_SUBBLOCKS];
+  SBlockInfo aSubBlock[1];
   SSmaInfo   smaInfo;
 };
 
@@ -652,11 +659,18 @@ struct SDelFWriter {
   uint8_t *aBuf[1];
 };
 
+#include "tarray2.h"
+//#include "tsdbFS2.h"
+// struct STFileSet;
+typedef struct STFileSet STFileSet;
+typedef TARRAY2(STFileSet *) TFileSetArray;
+
 struct STsdbReadSnap {
   SMemTable     *pMem;
   SQueryNode    *pNode;
   SMemTable     *pIMem;
   SQueryNode    *pINode;
+  TFileSetArray *pfSetArray;
   STsdbFS        fs;
 };
 
@@ -696,6 +710,7 @@ typedef struct {
 
 typedef struct SSttBlockLoadInfo {
   SBlockData blockData[2];
+  void      *pSttStatisBlkArray;
   SArray    *aSttBlk;
   int32_t    blockIndex[2];  // to denote the loaded block in the corresponding position.
   int32_t    currentLoadBlockIndex;
@@ -704,10 +719,9 @@ typedef struct SSttBlockLoadInfo {
   STSchema  *pSchema;
   int16_t   *colIds;
   int32_t    numOfCols;
-  bool       checkRemainingRow;
+  bool       checkRemainingRow;  // todo: no assign value?
   bool       isLast;
   bool       sttBlockLoaded;
-  int32_t    numOfStt;
 
   // keep the last access position, this position may be used to reduce the binary times for
   // starting last block data for a new table
@@ -768,8 +782,7 @@ struct SDiskDataBuilder {
 typedef struct SLDataIter {
   SRBTreeNode            node;
   SSttBlk               *pSttBlk;
-  SDataFReader      *pReader;
-  int32_t            iStt;
+  int32_t                iStt;  // for debug purpose
   int8_t                 backward;
   int32_t                iSttBlk;
   int32_t                iRow;
@@ -779,32 +792,79 @@ typedef struct SLDataIter {
   SVersionRange          verRange;
   SSttBlockLoadInfo     *pBlockLoadInfo;
   bool                   ignoreEarlierTs;
+  struct SSttFileReader *pReader;
 } SLDataIter;
 
 #define tMergeTreeGetRow(_t) (&((_t)->pIter->rInfo.row))
 int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
                        STimeWindow *pTimeWindow, SVersionRange *pVerRange, SSttBlockLoadInfo *pBlockLoadInfo,
                        bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter *pLDataIter);
+
+struct SSttFileReader;
+typedef int32_t (*_load_tomb_fn)(STsdbReader *pReader, struct SSttFileReader *pSttFileReader,
+                                 SSttBlockLoadInfo *pLoadInfo);
+
+typedef struct {
+  int8_t        backward;
+  STsdb        *pTsdb;
+  uint64_t      suid;
+  uint64_t      uid;
+  STimeWindow   timewindow;
+  SVersionRange verRange;
+  bool          strictTimeRange;
+  SArray       *pSttFileBlockIterArray;
+  void         *pCurrentFileset;
+  STSchema     *pSchema;
+  int16_t      *pCols;
+  int32_t       numOfCols;
+  _load_tomb_fn loadTombFn;
+  void         *pReader;
+  void         *idstr;
+} SMergeTreeConf;
+int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf);
+
 void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
 bool tMergeTreeNext(SMergeTree *pMTree);
 bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);
 void tMergeTreeClose(SMergeTree *pMTree);
 
 SSttBlockLoadInfo *tCreateLastBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols, int32_t numOfStt);
+SSttBlockLoadInfo *tCreateOneLastBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols);
 void               resetLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
 void               getLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo, int64_t *blocks, double *el);
 void              *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
+void              *destroySttBlockReader(SArray *pLDataIterArray, int64_t *blocks, double *el);
 
 // tsdbCache ==============================================================================================
+typedef enum {
+  READ_MODE_COUNT_ONLY = 0x1,
+  READ_MODE_ALL,
+} EReadMode;
+
+typedef struct STsdbReaderInfo {
+  uint64_t      suid;
+  STSchema     *pSchema;
+  EReadMode     readMode;
+  uint64_t      rowsNum;
+  STimeWindow   window;
+  SVersionRange verRange;
+  int16_t       order;
+} STsdbReaderInfo;
+
+typedef struct {
+  SArray *pTombData;
+} STableLoadInfo;
+
+struct SDataFileReader;
+
 typedef struct SCacheRowsReader {
   STsdb                  *pTsdb;
-  SVersionRange      verRange;
+  STsdbReaderInfo         info;
   TdThreadMutex           readerMutex;
   SVnode                 *pVnode;
   STSchema               *pSchema;
   STSchema               *pCurrSchema;
   uint64_t                uid;
-  uint64_t           suid;
   char                  **transferBuf;  // todo remove it soon
   int32_t                 numOfCols;
   SArray                 *pCidList;
@@ -813,12 +873,12 @@ typedef struct SCacheRowsReader {
   int32_t                 tableIndex;  // currently returned result tables
   STableKeyInfo          *pTableList;  // table id list
   int32_t                 numOfTables;
-  SSttBlockLoadInfo *pLoadInfo;
-  SLDataIter        *pDataIter;
+  uint64_t               *uidList;
+  SSHashObj              *pTableMap;
+  SArray                 *pLDataIterArray;
+  struct SDataFileReader *pFileReader;
   STsdbReadSnap          *pReadSnap;
-  SDataFReader      *pDataFReader;
-  SDataFReader      *pDataFReaderLast;
-  const char        *idstr;
+  char                   *idstr;
   int64_t                 lastTs;
 } SCacheRowsReader;
 

source/dnode/vnode/src/inc/vnd.h

@@ -49,7 +49,8 @@ int32_t vnodeEncodeConfig(const void* pObj, SJson* pJson);
 int32_t vnodeDecodeConfig(const SJson* pJson, void* pObj);
 
 // vnodeModule.c
-int32_t vnodeScheduleTask(int32_t (*execute)(void*), void* arg);
+int vnodeScheduleTask(int (*execute)(void*), void* arg);
+int vnodeScheduleTaskEx(int tpid, int (*execute)(void*), void* arg);
 
 // vnodeBufPool.c
 typedef struct SVBufPoolNode SVBufPoolNode;
@@ -87,7 +88,7 @@ void    vnodeBufPoolAddToFreeList(SVBufPool* pPool);
 int32_t vnodeBufPoolRecycle(SVBufPool* pPool);
 
 // vnodeOpen.c
-int32_t vnodeGetPrimaryDir(const char* relPath, STfs* pTfs, char* buf, size_t bufLen);
+int32_t vnodeGetPrimaryDir(const char* relPath, int32_t diskPrimary, STfs* pTfs, char* buf, size_t bufLen);
 
 // vnodeQuery.c
 int32_t vnodeQueryOpen(SVnode* pVnode);

source/dnode/vnode/src/inc/vnodeInt.h

@@ -93,6 +93,7 @@ typedef struct SQueryNode         SQueryNode;
 #define VNODE_BUFPOOL_SEGMENTS 3
 
 #define VND_INFO_FNAME "vnode.json"
+#define VND_INFO_FNAME_TMP "vnode_tmp.json"
 
 // vnd.h
 typedef int32_t (*_query_reseek_func_t)(void* pQHandle);
@@ -179,8 +180,8 @@ SArray*       metaGetSmaTbUids(SMeta* pMeta);
 void*         metaGetIdx(SMeta* pMeta);
 void*         metaGetIvtIdx(SMeta* pMeta);
 
-int64_t       metaGetTbNum(SMeta *pMeta);
-void          metaReaderDoInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags);
+int64_t metaGetTbNum(SMeta* pMeta);
+void    metaReaderDoInit(SMetaReader* pReader, SMeta* pMeta, int32_t flags);
 
 int32_t metaCreateTSma(SMeta* pMeta, int64_t version, SSmaCfg* pCfg);
 int32_t metaDropTSma(SMeta* pMeta, int64_t indexUid);
@@ -197,12 +198,12 @@ int32_t metaGetInfo(SMeta* pMeta, int64_t uid, SMetaInfo* pInfo, SMetaReader* pR
 int     tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeepCfg* pKeepCfg, int8_t rollback);
 int     tsdbClose(STsdb** pTsdb);
 int32_t tsdbBegin(STsdb* pTsdb);
-int32_t tsdbPrepareCommit(STsdb* pTsdb);
-int32_t tsdbCommit(STsdb* pTsdb, SCommitInfo* pInfo);
+// int32_t tsdbPrepareCommit(STsdb* pTsdb);
+// int32_t tsdbCommit(STsdb* pTsdb, SCommitInfo* pInfo);
 int32_t tsdbCacheCommit(STsdb* pTsdb);
 int32_t tsdbCompact(STsdb* pTsdb, SCompactInfo* pInfo);
-int32_t tsdbFinishCommit(STsdb* pTsdb);
-int32_t tsdbRollbackCommit(STsdb* pTsdb);
+// int32_t tsdbFinishCommit(STsdb* pTsdb);
+// int32_t tsdbRollbackCommit(STsdb* pTsdb);
 int     tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq2* pMsg);
 int     tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq2* pMsg, SSubmitRsp2* pRsp);
 int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitTbData* pSubmitTbData, int32_t* affectedRows);
@@ -249,8 +250,8 @@ int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskTransferStateReq(STQ* pTq, SRpcMsg* pMsg);
-int32_t tqProcessStreamTaskScanHistoryFinishReq(STQ* pTq, SRpcMsg* pMsg);
-int32_t tqProcessTaskRecoverFinishRsp(STQ* pTq, SRpcMsg* pMsg);
+int32_t tqProcessTaskScanHistoryFinishReq(STQ* pTq, SRpcMsg* pMsg);
+int32_t tqProcessTaskScanHistoryFinishRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqCheckLogInWal(STQ* pTq, int64_t version);
 
 // sma
@@ -385,6 +386,7 @@ struct SVnode {
   SVState   state;
   SVStatis  statis;
   STfs*     pTfs;
+  int32_t   diskPrimary;
   SMsgCb    msgCb;
 
   // Buffer Pool

source/dnode/vnode/src/meta/metaOpen.c

@@ -41,7 +41,7 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
   *ppMeta = NULL;
 
   // create handle
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, path, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, path, TSDB_FILENAME_LEN);
   offset = strlen(path);
   snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VNODE_META_DIR);
 

source/dnode/vnode/src/meta/metaTtl.c

@@ -379,7 +379,7 @@ _out:
 int ttlMgrFlush(STtlManger *pTtlMgr, TXN *pTxn) {
   ttlMgrWLock(pTtlMgr);
 
-  metaInfo("%s, ttl mgr flush start. dirty uids:%d", pTtlMgr->logPrefix, taosHashGetSize(pTtlMgr->pDirtyUids));
+  metaDebug("%s, ttl mgr flush start. dirty uids:%d", pTtlMgr->logPrefix, taosHashGetSize(pTtlMgr->pDirtyUids));
 
   int ret = -1;
 
@@ -433,7 +433,7 @@ int ttlMgrFlush(STtlManger *pTtlMgr, TXN *pTxn) {
 _out:
   ttlMgrULock(pTtlMgr);
 
-  metaInfo("%s, ttl mgr flush end.", pTtlMgr->logPrefix);
+  metaDebug("%s, ttl mgr flush end.", pTtlMgr->logPrefix);
 
   return ret;
 }

source/dnode/vnode/src/sma/smaCommit.c

@@ -103,15 +103,16 @@ _exit:
   return code;
 }
 
+extern int32_t tsdbCommitCommit(STsdb *tsdb);
 int32_t        smaFinishCommit(SSma *pSma) {
   int32_t code = 0;
   int32_t lino = 0;
   SVnode *pVnode = pSma->pVnode;
 
-  if (VND_RSMA1(pVnode) && (code = tsdbFinishCommit(VND_RSMA1(pVnode))) < 0) {
+  if (VND_RSMA1(pVnode) && (code = tsdbCommitCommit(VND_RSMA1(pVnode))) < 0) {
     TSDB_CHECK_CODE(code, lino, _exit);
   }
-  if (VND_RSMA2(pVnode) && (code = tsdbFinishCommit(VND_RSMA2(pVnode))) < 0) {
+  if (VND_RSMA2(pVnode) && (code = tsdbCommitCommit(VND_RSMA2(pVnode))) < 0) {
     TSDB_CHECK_CODE(code, lino, _exit);
   }
 _exit:
@@ -130,6 +131,7 @@ _exit:
  * @param isCommit
  * @return int32_t
  */
+extern int32_t tsdbPreCommit(STsdb *tsdb);
 static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
   int32_t code = 0;
   int32_t lino = 0;
@@ -186,11 +188,11 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
   // all rsma results are written completely
   STsdb *pTsdb = NULL;
   if ((pTsdb = VND_RSMA1(pSma->pVnode))) {
-    code = tsdbPrepareCommit(pTsdb);
+    code = tsdbPreCommit(pTsdb);
     TSDB_CHECK_CODE(code, lino, _exit);
   }
   if ((pTsdb = VND_RSMA2(pSma->pVnode))) {
-    code = tsdbPrepareCommit(pTsdb);
+    code = tsdbPreCommit(pTsdb);
     TSDB_CHECK_CODE(code, lino, _exit);
   }
 
@@ -207,6 +209,7 @@ _exit:
  * @param pSma
  * @return int32_t
  */
+extern int32_t tsdbCommitBegin(STsdb *tsdb, SCommitInfo *info);
 static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
   int32_t code = 0;
   int32_t lino = 0;
@@ -217,10 +220,10 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
     goto _exit;
   }
 
-  code = tsdbCommit(VND_RSMA1(pVnode), pInfo);
+  code = tsdbCommitBegin(VND_RSMA1(pVnode), pInfo);
   TSDB_CHECK_CODE(code, lino, _exit);
 
-  code = tsdbCommit(VND_RSMA2(pVnode), pInfo);
+  code = tsdbCommitBegin(VND_RSMA2(pVnode), pInfo);
   TSDB_CHECK_CODE(code, lino, _exit);
 
 _exit:

source/dnode/vnode/src/sma/smaRollup.c

@@ -264,7 +264,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
       return TSDB_CODE_FAILED;
     }
 
-    SReadHandle handle = { .vnode = pVnode, .initTqReader = 1, .pStateBackend = pStreamState };
+    SReadHandle handle = {.vnode = pVnode, .initTqReader = 1, .pStateBackend = pStreamState};
     initStorageAPI(&handle.api);
 
     pRSmaInfo->taskInfo[idx] = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle, TD_VID(pVnode));
@@ -572,8 +572,8 @@ int32_t smaDoRetention(SSma *pSma, int64_t now) {
 
   for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
     if (pSma->pRSmaTsdb[i]) {
-      code = tsdbDoRetention(pSma->pRSmaTsdb[i], now);
-      if (code) goto _end;
+      // code = tsdbDoRetention(pSma->pRSmaTsdb[i], now);
+      // if (code) goto _end;
     }
   }
 
@@ -612,7 +612,6 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
     blockDebugShowDataBlocks(pResList, flag);
 #endif
     for (int32_t i = 0; i < taosArrayGetSize(pResList); ++i) {
-
       output = taosArrayGetP(pResList, i);
       smaDebug("vgId:%d, result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%" PRIi64, SMA_VID(pSma),
                output->info.id.uid, output->info.id.groupId, output->info.rows);
@@ -1114,8 +1113,8 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
   }
 
   if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) {
-    smaWarn("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")",
-             smaMgmt.rsetId, pRSmaRef->refId);  // pRSmaRef freed in taosHashRemove
+    smaWarn("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
+            pRSmaRef->refId);  // pRSmaRef freed in taosHashRemove
     taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
     return;
   }

source/dnode/vnode/src/sma/smaUtil.c

@@ -26,7 +26,7 @@ void tdRSmaGetDirName(SVnode *pVnode, STfs *pTfs, bool endWithSep, char *outputN
   int32_t offset = 0;
 
   // vnode
-  vnodeGetPrimaryDir(pVnode->path, pTfs, outputName, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pTfs, outputName, TSDB_FILENAME_LEN);
   offset = strlen(outputName);
 
   // rsma

source/dnode/vnode/src/tq/tqRestore.c

@@ -80,11 +80,17 @@ int32_t tqStreamTasksStatusCheck(STQ* pTq) {
       continue;
     }
 
-    streamTaskCheckDownstreamTasks(pTask);
+    if (pTask->info.fillHistory == 1) {
+      tqDebug("s-task:%s fill-history task, wait for related stream task:0x%x to launch it", pTask->id.idStr,
+              pTask->streamTaskId.taskId);
+      continue;
+    }
+
+    streamTaskDoCheckDownstreamTasks(pTask);
     streamMetaReleaseTask(pMeta, pTask);
   }
-  taosArrayDestroy(pTaskList);
 
+  taosArrayDestroy(pTaskList);
   return 0;
 }
 
@@ -234,7 +240,9 @@ int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
     }
 
     int32_t status = pTask->status.taskStatus;
-    if (pTask->info.taskLevel != TASK_LEVEL__SOURCE) {
+
+    // non-source or fill-history tasks don't need to response the WAL scan action.
+    if (pTask->info.taskLevel != TASK_LEVEL__SOURCE || pTask->info.fillHistory == 1) {
       streamMetaReleaseTask(pStreamMeta, pTask);
       continue;
     }

source/dnode/vnode/src/tsdb/tsdbCacheRead.c

@@ -17,6 +17,7 @@
 #include "tarray.h"
 #include "tcommon.h"
 #include "tsdb.h"
+#include "tsdbDataFileRW.h"
 
 #define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))
 
@@ -124,11 +125,29 @@ int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOf
   pReader->numOfTables = numOfTables;
   pReader->lastTs = INT64_MIN;
 
-  resetLastBlockLoadInfo(pReader->pLoadInfo);
+  int64_t blocks;
+  double  elapse;
+  pReader->pLDataIterArray = destroySttBlockReader(pReader->pLDataIterArray, &blocks, &elapse);
+  pReader->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
 
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t uidComparFunc(const void* p1, const void* p2) {
+  uint64_t pu1 = *(uint64_t*)p1;
+  uint64_t pu2 = *(uint64_t*)p2;
+  if (pu1 == pu2) {
+    return 0;
+  } else {
+    return (pu1 < pu2) ? -1 : 1;
+  }
+}
+
+static void freeTableInfoFunc(void* param) {
+  void** p = (void**)param;
+  taosMemoryFreeClear(*p);
+}
+
 int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,
                                 SArray* pCidList, int32_t* pSlotIds, uint64_t suid, void** pReader, const char* idstr) {
   *pReader = NULL;
@@ -140,11 +159,11 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
   p->type = type;
   p->pVnode = pVnode;
   p->pTsdb = p->pVnode->pTsdb;
-  p->verRange = (SVersionRange){.minVer = 0, .maxVer = UINT64_MAX};
+  p->info.verRange = (SVersionRange){.minVer = 0, .maxVer = UINT64_MAX};
+  p->info.suid = suid;
   p->numOfCols = numOfCols;
   p->pCidList = pCidList;
   p->pSlotIds = pSlotIds;
-  p->suid = suid;
 
   if (numOfTables == 0) {
     *pReader = p;
@@ -154,6 +173,27 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
   p->pTableList = pTableIdList;
   p->numOfTables = numOfTables;
 
+  p->pTableMap = tSimpleHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
+  if (p->pTableMap == NULL) {
+    tsdbCacherowsReaderClose(p);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  p->uidList = taosMemoryMalloc(numOfTables * sizeof(uint64_t));
+  if (p->uidList == NULL) {
+    tsdbCacherowsReaderClose(p);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  for (int32_t i = 0; i < numOfTables; ++i) {
+    uint64_t uid = p->pTableList[i].uid;
+    p->uidList[i] = uid;
+    STableLoadInfo* pInfo = taosMemoryCalloc(1, sizeof(STableLoadInfo));
+    tSimpleHashPut(p->pTableMap, &uid, sizeof(uint64_t), &pInfo, POINTER_BYTES);
+  }
+
+  tSimpleHashSetFreeFp(p->pTableMap, freeTableInfoFunc);
+
+  taosSort(p->uidList, numOfTables, sizeof(uint64_t), uidComparFunc);
+
   int32_t code = setTableSchema(p, suid, idstr);
   if (code != TSDB_CODE_SUCCESS) {
     tsdbCacherowsReaderClose(p);
@@ -178,14 +218,8 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
 
   SVnodeCfg* pCfg = &((SVnode*)pVnode)->config;
   int32_t    numOfStt = pCfg->sttTrigger;
-  p->pLoadInfo = tCreateLastBlockLoadInfo(p->pSchema, NULL, 0, numOfStt);
-  if (p->pLoadInfo == NULL) {
-    tsdbCacherowsReaderClose(p);
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-
-  p->pDataIter = taosMemoryCalloc(pCfg->sttTrigger, sizeof(SLDataIter));
-  if (p->pDataIter == NULL) {
+  p->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
+  if (p->pLDataIterArray == NULL) {
     tsdbCacherowsReaderClose(p);
     return TSDB_CODE_OUT_OF_MEMORY;
   }
@@ -214,14 +248,34 @@ void* tsdbCacherowsReaderClose(void* pReader) {
     taosMemoryFree(p->pSchema);
   }
 
-  taosMemoryFree(p->pDataIter);
   taosMemoryFree(p->pCurrSchema);
 
-  destroyLastBlockLoadInfo(p->pLoadInfo);
+  int64_t loadBlocks = 0;
+  double  elapse = 0;
+  destroySttBlockReader(p->pLDataIterArray, &loadBlocks, &elapse);
+
+  if (p->pFileReader) {
+    tsdbDataFileReaderClose(&p->pFileReader);
+    p->pFileReader = NULL;
+  }
 
   taosMemoryFree((void*)p->idstr);
   taosThreadMutexDestroy(&p->readerMutex);
 
+  if (p->pTableMap) {
+    void*   pe = NULL;
+    int32_t iter = 0;
+    while ((pe = tSimpleHashIterate(p->pTableMap, pe, &iter)) != NULL) {
+      STableLoadInfo* pInfo = *(STableLoadInfo**)pe;
+      pInfo->pTombData = taosArrayDestroy(pInfo->pTombData);
+    }
+
+    tSimpleHashCleanup(p->pTableMap);
+  }
+  if (p->uidList) {
+    taosMemoryFree(p->uidList);
+  }
+
   taosMemoryFree(pReader);
   return NULL;
 }
@@ -298,12 +352,10 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
   }
 
   taosThreadMutexLock(&pr->readerMutex);
-  code = tsdbTakeReadSnap((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap);
+  code = tsdbTakeReadSnap2((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap);
   if (code != TSDB_CODE_SUCCESS) {
     goto _end;
   }
-  pr->pDataFReader = NULL;
-  pr->pDataFReaderLast = NULL;
 
   int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
 
@@ -424,11 +476,13 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
   }
 
 _end:
-  tsdbDataFReaderClose(&pr->pDataFReaderLast);
-  tsdbDataFReaderClose(&pr->pDataFReader);
+  tsdbUntakeReadSnap2((STsdbReader*)pr, pr->pReadSnap, true);
+
+  int64_t loadBlocks = 0;
+  double  elapse = 0;
+  pr->pLDataIterArray = destroySttBlockReader(pr->pLDataIterArray, &loadBlocks, &elapse);
+  pr->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
 
-  resetLastBlockLoadInfo(pr->pLoadInfo);
-  tsdbUntakeReadSnap((STsdbReader*)pr, pr->pReadSnap, true);
   taosThreadMutexUnlock(&pr->readerMutex);
 
   if (pRes != NULL) {

source/dnode/vnode/src/tsdb/tsdbCommit2.h

+/*
+ * 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 "tsdbDataFileRW.h"
+#include "tsdbFS2.h"
+#include "tsdbFSetRW.h"
+#include "tsdbIter.h"
+#include "tsdbSttFileRW.h"
+
+#ifndef _TSDB_COMMIT_H_
+#define _TSDB_COMMIT_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TSDB_COMMIT_H_*/
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbDataFileRW.h

+/*
+ * 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 "tsdbDef.h"
+#include "tsdbFSet2.h"
+#include "tsdbSttFileRW.h"
+#include "tsdbUtil2.h"
+
+#ifndef _TSDB_DATA_FILE_RW_H
+#define _TSDB_DATA_FILE_RW_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef TARRAY2(SBlockIdx) TBlockIdxArray;
+typedef TARRAY2(SDataBlk) TDataBlkArray;
+typedef TARRAY2(SColumnDataAgg) TColumnDataAggArray;
+
+typedef struct {
+  SFDataPtr brinBlkPtr[1];
+  SFDataPtr rsrvd[2];
+} SHeadFooter;
+
+typedef struct {
+  SFDataPtr tombBlkPtr[1];
+  SFDataPtr rsrvd[2];
+} STombFooter;
+
+// SDataFileReader =============================================
+typedef struct SDataFileReader SDataFileReader;
+typedef struct SDataFileReaderConfig {
+  STsdb  *tsdb;
+  int32_t szPage;
+  struct {
+    bool   exist;
+    STFile file;
+  } files[TSDB_FTYPE_MAX];
+  uint8_t **bufArr;
+} SDataFileReaderConfig;
+
+int32_t tsdbDataFileReaderOpen(const char *fname[/* TSDB_FTYPE_MAX */], const SDataFileReaderConfig *config,
+                               SDataFileReader **reader);
+int32_t tsdbDataFileReaderClose(SDataFileReader **reader);
+// .head
+int32_t tsdbDataFileReadBrinBlk(SDataFileReader *reader, const TBrinBlkArray **brinBlkArray);
+int32_t tsdbDataFileReadBrinBlock(SDataFileReader *reader, const SBrinBlk *brinBlk, SBrinBlock *brinBlock);
+// .data
+int32_t tsdbDataFileReadBlockData(SDataFileReader *reader, const SBrinRecord *record, SBlockData *bData);
+int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRecord *record, SBlockData *bData,
+                                          STSchema *pTSchema, int16_t cids[], int32_t ncid);
+// .sma
+int32_t tsdbDataFileReadBlockSma(SDataFileReader *reader, const SBrinRecord *record,
+                                 TColumnDataAggArray *columnDataAggArray);
+// .tomb
+int32_t tsdbDataFileReadTombBlk(SDataFileReader *reader, const TTombBlkArray **tombBlkArray);
+int32_t tsdbDataFileReadTombBlock(SDataFileReader *reader, const STombBlk *tombBlk, STombBlock *tData);
+
+// SDataFileWriter =============================================
+typedef struct SDataFileWriter SDataFileWriter;
+typedef struct SDataFileWriterConfig {
+  STsdb  *tsdb;
+  int8_t  cmprAlg;
+  int32_t maxRow;
+  int32_t szPage;
+  int32_t fid;
+  int64_t cid;
+  SDiskID did;
+  int64_t compactVersion;
+  struct {
+    bool   exist;
+    STFile file;
+  } files[TSDB_FTYPE_MAX];
+  SSkmInfo *skmTb;
+  SSkmInfo *skmRow;
+  uint8_t **bufArr;
+} SDataFileWriterConfig;
+
+int32_t tsdbDataFileWriterOpen(const SDataFileWriterConfig *config, SDataFileWriter **writer);
+int32_t tsdbDataFileWriterClose(SDataFileWriter **writer, bool abort, TFileOpArray *opArr);
+
+int32_t tsdbDataFileWriteRow(SDataFileWriter *writer, SRowInfo *row);
+int32_t tsdbDataFileWriteBlockData(SDataFileWriter *writer, SBlockData *bData);
+int32_t tsdbDataFileFlush(SDataFileWriter *writer);
+
+int32_t tsdbDataFileWriteTombRecord(SDataFileWriter *writer, const STombRecord *record);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TSDB_DATA_FILE_RW_H*/
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbFile.c

@@ -284,8 +284,9 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
 // SDelFile ===============================================
 void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]) {
   int32_t offset = 0;
+  SVnode *pVnode = pTsdb->pVnode;
 
-  vnodeGetPrimaryDir(pTsdb->path, pTsdb->pVnode->pTfs, fname, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pTsdb->path, pVnode->diskPrimary, pVnode->pTfs, fname, TSDB_FILENAME_LEN);
   offset = strlen(fname);
   snprintf((char *)fname + offset, TSDB_FILENAME_LEN - offset - 1, "%sv%dver%" PRId64 ".del", TD_DIRSEP,
            TD_VID(pTsdb->pVnode), pFile->commitID);