cachescanoperator.c 10.8 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "function.h"
H
Hongze Cheng 已提交
17
#include "os.h"
18 19 20 21 22 23 24 25 26 27
#include "tname.h"

#include "tdatablock.h"
#include "tmsg.h"

#include "executorimpl.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"

28
static SSDataBlock* doScanCache(SOperatorInfo* pOperator);
H
Hongze Cheng 已提交
29
static void         destroyLastrowScanOperator(void* param);
30
static int32_t      extractCacheScanSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds);
H
Haojun Liao 已提交
31
static int32_t      removeRedundantTsCol(SLastRowScanPhysiNode* pScanNode, SColMatchInfo* pColMatchInfo);
32

H
Haojun Liao 已提交
33 34
#define SCAN_ROW_TYPE(_t)  ((_t)? CACHESCAN_RETRIEVE_LAST : CACHESCAN_RETRIEVE_LAST_ROW)

35 36 37
SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SReadHandle* readHandle,
                                           SExecTaskInfo* pTaskInfo) {
  int32_t           code = TSDB_CODE_SUCCESS;
38 39 40
  SLastrowScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SLastrowScanInfo));
  SOperatorInfo*    pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
41
    code = TSDB_CODE_OUT_OF_MEMORY;
42 43 44 45
    goto _error;
  }

  pInfo->readHandle = *readHandle;
46 47 48

  SDataBlockDescNode* pDescNode = pScanNode->scan.node.pOutputDataBlockDesc;
  pInfo->pRes = createResDataBlock(pDescNode);
49 50

  int32_t numOfCols = 0;
51 52
  code =
      extractColMatchInfo(pScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);
53 54 55 56
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
57
  removeRedundantTsCol(pScanNode, &pInfo->matchInfo);
58

H
Haojun Liao 已提交
59
  code = extractCacheScanSlotId(pInfo->matchInfo.pList, pTaskInfo, &pInfo->pSlotIds);
60 61 62 63
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
64
  STableListInfo* pTableList = pTaskInfo->pTableInfoList;
65

H
Haojun Liao 已提交
66 67 68
  int32_t totalTables = tableListGetSize(pTableList);
  int32_t capacity = 0;

69 70
  pInfo->pUidList = taosArrayInit(4, sizeof(int64_t));

H
Haojun Liao 已提交
71 72 73
  // partition by tbname
  if (oneTableForEachGroup(pTableList) || (totalTables == 1)) {
    pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_ALL | SCAN_ROW_TYPE(pScanNode->ignoreNull);
74

H
Haojun Liao 已提交
75
    STableKeyInfo* pList = tableListGetInfo(pTableList, 0);
76

H
Haojun Liao 已提交
77 78
    size_t   num = tableListGetSize(pTableList);
    uint64_t suid = tableListGetSuid(pTableList);
79
    code = tsdbCacherowsReaderOpen(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,
H
Haojun Liao 已提交
80
                                   taosArrayGetSize(pInfo->matchInfo.pList), suid, &pInfo->pLastrowReader);
81 82 83 84
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }

H
Haojun Liao 已提交
85 86
    capacity = TMIN(totalTables, 4096);

87
    pInfo->pBufferredRes = createOneDataBlock(pInfo->pRes, false);
H
Haojun Liao 已提交
88
    blockDataEnsureCapacity(pInfo->pBufferredRes, capacity);
89
  } else {  // by tags
H
Haojun Liao 已提交
90 91
    pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_SINGLE | SCAN_ROW_TYPE(pScanNode->ignoreNull);
    capacity = 1;  // only one row output
92
  }
93

H
Haojun Liao 已提交
94 95 96
  initResultSizeInfo(&pOperator->resultInfo, capacity);
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

X
Xiaoyu Wang 已提交
97
  if (pScanNode->scan.pScanPseudoCols != NULL) {
98 99 100
    SExprSupp* p = &pInfo->pseudoExprSup;
    p->pExprInfo = createExprInfo(pScanNode->scan.pScanPseudoCols, NULL, &p->numOfExprs);
    p->pCtx = createSqlFunctionCtx(p->pExprInfo, p->numOfExprs, &p->rowEntryInfoOffset);
101
  }
102

H
Haojun Liao 已提交
103
  setOperatorInfo(pOperator, "CachedRowScanOperator", QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
104 105 106
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);

  pOperator->fpSet =
H
Haojun Liao 已提交
107
      createOperatorFpSet(operatorDummyOpenFn, doScanCache, NULL, destroyLastrowScanOperator, NULL);
108

109 110 111
  pOperator->cost.openCost = 0;
  return pOperator;

112
  _error:
113 114
  pTaskInfo->code = code;
  destroyLastrowScanOperator(pInfo);
115 116 117 118
  taosMemoryFree(pOperator);
  return NULL;
}

