Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/stream_compression

docs/en/14-reference/03-connector/06-rust.mdx

@@ -116,7 +116,7 @@ The parameters are described as follows:
 - **protocol**: Specify which connection method to use. For example, `taos+ws://localhost:6041` uses Websocket to establish connections.
 - **username/password**: Username and password used to create connections.
 - **host/port**: Specifies the server and port to establish a connection. If you do not specify a hostname or port, native connections default to `localhost:6030` and Websocket connections default to `localhost:6041`.
-- **database**: Specify the default database to connect to.
+- **database**: Specify the default database to connect to. It's optional.
 - **params**：Optional parameters.
 
 A sample DSN description string is as follows:

docs/en/14-reference/03-connector/09-csharp.mdx

@@ -151,7 +151,7 @@ The parameters are described as follows:
 * **protocol**: Specify which connection method to use (support http/ws). For example, `ws://localhost:6041` uses Websocket to establish connections. 
 * **username/password**: Username and password used to create connections.
 * **host/port**: Specifies the server and port to establish a connection. Websocket connections default to `localhost:6041`.
-* **database**: Specify the default database to connect to.
+* **database**: Specify the default database to connect to. It's optional.
 * **params**：Optional parameters.
 
 A sample DSN description string is as follows:

docs/zh/08-connector/26-rust.mdx

@@ -116,7 +116,7 @@ DSN 描述字符串基本结构如下：
 - **protocol**: 显示指定以何种方式建立连接，例如：`taos+ws://localhost:6041` 指定以 Websocket 方式建立连接。
 - **username/password**: 用于创建连接的用户名及密码。
 - **host/port**: 指定创建连接的服务器及端口，当不指定服务器地址及端口时（`taos://`），原生连接默认为 `localhost:6030`，Websocket 连接默认为 `localhost:6041` 。
-- **database**: 指定默认连接的数据库名。
+- **database**: 指定默认连接的数据库名，可选参数。
 - **params**：其他可选参数。
 
 一个完整的 DSN 描述字符串示例如下：

docs/zh/08-connector/40-csharp.mdx

@@ -154,7 +154,7 @@ namespace TDengineExample
   
 * **host/port**: 指定创建连接的服务器及端口，WebSocket 连接默认为 `localhost:6041` 。
   
-* **database**: 指定默认连接的数据库名。
+* **database**: 指定默认连接的数据库名，可选参数。
 
 * **params**：其他可选参数。
 

include/common/ttokendef.h

@@ -234,96 +234,100 @@
 #define TK_CURRENT_USER         216
 #define TK_COUNT                217
 #define TK_LAST_ROW             218
-#define TK_BETWEEN              219
-#define TK_IS                   220
-#define TK_NK_LT                221
-#define TK_NK_GT                222
-#define TK_NK_LE                223
-#define TK_NK_GE                224
-#define TK_NK_NE                225
-#define TK_MATCH                226
-#define TK_NMATCH               227
-#define TK_CONTAINS             228
-#define TK_IN                   229
-#define TK_JOIN                 230
-#define TK_INNER                231
-#define TK_SELECT               232
-#define TK_DISTINCT             233
-#define TK_WHERE                234
-#define TK_PARTITION            235
-#define TK_BY                   236
-#define TK_SESSION              237
-#define TK_STATE_WINDOW         238
-#define TK_SLIDING              239
-#define TK_FILL                 240
-#define TK_VALUE                241
-#define TK_NONE                 242
-#define TK_PREV                 243
-#define TK_LINEAR               244
-#define TK_NEXT                 245
-#define TK_HAVING               246
-#define TK_RANGE                247
-#define TK_EVERY                248
-#define TK_ORDER                249
-#define TK_SLIMIT               250
-#define TK_SOFFSET              251
-#define TK_LIMIT                252
-#define TK_OFFSET               253
-#define TK_ASC                  254
-#define TK_NULLS                255
-#define TK_ABORT                256
-#define TK_AFTER                257
-#define TK_ATTACH               258
-#define TK_BEFORE               259
-#define TK_BEGIN                260
-#define TK_BITAND               261
-#define TK_BITNOT               262
-#define TK_BITOR                263
-#define TK_BLOCKS               264
-#define TK_CHANGE               265
-#define TK_COMMA                266
-#define TK_COMPACT              267
-#define TK_CONCAT               268
-#define TK_CONFLICT             269
-#define TK_COPY                 270
-#define TK_DEFERRED             271
-#define TK_DELIMITERS           272
-#define TK_DETACH               273
-#define TK_DIVIDE               274
-#define TK_DOT                  275
-#define TK_EACH                 276
-#define TK_END                  277
-#define TK_FAIL                 278
-#define TK_FILE                 279
-#define TK_FOR                  280
-#define TK_GLOB                 281
-#define TK_ID                   282
-#define TK_IMMEDIATE            283
-#define TK_IMPORT               284
-#define TK_INITIALLY            285
-#define TK_INSTEAD              286
-#define TK_ISNULL               287
-#define TK_KEY                  288
-#define TK_NK_BITNOT            289
-#define TK_NK_SEMI              290
-#define TK_NOTNULL              291
-#define TK_OF                   292
-#define TK_PLUS                 293
-#define TK_PRIVILEGE            294
-#define TK_RAISE                295
-#define TK_REPLACE              296
-#define TK_RESTRICT             297
-#define TK_ROW                  298
-#define TK_SEMI                 299
-#define TK_STAR                 300
-#define TK_STATEMENT            301
-#define TK_STRING               302
-#define TK_TIMES                303
-#define TK_UPDATE               304
-#define TK_VALUES               305
-#define TK_VARIABLE             306
-#define TK_VIEW                 307
-#define TK_WAL                  308
+#define TK_CASE                 219
+#define TK_END                  220
+#define TK_WHEN                 221
+#define TK_THEN                 222
+#define TK_ELSE                 223
+#define TK_BETWEEN              224
+#define TK_IS                   225
+#define TK_NK_LT                226
+#define TK_NK_GT                227
+#define TK_NK_LE                228
+#define TK_NK_GE                229
+#define TK_NK_NE                230
+#define TK_MATCH                231
+#define TK_NMATCH               232
+#define TK_CONTAINS             233
+#define TK_IN                   234
+#define TK_JOIN                 235
+#define TK_INNER                236
+#define TK_SELECT               237
+#define TK_DISTINCT             238
+#define TK_WHERE                239
+#define TK_PARTITION            240
+#define TK_BY                   241
+#define TK_SESSION              242
+#define TK_STATE_WINDOW         243
+#define TK_SLIDING              244
+#define TK_FILL                 245
+#define TK_VALUE                246
+#define TK_NONE                 247
+#define TK_PREV                 248
+#define TK_LINEAR               249
+#define TK_NEXT                 250
+#define TK_HAVING               251
+#define TK_RANGE                252
+#define TK_EVERY                253
+#define TK_ORDER                254
+#define TK_SLIMIT               255
+#define TK_SOFFSET              256
+#define TK_LIMIT                257
+#define TK_OFFSET               258
+#define TK_ASC                  259
+#define TK_NULLS                260
+#define TK_ABORT                261
+#define TK_AFTER                262
+#define TK_ATTACH               263
+#define TK_BEFORE               264
+#define TK_BEGIN                265
+#define TK_BITAND               266
+#define TK_BITNOT               267
+#define TK_BITOR                268
+#define TK_BLOCKS               269
+#define TK_CHANGE               270
+#define TK_COMMA                271
+#define TK_COMPACT              272
+#define TK_CONCAT               273
+#define TK_CONFLICT             274
+#define TK_COPY                 275
+#define TK_DEFERRED             276
+#define TK_DELIMITERS           277
+#define TK_DETACH               278
+#define TK_DIVIDE               279
+#define TK_DOT                  280
+#define TK_EACH                 281
+#define TK_FAIL                 282
+#define TK_FILE                 283
+#define TK_FOR                  284
+#define TK_GLOB                 285
+#define TK_ID                   286
+#define TK_IMMEDIATE            287
+#define TK_IMPORT               288
+#define TK_INITIALLY            289
+#define TK_INSTEAD              290
+#define TK_ISNULL               291
+#define TK_KEY                  292
+#define TK_NK_BITNOT            293
+#define TK_NK_SEMI              294
+#define TK_NOTNULL              295
+#define TK_OF                   296
+#define TK_PLUS                 297
+#define TK_PRIVILEGE            298
+#define TK_RAISE                299
+#define TK_REPLACE              300
+#define TK_RESTRICT             301
+#define TK_ROW                  302
+#define TK_SEMI                 303
+#define TK_STAR                 304
+#define TK_STATEMENT            305
+#define TK_STRING               306
+#define TK_TIMES                307
+#define TK_UPDATE               308
+#define TK_VALUES               309
+#define TK_VARIABLE             310
+#define TK_VIEW                 311
+#define TK_WAL                  312
 
 #define TK_NK_SPACE   300
 #define TK_NK_COMMENT 301

