other: merge 3.0

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 2af2222
+        GIT_TAG 833b721
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

docs/en/14-reference/02-rest-api/02-rest-api.mdx

@@ -10,7 +10,7 @@ One difference from the native connector is that the REST interface is stateless
 
 ## Installation
 
-The REST interface does not rely on any TDengine native library, so the client application does not need to install any TDengine libraries. The client application's development language only needs to support the HTTP protocol.
+The REST interface does not rely on any TDengine native library, so the client application does not need to install any TDengine libraries. The client application's development language only needs to support the HTTP protocol. The REST interface is provided by [taosAdapter](../taosadapter), to use REST interface you need to make sure `taosAdapter` is running properly.
 
 ## Verification
 

docs/en/14-reference/03-connector/cpp.mdx → docs/en/14-reference/03-connector/03-cpp.mdx

 ---
-sidebar_position: 1
 sidebar_label: C/C++
 title: C/C++ Connector
 ---

docs/en/14-reference/03-connector/java.mdx → docs/en/14-reference/03-connector/04-java.mdx

 ---
 toc_max_heading_level: 4
-sidebar_position: 2
 sidebar_label: Java
 title: TDengine Java Connector
 description: The TDengine Java Connector is implemented on the standard JDBC API and provides native and REST connectors.

docs/en/14-reference/03-connector/go.mdx → docs/en/14-reference/03-connector/05-go.mdx

 ---
 toc_max_heading_level: 4
-sidebar_position: 4
 sidebar_label: Go
 title: TDengine Go Connector
 ---

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

 ---
 toc_max_heading_level: 4
-sidebar_position: 5
 sidebar_label: Rust
 title: TDengine Rust Connector
 ---

docs/en/14-reference/03-connector/python.mdx → docs/en/14-reference/03-connector/07-python.mdx

 ---
-sidebar_position: 3
 sidebar_label: Python
 title: TDengine Python Connector
 description: "taospy is the official Python connector for TDengine. taospy provides a rich API that makes it easy for Python applications to use TDengine. tasopy wraps both the native and REST interfaces of TDengine, corresponding to the two submodules of tasopy: taos and taosrest. In addition to wrapping the native and REST interfaces, taospy also provides a programming interface that conforms to the Python Data Access Specification (PEP 249), making it easy to integrate taospy with many third-party tools, such as SQLAlchemy and pandas."

docs/en/14-reference/03-connector/node.mdx → docs/en/14-reference/03-connector/08-node.mdx

 ---
 toc_max_heading_level: 4
-sidebar_position: 6
 sidebar_label: Node.js
 title: TDengine Node.js Connector
 ---

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

 ---
 toc_max_heading_level: 4
-sidebar_position: 7
 sidebar_label: C#
 title: C# Connector
 ---

docs/en/14-reference/03-connector/php.mdx → docs/en/14-reference/03-connector/10-php.mdx

 ---
-sidebar_position: 1
-sidebar_label: PHP (community contribution)
+sidebar_label: PHP
 title: PHP Connector
 ---
 

docs/zh/08-connector/02-rest-api.mdx

 ---
 title: REST API