119
SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
120 121 122 123 124 125
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SLastrowScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*    pTaskInfo = pOperator->pTaskInfo;
H
Haojun Liao 已提交
126 127 128 129
  STableListInfo*   pTableList = pTaskInfo->pTableInfoList;

  uint64_t suid = tableListGetSuid(pTableList);
  int32_t  size = tableListGetSize(pTableList);
130
  if (size == 0) {
H
Haojun Liao 已提交
131
    setOperatorCompleted(pOperator);
132 133 134
    return NULL;
  }

135 136
  blockDataCleanup(pInfo->pRes);

137
  // check if it is a group by tbname
138
  if ((pInfo->retrieveType & CACHESCAN_RETRIEVE_TYPE_ALL) == CACHESCAN_RETRIEVE_TYPE_ALL) {
139 140 141 142
    if (pInfo->indexOfBufferedRes >= pInfo->pBufferredRes->info.rows) {
      blockDataCleanup(pInfo->pBufferredRes);
      taosArrayClear(pInfo->pUidList);

H
Hongze Cheng 已提交
143 144
      int32_t code =
          tsdbRetrieveCacheRows(pInfo->pLastrowReader, pInfo->pBufferredRes, pInfo->pSlotIds, pInfo->pUidList);
145
      if (code != TSDB_CODE_SUCCESS) {
146
        T_LONG_JMP(pTaskInfo->env, code);
147 148 149 150
      }

      // check for tag values
      int32_t resultRows = pInfo->pBufferredRes->info.rows;
151 152 153

      // the results may be null, if last values are all null
      ASSERT(resultRows == 0 || resultRows == taosArrayGetSize(pInfo->pUidList));
154 155 156
      pInfo->indexOfBufferedRes = 0;
    }

157 158
    SSDataBlock* pRes = pInfo->pRes;

159
    if (pInfo->indexOfBufferedRes < pInfo->pBufferredRes->info.rows) {
H
Haojun Liao 已提交
160 161 162
      for (int32_t i = 0; i < taosArrayGetSize(pInfo->matchInfo.pList); ++i) {
        SColMatchItem* pMatchInfo = taosArrayGet(pInfo->matchInfo.pList, i);
        int32_t        slotId = pMatchInfo->dstSlotId;
163 164

        SColumnInfoData* pSrc = taosArrayGet(pInfo->pBufferredRes->pDataBlock, slotId);
165
        SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, slotId);
166 167

        char* p = colDataGetData(pSrc, pInfo->indexOfBufferedRes);
H
Hongze Cheng 已提交
168
        bool  isNull = colDataIsNull_s(pSrc, pInfo->indexOfBufferedRes);
169 170 171
        colDataAppend(pDst, 0, p, isNull);
      }

172 173
      pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, pInfo->indexOfBufferedRes);
      pRes->info.rows = 1;
174

H
Haojun Liao 已提交
175 176
      SExprSupp* pSup = &pInfo->pseudoExprSup;
      int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes,
177
                                            pRes->info.rows, GET_TASKID(pTaskInfo), NULL);
H
Haojun Liao 已提交
178 179 180
      if (code != TSDB_CODE_SUCCESS) {
        pTaskInfo->code = code;
        return NULL;
181 182
      }

183
      pRes->info.groupId = getTableGroupId(pTableList, pRes->info.uid);
184
      pInfo->indexOfBufferedRes += 1;
185
      return pRes;
186
    } else {
H
Haojun Liao 已提交
187
      setOperatorCompleted(pOperator);
188
      return NULL;
189
    }
190
  } else {
H
Haojun Liao 已提交
191
    size_t totalGroups = tableListGetOutputGroups(pTableList);
192 193

    while (pInfo->currentGroupIndex < totalGroups) {
194 195 196
      STableKeyInfo* pList = NULL;
      int32_t num = 0;

H
Haojun Liao 已提交
197
      int32_t code = tableListGetGroupList(pTableList, pInfo->currentGroupIndex, &pList, &num);
198 199 200 201 202
      if (code != TSDB_CODE_SUCCESS) {
        T_LONG_JMP(pTaskInfo->env, code);
      }

      tsdbCacherowsReaderOpen(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,
H
Haojun Liao 已提交
203
                              taosArrayGetSize(pInfo->matchInfo.pList), suid, &pInfo->pLastrowReader);
204 205
      taosArrayClear(pInfo->pUidList);

206
      code = tsdbRetrieveCacheRows(pInfo->pLastrowReader, pInfo->pRes, pInfo->pSlotIds, pInfo->pUidList);
207
      if (code != TSDB_CODE_SUCCESS) {
208
        T_LONG_JMP(pTaskInfo->env, code);
209
      }
210

211 212 213 214 215 216 217
      pInfo->currentGroupIndex += 1;

      // check for tag values
      if (pInfo->pRes->info.rows > 0) {
        if (pInfo->pseudoExprSup.numOfExprs > 0) {
          SExprSupp* pSup = &pInfo->pseudoExprSup;

218
          STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pTableList)[0];
219 220
          pInfo->pRes->info.groupId = pKeyInfo->groupId;

H
Haojun Liao 已提交
221 222 223 224
          if (taosArrayGetSize(pInfo->pUidList) > 0) {
            ASSERT((pInfo->retrieveType & CACHESCAN_RETRIEVE_LAST_ROW) == CACHESCAN_RETRIEVE_LAST_ROW);

            pInfo->pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, 0);
225
            code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes, pInfo->pRes->info.rows,
226
                                          GET_TASKID(pTaskInfo), NULL);
H
Haojun Liao 已提交
227 228 229 230
            if (code != TSDB_CODE_SUCCESS) {
              pTaskInfo->code = code;
              return NULL;
            }
231
          }
232 233
        }