include/libs/nodes/nodes.h

@@ -103,6 +103,8 @@ typedef enum ENodeType {
   QUERY_NODE_STREAM_OPTIONS,
   QUERY_NODE_LEFT_VALUE,
   QUERY_NODE_COLUMN_REF,
+  QUERY_NODE_WHEN_THEN,
+  QUERY_NODE_CASE_WHEN,
 
   // Statement nodes are used in parser and planner module.
   QUERY_NODE_SET_OPERATOR = 100,

include/libs/nodes/querynodes.h

@@ -241,6 +241,19 @@ typedef struct SFillNode {
   STimeWindow timeRange;
 } SFillNode;
 
+typedef struct SWhenThenNode {
+  SExprNode node;  // QUERY_NODE_WHEN_THEN
+  SNode*    pWhen;
+  SNode*    pThen;
+} SWhenThenNode;
+
+typedef struct SCaseWhenNode {
+  SExprNode  node;  // QUERY_NODE_CASE_WHEN
+  SNode*     pCase;
+  SNode*     pElse;
+  SNodeList* pWhenThenList;
+} SCaseWhenNode;
+
 typedef struct SSelectStmt {
   ENodeType   type;  // QUERY_NODE_SELECT_STMT
   bool        isDistinct;

source/common/src/tglobal.c

@@ -58,7 +58,7 @@ int32_t tsNumOfMnodeFetchThreads = 1;
 int32_t tsNumOfMnodeReadThreads = 1;
 int32_t tsNumOfVnodeQueryThreads = 4;
 int32_t tsNumOfVnodeStreamThreads = 2;
-int32_t tsNumOfVnodeFetchThreads = 4;
+int32_t tsNumOfVnodeFetchThreads = 1;
 int32_t tsNumOfVnodeWriteThreads = 2;
 int32_t tsNumOfVnodeSyncThreads = 2;
 int32_t tsNumOfVnodeRsmaThreads = 2;
@@ -165,8 +165,8 @@ int32_t tsMqRebalanceInterval = 2;
 int32_t tsTtlUnit = 86400;
 int32_t tsTtlPushInterval = 86400;
 int32_t tsGrantHBInterval = 60;
-int32_t tsUptimeInterval = 300;  // seconds
-char    tsUdfdResFuncs[1024] = ""; // udfd resident funcs that teardown when udfd exits
+int32_t tsUptimeInterval = 300;     // seconds
+char    tsUdfdResFuncs[1024] = "";  // udfd resident funcs that teardown when udfd exits
 
 #ifndef _STORAGE
 int32_t taosSetTfsCfg(SConfig *pCfg) {
@@ -371,9 +371,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   tsNumOfVnodeStreamThreads = TMAX(tsNumOfVnodeStreamThreads, 4);
   if (cfgAddInt32(pCfg, "numOfVnodeStreamThreads", tsNumOfVnodeStreamThreads, 4, 1024, 0) != 0) return -1;
 
-  tsNumOfVnodeFetchThreads = tsNumOfCores / 4;
-  tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4);
-  if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 4, 1024, 0) != 0) return -1;
+  tsNumOfVnodeFetchThreads = 1;
+  if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 1, 1024, 0) != 0) return -1;
 
   tsNumOfVnodeWriteThreads = tsNumOfCores;
   tsNumOfVnodeWriteThreads = TMAX(tsNumOfVnodeWriteThreads, 1);

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

@@ -479,6 +479,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
     tqInitDataRsp(&dataRsp, pReq, pHandle->execHandle.subType);
     tqScanData(pTq, pHandle, &dataRsp, &fetchOffsetNew);
 
+#if 1
+
+#endif
     if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
       code = -1;
     }

