taosdata
diff --git a/‎include/common/streamMsg.h‎
Lines changed: 22 additions & 0 deletions b/‎include/common/streamMsg.h‎
Lines changed: 22 additions & 0 deletions
diff --git a/‎include/common/tdataformat.h‎
Lines changed: 7 additions & 1 deletion b/‎include/common/tdataformat.h‎
Lines changed: 7 additions & 1 deletion
diff --git a/‎include/common/tmsg.h‎
Lines changed: 24 additions & 0 deletions b/‎include/common/tmsg.h‎
Lines changed: 24 additions & 0 deletions
diff --git a/‎include/common/tmsgdef.h‎
Lines changed: 2 additions & 1 deletion b/‎include/common/tmsgdef.h‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎include/libs/decimal/decimal.h‎
Lines changed: 3 additions & 0 deletions b/‎include/libs/decimal/decimal.h‎
Lines changed: 3 additions & 0 deletions
diff --git a/‎include/libs/function/functionMgt.h‎
Lines changed: 1 addition & 0 deletions b/‎include/libs/function/functionMgt.h‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎include/libs/new-stream/stream.h‎
Lines changed: 1 addition & 0 deletions b/‎include/libs/new-stream/stream.h‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎include/libs/new-stream/streamReader.h‎
Lines changed: 1 addition & 1 deletion b/‎include/libs/new-stream/streamReader.h‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎include/libs/nodes/cmdnodes.h‎
Lines changed: 1 addition & 1 deletion b/‎include/libs/nodes/cmdnodes.h‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎include/libs/nodes/querynodes.h‎
Lines changed: 0 additions & 1 deletion b/‎include/libs/nodes/querynodes.h‎
Lines changed: 0 additions & 1 deletion
@@ -716,6 +716,7 @@ typedef enum ESTriggerPullType {
   STRIGGER_PULL_WAL_DATA,
   STRIGGER_PULL_GROUP_COL_VALUE,
   STRIGGER_PULL_VTABLE_INFO,
+  STRIGGER_PULL_VTABLE_PSEUDO_COL,
   STRIGGER_PULL_OTABLE_INFO,
   STRIGGER_PULL_TYPE_MAX,
 } ESTriggerPullType;
@@ -806,6 +807,11 @@ typedef struct SSTriggerVirTableInfoRequest {
   SArray*              cids;  // SArray<col_id_t>, col ids of the virtual table
 } SSTriggerVirTableInfoRequest;
 
