tqRead.c 7.9 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

H
Hongze Cheng 已提交
16
#include "tq.h"
L
Liu Jicong 已提交
17 18

STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
wafwerar's avatar
wafwerar 已提交
19
  STqReadHandle* pReadHandle = taosMemoryMalloc(sizeof(STqReadHandle));
L
Liu Jicong 已提交
20 21 22 23 24 25 26 27
  if (pReadHandle == NULL) {
    return NULL;
  }
  pReadHandle->pVnodeMeta = pMeta;
  pReadHandle->pMsg = NULL;
  pReadHandle->ver = -1;
  pReadHandle->pColIdList = NULL;
  pReadHandle->sver = -1;
L
Liu Jicong 已提交
28
  pReadHandle->cachedSchemaUid = -1;
L
Liu Jicong 已提交
29 30
  pReadHandle->pSchema = NULL;
  pReadHandle->pSchemaWrapper = NULL;
L
Liu Jicong 已提交
31
  pReadHandle->tbIdHash = NULL;
L
Liu Jicong 已提交
32 33 34
  return pReadHandle;
}

L
Liu Jicong 已提交
35
int32_t tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t ver) {
L
Liu Jicong 已提交
36
  pReadHandle->pMsg = pMsg;
C
Cary Xu 已提交
37 38
  // pMsg->length = htonl(pMsg->length);
  // pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
L
Liu Jicong 已提交
39

L
Liu Jicong 已提交
40
  // iterate and convert
C
Cary Xu 已提交
41
  if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1;
L
Liu Jicong 已提交
42
  while (true) {
C
Cary Xu 已提交
43
    if (tGetSubmitMsgNext(&pReadHandle->msgIter, &pReadHandle->pBlock) < 0) return -1;
L
Liu Jicong 已提交
44 45
    if (pReadHandle->pBlock == NULL) break;

C
Cary Xu 已提交
46 47 48 49 50 51
    // pReadHandle->pBlock->uid = htobe64(pReadHandle->pBlock->uid);
    // pReadHandle->pBlock->suid = htobe64(pReadHandle->pBlock->suid);
    // pReadHandle->pBlock->sversion = htonl(pReadHandle->pBlock->sversion);
    // pReadHandle->pBlock->dataLen = htonl(pReadHandle->pBlock->dataLen);
    // pReadHandle->pBlock->schemaLen = htonl(pReadHandle->pBlock->schemaLen);
    // pReadHandle->pBlock->numOfRows = htons(pReadHandle->pBlock->numOfRows);
L
Liu Jicong 已提交
52 53
  }

C
Cary Xu 已提交
54
  if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1;
L
Liu Jicong 已提交
55 56
  pReadHandle->ver = ver;
  memset(&pReadHandle->blkIter, 0, sizeof(SSubmitBlkIter));
L
Liu Jicong 已提交
57
  return 0;
L
Liu Jicong 已提交
58 59 60 61
}

bool tqNextDataBlock(STqReadHandle* pHandle) {
  while (1) {
C
Cary Xu 已提交
62
    if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
L
Liu Jicong 已提交
63 64 65 66
      return false;
    }
    if (pHandle->pBlock == NULL) return false;

L
Liu Jicong 已提交
67
    /*pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);*/
L
Liu Jicong 已提交
68
    /*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
L
Liu Jicong 已提交
69 70 71
    if (pHandle->tbIdHash == NULL) {
      return true;
    }
C
Cary Xu 已提交
72
    void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->msgIter.uid, sizeof(int64_t));
L
Liu Jicong 已提交
73
    if (ret != NULL) {
74
      /*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/
L
Liu Jicong 已提交
75 76 77 78 79
      /*pHandle->pBlock->tid = htonl(pHandle->pBlock->tid);*/
      /*pHandle->pBlock->sversion = htonl(pHandle->pBlock->sversion);*/
      /*pHandle->pBlock->dataLen = htonl(pHandle->pBlock->dataLen);*/
      /*pHandle->pBlock->schemaLen = htonl(pHandle->pBlock->schemaLen);*/
      /*pHandle->pBlock->numOfRows = htons(pHandle->pBlock->numOfRows);*/
L
Liu Jicong 已提交
80
      return true;
L
Liu Jicong 已提交
81
      /*} else {*/
82
      /*printf("skip one tb %ld\n", pHandle->pBlock->uid);*/
L
Liu Jicong 已提交
83 84 85 86 87
    }
  }
  return false;
}

L
Liu Jicong 已提交
88 89
int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* pGroupId, uint64_t* pUid,
                            int32_t* pNumOfRows, int16_t* pNumOfCols) {
L
Liu Jicong 已提交
90 91
  /*int32_t         sversion = pHandle->pBlock->sversion;*/
  // TODO set to real sversion
92 93
  *pUid = 0;

H
Hongze Cheng 已提交
94
  int32_t sversion = 1;
L
Liu Jicong 已提交
95
  if (pHandle->sver != sversion || pHandle->cachedSchemaUid != pHandle->msgIter.suid) {
C
Cary Xu 已提交
96
    pHandle->pSchema = metaGetTbTSchema(pHandle->pVnodeMeta, pHandle->msgIter.uid, sversion);
L
Liu Jicong 已提交
97 98

    // this interface use suid instead of uid
C
Cary Xu 已提交
99
    pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.suid, sversion, true);
L
Liu Jicong 已提交
100
    pHandle->sver = sversion;
L
Liu Jicong 已提交
101
    pHandle->cachedSchemaUid = pHandle->msgIter.suid;
L
Liu Jicong 已提交
102 103 104 105 106
  }

  STSchema*       pTschema = pHandle->pSchema;
  SSchemaWrapper* pSchemaWrapper = pHandle->pSchemaWrapper;