source/libs/executor/src/executor.c

@@ -49,6 +49,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
 
     SStreamScanInfo* pInfo = pOperator->info;
 
+#if 0
     // TODO: if a block was set but not consumed,
     // prevent setting a different type of block
     pInfo->validBlockIndex = 0;
@@ -57,6 +58,10 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
     } else {
       taosArrayClear(pInfo->pBlockLists);
     }
+#endif
+
+    ASSERT(pInfo->validBlockIndex == 0);
+    ASSERT(taosArrayGetSize(pInfo->pBlockLists) == 0);
 
     if (type == STREAM_INPUT__MERGED_SUBMIT) {
       // ASSERT(numOfBlocks > 1);
@@ -79,7 +84,9 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
     } else if (type == STREAM_INPUT__DATA_BLOCK) {
       for (int32_t i = 0; i < numOfBlocks; ++i) {
         SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];
+        taosArrayPush(pInfo->pBlockLists, &pDataBlock);
 
+#if 0
         // TODO optimize
         SSDataBlock* p = createOneDataBlock(pDataBlock, false);
         p->info = pDataBlock->info;
@@ -87,6 +94,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
         taosArrayClear(p->pDataBlock);
         taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock);
         taosArrayPush(pInfo->pBlockLists, &p);
+#endif
       }
       pInfo->blockType = STREAM_INPUT__DATA_BLOCK;
     } else {
@@ -100,6 +108,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
 static FORCE_INLINE void streamInputBlockDataDestory(void* pBlock) { blockDataDestroy((SSDataBlock*)pBlock); }
 
 void tdCleanupStreamInputDataBlock(qTaskInfo_t tinfo) {
+#if 0
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
   if (!pTaskInfo || !pTaskInfo->pRoot || pTaskInfo->pRoot->numOfDownstream <= 0) {
     return;
@@ -116,6 +125,7 @@ void tdCleanupStreamInputDataBlock(qTaskInfo_t tinfo) {
   } else {
     ASSERT(0);
   }
+#endif
 }
 
 int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {

source/libs/executor/src/scanoperator.c

@@ -920,6 +920,19 @@ _error:
 }
 
 static void doClearBufferedBlocks(SStreamScanInfo* pInfo) {
+#if 0
+  if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
+    size_t total = taosArrayGetSize(pInfo->pBlockLists);
+    for (int32_t i = 0; i < total; i++) {
+      SSDataBlock* p = taosArrayGetP(pInfo->pBlockLists, i);
+      taosArrayDestroy(p->pDataBlock);
+      taosMemoryFree(p);
+    }
+  }
+#endif
+  taosArrayClear(pInfo->pBlockLists);
+  pInfo->validBlockIndex = 0;
+#if 0
   size_t total = taosArrayGetSize(pInfo->pBlockLists);
 
   pInfo->validBlockIndex = 0;
@@ -928,6 +941,7 @@ static void doClearBufferedBlocks(SStreamScanInfo* pInfo) {
     blockDataDestroy(p);
   }
   taosArrayClear(pInfo->pBlockLists);
+#endif
 }
 
 static bool isSessionWindow(SStreamScanInfo* pInfo) {
@@ -1580,9 +1594,10 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
 
   size_t total = taosArrayGetSize(pInfo->pBlockLists);
   // TODO: refactor
+FETCH_NEXT_BLOCK:
   if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
     if (pInfo->validBlockIndex >= total) {
-      /*doClearBufferedBlocks(pInfo);*/
+      doClearBufferedBlocks(pInfo);
       /*pOperator->status = OP_EXEC_DONE;*/
       return NULL;
     }
@@ -1606,27 +1621,40 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
       } break;
       case STREAM_DELETE_DATA: {
         printDataBlock(pBlock, "stream scan delete recv");
+        SSDataBlock* pDelBlock = NULL;
         if (pInfo->tqReader) {
-          SSDataBlock* pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);
+          pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);
           filterDelBlockByUid(pDelBlock, pBlock, pInfo);
-          pBlock = pDelBlock;
+        } else {
+          pDelBlock = pBlock;
         }
         printDataBlock(pBlock, "stream scan delete recv filtered");
         if (!isIntervalWindow(pInfo) && !isSessionWindow(pInfo) && !isStateWindow(pInfo)) {
-          generateDeleteResultBlock(pInfo, pBlock, pInfo->pDeleteDataRes);
+          generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes);
           pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT;
-          printDataBlock(pBlock, "stream scan delete result");
-          return pInfo->pDeleteDataRes;
+          printDataBlock(pDelBlock, "stream scan delete result");
+          if (pInfo->pDeleteDataRes->info.rows > 0) {
+            return pInfo->pDeleteDataRes;
+          } else {
+            goto FETCH_NEXT_BLOCK;
+          }
         } else {
           pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
           pInfo->updateResIndex = 0;
-          generateScanRange(pInfo, pBlock, pInfo->pUpdateRes);
+          generateScanRange(pInfo, pDelBlock, pInfo->pUpdateRes);
           prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
           copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
           pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
           pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
-          printDataBlock(pBlock, "stream scan delete data");
-          return pInfo->pDeleteDataRes;
+          printDataBlock(pDelBlock, "stream scan delete data");
+          if (pInfo->tqReader) {
+            blockDataDestroy(pDelBlock);
+          }
+          if (pInfo->pDeleteDataRes->info.rows > 0) {
+            return pInfo->pDeleteDataRes;
+          } else {
+            goto FETCH_NEXT_BLOCK;
+          }
         }
       } break;
       default:
@@ -1688,10 +1716,12 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
 
     int32_t totBlockNum = taosArrayGetSize(pInfo->pBlockLists);
 
