[td-11818] refactor ssdatablock, add plan in log.

include/common/common.h

@@ -38,6 +38,12 @@
 //  int16_t bytes;
 //} SSchema;
 
+typedef struct {
+  uint32_t  numOfTables;
+  SArray   *pGroupList;
+  SHashObj *map;  // speedup acquire the tableQueryInfo by table uid
+} STableGroupInfo;
+
 typedef struct SColumnDataAgg {
   int16_t colId;
   int64_t sum;
@@ -57,17 +63,12 @@ typedef struct SDataBlockInfo {
 
 typedef struct SConstantItem {
   SColumnInfo info;
-  int32_t     startIndex;    // run-length-encoding to save the space for multiple rows
-  int32_t     endIndex;
+  int32_t     startRow;    // run-length-encoding to save the space for multiple rows
+  int32_t     endRow;
   SVariant    value;
 } SConstantItem;
 
-typedef struct {
-  uint32_t  numOfTables;
-  SArray   *pGroupList;
-  SHashObj *map;  // speedup acquire the tableQueryInfo by table uid
-} STableGroupInfo;
-
+// info.numOfCols = taosArrayGetSize(pDataBlock) + taosArrayGetSize(pConstantList);
 typedef struct SSDataBlock {
   SColumnDataAgg *pBlockAgg;
   SArray         *pDataBlock;    // SArray<SColumnInfoData>
@@ -75,9 +76,12 @@ typedef struct SSDataBlock {
   SDataBlockInfo  info;
 } SSDataBlock;
 
+// pBlockAgg->numOfNull == info.rows, all data are null
+// pBlockAgg->numOfNull == 0, no data are null.
 typedef struct SColumnInfoData {
-  SColumnInfo info;     // TODO filter info needs to be removed
-  char       *pData;    // the corresponding block data in memory
+  SColumnInfo info;      // TODO filter info needs to be removed
+  char       *nullbitmap;//
+  char       *pData;     // the corresponding block data in memory
 } SColumnInfoData;
 
 //======================================================================================================================

source/client/src/clientImpl.c

@@ -218,12 +218,10 @@ int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag,
 
   if (pQueryNode->type == TSDB_SQL_SELECT) {
     setResSchemaInfo(&pRequest->body.resInfo, pSchema, numOfCols);
-    tfree(pSchema);
     pRequest->type = TDMT_VND_QUERY;
-  } else {
-    tfree(pSchema);
   }
 
+  tfree(pSchema);
   return code;
 }
 

source/common/src/tep.c

@@ -1,4 +1,5 @@
 #include "tep.h"
+#include "common.h"
 #include "tglobal.h"
 #include "tlockfree.h"
 
@@ -59,3 +60,99 @@ SEpSet getEpSet_s(SCorEpSet *pEpSet) {
   return ep;
 }
 
+bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row, SColumnDataAgg* pColAgg) {
+  if (pColAgg != NULL) {
+    if (pColAgg->numOfNull == totalRows) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return true;
+    } else if (pColAgg->numOfNull == 0) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return false;
+    }
+  }
+
+  if (pColumnInfoData->nullbitmap == NULL) {
+    return false;
+  }
+
+  uint8_t v = (pColumnInfoData->nullbitmap[row>>3] & (1<<(8 - (row&0x07))));
+  return (v == 1);
+}
+
+bool colDataIsNull_f(const char* bitmap, uint32_t row) {
+  return (bitmap[row>>3] & (1<<(8 - (row&0x07))));
+}
+
+void colDataSetNull_f(char* bitmap, uint32_t row) { // TODO
+  return;
+}
+
+void* colDataGet(const SColumnInfoData* pColumnInfoData, uint32_t row) {
+  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    uint32_t offset = ((uint32_t*)pColumnInfoData->pData)[row];
+    return (char*)(pColumnInfoData->pData) + offset; // the first part is the pointer to the true binary data
+  } else {
+    return (char*)(pColumnInfoData->pData) + (row * pColumnInfoData->info.bytes);
+  }
+}
+
+int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull) {
+  ASSERT(pColumnInfoData != NULL);
+
+  if (isNull) {
+    // TODO set null value in the nullbitmap
+    return 0;
+  }
+
+  int32_t type = pColumnInfoData->info.type;
+  if (IS_VAR_DATA_TYPE(type)) {
+    // TODO continue append var_type
+  } else {
+    char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
+    switch(type) {
+      case TSDB_DATA_TYPE_TINYINT:
+      case TSDB_DATA_TYPE_UTINYINT: {*(int8_t*) p = *(int8_t*) pData;break;}
+      default:
+        assert(0);
+    }
+
+  }
+
+  return 0;
+}
+
+size_t colDataGetCols(const SSDataBlock* pBlock) {
+  ASSERT(pBlock);
+
+  size_t constantCols = (pBlock->pConstantList != NULL)? taosArrayGetSize(pBlock->pConstantList):0;
+  ASSERT( pBlock->info.numOfCols == taosArrayGetSize(pBlock->pDataBlock) + constantCols);
+  return pBlock->info.numOfCols;
+}
+
+size_t colDataGetRows(const SSDataBlock* pBlock) {
+  return pBlock->info.rows;
+}
+
+int32_t colDataUpdateTsWindow(SSDataBlock* pDataBlock) {
+  if (pDataBlock == NULL || pDataBlock->info.rows <= 0) {
+    return 0;
+  }
+
+  if (pDataBlock->info.numOfCols <= 0) {
+    return -1;
+  }
+
+  SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, 0);
+  if (pColInfoData->info.type != TSDB_DATA_TYPE_TIMESTAMP) {
+    return 0;
+  }
+
+  ASSERT(pColInfoData->nullbitmap == NULL);
+  pDataBlock->info.window.skey = *(TSKEY*) colDataGet(pColInfoData, 0);
+  pDataBlock->info.window.ekey = *(TSKEY*) colDataGet(pColInfoData, (pDataBlock->info.rows - 1));
+  return 0;
+}
+
+
+
+