C
bug fix  
Cary Xu 已提交
107
  *pNumOfRows = pHandle->msgIter.numOfRows;
L
Liu Jicong 已提交
108 109
  int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);

L
Liu Jicong 已提交
110 111 112 113 114
  if (colNumNeed == 0) {
    *ppCols = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData));
    if (*ppCols == NULL) {
      return -1;
    }
L
Liu Jicong 已提交
115

L
Liu Jicong 已提交
116 117 118
    int32_t colMeta = 0;
    while (colMeta < pSchemaWrapper->nCols) {
      SSchema*        pColSchema = &pSchemaWrapper->pSchema[colMeta];
L
Liu Jicong 已提交
119 120 121 122 123
      SColumnInfoData colInfo = {0};
      colInfo.info.bytes = pColSchema->bytes;
      colInfo.info.colId = pColSchema->colId;
      colInfo.info.type = pColSchema->type;

L
Liu Jicong 已提交
124
      if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
L
Liu Jicong 已提交
125
        goto FAIL;
L
Liu Jicong 已提交
126
      }
L
Liu Jicong 已提交
127
      taosArrayPush(*ppCols, &colInfo);
L
Liu Jicong 已提交
128
      colMeta++;
L
Liu Jicong 已提交
129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162
    }
  } else {
    if (colNumNeed > pSchemaWrapper->nCols) {
      colNumNeed = pSchemaWrapper->nCols;
    }

    *ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
    if (*ppCols == NULL) {
      return -1;
    }

    int32_t colMeta = 0;
    int32_t colNeed = 0;
    while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
      SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
      col_id_t colIdSchema = pColSchema->colId;
      col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pHandle->pColIdList, colNeed);
      if (colIdSchema < colIdNeed) {
        colMeta++;
      } else if (colIdSchema > colIdNeed) {
        colNeed++;
      } else {
        SColumnInfoData colInfo = {0};
        colInfo.info.bytes = pColSchema->bytes;
        colInfo.info.colId = pColSchema->colId;
        colInfo.info.type = pColSchema->type;

        if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
          goto FAIL;
        }
        taosArrayPush(*ppCols, &colInfo);
        colMeta++;
        colNeed++;
      }
L
Liu Jicong 已提交
163 164
    }
  }
L
Liu Jicong 已提交
165

L
Liu Jicong 已提交
166
  int32_t colActual = taosArrayGetSize(*ppCols);
L
Liu Jicong 已提交
167
  *pNumOfCols = colActual;
L
Liu Jicong 已提交
168 169 170 171

  // TODO in stream shuffle case, fetch groupId
  *pGroupId = 0;

L
Liu Jicong 已提交
172 173 174 175
  STSRowIter iter = {0};
  tdSTSRowIterInit(&iter, pTschema);
  STSRow* row;
  int32_t curRow = 0;
176

C
Cary Xu 已提交
177
  tInitSubmitBlkIter(&pHandle->msgIter, pHandle->pBlock, &pHandle->blkIter);
5
54liuyao 已提交
178
  *pUid = pHandle->msgIter.uid;  // set the uid of table for submit block
179

C
Cary Xu 已提交
180
  while ((row = tGetSubmitBlkNext(&pHandle->blkIter)) != NULL) {
L
Liu Jicong 已提交
181 182
    tdSTSRowIterReset(&iter, row);
    // get all wanted col of that block
L
Liu Jicong 已提交
183 184
    for (int32_t i = 0; i < colActual; i++) {
      SColumnInfoData* pColData = taosArrayGet(*ppCols, i);
L
Liu Jicong 已提交
185 186 187 188
      SCellVal         sVal = {0};
      if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
        break;
      }
L
Liu Jicong 已提交
189
      if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {
L
Liu Jicong 已提交
190
        goto FAIL;
L
Liu Jicong 已提交
191
      }
L
Liu Jicong 已提交
192 193 194
    }
    curRow++;
  }
L
Liu Jicong 已提交
195 196 197 198
  return 0;
FAIL:
  taosArrayDestroy(*ppCols);
  return -1;
L
Liu Jicong 已提交
199
}
H
Hongze Cheng 已提交
200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237

void tqReadHandleSetColIdList(STqReadHandle* pReadHandle, SArray* pColIdList) { pReadHandle->pColIdList = pColIdList; }

int tqReadHandleSetTbUidList(STqReadHandle* pHandle, const SArray* tbUidList) {
  if (pHandle->tbIdHash) {
    taosHashClear(pHandle->tbIdHash);
  }

  pHandle->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
  if (pHandle->tbIdHash == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  for (int i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
    taosHashPut(pHandle->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
  }

  return 0;
}

int tqReadHandleAddTbUidList(STqReadHandle* pHandle, const SArray* tbUidList) {
  if (pHandle->tbIdHash == NULL) {
    pHandle->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
    if (pHandle->tbIdHash == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return -1;
    }
  }

  for (int i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
    taosHashPut(pHandle->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
  }

  return 0;
}