H
Haojun Liao 已提交
234
        pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
235 236
        return pInfo->pRes;
      }
237 238
    }

H
Haojun Liao 已提交
239
    setOperatorCompleted(pOperator);
240
    return NULL;
241 242 243
  }
}

244
void destroyLastrowScanOperator(void* param) {
245 246
  SLastrowScanInfo* pInfo = (SLastrowScanInfo*)param;
  blockDataDestroy(pInfo->pRes);
H
Haojun Liao 已提交
247 248 249
  blockDataDestroy(pInfo->pBufferredRes);
  taosMemoryFree(pInfo->pSlotIds);
  taosArrayDestroy(pInfo->pUidList);
H
Haojun Liao 已提交
250
  taosArrayDestroy(pInfo->matchInfo.pList);
H
Haojun Liao 已提交
251 252 253 254 255

  if (pInfo->pLastrowReader != NULL) {
    pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
  }

256 257 258
  taosMemoryFreeClear(param);
}

259
int32_t extractCacheScanSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds) {
H
Hongze Cheng 已提交
260
  size_t numOfCols = taosArrayGetSize(pColMatchInfo);
261 262

  *pSlotIds = taosMemoryMalloc(numOfCols * sizeof(int32_t));
H
Hongze Cheng 已提交
263
  if (*pSlotIds == NULL) {
264 265 266
    return TSDB_CODE_OUT_OF_MEMORY;
  }

267 268
  SSchemaWrapper* pWrapper = pTaskInfo->schemaInfo.sw;

269
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
270
    SColMatchItem* pColMatch = taosArrayGet(pColMatchInfo, i);
271
    for (int32_t j = 0; j < pWrapper->nCols; ++j) {
272
      if (pColMatch->colId == pWrapper->pSchema[j].colId && pColMatch->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
273
        (*pSlotIds)[pColMatch->dstSlotId] = -1;
274 275 276
        break;
      }

277
      if (pColMatch->colId == pWrapper->pSchema[j].colId) {
H
Haojun Liao 已提交
278
        (*pSlotIds)[pColMatch->dstSlotId] = j;
279 280 281 282 283 284
        break;
      }
    }
  }

  return TSDB_CODE_SUCCESS;
285
}
286

H
Haojun Liao 已提交
287
int32_t removeRedundantTsCol(SLastRowScanPhysiNode* pScanNode, SColMatchInfo* pColMatchInfo) {
288
  if (!pScanNode->ignoreNull) {  // retrieve cached last value
H
Haojun Liao 已提交
289
    return TSDB_CODE_SUCCESS;
290 291
  }

H
Haojun Liao 已提交
292
  size_t size = taosArrayGetSize(pColMatchInfo->pList);
293
  SArray* pMatchInfo = taosArrayInit(size, sizeof(SColMatchItem));
294

H
Haojun Liao 已提交
295 296
  for (int32_t i = 0; i < size; ++i) {
    SColMatchItem* pColInfo = taosArrayGet(pColMatchInfo->pList, i);
297

H
Haojun Liao 已提交
298
    int32_t    slotId = pColInfo->dstSlotId;
299 300 301 302 303 304 305 306
    SNodeList* pList = pScanNode->scan.node.pOutputDataBlockDesc->pSlots;

    SSlotDescNode* pDesc = (SSlotDescNode*)nodesListGetNode(pList, slotId);
    if (pDesc->dataType.type != TSDB_DATA_TYPE_TIMESTAMP) {
      taosArrayPush(pMatchInfo, pColInfo);
    }
  }

H
Haojun Liao 已提交
307 308 309
  taosArrayDestroy(pColMatchInfo->pList);
  pColMatchInfo->pList = pMatchInfo;
  return TSDB_CODE_SUCCESS;
310
}