source/libs/planner/src/physicalPlanJson.c

@@ -1125,8 +1125,6 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
   *str = cJSON_Print(json);
   cJSON_Delete(json);
 
-//  printf("====Physical plan:====\n");
-//  printf("%s\n", *str);
   *len = strlen(*str) + 1;
   return TSDB_CODE_SUCCESS;
 }

source/libs/planner/src/planner.c

@@ -40,7 +40,8 @@ void qDestroyQueryDag(struct SQueryDag* pDag) {
   tfree(pDag);
 }
 
-int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, SSchema** pResSchema, int32_t* numOfCols, SArray* pNodeList, uint64_t requestId) {
+int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, SSchema** pResSchema, int32_t* numOfCols, SArray* pNodeList,
+    uint64_t requestId) {
   SQueryPlanNode* pLogicPlan;
   int32_t code = createQueryPlan(pNode, &pLogicPlan);
   if (TSDB_CODE_SUCCESS != code) {
@@ -49,9 +50,10 @@ int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag,
   }
 
   if (pLogicPlan->info.type != QNODE_MODIFY) {
-//   char* str = NULL;
-//   queryPlanToString(pLogicPlan, &str);
-//   printf("%s\n", str);
+    char* str = NULL;
+    queryPlanToString(pLogicPlan, &str);
+    qDebug("reqId:0x%"PRIx64": %s", requestId, str);
+    tfree(str);
   }
 
   code = optimizeQueryPlan(pLogicPlan);

source/libs/scheduler/src/scheduler.c

@@ -1194,17 +1194,18 @@ int32_t schLaunchTask(SSchJob *pJob, SSchTask *pTask) {
     
     code = atomic_load_32(&pJob->errCode);
     SCH_ERR_RET(code);
-    
     SCH_RET(TSDB_CODE_SCH_STATUS_ERROR);
   }
   
   SSubplan *plan = pTask->plan;
 
-  if (NULL == pTask->msg) {
+  if (NULL == pTask->msg) { // TODO add more detailed reason for failure
     code = qSubPlanToString(plan, &pTask->msg, &pTask->msgLen);
     if (TSDB_CODE_SUCCESS != code || NULL == pTask->msg || pTask->msgLen <= 0) {
-      SCH_TASK_ELOG("subplanToString error, code:%x, msg:%p, len:%d", code, pTask->msg, pTask->msgLen);
+      SCH_TASK_ELOG("failed to create physical plan, code:%s, msg:%p, len:%d", tstrerror(code), pTask->msg, pTask->msgLen);
       SCH_ERR_JRET(code);
+    } else {
+      SCH_TASK_DLOG(" ===physical plan=== len:%d, %s", pTask->msgLen, pTask->msg);
     }
   }
   
@@ -1218,13 +1219,10 @@ int32_t schLaunchTask(SSchJob *pJob, SSchTask *pTask) {
   }
 
   SCH_ERR_JRET(schBuildAndSendMsg(pJob, pTask, NULL, plan->msgType));
-  
   return TSDB_CODE_SUCCESS;
 
 _return:
-
   SCH_ERR_RET(schProcessOnTaskFailure(pJob, pTask, code));
-  
   SCH_RET(code);
 }
 

tools/shell/src/shellEngine.c

@@ -360,6 +360,7 @@ void shellRunCommandOnServer(TAOS *con, char command[]) {
     }
   } else {
     int num_rows_affacted = taos_affected_rows(pSql);
+    taos_free_result(pSql);
     et = taosGetTimestampUs();
     printf("Query OK, %d of %d row(s) in database (%.6fs)\n", num_rows_affacted, num_rows_affacted, (et - st) / 1E6);
   }