+  NEXT_SUBMIT_BLK:
     while (1) {
       if (pInfo->tqReader->pMsg == NULL) {
         if (pInfo->validBlockIndex >= totBlockNum) {
           updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
+          doClearBufferedBlocks(pInfo);
           return NULL;
         }
 
@@ -1763,7 +1793,12 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
     }
 
     qDebug("scan rows: %d", pBlockInfo->rows);
-    return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
+    if (pBlockInfo->rows > 0) {
+      return pInfo->pRes;
+    } else {
+      goto NEXT_SUBMIT_BLK;
+    }
+    /*return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;*/
   } else {
     ASSERT(0);
     return NULL;

source/libs/executor/src/timewindowoperator.c

@@ -5658,7 +5658,6 @@ static void doStreamIntervalAggImpl2(SOperatorInfo* pOperatorInfo, SSDataBlock*
   TSKEY*          tsCols = NULL;
   SResultRow*     pResult = NULL;
   int32_t         forwardRows = 0;
-  int32_t         aa = 4;
 
   ASSERT(pSDataBlock->pDataBlock != NULL);
   SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);

source/libs/nodes/src/nodesCloneFuncs.c

@@ -324,6 +324,21 @@ static int32_t fillNodeCopy(const SFillNode* pSrc, SFillNode* pDst) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t whenThenNodeCopy(const SWhenThenNode* pSrc, SWhenThenNode* pDst) {
+  COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
+  CLONE_NODE_FIELD(pWhen);
+  CLONE_NODE_FIELD(pThen);
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t caseWhenNodeCopy(const SCaseWhenNode* pSrc, SCaseWhenNode* pDst) {
+  COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
+  CLONE_NODE_FIELD(pCase);
+  CLONE_NODE_FIELD(pElse);
+  CLONE_NODE_LIST_FIELD(pWhenThenList);
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t logicNodeCopy(const SLogicNode* pSrc, SLogicNode* pDst) {
   CLONE_NODE_LIST_FIELD(pTargets);
   CLONE_NODE_FIELD(pConditions);
@@ -711,6 +726,12 @@ SNode* nodesCloneNode(const SNode* pNode) {
     case QUERY_NODE_LEFT_VALUE:
       code = TSDB_CODE_SUCCESS;
       break;
+    case QUERY_NODE_WHEN_THEN:
+      code = whenThenNodeCopy((const SWhenThenNode*)pNode, (SWhenThenNode*)pDst);
+      break;
+    case QUERY_NODE_CASE_WHEN:
+      code = caseWhenNodeCopy((const SCaseWhenNode*)pNode, (SCaseWhenNode*)pDst);
+      break;
     case QUERY_NODE_SELECT_STMT:
       code = selectStmtCopy((const SSelectStmt*)pNode, (SSelectStmt*)pDst);
       break;

source/libs/nodes/src/nodesCodeFuncs.c

@@ -81,6 +81,10 @@ const char* nodesNodeName(ENodeType type) {
       return "IndexOptions";
     case QUERY_NODE_LEFT_VALUE:
       return "LeftValue";
+    case QUERY_NODE_WHEN_THEN:
+      return "WhenThen";
+    case QUERY_NODE_CASE_WHEN:
+      return "CaseWhen";
     case QUERY_NODE_SET_OPERATOR:
       return "SetOperator";
     case QUERY_NODE_SELECT_STMT:
@@ -3917,6 +3921,75 @@ static int32_t jsonToDatabaseOptions(const SJson* pJson, void* pObj) {
   return code;
 }
 
+static const char* jkWhenThenWhen = "When";
+static const char* jkWhenThenThen = "Then";
+
+static int32_t whenThenNodeToJson(const void* pObj, SJson* pJson) {
+  const SWhenThenNode* pNode = (const SWhenThenNode*)pObj;
+
+  int32_t code = exprNodeToJson(pObj, pJson);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkWhenThenWhen, nodeToJson, pNode->pWhen);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkWhenThenThen, nodeToJson, pNode->pThen);
+  }
+
+  return code;
+}
+
+static int32_t jsonToWhenThenNode(const SJson* pJson, void* pObj) {
+  SWhenThenNode* pNode = (SWhenThenNode*)pObj;
+
+  int32_t code = jsonToExprNode(pJson, pObj);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkWhenThenWhen, &pNode->pWhen);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkWhenThenThen, &pNode->pThen);
+  }
+
+  return code;
+}
+
+static const char* jkCaseWhenCase = "Case";
+static const char* jkCaseWhenWhenThenList = "WhenThenList";
+static const char* jkCaseWhenElse = "Else";
+
+static int32_t caseWhenNodeToJson(const void* pObj, SJson* pJson) {
+  const SCaseWhenNode* pNode = (const SCaseWhenNode*)pObj;
+
+  int32_t code = exprNodeToJson(pObj, pJson);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkCaseWhenCase, nodeToJson, pNode->pCase);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodeListToJson(pJson, jkCaseWhenWhenThenList, pNode->pWhenThenList);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkCaseWhenElse, nodeToJson, pNode->pElse);
+  }
+
+  return code;
+}
+
+static int32_t jsonToCaseWhenNode(const SJson* pJson, void* pObj) {
+  SCaseWhenNode* pNode = (SCaseWhenNode*)pObj;
+
+  int32_t code = jsonToExprNode(pJson, pObj);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkCaseWhenCase, &pNode->pCase);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeList(pJson, jkCaseWhenWhenThenList, &pNode->pWhenThenList);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkCaseWhenElse, &pNode->pElse);
+  }
+
+  return code;
+}
+
 static const char* jkDataBlockDescDataBlockId = "DataBlockId";
 static const char* jkDataBlockDescSlots = "Slots";
 static const char* jkDataBlockTotalRowSize = "TotalRowSize";
@@ -4399,6 +4472,10 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
       return databaseOptionsToJson(pObj, pJson);
     case QUERY_NODE_LEFT_VALUE:
       return TSDB_CODE_SUCCESS;  // SLeftValueNode has no fields to serialize.
+    case QUERY_NODE_WHEN_THEN:
+      return whenThenNodeToJson(pObj, pJson);
+    case QUERY_NODE_CASE_WHEN:
+      return caseWhenNodeToJson(pObj, pJson);
     case QUERY_NODE_SET_OPERATOR:
       return setOperatorToJson(pObj, pJson);
     case QUERY_NODE_SELECT_STMT:
@@ -4562,6 +4639,10 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
       return jsonToDatabaseOptions(pJson, pObj);
     case QUERY_NODE_LEFT_VALUE:
       return TSDB_CODE_SUCCESS;  // SLeftValueNode has no fields to deserialize.
+    case QUERY_NODE_WHEN_THEN:
+      return jsonToWhenThenNode(pJson, pObj);
+    case QUERY_NODE_CASE_WHEN:
+      return jsonToCaseWhenNode(pJson, pObj);
     case QUERY_NODE_SET_OPERATOR:
       return jsonToSetOperator(pJson, pObj);
     case QUERY_NODE_SELECT_STMT:

source/libs/nodes/src/nodesEqualFuncs.c

@@ -140,6 +140,19 @@ static bool functionNodeEqual(const SFunctionNode* a, const SFunctionNode* b) {
   return true;
 }
 
+static bool whenThenNodeEqual(const SWhenThenNode* a, const SWhenThenNode* b) {
+  COMPARE_NODE_FIELD(pWhen);
+  COMPARE_NODE_FIELD(pThen);
+  return true;
+}
+
+static bool caseWhenNodeEqual(const SCaseWhenNode* a, const SCaseWhenNode* b) {
+  COMPARE_NODE_FIELD(pCase);
+  COMPARE_NODE_FIELD(pElse);
+  COMPARE_NODE_LIST_FIELD(pWhenThenList);
+  return true;
+}
+
 bool nodesEqualNode(const SNode* a, const SNode* b) {
   if (a == b) {
     return true;
@@ -164,13 +177,17 @@ bool nodesEqualNode(const SNode* a, const SNode* b) {
       return logicConditionNodeEqual((const SLogicConditionNode*)a, (const SLogicConditionNode*)b);
     case QUERY_NODE_FUNCTION:
       return functionNodeEqual((const SFunctionNode*)a, (const SFunctionNode*)b);
+    case QUERY_NODE_WHEN_THEN:
+      return whenThenNodeEqual((const SWhenThenNode*)a, (const SWhenThenNode*)b);
+    case QUERY_NODE_CASE_WHEN:
+      return caseWhenNodeEqual((const SCaseWhenNode*)a, (const SCaseWhenNode*)b);
     case QUERY_NODE_REAL_TABLE:
     case QUERY_NODE_TEMP_TABLE:
     case QUERY_NODE_JOIN_TABLE:
     case QUERY_NODE_GROUPING_SET:
     case QUERY_NODE_ORDER_BY_EXPR:
     case QUERY_NODE_LIMIT:
-      return false;  // todo
+      return false;
     default:
       break;
   }

source/libs/nodes/src/nodesMsgFuncs.c

@@ -1733,6 +1733,92 @@ static int32_t msgToDownstreamSourceNode(STlvDecoder* pDecoder, void* pObj) {
   return code;
 }
 
+enum { WHEN_THEN_CODE_EXPR_BASE = 1, WHEN_THEN_CODE_WHEN, WHEN_THEN_CODE_THEN };
+
+static int32_t whenThenNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
+  const SWhenThenNode* pNode = (const SWhenThenNode*)pObj;
+
+  int32_t code = tlvEncodeObj(pEncoder, WHEN_THEN_CODE_EXPR_BASE, exprNodeToMsg, pNode);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, WHEN_THEN_CODE_WHEN, nodeToMsg, pNode->pWhen);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, WHEN_THEN_CODE_THEN, nodeToMsg, pNode->pThen);
+  }
+
+  return code;
+}
+
+static int32_t msgToWhenThenNode(STlvDecoder* pDecoder, void* pObj) {
+  SWhenThenNode* pNode = (SWhenThenNode*)pObj;
+
+  int32_t code = TSDB_CODE_SUCCESS;
+  STlv*   pTlv = NULL;
+  tlvForEach(pDecoder, pTlv, code) {
+    switch (pTlv->type) {
+      case WHEN_THEN_CODE_EXPR_BASE:
+        code = tlvDecodeObjFromTlv(pTlv, msgToExprNode, &pNode->node);
+        break;
+      case WHEN_THEN_CODE_WHEN:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pWhen);
+        break;
+      case WHEN_THEN_CODE_THEN:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pThen);
+        break;
+      default:
+        break;
+    }
+  }
+
+  return code;
+}
+
+enum { CASE_WHEN_CODE_EXPR_BASE = 1, CASE_WHEN_CODE_CASE, CASE_WHEN_CODE_ELSE, CASE_WHEN_CODE_WHEN_THEN_LIST };
+
+static int32_t caseWhenNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
+  const SCaseWhenNode* pNode = (const SCaseWhenNode*)pObj;
+
+  int32_t code = tlvEncodeObj(pEncoder, CASE_WHEN_CODE_EXPR_BASE, exprNodeToMsg, pNode);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, CASE_WHEN_CODE_CASE, nodeToMsg, pNode->pCase);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, CASE_WHEN_CODE_ELSE, nodeToMsg, pNode->pElse);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, CASE_WHEN_CODE_WHEN_THEN_LIST, nodeListToMsg, pNode->pWhenThenList);
+  }
+
+  return code;
+}
+
+static int32_t msgToCaseWhenNode(STlvDecoder* pDecoder, void* pObj) {
+  SCaseWhenNode* pNode = (SCaseWhenNode*)pObj;
+
+  int32_t code = TSDB_CODE_SUCCESS;
+  STlv*   pTlv = NULL;
+  tlvForEach(pDecoder, pTlv, code) {
+    switch (pTlv->type) {
+      case CASE_WHEN_CODE_EXPR_BASE:
+        code = tlvDecodeObjFromTlv(pTlv, msgToExprNode, &pNode->node);
+        break;
+      case CASE_WHEN_CODE_CASE:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pCase);
+        break;
+      case CASE_WHEN_CODE_ELSE:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pElse);
+        break;
+      case CASE_WHEN_CODE_WHEN_THEN_LIST:
+        code = msgToNodeListFromTlv(pTlv, (void**)&pNode->pWhenThenList);
+        break;
+      default:
+        break;
+    }
+  }
+
+  return code;
+}
+
 enum {
   PHY_NODE_CODE_OUTPUT_DESC = 1,
   PHY_NODE_CODE_CONDITIONS,
@@ -3434,9 +3520,16 @@ static int32_t specificNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
       code = slotDescNodeToMsg(pObj, pEncoder);
       break;
     case QUERY_NODE_DOWNSTREAM_SOURCE:
-      return downstreamSourceNodeToMsg(pObj, pEncoder);
+      code = downstreamSourceNodeToMsg(pObj, pEncoder);
+      break;
     case QUERY_NODE_LEFT_VALUE:
       break;
+    case QUERY_NODE_WHEN_THEN:
+      code = whenThenNodeToMsg(pObj, pEncoder);
+      break;
+    case QUERY_NODE_CASE_WHEN:
+      code = caseWhenNodeToMsg(pObj, pEncoder);
+      break;
     case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
     case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
       code = physiScanNodeToMsg(pObj, pEncoder);
@@ -3563,9 +3656,15 @@ static int32_t msgToSpecificNode(STlvDecoder* pDecoder, void* pObj) {
       code = msgToSlotDescNode(pDecoder, pObj);
       break;
     case QUERY_NODE_DOWNSTREAM_SOURCE:
-      return msgToDownstreamSourceNode(pDecoder, pObj);
+      code = msgToDownstreamSourceNode(pDecoder, pObj);
     case QUERY_NODE_LEFT_VALUE:
       break;
+    case QUERY_NODE_WHEN_THEN:
+      code = msgToWhenThenNode(pDecoder, pObj);
+      break;
+    case QUERY_NODE_CASE_WHEN:
+      code = msgToCaseWhenNode(pDecoder, pObj);
+      break;
     case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
     case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
       code = msgToPhysiScanNode(pDecoder, pObj);

source/libs/nodes/src/nodesTraverseFuncs.c

@@ -146,6 +146,25 @@ static EDealRes dispatchExpr(SNode* pNode, ETraversalOrder order, FNodeWalker wa
     case QUERY_NODE_TARGET:
       res = walkExpr(((STargetNode*)pNode)->pExpr, order, walker, pContext);
       break;
+    case QUERY_NODE_WHEN_THEN: {
+      SWhenThenNode* pWhenThen = (SWhenThenNode*)pNode;
+      res = walkExpr(pWhenThen->pWhen, order, walker, pContext);
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = walkExpr(pWhenThen->pThen, order, walker, pContext);
+      }
+      break;
+    }
+    case QUERY_NODE_CASE_WHEN: {
+      SCaseWhenNode* pCaseWhen = (SCaseWhenNode*)pNode;
+      res = walkExpr(pCaseWhen->pCase, order, walker, pContext);
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = walkExpr(pCaseWhen->pElse, order, walker, pContext);
+      }
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = walkExprs(pCaseWhen->pWhenThenList, order, walker, pContext);
+      }
+      break;
+    }
     default:
       break;
   }