+typedef struct SSTriggerVirTablePseudoColRequest {
+  SSTriggerPullRequest base;
+  int64_t              uid;
+  SArray*              cids;  // SArray<col_id_t>, -1 means tbname
+} SSTriggerVirTablePseudoColRequest;
 typedef struct OTableInfoRsp {
   int64_t  suid;
   int64_t  uid;
@@ -844,6 +850,7 @@ typedef union SSTriggerPullRequestUnion {
   SSTriggerWalDataRequest             walDataReq;
   SSTriggerGroupColValueRequest       groupColValueReq;
   SSTriggerVirTableInfoRequest        virTableInfoReq;
+  SSTriggerVirTablePseudoColRequest   virTablePseudoColReq;
   SSTriggerOrigTableInfoRequest       origTableInfoReq;
 } SSTriggerPullRequestUnion;
 
@@ -900,6 +907,21 @@ int32_t tDeserializeSTriggerCalcRequest(void* buf, int32_t bufLen, SSTriggerCalc
 void    tDestroySSTriggerCalcParam(void* ptr);
 void    tDestroySTriggerCalcRequest(SSTriggerCalcRequest* pReq);
 
+typedef enum ESTriggerCtrlType {
+  STRIGGER_CTRL_START = 0,
+  STRIGGER_CTRL_STOP = 1,
+} ESTriggerCtrlType;
+
+typedef struct SSTriggerCtrlRequest {
+  ESTriggerCtrlType type;
+  int64_t           streamId;
+  int64_t           taskId;
+  int64_t           sessionId;
+} SSTriggerCtrlRequest;
+
+int32_t tSerializeSTriggerCtrlRequest(void* buf, int32_t bufLen, const SSTriggerCtrlRequest* pReq);
+int32_t tDeserializeSTriggerCtrlRequest(void* buf, int32_t bufLen, SSTriggerCtrlRequest* pReq);
+
 typedef struct SStreamRuntimeFuncInfo {
   SArray* pStreamPesudoFuncVals;
   SArray* pStreamPartColVals;
 
@@ -107,6 +107,12 @@ const static uint8_t BIT2_MAP[4] = {0b11111100, 0b11110011, 0b11001111, 0b001111
 #define COL_VAL_IS_NULL(CV)  ((CV)->flag == CV_FLAG_NULL)
 #define COL_VAL_IS_VALUE(CV) ((CV)->flag == CV_FLAG_VALUE)
 
+// Strategies of merging rows with
+// same pk in single insert batch.
+typedef enum {
+  PREFER_NON_NULL = 0,  // choose latest non-null value for each column
+  KEEP_CONSISTENCY = 1  // choose latest row
+} ERowMergeStrategy;
 #define BSE_SEQUECE_SIZE sizeof(uint64_t)
 
 enum { TSDB_DATA_BLOB_VALUE = 0x1, TSDB_DATA_BLOB_EMPTY_VALUE = 0x2, TSDB_DATA_BLOB_NULL_VALUE = 0x4 };
@@ -177,7 +183,7 @@ void    tBlobSetSwap(SBlobSet *p1, SBlobSet *p2);
 int32_t tRowGetBlobSeq(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal, uint64_t *seq);
 void    tRowDestroy(SRow *pRow);
 int32_t tRowSort(SArray *aRowP);
-int32_t tRowMerge(SArray *aRowP, STSchema *pTSchema, int8_t flag);
+int32_t tRowMerge(SArray *aRowP, STSchema *pTSchema, ERowMergeStrategy strategy);
 int32_t tRowUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *aColData, int32_t nColData, int32_t flag);
 void    tRowGetPrimaryKey(SRow *pRow, SRowKey *key);
 int32_t tRowKeyCompare(const SRowKey *key1, const SRowKey *key2);
 
@@ -515,10 +515,19 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL,
   QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL,  // INACTIVE
   QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_1,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_2,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_3,
   QUERY_NODE_PHYSICAL_PLAN_FILL,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_4,
   QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_5,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_6,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_7,
   QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_8,
   QUERY_NODE_PHYSICAL_PLAN_PARTITION,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_9,
   QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC,
   QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC,
   QUERY_NODE_PHYSICAL_PLAN_DISPATCH,
@@ -529,12 +538,27 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN,
   QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN,
   QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_10,
   QUERY_NODE_PHYSICAL_PLAN_HASH_JOIN,
   QUERY_NODE_PHYSICAL_PLAN_GROUP_CACHE,
   QUERY_NODE_PHYSICAL_PLAN_DYN_QUERY_CTRL,
   QUERY_NODE_PHYSICAL_PLAN_MERGE_COUNT,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_11,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_12,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_13,
   QUERY_NODE_PHYSICAL_PLAN_MERGE_ANOMALY,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_14,
   QUERY_NODE_PHYSICAL_PLAN_FORECAST_FUNC,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_15,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_16,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_17,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_18,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_19,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_20,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_21,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_22,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_23,
+  QUERY_NODE_PHYSICAL_PLAN_UNUSED_24,
   QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN,
   QUERY_NODE_PHYSICAL_PLAN_EXTERNAL_WINDOW,
   QUERY_NODE_PHYSICAL_PLAN_HASH_EXTERNAL,
 
@@ -376,8 +376,9 @@
   TD_DEF_MSG_TYPE(TDMT_STREAM_UNUSED2, "stream-unused2", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_RETRIEVE_TRIGGER, "stream-retri-trigger", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_CONSEN_CHKPT, "stream-consen-chkpt", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_STREAM_CHKPT_EXEC, "stream-exec-chkpt", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_STREAM_CHKPT_EXEC, "stream-exec-chkpt", NULL, NULL)  //1059 1060
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_START, "stream-task-start", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_STREAM_TRIGGER_CTRL, "stream-trigger-ctl", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TRIGGER_PULL, "stream-trigger-pull", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TRIGGER_CALC, "stream_trigger-calc", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_FETCH_FROM_RUNNER, "stream-fetch-runner", NULL, NULL)
 
@@ -136,6 +136,9 @@ int32_t TEST_decimal128FromDecimal64(const Decimal64* pInput, uint8_t inputPrec,
 int32_t TEST_decimal128FromDecimal128(const Decimal128* pDec, uint8_t prec, uint8_t scale, Decimal128* pOutput,
                                       uint8_t outputPrec, uint8_t outputScale);
 
+int32_t decimal64ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen);
+int32_t decimal128ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen);
+
 #ifdef __cplusplus
 }
 #endif
 
