From 3799db4366d954d36bd8d1fa14140a8ed7201838 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Thu, 23 Jun 2022 20:46:11 +0800
Subject: [PATCH] feat:sort table group if needed

---
 source/dnode/vnode/src/tsdb/tsdbRead.c  |  4 +++-
 source/libs/executor/src/scanoperator.c | 31 +++++++++++++++++++++----
 tests/system-test/2-query/json_tag.py   |  7 ++++++
 3 files changed, 37 insertions(+), 5 deletions(-)

diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c
index a76f7af7ef..d6e339c92a 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead.c
@@ -226,7 +226,6 @@ int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle) {
 
 static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, SArray* pTableList) {
   size_t tableSize = taosArrayGetSize(pTableList);
-  assert(tableSize >= 1);
 
   // allocate buffer in order to load data blocks from file
   SArray* pTableCheckInfo = taosArrayInit(tableSize, sizeof(STableCheckInfo));
@@ -510,6 +509,9 @@ int32_t tsdbSetTableList(tsdbReaderT reader, SArray* tableList){
 
 tsdbReaderT tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* tableList, uint64_t qId,
                             uint64_t taskId) {
+  if(taosArrayGetSize(tableList) == 0){
+    return NULL;
+  }
   STsdbReadHandle* pTsdbReadHandle = tsdbQueryTablesImpl(pVnode, pCond, qId, taskId);
   if (pTsdbReadHandle == NULL) {
     return NULL;
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index bde438f9d5..ccc9f4a3ef 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -510,6 +510,10 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
 
   if(pInfo->currentGroupId == -1){
     pInfo->currentGroupId++;
+    if (pInfo->currentGroupId >= taosArrayGetSize(pTaskInfo->tableqinfoList.pGroupList)) {
+      doSetOperatorCompleted(pOperator);
+      return NULL;
+    }
     SArray *tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
     tsdbReaderT* pReader = tsdbReaderOpen(pInfo->readHandle.vnode, &pInfo->cond, tableList, pInfo->queryId, pInfo->taskId);
     pInfo->dataReader = pReader;
@@ -2200,19 +2204,34 @@ SArray* generateSortByTsInfo(int32_t order) {
   return pList;
 }
 
+static int32_t createMultipleDataReaders(SQueryTableDataCond* pQueryCond, SReadHandle* pHandle, SArray* tableList, SArray* arrayReader, uint64_t queryId,
+                                  uint64_t taskId) {
+  for (int32_t i = 0; i < taosArrayGetSize(tableList); ++i) {
+    SArray* tmp = taosArrayInit(1, sizeof(STableKeyInfo));
+    taosArrayPush(tmp, taosArrayGet(tableList, i));
+
+    tsdbReaderT* pReader = tsdbReaderOpen(pHandle->vnode, pQueryCond, tmp, queryId, taskId);
+    taosArrayPush(arrayReader, &pReader);
+
+    taosArrayDestroy(tmp);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
   STableMergeScanInfo* pInfo = pOperator->info;
   SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;
 
   SArray* tableList = taosArrayGetP(pInfo->tableListInfo->pGroupList, pInfo->currentGroupId);
 
-  tsdbReaderT* pReader = tsdbReaderOpen(pInfo->readHandle.vnode, &pInfo->cond, tableList, pInfo->queryId, pInfo->taskId);
-  taosArrayPush(pInfo->dataReaders, &pReader);
+  createMultipleDataReaders(&pInfo->cond, &pInfo->readHandle, tableList,
+                            pInfo->dataReaders, pInfo->queryId, pInfo->taskId);
 
   // todo the total available buffer should be determined by total capacity of buffer of this task.
   // the additional one is reserved for merge result
   int32_t tableLen = taosArrayGetSize(tableList);
-  pInfo->sortBufSize = pInfo->bufPageSize * (tableLen + 1);
+  pInfo->sortBufSize = pInfo->bufPageSize * ((tableLen==0?1:tableLen) + 1);
   int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
   pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
                                              pInfo->pSortInputBlock, pTaskInfo->id.str);
@@ -2302,6 +2321,10 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
 
   if (pInfo->currentGroupId == -1) {
     pInfo->currentGroupId++;
+    if (pInfo->currentGroupId >= taosArrayGetSize(pInfo->tableListInfo->pGroupList)) {
+      doSetOperatorCompleted(pOperator);
+      return NULL;
+    }
     startGroupTableMergeScan(pOperator);
   }
   SSDataBlock* pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
@@ -2313,8 +2336,8 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
     return pBlock;
   }
 
-  pInfo->currentGroupId++;
   stopGroupTableMergeScan(pOperator);
+  pInfo->currentGroupId++;
   if (pInfo->currentGroupId >= taosArrayGetSize(pInfo->tableListInfo->pGroupList)) {
     doSetOperatorCompleted(pOperator);
     return NULL;
diff --git a/tests/system-test/2-query/json_tag.py b/tests/system-test/2-query/json_tag.py
index 0c649f2008..6816d4a3a3 100644
--- a/tests/system-test/2-query/json_tag.py
+++ b/tests/system-test/2-query/json_tag.py
@@ -412,6 +412,13 @@ class TDTestCase:
         tdSql.checkColNameList(res, cname_list)
         #
         # test group by & order by  json tag
+        tdSql.query("select ts,jtag->'tag1' from jsons1 partition by jtag->'tag1' order by jtag->'tag1' desc")
+        tdSql.checkRows(11)
+        tdSql.checkData(0, 1, '"femail"')
+        tdSql.checkData(2, 1, '"收到货"')
+        tdSql.checkData(7, 1, "false")
+
+
         # tdSql.error("select count(*) from jsons1 group by jtag")
         # tdSql.error("select count(*) from jsons1 partition by jtag")
         # tdSql.error("select count(*) from jsons1 group by jtag order by jtag")
-- 
GitLab