@@ -291,6 +310,25 @@ static EDealRes rewriteExpr(SNode** pRawNode, ETraversalOrder order, FNodeRewrit
     case QUERY_NODE_TARGET:
       res = rewriteExpr(&(((STargetNode*)pNode)->pExpr), order, rewriter, pContext);
       break;
+    case QUERY_NODE_WHEN_THEN: {
+      SWhenThenNode* pWhenThen = (SWhenThenNode*)pNode;
+      res = rewriteExpr(&pWhenThen->pWhen, order, rewriter, pContext);
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = rewriteExpr(&pWhenThen->pThen, order, rewriter, pContext);
+      }
+      break;
+    }
+    case QUERY_NODE_CASE_WHEN: {
+      SCaseWhenNode* pCaseWhen = (SCaseWhenNode*)pNode;
+      res = rewriteExpr(&pCaseWhen->pCase, order, rewriter, pContext);
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = rewriteExpr(&pCaseWhen->pElse, order, rewriter, pContext);
+      }
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = rewriteExprs(pCaseWhen->pWhenThenList, order, rewriter, pContext);
+      }
+      break;
+    }
     default:
       break;
   }

source/libs/nodes/src/nodesUtilFuncs.c

@@ -291,6 +291,10 @@ SNode* nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SLeftValueNode));
     case QUERY_NODE_COLUMN_REF:
       return makeNode(type, sizeof(SColumnDefNode));
+    case QUERY_NODE_WHEN_THEN:
+      return makeNode(type, sizeof(SWhenThenNode));
+    case QUERY_NODE_CASE_WHEN:
+      return makeNode(type, sizeof(SCaseWhenNode));
     case QUERY_NODE_SET_OPERATOR:
       return makeNode(type, sizeof(SSetOperator));
     case QUERY_NODE_SELECT_STMT:
@@ -738,7 +742,21 @@ void nodesDestroyNode(SNode* pNode) {
       break;
     }
     case QUERY_NODE_LEFT_VALUE:  // no pointer field
+    case QUERY_NODE_COLUMN_REF:  // no pointer field
       break;