+sidebar_label: REST API
+description: 详细介绍 TDengine 提供的 RESTful API.
 ---
 
 为支持各种不同类型平台的开发，TDengine 提供符合 REST 设计标准的 API，即 REST API。为最大程度降低学习成本，不同于其他数据库 REST API 的设计方法，TDengine 直接通过 HTTP POST 请求 BODY 中包含的 SQL 语句来操作数据库，仅需要一个 URL。REST 连接器的使用参见 [视频教程](https://www.taosdata.com/blog/2020/11/11/1965.html)。
@@ -10,7 +12,7 @@ title: REST API
 
 ## 安装
 
-RESTful 接口不依赖于任何 TDengine 的库，因此客户端不需要安装任何 TDengine 的库，只要客户端的开发语言支持 HTTP 协议即可。
+RESTful 接口不依赖于任何 TDengine 的库，因此客户端不需要安装任何 TDengine 的库，只要客户端的开发语言支持 HTTP 协议即可。TDengine 的 RESTful API 由 [taosAdapter](../../reference/taosadapter) 提供，在使用 RESTful API 之前需要确保 `taosAdapter` 正常运行。
 
 ## 验证
 

docs/zh/08-connector/03-cpp.mdx

 ---
-sidebar_position: 1
 sidebar_label: C/C++
 title: C/C++ Connector
 ---

docs/zh/08-connector/04-java.mdx

 ---
 toc_max_heading_level: 4
-sidebar_position: 2
 sidebar_label: Java
 title: TDengine Java Connector
 description: TDengine Java 连接器基于标准 JDBC API 实现, 并提供原生连接与 REST连接两种连接器。

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

 ---
 toc_max_heading_level: 4
-sidebar_position: 5
 sidebar_label: Rust
 title: TDengine Rust Connector
 ---

docs/zh/08-connector/07-python.mdx

 ---
-sidebar_position: 3
 sidebar_label: Python
 title: TDengine Python Connector
 description: "taospy 是 TDengine 的官方 Python 连接器。taospy 提供了丰富的 API， 使得 Python 应用可以很方便地使用 TDengine。tasopy 对 TDengine 的原生接口和 REST 接口都进行了封装， 分别对应 tasopy 的两个子模块：tasos 和 taosrest。除了对原生接口和 REST 接口的封装，taospy 还提供了符合 Python 数据访问规范(PEP 249)的编程接口。这使得 taospy 和很多第三方工具集成变得简单，比如 SQLAlchemy 和 pandas"

docs/zh/08-connector/08-node.mdx

 ---
 toc_max_heading_level: 4
-sidebar_position: 6
 sidebar_label: Node.js
 title: TDengine Node.js Connector
 ---

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

 ---
 toc_max_heading_level: 4
-sidebar_position: 7
 sidebar_label: C#
 title: C# Connector
 ---

docs/zh/08-connector/10-php.mdx

 ---
-sidebar_position: 1
-sidebar_label: PHP（社区贡献）
+sidebar_label: PHP
 title: PHP Connector
 ---
 

examples/c/stream_demo.c

@@ -96,7 +96,7 @@ int32_t create_stream() {
   taos_free_result(pRes);
 
   pRes = taos_query(pConn,
-                    "create stream stream1 trigger at_once watermark 10s into outstb as select _wstart start, k from st1 partition by tbname state_window(k)");
+                    "create stream stream1 trigger at_once watermark 10s into outstb as select _wstart start, avg(k) from st1 partition by tbname interval(10s)");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;

include/common/tcommon.h

@@ -44,6 +44,30 @@ enum {
 )
 // clang-format on
 
+typedef struct {
+  TSKEY    ts;
+  uint64_t groupId;
+} SWinKey;
+
+static inline int SWinKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
+  SWinKey* pWin1 = (SWinKey*)pKey1;
+  SWinKey* pWin2 = (SWinKey*)pKey2;
+
+  if (pWin1->groupId > pWin2->groupId) {
+    return 1;
+  } else if (pWin1->groupId < pWin2->groupId) {
+    return -1;
+  }
+
+  if (pWin1->ts > pWin2->ts) {
+    return 1;
+  } else if (pWin1->ts < pWin2->ts) {
+    return -1;
+  }
+
+  return 0;
+}
+
 enum {
   TMQ_MSG_TYPE__DUMMY = 0,
   TMQ_MSG_TYPE__POLL_RSP,

include/libs/stream/tstream.h

@@ -551,16 +551,17 @@ typedef struct {
 } SStreamStateCur;
 
 #if 1
-int32_t streamStatePut(SStreamState* pState, const void* key, int32_t kLen, const void* value, int32_t vLen);
-int32_t streamStateGet(SStreamState* pState, const void* key, int32_t kLen, void** pVal, int32_t* pVLen);
-int32_t streamStateDel(SStreamState* pState, const void* key, int32_t kLen);
-
-SStreamStateCur* streamStateGetCur(SStreamState* pState, const void* key, int32_t kLen);
-SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const void* key, int32_t kLen);
-SStreamStateCur* streamStateSeekKeyPrev(SStreamState* pState, const void* key, int32_t kLen);
+int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
+int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
+int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
+void    streamFreeVal(void* val);
+
+SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key);
+SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key);
+SStreamStateCur* streamStateSeekKeyPrev(SStreamState* pState, const SWinKey* key);
 void             streamStateFreeCur(SStreamStateCur* pCur);
 
-int32_t streamGetKVByCur(SStreamStateCur* pCur, void** pKey, int32_t* pKLen, void** pVal, int32_t* pVLen);
+int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
 
 int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);

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