@@ -272,6 +272,7 @@ EFunctionType fmGetFuncType(const char* pFunc);
 bool fmIsAggFunc(int32_t funcId);
 bool fmIsScalarFunc(int32_t funcId);
 bool fmIsVectorFunc(int32_t funcId);
+bool fmIsStreamVectorFunc(int32_t funcId);
 bool fmIsIndefiniteRowsFunc(int32_t funcId);
 bool fmIsStringFunc(int32_t funcId);
 bool fmIsDateTimeFunc(int32_t funcId);
 
@@ -103,6 +103,7 @@ typedef struct SStreamRunnerTaskExecMgr {
   SList*        pFreeExecs;
   SList*        pRunningExecs;
   TdThreadMutex lock;
+  bool          lockInited;
 } SStreamRunnerTaskExecMgr;
 
 typedef struct SStreamTagInfo {
 
@@ -102,7 +102,7 @@ int32_t createStreamTask(void* pVnode, SStreamTriggerReaderTaskInnerOptions* opt
                          SSDataBlock* pResBlock, SHashObj* groupIdMap, SStorageAPI*  api);
 void*   qStreamGetReaderInfo(int64_t streamId, int64_t taskId, void** taskAddr);
 void    qStreamSetTaskRunning(int64_t streamId, int64_t taskId);
-int32_t streamBuildFetchRsp(SSDataBlock* pBlock, void** data, size_t* size, int8_t precision);
+int32_t streamBuildFetchRsp(SArray* pResList, bool hasNext, void** data, size_t* size, int8_t precision);
 
 #ifdef __cplusplus
 }
 
@@ -26,7 +26,7 @@ extern "C" {
 #define DESCRIBE_RESULT_COLS               4
 #define DESCRIBE_RESULT_COLS_COMPRESS      7
 #define DESCRIBE_RESULT_COLS_REF           5
-#define DESCRIBE_RESULT_FIELD_LEN          (TSDB_COL_NAME_LEN - 1 + VARSTR_HEADER_SIZE)
+#define DESCRIBE_RESULT_FIELD_LEN          (TSDB_DB_NAME_LEN + TSDB_NAME_DELIMITER_LEN + TSDB_COL_FNAME_LEN + VARSTR_HEADER_SIZE)
 #define DESCRIBE_RESULT_TYPE_LEN           (20 + VARSTR_HEADER_SIZE)
 #define DESCRIBE_RESULT_NOTE_LEN           (16 + VARSTR_HEADER_SIZE)
 #define DESCRIBE_RESULT_COPRESS_OPTION_LEN (TSDB_CL_COMPRESS_OPTION_LEN + VARSTR_HEADER_SIZE)
 
@@ -456,7 +456,6 @@ typedef struct SStreamTagDefNode {
   char      tagName[TSDB_COL_NAME_LEN];
   SDataType dataType;
   SNode*    pTagExpr;
-  SNode*    pComment;
 } SStreamTagDefNode;
 
 typedef struct SPeriodWindowNode {
Original file line number	Diff line number	Diff line change
`@@ -136,6 +136,9 @@ int32_t TEST_decimal128FromDecimal64(const Decimal64* pInput, uint8_t inputPrec,`
`136`	`136`	`int32_t TEST_decimal128FromDecimal128(const Decimal128* pDec, uint8_t prec, uint8_t scale, Decimal128* pOutput,`
`137`	`137`	`uint8_t outputPrec, uint8_t outputScale);`
`138`	`138`
	`139`	`+int32_t decimal64ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen);`
	`140`	`+int32_t decimal128ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen);`
	`141`	`+`
`139`	`142`	`#ifdef __cplusplus`
`140`	`143`	`}`
`141`	`144`	`#endif`
Original file line number	Diff line number	Diff line change
`@@ -102,7 +102,7 @@ int32_t createStreamTask(void* pVnode, SStreamTriggerReaderTaskInnerOptions* opt`
`102`	`102`	`SSDataBlock* pResBlock, SHashObj* groupIdMap, SStorageAPI* api);`
`103`	`103`	`void* qStreamGetReaderInfo(int64_t streamId, int64_t taskId, void** taskAddr);`
`104`	`104`	`void qStreamSetTaskRunning(int64_t streamId, int64_t taskId);`
`105`		`-int32_t streamBuildFetchRsp(SSDataBlock* pBlock, void** data, size_t* size, int8_t precision);`
	`105`	`+int32_t streamBuildFetchRsp(SArray* pResList, bool hasNext, void** data, size_t* size, int8_t precision);`
`106`	`106`
`107`	`107`	`#ifdef __cplusplus`
`108`	`108`	`}`