+    case QUERY_NODE_WHEN_THEN: {
+      SWhenThenNode* pStmt = (SWhenThenNode*)pNode;
+      nodesDestroyNode(pStmt->pWhen);
+      nodesDestroyNode(pStmt->pThen);
+      break;
+    }
+    case QUERY_NODE_CASE_WHEN: {
+      SCaseWhenNode* pStmt = (SCaseWhenNode*)pNode;
+      nodesDestroyNode(pStmt->pCase);
+      nodesDestroyNode(pStmt->pElse);
+      nodesDestroyList(pStmt->pWhenThenList);
+      break;
+    }
     case QUERY_NODE_SET_OPERATOR: {
       SSetOperator* pStmt = (SSetOperator*)pNode;
       nodesDestroyList(pStmt->pProjectionList);

source/libs/parser/inc/parAst.h

@@ -119,6 +119,8 @@ SNode* createIntervalWindowNode(SAstCreateContext* pCxt, SNode* pInterval, SNode
 SNode* createFillNode(SAstCreateContext* pCxt, EFillMode mode, SNode* pValues);
 SNode* createGroupingSetNode(SAstCreateContext* pCxt, SNode* pNode);
 SNode* createInterpTimeRange(SAstCreateContext* pCxt, SNode* pStart, SNode* pEnd);
+SNode* createWhenThenNode(SAstCreateContext* pCxt, SNode* pWhen, SNode* pThen);
+SNode* createCaseWhenNode(SAstCreateContext* pCxt, SNode* pCase, SNodeList* pWhenThenList, SNode* pElse);
 
 SNode* addWhereClause(SAstCreateContext* pCxt, SNode* pStmt, SNode* pWhere);
 SNode* addPartitionByClause(SAstCreateContext* pCxt, SNode* pStmt, SNodeList* pPartitionByList);

source/libs/parser/src/parAstCreater.c

@@ -647,6 +647,25 @@ SNode* createInterpTimeRange(SAstCreateContext* pCxt, SNode* pStart, SNode* pEnd
   return createBetweenAnd(pCxt, createPrimaryKeyCol(pCxt, NULL), pStart, pEnd);
 }
 
+SNode* createWhenThenNode(SAstCreateContext* pCxt, SNode* pWhen, SNode* pThen) {
+  CHECK_PARSER_STATUS(pCxt);
+  SWhenThenNode* pWhenThen = (SWhenThenNode*)nodesMakeNode(QUERY_NODE_WHEN_THEN);
+  CHECK_OUT_OF_MEM(pWhenThen);
+  pWhenThen->pWhen = pWhen;
+  pWhenThen->pThen = pThen;
+  return (SNode*)pWhenThen;
+}
+
+SNode* createCaseWhenNode(SAstCreateContext* pCxt, SNode* pCase, SNodeList* pWhenThenList, SNode* pElse) {
+  CHECK_PARSER_STATUS(pCxt);
+  SCaseWhenNode* pCaseWhen = (SCaseWhenNode*)nodesMakeNode(QUERY_NODE_CASE_WHEN);
+  CHECK_OUT_OF_MEM(pCaseWhen);
+  pCaseWhen->pCase = pCase;
+  pCaseWhen->pWhenThenList = pWhenThenList;
+  pCaseWhen->pElse = pElse;
+  return (SNode*)pCaseWhen;
+}
+
 SNode* setProjectionAlias(SAstCreateContext* pCxt, SNode* pNode, SToken* pAlias) {
   CHECK_PARSER_STATUS(pCxt);
   trimEscape(pAlias);

source/libs/parser/src/parTokenizer.c

@@ -54,6 +54,7 @@ static SKeyword keywordTable[] = {
     {"CACHE",                TK_CACHE},
     {"CACHEMODEL",           TK_CACHEMODEL},
     {"CACHESIZE",            TK_CACHESIZE},
+    {"CASE",                 TK_CASE},
     {"CAST",                 TK_CAST},
     {"CLIENT_VERSION",       TK_CLIENT_VERSION},
     {"CLUSTER",              TK_CLUSTER},
@@ -82,7 +83,9 @@ static SKeyword keywordTable[] = {
     {"DOUBLE",               TK_DOUBLE},
     {"DROP",                 TK_DROP},
     {"DURATION",             TK_DURATION},
+    {"ELSE",                 TK_ELSE},
     {"ENABLE",               TK_ENABLE},
+    {"END",                  TK_END},
     {"EXISTS",               TK_EXISTS},
     {"EXPIRED",              TK_EXPIRED},
     {"EXPLAIN",              TK_EXPLAIN},
@@ -205,6 +208,7 @@ static SKeyword keywordTable[] = {
     {"TAG",                  TK_TAG},
     {"TAGS",                 TK_TAGS},
     {"TBNAME",               TK_TBNAME},
+    {"THEN",                 TK_THEN},
     {"TIMESTAMP",            TK_TIMESTAMP},
     {"TIMEZONE",             TK_TIMEZONE},
     {"TINYINT",              TK_TINYINT},
@@ -240,6 +244,7 @@ static SKeyword keywordTable[] = {
     {"WAL_ROLL_PERIOD",      TK_WAL_ROLL_PERIOD},
     {"WAL_SEGMENT_SIZE",     TK_WAL_SEGMENT_SIZE},
     {"WATERMARK",            TK_WATERMARK},
+    {"WHEN",                 TK_WHEN},
     {"WHERE",                TK_WHERE},
     {"WINDOW_CLOSE",         TK_WINDOW_CLOSE},
     {"WITH",                 TK_WITH},

source/libs/parser/src/parTranslater.c

@@ -1813,6 +1813,59 @@ static EDealRes translateLogicCond(STranslateContext* pCxt, SLogicConditionNode*
   return DEAL_RES_CONTINUE;
 }
 
+static int32_t createCastFunc(STranslateContext* pCxt, SNode* pExpr, SDataType dt, SNode** pCast) {
+  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
+  if (NULL == pFunc) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  strcpy(pFunc->functionName, "cast");
+  pFunc->node.resType = dt;
+  if (TSDB_CODE_SUCCESS != nodesListMakeAppend(&pFunc->pParameterList, pExpr)) {
+    nodesDestroyNode((SNode*)pFunc);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  if (TSDB_CODE_SUCCESS != getFuncInfo(pCxt, pFunc)) {
+    nodesClearList(pFunc->pParameterList);
+    pFunc->pParameterList = NULL;
+    nodesDestroyNode((SNode*)pFunc);
+    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)pExpr)->aliasName);
+  }
+  *pCast = (SNode*)pFunc;
+  return TSDB_CODE_SUCCESS;
+}
+
+static EDealRes translateWhenThen(STranslateContext* pCxt, SWhenThenNode* pWhenThen) {
+  pWhenThen->node.resType = ((SExprNode*)pWhenThen->pThen)->resType;
+  return DEAL_RES_CONTINUE;
+}
+
+static EDealRes translateCaseWhen(STranslateContext* pCxt, SCaseWhenNode* pCaseWhen) {
+  bool   first = true;
+  SNode* pNode = NULL;
+  FOREACH(pNode, pCaseWhen->pWhenThenList) {
+    if (first) {
+      pCaseWhen->node.resType = ((SExprNode*)pNode)->resType;
+    } else if (!dataTypeEqual(&pCaseWhen->node.resType, &((SExprNode*)pNode)->resType)) {
+      SWhenThenNode* pWhenThen = (SWhenThenNode*)pNode;
+      SNode*         pCastFunc = NULL;
+      if (TSDB_CODE_SUCCESS != createCastFunc(pCxt, pWhenThen->pThen, pCaseWhen->node.resType, &pCastFunc)) {
+        return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "CASE WHEN data type mismatch");
+      }
+      pWhenThen->pThen = pCastFunc;
+      pWhenThen->node.resType = pCaseWhen->node.resType;
+    }
+  }
+  if (NULL != pCaseWhen->pElse && !dataTypeEqual(&pCaseWhen->node.resType, &((SExprNode*)pCaseWhen->pElse)->resType)) {
+    SNode* pCastFunc = NULL;
+    if (TSDB_CODE_SUCCESS != createCastFunc(pCxt, pCaseWhen->pElse, pCaseWhen->node.resType, &pCastFunc)) {
+      return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "CASE WHEN data type mismatch");
+    }
+    pCaseWhen->pElse = pCastFunc;
+    ((SExprNode*)pCaseWhen->pElse)->resType = pCaseWhen->node.resType;
+  }
+  return DEAL_RES_CONTINUE;
+}
+
 static EDealRes doTranslateExpr(SNode** pNode, void* pContext) {
   STranslateContext* pCxt = (STranslateContext*)pContext;
   switch (nodeType(*pNode)) {
@@ -1828,6 +1881,10 @@ static EDealRes doTranslateExpr(SNode** pNode, void* pContext) {
       return translateLogicCond(pCxt, (SLogicConditionNode*)*pNode);
     case QUERY_NODE_TEMP_TABLE:
       return translateExprSubquery(pCxt, ((STempTableNode*)*pNode)->pSubquery);
+    case QUERY_NODE_WHEN_THEN:
+      return translateWhenThen(pCxt, (SWhenThenNode*)*pNode);
+    case QUERY_NODE_CASE_WHEN:
+      return translateCaseWhen(pCxt, (SCaseWhenNode*)*pNode);
     default:
       break;
   }
@@ -3228,27 +3285,6 @@ static SNode* createSetOperProject(const char* pTableAlias, SNode* pNode) {
   return (SNode*)pCol;
 }
 
-static int32_t createCastFunc(STranslateContext* pCxt, SNode* pExpr, SDataType dt, SNode** pCast) {
-  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
-  if (NULL == pFunc) {
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-  strcpy(pFunc->functionName, "cast");
-  pFunc->node.resType = dt;
-  if (TSDB_CODE_SUCCESS != nodesListMakeAppend(&pFunc->pParameterList, pExpr)) {
-    nodesDestroyNode((SNode*)pFunc);
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-  if (TSDB_CODE_SUCCESS != getFuncInfo(pCxt, pFunc)) {
-    nodesClearList(pFunc->pParameterList);
-    pFunc->pParameterList = NULL;
-    nodesDestroyNode((SNode*)pFunc);
-    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)pExpr)->aliasName);
-  }
-  *pCast = (SNode*)pFunc;
-  return TSDB_CODE_SUCCESS;
-}
-
 static int32_t translateSetOperProject(STranslateContext* pCxt, SSetOperator* pSetOperator) {
   SNodeList* pLeftProjections = getProjectList(pSetOperator->pLeft);
   SNodeList* pRightProjections = getProjectList(pSetOperator->pRight);

source/libs/parser/test/parSelectTest.cpp

@@ -60,6 +60,9 @@ TEST_F(ParserSelectTest, expression) {
   run("SELECT ts > 0, c1 between 10 and 20 and c2 = 'qaz' FROM t1");
 
   run("SELECT c1 | 10, c2 & 20, c4 | c5 FROM t1");
+
+  run("SELECT CASE WHEN ts > '2020-1-1 10:10:10' THEN c1 + 10 ELSE c1 - 10 END FROM t1 "
+      "WHERE CASE c1 WHEN c3 + 20 THEN c3 - 1 WHEN c3 + 10 THEN c3 - 2 ELSE 10 END > 0");
 }
 
 TEST_F(ParserSelectTest, condition) {
@@ -312,6 +315,8 @@ TEST_F(ParserSelectTest, subquery) {
   run("SELECT _C0 FROM (SELECT _ROWTS, ts FROM st1s1)");
 
   run("SELECT ts FROM (SELECT t1.ts FROM st1s1 t1)");
+
+  run("(((SELECT t1.ts FROM st1s1 t1)))");
 }
 
 TEST_F(ParserSelectTest, subquerySemanticCheck) {

source/libs/planner/test/planBasicTest.cpp

@@ -40,6 +40,13 @@ TEST_F(PlanBasicTest, whereClause) {
   run("SELECT ts, c1 FROM t1 WHERE ts > NOW AND ts IS NULL AND (c1 > 0 OR c3 < 20)");
 }
 
+TEST_F(PlanBasicTest, caseWhen) {
+  useDb("root", "test");
+
+  run("SELECT CASE WHEN ts > '2020-1-1 10:10:10' THEN c1 + 10 ELSE c1 - 10 END FROM t1 "
+      "WHERE CASE c1 WHEN c2 + 20 THEN c4 - 1 WHEN c2 + 10 THEN c4 - 2 ELSE 10 END > 0");
+}
+
 TEST_F(PlanBasicTest, func) {
   useDb("root", "test");
 

tests/system-test/1-insert/database_pre_suf.py

@@ -107,7 +107,7 @@ class TDTestCase:
 
         
         # create stream
-        tdSql.execute('''create stream current_stream into stream_max_stable_1 as select _wstart as start, _wend as end, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 where ts is not null interval (5s);''')
+        tdSql.execute('''create stream current_stream into stream_max_stable_1 as select _wstart as start, _wend as wend, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 where ts is not null interval (5s);''')
         
         # insert data
         for i in range(num_random*n):        
@@ -185,20 +185,20 @@ class TDTestCase:
         tdSql.checkData(0,0,num_random*n)
         
         # stream data check
-        tdSql.query("select start,end,max_int from stream_max_stable_1 ;")
+        tdSql.query("select start,wend,max_int from stream_max_stable_1 ;")
         tdSql.checkRows(20)
         tdSql.query("select sum(max_int) from stream_max_stable_1 ;")
         stream_data_1 = tdSql.queryResult[0][0]
         tdSql.query("select sum(min_int) from stream_max_stable_1 ;")
         stream_data_2 = tdSql.queryResult[0][0]
-        tdSql.query("select sum(max_int),sum(min_int) from (select _wstart as start, _wend as end, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 where ts is not null interval (5s));")
+        tdSql.query("select sum(max_int),sum(min_int) from (select _wstart as start, _wend as wend, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 where ts is not null interval (5s));")
         sql_data_1 = tdSql.queryResult[0][0]
         sql_data_2 = tdSql.queryResult[0][1]
         
         self.stream_value_check(stream_data_1,sql_data_1)       
         self.stream_value_check(stream_data_2,sql_data_2)
         
-        tdSql.query("select sum(max_int),sum(min_int) from (select _wstart as start, _wend as end, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 interval (5s));")
+        tdSql.query("select sum(max_int),sum(min_int) from (select _wstart as start, _wend as wend, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 interval (5s));")
         sql_data_1 = tdSql.queryResult[0][0]
         sql_data_2 = tdSql.queryResult[0][1]