@@ -652,17 +652,28 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask) {
 
   // expand executor
   if (pTask->taskLevel == TASK_LEVEL__SOURCE) {
+    pTask->pState = streamStateOpen(pTq->pStreamMeta->path, pTask);
+    if (pTask->pState == NULL) {
+      return -1;
+    }
+
     SReadHandle handle = {
         .meta = pTq->pVnode->pMeta,
         .vnode = pTq->pVnode,
         .initTqReader = 1,
+        .pStateBackend = pTask->pState,
     };
     pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
     ASSERT(pTask->exec.executor);
   } else if (pTask->taskLevel == TASK_LEVEL__AGG) {
+    pTask->pState = streamStateOpen(pTq->pStreamMeta->path, pTask);
+    if (pTask->pState == NULL) {
+      return -1;
+    }
     SReadHandle mgHandle = {
         .vnode = NULL,
         .numOfVgroups = (int32_t)taosArrayGetSize(pTask->childEpInfo),
+        .pStateBackend = pTask->pState,
     };
     pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &mgHandle);
     ASSERT(pTask->exec.executor);

source/libs/executor/src/scanoperator.c

@@ -1281,6 +1281,42 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
   SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;
   SStreamScanInfo* pInfo = pOperator->info;
 
+#if 0
+  SStreamState* pState = pTaskInfo->streamInfo.pState;
+  if (pState) {
+    printf(">>>>>>>> stream write backend\n");
+    SWinKey key = {
+        .ts = 1,
+        .groupId = 2,
+    };
+    char tmp[100] = "abcdefg1";
+    if (streamStatePut(pState, &key, &tmp, strlen(tmp) + 1) < 0) {
+      ASSERT(0);
+    }
+
+    key.ts = 2;
+    char tmp2[100] = "abcdefg2";
+    if (streamStatePut(pState, &key, &tmp2, strlen(tmp2) + 1) < 0) {
+      ASSERT(0);
+    }
+
+    key.groupId = 5;
+    key.ts = 1;
+    char tmp3[100] = "abcdefg3";
+    if (streamStatePut(pState, &key, &tmp3, strlen(tmp3) + 1) < 0) {
+      ASSERT(0);
+    }
+
+    char*   val2 = NULL;
+    int32_t sz;
+    if (streamStateGet(pState, &key, (void**)&val2, &sz) < 0) {
+      ASSERT(0);
+    }
+    printf("stream read %s %d\n", val2, sz);
+    streamFreeVal(val2);
+  }
+#endif
+
   qDebug("stream scan called");
   if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) {
     while (1) {

source/libs/stream/src/streamDispatch.c

@@ -358,7 +358,7 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
   FAIL_SHUFFLE_DISPATCH:
     if (pReqs) {
       for (int32_t i = 0; i < vgSz; i++) {
-        taosArrayDestroy(pReqs[i].data);
+        taosArrayDestroyP(pReqs[i].data, taosMemoryFree);
         taosArrayDestroy(pReqs[i].dataLen);
       }
       taosMemoryFree(pReqs);

source/libs/stream/src/streamState.c

@@ -13,175 +13,104 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "executor.h"
-#include "streamInc.h"
-#include "ttimer.h"
-
-SStreamState* streamStateOpen(char* path, SStreamTask* pTask) {
-  SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState));
-  if (pState == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-  char statePath[200];
-  sprintf(statePath, "%s/%d", path, pTask->taskId);
-  if (tdbOpen(statePath, 16 * 1024, 1, &pState->db) < 0) {
-    goto _err;
-  }
-
-  // open state storage backend
-  if (tdbTbOpen("state.db", sizeof(int32_t), -1, NULL, pState->db, &pState->pStateDb) < 0) {
-    goto _err;
-  }
-
-  pState->pOwner = pTask;
-
-  return pState;
-
-_err:
-  if (pState->pStateDb) tdbTbClose(pState->pStateDb);
-  if (pState->db) tdbClose(pState->db);
-  taosMemoryFree(pState);
-  return NULL;
-}
-
-void streamStateClose(SStreamState* pState) {
-  tdbCommit(pState->db, &pState->txn);
-  tdbTbClose(pState->pStateDb);
-  tdbClose(pState->db);
-
-  taosMemoryFree(pState);
-}
-
-int32_t streamStateBegin(SStreamState* pState) {
-  if (tdbTxnOpen(&pState->txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
-      0) {
+// clang-format off
+#include <assert.h>
+#include <stdio.h>
+#include <string.h>
+#include <time.h>
+#include "taos.h"
+
+int32_t init_env() {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  if (pConn == NULL) {
     return -1;
   }
 
-  if (tdbBegin(pState->db, &pState->txn) < 0) {
+  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create db, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
-  return 0;
-}
+  taos_free_result(pRes);
 
-int32_t streamStateCommit(SStreamState* pState) {
-  if (tdbCommit(pState->db, &pState->txn) < 0) {
+#if 0
+  pRes = taos_query(pConn, "create database if not exists abc2 vgroups 20");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create db, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
-  memset(&pState->txn, 0, sizeof(TXN));
-  if (tdbTxnOpen(&pState->txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
-      0) {
+  taos_free_result(pRes);
+#endif
+
+  pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in use db, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
-  if (tdbBegin(pState->db, &pState->txn) < 0) {
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
-  return 0;
-}
+  taos_free_result(pRes);
 
-int32_t streamStateAbort(SStreamState* pState) {
-  if (tdbAbort(pState->db, &pState->txn) < 0) {
+  pRes = taos_query(pConn, "create table if not exists tu1 using st1 tags(1)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
-  memset(&pState->txn, 0, sizeof(TXN));
-  if (tdbTxnOpen(&pState->txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
-      0) {
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags(2)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
-  if (tdbBegin(pState->db, &pState->txn) < 0) {
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table if not exists tu3 using st1 tags(3)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table tu3, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
-  return 0;
-}
-
-int32_t streamStatePut(SStreamState* pState, const void* key, int32_t kLen, const void* value, int32_t vLen) {
-  return tdbTbUpsert(pState->pStateDb, key, kLen, value, vLen, &pState->txn);
-}
-int32_t streamStateGet(SStreamState* pState, const void* key, int32_t kLen, void** pVal, int32_t* pVLen) {
-  return tdbTbGet(pState->pStateDb, key, kLen, pVal, pVLen);
-}
-
-int32_t streamStateDel(SStreamState* pState, const void* key, int32_t kLen) {
-  return tdbTbDelete(pState->pStateDb, key, kLen, &pState->txn);
-}
-
-SStreamStateCur* streamStateGetCur(SStreamState* pState, const void* key, int32_t kLen) {
-  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
-  if (pCur == NULL) return NULL;
-  tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL);
+  taos_free_result(pRes);
 
-  int32_t c;
-  tdbTbcMoveTo(pCur->pCur, key, kLen, &c);
-  if (c != 0) {
-    taosMemoryFree(pCur);
-    return NULL;
-  }
   return 0;
 }
 
-int32_t streamGetKVByCur(SStreamStateCur* pCur, void** pKey, int32_t* pKLen, void** pVal, int32_t* pVLen) {
-  return tdbTbcGet(pCur->pCur, (const void**)pKey, pKLen, (const void**)pVal, pVLen);
-}
-
-int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur) {
-  //
-  return tdbTbcMoveToFirst(pCur->pCur);
-}
-
-int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur) {
-  //
-  return tdbTbcMoveToLast(pCur->pCur);
-}
-
-SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const void* key, int32_t kLen) {
-  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
-  if (pCur == NULL) {
-    return NULL;
+int32_t create_stream() {
+  printf("create stream\n");
+  TAOS_RES* pRes;
+  TAOS*     pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  if (pConn == NULL) {
+    return -1;
   }
 
-  int32_t c;
-  if (tdbTbcMoveTo(pCur->pCur, key, kLen, &c) < 0) {
-    taosMemoryFree(pCur);
-    return NULL;
+  pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in use db, reason:%s\n", taos_errstr(pRes));
+    return -1;
   }
-  if (c > 0) return pCur;
+  taos_free_result(pRes);
 
-  if (tdbTbcMoveToNext(pCur->pCur) < 0) {
-    taosMemoryFree(pCur);
-    return NULL;
+  pRes = taos_query(pConn,
+                    "create stream stream1 trigger at_once watermark 10s into outstb as select _wstart start, avg(k) from st1 partition by tbname interval(10s)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
+    return -1;
   }
-
-  return pCur;
+  taos_free_result(pRes);
+  taos_close(pConn);
+  return 0;
 }
 
-SStreamStateCur* streamStateSeekKeyPrev(SStreamState* pState, const void* key, int32_t kLen) {
-  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
-  if (pCur == NULL) {
-    return NULL;
-  }
-
-  int32_t c;
-  if (tdbTbcMoveTo(pCur->pCur, key, kLen, &c) < 0) {
-    taosMemoryFree(pCur);
-    return NULL;
-  }
-  if (c < 0) return pCur;
-
-  if (tdbTbcMoveToPrev(pCur->pCur) < 0) {
-    taosMemoryFree(pCur);
-    return NULL;
+int main(int argc, char* argv[]) {
+  int code;
+  if (argc > 1) {
+    printf("env init\n");
+    code = init_env();
   }
-
-  return pCur;
-}
-
-int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur) {
-  //
-  return tdbTbcMoveToNext(pCur->pCur);
-}
-
-int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) {
-  //
-  return tdbTbcMoveToPrev(pCur->pCur);
+  create_stream();
 }